Repository: pingcap/tidb-tools
Pull request #502 (2021-09-07 07:53:26)

Diff2: add unit test coverage
<!--
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/502/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/502/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/progress/progress.go b/sync_diff_inspector/progress/progress.go
index ff0cc08..44d406b 100644
--- a/sync_diff_inspector/progress/progress.go
+++ b/sync_diff_inspector/progress/progress.go
@@ -23,12 +23,11 @@ import (
 )
 
 type TableProgressPrinter struct {
-	tableList      *list.List
-	stateIsChanged bool
-	tableFailList  *list.List
-	tableMap       map[string]*list.Element
-	output         io.Writer
-	lines          int
+	tableList     *list.List
+	tableFailList *list.List
+	tableMap      map[string]*list.Element
+	output        io.Writer
+	lines         int
 
 	finishTableNums int
 	tableNums       int
@@ -87,7 +86,6 @@ type Operator struct {
 func NewTableProgressPrinter(tableNums int) *TableProgressPrinter {
 	tpp := &TableProgressPrinter{
 		tableList:       list.New(),
-		stateIsChanged:  false,
 		tableFailList:   list.New(),
 		tableMap:        make(map[string]*list.Element),
 		lines:           0,
@@ -224,11 +222,11 @@ func (tpp *TableProgressPrinter) serve() {
 	for {
 		select {
 		case <-tick.C:
-			tpp.flush()
+			tpp.flush(false)
 		case opt := <-tpp.optCh:
 			switch opt.optType {
 			case PROGRESS_OPT_CLOSE:
-				tpp.flush()
+				tpp.flush(false)
 				tpp.finishCh <- struct{}{}
 				return
 			case PROGRESS_OPT_ERROR:
@@ -243,8 +241,8 @@ func (tpp *TableProgressPrinter) serve() {
 						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)
+						tpp.flush(true)
 					}
 				}
 			case PROGRESS_OPT_REGISTER:
@@ -281,8 +279,7 @@ func (tpp *TableProgressPrinter) serve() {
 				} else {
 					delete(tpp.tableMap, opt.name)
 				}
-				tpp.stateIsChanged = true
-				tpp.flush()
+				tpp.flush(true)
 			case PROGRESS_OPT_UPDATE:
 				if e, ok := tpp.tableMap[opt.name]; ok {
 					tp := e.Value.(*TableProgress)
@@ -293,10 +290,9 @@ func (tpp *TableProgressPrinter) serve() {
 			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)
+					tp.state |= opt.state | TABLE_STATE_FINISH
+					tpp.flush(true)
+					// continue to increment chunk
 				}
 			}
 		}
@@ -304,7 +300,7 @@ func (tpp *TableProgressPrinter) serve() {
 }
 
 // flush flush info
-func (tpp *TableProgressPrinter) flush() {
+func (tpp *TableProgressPrinter) flush(stateIsChanged bool) {
 	/*
 	 * A total of 15 tables need to be compared
 	 *
@@ -317,7 +313,7 @@ func (tpp *TableProgressPrinter) flush() {
 	 *
 	 */
 
-	if tpp.stateIsChanged {
+	if stateIsChanged {
 		var cleanStr, fixStr, dynStr string
 		cleanStr = fmt.Sprintf("\x1b[%dA\x1b[J", tpp.lines)
 		tpp.lines = 2
@@ -355,7 +351,7 @@ func (tpp *TableProgressPrinter) flush() {
 					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
+					tp.state ^= TABLE_STATE_COMPARING | TABLE_STATE_PRESTART
 				}
 			case TABLE_STATE_COMPARING:
 				dynStr = fmt.Sprintf("%sComparing the table data of `%s` ...\n", dynStr, tp.name)
@@ -379,7 +375,6 @@ func (tpp *TableProgressPrinter) flush() {
 
 		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")
 	}
@@ -445,3 +440,9 @@ func Error(err error) {
 		progress_.Error(err)
 	}
 }
+
+func SetOutput(output io.Writer) {
+	if progress_ != nil {
+		progress_.SetOutput(output)
+	}
+}
diff --git a/sync_diff_inspector/report/report.go b/sync_diff_inspector/report/report.go
index 7b06928..c86717d 100644
--- a/sync_diff_inspector/report/report.go
+++ b/sync_diff_inspector/report/report.go
@@ -331,6 +331,10 @@ func (r *Report) SetTableMeetError(schema, table string, err error) {
 	defer r.Unlock()
 	if _, ok := r.TableResults[schema]; !ok {
 		r.TableResults[schema] = make(map[string]*TableResult)
+		r.TableResults[schema][table] = &TableResult{
+			MeetError: err,
+		}
+		return
 	}
 
 	r.TableResults[schema][table].MeetError = err


Test patch:
diff --git a/sync_diff_inspector/checkpoints/checkpoints_test.go b/sync_diff_inspector/checkpoints/checkpoints_test.go
index 2f04a46..6c435b6 100644
--- a/sync_diff_inspector/checkpoints/checkpoints_test.go
+++ b/sync_diff_inspector/checkpoints/checkpoints_test.go
@@ -46,21 +46,21 @@ func (cp *testCheckpointSuit) TestSaveChunk(c *C) {
 	rounds := 100
 	for i := 1; i < rounds; i++ {
 		wg.Add(1)
-		go func(i_ int) {
+		go func(i int) {
 			node := &Node{
 				ChunkRange: &chunk.Range{
-					ID: i_,
+					ID: i,
 					Bounds: []*chunk.Bound{
 						{
-							HasLower: i_ != 1,
-							Lower:    strconv.Itoa(i_ + 1000),
-							Upper:    strconv.Itoa(i_ + 1000 + 1),
-							HasUpper: i_ != rounds,
+							HasLower: i != 1,
+							Lower:    strconv.Itoa(i + 1000),
+							Upper:    strconv.Itoa(i + 1000 + 1),
+							HasUpper: i != rounds,
 						},
 					},
 				},
 
-				BucketID: i_,
+				BucketID: i,
 				State:    SuccessState,
 			}
 			if rand.Intn(4) == 0 {
diff --git a/sync_diff_inspector/config/config_test.go b/sync_diff_inspector/config/config_test.go
index 7860184..3f86787 100644
--- a/sync_diff_inspector/config/config_test.go
+++ b/sync_diff_inspector/config/config_test.go
@@ -66,4 +66,28 @@ func (s *testConfigSuite) TestError(c *C) {
 	c.Assert(tableConfig.Valid(), IsFalse)
 	tableConfig.Table = "234"
 	c.Assert(tableConfig.Valid(), IsTrue)
+
+	cfg := NewConfig()
+	// Parse
+	c.Assert(cfg.Parse([]string{"-config", "no_exist.toml"}), ErrorMatches, ".*no_exist.toml: no such file or directory.*")
+
+	// CheckConfig
+	cfg.Sample = 101
+	c.Assert(cfg.CheckConfig(), IsFalse)
+	cfg.Sample = -1
+	c.Assert(cfg.CheckConfig(), IsFalse)
+	cfg.Sample = 20
+	c.Assert(cfg.CheckConfig(), IsTrue)
+	cfg.CheckThreadCount = 0
+	c.Assert(cfg.CheckConfig(), IsFalse)
+	cfg.CheckThreadCount = 1
+	c.Assert(cfg.CheckConfig(), IsTrue)
+
+	// Init
+	cfg.DataSources = make(map[string]*DataSource)
+	cfg.DataSources["123"] = &DataSource{
+		RouteRules: []string{"111"},
+	}
+	err := cfg.Init()
+	c.Assert(err, ErrorMatches, "not found source routes for rule 111, please correct the config")
 }
diff --git a/sync_diff_inspector/progress/progress_test.go b/sync_diff_inspector/progress/progress_test.go
index b816173..f044a23 100644
--- a/sync_diff_inspector/progress/progress_test.go
+++ b/sync_diff_inspector/progress/progress_test.go
@@ -15,7 +15,9 @@ package progress
 
 import (
 	"bytes"
+	"errors"
 	"testing"
+	"time"
 
 	. "github.com/pingcap/check"
 )
@@ -33,17 +35,19 @@ func (s *testProgressSuite) TestProgress(c *C) {
 	p.RegisterTable("1", true, true)
 	p.StartTable("1", 50, true)
 	p.RegisterTable("2", true, false)
-	p.StartTable("2", 1, true)
+	p.StartTable("2", 2, true)
 	p.Inc("2")
 	p.RegisterTable("3", false, false)
 	p.StartTable("3", 1, false)
+	p.Inc("2")
 	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.Inc("4")
+	time.Sleep(500 * time.Millisecond)
 	p.Close()
 	buffer := new(bytes.Buffer)
 	p.SetOutput(buffer)
@@ -51,8 +55,51 @@ func (s *testProgressSuite) TestProgress(c *C) {
 	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"+
+		"\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"+
 			"You can view the comparison details through './output_dir/sync_diff_inspector.log'\n\n",
 	)
 }
+
+func (s *testProgressSuite) TestTableError(c *C) {
+	p := NewTableProgressPrinter(4)
+	p.RegisterTable("1", true, true)
+	p.StartTable("1", 50, true)
+	p.RegisterTable("2", true, true)
+	p.StartTable("2", 1, true)
+	p.Inc("2")
+	buffer := new(bytes.Buffer)
+	p.SetOutput(buffer)
+	p.Error(errors.New("[aaa]"))
+	time.Sleep(500 * time.Millisecond)
+	c.Assert(
+		buffer.String(),
+		DeepEquals,
+		"\x1b[0A\x1b[JComparing the table structure of `1` ... failure\n"+
+			"_____________________________________________________________________________\n"+
+			"Progress [===============>---------------------------------------------] 25% 0/0\n"+
+			"\x1b[2A\x1b[JComparing the table structure of `2` ... failure\n"+
+			"_____________________________________________________________________________\n"+
+			"Progress [==============================>------------------------------] 50% 0/0\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 (s *testProgressSuite) TestAllSuccess(c *C) {
+	Init(2)
+	RegisterTable("1", false, false)
+	StartTable("1", 1, true)
+	RegisterTable("2", false, false)
+	StartTable("2", 1, true)
+	Inc("1")
+	Inc("2")
+	Close()
+	buf := new(bytes.Buffer)
+	SetOutput(buf)
+	PrintSummary()
+	c.Assert(buf.String(), Equals, "\x1b[1A\x1b[J\nSummary:\n\n"+
+		"A total of 2 tables have been compared and all are equal.\n"+
+		"You can view the comparison details through './output_dir/sync_diff_inspector.log'\n\n",
+	)
+}
diff --git a/sync_diff_inspector/report/report_test.go b/sync_diff_inspector/report/report_test.go
index c0e0964..4b5cba8 100644
--- a/sync_diff_inspector/report/report_test.go
+++ b/sync_diff_inspector/report/report_test.go
@@ -17,6 +17,9 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"os"
+	"path"
+	"strings"
 	"testing"
 
 	"github.com/BurntSushi/toml"
@@ -24,6 +27,7 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/tidb-tools/pkg/dbutil"
+	"github.com/pingcap/tidb-tools/sync_diff_inspector/config"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 )
 
@@ -90,10 +94,12 @@ func (s *testReportSuite) TestReport(c *C) {
 	}
 	report.Init(tableDiffs, configsBytes[:2], configsBytes[2])
 
+	// Test CalculateTotal
 	mock.ExpectQuery("select sum.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("123"))
 	mock.ExpectQuery("select sum.*where table_schema=.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("456"))
 	report.CalculateTotalSize(ctx, db, 2, 4)
 
+	// Test Table Report
 	report.SetTableStructCheckResult("test", "tbl", true)
 	report.SetTableDataCheckResult("test", "tbl", true, 100, 200, 100)
 	report.SetTableMeetError("test", "tbl", errors.New("eeee"))
@@ -113,9 +119,13 @@ func (s *testReportSuite) TestReport(c *C) {
 	c.Assert(new_report.getSortedTables(), DeepEquals, []string{"`atest`.`atbl`", "`test`.`tbl`"})
 	c.Assert(new_report.getDiffRows(), DeepEquals, [][]string{})
 
-	new_report.SetTableStructCheckResult("atest", "atbl", false)
+	new_report.SetTableStructCheckResult("atest", "atbl", true)
 	new_report.SetTableDataCheckResult("atest", "atbl", false, 111, 222, 100)
 	c.Assert(new_report.getSortedTables(), DeepEquals, []string{"`test`.`tbl`"})
+	c.Assert(new_report.getDiffRows(), DeepEquals, [][]string{{"`atest`.`atbl`", "true", "+111/-222"}})
+
+	new_report.SetTableStructCheckResult("atest", "atbl", false)
+	c.Assert(new_report.getSortedTables(), DeepEquals, []string{"`test`.`tbl`"})
 	c.Assert(new_report.getDiffRows(), DeepEquals, [][]string{{"`atest`.`atbl`", "false", "+111/-222"}})
 
 	buf := new(bytes.Buffer)
@@ -127,3 +137,352 @@ func (s *testReportSuite) TestReport(c *C) {
 		"The patch file has been generated to './output_dir/patch.sql'\n"+
 		"You can view the comparision details through './output_dir/[123]'\n")
 }
+
+func (s *testReportSuite) TestCalculateTotal(c *C) {
+	ctx := context.Background()
+
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+
+	report := NewReport()
+	createTableSQL := "create table `test`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo, err := dbutil.GetTableInfoBySQL(createTableSQL, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiffs := []*common.TableDiff{
+		{
+			Schema:    "test",
+			Table:     "tbl",
+			Info:      tableInfo,
+			Collation: "[123]",
+		},
+	}
+	configs := []*ReportConfig{
+		{
+			Host: "127.0.0.1",
+			Port: 3306,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 3307,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 4000,
+			User: "root",
+		},
+	}
+
+	configsBytes := make([][]byte, 3)
+	for i := 0; i < 3; i++ {
+		buf := new(bytes.Buffer)
+		err := toml.NewEncoder(buf).Encode(configs[i])
+		c.Assert(err, IsNil)
+		configsBytes[i] = buf.Bytes()
+	}
+	report.Init(tableDiffs, configsBytes[:2], configsBytes[2])
+
+	// Normal
+	mock.ExpectQuery("select sum.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("123"))
+	report.CalculateTotalSize(ctx, db, 2, 4)
+	c.Assert(report.TotalSize, Equals, int64(123))
+
+	// Fail (and ...
+	// fail to analyze
+	mock.ExpectQuery("select sum.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("0"))
+	mock.ExpectQuery("select sum.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("1"))
+	report.CalculateTotalSize(ctx, db, 2, 4)
+	c.Assert(strings.Contains(report.TableResults["test"]["tbl"].MeetError.Error(), "ANALYZE TABLE"), IsTrue)
+	c.Assert(report.TotalSize, Equals, int64(124))
+
+	// fail to getSize
+	mock.ExpectQuery("select sum.*").WillReturnRows(sqlmock.NewRows([]string{"data"}).AddRow("0"))
+	mock.ExpectQuery("ANALYZE TABLE `test`.`tbl`.*")
+	report.CalculateTotalSize(ctx, db, 2, 4)
+	c.Assert(report.TableResults["test"]["tbl"].MeetError, ErrorMatches, "*select sum.*")
+	c.Assert(report.TotalSize, Equals, int64(124))
+}
+
+func (s *testReportSuite) TestPrint(c *C) {
+	report := NewReport()
+	createTableSQL := "create table `test`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo, err := dbutil.GetTableInfoBySQL(createTableSQL, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiffs := []*common.TableDiff{
+		{
+			Schema:    "test",
+			Table:     "tbl",
+			Info:      tableInfo,
+			Collation: "[123]",
+		},
+	}
+	configs := []*ReportConfig{
+		{
+			Host: "127.0.0.1",
+			Port: 3306,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 3307,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 4000,
+			User: "root",
+		},
+	}
+
+	configsBytes := make([][]byte, 3)
+	for i := 0; i < 3; i++ {
+		buf := new(bytes.Buffer)
+		err := toml.NewEncoder(buf).Encode(configs[i])
+		c.Assert(err, IsNil)
+		configsBytes[i] = buf.Bytes()
+	}
+	report.Init(tableDiffs, configsBytes[:2], configsBytes[2])
+
+	var buf *bytes.Buffer
+	// All Pass
+	report.SetTableStructCheckResult("test", "tbl", true)
+	report.SetTableDataCheckResult("test", "tbl", true, 0, 0, 0)
+	buf = new(bytes.Buffer)
+	report.Print("[123]", buf)
+	c.Assert(buf.String(), Equals, "A total of 0 table have been compared and all are equal.\n"+
+		"You can view the comparision details through './output_dir/[123]'\n")
+
+	// Error
+	report.SetTableMeetError("test", "tbl", errors.New("123"))
+	report.SetTableStructCheckResult("test", "tbl", false)
+	buf = new(bytes.Buffer)
+	report.Print("[123]", buf)
+	c.Assert(buf.String(), Equals, "Error in comparison process:\n"+
+		"123 error occured in `test`.`tbl`\n"+
+		"You can view the comparision details through './output_dir/[123]'\n")
+}
+
+func (s *testReportSuite) TestGetSnapshot(c *C) {
+	report := NewReport()
+	createTableSQL1 := "create table `test`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo1, err := dbutil.GetTableInfoBySQL(createTableSQL1, parser.New())
+	c.Assert(err, IsNil)
+	createTableSQL2 := "create table `atest`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo2, err := dbutil.GetTableInfoBySQL(createTableSQL2, parser.New())
+	c.Assert(err, IsNil)
+	createTableSQL3 := "create table `xtest`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo3, err := dbutil.GetTableInfoBySQL(createTableSQL3, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiffs := []*common.TableDiff{
+		{
+			Schema:    "test",
+			Table:     "tbl",
+			Info:      tableInfo1,
+			Collation: "[123]",
+		}, {
+			Schema:    "atest",
+			Table:     "tbl",
+			Info:      tableInfo2,
+			Collation: "[123]",
+		}, {
+			Schema:    "xtest",
+			Table:     "tbl",
+			Info:      tableInfo3,
+			Collation: "[123]",
+		},
+	}
+	configs := []*ReportConfig{
+		{
+			Host: "127.0.0.1",
+			Port: 3306,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 3307,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 4000,
+			User: "root",
+		},
+	}
+
+	configsBytes := make([][]byte, 3)
+	for i := 0; i < 3; i++ {
+		buf := new(bytes.Buffer)
+		err := toml.NewEncoder(buf).Encode(configs[i])
+		c.Assert(err, IsNil)
+		configsBytes[i] = buf.Bytes()
+	}
+	report.Init(tableDiffs, configsBytes[:2], configsBytes[2])
+
+	report.SetTableStructCheckResult("test", "tbl", true)
+	report.SetTableDataCheckResult("test", "tbl", false, 100, 100, 0)
+	report.SetTableDataCheckResult("test", "tbl", true, 0, 0, 1)
+	report.SetTableDataCheckResult("test", "tbl", false, 200, 200, 3)
+
+	report.SetTableStructCheckResult("atest", "tbl", true)
+	report.SetTableDataCheckResult("atest", "tbl", false, 100, 100, 0)
+	report.SetTableDataCheckResult("atest", "tbl", true, 0, 0, 1)
+	report.SetTableDataCheckResult("atest", "tbl", false, 200, 200, 3)
+
+	report.SetTableStructCheckResult("xtest", "tbl", true)
+	report.SetTableDataCheckResult("xtest", "tbl", false, 100, 100, 0)
+	report.SetTableDataCheckResult("xtest", "tbl", true, 0, 0, 1)
+	report.SetTableDataCheckResult("xtest", "tbl", false, 200, 200, 3)
+
+	report_snap, err := report.GetSnapshot(1, "test", "tbl")
+	c.Assert(err, IsNil)
+	c.Assert(report_snap.TotalSize, Equals, report.TotalSize)
+	c.Assert(report_snap.Result, Equals, report.Result)
+	for key, value := range report.TableResults {
+		if _, ok := report_snap.TableResults[key]; !ok {
+			v, ok := value["tbl"]
+			c.Assert(ok, IsTrue)
+			c.Assert(v.Schema, Equals, "atest")
+			continue
+		}
+
+		if _, ok := report_snap.TableResults[key]["tbl"]; !ok {
+			c.Assert(key, Equals, "atest")
+			continue
+		}
+
+		v1 := value["tbl"]
+		v2 := report_snap.TableResults[key]["tbl"]
+		c.Assert(v1.Schema, Equals, v2.Schema)
+		c.Assert(v1.Table, Equals, v2.Table)
+		c.Assert(v1.StructEqual, Equals, v2.StructEqual)
+		c.Assert(v1.DataEqual, Equals, v2.DataEqual)
+		c.Assert(v1.MeetError, Equals, v2.MeetError)
+
+		chunkMap1 := v1.ChunkMap
+		chunkMap2 := v2.ChunkMap
+		for id, r1 := range chunkMap1 {
+			if _, ok := chunkMap2[id]; !ok {
+				c.Assert(id, Equals, 3)
+				continue
+			}
+			c.Assert(id, LessEqual, 1)
+			r2 := chunkMap2[id]
+			c.Assert(r1.RowsAdd, Equals, r2.RowsAdd)
+			c.Assert(r1.RowsCnt, Equals, r2.RowsCnt)
+			c.Assert(r1.RowsDelete, Equals, r2.RowsDelete)
+		}
+
+	}
+}
+
+func (s *testReportSuite) TestCommitSummary(c *C) {
+	report := NewReport()
+	createTableSQL1 := "create table `test`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo1, err := dbutil.GetTableInfoBySQL(createTableSQL1, parser.New())
+	c.Assert(err, IsNil)
+	createTableSQL2 := "create table `atest`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo2, err := dbutil.GetTableInfoBySQL(createTableSQL2, parser.New())
+	c.Assert(err, IsNil)
+	createTableSQL3 := "create table `xtest`.`tbl`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo3, err := dbutil.GetTableInfoBySQL(createTableSQL3, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiffs := []*common.TableDiff{
+		{
+			Schema:    "test",
+			Table:     "tbl",
+			Info:      tableInfo1,
+			Collation: "[123]",
+		}, {
+			Schema:    "atest",
+			Table:     "tbl",
+			Info:      tableInfo2,
+			Collation: "[123]",
+		}, {
+			Schema:    "xtest",
+			Table:     "tbl",
+			Info:      tableInfo3,
+			Collation: "[123]",
+		},
+	}
+	configs := []*ReportConfig{
+		{
+			Host: "127.0.0.1",
+			Port: 3306,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 3307,
+			User: "root",
+		},
+		{
+			Host: "127.0.0.1",
+			Port: 4000,
+			User: "root",
+		},
+	}
+
+	configsBytes := make([][]byte, 3)
+	for i := 0; i < 3; i++ {
+		buf := new(bytes.Buffer)
+		err := toml.NewEncoder(buf).Encode(configs[i])
+		c.Assert(err, IsNil)
+		configsBytes[i] = buf.Bytes()
+	}
+	report.Init(tableDiffs, configsBytes[:2], configsBytes[2])
+
+	report.SetTableStructCheckResult("test", "tbl", true)
+	report.SetTableDataCheckResult("test", "tbl", true, 100, 200, 1)
+	report.SetRowsCnt("test", "tbl", 10000, 1)
+
+	report.SetTableStructCheckResult("atest", "tbl", true)
+	report.SetTableDataCheckResult("atest", "tbl", false, 100, 200, 2)
+	report.SetRowsCnt("atest", "tbl", 10000, 2)
+
+	report.SetTableStructCheckResult("xtest", "tbl", false)
+	report.SetTableDataCheckResult("xtest", "tbl", false, 100, 200, 3)
+	report.SetRowsCnt("xtest", "tbl", 10000, 3)
+
+	outputDir := "./"
+	err = report.CommitSummary(&config.TaskConfig{OutputDir: outputDir})
+	c.Assert(err, IsNil)
+	filename := path.Join(outputDir, "summary.txt")
+	file, err := os.Open(filename)
+	c.Assert(err, IsNil)
+
+	p := make([]byte, 1024)
+	file.Read(p)
+
+	c.Assert(string(p), Matches, "Summary\n\n\n\n"+
+		"Source Database\n\n\n\n"+
+		"host = \"127.0.0.1\"\n"+
+		"port = 3306\n"+
+		"user = \"root\"\n\n"+
+		"host = \"127.0.0.1\"\n"+
+		"port = 3307\n"+
+		"user = \"root\"\n\n"+
+		"Target Databases\n\n\n\n"+
+		"host = \"127.0.0.1\"\n"+
+		"port = 4000\n"+
+		"user = \"root\"\n\n"+
+		"Comparison Result\n\n\n\n"+
+		"The table structure and data in following tables are equivalent\n\n"+
+		"`test`.`tbl`\n\n"+
+		"The following tables contains inconsistent data\n\n"+
+		"+---------------+--------------------+----------------+\n"+
+		"|     TABLE     | STRUCTURE EQUALITY | DATA DIFF ROWS |\n"+
+		"+---------------+--------------------+----------------+\n"+
+		"| `atest`.`tbl` | true               | +100/-200      |\n"+
+		"| `xtest`.`tbl` | false              | +100/-200      |\n"+
+		"+---------------+--------------------+----------------+\n"+
+		"Time Cost:.*")
+
+	file.Close()
+	err = os.Remove(filename)
+	c.Assert(err, IsNil)
+}
diff --git a/sync_diff_inspector/source/source_test.go b/sync_diff_inspector/source/source_test.go
index dbaab59..fefa948 100644
--- a/sync_diff_inspector/source/source_test.go
+++ b/sync_diff_inspector/source/source_test.go
@@ -18,6 +18,8 @@ import (
 	"database/sql"
 	"database/sql/driver"
 	"fmt"
+	"os"
+	"strconv"
 	"testing"
 	"time"
 
@@ -32,6 +34,8 @@ import (
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/chunk"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/source/common"
 	"github.com/pingcap/tidb-tools/sync_diff_inspector/splitter"
+
+	_ "github.com/go-sql-driver/mysql"
 )
 
 func TestClient(t *testing.T) {
@@ -523,15 +527,23 @@ func prepareTiDBTables(c *C, tableCases []*tableCaseType) []*common.TableDiff {
 }
 
 func (s *testSourceSuite) TestSource(c *C) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
+	host, isExist := os.LookupEnv("MYSQL_HOST")
+	if host == "" || !isExist {
+		return
+	}
+	portstr, isExist := os.LookupEnv("MYSQL_PORT")
+	if portstr == "" || !isExist {
+		return
+	}
 
-	conn, _, err := sqlmock.New()
+	port, err := strconv.Atoi(portstr)
 	c.Assert(err, IsNil)
-	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 
 	cfg := &config.Config{
-		LogLevel:          "info",
+		LogLevel:          "debug",
 		Sample:            100,
 		CheckThreadCount:  4,
 		UseChecksum:       false,
@@ -541,20 +553,24 @@ func (s *testSourceSuite) TestSource(c *C) {
 		UseCheckpoint:     true,
 		DataSources: map[string]*config.DataSource{
 			"mysql1": {
-				Conn: conn,
+				Host: host,
+				Port: port,
+				User: "root",
 			},
 			"tidb": {
-				Conn: conn,
+				Host: host,
+				Port: port,
+				User: "root",
 			},
 		},
 		Routes: nil,
 		TableConfigs: map[string]*config.TableConfig{
 			"config1": {
 				Schema:          "schama1",
-				Table:           "table",
+				Table:           "tbl",
 				IgnoreColumns:   []string{"", ""},
 				Fields:          "",
-				Range:           "age > 10 AND age < 20",
+				Range:           "a > 10 AND a < 20",
 				IsSharding:      false,
 				TargetTableInfo: nil,
 				Collation:       "",
@@ -564,40 +580,28 @@ func (s *testSourceSuite) TestSource(c *C) {
 			Source:       []string{"mysql1"},
 			Routes:       nil,
 			Target:       []string{"tidb"},
-			CheckTables:  []string{"schema*.table*", "!c.*", "test2.t2"},
+			CheckTables:  []string{"schema*.tbl"},
 			TableConfigs: []string{"config1"},
 			OutputDir:    "./output",
 			SourceInstances: []*config.DataSource{
 				{
-					Host:       "127.0.0.1",
-					Port:       4567,
-					User:       "root",
-					Password:   "",
-					SqlMode:    "",
-					Snapshot:   "",
-					RouteRules: nil,
-					Router:     nil,
-					Conn:       nil,
+					Host: host,
+					Port: port,
+					User: "root",
 				},
 			},
 			TargetInstance: &config.DataSource{
-				Host:       "127.0.0.1",
-				Port:       4567,
-				User:       "root",
-				Password:   "",
-				SqlMode:    "",
-				Snapshot:   "",
-				RouteRules: nil,
-				Router:     nil,
-				Conn:       nil,
+				Host: host,
+				Port: port,
+				User: "root",
 			},
 			TargetTableConfigs: []*config.TableConfig{
 				{
-					Schema:          "schama1",
-					Table:           "table",
+					Schema:          "schema1",
+					Table:           "tbl",
 					IgnoreColumns:   []string{"", ""},
 					Fields:          "",
-					Range:           "age > 10 AND age < 20",
+					Range:           "a > 10 AND a < 20",
 					IsSharding:      false,
 					TargetTableInfo: nil,
 					Collation:       "",
@@ -611,10 +615,17 @@ func (s *testSourceSuite) TestSource(c *C) {
 		ConfigFile:   "config.toml",
 		PrintVersion: false,
 	}
-	cfg.Task.TargetCheckTables, err = filter.Parse([]string{"schema*.table*", "!c.*", "test2.t2"})
+	cfg.Task.TargetCheckTables, err = filter.Parse([]string{"schema*.tbl"})
+	c.Assert(err, IsNil)
+
+	// create table
+	conn, err := sql.Open("mysql", fmt.Sprintf("root:@tcp(%s:%d)/?charset=utf8mb4", host, port))
 	c.Assert(err, IsNil)
+
+	conn.Exec("CREATE DATABASE IF NOT EXISTS schema1")
+	conn.Exec("CREATE TABLE IF NOT EXISTS `schema1`.`tbl` (`a` int, `b` varchar(24), `c` float, `d` datetime, primary key(`a`, `b`))")
 	// create db connections refused.
 	// TODO unit_test covers source.go
 	_, _, err = NewSources(ctx, cfg)
-	c.Assert(err, NotNil)
+	c.Assert(err, IsNil)
 }
diff --git a/sync_diff_inspector/splitter/splitter_test.go b/sync_diff_inspector/splitter/splitter_test.go
index e9e0864..0f26478 100644
--- a/sync_diff_inspector/splitter/splitter_test.go
+++ b/sync_diff_inspector/splitter/splitter_test.go
@@ -769,7 +769,7 @@ func (s *testSplitterSuite) TestLimitSpliter(c *C) {
 	// Test Checkpoint
 	stopJ := 2
 	mock.ExpectQuery("SELECT COUNT\\(DISTINCT `a.*").WillReturnRows(sqlmock.NewRows([]string{"SEL"}).AddRow("123"))
-	createFakeResultForLimitSplit(mock, testCases[0].limitAValues[:stopJ], testCases[0].limitBValues[:stopJ], false)
+	createFakeResultForLimitSplit(mock, testCases[0].limitAValues[:stopJ], testCases[0].limitBValues[:stopJ], true)
 	iter, err := NewLimitIterator(ctx, "", tableDiff, db)
 	c.Assert(err, IsNil)
 	j := 0
@@ -833,3 +833,74 @@ func (s *testSplitterSuite) TestRangeInfo(c *C) {
 	c.Assert(rangeInfo2.GetTableIndex(), Equals, 1)
 
 }
+
+func (s *testSplitterSuite) TestChunkSize(c *C) {
+	ctx := context.Background()
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+
+	createTableSQL := "create table `test`.`test`(`a` int, `b` varchar(10), `c` float, `d` datetime, primary key(`a`, `b`))"
+	tableInfo, err := dbutil.GetTableInfoBySQL(createTableSQL, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiff := &common.TableDiff{
+		Schema:    "test",
+		Table:     "test",
+		Info:      tableInfo,
+		ChunkSize: 0,
+	}
+
+	// test bucket splitter chunksize
+	statsRows := sqlmock.NewRows([]string{"Db_name", "Table_name", "Column_name", "Is_index", "Bucket_id", "Count", "Repeats", "Lower_Bound", "Upper_Bound"})
+	statsRows.AddRow("test", "test", "PRIMARY", 1, 0, 1000000000, 1, "(1, 2)", "(2, 3)")
+	mock.ExpectQuery("SHOW STATS_BUCKETS").WillReturnRows(statsRows)
+	mock.ExpectQuery("SELECT COUNT\\(DISTINCT `a.*").WillReturnRows(sqlmock.NewRows([]string{"SEL"}).AddRow("123"))
+
+	bucketIter, err := NewBucketIterator(ctx, "", tableDiff, db)
+	c.Assert(err, IsNil)
+	c.Assert(bucketIter.chunkSize, Equals, int64(100000))
+
+	createFakeResultForBucketSplit(mock, nil, nil)
+	bucketIter, err = NewBucketIterator(ctx, "", tableDiff, db)
+	c.Assert(err, IsNil)
+	c.Assert(bucketIter.chunkSize, Equals, int64(50000))
+
+	// test random splitter chunksize
+	// chunkNum is only 1, so don't need randomValues
+	createFakeResultForRandomSplit(mock, 1000, nil)
+	randomIter, err := NewRandomIterator(ctx, "", tableDiff, db)
+	c.Assert(err, IsNil)
+	c.Assert(randomIter.chunkSize, Equals, int64(50000))
+
+	createFakeResultForRandomSplit(mock, 1000000000, [][]interface{}{
+		{1, 2, 3, 4, 5},
+		{"a", "b", "c", "d", "e"},
+	})
+	randomIter, err = NewRandomIterator(ctx, "", tableDiff, db)
+	c.Assert(err, IsNil)
+	c.Assert(randomIter.chunkSize, Equals, int64(100000))
+
+	createTableSQL = "create table `test`.`test`(`a` int, `b` varchar(10), `c` float, `d` datetime)"
+	tableInfo, err = dbutil.GetTableInfoBySQL(createTableSQL, parser.New())
+	c.Assert(err, IsNil)
+
+	tableDiff_noindex := &common.TableDiff{
+		Schema:    "test",
+		Table:     "test",
+		Info:      tableInfo,
+		ChunkSize: 0,
+	}
+	// no index
+	createFakeResultForRandomSplit(mock, 1000, nil)
+	randomIter, err = NewRandomIterator(ctx, "", tableDiff_noindex, db)
+	c.Assert(err, IsNil)
+	c.Assert(randomIter.chunkSize, Equals, int64(1000))
+
+	// test limit splitter chunksize
+	mock.ExpectQuery("SELECT COUNT\\(DISTINCT `a.*").WillReturnRows(sqlmock.NewRows([]string{"SEL"}).AddRow("123"))
+	createFakeResultForCount(mock, 1000)
+	mock.ExpectQuery("SELECT `a`,.*limit 50000.*").WillReturnRows(sqlmock.NewRows([]string{"a", "b"}))
+	_, err = NewLimitIterator(ctx, "", tableDiff, db)
+	c.Assert(err, IsNil)
+
+}
diff --git a/sync_diff_inspector/utils/utils_test.go b/sync_diff_inspector/utils/utils_test.go
index 3c55021..59213f6 100644
--- a/sync_diff_inspector/utils/utils_test.go
+++ b/sync_diff_inspector/utils/utils_test.go
@@ -425,3 +425,8 @@ func (s *testUtilsSuite) TestGetBetterIndex(c *C) {
 		c.Assert(indices[0].Name.O, Equals, "PRIMARY")
 	}
 }
+
+func (s *testUtilsSuite) TestCalculateChunkSize(c *C) {
+	c.Assert(CalculateChunkSize(1000), Equals, int64(50000))
+	c.Assert(CalculateChunkSize(1000000000), Equals, int64(100000))
+}