Repository: pingcap/tidb-tools
Pull request #492 (2021-08-25 05:46:45)

Diff2: add progress bar
<!--
Thank you for contributing to TiDB! Please read TiDB's [CONTRIBUTING](https://github.com/pingcap/tidb/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->


### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
 - No code

Code changes

 - Has exported function/method change
 - Has exported variable/fields change
 - Has interface methods change
 - Has persistent data change

Side effects

 - Possible performance regression
 - Increased code complexity
 - Breaking backward compatibility

Related changes

 - Need to cherry-pick to the release branch
 - Need to update the documentation
 - Need to be included in the release note

ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tidb-tools/pulls/492/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tidb-tools/pulls/492/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=pingcap%2Ftidb-tools).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: @Leavrth: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

Patch:
diff --git a/sync_diff_inspector/chunk/chunk.go b/sync_diff_inspector/chunk/chunk.go
index 65a7f8d..6f2acd9 100644
--- a/sync_diff_inspector/chunk/chunk.go
+++ b/sync_diff_inspector/chunk/chunk.go
@@ -257,20 +257,17 @@ func (c *Range) CopyAndUpdate(column, lower, upper string, updateLower, updateUp
 	return newChunk
 }
 
-func InitChunks(chunks []*Range, t ChunkType, chunkBeginID, bucketID int, collation, limits string) int {
+func InitChunks(chunks []*Range, t ChunkType, bucketID int, collation, limits string) {
 	if chunks == nil {
-		return chunkBeginID
+		return
 	}
 	for _, chunk := range chunks {
 		conditions, args := chunk.ToString(collation)
-		chunk.ID = chunkBeginID
-		chunkBeginID++
 		chunk.Where = fmt.Sprintf("((%s) AND %s)", conditions, limits)
 		chunk.Args = args
 		chunk.BucketID = bucketID
 		chunk.Type = t
 	}
-	return chunkBeginID
 }
 
 func InitChunk(chunk *Range, t ChunkType, bucketID int, collation, limits string) {
diff --git a/sync_diff_inspector/diff.go b/sync_diff_inspector/diff.go
index f6fbdea..ceeebc9 100644
--- a/sync_diff_inspector/diff.go
+++ b/sync_diff_inspector/diff.go
@@ -32,6 +32,7 @@ import (
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/checkpoints"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/config"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/progress"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/report"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/splitter"
@@ -133,6 +134,7 @@ func (df *Diff) init(ctx context.Context, cfg *config.Config) (err error) {
 		return errors.Trace(err)
 	}
 	df.report.Init(df.downstream.GetTables(), sourceConfigs, targetConfig)
+	progress.Init(len(df.workSource.GetTables()))
 	return nil
 }
 
@@ -223,7 +225,9 @@ func (df *Diff) Equal(ctx context.Context) error {
 			df.report.SetTableDataCheckResult(schema, table, isEqual)
 			if !isEqual {
 				df.report.SetTableDataCheckCount(schema, table, dml.rowAdd, dml.rowDelete)
+				progress.FailTable(c.ProgressID)
 			}
+			progress.Inc(c.ProgressID)
 		})
 	}
 
@@ -237,6 +241,7 @@ func (df *Diff) StructEqual(ctx context.Context) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+		progress.RegisterTable(dbutil.TableName(tables[tableIndex].Schema, tables[tableIndex].Table), !structEq, false)
 		df.report.SetTableStructCheckResult(tables[tableIndex].Schema, tables[tableIndex].Table, structEq)
 	}
 	return nil
@@ -736,5 +741,5 @@ func setTiDBCfg() {
 	tidbCfg.MaxIndexLength = 3027 * 4
 	tidbconfig.StoreGlobalConfig(tidbCfg)
 
-	fmt.Println("set tidb cfg")
+	log.Info("set tidb cfg")
 }
diff --git a/sync_diff_inspector/main.go b/sync_diff_inspector/main.go
index b8d1259..78ea815 100644
--- a/sync_diff_inspector/main.go
+++ b/sync_diff_inspector/main.go
@@ -18,6 +18,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"path/filepath"
 	"time"
 
 	_ "github.com/go-sql-driver/mysql"
@@ -25,6 +26,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb-tools/pkg/utils"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/config"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/progress"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/report"
 	"go.uber.org/zap"
 )
@@ -46,12 +48,16 @@ func main() {
 		return
 	}
 
-	l := zap.NewAtomicLevel()
-	if err := l.UnmarshalText([]byte(cfg.LogLevel)); err != nil {
-		log.Error("invalid log level", zap.String("log level", cfg.LogLevel))
+	conf := new(log.Config)
+	conf.Level = cfg.LogLevel
+
+	conf.File.Filename = filepath.Join(cfg.Task.OutputDir, "sync_diff.log")
+	lg, p, e := log.InitLogger(conf)
+	if e != nil {
+		log.Error("Log init failed!", zap.String("error", e.Error()))
 		return
 	}
-	log.SetLevel(l.Level())
+	log.ReplaceGlobals(lg, p)
 
 	utils.PrintInfo("sync_diff_inspector")
 
@@ -98,6 +104,8 @@ func checkSyncState(ctx context.Context, cfg *config.Config) bool {
 			log.Fatal("check data difference failed", zap.Error(err))
 		}
 	}
+	progress.Close()
+	//progress.PrintSummary()
 	if err := d.report.CalculateTotalSize(ctx, d.downstream.GetDB()); err != nil {
 		log.Warn("fail to calculate the total size", zap.Error(err))
 	}
