.*: add compactor for syncer #2261

GMHDBJD · 2021-10-25T10:38:26Z

What problem does this PR solve?

add compactor to compact dml

What is changed and how it works?

# Assuming that UPDATE statement does not update the primary key. If no primary key, choose a UNIQUE NOT NULL Key.
# If UPDATE statement update the primary key or unique key, split it into one DELETE and one INSERT statement.
# We list two successive DMLs of one row and the result that can be compacted. X means this situation will not happen.
INSERT + INSERT => X
INSERT + UPDATE => INSERT
INSERT + DELETE => DELETE
UPDATE + INSERT => X
UPDATE + UPDATE => UPDATE
UPDATE + DELETE => DELETE
DELETE + INSERT => UPDATE
DELETE + UPDATE => X
DELETE + DELETE => X

Check List

Tests

Integration test
UT

chaos/cases/conf/task-optimistic.yaml

dm/config/task.go

syncer/dml_worker.go

syncer/syncer.go

syncer/compactor.go

Co-authored-by: Ehco <[email protected]>

GMHDBJD · 2021-10-27T02:45:31Z

syncer/syncer.go

+	chanSize := s.cfg.QueueSize * s.cfg.WorkerCount / 2
+	if s.cfg.Compact {
+		chanSize /= 2
+	}
+	s.dmlJobCh = make(chan *job, chanSize)


DM originally cached s.cfg.QueueSize * s.cfg.WorkerCount dml jobs in memory. Now if compact: false, dmlJobCh will dmlWorker will both cached s.cfg.QueueSize * s.cfg.WorkerCount/2 jobs. If compact: true, dmlJobCh, compactor buffer, compactor output channel and dmlWorker will all cached s.cfg.QueueSize * s.cfg.WorkerCount/4 jobs.
Actually we can use a larger compact buffer-size, but if so, when user pause-task/stop-task, they may need to wait a longer time to wait all jobs flushed.

seems causality output channel didn't get adjusted

syncer/compactor_test.go

chaos/cases/conf/task-optimistic.yaml

Ehco1996

LGTM
if all chanSize and bufferSize is accept by other reviews, btw pelease add some comments when the parameters are determined

GMHDBJD · 2021-10-28T04:14:39Z

upgrade failed due to https://github.com/pingcap/dm/issues/1961

lance6716

will review after food

dm/config/task.go

syncer/compactor.go

lance6716 · 2021-10-28T04:15:47Z

syncer/dml.go

+	*delDML = *updateDML
+	delDML.op = del
+	// use oldValues of update as values of delete and reset oldValues
+	delDML.values = updateDML.oldValues


maybe in future we can remove the op of DML,

swicth DML:
only has new value: it's an INSERT
only has old value: it's a DELETE
else: it's a UPDATE

Maybe some problems like a table with all columns are generate columns, so new value and old value will both be null in insert statement.

maybe there should be at least one column, to let other columns generated from it.

I think this kind of implicit opType is no good for the readbility if our code? Are there any harm to reserve the explicit op field?

syncer/compactor.go

lance6716 · 2021-10-28T04:45:32Z

syncer/compactor.go

