Repository: pingcap/dumpling (go)

Issue:
Support `--where` clause
Support [the `--where` clause from mysqldump](https://dev.mysql.com/doc/refman/5.7/en/mysqldump.html#option_mysqldump_where). The condition is applied on all tables.

* [ ] The file size of the SQL dump should still be close to the configuration. If we dumped 1000 files of 256 MB each originally, and we use `--where` to filter out  half of the rows, the result should be 500 files of 256 MB each, not 1000 files of 128 MB each.
* [ ] If the expression cannot be evaluated, abort and report the error. 
* [ ] The condition used should be written out for SQL dumps (can be skipped for CSV and binary dumps) like:

    ```sql
    /* WHERE (id > 500) */
    INSERT INTO ...
    ```

* [ ] Even if all rows are filtered out, it should still generate an SQL file with content:

    ```sql
    /* All data are filtered out by this condition: */
    /* WHERE (id > 500) */
    ```

Should work just fine by adding the condition to `SELECT *`.

# Score 

* 600

Pull request description:
Support --rows and --where clause
try to resolve #9 #31 #12 
port from [mydumper](https://github.com/pingcap/mydumper/blob/master/mydumper.c)

Patch:
diff --git a/cmd/dumpling/main.go b/cmd/dumpling/main.go
index 665771f..2b5a2a6 100644
--- a/cmd/dumpling/main.go
+++ b/cmd/dumpling/main.go
@@ -40,6 +40,8 @@ var (
 	consistency   string
 	snapshot      string
 	noViews       bool
+	rows          uint64
+	where         string
 
 	rootCmd = &cobra.Command{
 		Use:   "dumpling",
@@ -71,6 +73,8 @@ func init() {
 	rootCmd.PersistentFlags().StringVar(&consistency, "consistency", "auto", "Consistency level during dumping: {auto|none|flush|lock|snapshot}")
 	rootCmd.PersistentFlags().StringVar(&snapshot, "snapshot", "", "Snapshot position. Valid only when consistency=snapshot")
 	rootCmd.PersistentFlags().BoolVarP(&noViews, "no-views", "W", true, "Do not dump views")
+	rootCmd.PersistentFlags().Uint64VarP(&rows, "rows", "r", export.UnspecifiedSize, "Split table into chunks of this many rows, default unlimited")
+	rootCmd.PersistentFlags().StringVar(&where, "where", "", "Dump only selected records")
 }
 
 func run() {
@@ -94,6 +98,8 @@ func run() {
 	conf.OutputDirPath = outputDir
 	conf.Consistency = consistency
 	conf.NoViews = noViews
+	conf.Rows = rows
+	conf.Where = where
 
 	err = export.Dump(conf)
 	if err != nil {
diff --git a/v4/export/config.go b/v4/export/config.go
index 210b880..9bad238 100644
--- a/v4/export/config.go
+++ b/v4/export/config.go
@@ -29,6 +29,8 @@ type Config struct {
 	NoViews       bool
 
 	BlackWhiteList BWListConf
+	Rows           uint64
+	Where          string
 }
 
 func DefaultConfig() *Config {
@@ -48,6 +50,8 @@ func DefaultConfig() *Config {
 		Snapshot:      "",
 		Consistency:   "auto",
 		NoViews:       true,
+		Rows:          UnspecifiedSize,
+		Where:         "",
 	}
 }
 
@@ -55,7 +59,10 @@ func (conf *Config) getDSN(db string) string {
 	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4", conf.User, conf.Password, conf.Host, conf.Port, db)
 }
 
-const UnspecifiedSize = 0
+const (
+	UnspecifiedSize    = 0
+	defaultDumpThreads = 128
+)
 
 type ServerInfo struct {
 	ServerType    ServerType
diff --git a/v4/export/dump.go b/v4/export/dump.go
index 0a1f6e7..4fc6645 100644
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -18,6 +18,8 @@ func Dump(conf *Config) (err error) {
 	}
 	defer pool.Close()
 
+	adjustConfig(conf)
+
 	conf.ServerInfo, err = detectServerInfo(pool)
 	if err != nil {
 		return err
@@ -124,13 +126,54 @@ func dumpTable(ctx context.Context, conf *Config, db *sql.DB, dbName string, tab
 		return err
 	}
 
+	if conf.Rows != UnspecifiedSize {
+		finished, err := concurrentDumpTable(ctx, writer, conf, db, dbName, tableName)
+		if err != nil || finished {
+			return err
+		}
+	}
 	tableIR, err := SelectAllFromTable(conf, db, dbName, tableName)
 	if err != nil {
 		return err
 	}
 
-	if err := writer.WriteTableData(ctx, tableIR); err != nil {
-		return err
+	return writer.WriteTableData(ctx, tableIR)
+}
+
+func concurrentDumpTable(ctx context.Context, writer Writer, conf *Config, db *sql.DB, dbName string, tableName string) (bool, error) {
+	// try dump table concurrently by split table to chunks
+	chunksIterCh := make(chan TableDataIR, defaultDumpThreads)
+	errCh := make(chan error, defaultDumpThreads)
+	linear := make(chan struct{})
+
+	ctx1, cancel1 := context.WithCancel(ctx)
+	defer cancel1()
+	var g errgroup.Group
+	g.Go(func() error {
+		splitTableDataIntoChunks(ctx1, chunksIterCh, errCh, linear, dbName, tableName, db, conf)
+		return nil
+	})
+
+Loop:
+	for {
+		select {
+		case <-ctx.Done():
+			return true, nil
+		case <-linear:
+			return false, nil
+		case chunksIter, ok := <-chunksIterCh:
+			if !ok {
+				break Loop
+			}
+			g.Go(func() error {
+				return writer.WriteTableData(ctx, chunksIter)
+			})
+		case err := <-errCh:
+			return false, err
+		}
 	}
-	return nil
+	if err := g.Wait(); err != nil {
+		return true, err
+	}
+	return true, nil
 }
diff --git a/v4/export/ir.go b/v4/export/ir.go
index 8bf9f74..e35b20e 100644
--- a/v4/export/ir.go
+++ b/v4/export/ir.go
@@ -6,6 +6,7 @@ import "database/sql"
 type TableDataIR interface {
 	DatabaseName() string
 	TableName() string
+	ChunkIndex() int
 	ColumnCount() uint
 	ColumnTypes() []string
 
@@ -19,6 +20,8 @@ type SQLRowIter interface {
 	HasNext() bool
 	HasNextSQLRowIter() bool
 	NextSQLRowIter() SQLRowIter
+	// release SQLRowIter
+	Close() error
 }
 
 type RowReceiverStringer interface {
diff --git a/v4/export/ir_impl.go b/v4/export/ir_impl.go
index 01e3482..c915461 100644
--- a/v4/export/ir_impl.go
+++ b/v4/export/ir_impl.go
@@ -1,7 +1,15 @@
 package export
 
 import (
+	"context"
 	"database/sql"
+	"fmt"
+	"strconv"
+
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+
+	"github.com/pingcap/dumpling/v4/log"
 )
 
 // rowIter implements the SQLRowIter interface.
@@ -22,6 +30,10 @@ func newRowIter(rows *sql.Rows, argLen int) *rowIter {
 	return r
 }
 
+func (iter *rowIter) Close() error {
+	return iter.rows.Close()
+}
+
 func (iter *rowIter) Next(row RowReceiver) error {
 	err := decodeFromRows(iter.rows, iter.args, row)
 	iter.hasNext = iter.rows.Next()
@@ -49,6 +61,10 @@ type fileRowIter struct {
 	currentFileSize      uint64
 }
 
+func (c *fileRowIter) Close() error {
+	return c.rowIter.Close()
+}
+
 func (c *fileRowIter) Next(row RowReceiver) error {
 	err := c.rowIter.Next(row)
 	if err != nil {
@@ -115,11 +131,12 @@ func (m *stringIter) HasNext() bool {
 }
 
 type tableData struct {
-	database string
-	table    string
-	rows     *sql.Rows
-	colTypes []*sql.ColumnType
-	specCmts []string
+	database   string
+	table      string
+	chunkIndex int
+	rows       *sql.Rows
+	colTypes   []*sql.ColumnType
+	specCmts   []string
 }
 
 func (td *tableData) ColumnTypes() []string {
@@ -138,6 +155,10 @@ func (td *tableData) TableName() string {
 	return td.table
 }
 
+func (td *tableData) ChunkIndex() int {
+	return td.chunkIndex
+}
+
 func (td *tableData) ColumnCount() uint {
 	return uint(len(td.colTypes))
 }
@@ -169,7 +190,7 @@ func (t *tableDataChunks) Rows() SQLRowIter {
 	}
 }
 
-func splitTableDataIntoChunks(td TableDataIR, chunkSize uint64, statementSize uint64) *tableDataChunks {
+func buildChunksIter(td TableDataIR, chunkSize uint64, statementSize uint64) *tableDataChunks {
 	return &tableDataChunks{
 		TableDataIR:        td,
 		chunkSizeLimit:     chunkSize,
@@ -177,6 +198,117 @@ func splitTableDataIntoChunks(td TableDataIR, chunkSize uint64, statementSize ui
 	}
 }
 
+func splitTableDataIntoChunks(
+	ctx context.Context,
+	tableDataIRCh chan TableDataIR,
+	errCh chan error,
+	linear chan struct{},
+	dbName, tableName string, db *sql.DB, conf *Config) {
+	field, err := pickupPossibleField(dbName, tableName, db, conf)
+	if err != nil {
+		errCh <- withStack(err)
+		return
+	}
+	if field == "" {
+		// skip split chunk logic if not found proper field
+		log.Zap().Debug("skip concurrent dump due to no proper field", zap.String("field", field))
+		linear <- struct{}{}
+		return
+	}
+
+	query := fmt.Sprintf("SELECT MIN(`%s`),MAX(`%s`) FROM `%s`.`%s` ",
+		field, field, dbName, tableName)
+	if conf.Where != "" {
+		query = fmt.Sprintf("%s WHERE %s", query, conf.Where)
+	}
+	log.Zap().Debug("split chunks", zap.String("query", query))
+
+	var smin sql.NullString
+	var smax sql.NullString
+	row := db.QueryRow(query)
+	err = row.Scan(&smin, &smax)
+	if err != nil {
+		log.Zap().Error("split chunks - get max min failed", zap.String("query", query), zap.Error(err))
+		errCh <- withStack(err)
+		return
+	}
+	if !smax.Valid || !smin.Valid {
+		// found no data
+		log.Zap().Warn("no data to dump", zap.String("schema", dbName), zap.String("table", tableName))
+		close(tableDataIRCh)
+		return
+	}
+
+	var max uint64
+	var min uint64
+	if max, err = strconv.ParseUint(smax.String, 10, 64); err != nil {
+		errCh <- errors.WithMessagef(err, "fail to convert max value %s in query %s", smax.String, query)
+		return
+	}
+	if min, err = strconv.ParseUint(smin.String, 10, 64); err != nil {
+		errCh <- errors.WithMessagef(err, "fail to convert min value %s in query %s", smin.String, query)
+		return
+	}
+
+	count := estimateCount(dbName, tableName, db, field, conf)
+	if count < conf.Rows {
+		// skip chunk logic if estimates are low
+		log.Zap().Debug("skip concurrent dump due to estimate count < rows",
+			zap.Uint64("estimate count", count),
+			zap.Uint64("conf.rows", conf.Rows),
+		)
+		linear <- struct{}{}
+		return
+	}
+
+	// every chunk would have eventual adjustments
+	estimatedChunks := count / conf.Rows
+	estimatedStep := (max-min)/estimatedChunks + 1
+	cutoff := min
+
+	colTypes, err := GetColumnTypes(db, dbName, tableName)
+	if err != nil {
+		errCh <- withStack(err)
+		return
+	}
+	orderByClause, err := buildOrderByClause(conf, db, dbName, tableName)
+	if err != nil {
+		errCh <- withStack(err)
+		return
+	}
+
+	chunkIndex := 0
+LOOP:
+	for cutoff <= max {
+		chunkIndex += 1
+		where := fmt.Sprintf("(`%s` >= %d AND `%s` < %d)", field, cutoff, field, cutoff+estimatedStep)
+		query = buildSelectQuery(dbName, tableName, buildWhereCondition(conf, where), orderByClause)
+		rows, err := db.Query(query)
+		if err != nil {
+			errCh <- errors.WithMessage(err, query)
+			return
+		}
+
+		td := &tableData{
+			database:   dbName,
+			table:      tableName,
+			rows:       rows,
+			chunkIndex: chunkIndex,
+			colTypes:   colTypes,
+			specCmts: []string{
+				"/*!40101 SET NAMES binary*/;",
+			},
+		}
+		cutoff += estimatedStep
+		select {
+		case <-ctx.Done():
+			break LOOP
+		case tableDataIRCh <- td:
+		}
+	}
+	close(tableDataIRCh)
+}
+
 type metaData struct {
 	target   string
 	metaSQL  string
diff --git a/v4/export/prepare.go b/v4/export/prepare.go
index 944181f..f480d4e 100644
--- a/v4/export/prepare.go
+++ b/v4/export/prepare.go
@@ -7,6 +7,13 @@ import (
 	"github.com/pingcap/dumpling/v4/log"
 )
 
+func adjustConfig(conf *Config) {
+	if conf.Rows != UnspecifiedSize {
+		// Disable filesize if rows was set
+		conf.FileSize = UnspecifiedSize
+	}
+}
+
 func detectServerInfo(db *sql.DB) (ServerInfo, error) {
 	versionStr, err := SelectVersion(db)
 	if err != nil {
diff --git a/v4/export/sql.go b/v4/export/sql.go
index 1c9c220..91c3fb1 100644
--- a/v4/export/sql.go
+++ b/v4/export/sql.go
@@ -3,9 +3,13 @@ package export
 import (
 	"database/sql"
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/pkg/errors"
+	"go.uber.org/zap"
+
+	"github.com/pingcap/dumpling/v4/log"
 )
 
 func ShowDatabases(db *sql.DB) ([]string, error) {
@@ -100,10 +104,12 @@ func SelectAllFromTable(conf *Config, db *sql.DB, database, table string) (Table
 		return nil, err
 	}
 
-	query, err := buildSelectAllQuery(conf, db, database, table)
+	orderByClause, err := buildOrderByClause(conf, db, database, table)
 	if err != nil {
 		return nil, err
 	}
+
+	query := buildSelectQuery(database, table, buildWhereCondition(conf, ""), orderByClause)
 	rows, err := db.Query(query)
 	if err != nil {
 		return nil, withStack(errors.WithMessage(err, query))
@@ -120,30 +126,34 @@ func SelectAllFromTable(conf *Config, db *sql.DB, database, table string) (Table
 	}, nil
 }
 
-func buildSelectAllQuery(conf *Config, db *sql.DB, database, table string) (string, error) {
+func buildSelectQuery(database, table string, where string, orderByClause string) string {
 	var query strings.Builder
 	query.WriteString("SELECT * FROM ")
 	query.WriteString(database)
 	query.WriteString(".")
 	query.WriteString(table)
-	if conf.SortByPk {
-		orderByClause, err := buildOrderByClause(conf, db, database, table)
-		if err != nil {
-			return "", err
-		}
-		if orderByClause != "" {
-			query.WriteString(" ")
-			query.WriteString(orderByClause)
-		}
+
+	if where != "" {
+		query.WriteString(" ")
+		query.WriteString(where)
+	}
+
+	if orderByClause != "" {
+		query.WriteString(" ")
+		query.WriteString(orderByClause)
 	}
-	return query.String(), nil
+
+	return query.String()
 }
 
 func buildOrderByClause(conf *Config, db *sql.DB, database, table string) (string, error) {
+	if !conf.SortByPk {
+		return "", nil
+	}
 	if conf.ServerInfo.ServerType == ServerTypeTiDB {
 		ok, err := SelectTiDBRowID(db, database, table)
 		if err != nil {
-			return "", err
+			return "", withStack(err)
 		}
 		if ok {
 			return "ORDER BY _tidb_rowid", nil
@@ -153,7 +163,7 @@ func buildOrderByClause(conf *Config, db *sql.DB, database, table string) (strin
 	}
 	pkName, err := GetPrimaryKeyName(db, database, table)
 	if err != nil {
-		return "", err
+		return "", withStack(err)
 	}
 	tableContainsPriKey := pkName != ""
 	if tableContainsPriKey {
@@ -209,6 +219,21 @@ func GetPrimaryKeyName(db *sql.DB, database, table string) (string, error) {
 	return colName, nil
 }
 
+func GetUniqueIndexName(db *sql.DB, database, table string) (string, error) {
+	uniKeyQuery := "SELECT column_name FROM information_schema.columns "+
+		"WHERE table_schema = ? AND table_name = ? AND column_key = 'UNI';"
+	var colName string
+	row := db.QueryRow(uniKeyQuery, database, table)
+	if err := row.Scan(&colName); err != nil {
+		if err == sql.ErrNoRows {
+			return "", nil
+		} else {
+			return "", withStack(errors.WithMessage(err, uniKeyQuery))
+		}
+	}
+	return colName, nil
+}
+
 func FlushTableWithReadLock(db *sql.DB) error {
 	_, err := db.Exec("FLUSH TABLES WITH READ LOCK")
 	return withStack(err)
@@ -277,3 +302,138 @@ func simpleQuery(db *sql.DB, sql string, handleOneRow func(*sql.Rows) error) err
 	}
 	return nil
 }
+
+func pickupPossibleField(dbName, tableName string, db *sql.DB, conf *Config) (string, error) {
+	// If detected server is TiDB, try using _tidb_rowid
+	if conf.ServerInfo.ServerType == ServerTypeTiDB {
+		ok, err := SelectTiDBRowID(db, dbName, tableName)
+		if err != nil {
+			return "", nil
+		}
+		if ok {
+			return "_tidb_rowid", nil
+		}
+	}
+	// try to use pk
+	fieldName, err := GetPrimaryKeyName(db, dbName, tableName)
+	if err != nil {
+		return "", err
+	}
+	// try to use first uniqueIndex
+	if fieldName == "" {
+		fieldName, err = GetUniqueIndexName(db, dbName, tableName)
+		if err != nil {
+			return "", err
+		}
+	}
+
+	// there is no proper index
+	if fieldName == "" {
+		return "", nil
+	}
+
+	query := "SELECT DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS "+
+		"WHERE TABLE_NAME = ? AND COLUMN_NAME = ?"
+	var fieldType string
+	row := db.QueryRow(query, tableName, fieldName)
+	err = row.Scan(&fieldType)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return "", nil
+		} else {
+			return "", withStack(errors.WithMessage(err, query))
+		}
+	}
+	switch strings.ToLower(fieldType) {
+	case "int", "bigint":
+		return fieldName, nil
+	}
+	return "", nil
+}
+
+func estimateCount(dbName, tableName string, db *sql.DB, field string, conf *Config) uint64 {
+	query := fmt.Sprintf("EXPLAIN SELECT `%s` FROM `%s`.`%s`", field, dbName, tableName)
+
+	if conf.Where != "" {
+		query += " WHERE "
+		query += conf.Where
+	}
+
+	var estRows int
+	if conf.ServerInfo.ServerType == ServerTypeTiDB {
+		/* tidb results field name is estrows
+		+-----------------------+----------+-----------+-----------------------------------------+
+		| id                    | estrows  | task      | operator info                           |
+		+-----------------------+----------+-----------+-----------------------------------------+
+		| tablereader_5         | 10000.00 | root      | data:tablefullscan_4                    |
+		| └─tablefullscan_4     | 10000.00 | cop[tikv] | table:a, keep order:false, stats:pseudo |
+		+-----------------------+----------+-----------+----------------------------------------
+		*/
+		estRows = detectEstimateRows(db, query, "tidb", 1, 4)
+	} else if conf.ServerInfo.ServerType == ServerTypeMariaDB {
+		/* mariadb result field name is rows
+		+------+-------------+---------+-------+---------------+------+---------+------+----------+-------------+
+		| id   | select_type | table   | type  | possible_keys | key  | key_len | ref  | rows     | Extra       |
+		+------+-------------+---------+-------+---------------+------+---------+------+----------+-------------+
+		|    1 | SIMPLE      | sbtest1 | index | NULL          | k_1  | 4       | NULL | 15000049 | Using index |
+		+------+-------------+---------+-------+---------------+------+---------+------+----------+-------------+
+		*/
+		estRows = detectEstimateRows(db, query, "mariadb", 8, 10)
+	} else if conf.ServerInfo.ServerType == ServerTypeMySQL {
+		/* mysql result field name is rows
+		+----+-------------+-------+------------+-------+---------------+-----------+---------+------+------+----------+-------------+
+		| id | select_type | table | partitions | type  | possible_keys | key       | key_len | ref  | rows | filtered | Extra       |
+		+----+-------------+-------+------------+-------+---------------+-----------+---------+------+------+----------+-------------+
+		|  1 | SIMPLE      | t1    | NULL       | index | NULL          | multi_col | 10      | NULL |    5 |   100.00 | Using index |
+		+----+-------------+-------+------------+-------+---------------+-----------+---------+------+------+----------+-------------+
+		*/
+		estRows = detectEstimateRows(db, query, "mysql", 9, 12)
+	}
+	if estRows > 0 {
+		return uint64(estRows)
+	}
+	return 0
+}
+
+func detectEstimateRows(db *sql.DB, query string, dbType string, fieldIndex int, fieldLength int) int {
+	oneRow := make([]sql.NullString, fieldLength)
+	addr := make([]interface{}, fieldLength)
+	for i := range oneRow{
+		addr[i] = &oneRow[i]
+	}
+	row := db.QueryRow(query)
+	err := row.Scan(addr...)
+	if err != nil {
+		log.Zap().Warn("can't get estimate count from db",
+			zap.String("db", dbType),
+			zap.String("query", query), zap.Error(err))
+		return 0
+	}
+	estRows, err := strconv.ParseFloat(oneRow[fieldIndex].String, 64)
+	if err != nil {
+		log.Zap().Warn("can't get parse rows from db",
+			zap.String("db", dbType),
+			zap.String("query", query), zap.Error(err))
+		return 0
+	}
+	return int(estRows)
+}
+
+func buildWhereCondition(conf *Config, where string) string {
+	var query strings.Builder
+	separator := "WHERE"
+	if conf.Where != "" {
+		query.WriteString(" ")
+		query.WriteString(separator)
+		query.WriteString(" ")
+		query.WriteString(conf.Where)
+		separator = "AND"
+	}
+	if where != "" {
+		query.WriteString(" ")
+		query.WriteString(separator)
+		query.WriteString(" ")
+		query.WriteString(where)
+	}
+	return query.String()
+}
diff --git a/v4/export/writer.go b/v4/export/writer.go
index ff2018b..15d49b1 100644
--- a/v4/export/writer.go
+++ b/v4/export/writer.go
@@ -6,8 +6,9 @@ import (
 	"os"
 	"path"
 
-	"github.com/pingcap/dumpling/v4/log"
 	"go.uber.org/zap"
+
+	"github.com/pingcap/dumpling/v4/log"
 )
 
 type Writer interface {
@@ -40,18 +41,24 @@ func (f *SimpleWriter) WriteTableMeta(ctx context.Context, db, table, createSQL
 func (f *SimpleWriter) WriteTableData(ctx context.Context, ir TableDataIR) error {
 	log.Zap().Debug("start dumping table...", zap.String("table", ir.TableName()))
 
-	chunks := splitTableDataIntoChunks(ir, f.cfg.FileSize, f.cfg.StatementSize)
-	chunkCount := 0
-	fileName := fmt.Sprintf("%s.%s.sql", ir.DatabaseName(), ir.TableName())
-	if f.cfg.FileSize != UnspecifiedSize {
-		fileName = fmt.Sprintf("%s.%s.%d.sql", ir.DatabaseName(), ir.TableName(), chunkCount)
-	}
+	chunkIndex := ir.ChunkIndex()
+	fileName := fmt.Sprintf("%s.%s.%d.sql", ir.DatabaseName(), ir.TableName(), ir.ChunkIndex())
+	// just let `database.table.sql` be `database.table.0.sql`
+	/*if fileName == "" {
+		// set initial file name
+		fileName = fmt.Sprintf("%s.%s.sql", ir.DatabaseName(), ir.TableName())
+		if f.cfg.FileSize != UnspecifiedSize {
+			fileName = fmt.Sprintf("%s.%s.%d.sql", ir.DatabaseName(), ir.TableName(), 0)
+		}
+	}*/
+	chunksIter := buildChunksIter(ir, f.cfg.FileSize, f.cfg.StatementSize)
+	defer chunksIter.Rows().Close()
 
 	for {
 		filePath := path.Join(f.cfg.OutputDirPath, fileName)
 		fileWriter, tearDown := buildLazyFileWriter(filePath)
 		intWriter := &InterceptStringWriter{StringWriter: fileWriter}
-		err := WriteInsert(chunks, intWriter)
+		err := WriteInsert(chunksIter, intWriter)
 		tearDown()
 		if err != nil {
 			return err
@@ -64,8 +71,8 @@ func (f *SimpleWriter) WriteTableData(ctx context.Context, ir TableDataIR) error
 		if f.cfg.FileSize == UnspecifiedSize {
 			break
 		}
-		chunkCount += 1
-		fileName = fmt.Sprintf("%s.%s.%d.sql", ir.DatabaseName(), ir.TableName(), chunkCount)
+		chunkIndex += 1
+		fileName = fmt.Sprintf("%s.%s.%d.sql", ir.DatabaseName(), ir.TableName(), chunkIndex)
 	}
 	log.Zap().Debug("dumping table successfully",
 		zap.String("table", ir.TableName()))


Test patch:
diff --git a/tests/naughty_strings/run.sh b/tests/naughty_strings/run.sh
index 4a96c9c..8fe5412 100755
--- a/tests/naughty_strings/run.sh
+++ b/tests/naughty_strings/run.sh
@@ -9,4 +9,4 @@ run_sql_file "$DUMPLING_BASE_NAME/data/naughty_strings.t-schema.sql"
 run_sql_file "$DUMPLING_BASE_NAME/data/naughty_strings.t.sql"
 run_dumpling
 # FIXME should compare the schemas too, but they differ too much among MySQL versions.
-diff "$DUMPLING_BASE_NAME/data/naughty_strings.t.sql" "$DUMPLING_OUTPUT_DIR/naughty_strings.t.sql"
+diff "$DUMPLING_BASE_NAME/data/naughty_strings.t.sql" "$DUMPLING_OUTPUT_DIR/naughty_strings.t.0.sql"
diff --git a/tests/primary_key/run.sh b/tests/primary_key/run.sh
index f2e25a0..04cb9f2 100644
--- a/tests/primary_key/run.sh
+++ b/tests/primary_key/run.sh
@@ -16,6 +16,6 @@ for file_path in "$DUMPLING_BASE_NAME"/data/*; do
   base_name=$(basename "$file_path")
   table_name="${base_name%.sql}"
   file_should_exist "$DUMPLING_BASE_NAME/result/$table_name.sql"
-  file_should_exist "$DUMPLING_OUTPUT_DIR/primary_key.$table_name.sql"
-  diff "$DUMPLING_BASE_NAME/result/$table_name.sql" "$DUMPLING_OUTPUT_DIR/primary_key.$table_name.sql"
+  file_should_exist "$DUMPLING_OUTPUT_DIR/primary_key.$table_name.0.sql"
+  diff "$DUMPLING_BASE_NAME/result/$table_name.sql" "$DUMPLING_OUTPUT_DIR/primary_key.$table_name.0.sql"
 done
diff --git a/tests/rows/conf/diff_config.toml b/tests/rows/conf/diff_config.toml
new file mode 100644
index 0000000..96c3c2a
--- /dev/null
+++ b/tests/rows/conf/diff_config.toml
@@ -0,0 +1,42 @@
+# diff Configuration.
+
+log-level = "info"
+
+chunk-size = 1000
+
+check-thread-count = 4
+
+sample-percent = 100
+
+use-rowid = false
+
+use-checksum = true
+
+fix-sql-file = "fix.sql"
+
+# tables need to check.
+[[check-tables]]
+schema = "rows"
+tables = ["~t.*"]
+
+[[table-config]]
+schema = "rows"
+table = "t"
+
+[[table-config.source-tables]]
+instance-id = "source-1"
+schema = "rows"
+table  = "t"
+
+[[source-db]]
+host = "127.0.0.1"
+port = 3306
+user = "root"
+password = ""
+instance-id = "source-1"
+
+[target-db]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+password = ""
diff --git a/tests/rows/conf/lightning.toml b/tests/rows/conf/lightning.toml
new file mode 100644
index 0000000..322049b
--- /dev/null
+++ b/tests/rows/conf/lightning.toml
@@ -0,0 +1,20 @@
+### tidb-lightning config
+
+[lightning]
+server-mode = false
+level = "error"
+check-requirements = false
+
+[tikv-importer]
+backend="tidb"
+on-duplicate = "error"
+
+[mydumper]
+data-source-dir = "/tmp/dumpling_test_result/sql_res.rows"
+
+[tidb]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+password = ""
+status-port = 10080
diff --git a/tests/rows/run.sh b/tests/rows/run.sh
new file mode 100644
index 0000000..add1490
--- /dev/null
+++ b/tests/rows/run.sh
@@ -0,0 +1,43 @@
+#!/bin/sh
+
+set -eu
+cur=$(cd `dirname $0`; pwd)
+
+DB_NAME="rows"
+TABLE_NAME="t"
+
+# drop database on tidb
+export DUMPLING_TEST_PORT=4000
+run_sql "drop database if exists $DB_NAME;"
+
+# drop database on mysql
+export DUMPLING_TEST_PORT=3306
+run_sql "drop database if exists $DB_NAME;"
+
+# build data on mysql
+run_sql "create database $DB_NAME;"
+run_sql "create table $DB_NAME.$TABLE_NAME (id int not null auto_increment primary key, a varchar(8));"
+
+# insert 100 records
+run_sql "insert into $DB_NAME.$TABLE_NAME (a) values $(seq -s, 100 | sed 's/,*$//g' | sed "s/[0-9]*/('1111_')/g");"
+
+# dumping
+export DUMPLING_TEST_DATABASE=$DB_NAME
+run_dumpling --rows 10 --loglevel debug
+
+# the dumping result is expected to be:
+# 10 files for insertion
+# FIXME the result of EXPLAIN SELECT `id` FROM `rows`.`t` randomly equal to 1 or 100, this could affect on file num.
+# file_num=$(find "$DUMPLING_OUTPUT_DIR" -maxdepth 1 -iname "$DB_NAME.$TABLE_NAME.*.sql" | wc -l)
+# if [ "$file_num" -ne 10 ]; then
+#   echo "obtain file number: $file_num, but expect: 10" && exit 1
+# fi
+
+cat "$cur/conf/lightning.toml"
+# use lightning import data to tidb
+run_lightning $cur/conf/lightning.toml
+
+# check mysql and tidb data
+check_sync_diff $cur/conf/diff_config.toml
+
+
diff --git a/v4/export/dump_test.go b/v4/export/dump_test.go
index 213e675..e7c9cc0 100644
--- a/v4/export/dump_test.go
+++ b/v4/export/dump_test.go
@@ -3,6 +3,7 @@ package export
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
@@ -98,9 +99,59 @@ func (s *testDumpSuite) TestDumpTable(c *C) {
 	c.Assert(specCmts.HasNext(), IsTrue)
 	c.Assert(specCmts.Next(), Equals, "/*!40101 SET NAMES binary*/;")
 	c.Assert(specCmts.HasNext(), IsFalse)
-	c.Assert(tbDataRes.Rows().HasNext(), Equals, true)
+	rowIter := tbDataRes.Rows()
+	c.Assert(rowIter.HasNext(), IsTrue)
 	receiver := newSimpleRowReceiver(1)
-	c.Assert(tbDataRes.Rows().Next(receiver), IsNil)
+	c.Assert(rowIter.Next(receiver), IsNil)
+	c.Assert(receiver.data[0], Equals, "1")
+	c.Assert(rowIter.Next(receiver), IsNil)
 	c.Assert(receiver.data[0], Equals, "2")
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 }
+
+func (s *testDumpSuite) TestDumpTableWhereClause(c *C) {
+	mockConfig := DefaultConfig()
+	mockConfig.SortByPk = false
+	mockConfig.Where = "a > 3 and a < 9"
+	db, mock, err := sqlmock.New()
+	c.Assert(err, IsNil)
+
+	showCreateTableResult := "CREATE TABLE t (a INT)"
+	rows := mock.NewRows([]string{"Table", "Create Table"}).AddRow("t", showCreateTableResult)
+	mock.ExpectQuery("SHOW CREATE TABLE test.t").WillReturnRows(rows)
+
+	rows = mock.NewRows([]string{"a"}).AddRow(1)
+	mock.ExpectQuery("SELECT (.) FROM test.t LIMIT 1").WillReturnRows(rows)
+
+	rows = mock.NewRows([]string{"a"})
+	for i := 4; i < 9; i++ {
+		rows.AddRow(i)
+	}
+	mock.ExpectQuery("SELECT (.) FROM test.t WHERE a > 3 and a < 9").WillReturnRows(rows)
+
+	mockWriter := newMockWriter()
+	err = dumpTable(context.Background(), mockConfig, db, "test", &TableInfo{Name: "t"}, mockWriter)
+	c.Assert(err, IsNil)
+
+	c.Assert(mockWriter.tableMeta["test.t"], Equals, showCreateTableResult)
+	c.Assert(len(mockWriter.tableData), Equals, 1)
+	tbDataRes := mockWriter.tableData[0]
+	c.Assert(tbDataRes.DatabaseName(), Equals, "test")
+	c.Assert(tbDataRes.TableName(), Equals, "t")
+	c.Assert(tbDataRes.ColumnCount(), Equals, uint(1))
+	specCmts := tbDataRes.SpecialComments()
+	c.Assert(specCmts.HasNext(), IsTrue)
+	c.Assert(specCmts.Next(), Equals, "/*!40101 SET NAMES binary*/;")
+	c.Assert(specCmts.HasNext(), IsFalse)
+	rowIter := tbDataRes.Rows()
+	c.Assert(rowIter.HasNext(), IsTrue)
+	receiver := newSimpleRowReceiver(1)
+
+	for i := 4; i < 9; i++ {
+		c.Assert(rowIter.HasNext(), IsTrue)
+		c.Assert(rowIter.Next(receiver), IsNil)
+		c.Assert(receiver.data[0], Equals, strconv.Itoa(i))
+	}
+	c.Assert(tbDataRes.Rows().HasNext(), IsFalse)
+	c.Assert(mock.ExpectationsWereMet(), IsNil)
+}
diff --git a/v4/export/ir_impl_test.go b/v4/export/ir_impl_test.go
index 9e002e3..9fd3f35 100644
--- a/v4/export/ir_impl_test.go
+++ b/v4/export/ir_impl_test.go
@@ -83,10 +83,10 @@ func (s *testIRImplSuite) TestChunkRowIter(c *C) {
 		testStatementSize uint64 = 101
 
 		expectedSize = [][]uint64{
-			[]uint64{50, 50},
-			[]uint64{100, 100},
-			[]uint64{150, 150},
-			[]uint64{200, 50},
+			{50, 50},
+			{100, 100},
+			{150, 150},
+			{200, 50},
 		}
 	)
 
diff --git a/v4/export/sql_test.go b/v4/export/sql_test.go
index 5dcb45c..10856cb 100644
--- a/v4/export/sql_test.go
+++ b/v4/export/sql_test.go
@@ -64,13 +64,17 @@ func (s *testDumpSuite) TestBuildSelectAllQuery(c *C) {
 	// _tidb_rowid is available.
 	mock.ExpectExec("SELECT _tidb_rowid from test.t").
 		WillReturnResult(sqlmock.NewResult(0, 0))
-	q, err := buildSelectAllQuery(mockConf, db, "test", "t")
+	orderByClause, err := buildOrderByClause(mockConf, db, "test", "t")
+	c.Assert(err, IsNil)
+	q := buildSelectQuery("test", "t", "", orderByClause)
 	c.Assert(q, Equals, "SELECT * FROM test.t ORDER BY _tidb_rowid")
 
 	// _tidb_rowid is unavailable, or PKIsHandle.
 	mock.ExpectExec("SELECT _tidb_rowid from test.t").
 		WillReturnError(errors.New(`1054, "Unknown column '_tidb_rowid' in 'field list'"`))
-	q, err = buildSelectAllQuery(mockConf, db, "test", "t")
+	orderByClause, err = buildOrderByClause(mockConf, db, "test", "t")
+	c.Assert(err, IsNil)
+	q = buildSelectQuery("test", "t", "", orderByClause)
 	c.Assert(q, Equals, "SELECT * FROM test.t")
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 
@@ -84,9 +88,9 @@ func (s *testDumpSuite) TestBuildSelectAllQuery(c *C) {
 		mock.ExpectPrepare("SELECT column_name FROM information_schema.columns").
 			ExpectQuery().WithArgs("test", "t").
 			WillReturnRows(sqlmock.NewRows([]string{"column_name"}).AddRow("id"))
-
-		q, err := buildSelectAllQuery(mockConf, db, "test", "t")
+		orderByClause, err := buildOrderByClause(mockConf, db, "test", "t")
 		c.Assert(err, IsNil, cmt)
+		q = buildSelectQuery("test", "t", "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM test.t ORDER BY id", cmt)
 		err = mock.ExpectationsWereMet()
 		c.Assert(err, IsNil, cmt)
@@ -101,8 +105,9 @@ func (s *testDumpSuite) TestBuildSelectAllQuery(c *C) {
 			ExpectQuery().WithArgs("test", "t").
 			WillReturnRows(sqlmock.NewRows([]string{"column_name"}))
 
-		q, err := buildSelectAllQuery(mockConf, db, "test", "t")
+		orderByClause, err := buildOrderByClause(mockConf, db, "test", "t")
 		c.Assert(err, IsNil, cmt)
+		q := buildSelectQuery("test", "t", "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM test.t", cmt)
 		err = mock.ExpectationsWereMet()
 		c.Assert(err, IsNil, cmt)
@@ -115,8 +120,7 @@ func (s *testDumpSuite) TestBuildSelectAllQuery(c *C) {
 		mockConf.ServerInfo.ServerType = ServerType(tp)
 		cmt := Commentf("current server type: ", tp)
 
-		q, err := buildSelectAllQuery(mockConf, db, "test", "t")
-		c.Assert(err, IsNil, cmt)
+		q := buildSelectQuery("test", "t", "", "")
 		c.Assert(q, Equals, "SELECT * FROM test.t", cmt)
 		c.Assert(mock.ExpectationsWereMet(), IsNil, cmt)
 	}
diff --git a/v4/export/test_util.go b/v4/export/test_util.go
index e937855..40cf50b 100644
--- a/v4/export/test_util.go
+++ b/v4/export/test_util.go
@@ -75,11 +75,12 @@ func newMockMetaIR(targetName string, meta string, specialComments []string) Met
 }
 
 type mockTableIR struct {
-	dbName   string
-	tblName  string
-	data     [][]driver.Value
-	specCmt  []string
-	colTypes []string
+	dbName    string
+	tblName   string
+	chunIndex int
+	data      [][]driver.Value
+	specCmt   []string
+	colTypes  []string
 }
 
 func (m *mockTableIR) DatabaseName() string {
@@ -90,6 +91,10 @@ func (m *mockTableIR) TableName() string {
 	return m.tblName
 }
 
+func (m *mockTableIR) ChunkIndex() int {
+	return m.chunIndex
+}
+
 func (m *mockTableIR) ColumnCount() uint {
 	return uint(len(m.colTypes))
 }
diff --git a/v4/export/writer_test.go b/v4/export/writer_test.go
index 52465bd..3e6b8d3 100644
--- a/v4/export/writer_test.go
+++ b/v4/export/writer_test.go
@@ -84,7 +84,7 @@ func (s *testDumpSuite) TestWriteTableData(c *C) {
 	err = writer.WriteTableData(ctx, tableIR)
 	c.Assert(err, IsNil)
 
-	p := path.Join(dir, "test.employee.sql")
+	p := path.Join(dir, "test.employee.0.sql")
 	_, err = os.Stat(p)
 	c.Assert(err, IsNil)
 	bytes, err := ioutil.ReadFile(p)
@@ -181,7 +181,7 @@ func (s *testDumpSuite) TestWriteTableDataWithStatementSize(c *C) {
 
 	// only with statement size
 	cases := map[string]string{
-		"test.employee.sql": "/*!40101 SET NAMES binary*/;\n" +
+		"test.employee.0.sql": "/*!40101 SET NAMES binary*/;\n" +
 			"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
 			"INSERT INTO `employee` VALUES\n" +
 			"(1, 'male', 'bob@mail.com', '020-1234', NULL),\n" +