sync: refactor dml.go

syncer/causality.go

@@ -66,14 +66,15 @@ func (c *causality) run() {
 		if j.tp == flush {
 			c.reset()
 		} else {
+			keys := j.dml.identifyKeys()
 			// detectConflict before add
-			if c.detectConflict(j.keys) {
-				c.logger.Debug("meet causality key, will generate a conflict job to flush all sqls", zap.Strings("keys", j.keys))
+			if c.detectConflict(keys) {
+				c.logger.Debug("meet causality key, will generate a conflict job to flush all sqls", zap.Strings("keys", keys))
 				c.outCh <- newConflictJob()
 				c.reset()
 			}
-			j.key = c.add(j.keys)
-			c.logger.Debug("key for keys", zap.String("key", j.key), zap.Strings("keys", j.keys))
+			j.dml.key = c.add(keys)
+			c.logger.Debug("key for keys", zap.String("key", j.dml.key), zap.Strings("keys", keys))
 		}
 		metrics.ConflictDetectDurationHistogram.WithLabelValues(c.task, c.source).Observe(time.Since(startTime).Seconds())
 

syncer/causality_test.go

@@ -71,28 +71,30 @@ func (s *testSyncerSuite) TestCasuality(c *C) {
 	}
 	causalityCh := causalityWrap(jobCh, syncer)
 	testCases := []struct {
-		op   opType
-		vals [][]interface{}
+		op      opType
+		oldVals []interface{}
+		vals    []interface{}
 	}{
 		{
 			op:   insert,
-			vals: [][]interface{}{{1, 2}},
+			vals: []interface{}{1, 2},
 		},
 		{
 			op:   insert,
-			vals: [][]interface{}{{2, 3}},
+			vals: []interface{}{2, 3},
 		},
 		{
-			op:   update,
-			vals: [][]interface{}{{2, 3}, {3, 4}},
+			op:      update,
+			oldVals: []interface{}{2, 3},
+			vals:    []interface{}{3, 4},
 		},
 		{
 			op:   del,
-			vals: [][]interface{}{{1, 2}},
+			vals: []interface{}{1, 2},
 		},
 		{
 			op:   insert,
-			vals: [][]interface{}{{1, 3}},
+			vals: []interface{}{1, 3},
 		},
 	}
 	results := []opType{insert, insert, update, del, conflict, insert}
@@ -101,11 +103,7 @@ func (s *testSyncerSuite) TestCasuality(c *C) {
 	ec := &eventContext{startLocation: &location, currentLocation: &location, lastLocation: &location}
 
 	for _, tc := range testCases {
-		var keys []string
-		for _, val := range tc.vals {
-			keys = append(keys, genMultipleKeys(ti, val, "tb")...)
-		}
-		job := newDMLJob(tc.op, table, table, "", nil, keys, ec)
+		job := newDMLJob(tc.op, table, table, newDML(tc.op, false, "", table, tc.oldVals, tc.vals, tc.oldVals, tc.vals, ti.Columns, ti), ec)
 		jobCh <- job
 	}