Repository: pingcap/dumpling
Pull request #279 (2021-04-30 03:34:51)

*: cherry-pick serveral PRs to release-4.0
<!--
Thank you for contributing to Dumpling! Please read the [CONTRIBUTING](https://github.com/pingcap/dumpling/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.

If you want to open the **Challenge Program** pull request, please use the following template:
https://raw.githubusercontent.com/pingcap/.github/master/.github/PULL_REQUEST_TEMPLATE/challenge-program.md

You can use it with query parameters: https://github.com/pingcap/dumpling/compare/master...${you branch}?template=challenge-program.md
-->

### What is changed and how it works?
cherry-pick master's PRs to prepare for a release.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

### Release note

<!-- bugfixes or new feature need a release note, must in the form of a list, such as

- support -T/--tables-list argument

or if no need to be included in the release note, just add the following line

- No release note
-->
- No 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/dumpling/pulls/279/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/dumpling/pulls/279/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%2Fdumpling).

<details>

Reviewer can indicate their review by writing `/lgtm` in a comment.
Reviewer can cancel approval by writing `/lgtm cancel` in a comment.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has been approved by:

- lance6716




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/dumpling/pulls/279/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/dumpling/pulls/279/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%2Fdumpling).

<details>

Reviewer can indicate their review by writing `/lgtm` in a comment.
Reviewer can cancel approval by writing `/lgtm cancel` in a comment.
</details>

<!--Review Notification Identifier-->
lichunzhu: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: 012f166ec47b2db289d024782e9f80253dc957dc</details>

Patch:
diff --git a/v4/export/config.go b/v4/export/config.go
index 00d8680..a5b99f7 100644
--- a/v4/export/config.go
+++ b/v4/export/config.go
@@ -565,6 +565,7 @@ var (
 
 // ServerInfo is the combination of ServerType and ServerInfo
 type ServerInfo struct {
+	HasTiKV       bool
 	ServerType    ServerType
 	ServerVersion *semver.Version
 }
diff --git a/v4/export/dump.go b/v4/export/dump.go
index 6eb5670..1ad1ca6 100755
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/br/pkg/summary"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	pclog "github.com/pingcap/log"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -399,13 +400,13 @@ func (d *Dumper) buildConcatTask(tctx *tcontext.Context, conn *sql.Conn, meta Ta
 	}
 }
 
-func (d *Dumper) dumpWholeTableDirectly(tctx *tcontext.Context, conn *sql.Conn, meta TableMeta, taskChan chan<- Task) error {
+func (d *Dumper) dumpWholeTableDirectly(tctx *tcontext.Context, conn *sql.Conn, meta TableMeta, taskChan chan<- Task, partition string, currentChunk, totalChunks int) error {
 	conf := d.conf
-	tableIR, err := SelectAllFromTable(conf, conn, meta)
+	tableIR, err := SelectAllFromTable(conf, conn, meta, partition)
 	if err != nil {
 		return err
 	}
-	task := NewTaskTableData(meta, tableIR, 0, 1)
+	task := NewTaskTableData(meta, tableIR, currentChunk, totalChunks)
 	ctxDone := d.sendTaskToChan(tctx, task, taskChan)
 	if ctxDone {
 		return tctx.Err()
@@ -431,7 +432,7 @@ func (d *Dumper) sequentialDumpTable(tctx *tcontext.Context, conn *sql.Conn, met
 			zap.String("database", meta.DatabaseName()),
 			zap.String("table", meta.TableName()))
 	}
-	return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan)
+	return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan, "", 0, 1)
 }
 
 // concurrentDumpTable tries to split table into several chunks to dump
@@ -440,9 +441,8 @@ func (d *Dumper) concurrentDumpTable(tctx *tcontext.Context, conn *sql.Conn, met
 	db, tbl := meta.DatabaseName(), meta.TableName()
 	if conf.ServerInfo.ServerType == ServerTypeTiDB &&
 		conf.ServerInfo.ServerVersion != nil &&
-		conf.ServerInfo.ServerVersion.Compare(*tableSampleVersion) >= 0 {
-		tctx.L().Debug("dumping TiDB tables with TABLESAMPLE",
-			zap.String("database", db), zap.String("table", tbl))
+		(conf.ServerInfo.ServerVersion.Compare(*tableSampleVersion) >= 0 ||
+			(conf.ServerInfo.HasTiKV && conf.ServerInfo.ServerVersion.Compare(*gcSafePointVersion) >= 0)) {
 		return d.concurrentDumpTiDBTables(tctx, conn, meta, taskChan)
 	}
 	field, err := pickupPossibleField(db, tbl, conn, conf)
@@ -453,7 +453,7 @@ func (d *Dumper) concurrentDumpTable(tctx *tcontext.Context, conn *sql.Conn, met
 		// skip split chunk logic if not found proper field
 		tctx.L().Warn("fallback to sequential dump due to no proper field",
 			zap.String("database", db), zap.String("table", tbl))
-		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan)
+		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan, "", 0, 1)
 	}
 
 	min, max, err := d.selectMinAndMaxIntValue(conn, db, tbl, field)
@@ -476,7 +476,7 @@ func (d *Dumper) concurrentDumpTable(tctx *tcontext.Context, conn *sql.Conn, met
 			zap.Uint64("conf.rows", conf.Rows),
 			zap.String("database", db),
 			zap.String("table", tbl))
-		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan)
+		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan, "", 0, 1)
 	}
 
 	// every chunk would have eventual adjustments
