sync-diff-inspector: skip validation for tables that exist only upstream or downstream and print skipped information in summary and progress

sync_diff_inspector/config/config.go

@@ -374,6 +374,8 @@ type Config struct {
 	CheckStructOnly bool `toml:"check-struct-only" json:"check-struct-only"`
 	// experimental feature: only check table data without table struct
 	CheckDataOnly bool `toml:"check-data-only" json:"-"`
+	// skip validation for tables that don't exist upstream or downstream
+	SkipNonExistingTable bool `toml:"skip-non-existing-table" json:"-"`
 	// 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"`
@@ -411,6 +413,7 @@ func NewConfig() *Config {
 	fs.IntVar(&cfg.CheckThreadCount, "check-thread-count", 4, "how many goroutines are created to check data")
 	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.BoolVar(&cfg.SkipNonExistingTable, "skip-non-existing-table", false, "skip validation for tables that don't exist upstream or downstream")
 	fs.BoolVar(&cfg.CheckDataOnly, "check-data-only", false, "ignore check table's struct")
 
 	_ = fs.MarkHidden("check-data-only")

sync_diff_inspector/diff.go

@@ -297,12 +297,19 @@ func (df *Diff) StructEqual(ctx context.Context) error {
 		tableIndex = df.startRange.ChunkRange.Index.TableIndex
 	}
 	for ; tableIndex < len(tables); tableIndex++ {
-		isEqual, isSkip, err := df.compareStruct(ctx, tableIndex)
-		if err != nil {
-			return errors.Trace(err)
+		var isEqual, isSkip bool
+		isAllTableExist := tables[tableIndex].NeedSkippedTable
+		if source.AllTableExist(tables[tableIndex]) {
+			var err error
+			isEqual, isSkip, err = df.compareStruct(ctx, tableIndex)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		} else {
+			isEqual, isSkip = false, true
 		}
-		progress.RegisterTable(dbutil.TableName(tables[tableIndex].Schema, tables[tableIndex].Table), !isEqual, isSkip)
-		df.report.SetTableStructCheckResult(tables[tableIndex].Schema, tables[tableIndex].Table, isEqual, isSkip)
+		progress.RegisterTable(dbutil.TableName(tables[tableIndex].Schema, tables[tableIndex].Table), !isEqual, isSkip, isAllTableExist)
+		df.report.SetTableStructCheckResult(tables[tableIndex].Schema, tables[tableIndex].Table, isEqual, isSkip, isAllTableExist)
 	}
 	return nil
 }
@@ -411,12 +418,21 @@ func (df *Diff) consume(ctx context.Context, rangeInfo *splitter.RangeInfo) bool
 		node: rangeInfo.ToNode(),
 	}
 	defer func() { df.sqlCh <- dml }()
+	tableDiff := df.downstream.GetTables()[rangeInfo.GetTableIndex()]
+	schema, table := tableDiff.Schema, tableDiff.Table
+	id := rangeInfo.ChunkRange.Index
 	if rangeInfo.ChunkRange.Type == chunk.Empty {
 		dml.node.State = checkpoints.IgnoreState
+		// for tables that don't exist upstream or downstream
+		if !source.AllTableExist(tableDiff) {
+			upCount, _ := dbutil.GetRowCount(ctx, df.upstream.GetDB(), schema, table, "", nil)
+			downCount, _ := dbutil.GetRowCount(ctx, df.downstream.GetDB(), schema, table, "", nil)
+			df.report.SetTableDataCheckResult(schema, table, false, int(upCount), int(downCount), upCount, downCount, id)
+			return false
+		}
 		return true
 	}
-	tableDiff := df.downstream.GetTables()[rangeInfo.GetTableIndex()]
-	schema, table := tableDiff.Schema, tableDiff.Table
+
 	var state string = checkpoints.SuccessState
 
 	isEqual, upCount, downCount, err := df.compareChecksumAndGetCount(ctx, rangeInfo)
@@ -447,7 +463,6 @@ func (df *Diff) consume(ctx context.Context, rangeInfo *splitter.RangeInfo) bool
 		isEqual = isDataEqual
 	}
 	dml.node.State = state
-	id := rangeInfo.ChunkRange.Index
 	df.report.SetTableDataCheckResult(schema, table, isEqual, dml.rowAdd, dml.rowDelete, upCount, downCount, id)
 	return isEqual
 }