diff --git a/sync_diff_inspector/progress/progress.go b/sync_diff_inspector/progress/progress.go
new file mode 100644
index 0000000..ff0cc08
--- /dev/null
+++ b/sync_diff_inspector/progress/progress.go
@@ -0,0 +1,447 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package progress
+
+import (
+	"container/list"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+)
+
+type TableProgressPrinter struct {
+	tableList      *list.List
+	stateIsChanged bool
+	tableFailList  *list.List
+	tableMap       map[string]*list.Element
+	output         io.Writer
+	lines          int
+
+	finishTableNums int
+	tableNums       int
+
+	progress int
+	total    int
+
+	optCh    chan Operator
+	finishCh chan struct{}
+}
+
+type table_state_t int
+
+const (
+	TABLE_STATE_REGISTER                       table_state_t = 0x1
+	TABLE_STATE_PRESTART                       table_state_t = 0x2
+	TABLE_STATE_COMPARING                      table_state_t = 0x4
+	TABLE_STATE_FINISH                         table_state_t = 0x8
+	TABLE_STATE_RESULT_OK                      table_state_t = 0x00
+	TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE     table_state_t = 0x10
+	TABLE_STATE_RESULT_FAIL_STRUCTURE_CONTINUE table_state_t = 0x20
+	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
+)
+
+type TableProgress struct {
+	name            string
+	progress        int
+	total           int
+	state           table_state_t
+	totalStopUpdate bool
+}
+
+type progress_opt_t int
+
+const (
+	PROGRESS_OPT_INC progress_opt_t = iota
+	PROGRESS_OPT_UPDATE
+	PROGRESS_OPT_REGISTER
+	PROGRESS_OPT_START
+	PROGRESS_OPT_FAIL
+	PROGRESS_OPT_CLOSE
+	PROGRESS_OPT_ERROR
+)
+
+type Operator struct {
+	optType         progress_opt_t
+	name            string
+	total           int
+	state           table_state_t
+	totalStopUpdate bool
+}
+
+func NewTableProgressPrinter(tableNums int) *TableProgressPrinter {
+	tpp := &TableProgressPrinter{
+		tableList:       list.New(),
+		stateIsChanged:  false,
+		tableFailList:   list.New(),
+		tableMap:        make(map[string]*list.Element),
+		lines:           0,
+		finishTableNums: 0,
+		tableNums:       tableNums,
+
+		progress: 0,
+		total:    0,
+
+		optCh:    make(chan Operator, 16),
+		finishCh: make(chan struct{}),
+	}
+	tpp.init()
+	go tpp.serve()
+	fmt.Fprintf(tpp.output, "A total of %d tables need to be compared\n\n\n", tableNums)
+	return tpp
+}
+
+func (tpp *TableProgressPrinter) SetOutput(output io.Writer) {
+	tpp.output = output
+}
+
+func (tpp *TableProgressPrinter) Inc(name string) {
+	tpp.optCh <- Operator{
+		optType: PROGRESS_OPT_INC,
+		name:    name,
+	}
+}
+
+func (tpp *TableProgressPrinter) UpdateTotal(name string, total int, stopUpdate bool) {
+	tpp.optCh <- Operator{
+		optType:         PROGRESS_OPT_UPDATE,
+		name:            name,
+		total:           total,
+		totalStopUpdate: stopUpdate,
+	}
+}
+
+func (tpp *TableProgressPrinter) RegisterTable(name string, isFailed bool, isDone bool) {
+	var state table_state_t
+	if isFailed {
+		if isDone {
+			state = TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE | TABLE_STATE_REGISTER
+		} else {
+			state = TABLE_STATE_RESULT_FAIL_STRUCTURE_CONTINUE | TABLE_STATE_REGISTER
+		}
+	} else {
+		state = TABLE_STATE_REGISTER
+	}
+	tpp.optCh <- Operator{
+		optType: PROGRESS_OPT_REGISTER,
+		name:    name,
+		state:   state,
+	}
+}
+
+func (tpp *TableProgressPrinter) StartTable(name string, total int, stopUpdate bool) {
+	tpp.optCh <- Operator{
+		optType:         PROGRESS_OPT_START,
+		name:            name,
+		total:           total,
+		state:           TABLE_STATE_PRESTART,
+		totalStopUpdate: stopUpdate,
+	}
+}
+
+func (tpp *TableProgressPrinter) FailTable(name string) {
+	tpp.optCh <- Operator{
+		optType: PROGRESS_OPT_FAIL,
+		name:    name,
+		state:   TABLE_STATE_RESULT_DIFFERENT,
+	}
+}
+
+func (tpp *TableProgressPrinter) Close() {
+	tpp.optCh <- Operator{
+		optType: PROGRESS_OPT_CLOSE,
+	}
+	<-tpp.finishCh
+}
+
+func (tpp *TableProgressPrinter) PrintSummary() {
+	var cleanStr, fixStr string
+	cleanStr = "\x1b[1A\x1b[J"
+	fixStr = "\nSummary:\n\n"
+	if tpp.tableFailList.Len() == 0 {
+		fixStr = fmt.Sprintf(
+			"%sA total of %d tables have been compared and all are equal.\nYou can view the comparison details through './output_dir/sync_diff_inspector.log'\n",
+			fixStr,
+			tpp.tableNums,
+		)
+	} else {
+		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 {
+				fixStr = fmt.Sprintf("%sThe structure of `%s` is not equal.\n", fixStr, tp.name)
+			}
+			if tp.state&(TABLE_STATE_RESULT_DIFFERENT) != 0 {
+				fixStr = fmt.Sprintf("%sThe data of `%s` is not equal.\n", fixStr, tp.name)
+			}
+		}
+		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,
+		)
+	}
+
+	fmt.Fprintf(tpp.output, "%s%s\n", cleanStr, fixStr)
+
+}
+
+func (tpp *TableProgressPrinter) Error(err error) {
+	tpp.optCh <- Operator{
+		optType: PROGRESS_OPT_ERROR,
+	}
+	<-tpp.finishCh
+	var cleanStr, fixStr string
+	cleanStr = "\x1b[1A\x1b[J"
+	fixStr = fmt.Sprintf("\nError in comparison process:\n%v\n\nYou can view the comparison details through './output_dir/sync_diff_inspector.log'\n", err)
+	fmt.Fprintf(tpp.output, "%s%s", cleanStr, fixStr)
+}
+
+func (tpp *TableProgressPrinter) init() {
+	tpp.tableList.PushBack(&TableProgress{
+		state: TABLE_STATE_HEAD,
+	})
+
+	tpp.output = os.Stdout
+}
+
+func (tpp *TableProgressPrinter) serve() {
+	tick := time.NewTicker(200 * time.Millisecond)
+
+	for {
+		select {
+		case <-tick.C:
+			tpp.flush()
+		case opt := <-tpp.optCh:
+			switch opt.optType {
+			case PROGRESS_OPT_CLOSE:
+				tpp.flush()
+				tpp.finishCh <- struct{}{}
+				return
+			case PROGRESS_OPT_ERROR:
+				tpp.finishCh <- struct{}{}
+				return
+			case PROGRESS_OPT_INC:
+				if e, ok := tpp.tableMap[opt.name]; ok {
+					tp := e.Value.(*TableProgress)
+					tp.progress++
+					tpp.progress++
+					if tp.progress >= tp.total && tp.totalStopUpdate {
+						tp.state = (tp.state & TABLE_STATE_RESULT_MASK) | TABLE_STATE_FINISH
+						tpp.progress -= tp.progress
+						tpp.total -= tp.total
+						tpp.stateIsChanged = true
+						delete(tpp.tableMap, opt.name)
+					}
+				}
+			case PROGRESS_OPT_REGISTER:
+				if _, ok := tpp.tableMap[opt.name]; !ok {
+					e := tpp.tableList.PushBack(&TableProgress{
+						name:            opt.name,
+						progress:        0,
+						total:           opt.total,
+						state:           opt.state,
+						totalStopUpdate: opt.totalStopUpdate,
+					})
+					tpp.tableMap[opt.name] = e
+				}
+			case PROGRESS_OPT_START:
+				e, ok := tpp.tableMap[opt.name]
+				if !ok {
+					e = tpp.tableList.PushBack(&TableProgress{
+						name:            opt.name,
+						progress:        0,
+						total:           opt.total,
+						state:           opt.state | TABLE_STATE_RESULT_FAIL_STRUCTURE_PASS,
+						totalStopUpdate: opt.totalStopUpdate,
+					})
+					tpp.tableMap[opt.name] = e
+				} else {
+					tp := e.Value.(*TableProgress)
+					tp.state ^= TABLE_STATE_REGISTER | opt.state
+					tp.progress = 0
+					tp.total = opt.total
+					tp.totalStopUpdate = opt.totalStopUpdate
+				}
+				if e.Value.(*TableProgress).state&TABLE_STATE_RESULT_FAIL_STRUCTURE_DONE == 0 {
+					tpp.total += opt.total
+				} else {
+					delete(tpp.tableMap, opt.name)
+				}
+				tpp.stateIsChanged = true
+				tpp.flush()
+			case PROGRESS_OPT_UPDATE:
+				if e, ok := tpp.tableMap[opt.name]; ok {
+					tp := e.Value.(*TableProgress)
+					tpp.total += opt.total
+					tp.total += opt.total
+					tp.totalStopUpdate = opt.totalStopUpdate
+				}
+			case PROGRESS_OPT_FAIL:
+				if e, ok := tpp.tableMap[opt.name]; ok {
+					tp := e.Value.(*TableProgress)
+					tp.state |= opt.state
+					//tpp.total -= tp.total
+					//tpp.progress -= tp.progress
+					//delete(tpp.tableMap, opt.name)
+				}
+			}
+		}
+	}
+}
+
+// flush flush info
+func (tpp *TableProgressPrinter) flush() {
+	/*
+	 * A total of 15 tables need to be compared
+	 *
+	 * Comparing the table structure of `schema1.table1` ... equivalent
+	 * Comparing the table data of `schema1.table1` ... equivalent
+	 * Comparing the table structure of `schema2.table2` ... equivalent
+	 * Comparing the table data of `schema2.table2` ...
+	 * _____________________________________________________________________________
+	 * Progress [===================>-----------------------------------------] 35%
+	 *
+	 */
+
+	if tpp.stateIsChanged {
+		var cleanStr, fixStr, dynStr string
+		cleanStr = fmt.Sprintf("\x1b[%dA\x1b[J", tpp.lines)
+		tpp.lines = 2
+		/* PRESTART/COMPARING/FINISH OK/DIFFERENT */
+		for p := tpp.tableList.Front(); p != nil; p = p.Next() {
+			tp := p.Value.(*TableProgress)
+			// There are 5 situations:
+			// 1. structure is same and data is same
+			// 2. structure is same and data is different
+			// 3. structure is different and we won't compare data
+			// 4. structure is different and data is same
+			// 5. structure is different and data is different
+			switch tp.state & 0xf {
+			case TABLE_STATE_PRESTART:
+				switch tp.state & TABLE_STATE_RESULT_MASK {
+				case TABLE_STATE_RESULT_OK:
+					fixStr = fmt.Sprintf("%sComparing the table structure of `%s` ... equivalent\n", fixStr, tp.name)
+					dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...\n", dynStr, tp.name)
+					tpp.lines++
+					tp.state = TABLE_STATE_COMPARING
+				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)
+					// we have empty node as list head, so p is not nil
+					preNode := p.Prev()
+					tpp.tableList.Remove(p)
+					p = preNode
+					tpp.finishTableNums++
+				case TABLE_STATE_RESULT_FAIL_STRUCTURE_CONTINUE:
+					fixStr = fmt.Sprintf("%sComparing the table structure of `%s` ... failure\n", fixStr, tp.name)
+					dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...\n", dynStr, tp.name)
+					tpp.lines++
+					tp.state ^= TABLE_STATE_COMPARING | TABLE_STATE_PRESTART
+				case TABLE_STATE_RESULT_FAIL_STRUCTURE_PASS:
+					fixStr = fmt.Sprintf("%sComparing the table structure of `%s` ... pass\n", fixStr, tp.name)
+					dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...\n", dynStr, tp.name)
+					tpp.lines++
+					tp.state = TABLE_STATE_PRESTART
+				}
+			case TABLE_STATE_COMPARING:
+				dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...\n", dynStr, tp.name)
+				tpp.lines++
+			case TABLE_STATE_FINISH:
+				if tp.state&TABLE_STATE_RESULT_DIFFERENT == 0 {
+					fixStr = fmt.Sprintf("%sComparing the table data of `%s` ... equivalent\n", fixStr, tp.name)
+				} else {
+					fixStr = fmt.Sprintf("%sComparing the table data of `%s` ... failure\n", fixStr, tp.name)
+				}
+				if tp.state&TABLE_STATE_RESULT_MASK != 0 {
+					tpp.tableFailList.PushBack(tp)
+				}
+				// we have empty node as list head, so p is not nil
+				preNode := p.Prev()
+				tpp.tableList.Remove(p)
+				p = preNode
+				tpp.finishTableNums++
+			}
+		}
+
+		dynStr = fmt.Sprintf("%s_____________________________________________________________________________\n", dynStr)
+		fmt.Fprintf(tpp.output, "%s%s%s", cleanStr, fixStr, dynStr)
+		tpp.stateIsChanged = false
+	} else {
+		fmt.Fprint(tpp.output, "\x1b[1A\x1b[J")
+	}
+	// show bar
+	// 60 '='+'-'
+	leftTableNums := 1000 * (tpp.tableNums - tpp.finishTableNums)
+	coe := float32(leftTableNums*tpp.progress)/float32(tpp.tableNums*(leftTableNums+1)*(tpp.total+1)) + float32(tpp.finishTableNums)/float32(tpp.tableNums)
+	numLeft := int(60 * coe)
+	percent := int(100 * coe)
+	fmt.Fprintf(tpp.output, "Progress [%s>%s] %d%% %d/%d\n", strings.Repeat("=", numLeft), strings.Repeat("-", 60-numLeft), percent, tpp.progress, tpp.total)
+}
+
+var progress_ *TableProgressPrinter = nil
+
+func Init(tableNums int) {
+	progress_ = NewTableProgressPrinter(tableNums)
+}
+
+func Inc(name string) {
+	if progress_ != nil {
+		progress_.Inc(name)
+	}
+}
+
+func UpdateTotal(name string, total int, stopUpdate bool) {
+	if progress_ != nil {
+		progress_.UpdateTotal(name, total, stopUpdate)
+	}
+}
+
+func RegisterTable(name string, isFailed bool, isDone bool) {
+	if progress_ != nil {
+		progress_.RegisterTable(name, isFailed, isDone)
+	}
+}
+
+func StartTable(name string, total int, stopUpdate bool) {
+	if progress_ != nil {
+		progress_.StartTable(name, total, stopUpdate)
+	}
+}
+
+func FailTable(name string) {
+	if progress_ != nil {
+		progress_.FailTable(name)
+	}
+}
+
+func Close() {
+	if progress_ != nil {
+		progress_.Close()
+	}
+}
+
+func PrintSummary() {
+	if progress_ != nil {
+		progress_.PrintSummary()
+	}
+}
+
+func Error(err error) {
+	if progress_ != nil {
+		progress_.Error(err)
+	}
+}
diff --git a/sync_diff_inspector/source/chunks_iter.go b/sync_diff_inspector/source/chunks_iter.go
index db8a78d..2575556 100644
--- a/sync_diff_inspector/source/chunks_iter.go
+++ b/sync_diff_inspector/source/chunks_iter.go
@@ -16,7 +16,9 @@ package source
 import (
 	"context"
 	"database/sql"
+
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/splitter"
 )