+		metrics.QueueSizeGauge.WithLabelValues(c.task, "compactor_input", c.source).Set(float64(len(c.inCh)))
+
+		if j.tp == flush {
+			c.flushBuffer()


this will block the consumption of c.inCh. if we has a lot of jobs to be flushed, we are now blocked to wait every job sent to next stage. This will increase the risk of upstream read timeout.

Maybe we can implement ping-pong buffer in future.

I think this is not the same problem. Upstream read timeout may also caused by the dmlJobCh is full. And after we support async flush checkpoint, we only need flush job when stop-task/pause-task?

when DM write 1 job per second to downstream and we have 100 jobs in compactor to flush, we now have to wait 100 seconds before reading upstream. Without compactor we only wait 1 second.

dm/syncer/compactor.go

Line 102 in a734fe0

if len(c.inCh) == 0 || len(c.buffer) >= c.bufferSize || len(c.outCh) == 0 {

Does len(c.in)==0 meets requirements?

Co-authored-by: lance6716 <[email protected]>

lance6716

will review tests later

syncer/compactor.go

lance6716 · 2021-10-28T07:03:51Z

syncer/syncer.go

+	chanSize := s.cfg.QueueSize * s.cfg.WorkerCount / 2
+	if s.cfg.Compact {
+		chanSize /= 2
+	}
+	s.dmlJobCh = make(chan *job, chanSize)


seems causality output channel didn't get adjusted

syncer/compactor.go

lance6716 · 2021-10-28T07:19:58Z

syncer/dml.go

+	}
+
+	for i := 0; i < len(values); i++ {
+		if values[i] != oldValues[i] {


we're comparing interface{}, so I guess they are not equal for most time?

Interface values are comparable. Two interface values are equal if they have identical dynamic types and equal dynamic values or if both have value nil.

https://stackoverflow.com/questions/34245932/checking-equality-of-interface

GMHDBJD · 2021-10-28T08:11:15Z

seems causality output channel didn't get adjusted

I think causality output channel no need to adjust, because it has no buffer and default queue size 1024 is enough.

lance6716

rest lgtm

tests/shardddl1/run.sh

glorv · 2021-10-28T12:32:50Z

syncer/compactor.go

+			if c.safeMode {
+				j.dml.safeMode = true
+			}


Suggested change

if c.safeMode {

j.dml.safeMode = true

}

j.dml.safeMode = c.safeMode

update in https://github.com/pingcap/ticdc/pull/3162

glorv · 2021-10-28T12:41:05Z

syncer/compactor.go

+	key := j.dml.identifyKey()
+	prevPos, ok := tableKeyMap[key]
+	// if no such key in the buffer, add it
+	if !ok || prevPos >= len(c.buffer) {


How can prevPos bigger than buffer length

just avoid panic. have removed it

syncer/compactor.go

glorv · 2021-10-28T12:48:08Z

syncer/dml.go

+	*delDML = *updateDML
+	delDML.op = del
+	// use oldValues of update as values of delete and reset oldValues
+	delDML.values = updateDML.oldValues


I think this kind of implicit opType is no good for the readbility if our code? Are there any harm to reserve the explicit op field?

glorv · 2021-10-28T12:58:52Z

syncer/dml_worker.go

@@ -59,10 +59,14 @@ type DMLWorker struct {

 // dmlWorkerWrap creates and runs a dmlWorker instance and returns flush job channel.
 func dmlWorkerWrap(inCh chan *job, syncer *Syncer) chan *job {
+	chanSize := syncer.cfg.QueueSize / 2


Please add a comment for this special logic? I'm curious if we can change the default value of this field to avoid this change. BTW, doesn't anyone know the effect of the chan size change? How can a user know he/she need to adjust its value and what is the proper value?

update comment in pingcap/ticdc#3162

I think no user will change the queue-size... so I keep the original value and didn't add a new config item like compact-buffer-size

GMHDBJD · 2021-10-28T13:06:04Z

Please comment in https://github.com/pingcap/ticdc/pull/3162

dianqihanwangzi · 2022-01-24T13:27:18Z

/run-unit-test

GMHDBJD and others added 30 commits August 29, 2021 09:39

refine sync dml

52e9f13

fix bug

96008ac

fix data race

b54a927

fix data race

53de6ce

fix test

22dad39

update

9606c90

update

57145f9

fix test

90f17bc

fix test

34d4989

fix ut

687d4e0

fix it

0ae0f33

save work

3163999

fix wait

6f1fe91

fix

4851cd4

Merge remote-tracking branch 'upstream/master' into refineSyncDML

b7cf22d

update causality and dml worker

30be49c

Merge remote-tracking branch 'upstream/master' into refineSyncDML

ff65707

refine

2ca637c

fix ut

9b5c1fe

Merge branch 'master' into refineSyncDML

735a087

fix causality

cb7a298

Merge branch 'master' into refineSyncDML

53ed0e5

refine queue size metrics

1401f4d

fix

5da87a1

Merge branch 'master' into refineSyncDML

f7392fb

debug ci

06dda5d

fix

f31328e

revert

f02faeb

add test back

7c44aec

update metrics

22c9374

GMHDBJD removed the status/WIP This PR is still work in progress label Oct 26, 2021

update channel size

4d7f6ab

Ehco1996 reviewed Oct 27, 2021

View reviewed changes

GMHDBJD and others added 2 commits October 27, 2021 10:38

Update syncer/compactor.go

0c89e36

Co-authored-by: Ehco <[email protected]>

Update syncer/compactor.go

9395b6a

Co-authored-by: Ehco <[email protected]>

GMHDBJD commented Oct 27, 2021

View reviewed changes

update safemode

716c9c2

Ehco1996 reviewed Oct 27, 2021

View reviewed changes

syncer/compactor_test.go Show resolved Hide resolved

syncer/compactor_test.go Show resolved Hide resolved

GMHDBJD added 2 commits October 27, 2021 01:11

test safmode

d51e49a

fix lint

b30f06a

lance6716 reviewed Oct 27, 2021

View reviewed changes

chaos/cases/conf/task-optimistic.yaml Show resolved Hide resolved

Merge branch 'master' into newCompactor

1bbc72d

Ehco1996 approved these changes Oct 28, 2021

View reviewed changes

ti-chi-bot added the status/LGT1 One reviewer already commented LGTM label Oct 28, 2021

lance6716 reviewed Oct 28, 2021

View reviewed changes

GMHDBJD and others added 2 commits October 28, 2021 13:51

Update dm/config/task.go

c5d5a22

Co-authored-by: lance6716 <[email protected]>

address comment

965b2ce

lance6716 reviewed Oct 28, 2021

View reviewed changes

tests/shardddl1/run.sh Outdated Show resolved Hide resolved

address comment

3d653a9

lance6716 approved these changes Oct 28, 2021

View reviewed changes

ti-chi-bot added status/LGT2 Two reviewers already commented LGTM, ready for merge and removed status/LGT1 One reviewer already commented LGTM labels Oct 28, 2021

Merge branch 'master' into newCompactor

a734fe0

glorv reviewed Oct 28, 2021

View reviewed changes

Ehco1996 mentioned this pull request Nov 11, 2021

add new added config part in openapi spec pingcap/tiflow#3397

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

.*: add compactor for syncer #2261

.*: add compactor for syncer #2261

GMHDBJD commented Oct 25, 2021 •

edited

Loading

GMHDBJD Oct 27, 2021

lance6716 Oct 28, 2021

Ehco1996 left a comment

GMHDBJD commented Oct 28, 2021

lance6716 left a comment

lance6716 Oct 28, 2021

GMHDBJD Oct 28, 2021

lance6716 Oct 28, 2021

glorv Oct 28, 2021

lance6716 Oct 28, 2021

GMHDBJD Oct 28, 2021

lance6716 Oct 28, 2021

GMHDBJD Oct 28, 2021 •

edited

Loading

lance6716 left a comment

lance6716 Oct 28, 2021

lance6716 Oct 28, 2021

GMHDBJD Oct 28, 2021

GMHDBJD commented Oct 28, 2021

lance6716 left a comment

glorv Oct 28, 2021

GMHDBJD Oct 28, 2021

glorv Oct 28, 2021

GMHDBJD Oct 28, 2021

glorv Oct 28, 2021

glorv Oct 28, 2021

GMHDBJD Oct 28, 2021

GMHDBJD Oct 28, 2021 •

edited

Loading

GMHDBJD commented Oct 28, 2021

dianqihanwangzi commented Jan 24, 2022

.*: add compactor for syncer #2261

Are you sure you want to change the base?

.*: add compactor for syncer #2261

Conversation

GMHDBJD commented Oct 25, 2021 • edited Loading

What problem does this PR solve?

What is changed and how it works?

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ehco1996 left a comment

Choose a reason for hiding this comment

GMHDBJD commented Oct 28, 2021

lance6716 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GMHDBJD Oct 28, 2021 • edited Loading

Choose a reason for hiding this comment

lance6716 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GMHDBJD commented Oct 28, 2021

lance6716 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GMHDBJD Oct 28, 2021 • edited Loading

Choose a reason for hiding this comment

GMHDBJD commented Oct 28, 2021

dianqihanwangzi commented Jan 24, 2022

GMHDBJD commented Oct 25, 2021 •

edited

Loading

GMHDBJD Oct 28, 2021 •

edited

Loading

GMHDBJD Oct 28, 2021 •

edited

Loading