sync_diff_inspector: add sum aggregate

sync_diff_inspector/config/config.go

@@ -52,6 +52,11 @@ const (
 	UnifiedTimeZone string = "+0:00"
 )
 
+const (
+	AggregateOpBITXOR = "BIT_XOR"
+	AggregateOpSUM    = "SUM"
+)
+
 // TableConfig is the config of table.
 type TableConfig struct {
 	// table's filter to tell us which table should adapt to this config.
@@ -363,6 +368,8 @@ type Config struct {
 	ExportFixSQL bool `toml:"export-fix-sql" json:"export-fix-sql"`
 	// only check table struct without table data.
 	CheckStructOnly bool `toml:"check-struct-only" json:"check-struct-only"`
+	// aggregate operator, default value is `BIT_XOR`
+	AggregateOp string `toml:"aggregate-op"`
 	// DMAddr is dm-master's address, the format should like "http://127.0.0.1:8261"
 	DMAddr string `toml:"dm-addr" json:"dm-addr"`
 	// DMTask string `toml:"dm-task" json:"dm-task"`
@@ -401,6 +408,8 @@ func NewConfig() *Config {
 	fs.BoolVar(&cfg.ExportFixSQL, "export-fix-sql", true, "set true if want to compare rows or set to false will only compare checksum")
 	fs.BoolVar(&cfg.CheckStructOnly, "check-struct-only", false, "ignore check table's data")
 
+	fs.StringVarP(&cfg.AggregateOp, "aggregate-op", "", "BIT_XOR", "aggregate operator, default value is `BIT_XOR`")
+	_ = fs.MarkHidden("aggregate-op")
 	fs.SortFlags = false
 	return cfg
 }
@@ -582,11 +591,23 @@ func (c *Config) Init() (err error) {
 	return nil
 }
 
+func checkAggregateOp(op string) bool {
+	switch op {
+	case AggregateOpBITXOR, AggregateOpSUM:
+		return true
+	}
+	return false
+}
+
 func (c *Config) CheckConfig() bool {
 	if c.CheckThreadCount <= 0 {
 		log.Error("check-thread-count must greater than 0!")
 		return false
 	}
+	if !checkAggregateOp(c.AggregateOp) {
+		log.Error("illegal aggregate operator. please select one from [BIT_XOR|SUM]")
+		return false
+	}
 	if len(c.DMAddr) != 0 {
 		u, err := url.Parse(c.DMAddr)
 		if err != nil || u.Scheme == "" || u.Host == "" {

sync_diff_inspector/diff.go

@@ -81,18 +81,21 @@ type Diff struct {
 	cp         *checkpoints.Checkpoint
 	startRange *splitter.RangeInfo
 	report     *report.Report
+
+	checksumAggregateOp string
 }
 
 // NewDiff returns a Diff instance.
 func NewDiff(ctx context.Context, cfg *config.Config) (diff *Diff, err error) {
 	diff = &Diff{
-		checkThreadCount: cfg.CheckThreadCount,
-		splitThreadCount: cfg.SplitThreadCount,
-		exportFixSQL:     cfg.ExportFixSQL,
-		ignoreDataCheck:  cfg.CheckStructOnly,
-		sqlCh:            make(chan *ChunkDML, splitter.DefaultChannelBuffer),
-		cp:               new(checkpoints.Checkpoint),
-		report:           report.NewReport(&cfg.Task),
+		checkThreadCount:    cfg.CheckThreadCount,
+		splitThreadCount:    cfg.SplitThreadCount,
+		exportFixSQL:        cfg.ExportFixSQL,
+		ignoreDataCheck:     cfg.CheckStructOnly,
+		sqlCh:               make(chan *ChunkDML, splitter.DefaultChannelBuffer),
+		cp:                  new(checkpoints.Checkpoint),
+		report:              report.NewReport(&cfg.Task),
+		checksumAggregateOp: cfg.AggregateOp,
 	}
 	if err = diff.init(ctx, cfg); err != nil {
 		diff.Close()
@@ -573,9 +576,9 @@ func (df *Diff) compareChecksumAndGetCount(ctx context.Context, tableRange *spli
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		upstreamInfo = df.upstream.GetCountAndCrc32(ctx, tableRange)
+		upstreamInfo = df.upstream.GetCountAndCrc32(ctx, tableRange, df.checksumAggregateOp)
 	}()
-	downstreamInfo = df.downstream.GetCountAndCrc32(ctx, tableRange)
+	downstreamInfo = df.downstream.GetCountAndCrc32(ctx, tableRange, df.checksumAggregateOp)
 	wg.Wait()
 
 	if upstreamInfo.Err != nil {

sync_diff_inspector/source/mysql_shard.go

@@ -93,7 +93,7 @@ func (s *MySQLSources) Close() {
 	}
 }
 
-func (s *MySQLSources) GetCountAndCrc32(ctx context.Context, tableRange *splitter.RangeInfo) *ChecksumInfo {
+func (s *MySQLSources) GetCountAndCrc32(ctx context.Context, tableRange *splitter.RangeInfo, op string) *ChecksumInfo {
 	beginTime := time.Now()
 	table := s.tableDiffs[tableRange.GetTableIndex()]
 	chunk := tableRange.GetChunk()
@@ -103,7 +103,7 @@ func (s *MySQLSources) GetCountAndCrc32(ctx context.Context, tableRange *splitte
 
 	for _, ms := range matchSources {
 		go func(ms *common.TableShardSource) {
-			count, checksum, err := utils.GetCountAndCRC32Checksum(ctx, ms.DBConn, ms.OriginSchema, ms.OriginTable, table.Info, chunk.Where, chunk.Args)
+			count, checksum, err := utils.GetCountAndCRC32Checksum(ctx, ms.DBConn, ms.OriginSchema, ms.OriginTable, table.Info, chunk.Where, chunk.Args, op)
 			infoCh <- &ChecksumInfo{
 				Checksum: checksum,
 				Count:    count,

sync_diff_inspector/source/source.go

@@ -82,7 +82,7 @@ type Source interface {
 	GetRangeIterator(context.Context, *splitter.RangeInfo, TableAnalyzer, int) (RangeIterator, error)
 
 	// GetCountAndCrc32 gets the crc32 result and the count from given range.
-	GetCountAndCrc32(context.Context, *splitter.RangeInfo) *ChecksumInfo
+	GetCountAndCrc32(context.Context, *splitter.RangeInfo, string) *ChecksumInfo
 
 	// GetRowsIterator gets the row data iterator from given range.
 	GetRowsIterator(context.Context, *splitter.RangeInfo) (RowDataIterator, error)

sync_diff_inspector/source/source_test.go

@@ -184,7 +184,7 @@ func TestTiDBSource(t *testing.T) {
 		require.Equal(t, n, tableCase.rangeInfo.GetTableIndex())
 		countRows := sqlmock.NewRows([]string{"CNT", "CHECKSUM"}).AddRow(123, 456)
 		mock.ExpectQuery("SELECT COUNT.*").WillReturnRows(countRows)
-		checksum := tidb.GetCountAndCrc32(ctx, tableCase.rangeInfo)
+		checksum := tidb.GetCountAndCrc32(ctx, tableCase.rangeInfo, config.AggregateOpBITXOR)
 		require.NoError(t, checksum.Err)
 		require.Equal(t, checksum.Count, int64(123))
 		require.Equal(t, checksum.Checksum, int64(456))
@@ -397,7 +397,7 @@ func TestMysqlShardSources(t *testing.T) {
 			mock.ExpectQuery("SELECT COUNT.*").WillReturnRows(countRows)
 		}
 
-		checksum := shard.GetCountAndCrc32(ctx, tableCase.rangeInfo)
+		checksum := shard.GetCountAndCrc32(ctx, tableCase.rangeInfo, config.AggregateOpBITXOR)
 		require.NoError(t, checksum.Err)
 		require.Equal(t, checksum.Count, int64(len(dbs)))
 		require.Equal(t, checksum.Checksum, resChecksum)

sync_diff_inspector/source/tidb.go

@@ -120,13 +120,13 @@ func (s *TiDBSource) GetRangeIterator(ctx context.Context, r *splitter.RangeInfo
 func (s *TiDBSource) Close() {
 	s.dbConn.Close()
 }
-func (s *TiDBSource) GetCountAndCrc32(ctx context.Context, tableRange *splitter.RangeInfo) *ChecksumInfo {
+func (s *TiDBSource) GetCountAndCrc32(ctx context.Context, tableRange *splitter.RangeInfo, op string) *ChecksumInfo {
 	beginTime := time.Now()
 	table := s.tableDiffs[tableRange.GetTableIndex()]
 	chunk := tableRange.GetChunk()
 
 	matchSource := getMatchSource(s.sourceTableMap, table)
-	count, checksum, err := utils.GetCountAndCRC32Checksum(ctx, s.dbConn, matchSource.OriginSchema, matchSource.OriginTable, table.Info, chunk.Where, chunk.Args)
+	count, checksum, err := utils.GetCountAndCRC32Checksum(ctx, s.dbConn, matchSource.OriginSchema, matchSource.OriginTable, table.Info, chunk.Where, chunk.Args, op)
 
 	cost := time.Since(beginTime)
 	return &ChecksumInfo{

sync_diff_inspector/utils/utils.go

@@ -744,7 +744,8 @@ func GetTableSize(ctx context.Context, db *sql.DB, schemaName, tableName string)
 }
 
 // GetCountAndCRC32Checksum returns checksum code and count of some data by given condition
-func GetCountAndCRC32Checksum(ctx context.Context, db *sql.DB, schemaName, tableName string, tbInfo *model.TableInfo, limitRange string, args []interface{}) (int64, int64, error) {
+// op can be [BIT_XOR|SUM]
+func GetCountAndCRC32Checksum(ctx context.Context, db *sql.DB, schemaName, tableName string, tbInfo *model.TableInfo, limitRange string, args []interface{}, op string) (int64, int64, error) {
 	/*
 		calculate CRC32 checksum and count example:
 		mysql> select count(*) as CNT, BIT_XOR(CAST(CRC32(CONCAT_WS(',', id, name, age, CONCAT(ISNULL(id), ISNULL(name), ISNULL(age))))AS UNSIGNED)) as CHECKSUM from test.test where id > 0;
@@ -770,8 +771,8 @@ func GetCountAndCRC32Checksum(ctx context.Context, db *sql.DB, schemaName, table
 		columnIsNull = append(columnIsNull, fmt.Sprintf("ISNULL(%s)", name))
 	}
 
-	query := fmt.Sprintf("SELECT COUNT(*) as CNT, BIT_XOR(CAST(CRC32(CONCAT_WS(',', %s, CONCAT(%s)))AS UNSIGNED)) as CHECKSUM FROM %s WHERE %s;",
-		strings.Join(columnNames, ", "), strings.Join(columnIsNull, ", "), dbutil.TableName(schemaName, tableName), limitRange)
+	query := fmt.Sprintf("SELECT COUNT(*) as CNT, %s(CAST(CRC32(CONCAT_WS(',', %s, CONCAT(%s)))AS UNSIGNED)) as CHECKSUM FROM %s WHERE %s;",
+		op, strings.Join(columnNames, ", "), strings.Join(columnIsNull, ", "), dbutil.TableName(schemaName, tableName), limitRange)
 	log.Debug("count and checksum", zap.String("sql", query), zap.Reflect("args", args))
 
 	var count sql.NullInt64

sync_diff_inspector/utils/utils_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/config"
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/stretchr/testify/require"
@@ -243,7 +244,7 @@ func TestGetCountAndCRC32Checksum(t *testing.T) {
 
 	mock.ExpectQuery("SELECT COUNT.*FROM `test_schema`\\.`test_table` WHERE \\[23 45\\].*").WithArgs("123", "234").WillReturnRows(sqlmock.NewRows([]string{"CNT", "CHECKSUM"}).AddRow(123, 456))
 
-	count, checksum, err := GetCountAndCRC32Checksum(ctx, conn, "test_schema", "test_table", tableInfo, "[23 45]", []interface{}{"123", "234"})
+	count, checksum, err := GetCountAndCRC32Checksum(ctx, conn, "test_schema", "test_table", tableInfo, "[23 45]", []interface{}{"123", "234"}, config.AggregateOpBITXOR)
 	require.NoError(t, err)
 	require.Equal(t, count, int64(123))
 	require.Equal(t, checksum, int64(456))