Repository: pingcap/tidb-tools
Pull request #512 (2021-09-17 06:55:02)

Diff2: progress bar support load checkpoint
<!--
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/512/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/512/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-->

Patch:
diff --git a/sync_diff_inspector/diff.go b/sync_diff_inspector/diff.go
index 77b97c0..d915dda 100644
--- a/sync_diff_inspector/diff.go
+++ b/sync_diff_inspector/diff.go
@@ -80,9 +80,10 @@ type Diff struct {
 	FixSQLDir     string
 	CheckpointDir string
 
-	sqlCh  chan *ChunkDML
-	cp     *checkpoints.Checkpoint
-	report *report.Report
+	sqlCh      chan *ChunkDML
+	cp         *checkpoints.Checkpoint
+	startRange *splitter.RangeInfo
+	report     *report.Report
 }
 
 // NewDiff returns a Diff instance.
@@ -149,13 +150,56 @@ func (df *Diff) init(ctx context.Context, cfg *config.Config) (err error) {
 	df.workSource = df.pickSource(ctx)
 	df.FixSQLDir = cfg.Task.FixDir
 	df.CheckpointDir = cfg.Task.CheckpointDir
-	df.cp.Init()
+	if err := df.initCheckpoint(); err != nil {
+		return errors.Trace(err)
+	}
 	sourceConfigs, targetConfig, err := getConfigsForReport(cfg)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	df.report.Init(df.downstream.GetTables(), sourceConfigs, targetConfig)
-	progress.Init(len(df.workSource.GetTables()))
+	return nil
+}
+
+func (df *Diff) initCheckpoint() error {
+	df.cp.Init()
+
+	finishTableNums := 0
+	if df.useCheckpoint {
+		path := filepath.Join(df.CheckpointDir, checkpointFile)
+		if ioutil2.FileExists(path) {
+			node, reportInfo, err := df.cp.LoadChunk(path)
+			if err != nil {
+				return errors.Annotate(err, "the checkpoint load process failed")
+			} else {
+				// this need not be synchronized, because at the moment, the is only one thread access the section
+				log.Info("load checkpoint",
+					zap.Any("chunk index", node.GetID()),
+					zap.Reflect("chunk", node),
+					zap.String("state", node.GetState()))
+				df.cp.SetCurrentSavedID(node)
+			}
+			if node != nil {
+				// remove the sql file that ID bigger than node.
+				// cause we will generate these sql again.
+				err = df.removeSQLFiles(node.GetID())
+				if err != nil {
+					return errors.Trace(err)
+				}
+				df.startRange = splitter.FromNode(node)
+				df.report.LoadReport(reportInfo)
+				finishTableNums = df.startRange.GetTableIndex()
+			}
+		} else {
+			log.Info("not found checkpoint file, start from beginning")
+			id := &chunk.ChunkID{TableIndex: 0, BucketIndexLeft: 0, BucketIndexRight: 0, ChunkIndex: 0, ChunkCnt: 0}
+			err := df.removeSQLFiles(id)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
+	}
+	progress.Init(len(df.workSource.GetTables()), finishTableNums)
 	return nil
 }
 
@@ -313,42 +357,7 @@ func (df *Diff) pickSource(ctx context.Context) source.Source {
 }
 
 func (df *Diff) generateChunksIterator(ctx context.Context) (source.RangeIterator, error) {
-	var startRange *splitter.RangeInfo
-	if df.useCheckpoint {
-		path := filepath.Join(df.CheckpointDir, checkpointFile)
-		if ioutil2.FileExists(path) {
-			node, reportInfo, err := df.cp.LoadChunk(path)
-			if err != nil {
-				return nil, errors.Annotate(err, "the checkpoint load process failed")
-			} else {
-				// this need not be synchronized, because at the moment, the is only one thread access the section
-				log.Info("load checkpoint",
-					zap.Any("chunk index", node.GetID()),
-					zap.Reflect("chunk", node),
-					zap.String("state", node.GetState()))
-				df.cp.SetCurrentSavedID(node)
-			}
-			if node != nil {
-				// remove the sql file that ID bigger than node.
-				// cause we will generate these sql again.
-				err = df.removeSQLFiles(node.GetID())
-				if err != nil {
-					return nil, errors.Trace(err)
-				}
-				startRange = splitter.FromNode(node)
-				df.report.LoadReport(reportInfo)
-			}
-		} else {
-			log.Info("not found checkpoint file, start from beginning")
-			id := &chunk.ChunkID{TableIndex: 0, BucketIndexLeft: 0, BucketIndexRight: 0, ChunkIndex: 0, ChunkCnt: 0}
-			err := df.removeSQLFiles(id)
-			if err != nil {
-				return nil, errors.Trace(err)
-			}
-		}
-	}
-
-	return df.workSource.GetRangeIterator(ctx, startRange, df.workSource.GetTableAnalyzer())
+	return df.workSource.GetRangeIterator(ctx, df.startRange, df.workSource.GetTableAnalyzer())
 }
 
 func (df *Diff) handleCheckpoints(ctx context.Context, stopCh chan struct{}) {
diff --git a/sync_diff_inspector/progress/progress.go b/sync_diff_inspector/progress/progress.go
index 44d406b..9f0ffb9 100644
--- a/sync_diff_inspector/progress/progress.go
+++ b/sync_diff_inspector/progress/progress.go
@@ -29,8 +29,9 @@ type TableProgressPrinter struct {
 	output        io.Writer
 	lines         int
 
-	finishTableNums int
-	tableNums       int
+	progressTableNums int
+	finishTableNums   int
+	tableNums         int
 
 	progress int
 	total    int
@@ -83,14 +84,16 @@ type Operator struct {
 	totalStopUpdate bool
 }
 
-func NewTableProgressPrinter(tableNums int) *TableProgressPrinter {
+func NewTableProgressPrinter(tableNums int, finishTableNums int) *TableProgressPrinter {
 	tpp := &TableProgressPrinter{
-		tableList:       list.New(),
-		tableFailList:   list.New(),
-		tableMap:        make(map[string]*list.Element),
-		lines:           0,
-		finishTableNums: 0,
-		tableNums:       tableNums,
+		tableList:     list.New(),
+		tableFailList: list.New(),
+		tableMap:      make(map[string]*list.Element),
+		lines:         0,
+
+		progressTableNums: 0,
+		finishTableNums:   finishTableNums,
+		tableNums:         tableNums,
 
 		progress: 0,
 		total:    0,
@@ -333,6 +336,7 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 					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++
+					tpp.progressTableNums++
 					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)
@@ -346,11 +350,13 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 					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++
+					tpp.progressTableNums++
 					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++
+					tpp.progressTableNums++
 					tp.state ^= TABLE_STATE_COMPARING | TABLE_STATE_PRESTART
 				}
 			case TABLE_STATE_COMPARING:
@@ -369,6 +375,7 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 				preNode := p.Prev()
 				tpp.tableList.Remove(p)
 				p = preNode
+				tpp.progressTableNums--
 				tpp.finishTableNums++
 			}
 		}
@@ -380,8 +387,7 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 	}
 	// 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)
+	coe := float32(tpp.progressTableNums*tpp.progress)/float32(tpp.tableNums*(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)
@@ -389,8 +395,8 @@ func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 
 var progress_ *TableProgressPrinter = nil
 
-func Init(tableNums int) {
-	progress_ = NewTableProgressPrinter(tableNums)
+func Init(tableNums, finishTableNums int) {
+	progress_ = NewTableProgressPrinter(tableNums, finishTableNums)
 }
 
 func Inc(name string) {


Test patch:
diff --git a/sync_diff_inspector/progress/progress_test.go b/sync_diff_inspector/progress/progress_test.go
index f044a23..4b1d28f 100644
--- a/sync_diff_inspector/progress/progress_test.go
+++ b/sync_diff_inspector/progress/progress_test.go
@@ -31,7 +31,7 @@ var _ = Suite(&testProgressSuite{})
 type testProgressSuite struct{}
 
 func (s *testProgressSuite) TestProgress(c *C) {
-	p := NewTableProgressPrinter(4)
+	p := NewTableProgressPrinter(4, 0)
 	p.RegisterTable("1", true, true)
 	p.StartTable("1", 50, true)
 	p.RegisterTable("2", true, false)
@@ -62,7 +62,7 @@ func (s *testProgressSuite) TestProgress(c *C) {
 }
 
 func (s *testProgressSuite) TestTableError(c *C) {
-	p := NewTableProgressPrinter(4)
+	p := NewTableProgressPrinter(4, 0)
 	p.RegisterTable("1", true, true)
 	p.StartTable("1", 50, true)
 	p.RegisterTable("2", true, true)
@@ -87,7 +87,7 @@ func (s *testProgressSuite) TestTableError(c *C) {
 }
 
 func (s *testProgressSuite) TestAllSuccess(c *C) {
-	Init(2)
+	Init(2, 0)
 	RegisterTable("1", false, false)
 	StartTable("1", 1, true)
 	RegisterTable("2", false, false)