sync_diff_inspector/progress/progress.go

@@ -20,6 +20,8 @@ import (
 	"os"
 	"strings"
 	"time"
+
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 )
 
 type TableProgressPrinter struct {
@@ -53,7 +55,9 @@ const (
 	TABLE_STATE_RESULT_FAIL_STRUCTURE_PASS     table_state_t = 0x40
 	TABLE_STATE_RESULT_DIFFERENT               table_state_t = 0x80
 	TABLE_STATE_HEAD                           table_state_t = 0xff
-	TABLE_STATE_RESULT_MASK                    table_state_t = 0xf0
+	TABLE_STATE_RESULT_MASK                    table_state_t = 0xff0
+	TABLE_STATE_NOT_EXSIT_UPSTREAM             table_state_t = 0x100
+	TABLE_STATE_NOT_EXSIT_DOWNSTREAM           table_state_t = 0x200
 )
 
 type TableProgress struct {
@@ -127,11 +131,17 @@ func (tpp *TableProgressPrinter) UpdateTotal(name string, total int, stopUpdate
 	}
 }
 
-func (tpp *TableProgressPrinter) RegisterTable(name string, isFailed bool, isDone bool) {
+func (tpp *TableProgressPrinter) RegisterTable(name string, isFailed bool, isDone bool, isExist int) {
 	var state table_state_t
 	if isFailed {
 		if isDone {
-			state = TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE | TABLE_STATE_REGISTER
+			if isExist == common.UpstreamTableLackFlag {
+				state = TABLE_STATE_NOT_EXSIT_UPSTREAM | TABLE_STATE_REGISTER
+			} else if isExist == common.DownstreamTableLackFlag {
+				state = TABLE_STATE_NOT_EXSIT_DOWNSTREAM | TABLE_STATE_REGISTER
+			} else {
+				state = TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE | TABLE_STATE_REGISTER
+			}
 		} else {
 			state = TABLE_STATE_RESULT_FAIL_STRUCTURE_CONTINUE | TABLE_STATE_REGISTER
 		}
@@ -181,6 +191,7 @@ func (tpp *TableProgressPrinter) PrintSummary() {
 			tpp.tableNums,
 		)
 	} else {
+		SkippedNum := 0
 		for p := tpp.tableFailList.Front(); p != nil; p = p.Next() {
 			tp := p.Value.(*TableProgress)
 			if tp.state&(TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE|TABLE_STATE_RESULT_FAIL_STRUCTURE_CONTINUE) != 0 {
@@ -189,10 +200,18 @@ func (tpp *TableProgressPrinter) PrintSummary() {
 			if tp.state&(TABLE_STATE_RESULT_DIFFERENT) != 0 {
 				fixStr = fmt.Sprintf("%sThe data of `%s` is not equal.\n", fixStr, tp.name)
 			}
+			if tp.state&(TABLE_STATE_NOT_EXSIT_DOWNSTREAM) != 0 {
+				fixStr = fmt.Sprintf("%sThe data of `%s` does not exist in downstream database.\n", fixStr, tp.name)
+				SkippedNum++
+			}
+			if tp.state&(TABLE_STATE_NOT_EXSIT_UPSTREAM) != 0 {
+				fixStr = fmt.Sprintf("%sThe data of `%s` does not exist in upstream database.\n", fixStr, tp.name)
+				SkippedNum++
+			}
 		}
 		fixStr = fmt.Sprintf(
-			"%s\nThe rest of the tables are all equal.\nThe patch file has been generated to './output_dir/patch.sql'\nYou can view the comparison details through './output_dir/sync_diff_inspector.log'\n",
-			fixStr,
+			"%s\nThe rest of the tables are all equal.\nA total of %d tables have been compared, %d tables finished, %d tables failed, %d tables skipped.\nThe patch file has been generated to './output_dir/patch.sql'\nYou can view the comparison details through './output_dir/sync_diff_inspector.log'\n",
+			fixStr, tpp.tableNums, tpp.tableNums-tpp.tableFailList.Len(), tpp.tableFailList.Len()-SkippedNum, SkippedNum,
 		)
 	}
 
@@ -337,6 +356,13 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 					tpp.lines++
 					tpp.progressTableNums++
 					tp.state = TABLE_STATE_COMPARING
+				case TABLE_STATE_NOT_EXSIT_UPSTREAM, TABLE_STATE_NOT_EXSIT_DOWNSTREAM:
+					dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...skipped\n", dynStr, tp.name)
+					tpp.tableFailList.PushBack(tp)
+					preNode := p.Prev()
+					tpp.tableList.Remove(p)
+					p = preNode
+					tpp.finishTableNums++
 				case TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE:
 					fixStr = fmt.Sprintf("%sComparing the table structure of `%s` ... failure\n", fixStr, tp.name)
 					tpp.tableFailList.PushBack(tp)
@@ -410,9 +436,9 @@ func UpdateTotal(name string, total int, stopUpdate bool) {
 	}
 }
 
-func RegisterTable(name string, isFailed bool, isDone bool) {
+func RegisterTable(name string, isFailed bool, isDone bool, isExist int) {
 	if progress_ != nil {
-		progress_.RegisterTable(name, isFailed, isDone)
+		progress_.RegisterTable(name, isFailed, isDone, isExist)
 	}
 }
 

sync_diff_inspector/progress/progress_test.go

@@ -24,12 +24,12 @@ import (
 
 func TestProgress(t *testing.T) {
 	p := NewTableProgressPrinter(4, 0)
-	p.RegisterTable("1", true, true)
+	p.RegisterTable("1", true, true, 1)
 	p.StartTable("1", 50, true)
-	p.RegisterTable("2", true, false)
+	p.RegisterTable("2", true, false, 0)
 	p.StartTable("2", 2, true)
 	p.Inc("2")
-	p.RegisterTable("3", false, false)
+	p.RegisterTable("3", false, false, 0)
 	p.StartTable("3", 1, false)
 	p.Inc("2")
 	p.Inc("3")
@@ -47,18 +47,21 @@ func TestProgress(t *testing.T) {
 	require.Equal(
 		t,
 		buffer.String(),
-		"\x1b[1A\x1b[J\nSummary:\n\nThe structure of `1` is not equal.\nThe structure of `2` is not equal.\nThe data of `4` is not equal.\n"+
-			"\nThe rest of the tables are all equal.\nThe patch file has been generated to './output_dir/patch.sql'\n"+
+		"\x1b[1A\x1b[J\nSummary:\n\nThe data of `1` does not exist in upstream database.\nThe structure of `2` is not equal.\nThe data of `4` is not equal.\n"+
+			"\nThe rest of the tables are all equal.\nA total of 4 tables have been compared, 1 tables finished, 2 tables failed, 1 tables skipped.\nThe patch file has been generated to './output_dir/patch.sql'\n"+
 			"You can view the comparison details through './output_dir/sync_diff_inspector.log'\n\n",
 	)
 }
 
 func TestTableError(t *testing.T) {
 	p := NewTableProgressPrinter(4, 0)
-	p.RegisterTable("1", true, true)
+	p.RegisterTable("1", true, true, 0)
 	p.StartTable("1", 50, true)
-	p.RegisterTable("2", true, true)
+	p.RegisterTable("2", true, true, 0)
 	p.StartTable("2", 1, true)
+	p.RegisterTable("3", true, true, -1)
+	p.StartTable("3", 1, true)
+
 	p.Inc("2")
 	buffer := new(bytes.Buffer)
 	p.SetOutput(buffer)
@@ -73,16 +76,19 @@ func TestTableError(t *testing.T) {
 			"\x1b[2A\x1b[JComparing the table structure of `2` ... failure\n"+
 			"_____________________________________________________________________________\n"+
 			"Progress [==============================>------------------------------] 50% 0/0\n"+
+			"\x1b[2A\x1b[JComparing the table data of `3` ...skipped\n"+
+			"_____________________________________________________________________________\n"+
+			"Progress [=============================================>---------------] 75% 0/1\n"+
 			"\x1b[1A\x1b[J\nError in comparison process:\n[aaa]\n\n"+
 			"You can view the comparison details through './output_dir/sync_diff_inspector.log'\n",
 	)
 }
 
 func TestAllSuccess(t *testing.T) {
 	Init(2, 0)
-	RegisterTable("1", false, false)
+	RegisterTable("1", false, false, 0)
 	StartTable("1", 1, true)
-	RegisterTable("2", false, false)
+	RegisterTable("2", false, false, 0)
 	StartTable("2", 1, true)
 	Inc("1")
 	Inc("2")