@@ -32,7 +34,8 @@ type ChunksIterator struct {
 	limit  int
 	dbConn *sql.DB
 
-	tableIter splitter.ChunkIterator
+	tableIter  splitter.ChunkIterator
+	progressID string
 }
 
 func (t *ChunksIterator) Next(ctx context.Context) (*splitter.RangeInfo, error) {
@@ -53,6 +56,7 @@ func (t *ChunksIterator) Next(ctx context.Context) (*splitter.RangeInfo, error)
 			ChunkRange: c,
 			TableIndex: curIndex,
 			IndexID:    t.getCurTableIndexID(),
+			ProgressID: t.progressID,
 		}, nil
 	}
 	err = t.nextTable(ctx, nil)
@@ -73,6 +77,7 @@ func (t *ChunksIterator) Next(ctx context.Context) (*splitter.RangeInfo, error)
 		ChunkRange: c,
 		TableIndex: curIndex,
 		IndexID:    t.getCurTableIndexID(),
+		ProgressID: t.progressID,
 	}, nil
 }
 
@@ -102,15 +107,17 @@ func (t *ChunksIterator) nextTable(ctx context.Context, startRange *splitter.Ran
 	}
 	curTable := t.TableDiffs[t.nextTableIndex]
 	t.nextTableIndex++