@@ -507,7 +507,7 @@ func (d *Dumper) concurrentDumpTable(tctx *tcontext.Context, conn *sql.Conn, met
 	for max.Cmp(cutoff) >= 0 {
 		nextCutOff := new(big.Int).Add(cutoff, bigEstimatedStep)
 		where := fmt.Sprintf("%s(`%s` >= %d AND `%s` < %d)", nullValueCondition, escapeString(field), cutoff, escapeString(field), nextCutOff)
-		query := buildSelectQuery(db, tbl, selectField, buildWhereCondition(conf, where), orderByClause)
+		query := buildSelectQuery(db, tbl, selectField, "", buildWhereCondition(conf, where), orderByClause)
 		if len(nullValueCondition) > 0 {
 			nullValueCondition = ""
 		}
@@ -571,16 +571,76 @@ func (d *Dumper) selectMinAndMaxIntValue(conn *sql.Conn, db, tbl, field string)
 }
 
 func (d *Dumper) concurrentDumpTiDBTables(tctx *tcontext.Context, conn *sql.Conn, meta TableMeta, taskChan chan<- Task) error {
-	conf := d.conf
 	db, tbl := meta.DatabaseName(), meta.TableName()
 
-	handleColNames, handleVals, err := selectTiDBTableSample(conn, db, tbl)
+	var (
+		handleColNames []string
+		handleVals     [][]string
+		err            error
+	)
+	if d.conf.ServerInfo.ServerVersion.Compare(*tableSampleVersion) >= 0 {
+		tctx.L().Debug("dumping TiDB tables with TABLESAMPLE",
+			zap.String("database", db), zap.String("table", tbl))
+		handleColNames, handleVals, err = selectTiDBTableSample(tctx, conn, db, tbl)
+	} else {
+		tctx.L().Debug("dumping TiDB tables with TABLE REGIONS",
+			zap.String("database", db), zap.String("table", tbl))
+		var partitions []string
+		partitions, err = GetPartitionNames(conn, db, tbl)
+		if err == nil {
+			if len(partitions) == 0 {
+				handleColNames, handleVals, err = selectTiDBTableRegion(tctx, conn, db, tbl)
+			} else {
+				return d.concurrentDumpTiDBPartitionTables(tctx, conn, meta, taskChan, partitions)
+			}
+		}
+	}
 	if err != nil {
 		return err
 	}
+	return d.sendConcurrentDumpTiDBTasks(tctx, conn, meta, taskChan, handleColNames, handleVals, "", 0, len(handleVals)+1)
+}
+
+func (d *Dumper) concurrentDumpTiDBPartitionTables(tctx *tcontext.Context, conn *sql.Conn, meta TableMeta, taskChan chan<- Task, partitions []string) error {
+	db, tbl := meta.DatabaseName(), meta.TableName()
+	tctx.L().Debug("dumping TiDB tables with TABLE REGIONS for partition table",
+		zap.String("database", db), zap.String("table", tbl), zap.Strings("partitions", partitions))
+
+	startChunkIdx := 0
+	totalChunk := 0
+	cachedHandleVals := make([][][]string, len(partitions))
+
+	handleColNames, _, err := selectTiDBRowKeyFields(conn, db, tbl, checkTiDBTableRegionPkFields)
+	if err != nil {
+		return err
+	}
+	// cache handleVals here to calculate the total chunks
+	for i, partition := range partitions {
+		handleVals, err := selectTiDBPartitionRegion(tctx, conn, db, tbl, partition)
+		if err != nil {
+			return err
+		}
+		totalChunk += len(handleVals) + 1
+		cachedHandleVals[i] = handleVals
+	}
+	for i, partition := range partitions {
+		err := d.sendConcurrentDumpTiDBTasks(tctx, conn, meta, taskChan, handleColNames, cachedHandleVals[i], partition, startChunkIdx, totalChunk)
+		if err != nil {
+			return err
+		}
+		startChunkIdx += len(cachedHandleVals[i]) + 1
+	}
+	return nil
+}
+
+func (d *Dumper) sendConcurrentDumpTiDBTasks(tctx *tcontext.Context,
+	conn *sql.Conn, meta TableMeta, taskChan chan<- Task,
+	handleColNames []string, handleVals [][]string, partition string, startChunkIdx, totalChunk int) error {
 	if len(handleVals) == 0 {
-		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan)
+		return d.dumpWholeTableDirectly(tctx, conn, meta, taskChan, partition, startChunkIdx, totalChunk)
 	}
+	conf := d.conf
+	db, tbl := meta.DatabaseName(), meta.TableName()
 	selectField, selectLen, err := buildSelectField(conn, db, tbl, conf.CompleteInsert)
 	if err != nil {
 		return err
@@ -589,8 +649,8 @@ func (d *Dumper) concurrentDumpTiDBTables(tctx *tcontext.Context, conn *sql.Conn
 	orderByClause := buildOrderByClauseString(handleColNames)
 
 	for i, w := range where {
-		query := buildSelectQuery(db, tbl, selectField, buildWhereCondition(conf, w), orderByClause)
-		task := NewTaskTableData(meta, newTableData(query, selectLen, false), i, len(where))
+		query := buildSelectQuery(db, tbl, selectField, partition, buildWhereCondition(conf, w), orderByClause)
+		task := NewTaskTableData(meta, newTableData(query, selectLen, false), i+startChunkIdx, totalChunk)
 		ctxDone := d.sendTaskToChan(tctx, task, taskChan)
 		if ctxDone {
 			return tctx.Err()
@@ -604,20 +664,14 @@ func (d *Dumper) L() log.Logger {
 	return d.tctx.L()
 }
 
-func selectTiDBTableSample(conn *sql.Conn, dbName, tableName string) (pkFields []string, pkVals [][]string, err error) {
-	pkFields, pkColTypes, err := GetPrimaryKeyAndColumnTypes(conn, dbName, tableName)
-	if err != nil {
-		return nil, nil, errors.Trace(err)
-	}
-	hasImplicitRowID, err := SelectTiDBRowID(conn, dbName, tableName)
+func selectTiDBTableSample(tctx *tcontext.Context, conn *sql.Conn, dbName, tableName string) (pkFields []string, pkVals [][]string, err error) {
+	pkFields, pkColTypes, err := selectTiDBRowKeyFields(conn, dbName, tableName, nil)
 	if err != nil {
 		return nil, nil, errors.Trace(err)
 	}
-	if hasImplicitRowID {
-		pkFields, pkColTypes = []string{"_tidb_rowid"}, []string{"BIGINT"}
-	}
+
 	query := buildTiDBTableSampleQuery(pkFields, dbName, tableName)
-	rows, err := conn.QueryContext(context.Background(), query)
+	rows, err := conn.QueryContext(tctx, query)
 	if err != nil {
 		return nil, nil, errors.Trace(err)
 	}
@@ -638,11 +692,11 @@ func selectTiDBTableSample(conn *sql.Conn, dbName, tableName string) (pkFields [
 			pkValRow = append(pkValRow, buf.String())
 			buf.Reset()
 		}
-		rowRec.WriteToBuffer(buf, true)
 		pkVals = append(pkVals, pkValRow)
 		iter.Next()
 	}
-	return pkFields, pkVals, nil
+	iter.Close()
+	return pkFields, pkVals, iter.Error()
 }
 
 func buildTiDBTableSampleQuery(pkFields []string, dbName, tblName string) string {
@@ -655,6 +709,124 @@ func buildTiDBTableSampleQuery(pkFields []string, dbName, tblName string) string
 	return fmt.Sprintf(template, pks, escapeString(dbName), escapeString(tblName), pks)
 }
 
+func selectTiDBRowKeyFields(conn *sql.Conn, dbName, tableName string, checkPkFields func([]string, []string) error) (pkFields, pkColTypes []string, err error) {
+	hasImplicitRowID, err := SelectTiDBRowID(conn, dbName, tableName)
+	if err != nil {
+		return
+	}
+	if hasImplicitRowID {
+		pkFields, pkColTypes = []string{"_tidb_rowid"}, []string{"BIGINT"}
+	} else {
+		pkFields, pkColTypes, err = GetPrimaryKeyAndColumnTypes(conn, dbName, tableName)
+		if err == nil {
+			if checkPkFields != nil {
+				err = checkPkFields(pkFields, pkColTypes)
+			}
+		}
+	}
+	return
+}
+
+func checkTiDBTableRegionPkFields(pkFields, pkColTypes []string) (err error) {
+	if len(pkFields) != 1 || len(pkColTypes) != 1 {
+		err = errors.Errorf("unsupported primary key for selectTableRegion. pkFields: [%s], pkColTypes: [%s]", strings.Join(pkFields, ", "), strings.Join(pkColTypes, ", "))
+		return
+	}
+	if _, ok := dataTypeNum[pkColTypes[0]]; !ok {
+		err = errors.Errorf("unsupported primary key type for selectTableRegion. pkFields: [%s], pkColTypes: [%s]", strings.Join(pkFields, ", "), strings.Join(pkColTypes, ", "))
+	}
+	return
+}
+
+func selectTiDBTableRegion(tctx *tcontext.Context, conn *sql.Conn, dbName, tableName string) (pkFields []string, pkVals [][]string, err error) {
+	pkFields, _, err = selectTiDBRowKeyFields(conn, dbName, tableName, checkTiDBTableRegionPkFields)
+	if err != nil {
+		return
+	}
+
+	var (
+		startKey, decodedKey sql.NullString
+		rowID                = -1
+	)
+	const (
+		tableRegionSQL = "SELECT START_KEY,tidb_decode_key(START_KEY) from INFORMATION_SCHEMA.TIKV_REGION_STATUS s WHERE s.DB_NAME = ? AND s.TABLE_NAME = ? AND IS_INDEX = 0 ORDER BY START_KEY;"
+		tidbRowID      = "_tidb_rowid="
+	)
+	logger := tctx.L().With(zap.String("database", dbName), zap.String("table", tableName))
+	err = simpleQueryWithArgs(conn, func(rows *sql.Rows) error {
+		rowID++
+		err = rows.Scan(&startKey, &decodedKey)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		// first region's start key has no use. It may come from another table or might be invalid
+		if rowID == 0 {
+			return nil
+		}
+		if !startKey.Valid {
+			logger.Debug("meet invalid start key", zap.Int("rowID", rowID))
+			return nil
+		}
+		if !decodedKey.Valid {
+			logger.Debug("meet invalid decoded start key", zap.Int("rowID", rowID), zap.String("startKey", startKey.String))
+			return nil
+		}
+		pkVal, err2 := extractTiDBRowIDFromDecodedKey(tidbRowID, decodedKey.String)
+		if err2 != nil {
+			logger.Debug("fail to extract pkVal from decoded start key",
+				zap.Int("rowID", rowID), zap.String("startKey", startKey.String), zap.String("decodedKey", decodedKey.String), zap.Error(err2))
+		} else {
+			pkVals = append(pkVals, []string{pkVal})
+		}
+		return nil
+	}, tableRegionSQL, dbName, tableName)
+
+	return pkFields, pkVals, errors.Trace(err)
+}
+
+func selectTiDBPartitionRegion(tctx *tcontext.Context, conn *sql.Conn, dbName, tableName, partition string) (pkVals [][]string, err error) {
+	var (
+		rows      *sql.Rows
+		startKeys []string
+	)
+	const (
+		partitionRegionSQL = "SHOW TABLE `%s`.`%s` PARTITION(`%s`) REGIONS"
+		regionRowKey       = "r_"
+	)
+	logger := tctx.L().With(zap.String("database", dbName), zap.String("table", tableName), zap.String("partition", partition))
+	rows, err = conn.QueryContext(tctx, fmt.Sprintf(partitionRegionSQL, escapeString(dbName), escapeString(tableName), escapeString(partition)))
+	if err != nil {
+		err = errors.Trace(err)
+		return
+	}
+	startKeys, err = GetSpecifiedColumnValueAndClose(rows, "START_KEY")
+	if err != nil {
+		return
+	}
+	for rowID, startKey := range startKeys {
+		if rowID == 0 {
+			continue
+		}
+		pkVal, err2 := extractTiDBRowIDFromDecodedKey(regionRowKey, startKey)
+		if err2 != nil {
+			logger.Debug("show table region start key doesn't have rowID",
+				zap.Int("rowID", rowID), zap.String("startKey", startKey), zap.Error(err2))
+		} else {
+			pkVals = append(pkVals, []string{pkVal})
+		}
+	}
+
+	return pkVals, err
+}
+
+func extractTiDBRowIDFromDecodedKey(indexField, key string) (string, error) {
+	if p := strings.Index(key, indexField); p != -1 {
+		p += len(indexField)
+		return key[p:], nil
+	}
+	return "", errors.Errorf("decoded key %s doesn't have %s field", key, indexField)
+}
+
 func prepareTableListToDump(tctx *tcontext.Context, conf *Config, db *sql.Conn) error {
 	databases, err := prepareDumpingDatabases(conf, db)
 	if err != nil {
@@ -764,12 +936,16 @@ func runSteps(d *Dumper, steps ...func(*Dumper) error) error {
 
 func initLogger(d *Dumper) error {
 	conf := d.conf
-	var logger log.Logger
+	var (
+		logger log.Logger
+		err    error
+		props  *pclog.ZapProperties
+	)
+	// conf.Logger != nil means dumpling is used as a library
 	if conf.Logger != nil {
 		logger = log.NewAppLogger(conf.Logger)
 	} else {
-		var err error
-		logger, err = log.InitAppLogger(&log.Config{
+		logger, props, err = log.InitAppLogger(&log.Config{
 			Level:  conf.LogLevel,
 			File:   conf.LogFile,
 			Format: conf.LogFormat,
@@ -777,6 +953,7 @@ func initLogger(d *Dumper) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+		pclog.ReplaceGlobals(logger.Logger, props)
 		cli.LogLongVersion(logger)
 	}
 	d.tctx = d.tctx.WithLogger(logger)
@@ -960,21 +1137,21 @@ func setSessionParam(d *Dumper) error {
 	if si.ServerType == ServerTypeTiDB && conf.TiDBMemQuotaQuery != UnspecifiedSize {
 		sessionParam[TiDBMemQuotaQueryName] = conf.TiDBMemQuotaQuery
 	}
+	var err error
 	if snapshot != "" {
 		if si.ServerType != ServerTypeTiDB {
 			return errors.New("snapshot consistency is not supported for this server")
 		}
 		if consistency == consistencyTypeSnapshot {
-			hasTiKV, err := CheckTiDBWithTiKV(pool)
+			conf.ServerInfo.HasTiKV, err = CheckTiDBWithTiKV(pool)
 			if err != nil {
 				return err
 			}
-			if hasTiKV {
+			if conf.ServerInfo.HasTiKV {
 				sessionParam["tidb_snapshot"] = snapshot
 			}
 		}
 	}
-	var err error
 	if d.dbHandle, err = resetDBWithSessionParams(d.tctx, pool, conf.GetDSN(""), conf.SessionParams); err != nil {
 		return errors.Trace(err)
 	}
diff --git a/v4/export/metrics.go b/v4/export/metrics.go
index 649a5ee..a2d671d 100644
--- a/v4/export/metrics.go
+++ b/v4/export/metrics.go
@@ -40,7 +40,7 @@ func InitMetricsVector(labels prometheus.Labels) {
 			Subsystem: "dump",
 			Name:      "estimate_total_rows",
 			Help:      "estimate total rows for dumpling tables",
-		}, []string{})
+		}, labelNames)
 	finishedRowsCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "dumpling",
diff --git a/v4/export/sql.go b/v4/export/sql.go
index 78c8652..655d9c2 100644
--- a/v4/export/sql.go
+++ b/v4/export/sql.go
@@ -18,6 +18,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const orderByTiDBRowID = "ORDER BY `_tidb_rowid`"
+
 // ShowDatabases shows the databases of a database server.
 func ShowDatabases(db *sql.Conn) ([]string, error) {
 	var res oneStrColumnTable
@@ -186,7 +188,7 @@ func SelectVersion(db *sql.DB) (string, error) {
 }
 
 // SelectAllFromTable dumps data serialized from a specified table
-func SelectAllFromTable(conf *Config, db *sql.Conn, meta TableMeta) (TableDataIR, error) {
+func SelectAllFromTable(conf *Config, db *sql.Conn, meta TableMeta, partition string) (TableDataIR, error) {
 	database, table := meta.DatabaseName(), meta.TableName()
 	selectedField, selectLen, err := buildSelectField(db, database, table, conf.CompleteInsert)
 	if err != nil {
@@ -197,7 +199,7 @@ func SelectAllFromTable(conf *Config, db *sql.Conn, meta TableMeta) (TableDataIR
 	if err != nil {
 		return nil, err
 	}
-	query := buildSelectQuery(database, table, selectedField, buildWhereCondition(conf, ""), orderByClause)
+	query := buildSelectQuery(database, table, selectedField, partition, buildWhereCondition(conf, ""), orderByClause)
 
 	return &tableData{
 		query:  query,
@@ -205,7 +207,7 @@ func SelectAllFromTable(conf *Config, db *sql.Conn, meta TableMeta) (TableDataIR
 	}, nil
 }
 
-func buildSelectQuery(database, table string, fields string, where string, orderByClause string) string {
+func buildSelectQuery(database, table, fields, partition, where, orderByClause string) string {
 	var query strings.Builder
 	query.WriteString("SELECT ")
 	if fields == "" {
@@ -218,7 +220,12 @@ func buildSelectQuery(database, table string, fields string, where string, order
 	query.WriteString(escapeString(database))
 	query.WriteString("`.`")
 	query.WriteString(escapeString(table))
-	query.WriteString("`")
+	query.WriteByte('`')
+	if partition != "" {
+		query.WriteString(" PARTITION(`")
+		query.WriteString(escapeString(partition))
+		query.WriteString("`)")
+	}
 
 	if where != "" {
 		query.WriteString(" ")
@@ -243,7 +250,7 @@ func buildOrderByClause(conf *Config, db *sql.Conn, database, table string) (str
 			return "", errors.Trace(err)
 		}
 		if ok {
-			return "ORDER BY `_tidb_rowid`", nil
+			return orderByTiDBRowID, nil
 		}
 	}
 	cols, err := GetPrimaryKeyColumns(db, database, table)
@@ -433,8 +440,13 @@ func ShowMasterStatus(db *sql.Conn) ([]string, error) {
 	return oneRow, nil
 }
 
-// GetSpecifiedColumnValue get columns' values whose name is equal to columnName
-func GetSpecifiedColumnValue(rows *sql.Rows, columnName string) ([]string, error) {
+// GetSpecifiedColumnValueAndClose get columns' values whose name is equal to columnName and close the given rows
+func GetSpecifiedColumnValueAndClose(rows *sql.Rows, columnName string) ([]string, error) {
+	if rows == nil {
+		return []string{}, nil
+	}
+	defer rows.Close()
+	columnName = strings.ToUpper(columnName)
 	var strs []string
 	columns, _ := rows.Columns()
 	addr := make([]interface{}, len(columns))
@@ -458,7 +470,7 @@ func GetSpecifiedColumnValue(rows *sql.Rows, columnName string) ([]string, error
 			strs = append(strs, oneRow[fieldIndex].String)
 		}
 	}
-	return strs, nil
+	return strs, errors.Trace(rows.Err())
 }
 
 // GetPdAddrs gets PD address from TiDB
@@ -470,8 +482,7 @@ func GetPdAddrs(tctx *tcontext.Context, db *sql.DB) ([]string, error) {
 			zap.String("query", query), zap.Error(err))
 		return []string{}, errors.Annotatef(err, "sql: %s", query)
 	}
-	defer rows.Close()
-	return GetSpecifiedColumnValue(rows, "STATUS_ADDRESS")
+	return GetSpecifiedColumnValueAndClose(rows, "STATUS_ADDRESS")
 }
 
 // GetTiDBDDLIDs gets DDL IDs from TiDB
@@ -483,8 +494,7 @@ func GetTiDBDDLIDs(tctx *tcontext.Context, db *sql.DB) ([]string, error) {
 			zap.String("query", query), zap.Error(err))
 		return []string{}, errors.Annotatef(err, "sql: %s", query)
 	}
-	defer rows.Close()
-	return GetSpecifiedColumnValue(rows, "DDL_ID")
+	return GetSpecifiedColumnValueAndClose(rows, "DDL_ID")
 }
 
 // CheckTiDBWithTiKV use sql to check whether current TiDB has TiKV
@@ -514,8 +524,20 @@ func isUnknownSystemVariableErr(err error) bool {
 func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, params map[string]interface{}) (*sql.DB, error) {
 	support := make(map[string]interface{})
 	for k, v := range params {
+		var pv interface{}
+		if str, ok := v.(string); ok {
+			if pvi, err := strconv.ParseInt(str, 10, 64); err == nil {
+				pv = pvi
+			} else if pvf, err := strconv.ParseFloat(str, 64); err == nil {
+				pv = pvf
+			} else {
+				pv = str
+			}
+		} else {
+			pv = v
+		}
 		s := fmt.Sprintf("SET SESSION %s = ?", k)
-		_, err := db.ExecContext(tctx, s, v)
+		_, err := db.ExecContext(tctx, s, pv)
 		if err != nil {
 			if isUnknownSystemVariableErr(err) {
 				tctx.L().Info("session variable is not supported by db", zap.String("variable", k), zap.Reflect("value", v))
@@ -524,7 +546,7 @@ func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, dsn string, pa
 			return nil, errors.Trace(err)
 		}
 
-		support[k] = v
+		support[k] = pv
 	}
 
 	for k, v := range support {
@@ -603,7 +625,7 @@ func buildSelectField(db *sql.Conn, dbName, tableName string, completeInsert boo
 }
 
 func buildWhereClauses(handleColNames []string, handleVals [][]string) []string {
-	if len(handleColNames) == 0 {
+	if len(handleColNames) == 0 || len(handleVals) == 0 {
 		return nil
 	}
 	quotaCols := make([]string, len(handleColNames))
@@ -663,26 +685,14 @@ func buildCompareClause(buf *bytes.Buffer, quotaCols []string, bound []string, c
 	}
 }
 
-// Compare returns an integer comparing two string arrays lexicographically.
-// return (compare result, common prefix length)
-// return -1 if low < up
-// return  0 if low == up
-// return  1 if low > up
-func compareArrString(low []string, up []string) (compareRes, commonLength int) {
-	l := len(low)
-	c := -1
-	if u := len(up); u < l {
-		l = u
-		c = 1
-	} else if u == l {
-		c = 0
-	}
-	for i := 0; i < l; i++ {
-		if d := strings.Compare(low[i], up[i]); d != 0 {
-			return d, i
+// getCommonLength returns the common length of low and up
+func getCommonLength(low []string, up []string) int {
+	for i := range low {
+		if low[i] != up[i] {
+			return i
 		}
 	}
-	return c, l
+	return len(low)
 }
 
 // buildBetweenClause build clause in a specified table range.
@@ -701,13 +711,13 @@ func buildBetweenClause(buf *bytes.Buffer, quotaCols []string, low []string, up
 		buf.WriteString(up[0])
 	}
 	// handle special cases with common prefix
-	compare, commonLen := compareArrString(low, up)
-	// unexpected case for low >= up, return empty result
-	if compare >= 0 {
-		buf.WriteString("false")
-		return
-	}
+	commonLen := getCommonLength(low, up)
 	if commonLen > 0 {
+		// unexpected case for low == up, return empty result
+		if commonLen == len(low) {
+			buf.WriteString("false")
+			return
+		}
 		for i := 0; i < commonLen; i++ {
 			if i > 0 {
 				buf.WriteString(" and ")
@@ -817,7 +827,7 @@ func simpleQueryWithArgs(conn *sql.Conn, handleOneRow func(*sql.Rows) error, sql
 		}
 	}
 	rows.Close()
-	return rows.Err()
+	return errors.Annotatef(rows.Err(), "sql: %s", sql)
 }
 
 func pickupPossibleField(dbName, tableName string, db *sql.Conn, conf *Config) (string, error) {
@@ -964,14 +974,14 @@ func buildWhereCondition(conf *Config, where string) string {
 	var query strings.Builder
 	separator := "WHERE"
 	if conf.Where != "" {
-		query.WriteString(" ")
 		query.WriteString(separator)
-		query.WriteString(" ")
+		query.WriteByte(' ')
 		query.WriteString(conf.Where)
+		query.WriteByte(' ')
 		separator = "AND"
+		query.WriteByte(' ')
 	}
 	if where != "" {
-		query.WriteString(" ")
 		query.WriteString(separator)
 		query.WriteString(" ")
 		query.WriteString(where)
@@ -982,3 +992,20 @@ func buildWhereCondition(conf *Config, where string) string {
 func escapeString(s string) string {
 	return strings.ReplaceAll(s, "`", "``")
 }
+
+// GetPartitionNames get partition names from a specified table
+func GetPartitionNames(db *sql.Conn, schema, table string) (partitions []string, err error) {
+	partitions = make([]string, 0)
+	var partitionName sql.NullString
+	err = simpleQueryWithArgs(db, func(rows *sql.Rows) error {
+		err := rows.Scan(&partitionName)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		if partitionName.Valid {
+			partitions = append(partitions, partitionName.String)
+		}
+		return nil
+	}, "SELECT PARTITION_NAME from INFORMATION_SCHEMA.PARTITIONS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?", schema, table)
+	return
+}
diff --git a/v4/export/sql_type.go b/v4/export/sql_type.go
index 6e117f0..6945096 100644
--- a/v4/export/sql_type.go
+++ b/v4/export/sql_type.go
@@ -17,38 +17,49 @@ var (
 	doubleQuotationMark = []byte{'"'}
 )
 
+// There are two kinds of scenes to use this dataType
+// The first is to be the receiver of table sample, which will use tidb's INFORMATION_SCHEMA.COLUMNS's DATA_TYPE column, which is from
+// https://github.com/pingcap/tidb/blob/619c4720059ea619081b01644ef3084b426d282f/executor/infoschema_reader.go#L654
+// https://github.com/pingcap/parser/blob/8e8ed7927bde11c4cf0967afc5e05ab5aeb14cc7/types/etc.go#L44-70
+// The second is to be the receiver of select row type, which will use sql.DB's rows.DatabaseTypeName(), which is from
+// https://github.com/go-sql-driver/mysql/blob/v1.5.0/fields.go#L17-97
 func initColTypeRowReceiverMap() {
-	for _, s := range dataTypeString {
+	dataTypeStringArr := []string{
+		"CHAR", "NCHAR", "VARCHAR", "NVARCHAR", "CHARACTER", "VARCHARACTER",
+		"TIMESTAMP", "DATETIME", "DATE", "TIME", "YEAR", "SQL_TSI_YEAR",
+		"TEXT", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT",
+		"ENUM", "SET", "JSON", "NULL", "VAR_STRING",
+	}
+
+	dataTypeNumArr := []string{
+		"INTEGER", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT",
+		"INT", "INT1", "INT2", "INT3", "INT8",
+		"FLOAT", "REAL", "DOUBLE", "DOUBLE PRECISION",
+		"DECIMAL", "NUMERIC", "FIXED",
+		"BOOL", "BOOLEAN",
+	}
+
+	dataTypeBinArr := []string{
+		"BLOB", "TINYBLOB", "MEDIUMBLOB", "LONGBLOB", "LONG",
+		"BINARY", "VARBINARY",
+		"BIT", "GEOMETRY",
+	}
+
+	for _, s := range dataTypeStringArr {
+		dataTypeString[s] = struct{}{}
 		colTypeRowReceiverMap[s] = SQLTypeStringMaker
 	}
-	for _, s := range dataTypeNum {
+	for _, s := range dataTypeNumArr {
+		dataTypeNum[s] = struct{}{}
 		colTypeRowReceiverMap[s] = SQLTypeNumberMaker
 	}
-	for _, s := range dataTypeBin {
+	for _, s := range dataTypeBinArr {
+		dataTypeBin[s] = struct{}{}
 		colTypeRowReceiverMap[s] = SQLTypeBytesMaker
 	}
 }
 
-var dataTypeString = []string{
-	"CHAR", "NCHAR", "VARCHAR", "NVARCHAR", "CHARACTER", "VARCHARACTER",
-	"TIMESTAMP", "DATETIME", "DATE", "TIME", "YEAR", "SQL_TSI_YEAR",
-	"TEXT", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT",
-	"ENUM", "SET", "JSON",
-}
-
-var dataTypeNum = []string{
-	"INTEGER", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT",
-	"INT", "INT1", "INT2", "INT3", "INT8",
-	"FLOAT", "REAL", "DOUBLE", "DOUBLE PRECISION",
-	"DECIMAL", "NUMERIC", "FIXED",
-	"BOOL", "BOOLEAN",
-}
-
-var dataTypeBin = []string{
-	"BLOB", "TINYBLOB", "MEDIUMBLOB", "LONGBLOB", "LONG",
-	"BINARY", "VARBINARY",
-	"BIT",
-}
+var dataTypeString, dataTypeNum, dataTypeBin = make(map[string]struct{}), make(map[string]struct{}), make(map[string]struct{})
 
 func escapeBackslashSQL(s []byte, bf *bytes.Buffer) {
 	var (
diff --git a/v4/export/status.go b/v4/export/status.go
index 3f93f7b..909469c 100644
--- a/v4/export/status.go
+++ b/v4/export/status.go
@@ -77,6 +77,6 @@ func (d *Dumper) getEstimateTotalRowsCount(tctx *tcontext.Context, conn *sql.Con
 			}
 		}
 	}
-	estimateTotalRowsCounter.With(conf.Labels).Add(float64(totalCount))
+	AddCounter(estimateTotalRowsCounter, conf.Labels, float64(totalCount))
 	return nil
 }
diff --git a/v4/log/log.go b/v4/log/log.go
index 4632634..79f840d 100644
--- a/v4/log/log.go
+++ b/v4/log/log.go
@@ -38,8 +38,8 @@ type Config struct {
 }
 
 // InitAppLogger inits the wrapped logger from config.
-func InitAppLogger(cfg *Config) (Logger, error) {
-	logger, _, err := pclog.InitLogger(&pclog.Config{
+func InitAppLogger(cfg *Config) (Logger, *pclog.ZapProperties, error) {
+	logger, props, err := pclog.InitLogger(&pclog.Config{
 		Level: cfg.Level,
 		File: pclog.FileLogConfig{
 			Filename:   cfg.File,
@@ -50,9 +50,9 @@ func InitAppLogger(cfg *Config) (Logger, error) {
 		Format: cfg.Format,
 	})
 	if err != nil {
-		return appLogger, errors.Trace(err)
+		return appLogger, props, errors.Trace(err)
 	}
-	return Logger{logger.WithOptions(zap.AddStacktrace(zap.DPanicLevel))}, nil
+	return Logger{logger.WithOptions(zap.AddStacktrace(zap.DPanicLevel))}, props, nil
 }
 
 // NewAppLogger returns the wrapped logger from config.


Test patch:
diff --git a/tests/primary_key/data/pk_case_3.sql b/tests/primary_key/data/pk_case_3.sql
index 142102c..ab69e5b 100644
--- a/tests/primary_key/data/pk_case_3.sql
+++ b/tests/primary_key/data/pk_case_3.sql
@@ -1,14 +1,14 @@
 # test random order and no primary key
-create table `pk_case_3` (a int, b int);
+create table `pk_case_3` (a int, b int, g geometry);
 insert into `pk_case_3` values
-(6, 4),
-(4, 6),
-(8, 2),
-(3, 7),
-(1, 9),
-(2, 8),
-(5, 5),
-(10, 0),
-(0, 10),
-(9, 1),
-(7, 3);
+(6, 4, ST_GeomFromText('POINT(1 1)')),
+(4, 6, ST_GeomFromText('LINESTRING(2 1, 6 6)')),
+(8, 2, NULL),
+(3, 7, NULL),
+(1, 9, NULL),
+(2, 8, NULL),
+(5, 5, NULL),
+(10, 0, NULL),
+(0, 10, NULL),
+(9, 1, NULL),
+(7, 3, NULL);
diff --git a/tests/primary_key/result/pk_case_3.sql b/tests/primary_key/result/pk_case_3.sql
index 0698d1a..a29e568 100644
--- a/tests/primary_key/result/pk_case_3.sql
+++ b/tests/primary_key/result/pk_case_3.sql
@@ -1,13 +1,13 @@
 /*!40101 SET NAMES binary*/;
 INSERT INTO `pk_case_3` VALUES
-(6,4),
-(4,6),
-(8,2),
-(3,7),
-(1,9),
-(2,8),
-(5,5),
-(10,0),
-(0,10),
-(9,1),
-(7,3);
+(6,4,x'000000000101000000000000000000f03f000000000000f03f'),
+(4,6,x'000000000102000000020000000000000000000040000000000000f03f00000000000018400000000000001840'),
+(8,2,NULL),
+(3,7,NULL),
+(1,9,NULL),
+(2,8,NULL),
+(5,5,NULL),
+(10,0,NULL),
+(0,10,NULL),
+(9,1,NULL),
+(7,3,NULL);
diff --git a/v4/export/dump_test.go b/v4/export/dump_test.go
index b9d8ec6..64cccc2 100644
--- a/v4/export/dump_test.go
+++ b/v4/export/dump_test.go
@@ -25,7 +25,7 @@ func (s *testSQLSuite) TestDumpBlock(c *C) {
 		WillReturnRows(sqlmock.NewRows([]string{"Database", "Create Database"}).
 			AddRow("test", "CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET utf8mb4 */"))
 
-	tctx, cancel := tcontext.Background().WithCancel()
+	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
 	defer cancel()
 	conn, err := db.Conn(tctx)
 	c.Assert(err, IsNil)
diff --git a/v4/export/sql_test.go b/v4/export/sql_test.go
index bdecade..20d5e49 100644
--- a/v4/export/sql_test.go
+++ b/v4/export/sql_test.go
@@ -5,13 +5,16 @@ package export
 import (
 	"context"
 	"database/sql"
-	"errors"
+	"database/sql/driver"
+	"fmt"
+	"strings"
 
 	tcontext "github.com/pingcap/dumpling/v4/context"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/coreos/go-semver/semver"
 	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
 )
 
 var _ = Suite(&testSQLSuite{})
@@ -83,7 +86,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 
 	selectedField, _, err := buildSelectField(conn, "test", "t", false)
 	c.Assert(err, IsNil)
-	q := buildSelectQuery("test", "t", selectedField, "", orderByClause)
+	q := buildSelectQuery("test", "t", selectedField, "", "", orderByClause)
 	c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY `_tidb_rowid`")
 
 	// _tidb_rowid is unavailable, or PKIsHandle.
@@ -103,7 +106,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 
 	selectedField, _, err = buildSelectField(conn, "test", "t", false)
 	c.Assert(err, IsNil)
-	q = buildSelectQuery("test", "t", selectedField, "", orderByClause)
+	q = buildSelectQuery("test", "t", selectedField, "", "", orderByClause)
 	c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY `id`")
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 
@@ -126,7 +129,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 
 		selectedField, _, err = buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
-		q = buildSelectQuery("test", "t", selectedField, "", orderByClause)
+		q = buildSelectQuery("test", "t", selectedField, "", "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY `id`", cmt)
 		err = mock.ExpectationsWereMet()
 		c.Assert(err, IsNil, cmt)
@@ -150,7 +153,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 
 		selectedField, _, err = buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
-		q := buildSelectQuery("test", "t", selectedField, "", orderByClause)
+		q := buildSelectQuery("test", "t", selectedField, "", "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t`", cmt)
 		err = mock.ExpectationsWereMet()
 		c.Assert(err, IsNil, cmt)
@@ -169,7 +172,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 
 		selectedField, _, err := buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
-		q := buildSelectQuery("test", "t", selectedField, "", "")
+		q := buildSelectQuery("test", "t", selectedField, "", "", "")
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t`", cmt)
 		c.Assert(mock.ExpectationsWereMet(), IsNil, cmt)
 	}
@@ -194,7 +197,7 @@ func (s *testSQLSuite) TestBuildOrderByClause(c *C) {
 
 	orderByClause, err := buildOrderByClause(mockConf, conn, "test", "t")
 	c.Assert(err, IsNil)
-	c.Assert(orderByClause, Equals, "ORDER BY `_tidb_rowid`")
+	c.Assert(orderByClause, Equals, orderByTiDBRowID)
 
 	// _tidb_rowid is unavailable, or PKIsHandle.
 	mock.ExpectExec("SELECT _tidb_rowid from `test`.`t`").
@@ -354,11 +357,13 @@ func (s *testSQLSuite) TestGetSuitableRows(c *C) {
 	defer db.Close()
 	conn, err := db.Conn(context.Background())
 	c.Assert(err, IsNil)
-	const query = "select AVG_ROW_LENGTH from INFORMATION_SCHEMA.TABLES where table_schema=\\? and table_name=\\?;"
 	tctx, cancel := tcontext.Background().WithCancel()
 	defer cancel()
-	database := "foo"
-	table := "bar"
+	const (
+		query    = "select AVG_ROW_LENGTH from INFORMATION_SCHEMA.TABLES where table_schema=\\? and table_name=\\?;"
+		database = "foo"
+		table    = "bar"
+	)
 
 	testCases := []struct {
 		avgRowLength uint64
@@ -405,42 +410,99 @@ func (s *testSQLSuite) TestGetSuitableRows(c *C) {
 	}
 }
 
-func (s *testSQLSuite) TestBuildWhereClauses(c *C) {
+func (s *testSQLSuite) TestBuildTableSampleQueries(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+	conn, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
+
+	d := &Dumper{
+		tctx:      tctx,
+		conf:      DefaultConfig(),
+		cancelCtx: cancel,
+	}
+	d.conf.ServerInfo = ServerInfo{
+		HasTiKV:       true,
+		ServerType:    ServerTypeTiDB,
+		ServerVersion: tableSampleVersion,
+	}
+
+	const (
+		database = "foo"
+		table    = "bar"
+	)
+
 	testCases := []struct {
 		handleColNames       []string
-		handleVals           [][]string
+		handleColTypes       []string
+		handleVals           [][]driver.Value
 		expectedWhereClauses []string
+		hasTiDBRowID         bool
 	}{
 		{
 			[]string{},
-			[][]string{},
+			[]string{},
+			[][]driver.Value{},
 			nil,
+			false,
 		},
 		{
 			[]string{"a"},
-			[][]string{{"1"}},
+			[]string{"bigint"},
+			[][]driver.Value{{1}},
 			[]string{"`a`<1", "`a`>=1"},
+			false,
 		},
+		// check whether dumpling can turn to dump whole table
 		{
 			[]string{"a"},
-			[][]string{
-				{"1"},
-				{"2"},
-				{"3"},
+			[]string{"bigint"},
+			[][]driver.Value{},
+			nil,
+			false,
+		},
+		// check whether dumpling can turn to dump whole table
+		{
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[][]driver.Value{},
+			nil,
+			true,
+		},
+		{
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[][]driver.Value{{1}},
+			[]string{"`_tidb_rowid`<1", "`_tidb_rowid`>=1"},
+			true,
+		},
+		{
+			[]string{"a"},
+			[]string{"bigint"},
+			[][]driver.Value{
+				{1},
+				{2},
+				{3},
 			},
 			[]string{"`a`<1", "`a`>=1 and `a`<2", "`a`>=2 and `a`<3", "`a`>=3"},
+			false,
 		},
 		{
 			[]string{"a", "b"},
-			[][]string{{"1", "2"}},
+			[]string{"bigint", "bigint"},
+			[][]driver.Value{{1, 2}},
 			[]string{"`a`<1 or(`a`=1 and `b`<2)", "`a`>1 or(`a`=1 and `b`>=2)"},
+			false,
 		},
 		{
 			[]string{"a", "b"},
-			[][]string{
-				{"1", "2"},
-				{"3", "4"},
-				{"5", "6"},
+			[]string{"bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2},
+				{3, 4},
+				{5, 6},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)",
@@ -448,85 +510,658 @@ func (s *testSQLSuite) TestBuildWhereClauses(c *C) {
 				"(`a`>3 and `a`<5)or(`a`=3 and(`b`>=4))or(`a`=5 and(`b`<6))",
 				"`a`>5 or(`a`=5 and `b`>=6)",
 			},
+			false,
 		},
 		{
 			[]string{"a", "b", "c"},
-			[][]string{
-				{"1", "2", "3"},
-				{"4", "5", "6"},
+			[]string{"bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2, 3},
+				{4, 5, 6},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<3)",
 				"(`a`>1 and `a`<4)or(`a`=1 and(`b`>2 or(`b`=2 and `c`>=3)))or(`a`=4 and(`b`<5 or(`b`=5 and `c`<6)))",
 				"`a`>4 or(`a`=4 and `b`>5)or(`a`=4 and `b`=5 and `c`>=6)",
 			},
+			false,
 		},
 		{
 			[]string{"a", "b", "c"},
-			[][]string{
-				{"1", "2", "3"},
-				{"1", "4", "5"},
+			[]string{"bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2, 3},
+				{1, 4, 5},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<3)",
 				"`a`=1 and((`b`>2 and `b`<4)or(`b`=2 and(`c`>=3))or(`b`=4 and(`c`<5)))",
 				"`a`>1 or(`a`=1 and `b`>4)or(`a`=1 and `b`=4 and `c`>=5)",
 			},
+			false,
 		},
 		{
 			[]string{"a", "b", "c"},
-			[][]string{
-				{"1", "2", "3"},
-				{"1", "2", "8"},
+			[]string{"bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2, 3},
+				{1, 2, 8},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<3)",
 				"`a`=1 and `b`=2 and(`c`>=3 and `c`<8)",
 				"`a`>1 or(`a`=1 and `b`>2)or(`a`=1 and `b`=2 and `c`>=8)",
 			},
+			false,
 		},
 		// special case: avoid return same samples
 		{
 			[]string{"a", "b", "c"},
-			[][]string{
-				{"1", "2", "3"},
-				{"1", "2", "3"},
+			[]string{"bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2, 3},
+				{1, 2, 3},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<3)",
 				"false",
 				"`a`>1 or(`a`=1 and `b`>2)or(`a`=1 and `b`=2 and `c`>=3)",
 			},
+			false,
+		},
+		// special case: numbers has bigger lexicographically order but lower number
+		{
+			[]string{"a", "b", "c"},
+			[]string{"bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{12, 2, 3},
+				{111, 4, 5},
+			},
+			[]string{
+				"`a`<12 or(`a`=12 and `b`<2)or(`a`=12 and `b`=2 and `c`<3)",
+				"(`a`>12 and `a`<111)or(`a`=12 and(`b`>2 or(`b`=2 and `c`>=3)))or(`a`=111 and(`b`<4 or(`b`=4 and `c`<5)))", // should return sql correctly
+				"`a`>111 or(`a`=111 and `b`>4)or(`a`=111 and `b`=4 and `c`>=5)",
+			},
+			false,
 		},
 		// test string fields
 		{
 			[]string{"a", "b", "c"},
-			[][]string{
-				{"1", "2", "\"3\""},
-				{"1", "4", "\"5\""},
+			[]string{"bigint", "bigint", "varchar"},
+			[][]driver.Value{
+				{1, 2, "3"},
+				{1, 4, "5"},
 			},
 			[]string{
-				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<\"3\")",
-				"`a`=1 and((`b`>2 and `b`<4)or(`b`=2 and(`c`>=\"3\"))or(`b`=4 and(`c`<\"5\")))",
-				"`a`>1 or(`a`=1 and `b`>4)or(`a`=1 and `b`=4 and `c`>=\"5\")",
+				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<'3')",
+				"`a`=1 and((`b`>2 and `b`<4)or(`b`=2 and(`c`>='3'))or(`b`=4 and(`c`<'5')))",
+				"`a`>1 or(`a`=1 and `b`>4)or(`a`=1 and `b`=4 and `c`>='5')",
 			},
+			false,
 		},
 		{
 			[]string{"a", "b", "c", "d"},
-			[][]string{
-				{"1", "2", "3", "4"},
-				{"5", "6", "7", "8"},
+			[]string{"bigint", "bigint", "bigint", "bigint"},
+			[][]driver.Value{
+				{1, 2, 3, 4},
+				{5, 6, 7, 8},
 			},
 			[]string{
 				"`a`<1 or(`a`=1 and `b`<2)or(`a`=1 and `b`=2 and `c`<3)or(`a`=1 and `b`=2 and `c`=3 and `d`<4)",
 				"(`a`>1 and `a`<5)or(`a`=1 and(`b`>2 or(`b`=2 and `c`>3)or(`b`=2 and `c`=3 and `d`>=4)))or(`a`=5 and(`b`<6 or(`b`=6 and `c`<7)or(`b`=6 and `c`=7 and `d`<8)))",
 				"`a`>5 or(`a`=5 and `b`>6)or(`a`=5 and `b`=6 and `c`>7)or(`a`=5 and `b`=6 and `c`=7 and `d`>=8)",
 			},
+			false,
 		},
 	}
-	for _, testCase := range testCases {
-		whereClauses := buildWhereClauses(testCase.handleColNames, testCase.handleVals)
+	transferHandleValStrings := func(handleColTypes []string, handleVals [][]driver.Value) [][]string {
+		handleValStrings := make([][]string, 0, len(handleVals))
+		for _, handleVal := range handleVals {
+			handleValString := make([]string, 0, len(handleVal))
+			for i, val := range handleVal {
+				rec := colTypeRowReceiverMap[strings.ToUpper(handleColTypes[i])]()
+				var valStr string
+				switch rec.(type) {
+				case *SQLTypeString:
+					valStr = fmt.Sprintf("'%s'", val)
+				case *SQLTypeBytes:
+					valStr = fmt.Sprintf("x'%x'", val)
+				case *SQLTypeNumber:
+					valStr = fmt.Sprintf("%d", val)
+				}
+				handleValString = append(handleValString, valStr)
+			}
+			handleValStrings = append(handleValStrings, handleValString)
+		}
+		return handleValStrings
+	}
+
+	for caseID, testCase := range testCases {
+		c.Log(fmt.Sprintf("case #%d", caseID))
+		handleColNames := testCase.handleColNames
+		handleColTypes := testCase.handleColTypes
+		handleVals := testCase.handleVals
+		handleValStrings := transferHandleValStrings(handleColTypes, handleVals)
+
+		// Test build whereClauses
+		whereClauses := buildWhereClauses(handleColNames, handleValStrings)
 		c.Assert(whereClauses, DeepEquals, testCase.expectedWhereClauses)
+
+		// Test build tasks through table sample
+		if len(handleColNames) > 0 {
+			taskChan := make(chan Task, 128)
+			quotaCols := make([]string, 0, len(handleColNames))
+			for _, col := range quotaCols {
+				quotaCols = append(quotaCols, wrapBackTicks(col))
+			}
+			selectFields := strings.Join(quotaCols, ",")
+			meta := &tableMeta{
+				database:      database,
+				table:         table,
+				selectedField: selectFields,
+				specCmts: []string{
+					"/*!40101 SET NAMES binary*/;",
+				},
+			}
+
+			if testCase.hasTiDBRowID {
+				mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+					WillReturnResult(sqlmock.NewResult(0, 0))
+			} else {
+				mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+					WillReturnError(errors.Errorf("ERROR %d (%s): %s", 1054, "42S22", "Unknown column '_tidb_rowid' in 'field list'"))
+				rows := sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE"})
+				for i := range handleColNames {
+					rows.AddRow(handleColNames[i], handleColTypes[i])
+				}
+				mock.ExpectQuery("SELECT c.COLUMN_NAME, DATA_TYPE FROM").WithArgs(database, table).WillReturnRows(rows)
+			}
+
+			rows := sqlmock.NewRows(handleColNames)
+			for _, handleVal := range handleVals {
+				rows.AddRow(handleVal...)
+			}
+			mock.ExpectQuery(fmt.Sprintf("SELECT .* FROM `%s`.`%s` TABLESAMPLE REGIONS", database, table)).WillReturnRows(rows)
+
+			rows = sqlmock.NewRows([]string{"COLUMN_NAME", "EXTRA"})
+			for _, handleCol := range handleColNames {
+				rows.AddRow(handleCol, "")
+			}
+			mock.ExpectQuery("SELECT COLUMN_NAME,EXTRA FROM INFORMATION_SCHEMA.COLUMNS").WithArgs(database, table).
+				WillReturnRows(rows)
+			// special case, no value found, will scan whole table and try build order clause
+			if len(handleVals) == 0 {
+				mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+					WillReturnResult(sqlmock.NewResult(0, 0))
+			}
+
+			c.Assert(d.concurrentDumpTable(tctx, conn, meta, taskChan), IsNil)
+			c.Assert(mock.ExpectationsWereMet(), IsNil)
+			orderByClause := buildOrderByClauseString(handleColNames)
+
+			checkQuery := func(i int, query string) {
+				task := <-taskChan
+				taskTableData, ok := task.(*TaskTableData)
+				c.Assert(ok, IsTrue)
+				c.Assert(taskTableData.ChunkIndex, Equals, i)
+				data, ok := taskTableData.Data.(*tableData)
+				c.Assert(ok, IsTrue)
+				c.Assert(data.query, Equals, query)
+			}
+
+			// special case, no value found
+			if len(handleVals) == 0 {
+				orderByClause = orderByTiDBRowID
+				query := buildSelectQuery(database, table, "*", "", "", orderByClause)
+				checkQuery(0, query)
+				continue
+			}
+
+			for i, w := range testCase.expectedWhereClauses {
+				query := buildSelectQuery(database, table, "*", "", buildWhereCondition(d.conf, w), orderByClause)
+				checkQuery(i, query)
+			}
+		}
+	}
+}
+
+func (s *testSQLSuite) TestBuildPartitionClauses(c *C) {
+	const (
+		dbName        = "test"
+		tbName        = "t"
+		fields        = "*"
+		partition     = "p0"
+		where         = "WHERE a > 10"
+		orderByClause = "ORDER BY a"
+	)
+	testCases := []struct {
+		partition     string
+		where         string
+		orderByClause string
+		expectedQuery string
+	}{
+		{
+			"",
+			"",
+			"",
+			"SELECT * FROM `test`.`t`",
+		},
+		{
+			partition,
+			"",
+			"",
+			"SELECT * FROM `test`.`t` PARTITION(`p0`)",
+		},
+		{
+			partition,
+			where,
+			"",
+			"SELECT * FROM `test`.`t` PARTITION(`p0`) WHERE a > 10",
+		},
+		{
+			partition,
+			"",
+			orderByClause,
+			"SELECT * FROM `test`.`t` PARTITION(`p0`) ORDER BY a",
+		},
+		{
+			partition,
+			where,
+			orderByClause,
+			"SELECT * FROM `test`.`t` PARTITION(`p0`) WHERE a > 10 ORDER BY a",
+		},
+		{
+			"",
+			where,
+			orderByClause,
+			"SELECT * FROM `test`.`t` WHERE a > 10 ORDER BY a",
+		},
+	}
+	for _, testCase := range testCases {
+		query := buildSelectQuery(dbName, tbName, fields, testCase.partition, testCase.where, testCase.orderByClause)
+		c.Assert(query, Equals, testCase.expectedQuery)
+	}
+}
+
+func (s *testSQLSuite) TestBuildRegionQueriesWithoutPartition(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+	conn, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
+
+	d := &Dumper{
+		tctx:      tctx,
+		conf:      DefaultConfig(),
+		cancelCtx: cancel,
+	}
+	d.conf.ServerInfo = ServerInfo{
+		HasTiKV:       true,
+		ServerType:    ServerTypeTiDB,
+		ServerVersion: gcSafePointVersion,
+	}
+	database := "foo"
+	table := "bar"
+
+	testCases := []struct {
+		regionResults        [][]driver.Value
+		handleColNames       []string
+		handleColTypes       []string
+		expectedWhereClauses []string
+		hasTiDBRowID         bool
+	}{
+		{
+			[][]driver.Value{
+				{"7480000000000000FF3300000000000000F8", "7480000000000000FF3300000000000000F8"},
+			},
+			[]string{"a"},
+			[]string{"bigint"},
+			[]string{
+				"",
+			},
+			false,
+		},
+		{
+			[][]driver.Value{
+				{"7480000000000000FF3300000000000000F8", "7480000000000000FF3300000000000000F8"},
+			},
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[]string{
+				"",
+			},
+			true,
+		},
+		{
+			[][]driver.Value{
+				{"7480000000000000FF3300000000000000F8", "7480000000000000FF3300000000000000F8"},
+				{"7480000000000000FF335F728000000000FF0EA6010000000000FA", "tableID=51, _tidb_rowid=960001"},
+				{"7480000000000000FF335F728000000000FF1D4C010000000000FA", "tableID=51, _tidb_rowid=1920001"},
+				{"7480000000000000FF335F728000000000FF2BF2010000000000FA", "tableID=51, _tidb_rowid=2880001"},
+			},
+			[]string{"a"},
+			[]string{"bigint"},
+			[]string{
+				"`a`<960001",
+				"`a`>=960001 and `a`<1920001",
+				"`a`>=1920001 and `a`<2880001",
+				"`a`>=2880001",
+			},
+			false,
+		},
+		{
+			[][]driver.Value{
+				{"7480000000000000FF3300000000000000F8", "7480000000000000FF3300000000000000F8"},
+				{"7480000000000000FF335F728000000000FF0EA6010000000000FA", "tableID=51, _tidb_rowid=960001"},
+				// one invalid key
+				{"7520000000000000FF335F728000000000FF0EA6010000000000FA", "7520000000000000FF335F728000000000FF0EA6010000000000FA"},
+				{"7480000000000000FF335F728000000000FF1D4C010000000000FA", "tableID=51, _tidb_rowid=1920001"},
+				{"7480000000000000FF335F728000000000FF2BF2010000000000FA", "tableID=51, _tidb_rowid=2880001"},
+			},
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[]string{
+				"`_tidb_rowid`<960001",
+				"`_tidb_rowid`>=960001 and `_tidb_rowid`<1920001",
+				"`_tidb_rowid`>=1920001 and `_tidb_rowid`<2880001",
+				"`_tidb_rowid`>=2880001",
+			},
+			true,
+		},
+	}
+
+	for caseID, testCase := range testCases {
+		c.Log(fmt.Sprintf("case #%d", caseID))
+		handleColNames := testCase.handleColNames
+		handleColTypes := testCase.handleColTypes
+		regionResults := testCase.regionResults
+
+		// Test build tasks through table region
+		taskChan := make(chan Task, 128)
+		quotaCols := make([]string, 0, len(handleColNames))
+		for _, col := range quotaCols {
+			quotaCols = append(quotaCols, wrapBackTicks(col))
+		}
+		selectFields := strings.Join(quotaCols, ",")
+		meta := &tableMeta{
+			database:      database,
+			table:         table,
+			selectedField: selectFields,
+			specCmts: []string{
+				"/*!40101 SET NAMES binary*/;",
+			},
+		}
+
+		mock.ExpectQuery("SELECT PARTITION_NAME from INFORMATION_SCHEMA.PARTITIONS").
+			WithArgs(database, table).WillReturnRows(sqlmock.NewRows([]string{"PARTITION_NAME"}).AddRow(nil))
+
+		if testCase.hasTiDBRowID {
+			mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+				WillReturnResult(sqlmock.NewResult(0, 0))
+		} else {
+			mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+				WillReturnError(errors.Errorf("ERROR %d (%s): %s", 1054, "42S22", "Unknown column '_tidb_rowid' in 'field list'"))
+			rows := sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE"})
+			for i := range handleColNames {
+				rows.AddRow(handleColNames[i], handleColTypes[i])
+			}
+			mock.ExpectQuery("SELECT c.COLUMN_NAME, DATA_TYPE FROM").WithArgs(database, table).WillReturnRows(rows)
+		}
+
+		rows := sqlmock.NewRows([]string{"START_KEY", "tidb_decode_key(START_KEY)"})
+		for _, regionResult := range regionResults {
+			rows.AddRow(regionResult...)
+		}
+		mock.ExpectQuery("SELECT START_KEY,tidb_decode_key\\(START_KEY\\) from INFORMATION_SCHEMA.TIKV_REGION_STATUS").
+			WithArgs(database, table).WillReturnRows(rows)
+
+		rows = sqlmock.NewRows([]string{"COLUMN_NAME", "EXTRA"})
+		for _, handleCol := range handleColNames {
+			rows.AddRow(handleCol, "")
+		}
+		mock.ExpectQuery("SELECT COLUMN_NAME,EXTRA FROM INFORMATION_SCHEMA.COLUMNS").WithArgs(database, table).
+			WillReturnRows(rows)
+
+		orderByClause := buildOrderByClauseString(handleColNames)
+		// special case, no enough value to split chunks
+		if len(regionResults) <= 1 {
+			mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+				WillReturnResult(sqlmock.NewResult(0, 0))
+			orderByClause = orderByTiDBRowID
+		}
+		c.Assert(d.concurrentDumpTable(tctx, conn, meta, taskChan), IsNil)
+		c.Assert(mock.ExpectationsWereMet(), IsNil)
+
+		for i, w := range testCase.expectedWhereClauses {
+			query := buildSelectQuery(database, table, "*", "", buildWhereCondition(d.conf, w), orderByClause)
+			task := <-taskChan
+			taskTableData, ok := task.(*TaskTableData)
+			c.Assert(ok, IsTrue)
+			c.Assert(taskTableData.ChunkIndex, Equals, i)
+			data, ok := taskTableData.Data.(*tableData)
+			c.Assert(ok, IsTrue)
+			c.Assert(data.query, Equals, query)
+		}
+	}
+}
+
+func (s *testSQLSuite) TestBuildRegionQueriesWithPartitions(c *C) {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	defer db.Close()
+	conn, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
+
+	d := &Dumper{
+		tctx:      tctx,
+		conf:      DefaultConfig(),
+		cancelCtx: cancel,
+	}
+	d.conf.ServerInfo = ServerInfo{
+		HasTiKV:       true,
+		ServerType:    ServerTypeTiDB,
+		ServerVersion: gcSafePointVersion,
+	}
+	database := "foo"
+	table := "bar"
+	partitions := []string{"p0", "p1", "p2"}
+
+	testCases := []struct {
+		regionResults        [][][]driver.Value
+		handleColNames       []string
+		handleColTypes       []string
+		expectedWhereClauses [][]string
+		hasTiDBRowID         bool
+		dumpWholeTable       bool
+	}{
+		{
+			[][][]driver.Value{
+				{
+					{6009, "t_121_i_1_0380000000000ea6010380000000000ea601", "t_121_", 6010, 1, 6010, 0, 0, 0, 74, 1052002},
+					{6011, "t_121_", "t_121_i_1_0380000000000ea6010380000000000ea601", 6012, 1, 6012, 0, 0, 0, 68, 972177},
+				},
+				{
+					{6015, "t_122_i_1_0380000000002d2a810380000000002d2a81", "t_122_", 6016, 1, 6016, 0, 0, 0, 77, 1092962},
+					{6017, "t_122_", "t_122_i_1_0380000000002d2a810380000000002d2a81", 6018, 1, 6018, 0, 0, 0, 66, 939975},
+				},
+				{
+					{6021, "t_123_i_1_0380000000004baf010380000000004baf01", "t_123_", 6022, 1, 6022, 0, 0, 0, 85, 1206726},
+					{6023, "t_123_", "t_123_i_1_0380000000004baf010380000000004baf01", 6024, 1, 6024, 0, 0, 0, 65, 927576},
+				},
+			},
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[][]string{
+				{""}, {""}, {""},
+			},
+			true,
+			true,
+		},
+		{
+			[][][]driver.Value{
+				{
+					{6009, "t_121_i_1_0380000000000ea6010380000000000ea601", "t_121_r_10001", 6010, 1, 6010, 0, 0, 0, 74, 1052002},
+					{6013, "t_121_r_10001", "t_121_r_970001", 6014, 1, 6014, 0, 0, 0, 75, 975908},
+					{6003, "t_121_r_970001", "t_122_", 6004, 1, 6004, 0, 0, 0, 79, 1022285},
+					{6011, "t_121_", "t_121_i_1_0380000000000ea6010380000000000ea601", 6012, 1, 6012, 0, 0, 0, 68, 972177},
+				},
+				{
+					{6015, "t_122_i_1_0380000000002d2a810380000000002d2a81", "t_122_r_2070760", 6016, 1, 6016, 0, 0, 0, 77, 1092962},
+					{6019, "t_122_r_2070760", "t_122_r_3047115", 6020, 1, 6020, 0, 0, 0, 75, 959650},
+					{6005, "t_122_r_3047115", "t_123_", 6006, 1, 6006, 0, 0, 0, 77, 992339},
+					{6017, "t_122_", "t_122_i_1_0380000000002d2a810380000000002d2a81", 6018, 1, 6018, 0, 0, 0, 66, 939975},
+				},
+				{
+					{6021, "t_123_i_1_0380000000004baf010380000000004baf01", "t_123_r_4186953", 6022, 1, 6022, 0, 0, 0, 85, 1206726},
+					{6025, "t_123_r_4186953", "t_123_r_5165682", 6026, 1, 6026, 0, 0, 0, 74, 951379},
+					{6007, "t_123_r_5165682", "t_124_", 6008, 1, 6008, 0, 0, 0, 71, 918488},
+					{6023, "t_123_", "t_123_i_1_0380000000004baf010380000000004baf01", 6024, 1, 6024, 0, 0, 0, 65, 927576},
+				},
+			},
+			[]string{"_tidb_rowid"},
+			[]string{"bigint"},
+			[][]string{
+				{
+					"`_tidb_rowid`<10001",
+					"`_tidb_rowid`>=10001 and `_tidb_rowid`<970001",
+					"`_tidb_rowid`>=970001",
+				},
+				{
+					"`_tidb_rowid`<2070760",
+					"`_tidb_rowid`>=2070760 and `_tidb_rowid`<3047115",
+					"`_tidb_rowid`>=3047115",
+				},
+				{
+					"`_tidb_rowid`<4186953",
+					"`_tidb_rowid`>=4186953 and `_tidb_rowid`<5165682",
+					"`_tidb_rowid`>=5165682",
+				},
+			},
+			true,
+			false,
+		},
+		{
+			[][][]driver.Value{
+				{
+					{6041, "t_134_", "t_134_r_960001", 6042, 1, 6042, 0, 0, 0, 69, 964987},
+					{6035, "t_134_r_960001", "t_135_", 6036, 1, 6036, 0, 0, 0, 75, 1052130},
+				},
+				{
+					{6043, "t_135_", "t_135_r_2960001", 6044, 1, 6044, 0, 0, 0, 69, 969576},
+					{6037, "t_135_r_2960001", "t_136_", 6038, 1, 6038, 0, 0, 0, 72, 1014464},
+				},
+				{
+					{6045, "t_136_", "t_136_r_4960001", 6046, 1, 6046, 0, 0, 0, 68, 957557},
+					{6039, "t_136_r_4960001", "t_137_", 6040, 1, 6040, 0, 0, 0, 75, 1051579},
+				},
+			},
+			[]string{"a"},
+			[]string{"bigint"},
+			[][]string{
+
+				{
+					"`a`<960001",
+					"`a`>=960001",
+				},
+				{
+					"`a`<2960001",
+					"`a`>=2960001",
+				},
+				{
+					"`a`<4960001",
+					"`a`>=4960001",
+				},
+			},
+			false,
+			false,
+		},
+	}
+
+	for i, testCase := range testCases {
+		c.Log(fmt.Sprintf("case #%d", i))
+		handleColNames := testCase.handleColNames
+		handleColTypes := testCase.handleColTypes
+		regionResults := testCase.regionResults
+
+		// Test build tasks through table region
+		taskChan := make(chan Task, 128)
+		quotaCols := make([]string, 0, len(handleColNames))
+		for _, col := range quotaCols {
+			quotaCols = append(quotaCols, wrapBackTicks(col))
+		}
+		selectFields := strings.Join(quotaCols, ",")
+		meta := &tableMeta{
+			database:      database,
+			table:         table,
+			selectedField: selectFields,
+			specCmts: []string{
+				"/*!40101 SET NAMES binary*/;",
+			},
+		}
+
+		rows := sqlmock.NewRows([]string{"PARTITION_NAME"})
+		for _, partition := range partitions {
+			rows.AddRow(partition)
+		}
+		mock.ExpectQuery("SELECT PARTITION_NAME from INFORMATION_SCHEMA.PARTITIONS").
+			WithArgs(database, table).WillReturnRows(rows)
+
+		if testCase.hasTiDBRowID {
+			mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+				WillReturnResult(sqlmock.NewResult(0, 0))
+		} else {
+			mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+				WillReturnError(errors.Errorf("ERROR %d (%s): %s", 1054, "42S22", "Unknown column '_tidb_rowid' in 'field list'"))
+			rows = sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE"})
+			for i := range handleColNames {
+				rows.AddRow(handleColNames[i], handleColTypes[i])
+			}
+			mock.ExpectQuery("SELECT c.COLUMN_NAME, DATA_TYPE FROM").WithArgs(database, table).WillReturnRows(rows)
+		}
+
+		for i, partition := range partitions {
+			rows = sqlmock.NewRows([]string{"REGION_ID", "START_KEY", "END_KEY", "LEADER_ID", "LEADER_STORE_ID", "PEERS", "SCATTERING", "WRITTEN_BYTES", "READ_BYTES", "APPROXIMATE_SIZE(MB)", "APPROXIMATE_KEYS"})
+			for _, regionResult := range regionResults[i] {
+				rows.AddRow(regionResult...)
+			}
+			mock.ExpectQuery(fmt.Sprintf("SHOW TABLE `%s`.`%s` PARTITION\\(`%s`\\) REGIONS", escapeString(database), escapeString(table), escapeString(partition))).
+				WillReturnRows(rows)
+		}
+
+		for range partitions {
+			rows = sqlmock.NewRows([]string{"COLUMN_NAME", "EXTRA"})
+			for _, handleCol := range handleColNames {
+				rows.AddRow(handleCol, "")
+			}
+			mock.ExpectQuery("SELECT COLUMN_NAME,EXTRA FROM INFORMATION_SCHEMA.COLUMNS").WithArgs(database, table).
+				WillReturnRows(rows)
+			// special case, dump whole table
+			if testCase.dumpWholeTable {
+				mock.ExpectExec(fmt.Sprintf("SELECT _tidb_rowid from `%s`.`%s` LIMIT 0", database, table)).
+					WillReturnResult(sqlmock.NewResult(0, 0))
+			}
+		}
+
+		orderByClause := buildOrderByClauseString(handleColNames)
+		c.Assert(d.concurrentDumpTable(tctx, conn, meta, taskChan), IsNil)
+		c.Assert(mock.ExpectationsWereMet(), IsNil)
+
+		chunkIdx := 0
+		for i, partition := range partitions {
+			for _, w := range testCase.expectedWhereClauses[i] {
+				query := buildSelectQuery(database, table, "*", partition, buildWhereCondition(d.conf, w), orderByClause)
+				task := <-taskChan
+				taskTableData, ok := task.(*TaskTableData)
+				c.Assert(ok, IsTrue)
+				c.Assert(taskTableData.ChunkIndex, Equals, chunkIdx)
+				data, ok := taskTableData.Data.(*tableData)
+				c.Assert(ok, IsTrue)
+				c.Assert(data.query, Equals, query)
+				chunkIdx++
+			}
+		}
 	}
 }
 
diff --git a/v4/export/writer_util_test.go b/v4/export/writer_util_test.go
index ae6453e..ff249c5 100644
--- a/v4/export/writer_util_test.go
+++ b/v4/export/writer_util_test.go
@@ -9,14 +9,27 @@ import (
 	"testing"
 
 	tcontext "github.com/pingcap/dumpling/v4/context"
+	"github.com/pingcap/dumpling/v4/log"
 
 	"github.com/pingcap/br/pkg/storage"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 )
 
+var appLogger log.Logger
+
 func TestT(t *testing.T) {
 	initColTypeRowReceiverMap()
+	logger, _, err := log.InitAppLogger(&log.Config{
+		Level:  "info",
+		File:   "",
+		Format: "text",
+	})
+	if err != nil {
+		t.Log("fail to init logger, err: " + err.Error())
+		t.Fail()
+	}
+	appLogger = logger
 	TestingT(t)
 }