+	t.progressID = dbutil.TableName(curTable.Schema, curTable.Table)
 
 	// reads table index from checkpoint at the beginning
 	if startRange != nil {
 		curIndex := startRange.GetTableIndex()
 		curTable = t.TableDiffs[curIndex]
 		t.nextTableIndex = curIndex + 1
+		t.progressID = startRange.ProgressID
 	}
 
-	chunkIter, err := t.tableAnalyzer.AnalyzeSplitter(ctx, curTable, startRange)
+	chunkIter, err := t.tableAnalyzer.AnalyzeSplitter(ctx, t.progressID, curTable, startRange)
 	if err != nil {
 		return errors.Trace(err)
 	}
diff --git a/sync_diff_inspector/source/mysql_shard.go b/sync_diff_inspector/source/mysql_shard.go
index e3d7462..62f8e8e 100644
--- a/sync_diff_inspector/source/mysql_shard.go
+++ b/sync_diff_inspector/source/mysql_shard.go
@@ -36,7 +36,7 @@ type MySQLTableAnalyzer struct {
 	sourceTableMap map[string][]*common.TableShardSource
 }
 
-func (a *MySQLTableAnalyzer) AnalyzeSplitter(ctx context.Context, table *common.TableDiff, startRange *splitter.RangeInfo) (splitter.ChunkIterator, error) {
+func (a *MySQLTableAnalyzer) AnalyzeSplitter(ctx context.Context, progressID string, table *common.TableDiff, startRange *splitter.RangeInfo) (splitter.ChunkIterator, error) {
 	chunkSize := 1000
 	matchedSources := getMatchedSourcesForTable(a.sourceTableMap, table)
 
@@ -47,7 +47,7 @@ func (a *MySQLTableAnalyzer) AnalyzeSplitter(ctx context.Context, table *common.
 	table.Schema = matchedSources[0].OriginSchema
 	table.Table = matchedSources[0].OriginTable
 	// use random splitter if we cannot use bucket splitter, then we can simply choose target table to generate chunks.
-	randIter, err := splitter.NewRandomIteratorWithCheckpoint(ctx, table, matchedSources[0].DBConn, chunkSize, startRange)
+	randIter, err := splitter.NewRandomIteratorWithCheckpoint(ctx, progressID, table, matchedSources[0].DBConn, chunkSize, startRange)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/sync_diff_inspector/source/source.go b/sync_diff_inspector/source/source.go
index b5ea3f4..35c2b5c 100644
--- a/sync_diff_inspector/source/source.go
+++ b/sync_diff_inspector/source/source.go
@@ -60,7 +60,7 @@ type RowDataIterator interface {
 // each source has its own analyze function.
 type TableAnalyzer interface {
 	// AnalyzeSplitter picks the proper splitter.ChunkIterator according to table and source.
-	AnalyzeSplitter(context.Context, *common.TableDiff, *splitter.RangeInfo) (splitter.ChunkIterator, error)
+	AnalyzeSplitter(context.Context, string, *common.TableDiff, *splitter.RangeInfo) (splitter.ChunkIterator, error)
 }
 
 type Source interface {
diff --git a/sync_diff_inspector/source/tidb.go b/sync_diff_inspector/source/tidb.go
index 02903f1..6a845dd 100644
--- a/sync_diff_inspector/source/tidb.go
+++ b/sync_diff_inspector/source/tidb.go
@@ -36,7 +36,7 @@ type TiDBTableAnalyzer struct {
 	sourceTableMap map[string]*common.TableSource
 }
 
-func (a *TiDBTableAnalyzer) AnalyzeSplitter(ctx context.Context, table *common.TableDiff, startRange *splitter.RangeInfo) (splitter.ChunkIterator, error) {
+func (a *TiDBTableAnalyzer) AnalyzeSplitter(ctx context.Context, progressID string, table *common.TableDiff, startRange *splitter.RangeInfo) (splitter.ChunkIterator, error) {
 	chunkSize := 0
 	matchedSource := getMatchSource(a.sourceTableMap, table)
 	table.Schema = matchedSource.OriginSchema
@@ -44,7 +44,7 @@ func (a *TiDBTableAnalyzer) AnalyzeSplitter(ctx context.Context, table *common.T
 	// if we decide to use bucket to split chunks
 	// we always use bucksIter even we load from checkpoint is not bucketNode
 	// TODO check whether we can use bucket for this table to split chunks.
-	bucketIter, err := splitter.NewBucketIteratorWithCheckpoint(ctx, table, a.dbConn, chunkSize, startRange)
+	bucketIter, err := splitter.NewBucketIteratorWithCheckpoint(ctx, progressID, table, a.dbConn, chunkSize, startRange)
 	if err == nil {
 		return bucketIter, nil
 	}
@@ -52,7 +52,7 @@ func (a *TiDBTableAnalyzer) AnalyzeSplitter(ctx context.Context, table *common.T
 	// fall back to random splitter
 
 	// use random splitter if we cannot use bucket splitter, then we can simply choose target table to generate chunks.
-	randIter, err := splitter.NewRandomIteratorWithCheckpoint(ctx, table, a.dbConn, chunkSize, startRange)
+	randIter, err := splitter.NewRandomIteratorWithCheckpoint(ctx, progressID, table, a.dbConn, chunkSize, startRange)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/sync_diff_inspector/splitter/bucket.go b/sync_diff_inspector/splitter/bucket.go
index 82246b8..beb4ffc 100644
--- a/sync_diff_inspector/splitter/bucket.go
+++ b/sync_diff_inspector/splitter/bucket.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/progress"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/utils"
 	"go.uber.org/zap"
@@ -40,25 +41,31 @@ type BucketIterator struct {
 	errCh        chan error
 	indexID      int64
 
+	progressID string
+
 	dbConn *sql.DB
 }
 
-func NewBucketIterator(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*BucketIterator, error) {
-	return NewBucketIteratorWithCheckpoint(ctx, table, dbConn, chunkSize, nil)
+func NewBucketIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*BucketIterator, error) {
+	return NewBucketIteratorWithCheckpoint(ctx, progressID, table, dbConn, chunkSize, nil)
 }
 
-func NewBucketIteratorWithCheckpoint(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*BucketIterator, error) {
+func NewBucketIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*BucketIterator, error) {
 	bs := &BucketIterator{
 		table:     table,
 		chunkSize: int64(chunkSize),
 		chunksCh:  make(chan []*chunk.Range, DefaultChannelBuffer),
 		errCh:     make(chan error, 1),
 		dbConn:    dbConn,
+
+		progressID: progressID,
 	}
 
 	if err := bs.init(startRange); err != nil {
 		return nil, errors.Trace(err)
 	}
+
+	progress.StartTable(bs.progressID, 0, false)
 	go bs.produceChunks(ctx, startRange)
 
 	return bs, nil
@@ -145,7 +152,10 @@ func (s *BucketIterator) Close() {
 }
 
 func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInfo) {
-	defer close(s.chunksCh)
+	defer func() {
+		progress.UpdateTotal(s.progressID, 0, true)
+		close(s.chunksCh)
+	}()
 	var (
 		lowerValues, upperValues []string
 		latestCount              int64
@@ -156,7 +166,6 @@ func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInf
 	table := s.table
 	buckets := s.buckets
 	indexColumns := s.indexColumns
-	chunkID := 0
 	beginBucket := 0
 	if startRange != nil {
 		chunkRange := chunk.NewChunkRange()
@@ -173,7 +182,6 @@ func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInf
 		}
 
 		beginBucket = int(c.BucketID + 1)
-		chunkID = c.ID + 1
 		if c.BucketID < len(buckets) {
 			nextUpperValues, err := dbutil.AnalyzeValuesFromBuckets(buckets[c.BucketID].UpperBound, indexColumns)
 			if err != nil {
@@ -205,7 +213,8 @@ func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInf
 				s.errCh <- errors.Trace(err)
 				return
 			}
-			chunkID = chunk.InitChunks(chunks, chunk.Bucket, chunkID, c.BucketID, table.Collation, table.Range)
+			chunk.InitChunks(chunks, chunk.Bucket, c.BucketID, table.Collation, table.Range)
+			progress.UpdateTotal(s.progressID, len(chunks), false)
 			s.chunksCh <- chunks
 
 		}
@@ -251,7 +260,8 @@ func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInf
 		chunkRange = chunk.NewChunkRange()
 		latestCount = buckets[i].Count
 		lowerValues = upperValues
-		chunkID = chunk.InitChunks(chunks, chunk.Bucket, chunkID, i, table.Collation, table.Range)
+		chunk.InitChunks(chunks, chunk.Bucket, i, table.Collation, table.Range)
+		progress.UpdateTotal(s.progressID, len(chunks), false)
 		s.chunksCh <- chunks
 	}
 
@@ -261,7 +271,8 @@ func (s *BucketIterator) produceChunks(ctx context.Context, startRange *RangeInf
 			chunkRange.Update(column.Name.O, lowerValues[j], "", true, false)
 		}
 		chunks := []*chunk.Range{chunkRange}
-		chunkID = chunk.InitChunks(chunks, chunk.Bucket, chunkID, len(buckets), table.Collation, table.Range)
+		chunk.InitChunks(chunks, chunk.Bucket, len(buckets), table.Collation, table.Range)
+		progress.UpdateTotal(s.progressID, len(chunks), false)
 		s.chunksCh <- chunks
 	}
 }
diff --git a/sync_diff_inspector/splitter/limit.go b/sync_diff_inspector/splitter/limit.go
index e02b6f0..c5fc10b 100644
--- a/sync_diff_inspector/splitter/limit.go
+++ b/sync_diff_inspector/splitter/limit.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/progress"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/utils"
 	"go.uber.org/zap"
@@ -40,13 +41,15 @@ type LimitIterator struct {
 	errCh    chan error
 	cancel   context.CancelFunc
 	dbConn   *sql.DB
+
+	progressID string
 }
 
-func NewLimitIterator(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*LimitIterator, error) {
-	return NewLimitIteratorWithCheckpoint(ctx, table, dbConn, chunkSize, nil)
+func NewLimitIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*LimitIterator, error) {
+	return NewLimitIteratorWithCheckpoint(ctx, progressID, table, dbConn, chunkSize, nil)
 }
 
-func NewLimitIteratorWithCheckpoint(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*LimitIterator, error) {
+func NewLimitIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*LimitIterator, error) {
 	indices, err := utils.GetBetterIndex(ctx, dbConn, table.Schema, table.Table, table.Info)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -129,8 +132,11 @@ func NewLimitIteratorWithCheckpoint(ctx context.Context, table *common.TableDiff
 
 		cancel,
 		dbConn,
+
+		progressID,
 	}
 
+	progress.StartTable(progressID, 0, false)
 	go limitIterator.produceChunks(ctxx)
 
 	return limitIterator, nil
@@ -174,6 +180,7 @@ func (lmt *LimitIterator) produceChunks(ctx context.Context) {
 		if dataMap == nil {
 			// there is no row in result set
 			chunk.InitChunk(chunkRange, chunk.Limit, 0, lmt.table.Collation, lmt.table.Range)
+			progress.UpdateTotal(lmt.progressID, 1, true)
 			select {
 			case <-ctx.Done():
 			case lmt.chunksCh <- chunkRange:
@@ -189,6 +196,7 @@ func (lmt *LimitIterator) produceChunks(ctx context.Context) {
 		}
 
 		chunk.InitChunk(chunkRange, chunk.Limit, 0, lmt.table.Collation, lmt.table.Range)
+		progress.UpdateTotal(lmt.progressID, 1, false)
 		select {
 		case <-ctx.Done():
 			return
diff --git a/sync_diff_inspector/splitter/random.go b/sync_diff_inspector/splitter/random.go
index 11743cc..3117d8f 100644
--- a/sync_diff_inspector/splitter/random.go
+++ b/sync_diff_inspector/splitter/random.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/progress"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/utils"
 	"go.uber.org/zap"
@@ -38,11 +39,11 @@ type RandomIterator struct {
 	dbConn *sql.DB
 }
 
-func NewRandomIterator(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*RandomIterator, error) {
-	return NewRandomIteratorWithCheckpoint(ctx, table, dbConn, chunkSize, nil)
+func NewRandomIterator(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int) (*RandomIterator, error) {
+	return NewRandomIteratorWithCheckpoint(ctx, progressID, table, dbConn, chunkSize, nil)
 }
 
-func NewRandomIteratorWithCheckpoint(ctx context.Context, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*RandomIterator, error) {
+func NewRandomIteratorWithCheckpoint(ctx context.Context, progressID string, table *common.TableDiff, dbConn *sql.DB, chunkSize int, startRange *RangeInfo) (*RandomIterator, error) {
 	// get the chunk count by data count and chunk size
 	var splitFieldArr []string
 	if len(table.Fields) != 0 {
@@ -110,14 +111,9 @@ func NewRandomIteratorWithCheckpoint(ctx context.Context, table *common.TableDif
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	var chunk_id int
-	if startRange != nil {
-		chunk_id = startRange.GetChunk().ID + 1
-	} else {
-		chunk_id = 0
-	}
-	chunk.InitChunks(chunks, chunk.Random, chunk_id, 0, table.Collation, table.Range)
+	chunk.InitChunks(chunks, chunk.Random, 0, table.Collation, table.Range)
 
+	progress.StartTable(progressID, len(chunks), true)
 	return &RandomIterator{
 		table:     table,
 		chunkSize: chunkSize,
diff --git a/sync_diff_inspector/splitter/splitter.go b/sync_diff_inspector/splitter/splitter.go
index 36b4875..64436a8 100644
--- a/sync_diff_inspector/splitter/splitter.go
+++ b/sync_diff_inspector/splitter/splitter.go
@@ -39,6 +39,8 @@ type RangeInfo struct {
 	TableIndex int          `json:"table-index"`
 	// for bucket checkpoint
 	IndexID int64 `json:"index-id"`
+
+	ProgressID string `json:"progress-id"`
 }
 
 func (r *RangeInfo) GetChunk() *chunk.Range {
diff --git a/sync_diff_inspector/utils/utils.go b/sync_diff_inspector/utils/utils.go
index da6b773..e052f8d 100644
--- a/sync_diff_inspector/utils/utils.go
+++ b/sync_diff_inspector/utils/utils.go
@@ -103,8 +103,8 @@ func GetChar() (ascii int, keyCode int, err error) {
 // Apply executes a task.
 func (pool *WorkerPool) Apply(fn taskFunc) {
 	worker := pool.apply()
+	pool.wg.Add(1)
 	go func() {
-		pool.wg.Add(1)
 		defer pool.wg.Done()
 		defer pool.recycle(worker)
 		fn()
@@ -114,8 +114,8 @@ func (pool *WorkerPool) Apply(fn taskFunc) {
 // ApplyWithID execute a task and provides it with the worker ID.
 func (pool *WorkerPool) ApplyWithID(fn identifiedTaskFunc) {
 	worker := pool.apply()
+	pool.wg.Add(1)
 	go func() {
-		pool.wg.Add(1)
 		defer pool.wg.Done()
 		defer pool.recycle(worker)
 		fn(worker.ID)
@@ -125,8 +125,8 @@ func (pool *WorkerPool) ApplyWithID(fn identifiedTaskFunc) {
 // ApplyOnErrorGroup executes a task in an errgroup.
 func (pool *WorkerPool) ApplyOnErrorGroup(eg *errgroup.Group, fn func() error) {
 	worker := pool.apply()
+	pool.wg.Add(1)
 	eg.Go(func() error {
-		pool.wg.Add(1)
 		defer pool.wg.Done()
 		defer pool.recycle(worker)
 		return fn()
@@ -136,8 +136,8 @@ func (pool *WorkerPool) ApplyOnErrorGroup(eg *errgroup.Group, fn func() error) {
 // ApplyWithIDInErrorGroup executes a task in an errgroup and provides it with the worker ID.
 func (pool *WorkerPool) ApplyWithIDInErrorGroup(eg *errgroup.Group, fn func(id uint64) error) {
 	worker := pool.apply()
+	pool.wg.Add(1)
 	eg.Go(func() error {
-		pool.wg.Add(1)
 		defer pool.wg.Done()
 		defer pool.recycle(worker)
 		return fn(worker.ID)


Test patch:
diff --git a/sync_diff_inspector/chunk/chunk_test.go b/sync_diff_inspector/chunk/chunk_test.go
index 043184d..a1b5a04 100644
--- a/sync_diff_inspector/chunk/chunk_test.go
+++ b/sync_diff_inspector/chunk/chunk_test.go
@@ -169,8 +169,7 @@ func (*testChunkSuite) TestChunkInit(c *C) {
 		},
 	}
 
-	i := InitChunks(chunks, Others, 0, 1, "[123]", "[sdfds fsd fd gd]")
-	c.Assert(i, Equals, 2)
+	InitChunks(chunks, Others, 1, "[123]", "[sdfds fsd fd gd]")
 	c.Assert(chunks[0].ID, Equals, 0)
 	c.Assert(chunks[1].ID, Equals, 1)
 	c.Assert(chunks[0].Where, Equals, "((((`a` COLLATE '[123]' > ?) OR (`a` = ? AND `b` COLLATE '[123]' > ?) OR (`a` = ? AND `b` = ? AND `c` COLLATE '[123]' > ?)) AND ((`a` COLLATE '[123]' < ?) OR (`a` = ? AND `b` COLLATE '[123]' < ?) OR (`a` = ? AND `b` = ? AND `c` COLLATE '[123]' <= ?))) AND [sdfds fsd fd gd])")
diff --git a/sync_diff_inspector/progress/progress_test.go b/sync_diff_inspector/progress/progress_test.go
new file mode 100644
index 0000000..b816173
--- /dev/null
+++ b/sync_diff_inspector/progress/progress_test.go
@@ -0,0 +1,58 @@
+// Copyright 2021 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package progress
+
+import (
+	"bytes"
+	"testing"
+
+	. "github.com/pingcap/check"
+)
+
+func TestClient(t *testing.T) {
+	TestingT(t)
+}
+
+var _ = Suite(&testProgressSuite{})
+
+type testProgressSuite struct{}
+
+func (s *testProgressSuite) TestProgress(c *C) {
+	p := NewTableProgressPrinter(4)
+	p.RegisterTable("1", true, true)
+	p.StartTable("1", 50, true)
+	p.RegisterTable("2", true, false)
+	p.StartTable("2", 1, true)
+	p.Inc("2")
+	p.RegisterTable("3", false, false)
+	p.StartTable("3", 1, false)
+	p.Inc("3")
+	p.UpdateTotal("3", 1, true)
+	p.Inc("3")
+	p.RegisterTable("4", false, false)
+	p.StartTable("4", 1, true)
+	p.FailTable("4")
+	p.Inc("3")
+	p.Close()
+	buffer := new(bytes.Buffer)
+	p.SetOutput(buffer)
+	p.PrintSummary()
+	c.Assert(
+		buffer.String(),
+		DeepEquals,
+		"\x1b[1A\x1b[J\nSummary:\n\nThe structure of `1` is not equal.\nThe structure of `2` is not equal.\n"+
+			"\nThe rest of the tables are all equal.\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",
+	)
+}
diff --git a/sync_diff_inspector/source/source_test.go b/sync_diff_inspector/source/source_test.go
index 5f3a98f..866c029 100644
--- a/sync_diff_inspector/source/source_test.go
+++ b/sync_diff_inspector/source/source_test.go
@@ -26,7 +26,6 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser"
-	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
@@ -80,7 +79,7 @@ func (m *MockChunkIterator) Close() {
 type MockAnalyzer struct {
 }
 
-func (m *MockAnalyzer) AnalyzeSplitter(ctx context.Context, tableDiff *common.TableDiff, rangeInfo *splitter.RangeInfo) (splitter.ChunkIterator, error) {
+func (m *MockAnalyzer) AnalyzeSplitter(ctx context.Context, progressID string, tableDiff *common.TableDiff, rangeInfo *splitter.RangeInfo) (splitter.ChunkIterator, error) {
 	i := 0
 	return &MockChunkIterator{
 		ctx,
@@ -313,12 +312,6 @@ func prepareTiDBTables(c *C, tableCases []*tableCaseType) []*common.TableDiff {
 	for n, tableCase := range tableCases {
 		tableInfo, err := dbutil.GetTableInfoBySQL(tableCase.createTableSQL, parser.New())
 		c.Assert(err, IsNil)
-		orderKeyCols := make([]*model.ColumnInfo, 0, len(tableCase.rowColumns))
-		for _, column := range tableCase.rowColumns {
-			orderKeyCols = append(orderKeyCols, &model.ColumnInfo{
-				Name: model.CIStr{O: column},
-			})
-		}
 		tableDiffs = append(tableDiffs, &common.TableDiff{
 			Schema: "source_test",
 			Table:  fmt.Sprintf("test%d", n+1),
@@ -330,7 +323,7 @@ func prepareTiDBTables(c *C, tableCases []*tableCaseType) []*common.TableDiff {
 			chunkRange.Update(column, tableCase.rangeLeft[i], tableCase.rangeRight[i], true, true)
 		}
 
-		chunk.InitChunks([]*chunk.Range{chunkRange}, chunk.Others, 0, 0, "", "")
+		chunk.InitChunks([]*chunk.Range{chunkRange}, chunk.Others, 0, "", "")
 		rangeInfo := &splitter.RangeInfo{
 			ChunkRange: chunkRange,
 			TableIndex: n,
diff --git a/sync_diff_inspector/splitter/splitter_test.go b/sync_diff_inspector/splitter/splitter_test.go
index 362d3f3..016d1fb 100644
--- a/sync_diff_inspector/splitter/splitter_test.go
+++ b/sync_diff_inspector/splitter/splitter_test.go
@@ -223,7 +223,7 @@ func (s *testSplitterSuite) TestRandomSpliter(c *C) {
 
 		createFakeResultForRandomSplit(mock, testCase.count, testCase.randomValues)
 
-		iter, err := NewRandomIterator(ctx, tableDiff, db, 5)
+		iter, err := NewRandomIterator(ctx, "", tableDiff, db, 5)
 		c.Assert(err, IsNil)
 
 		j := 0
@@ -256,7 +256,7 @@ func (s *testSplitterSuite) TestRandomSpliter(c *C) {
 
 	createFakeResultForRandomSplit(mock, testCases[0].count, testCases[0].randomValues)
 
-	iter, err := NewRandomIterator(ctx, tableDiff, db, 5)
+	iter, err := NewRandomIterator(ctx, "", tableDiff, db, 5)
 	c.Assert(err, IsNil)
 
 	var chunk *chunk.Range
@@ -273,7 +273,7 @@ func (s *testSplitterSuite) TestRandomSpliter(c *C) {
 
 	createFakeResultForRandomSplit(mock, testCases[0].count, testCases[0].randomValues)
 
-	iter, err = NewRandomIteratorWithCheckpoint(ctx, tableDiff, db, 5, rangeInfo)
+	iter, err = NewRandomIteratorWithCheckpoint(ctx, "", tableDiff, db, 5, rangeInfo)
 	c.Assert(err, IsNil)
 
 	chunk, err = iter.Next()
@@ -296,8 +296,6 @@ func createFakeResultForRandomSplit(mock sqlmock.Sqlmock, count int, randomValue
 		}
 		mock.ExpectQuery("ORDER BY rand_value").WillReturnRows(randomRows)
 	}
-
-	return
 }
 
 func (s *testSplitterSuite) TestBucketSpliter(c *C) {
@@ -479,7 +477,7 @@ func (s *testSplitterSuite) TestBucketSpliter(c *C) {
 	for i, testCase := range testCases {
 		createFakeResultForBucketSplit(mock, testCase.aRandomValues, testCase.bRandomValues)
 
-		iter, err := NewBucketIterator(ctx, tableDiff, db, testCase.chunkSize)
+		iter, err := NewBucketIterator(ctx, "", tableDiff, db, testCase.chunkSize)
 		c.Assert(err, IsNil)
 
 		j := 0
@@ -500,7 +498,7 @@ func (s *testSplitterSuite) TestBucketSpliter(c *C) {
 	// Test Checkpoint
 	stopJ := 3
 	createFakeResultForBucketSplit(mock, testCases[0].aRandomValues[:stopJ], testCases[0].bRandomValues[:stopJ])
-	iter, err := NewBucketIterator(ctx, tableDiff, db, testCases[0].chunkSize)
+	iter, err := NewBucketIterator(ctx, "", tableDiff, db, testCases[0].chunkSize)
 	c.Assert(err, IsNil)
 	j := 0
 	var chunk *chunk.Range
@@ -519,7 +517,7 @@ func (s *testSplitterSuite) TestBucketSpliter(c *C) {
 	createFakeResultForBucketSplit(mock, nil, nil)
 	createFakeResultForCount(mock, 64)
 	createFakeResultForRandom(mock, testCases[0].aRandomValues[stopJ:], testCases[0].bRandomValues[stopJ:])
-	iter, err = NewBucketIteratorWithCheckpoint(ctx, tableDiff, db, testCases[0].chunkSize, rangeInfo)
+	iter, err = NewBucketIteratorWithCheckpoint(ctx, "", tableDiff, db, testCases[0].chunkSize, rangeInfo)
 	c.Assert(err, IsNil)
 	chunk, err = iter.Next()
 	c.Assert(err, IsNil)
@@ -549,8 +547,6 @@ func createFakeResultForBucketSplit(mock sqlmock.Sqlmock, aRandomValues, bRandom
 	mock.ExpectQuery("SHOW STATS_BUCKETS").WillReturnRows(statsRows)
 
 	createFakeResultForRandom(mock, aRandomValues, bRandomValues)
-
-	return
 }
 
 func createFakeResultForCount(mock sqlmock.Sqlmock, count int) {
@@ -623,7 +619,7 @@ func (s *testSplitterSuite) TestLimitSpliter(c *C) {
 	for i, testCase := range testCases {
 		createFakeResultForLimitSplit(mock, testCase.limitAValues, testCase.limitBValues, true)
 
-		iter, err := NewLimitIterator(ctx, tableDiff, db, testCase.chunkSize)
+		iter, err := NewLimitIterator(ctx, "", tableDiff, db, testCase.chunkSize)
 		c.Assert(err, IsNil)
 
 		j := 0
@@ -644,7 +640,7 @@ func (s *testSplitterSuite) TestLimitSpliter(c *C) {
 	// Test Checkpoint
 	stopJ := 2
 	createFakeResultForLimitSplit(mock, testCases[0].limitAValues[:stopJ], testCases[0].limitBValues[:stopJ], false)
-	iter, err := NewLimitIterator(ctx, tableDiff, db, testCases[0].chunkSize)
+	iter, err := NewLimitIterator(ctx, "", tableDiff, db, testCases[0].chunkSize)
 	c.Assert(err, IsNil)
 	j := 0
 	var chunk *chunk.Range
@@ -660,7 +656,7 @@ func (s *testSplitterSuite) TestLimitSpliter(c *C) {
 	}
 
 	createFakeResultForLimitSplit(mock, testCases[0].limitAValues[stopJ:], testCases[0].limitBValues[stopJ:], true)
-	iter, err = NewLimitIteratorWithCheckpoint(ctx, tableDiff, db, testCases[0].chunkSize, rangeInfo)
+	iter, err = NewLimitIteratorWithCheckpoint(ctx, "", tableDiff, db, testCases[0].chunkSize, rangeInfo)
 	c.Assert(err, IsNil)
 	chunk, err = iter.Next()
 	c.Assert(err, IsNil)