Repository: pingcap/dumpling
Pull request #205 (2020-11-27 10:54:19)

refactor writer.go and support concurrent dump by tablesample
<!--
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 problem does this PR solve? <!--add issue link with summary if exists-->

This PR utilize the `TABLESAMPLE` feature provided by https://github.com/pingcap/tidb/pull/20883 to implement concurrent dumping on a single table.

There are 2 ways to dump tables: `sequentialDumpTable` and `concurrentDumpTable`. For the TiDBs with version >= `5.0.0`(which contain the `TABLESAMPLE` feature), a specialized function `concurrentDumpTiDBTables` is used.

### What is changed and how it works?
- The `Writer` interface is changed to `Writer` struct because this abstraction is neither extensible nor convienent.
- During dumping, 
  - multiple `Writer`s are created and there is one connection(`*sql.Conn`) per `Writer`.
  - The database/table meta information and table data are packed into a unit `Task` and sent to a global task channel.
  - Each writer pulls the task from channel constantly, and writes them to the storage.
- This PR also refactors some of the code.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test
   1. Run `make integration_test` first to generate testing data.
   1. Compile the TiDB from https://github.com/pingcap/tidb/pull/20883 and replace it into `$dumpling_base_dir/bin/tidb-server`.
   1. Change the `tableSampleVersion` in `config.go` to a value that <= `4.0.0` (like `3.0.0`)
   1. Dump the testing data.
   1. Run `make integration_test`.

A simple manual test for clustered index is also passed.

Side effects

N/A

Related changes

N/A
 
### Release note

- No 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
-->


kennytm: (blocked by pingcap/tidb#20883)
lichunzhu: @kennytm PTAL

Patch:
diff --git a/v4/export/config.go b/v4/export/config.go
index d9be681..d77d932 100644
--- a/v4/export/config.go
+++ b/v4/export/config.go
@@ -542,7 +542,10 @@ const (
 	defaultEtcdDialTimeOut     = 3 * time.Second
 )
 
-var gcSafePointVersion, _ = semver.NewVersion("4.0.0")
+var (
+	gcSafePointVersion = semver.New("4.0.0")
+	tableSampleVersion = semver.New("5.0.0")
+)
 
 type ServerInfo struct {
 	ServerType    ServerType
@@ -669,3 +672,12 @@ func validateSpecifiedSQL(conf *Config) error {
 	}
 	return nil
 }
+
+func validateFileFormat(conf *Config) error {
+	conf.FileType = strings.ToLower(conf.FileType)
+	switch conf.FileType {
+	case "sql", "csv":
+		return nil
+	}
+	return errors.Errorf("unknown config.FileType '%s'", conf.FileType)
+}
diff --git a/v4/export/dump.go b/v4/export/dump.go
index 1293d22..fb8c0eb 100755
--- a/v4/export/dump.go
+++ b/v4/export/dump.go
@@ -3,8 +3,11 @@
 package export
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
+	"fmt"
+	"math/big"
 	"strings"
 	"time"
 
@@ -13,7 +16,6 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/br/pkg/summary"
-	"github.com/pingcap/br/pkg/utils"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	pd "github.com/tikv/pd/client"
@@ -42,7 +44,8 @@ func NewDumper(ctx context.Context, conf *Config) (*Dumper, error) {
 	err := adjustConfig(conf,
 		initLogger,
 		registerTLSConfig,
-		validateSpecifiedSQL)
+		validateSpecifiedSQL,
+		validateFileFormat)
 	if err != nil {
 		return nil, err
 	}
@@ -114,164 +117,384 @@ func (d *Dumper) Dump() (err error) {
 		conn.Close()
 	}
 
-	connectPool, err := newConnectionsPool(ctx, conf.Threads, pool)
+	metaConn, err := createConnWithConsistency(ctx, pool)
 	if err != nil {
 		return err
 	}
-	defer connectPool.Close()
 
 	if conf.PosAfterConnect {
 		// record again, to provide a location to exit safe mode for DM
-		err = m.recordGlobalMetaData(connectPool.extraConn(), conf.ServerInfo.ServerType, true)
+		err = m.recordGlobalMetaData(metaConn, conf.ServerInfo.ServerType, true)
 		if err != nil {
 			log.Info("get global metadata (after connection pool established) failed", zap.Error(err))
 		}
 	}
 
 	if conf.Consistency != consistencyTypeLock {
-		if err = prepareTableListToDump(conf, connectPool.extraConn()); err != nil {
+		if err = prepareTableListToDump(conf, metaConn); err != nil {
 			return err
 		}
 	}
 
-	if conf.TransactionalConsistency {
-		if conf.Consistency == consistencyTypeFlush || conf.Consistency == consistencyTypeLock {
-			log.Info("All the dumping transactions have started. Start to unlock tables")
+	failpoint.Inject("ConsistencyCheck", nil)
+	rebuildConn := func(conn *sql.Conn) (*sql.Conn, error) {
+		// make sure that the lock connection is still alive
+		err := conCtrl.PingContext(ctx)
+		if err != nil {
+			return conn, err
 		}
-		if err = conCtrl.TearDown(ctx); err != nil {
-			return err
+		// give up the last broken connection
+		conn.Close()
+		newConn, err := createConnWithConsistency(ctx, pool)
+		if err != nil {
+			return conn, err
 		}
+		conn = newConn
+		// renew the master status after connection. dm can't close safe-mode until dm reaches current pos
+		if conf.PosAfterConnect {
+			err = m.recordGlobalMetaData(conn, conf.ServerInfo.ServerType, true)
+			if err != nil {
+				return conn, err
+			}
+		}
+		return conn, nil
 	}
 
-	failpoint.Inject("ConsistencyCheck", nil)
-
-	simpleWriter, err := NewSimpleWriter(conf, d.extStore)
+	taskChan := make(chan Task, defaultDumpThreads)
+	taskChannelCapacity.With(conf.Labels).Add(defaultDumpThreads)
+	wg, writingCtx := errgroup.WithContext(ctx)
+	writers, tearDownWriters, err := d.startWriters(writingCtx, wg, taskChan, rebuildConn)
 	if err != nil {
 		return err
 	}
-	var writer Writer
-	switch strings.ToLower(conf.FileType) {
-	case "sql":
-		writer = SQLWriter{SimpleWriter: simpleWriter}
-	case "csv":
-		writer = CSVWriter{SimpleWriter: simpleWriter}
-	default:
-		return errors.Errorf("unsupported filetype %s", conf.FileType)
+	defer tearDownWriters()
+	if conf.TransactionalConsistency {
+		if conf.Consistency == consistencyTypeFlush || conf.Consistency == consistencyTypeLock {
+			log.Info("All the dumping transactions have started. Start to unlock tables")
+		}
+		if err = conCtrl.TearDown(ctx); err != nil {
+			return err
+		}
 	}
 
 	summary.SetLogCollector(summary.NewLogCollector(log.Info))
 	summary.SetUnit(summary.BackupUnit)
 	defer summary.Summary(summary.BackupUnit)
+
+	tableDataStartTime := time.Now()
 	if conf.Sql == "" {
-		if err = dumpDatabases(ctx, conf, connectPool, writer, func(conn *sql.Conn) (*sql.Conn, error) {
-			// make sure that the lock connection is still alive
-			err := conCtrl.PingContext(ctx)
-			if err != nil {
-				return conn, err
-			}
-			// give up the last broken connection
-			conn.Close()
-			newConn, err := createConnWithConsistency(ctx, pool)
-			if err != nil {
-				return conn, err
-			}
-			conn = newConn
-			// renew the master status after connection. dm can't close safe-mode until dm reaches current pos
-			if conf.PosAfterConnect {
-				err = m.recordGlobalMetaData(conn, conf.ServerInfo.ServerType, true)
-				if err != nil {
-					return conn, err
-				}
-			}
-			return conn, nil
-		}); err != nil {
+		if err = d.dumpDatabases(metaConn, taskChan); err != nil {
 			return err
 		}
 	} else {
-		if err = dumpSql(ctx, conf, connectPool, writer); err != nil {
+		if err = d.dumpSql(metaConn, taskChan); err != nil {
 			return err
 		}
 	}
+	close(taskChan)
+	if err := wg.Wait(); err != nil {
+		summary.CollectFailureUnit("dump table data", err)
+		return err
+	}
+	summary.CollectSuccessUnit("dump cost", countTotalTask(writers), time.Since(tableDataStartTime))
 
 	summary.SetSuccessStatus(true)
 	m.recordFinishTime(time.Now())
 	return nil
 }
 
-func dumpDatabases(pCtx context.Context, conf *Config, connectPool *connectionsPool, writer Writer, rebuildConnFunc func(*sql.Conn) (*sql.Conn, error)) error {
+func (d *Dumper) startWriters(ctx context.Context, wg *errgroup.Group, taskChan <-chan Task,
+	rebuildConnFn func(*sql.Conn) (*sql.Conn, error)) ([]*Writer, func(), error) {
+	conf, pool := d.conf, d.dbHandle
+	writers := make([]*Writer, conf.Threads)
+	for i := 0; i < conf.Threads; i++ {
+		conn, err := createConnWithConsistency(ctx, pool)
+		if err != nil {
+			return nil, func() {}, err
+		}
+		writer := NewWriter(int64(i), ctx, conf, conn, d.extStore)
+		writer.rebuildConnFn = rebuildConnFn
+		writer.finishTableCallBack = func(task Task) {
+			if td, ok := task.(*TaskTableData); ok {
+				log.Debug("finished dumping table data",
+					zap.String("database", td.Meta.DatabaseName()),
+					zap.String("table", td.Meta.TableName()))
+			}
+		}
+		writer.finishTaskCallBack = func(task Task) {
+			taskChannelCapacity.With(conf.Labels).Inc()
+		}
+		wg.Go(func() error {
+			return writer.run(taskChan)
+		})
+		writers[i] = writer
+	}
+	tearDown := func() {
+		for _, w := range writers {
+			w.conn.Close()
+		}
+	}
+	return writers, tearDown, nil
+}
+
+func (d *Dumper) dumpDatabases(metaConn *sql.Conn, taskChan chan<- Task) error {
+	conf := d.conf
 	allTables := conf.Tables
-	g, ctx := errgroup.WithContext(pCtx)
-	tableDataIRTotal := make([]TableDataIR, 0, len(allTables))
-	splitChunkStart := time.Now()
 	for dbName, tables := range allTables {
-		createDatabaseSQL, err := ShowCreateDatabase(connectPool.extraConn(), dbName)
+		createDatabaseSQL, err := ShowCreateDatabase(metaConn, dbName)
 		if err != nil {
 			return err
 		}
-		if err := writer.WriteDatabaseMeta(ctx, dbName, createDatabaseSQL); err != nil {
-			return err
-		}
+		task := NewTaskDatabaseMeta(dbName, createDatabaseSQL)
+		d.sendTaskToChan(task, taskChan)
 
-		if len(tables) == 0 {
-			continue
-		}
 		for _, table := range tables {
-			table := table
-			tableDataIRArray, err := dumpTable(ctx, conf, connectPool.extraConn(), dbName, table, writer)
+			log.Debug("start dumping table...",
+				zap.String("table", table.Name))
+			meta, err := dumpTableMeta(conf, metaConn, dbName, table)
+			if err != nil {
+				return err
+			}
+
+			if table.Type == TableTypeView {
+				task := NewTaskViewMeta(dbName, table.Name, meta.ShowCreateTable(), meta.ShowCreateView())
+				d.sendTaskToChan(task, taskChan)
+			} else {
+				task := NewTaskTableMeta(dbName, table.Name, meta.ShowCreateTable())
+				d.sendTaskToChan(task, taskChan)
+			}
+			err = d.dumpTableData(metaConn, meta, taskChan)
 			if err != nil {
 				return err
 			}
-			tableDataIRTotal = append(tableDataIRTotal, tableDataIRArray...)
 		}
 	}
-	summary.CollectDuration("split chunks", time.Since(splitChunkStart))
-	progressPrinter := utils.StartProgress(ctx, "dumpling", int64(len(tableDataIRTotal)), shouldRedirectLog(conf), log.Info)
-	defer progressPrinter.Close()
-	tableDataStartTime := time.Now()
-	for _, tableIR := range tableDataIRTotal {
-		tableIR := tableIR
-		g.Go(func() error {
-			conn := connectPool.getConn()
-			defer func() {
-				connectPool.releaseConn(conn)
-			}()
-			retryTime := 0
-			var lastErr error
-			return utils.WithRetry(ctx, func() (err error) {
-				defer func() {
-					lastErr = err
-					if err == nil {
-						progressPrinter.Inc()
-					} else {
-						errorCount.With(conf.Labels).Inc()
-					}
-				}()
-				retryTime += 1
-				log.Debug("trying to dump table chunk", zap.Int("retryTime", retryTime), zap.String("db", tableIR.DatabaseName()),
-					zap.String("table", tableIR.TableName()), zap.Int("chunkIndex", tableIR.ChunkIndex()), zap.NamedError("lastError", lastErr))
-				if retryTime > 1 {
-					conn, err = rebuildConnFunc(conn)
-					if err != nil {
-						return
-					}
-				}
-				err = tableIR.Start(ctx, conn)
-				if err != nil {
-					return
-				}
-				return writer.WriteTableData(ctx, tableIR)
-			}, newDumpChunkBackoffer(canRebuildConn(conf.Consistency, conf.TransactionalConsistency)))
-		})
+	return nil
+}
+
+func (d *Dumper) dumpTableData(conn *sql.Conn, meta TableMeta, taskChan chan<- Task) error {
+	conf := d.conf
+	if conf.NoData {
+		return nil
+	}
+	if conf.Rows == UnspecifiedSize {
+		return d.sequentialDumpTable(conn, meta, taskChan)
 	}
-	if err := g.Wait(); err != nil {
-		summary.CollectFailureUnit("dump", err)
+	return d.concurrentDumpTable(conn, meta, taskChan)
+}
+
+func (d *Dumper) sequentialDumpTable(conn *sql.Conn, meta TableMeta, taskChan chan<- Task) error {
+	conf := d.conf
+	db, tbl := meta.DatabaseName(), meta.TableName()
+	tableIR, err := SelectAllFromTable(conf, conn, db, tbl)
+	if err != nil {
 		return err
-	} else {
-		summary.CollectSuccessUnit("dump cost", len(tableDataIRTotal), time.Since(tableDataStartTime))
 	}
+	task := NewTaskTableData(meta, tableIR, 0, 1)
+	d.sendTaskToChan(task, taskChan)
 	return nil
 }
 
+func (d *Dumper) concurrentDumpTable(conn *sql.Conn, meta TableMeta, taskChan chan<- Task) error {
+	conf := d.conf
+	db, tbl := meta.DatabaseName(), meta.TableName()
+	if conf.ServerInfo.ServerType == ServerTypeTiDB &&
+		conf.ServerInfo.ServerVersion != nil &&
+		conf.ServerInfo.ServerVersion.Compare(*tableSampleVersion) >= 0 {
+		log.Debug("dumping TiDB tables with TABLESAMPLE",
+			zap.String("database", db), zap.String("table", tbl))
+		return d.concurrentDumpTiDBTables(conn, meta, taskChan)
+	}
+	field, err := pickupPossibleField(db, tbl, conn, conf)
+	if err != nil {
+		return nil
+	}
+	if field == "" {
+		// skip split chunk logic if not found proper field
+		log.Debug("fallback to sequential dump due to no proper field", zap.String("field", field))
+		return d.sequentialDumpTable(conn, meta, taskChan)
+	}
+
+	min, max, err := d.selectMinAndMaxIntValue(conn, db, tbl, field)
+	if err != nil {
+		return err
+	}
+	log.Debug("get int bounding values",
+		zap.String("lower", min.String()),
+		zap.String("upper", max.String()))
+
+	count := estimateCount(db, tbl, conn, field, conf)
+	log.Info("get estimated rows count", zap.Uint64("estimateCount", count))
+	if count < conf.Rows {
+		// skip chunk logic if estimates are low
+		log.Debug("skip concurrent dump due to estimate count < rows",
+			zap.Uint64("estimate count", count),
+			zap.Uint64("conf.rows", conf.Rows),
+		)
+		return d.sequentialDumpTable(conn, meta, taskChan)
+	}
+
+	// every chunk would have eventual adjustments
+	estimatedChunks := count / conf.Rows
+	estimatedStep := new(big.Int).Sub(max, min).Uint64()/estimatedChunks + 1
+	bigEstimatedStep := new(big.Int).SetUint64(estimatedStep)
+	cutoff := new(big.Int).Set(min)
+	totalChunks := estimatedChunks
+	if estimatedStep == 1 {
+		totalChunks = new(big.Int).Sub(max, min).Uint64()
+	}
+
+	selectField, selectLen, err := buildSelectField(conn, db, tbl, conf.CompleteInsert)
+	if err != nil {
+		return err
+	}
+
+	orderByClause, err := buildOrderByClause(conf, conn, db, tbl)
+	if err != nil {
+		return err
+	}
+
+	chunkIndex := 0
+	nullValueCondition := fmt.Sprintf("`%s` IS NULL OR ", escapeString(field))
+	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)
+		if len(nullValueCondition) > 0 {
+			nullValueCondition = ""
+		}
+		task := NewTaskTableData(meta, newTableData(query, selectLen), chunkIndex, int(totalChunks))
+		ctxDone := d.sendTaskToChan(task, taskChan)
+		if ctxDone {
+			break
+		}
+		cutoff = nextCutOff
+		chunkIndex++
+	}
+	return nil
+}
+
+func (d *Dumper) sendTaskToChan(task Task, taskChan chan<- Task) (ctxDone bool) {
+	ctx, conf := d.ctx, d.conf
+	select {
+	case <-ctx.Done():
+		return true
+	case taskChan <- task:
+		log.Debug("send task to writer",
+			zap.String("task", task.Brief()))
+		taskChannelCapacity.With(conf.Labels).Desc()
+		return false
+	}
+}
+
+func (d *Dumper) selectMinAndMaxIntValue(conn *sql.Conn, db, tbl, field string) (*big.Int, *big.Int, error) {
+	ctx, conf, zero := d.ctx, d.conf, &big.Int{}
+	query := fmt.Sprintf("SELECT MIN(`%s`),MAX(`%s`) FROM `%s`.`%s`",
+		escapeString(field), escapeString(field), escapeString(db), escapeString(tbl))
+	if conf.Where != "" {
+		query = fmt.Sprintf("%s WHERE %s", query, conf.Where)
+	}
+	log.Debug("split chunks", zap.String("query", query))
+
+	var smin sql.NullString
+	var smax sql.NullString
+	row := conn.QueryRowContext(ctx, query)
+	err := row.Scan(&smin, &smax)
+	if err != nil {
+		log.Error("split chunks - get max min failed", zap.String("query", query), zap.Error(err))
+		return zero, zero, err
+	}
+	if !smax.Valid || !smin.Valid {
+		// found no data
+		log.Warn("no data to dump", zap.String("schema", db), zap.String("table", tbl))
+		return zero, zero, nil
+	}
+
+	max := new(big.Int)
+	min := new(big.Int)
+	var ok bool
+	if max, ok = max.SetString(smax.String, 10); !ok {
+		return zero, zero, errors.Errorf("fail to convert max value %s in query %s", smax.String, query)
+	}
+	if min, ok = min.SetString(smin.String, 10); !ok {
+		return zero, zero, errors.Errorf("fail to convert min value %s in query %s", smin.String, query)
+	}
+	return min, max, nil
+}
+
+func (d *Dumper) concurrentDumpTiDBTables(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)
+	if err != nil {
+		return err
+	}
+	if len(handleVals) == 0 {
+		return nil
+	}
+	selectField, selectLen, err := buildSelectField(conn, db, tbl, conf.CompleteInsert)
+	if err != nil {
+		return err
+	}
+	where := buildWhereClauses(handleColNames, handleVals)
+	orderByClause := buildOrderByClauseString(handleColNames)
+
+	for i, w := range where {
+		query := buildSelectQuery(db, tbl, selectField, buildWhereCondition(conf, w), orderByClause)
+		task := NewTaskTableData(meta, newTableData(query, selectLen), i, len(where))
+		ctxDone := d.sendTaskToChan(task, taskChan)
+		if ctxDone {
+			break
+		}
+	}
+	return nil
+}
+
+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)
+	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)
+	if err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+	iter := newRowIter(rows, len(pkFields))
+	defer iter.Close()
+	rowRec := MakeRowReceiver(pkColTypes)
+	buf := new(bytes.Buffer)
+	for iter.HasNext() {
+		err = iter.Decode(rowRec)
+		if err != nil {
+			return nil, nil, errors.Trace(err)
+		}
+		rowRec.WriteToBuffer(buf, true)
+		pkVals = append(pkVals, buf.String())
+		buf.Reset()
+		iter.Next()
+	}
+	return pkFields, pkVals, nil
+}
+
+func buildTiDBTableSampleQuery(pkFields []string, dbName, tblName string) string {
+	template := "SELECT %s FROM `%s`.`%s` TABLESAMPLE REGIONS() ORDER BY %s"
+	quotaPk := make([]string, len(pkFields))
+	for i, s := range pkFields {
+		quotaPk[i] = fmt.Sprintf("`%s`", escapeString(s))
+	}
+	pks := strings.Join(quotaPk, ",")
+	return fmt.Sprintf(template, pks, escapeString(dbName), escapeString(tblName), pks)
+}
+
 func prepareTableListToDump(conf *Config, pool *sql.Conn) error {
 	databases, err := prepareDumpingDatabases(conf, pool)
 	if err != nil {
@@ -295,96 +518,65 @@ func prepareTableListToDump(conf *Config, pool *sql.Conn) error {
 	return nil
 }
 
-func dumpSql(ctx context.Context, conf *Config, connectPool *connectionsPool, writer Writer) error {
-	tableIR, err := SelectFromSql(conf, connectPool.extraConn())
+func dumpTableMeta(conf *Config, conn *sql.Conn, db string, table *TableInfo) (TableMeta, error) {
+	tbl := table.Name
+	selectField, _, err := buildSelectField(conn, db, tbl, conf.CompleteInsert)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	tableDataStartTime := time.Now()
-	err = writer.WriteTableData(ctx, tableIR)
-	if err != nil {
-		summary.CollectFailureUnit("dump", err)
-		return err
+	var colTypes []*sql.ColumnType
+	// If all columns are generated
+	if selectField == "" {
+		colTypes, err = GetColumnTypes(conn, "*", db, tbl)
 	} else {
-		summary.CollectSuccessUnit("dump cost", 1, time.Since(tableDataStartTime))
+		colTypes, err = GetColumnTypes(conn, selectField, db, tbl)
 	}
-	return nil
-}
-
-func dumpTable(ctx context.Context, conf *Config, db *sql.Conn, dbName string, table *TableInfo, writer Writer) ([]TableDataIR, error) {
-	tableName := table.Name
-	if !conf.NoSchemas {
-		if table.Type == TableTypeView {
-			viewName := table.Name
-			createTableSQL, createViewSQL, err := ShowCreateView(db, dbName, viewName)
-			if err != nil {
-				return nil, err
-			}
-			return nil, writer.WriteViewMeta(ctx, dbName, viewName, createTableSQL, createViewSQL)
-		}
-		createTableSQL, err := ShowCreateTable(db, dbName, tableName)
-		if err != nil {
-			return nil, err
-		}
-		if err := writer.WriteTableMeta(ctx, dbName, tableName, createTableSQL); err != nil {
-			return nil, err
-		}
+	if err != nil {
+		return nil, err
 	}
-	// Do not dump table data and return nil
-	if conf.NoData {
-		return nil, nil
+
+	meta := &tableMeta{
+		database:      db,
+		table:         tbl,
+		colTypes:      colTypes,
+		selectedField: selectField,
+		specCmts: []string{
+			"/*!40101 SET NAMES binary*/;",
+		},
 	}
 
-	if conf.Rows != UnspecifiedSize {
-		finished, chunksIterArray, err := concurrentDumpTable(ctx, conf, db, dbName, tableName)
-		if err != nil || finished {
-			return chunksIterArray, err
+	if conf.NoSchemas {
+		return meta, nil
+	}
+	if table.Type == TableTypeView {
+		viewName := table.Name
+		createTableSQL, createViewSQL, err := ShowCreateView(conn, db, viewName)
+		if err != nil {
+			return meta, err
 		}
+		meta.showCreateTable = createTableSQL
+		meta.showCreateView = createViewSQL
+		return meta, nil
 	}
-	tableIR, err := SelectAllFromTable(conf, db, dbName, tableName)
+	createTableSQL, err := ShowCreateTable(conn, db, tbl)
 	if err != nil {
 		return nil, err
 	}
-
-	return []TableDataIR{tableIR}, nil
+	meta.showCreateTable = createTableSQL
+	return meta, nil
 }
 
-func concurrentDumpTable(ctx context.Context, conf *Config, db *sql.Conn, dbName string, tableName string) (bool, []TableDataIR, 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
-	chunksIterArray := make([]TableDataIR, 0)
-	g.Go(func() error {
-		splitTableDataIntoChunks(ctx1, chunksIterCh, errCh, linear, dbName, tableName, db, conf)
-		return nil
-	})
-
-Loop:
-	for {
-		select {
-		case <-ctx.Done():
-			return true, chunksIterArray, nil
-		case <-linear:
-			return false, chunksIterArray, nil
-		case chunksIter, ok := <-chunksIterCh:
-			if !ok {
-				break Loop
-			}
-			chunksIterArray = append(chunksIterArray, chunksIter)
-		case err := <-errCh:
-			return false, chunksIterArray, err
-		}
-	}
-	if err := g.Wait(); err != nil {
-		return true, chunksIterArray, err
+func (d *Dumper) dumpSql(metaConn *sql.Conn, taskChan chan<- Task) error {
+	conf := d.conf
+	meta, tableIR, err := SelectFromSql(conf, metaConn)
+	if err != nil {
+		return err
 	}
-	return true, chunksIterArray, nil
+
+	task := NewTaskTableData(meta, tableIR, 0, 1)
+	d.sendTaskToChan(task, taskChan)
+	return nil
 }
 
 func canRebuildConn(consistency string, trxConsistencyOnly bool) bool {
diff --git a/v4/export/ir.go b/v4/export/ir.go
index e003759..0901e3d 100644
--- a/v4/export/ir.go
+++ b/v4/export/ir.go
@@ -11,20 +11,24 @@ import (
 )
 
 // TableDataIR is table data intermediate representation.
+// A table may be split into multiple TableDataIRs.
 type TableDataIR interface {
 	Start(context.Context, *sql.Conn) error
+	Rows() SQLRowIter
+	Close() error
+}
+
+// TableMeta contains the meta information of a table.
+type TableMeta interface {
 	DatabaseName() string
 	TableName() string
-	ChunkIndex() int
 	ColumnCount() uint
 	ColumnTypes() []string
 	ColumnNames() []string
 	SelectedField() string
-	EscapeBackSlash() bool
-
 	SpecialComments() StringIter
-	Rows() SQLRowIter
-	Close() error
+	ShowCreateTable() string
+	ShowCreateView() string
 }
 
 // SQLRowIter is the iterator on a collection of sql.Row.
diff --git a/v4/export/ir_impl.go b/v4/export/ir_impl.go
index b9f378c..5a9e0dc 100644
--- a/v4/export/ir_impl.go
+++ b/v4/export/ir_impl.go
@@ -6,12 +6,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"math/big"
 	"strings"
-
-	"github.com/pingcap/dumpling/v4/log"
-	"github.com/pingcap/errors"
-	"go.uber.org/zap"
 )
 
 // rowIter implements the SQLRowIter interface.
@@ -78,22 +73,20 @@ func (m *stringIter) HasNext() bool {
 }
 
 type tableData struct {
-	database        string
-	table           string
-	query           string
-	chunkIndex      int
-	rows            *sql.Rows
-	colTypes        []*sql.ColumnType
-	selectedField   string
-	specCmts        []string
-	escapeBackslash bool
-	cancel          context.CancelFunc
+	query  string
+	rows   *sql.Rows
+	colLen int
 	SQLRowIter
 }
 
-func (td *tableData) Start(pCtx context.Context, conn *sql.Conn) error {
-	var ctx context.Context
-	ctx, td.cancel = context.WithCancel(pCtx)
+func newTableData(query string, colLen int) *tableData {
+	return &tableData{
+		query:  query,
+		colLen: colLen,
+	}
+}
+
+func (td *tableData) Start(ctx context.Context, conn *sql.Conn) error {
 	rows, err := conn.QueryContext(ctx, td.query)
 	if err != nil {
 		return err
@@ -103,186 +96,72 @@ func (td *tableData) Start(pCtx context.Context, conn *sql.Conn) error {
 	return nil
 }
 
-func (td *tableData) ColumnTypes() []string {
-	colTypes := make([]string, len(td.colTypes))
-	for i, ct := range td.colTypes {
-		colTypes[i] = ct.DatabaseTypeName()
+func (td *tableData) Rows() SQLRowIter {
+	if td.SQLRowIter == nil {
+		td.SQLRowIter = newRowIter(td.rows, td.colLen)
 	}
-	return colTypes
+	return td.SQLRowIter
 }
 
-func (td *tableData) ColumnNames() []string {
-	colNames := make([]string, len(td.colTypes))
-	for i, ct := range td.colTypes {
-		colNames[i] = ct.Name()
-	}
-	return colNames
+func (td *tableData) Close() error {
+	return td.rows.Close()
 }
 
-func (td *tableData) DatabaseName() string {
-	return td.database
+type tableMeta struct {
+	database        string
+	table           string
+	colTypes        []*sql.ColumnType
+	selectedField   string
+	specCmts        []string
+	showCreateTable string
+	showCreateView  string
 }
 
-func (td *tableData) TableName() string {
-	return td.table
+func (tm *tableMeta) ColumnTypes() []string {
+	colTypes := make([]string, len(tm.colTypes))
+	for i, ct := range tm.colTypes {
+		colTypes[i] = ct.DatabaseTypeName()
+	}
+	return colTypes
 }
 
-func (td *tableData) ChunkIndex() int {
-	return td.chunkIndex
+func (tm *tableMeta) ColumnNames() []string {
+	colNames := make([]string, len(tm.colTypes))
+	for i, ct := range tm.colTypes {
+		colNames[i] = ct.Name()
+	}
+	return colNames
 }
 
-func (td *tableData) ColumnCount() uint {
-	return uint(len(td.colTypes))
+func (tm *tableMeta) DatabaseName() string {
+	return tm.database
 }
 
-func (td *tableData) Rows() SQLRowIter {
-	if td.SQLRowIter == nil {
-		td.SQLRowIter = newRowIter(td.rows, len(td.colTypes))
-	}
-	return td.SQLRowIter
+func (tm *tableMeta) TableName() string {
+	return tm.table
 }
 
-func (td *tableData) Close() error {
-	if td.cancel != nil {
-		td.cancel()
-	}
-	return td.Rows().Close()
+func (tm *tableMeta) ColumnCount() uint {
+	return uint(len(tm.colTypes))
 }
 
-func (td *tableData) SelectedField() string {
-	if td.selectedField == "*" || td.selectedField == "" {
-		return td.selectedField
+func (tm *tableMeta) SelectedField() string {
+	if tm.selectedField == "*" || tm.selectedField == "" {
+		return tm.selectedField
 	}
-	return fmt.Sprintf("(%s)", td.selectedField)
+	return fmt.Sprintf("(%s)", tm.selectedField)
 }
 
-func (td *tableData) SpecialComments() StringIter {
-	return newStringIter(td.specCmts...)
+func (tm *tableMeta) SpecialComments() StringIter {
+	return newStringIter(tm.specCmts...)
 }
 
-func (td *tableData) EscapeBackSlash() bool {
-	return td.escapeBackslash
+func (tm *tableMeta) ShowCreateTable() string {
+	return tm.showCreateTable
 }
 
-func splitTableDataIntoChunks(
-	ctx context.Context,
-	tableDataIRCh chan TableDataIR,
-	errCh chan error,
-	linear chan struct{},
-	dbName, tableName string, db *sql.Conn, conf *Config) {
-	field, err := pickupPossibleField(dbName, tableName, db, conf)
-	if err != nil {
-		errCh <- errors.Trace(err)
-		return
-	}
-	if field == "" {
-		// skip split chunk logic if not found proper field
-		log.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` ",
-		escapeString(field), escapeString(field), escapeString(dbName), escapeString(tableName))
-	if conf.Where != "" {
-		query = fmt.Sprintf("%s WHERE %s", query, conf.Where)
-	}
-	log.Debug("split chunks", zap.String("query", query))
-
-	var smin sql.NullString
-	var smax sql.NullString
-	row := db.QueryRowContext(ctx, query)
-	err = row.Scan(&smin, &smax)
-	if err != nil {
-		log.Error("split chunks - get max min failed", zap.String("query", query), zap.Error(err))
-		errCh <- errors.Trace(err)
-		return
-	}
-	if !smax.Valid || !smin.Valid {
-		// smax and smin are not valid, but there can also be data to dump, so just skip split chunk logic.
-		log.Debug("skip concurrent dump due to no valid smax or smin", zap.String("schema", dbName), zap.String("table", tableName))
-		linear <- struct{}{}
-		return
-	}
-
-	max := new(big.Int)
-	min := new(big.Int)
-	var ok bool
-	if max, ok = max.SetString(smax.String, 10); !ok {
-		errCh <- errors.Errorf("fail to convert max value %s in query %s", smax.String, query)
-		return
-	}
-	if min, ok = min.SetString(smin.String, 10); !ok {
-		errCh <- errors.Errorf("fail to convert min value %s in query %s", smin.String, query)
-		return
-	}
-
-	count := estimateCount(dbName, tableName, db, field, conf)
-	log.Info("get estimated rows count", zap.Uint64("estimateCount", count))
-	if count < conf.Rows {
-		// skip chunk logic if estimates are low
-		log.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 := new(big.Int).Sub(max, min).Uint64()/estimatedChunks + 1
-	bigEstimatedStep := new(big.Int).SetUint64(estimatedStep)
-	cutoff := new(big.Int).Set(min)
-	nextCutoff := new(big.Int)
-
-	selectedField, err := buildSelectField(db, dbName, tableName, conf.CompleteInsert)
-	if err != nil {
-		errCh <- errors.Trace(err)
-		return
-	}
-
-	colTypes, err := GetColumnTypes(db, selectedField, dbName, tableName)
-	if err != nil {
-		errCh <- errors.Trace(err)
-		return
-	}
-	orderByClause, err := buildOrderByClause(conf, db, dbName, tableName)
-	if err != nil {
-		errCh <- errors.Trace(err)
-		return
-	}
-
-	chunkIndex := 0
-	nullValueCondition := fmt.Sprintf("`%s` IS NULL OR ", escapeString(field))
-LOOP:
-	for max.Cmp(cutoff) >= 0 {
-		chunkIndex += 1
-		where := fmt.Sprintf("%s(`%s` >= %d AND `%s` < %d)", nullValueCondition, escapeString(field), cutoff, escapeString(field), nextCutoff.Add(cutoff, bigEstimatedStep))
-		query = buildSelectQuery(dbName, tableName, selectedField, buildWhereCondition(conf, where), orderByClause)
-		if len(nullValueCondition) > 0 {
-			nullValueCondition = ""
-		}
-
-		td := &tableData{
-			database:        dbName,
-			table:           tableName,
-			query:           query,
-			chunkIndex:      chunkIndex,
-			colTypes:        colTypes,
-			selectedField:   selectedField,
-			escapeBackslash: conf.EscapeBackslash,
-			specCmts: []string{
-				"/*!40101 SET NAMES binary*/;",
-			},
-		}
-		cutoff.Set(nextCutoff)
-		select {
-		case <-ctx.Done():
-			break LOOP
-		case tableDataIRCh <- td:
-		}
-	}
-	close(tableDataIRCh)
+func (tm *tableMeta) ShowCreateView() string {
+	return tm.showCreateView
 }
 
 type metaData struct {
diff --git a/v4/export/metrics.go b/v4/export/metrics.go
index b5dab48..5f88646 100644
--- a/v4/export/metrics.go
+++ b/v4/export/metrics.go
@@ -44,6 +44,13 @@ var (
 			Name:      "error_count",
 			Help:      "Total error count during dumping progress",
 		}, []string{})
+	taskChannelCapacity = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "dumpling",
+			Subsystem: "dump",
+			Name:      "channel_capacity",
+			Help:      "The task channel capacity during dumping progress",
+		}, []string{})
 )
 
 // RegisterMetrics registers metrics.
@@ -53,6 +60,7 @@ func RegisterMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(writeTimeHistogram)
 	registry.MustRegister(receiveWriteChunkTimeHistogram)
 	registry.MustRegister(errorCount)
+	registry.MustRegister(taskChannelCapacity)
 }
 
 func RemoveLabelValuesWithTaskInMetrics(labels prometheus.Labels) {
@@ -61,4 +69,5 @@ func RemoveLabelValuesWithTaskInMetrics(labels prometheus.Labels) {
 	writeTimeHistogram.Delete(labels)
 	receiveWriteChunkTimeHistogram.Delete(labels)
 	errorCount.Delete(labels)
+	taskChannelCapacity.Delete(labels)
 }
diff --git a/v4/export/sql.go b/v4/export/sql.go
index 2b43606..5f0b7e7 100644
--- a/v4/export/sql.go
+++ b/v4/export/sql.go
@@ -173,18 +173,7 @@ func SelectVersion(db *sql.DB) (string, error) {
 }
 
 func SelectAllFromTable(conf *Config, db *sql.Conn, database, table string) (TableDataIR, error) {
-	selectedField, err := buildSelectField(db, database, table, conf.CompleteInsert)
-	if err != nil {
-		return nil, err
-	}
-
-	var colTypes []*sql.ColumnType
-	// If all columns are generated
-	if selectedField == "" {
-		colTypes, err = GetColumnTypes(db, "*", database, table)
-	} else {
-		colTypes, err = GetColumnTypes(db, selectedField, database, table)
-	}
+	selectedField, selectLen, err := buildSelectField(db, database, table, conf.CompleteInsert)
 	if err != nil {
 		return nil, err
 	}
@@ -193,60 +182,55 @@ func SelectAllFromTable(conf *Config, db *sql.Conn, database, table string) (Tab
 	if err != nil {
 		return nil, err
 	}
-
-	queryField := selectedField
-	// If all columns are generated
-	if queryField == "" {
-		queryField = "''"
-	}
-	query := buildSelectQuery(database, table, queryField, buildWhereCondition(conf, ""), orderByClause)
+	query := buildSelectQuery(database, table, selectedField, buildWhereCondition(conf, ""), orderByClause)
 
 	return &tableData{
-		database:        database,
-		table:           table,
-		query:           query,
-		colTypes:        colTypes,
-		selectedField:   selectedField,
-		escapeBackslash: conf.EscapeBackslash,
-		specCmts: []string{
-			"/*!40101 SET NAMES binary*/;",
-		},
+		query:  query,
+		colLen: selectLen,
 	}, nil
 }
 
-func SelectFromSql(conf *Config, db *sql.Conn) (TableDataIR, error) {
+func SelectFromSql(conf *Config, conn *sql.Conn) (TableMeta, TableDataIR, error) {
 	log.Info("dump data from sql", zap.String("sql", conf.Sql))
-	rows, err := db.QueryContext(context.Background(), conf.Sql)
+	rows, err := conn.QueryContext(context.Background(), conf.Sql)
 	if err != nil {
-		return nil, errors.Annotatef(err, "sql: %s", conf.Sql)
+		return nil, nil, errors.Annotatef(err, "sql: %s", conf.Sql)
 	}
+	defer rows.Close()
 	colTypes, err := rows.ColumnTypes()
 	if err != nil {
-		return nil, errors.Annotatef(err, "sql: %s", conf.Sql)
+		return nil, nil, errors.Annotatef(err, "sql: %s", conf.Sql)
 	}
 	cols, err := rows.Columns()
 	if err != nil {
-		return nil, errors.Annotatef(err, "sql: %s", conf.Sql)
+		return nil, nil, errors.Annotatef(err, "sql: %s", conf.Sql)
 	}
 	for i := range cols {
 		cols[i] = wrapBackTicks(cols[i])
 	}
-	return &tableData{
-		database:        "",
-		table:           "",
-		rows:            rows,
-		colTypes:        colTypes,
-		selectedField:   strings.Join(cols, ","),
-		escapeBackslash: conf.EscapeBackslash,
+	meta := &tableMeta{
+		colTypes:      colTypes,
+		selectedField: strings.Join(cols, ","),
 		specCmts: []string{
 			"/*!40101 SET NAMES binary*/;",
 		},
-	}, nil
+	}
+	data := &tableData{
+		query:  conf.Sql,
+		colLen: len(cols),
+	}
+
+	return meta, data, nil
 }
 
 func buildSelectQuery(database, table string, fields string, where string, orderByClause string) string {
 	var query strings.Builder
 	query.WriteString("SELECT ")
+	if fields == "" {
+		// If all of the columns are generated,
+		// we need to make sure the query is valid.
+		fields = "''"
+	}
 	query.WriteString(fields)
 	query.WriteString(" FROM `")
 	query.WriteString(escapeString(database))
@@ -277,23 +261,14 @@ func buildOrderByClause(conf *Config, db *sql.Conn, database, table string) (str
 			return "", errors.Trace(err)
 		}
 		if ok {
-			return "ORDER BY _tidb_rowid", nil
+			return "ORDER BY `_tidb_rowid`", nil
 		}
 	}
 	cols, err := GetPrimaryKeyColumns(db, database, table)
 	if err != nil {
 		return "", errors.Trace(err)
 	}
-	tableContainsPriKey := len(cols) != 0
-	if tableContainsPriKey {
-		separator := ", "
-		quotaCols := make([]string, len(cols))
-		for i, col := range cols {
-			quotaCols[i] = fmt.Sprintf("`%s`", escapeString(col))
-		}
-		return fmt.Sprintf("ORDER BY %s", strings.Join(quotaCols, separator)), nil
-	}
-	return "", nil
+	return buildOrderByClauseString(cols), nil
 }
 
 func SelectTiDBRowID(db *sql.Conn, database, table string) (bool, error) {
@@ -320,6 +295,32 @@ func GetColumnTypes(db *sql.Conn, fields, database, table string) ([]*sql.Column
 	return rows.ColumnTypes()
 }
 
+func GetPrimaryKeyAndColumnTypes(conn *sql.Conn, database, table string) ([]string, []string, error) {
+	query :=
+		`SELECT c.COLUMN_NAME, DATA_TYPE FROM
+INFORMATION_SCHEMA.COLUMNS c INNER JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE k ON
+c.column_name = k.column_name and
+c.table_schema = k.table_schema and
+c.table_name = k.table_name and
+c.table_schema = ? and
+c.table_name = ?
+WHERE COLUMN_KEY = 'PRI'
+ORDER BY k.ORDINAL_POSITION;`
+	var colNames, colTypes []string
+	if err := simpleQueryWithArgs(conn, func(rows *sql.Rows) error {
+		var colName, colType string
+		if err := rows.Scan(&colName, &colType); err != nil {
+			return errors.Trace(err)
+		}
+		colNames = append(colNames, colName)
+		colTypes = append(colTypes, strings.ToUpper(colType))
+		return nil
+	}, query, database, table); err != nil {
+		return nil, nil, errors.Annotatef(err, "sql: %s", query)
+	}
+	return colNames, colTypes, nil
+}
+
 func GetPrimaryKeyColumns(db *sql.Conn, database, table string) ([]string, error) {
 	priKeyColsQuery := "SELECT column_name FROM information_schema.KEY_COLUMN_USAGE " +
 		"WHERE table_schema = ? AND table_name = ? AND CONSTRAINT_NAME = 'PRIMARY' order by ORDINAL_POSITION;"
@@ -539,11 +540,13 @@ func createConnWithConsistency(ctx context.Context, db *sql.DB) (*sql.Conn, erro
 	return conn, nil
 }
 
-func buildSelectField(db *sql.Conn, dbName, tableName string, completeInsert bool) (string, error) {
+// buildSelectField returns the selecting fields' string(joined by comma(`,`)),
+// and the number of writable fields.
+func buildSelectField(db *sql.Conn, dbName, tableName string, completeInsert bool) (string, int, error) {
 	query := `SELECT COLUMN_NAME,EXTRA FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA=? AND TABLE_NAME=? ORDER BY ORDINAL_POSITION;`
 	rows, err := db.QueryContext(context.Background(), query, dbName, tableName)
 	if err != nil {
-		return "", errors.Annotatef(err, "sql: %s", query)
+		return "", 0, errors.Annotatef(err, "sql: %s", query)
 	}
 	defer rows.Close()
 	availableFields := make([]string, 0)
@@ -554,7 +557,7 @@ func buildSelectField(db *sql.Conn, dbName, tableName string, completeInsert boo
 	for rows.Next() {
 		err = rows.Scan(&fieldName, &extra)
 		if err != nil {
-			return "", errors.Annotatef(err, "sql: %s", query)
+			return "", 0, errors.Annotatef(err, "sql: %s", query)
 		}
 		switch extra {
 		case "STORED GENERATED", "VIRTUAL GENERATED":
@@ -564,9 +567,40 @@ func buildSelectField(db *sql.Conn, dbName, tableName string, completeInsert boo
 		availableFields = append(availableFields, wrapBackTicks(escapeString(fieldName)))
 	}
 	if completeInsert || hasGenerateColumn {
-		return strings.Join(availableFields, ","), nil
+		return strings.Join(availableFields, ","), len(availableFields), nil
+	}
+	return "*", len(availableFields), nil
+}
+
+func buildWhereClauses(handleColNames, handleVals []string) []string {
+	if len(handleVals) == 0 {
+		return nil
+	}
+	quotaCols := make([]string, len(handleColNames))
+	for i, s := range handleColNames {
+		quotaCols[i] = fmt.Sprintf("`%s`", escapeString(s))
+	}
+	handleCols := strings.Join(quotaCols, ",")
+	where := make([]string, 0, len(handleVals)+1)
+	where = append(where, fmt.Sprintf("(%s) < %s", handleCols, handleVals[0]))
+	for i := 1; i < len(handleVals); i++ {
+		low, up := handleVals[i-1], handleVals[i]
+		where = append(where, fmt.Sprintf("(%s) >= (%s) AND (%s) < (%s)", handleCols, low, handleCols, up))
+	}
+	where = append(where, fmt.Sprintf("(%s) >= (%s)", handleCols, handleVals[len(handleVals)-1]))
+	return where
+}
+
+func buildOrderByClauseString(handleColNames []string) string {
+	if len(handleColNames) == 0 {
+		return ""
+	}
+	separator := ","
+	quotaCols := make([]string, len(handleColNames))
+	for i, col := range handleColNames {
+		quotaCols[i] = fmt.Sprintf("`%s`", escapeString(col))
 	}
-	return "*", nil
+	return fmt.Sprintf("ORDER BY %s", strings.Join(quotaCols, separator))
 }
 
 type oneStrColumnTable struct {
diff --git a/v4/export/task.go b/v4/export/task.go
new file mode 100644
index 0000000..6c11bca
--- /dev/null
+++ b/v4/export/task.go
@@ -0,0 +1,87 @@
+package export
+
+import "fmt"
+
+type Task interface {
+	Brief() string
+}
+
+type TaskDatabaseMeta struct {
+	Task
+	DatabaseName      string
+	CreateDatabaseSQL string
+}
+
+type TaskTableMeta struct {
+	Task
+	DatabaseName   string
+	TableName      string
+	CreateTableSQL string
+}
+
+type TaskViewMeta struct {
+	Task
+	DatabaseName   string
+	ViewName       string
+	CreateTableSQL string
+	CreateViewSQL  string
+}
+
+type TaskTableData struct {
+	Task
+	Meta        TableMeta
+	Data        TableDataIR
+	ChunkIndex  int
+	TotalChunks int
+}
+
+func NewTaskDatabaseMeta(dbName, createSQL string) *TaskDatabaseMeta {
+	return &TaskDatabaseMeta{
+		DatabaseName:      dbName,
+		CreateDatabaseSQL: createSQL,
+	}
+}
+
+func NewTaskTableMeta(dbName, tblName, createSQL string) *TaskTableMeta {
+	return &TaskTableMeta{
+		DatabaseName:   dbName,
+		TableName:      tblName,
+		CreateTableSQL: createSQL,
+	}
+}
+
+func NewTaskViewMeta(dbName, tblName, createTableSQL, createViewSQL string) *TaskViewMeta {
+	return &TaskViewMeta{
+		DatabaseName:   dbName,
+		ViewName:       tblName,
+		CreateTableSQL: createTableSQL,
+		CreateViewSQL:  createViewSQL,
+	}
+}
+
+func NewTaskTableData(meta TableMeta, data TableDataIR, currentChunk, totalChunks int) *TaskTableData {
+	return &TaskTableData{
+		Meta:        meta,
+		Data:        data,
+		ChunkIndex:  currentChunk,
+		TotalChunks: totalChunks,
+	}
+}
+
+func (t *TaskDatabaseMeta) Brief() string {
+	return fmt.Sprintf("meta of dababase '%s'", t.DatabaseName)
+}
+
+func (t *TaskTableMeta) Brief() string {
+	return fmt.Sprintf("meta of table '%s'.'%s'", t.DatabaseName, t.TableName)
+}
+
+func (t *TaskViewMeta) Brief() string {
+	return fmt.Sprintf("meta of view '%s'.'%s'", t.DatabaseName, t.ViewName)
+}
+
+func (t *TaskTableData) Brief() string {
+	db, tbl := t.Meta.DatabaseName(), t.Meta.TableName()
+	idx, total := t.ChunkIndex, t.TotalChunks
+	return fmt.Sprintf("data of table '%s'.'%s'(%d/%d)", db, tbl, idx, total)
+}
diff --git a/v4/export/writer.go b/v4/export/writer.go
index a610b36..c891397 100644
--- a/v4/export/writer.go
+++ b/v4/export/writer.go
@@ -5,84 +5,187 @@ package export
 import (
 	"bytes"
 	"context"
+	"database/sql"
 	"fmt"
 	"strings"
 	"text/template"
 
 	"github.com/pingcap/br/pkg/storage"
+	"github.com/pingcap/br/pkg/utils"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/dumpling/v4/log"
 )
 
-type Writer interface {
-	WriteDatabaseMeta(ctx context.Context, db, createSQL string) error
-	WriteTableMeta(ctx context.Context, db, table, createSQL string) error
-	WriteViewMeta(ctx context.Context, db, table, createTableSQL, createViewSQL string) error
-	WriteTableData(ctx context.Context, ir TableDataIR) error
+type Writer struct {
+	id         int64
+	ctx        context.Context
+	conf       *Config
+	conn       *sql.Conn
+	extStorage storage.ExternalStorage
+	fileFmt    FileFormat
+
+	receivedTaskCount int
+
+	rebuildConnFn       func(*sql.Conn) (*sql.Conn, error)
+	finishTaskCallBack  func(Task)
+	finishTableCallBack func(Task)
 }
 
-type SimpleWriter struct {
-	cfg      *Config
-	extStore storage.ExternalStorage
+func NewWriter(id int64, ctx context.Context, config *Config, conn *sql.Conn, externalStore storage.ExternalStorage) *Writer {
+	sw := &Writer{
+		id:                  id,
+		ctx:                 ctx,
+		conf:                config,
+		conn:                conn,
+		extStorage:          externalStore,
+		finishTaskCallBack:  func(Task) {},
+		finishTableCallBack: func(Task) {},
+	}
+	switch strings.ToLower(config.FileType) {
+	case "sql":
+		sw.fileFmt = FileFormatSQLText
+	case "csv":
+		sw.fileFmt = FileFormatCSV
+	}
+	return sw
 }
 
-func NewSimpleWriter(config *Config, externalStore storage.ExternalStorage) (SimpleWriter, error) {
-	sw := SimpleWriter{
-		cfg:      config,
-		extStore: externalStore,
+func (w *Writer) setFinishTaskCallBack(fn func(Task)) {
+	w.finishTaskCallBack = fn
+}
+
+func (w *Writer) setFinishTableCallBack(fn func(Task)) {
+	w.finishTaskCallBack = fn
+}
+
+func countTotalTask(writers []*Writer) int {
+	sum := 0
+	for _, w := range writers {
+		sum += w.receivedTaskCount
+	}
+	return sum
+}
+
+func (w *Writer) run(taskStream <-chan Task) error {
+	for {
+		select {
+		case <-w.ctx.Done():
+			log.Warn("context has been done, the writer will exit",
+				zap.Int64("writer ID", w.id))
+			return nil
+		case task, ok := <-taskStream:
+			if !ok {
+				return nil
+			}
+			w.receivedTaskCount++
+			err := w.handleTask(task)
+			if err != nil {
+				return err
+			}
+			w.finishTaskCallBack(task)
+		}
 	}
-	return sw, nil
 }
 
-func (f SimpleWriter) WriteDatabaseMeta(ctx context.Context, db, createSQL string) error {
-	fileName, err := (&outputFileNamer{DB: db}).render(f.cfg.OutputFileTemplate, outputFileTemplateSchema)
+func (w *Writer) handleTask(task Task) error {
+	switch t := task.(type) {
+	case *TaskDatabaseMeta:
+		return w.WriteDatabaseMeta(t.DatabaseName, t.CreateDatabaseSQL)
+	case *TaskTableMeta:
+		return w.WriteTableMeta(t.DatabaseName, t.TableName, t.CreateTableSQL)
+	case *TaskViewMeta:
+		return w.WriteViewMeta(t.DatabaseName, t.ViewName, t.CreateTableSQL, t.CreateViewSQL)
+	case *TaskTableData:
+		err := w.WriteTableData(t.Meta, t.Data, t.ChunkIndex)
+		if err != nil {
+			return err
+		}
+		if t.ChunkIndex+1 == t.TotalChunks {
+			w.finishTableCallBack(task)
+		}
+		return nil
+	default:
+		log.Warn("unsupported writer task type", zap.String("type", fmt.Sprintf("%T", t)))
+		return nil
+	}
+}
+
+func (w *Writer) WriteDatabaseMeta(db, createSQL string) error {
+	ctx, conf := w.ctx, w.conf
+	fileName, err := (&outputFileNamer{DB: db}).render(conf.OutputFileTemplate, outputFileTemplateSchema)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createSQL, f.extStore, fileName+".sql", f.cfg.CompressType)
+	return writeMetaToFile(ctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
 }
 
-func (f SimpleWriter) WriteTableMeta(ctx context.Context, db, table, createSQL string) error {
-	fileName, err := (&outputFileNamer{DB: db, Table: table}).render(f.cfg.OutputFileTemplate, outputFileTemplateTable)
+func (w *Writer) WriteTableMeta(db, table, createSQL string) error {
+	ctx, conf := w.ctx, w.conf
+	fileName, err := (&outputFileNamer{DB: db, Table: table}).render(conf.OutputFileTemplate, outputFileTemplateTable)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createSQL, f.extStore, fileName+".sql", f.cfg.CompressType)
+	return writeMetaToFile(ctx, db, createSQL, w.extStorage, fileName+".sql", conf.CompressType)
 }
 
-func (f SimpleWriter) WriteViewMeta(ctx context.Context, db, view, createTableSQL, createViewSQL string) error {
-	fileNameTable, err := (&outputFileNamer{DB: db, Table: view}).render(f.cfg.OutputFileTemplate, outputFileTemplateTable)
+func (w *Writer) WriteViewMeta(db, view, createTableSQL, createViewSQL string) error {
+	ctx, conf := w.ctx, w.conf
+	fileNameTable, err := (&outputFileNamer{DB: db, Table: view}).render(conf.OutputFileTemplate, outputFileTemplateTable)
 	if err != nil {
 		return err
 	}
-	fileNameView, err := (&outputFileNamer{DB: db, Table: view}).render(f.cfg.OutputFileTemplate, outputFileTemplateView)
+	fileNameView, err := (&outputFileNamer{DB: db, Table: view}).render(conf.OutputFileTemplate, outputFileTemplateView)
 	if err != nil {
 		return err
 	}
-	err = writeMetaToFile(ctx, db, createTableSQL, f.extStore, fileNameTable+".sql", f.cfg.CompressType)
+	err = writeMetaToFile(ctx, db, createTableSQL, w.extStorage, fileNameTable+".sql", conf.CompressType)
 	if err != nil {
 		return err
 	}
-	return writeMetaToFile(ctx, db, createViewSQL, f.extStore, fileNameView+".sql", f.cfg.CompressType)
+	return writeMetaToFile(ctx, db, createViewSQL, w.extStorage, fileNameView+".sql", conf.CompressType)
 }
 
-type SQLWriter struct{ SimpleWriter }
-
-func (f SQLWriter) WriteTableData(ctx context.Context, ir TableDataIR) (err error) {
-	log.Debug("start dumping table...", zap.String("table", ir.TableName()))
+func (w *Writer) WriteTableData(meta TableMeta, ir TableDataIR, currentChunk int) error {
+	ctx, conf, conn := w.ctx, w.conf, w.conn
+	retryTime := 0
+	var lastErr error
+	return utils.WithRetry(ctx, func() (err error) {
+		defer func() {
+			lastErr = err
+			if err != nil {
+				errorCount.With(conf.Labels).Inc()
+			}
+		}()
+		retryTime += 1
+		log.Debug("trying to dump table chunk", zap.Int("retryTime", retryTime), zap.String("db", meta.DatabaseName()),
+			zap.String("table", meta.TableName()), zap.Int("chunkIndex", currentChunk), zap.NamedError("lastError", lastErr))
+		if retryTime > 1 {
+			conn, err = w.rebuildConnFn(conn)
+			if err != nil {
+				return
+			}
+		}
+		err = ir.Start(ctx, conn)
+		if err != nil {
+			return
+		}
+		defer ir.Close()
+		return w.writeTableData(ctx, meta, ir, currentChunk)
+	}, newDumpChunkBackoffer(canRebuildConn(conf.Consistency, conf.TransactionalConsistency)))
+}
 
-	defer ir.Close()
-	namer := newOutputFileNamer(ir, f.cfg.Rows != UnspecifiedSize, f.cfg.FileSize != UnspecifiedSize)
-	fileType := strings.ToLower(f.cfg.FileType)
-	fileName, err := namer.NextName(f.cfg.OutputFileTemplate, fileType)
+func (w *Writer) writeTableData(ctx context.Context, meta TableMeta, ir TableDataIR, curChkIdx int) error {
+	conf, format := w.conf, w.fileFmt
+	namer := newOutputFileNamer(meta, curChkIdx, conf.Rows != UnspecifiedSize, conf.FileSize != UnspecifiedSize)
+	fileName, err := namer.NextName(conf.OutputFileTemplate, w.fileFmt.Extension())
 	if err != nil {
 		return err
 	}
 
 	for {
-		fileWriter, tearDown := buildInterceptFileWriter(f.extStore, fileName, f.cfg.CompressType)
-		err = WriteInsert(ctx, ir, fileWriter, f.cfg)
+		fileWriter, tearDown := buildInterceptFileWriter(w.extStorage, fileName, conf.CompressType)
+		err = format.WriteInsert(ctx, conf, meta, ir, fileWriter)
 		tearDown(ctx)
 		if err != nil {
 			return err
@@ -92,16 +195,14 @@ func (f SQLWriter) WriteTableData(ctx context.Context, ir TableDataIR) (err erro
 			break
 		}
 
-		if f.cfg.FileSize == UnspecifiedSize {
+		if conf.FileSize == UnspecifiedSize {
 			break
 		}
-		fileName, err = namer.NextName(f.cfg.OutputFileTemplate, fileType)
+		fileName, err = namer.NextName(conf.OutputFileTemplate, w.fileFmt.Extension())
 		if err != nil {
 			return err
 		}
 	}
-	log.Debug("dumping table successfully",
-		zap.String("table", ir.TableName()))
 	return nil
 }
 
@@ -121,8 +222,6 @@ func writeMetaToFile(ctx context.Context, target, metaSQL string, s storage.Exte
 	}, fileWriter)
 }
 
-type CSVWriter struct{ SimpleWriter }
-
 type outputFileNamer struct {
 	ChunkIndex int
 	FileIndex  int
@@ -137,12 +236,12 @@ type csvOption struct {
 	delimiter []byte
 }
 
-func newOutputFileNamer(ir TableDataIR, rows, fileSize bool) *outputFileNamer {
+func newOutputFileNamer(meta TableMeta, chunkIdx int, rows, fileSize bool) *outputFileNamer {
 	o := &outputFileNamer{
-		DB:    ir.DatabaseName(),
-		Table: ir.TableName(),
+		DB:    meta.DatabaseName(),
+		Table: meta.TableName(),
 	}
-	o.ChunkIndex = ir.ChunkIndex()
+	o.ChunkIndex = chunkIdx
 	o.FileIndex = 0
 	if rows && fileSize {
 		o.format = "%09d%04d"
@@ -171,39 +270,3 @@ func (namer *outputFileNamer) NextName(tmpl *template.Template, fileType string)
 	namer.FileIndex++
 	return res + "." + fileType, err
 }
-
-func (f CSVWriter) WriteTableData(ctx context.Context, ir TableDataIR) (err error) {
-	log.Debug("start dumping table in csv format...", zap.String("table", ir.TableName()))
-
-	defer ir.Close()
-	namer := newOutputFileNamer(ir, f.cfg.Rows != UnspecifiedSize, f.cfg.FileSize != UnspecifiedSize)
-	fileType := strings.ToLower(f.cfg.FileType)
-	fileName, err := namer.NextName(f.cfg.OutputFileTemplate, fileType)
-	if err != nil {
-		return err
-	}
-
-	for {
-		fileWriter, tearDown := buildInterceptFileWriter(f.extStore, fileName, f.cfg.CompressType)
-		err = WriteInsertInCsv(ctx, ir, fileWriter, f.cfg)
-		tearDown(ctx)
-		if err != nil {
-			return err
-		}
-
-		if w, ok := fileWriter.(*InterceptFileWriter); ok && !w.SomethingIsWritten {
-			break
-		}
-
-		if f.cfg.FileSize == UnspecifiedSize {
-			break
-		}
-		fileName, err = namer.NextName(f.cfg.OutputFileTemplate, fileType)
-		if err != nil {
-			return err
-		}
-	}
-	log.Debug("dumping table in csv format successfully",
-		zap.String("table", ir.TableName()))
-	return nil
-}
diff --git a/v4/export/writer_util.go b/v4/export/writer_util.go
index e9adc7d..cb14417 100644
--- a/v4/export/writer_util.go
+++ b/v4/export/writer_util.go
@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path"
 	"strings"
 	"sync"
 	"time"
@@ -134,7 +135,7 @@ func WriteMeta(ctx context.Context, meta MetaIR, w storage.Writer) error {
 	return nil
 }
 
-func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg *Config) error {
+func WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.Writer) error {
 	fileRowIter := tblIR.Rows()
 	if !fileRowIter.HasNext() {
 		return nil
@@ -159,7 +160,7 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg
 		wg.Wait()
 	}()
 
-	specCmtIter := tblIR.SpecialComments()
+	specCmtIter := meta.SpecialComments()
 	for specCmtIter.HasNext() {
 		bf.WriteString(specCmtIter.Next())
 		bf.WriteByte('\n')
@@ -168,22 +169,22 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg
 
 	var (
 		insertStatementPrefix string
-		row                   = MakeRowReceiver(tblIR.ColumnTypes())
+		row                   = MakeRowReceiver(meta.ColumnTypes())
 		counter               uint64
 		lastCounter           uint64
-		escapeBackSlash       = tblIR.EscapeBackSlash()
+		escapeBackSlash       = cfg.EscapeBackslash
 		err                   error
 	)
 
-	selectedField := tblIR.SelectedField()
+	selectedField := meta.SelectedField()
 
 	// if has generated column
 	if selectedField != "" && selectedField != "*" {
 		insertStatementPrefix = fmt.Sprintf("INSERT INTO %s %s VALUES\n",
-			wrapBackTicks(escapeString(tblIR.TableName())), selectedField)
+			wrapBackTicks(escapeString(meta.TableName())), selectedField)
 	} else {
 		insertStatementPrefix = fmt.Sprintf("INSERT INTO %s VALUES\n",
-			wrapBackTicks(escapeString(tblIR.TableName())))
+			wrapBackTicks(escapeString(meta.TableName())))
 	}
 	insertStatementPrefixLen := uint64(len(insertStatementPrefix))
 
@@ -207,7 +208,6 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg
 			counter += 1
 			wp.AddFileSize(uint64(bf.Len()-lastBfSize) + 2) // 2 is for ",\n" and ";\n"
 			failpoint.Inject("ChaosBrokenMySQLConn", func(_ failpoint.Value) {
-				tblIR.Close()
 				failpoint.Return(errors.New("connection is closed"))
 			})
 
@@ -243,7 +243,7 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg
 		}
 	}
 	log.Debug("dumping table",
-		zap.String("table", tblIR.TableName()),
+		zap.String("table", meta.TableName()),
 		zap.Uint64("record counts", counter))
 	if bf.Len() > 0 {
 		wp.input <- bf
@@ -259,7 +259,7 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg
 	return wp.Error()
 }
 
-func WriteInsertInCsv(pCtx context.Context, tblIR TableDataIR, w storage.Writer, cfg *Config) error {
+func WriteInsertInCsv(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.Writer) error {
 	fileRowIter := tblIR.Rows()
 	if !fileRowIter.HasNext() {
 		return nil
@@ -290,20 +290,20 @@ func WriteInsertInCsv(pCtx context.Context, tblIR TableDataIR, w storage.Writer,
 	}()
 
 	var (
-		row             = MakeRowReceiver(tblIR.ColumnTypes())
+		row             = MakeRowReceiver(meta.ColumnTypes())
 		counter         uint64
 		lastCounter     uint64
-		escapeBackSlash = tblIR.EscapeBackSlash()
-		selectedFields  = tblIR.SelectedField()
+		escapeBackSlash = cfg.EscapeBackslash
+		selectedFields  = meta.SelectedField()
 		err             error
 	)
 
-	if !cfg.NoHeader && len(tblIR.ColumnNames()) != 0 && selectedFields != "" {
-		for i, col := range tblIR.ColumnNames() {
+	if !cfg.NoHeader && len(meta.ColumnNames()) != 0 && selectedFields != "" {
+		for i, col := range meta.ColumnNames() {
 			bf.Write(opt.delimiter)
 			escape([]byte(col), bf, getEscapeQuotation(escapeBackSlash, opt.delimiter))
 			bf.Write(opt.delimiter)
-			if i != len(tblIR.ColumnTypes())-1 {
+			if i != len(meta.ColumnTypes())-1 {
 				bf.Write(opt.separator)
 			}
 		}
@@ -347,8 +347,7 @@ func WriteInsertInCsv(pCtx context.Context, tblIR TableDataIR, w storage.Writer,
 	}
 
 	log.Debug("dumping table",
-		zap.String("table", tblIR.TableName()),
-		zap.Int("chunkIndex", tblIR.ChunkIndex()),
+		zap.String("table", meta.TableName()),
 		zap.Uint64("record counts", counter))
 	if bf.Len() > 0 {
 		wp.input <- bf
@@ -395,10 +394,10 @@ func writeBytes(ctx context.Context, writer storage.Writer, p []byte) error {
 	return err
 }
 
-func buildFileWriter(ctx context.Context, s storage.ExternalStorage, path string, compressType storage.CompressType) (storage.Writer, func(ctx context.Context), error) {
-	path += compressFileSuffix(compressType)
-	fullPath := s.URI() + path
-	uploader, err := s.CreateUploader(ctx, path)
+func buildFileWriter(ctx context.Context, s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.Writer, func(ctx context.Context), error) {
+	fileName += compressFileSuffix(compressType)
+	fullPath := path.Join(s.URI(), fileName)
+	uploader, err := s.CreateUploader(ctx, fileName)
 	if err != nil {
 		log.Error("open file failed",
 			zap.String("path", fullPath),
@@ -419,13 +418,13 @@ func buildFileWriter(ctx context.Context, s storage.ExternalStorage, path string
 	return writer, tearDownRoutine, nil
 }
 
-func buildInterceptFileWriter(s storage.ExternalStorage, path string, compressType storage.CompressType) (storage.Writer, func(context.Context)) {
-	path += compressFileSuffix(compressType)
+func buildInterceptFileWriter(s storage.ExternalStorage, fileName string, compressType storage.CompressType) (storage.Writer, func(context.Context)) {
+	fileName += compressFileSuffix(compressType)
 	var writer storage.Writer
-	fullPath := s.URI() + path
+	fullPath := path.Join(s.URI(), fileName)
 	fileWriter := &InterceptFileWriter{}
 	initRoutine := func(ctx context.Context) error {
-		uploader, err := s.CreateUploader(ctx, path)
+		uploader, err := s.CreateUploader(ctx, fileName)
 		if err != nil {
 			log.Error("open file failed",
 				zap.String("path", fullPath),
@@ -531,3 +530,49 @@ func compressFileSuffix(compressType storage.CompressType) string {
 		return ""
 	}
 }
+
+// FileFormat is the format that output to file. Currently we support SQL text and CSV file format.
+type FileFormat int32
+
+const (
+	FileFormatUnknown FileFormat = iota
+	FileFormatSQLText
+	FileFormatCSV
+)
+
+// String implement Stringer.String method.
+func (f FileFormat) String() string {
+	switch f {
+	case FileFormatSQLText:
+		return "SQL"
+	case FileFormatCSV:
+		return "CSV"
+	default:
+		return "unknown"
+	}
+}
+
+// Extension returns the extension for specific format.
+//  text -> "sql"
+//  csv  -> "csv"
+func (f FileFormat) Extension() string {
+	switch f {
+	case FileFormatSQLText:
+		return "sql"
+	case FileFormatCSV:
+		return "csv"
+	default:
+		return "unknown_format"
+	}
+}
+
+func (f FileFormat) WriteInsert(pCtx context.Context, cfg *Config, meta TableMeta, tblIR TableDataIR, w storage.Writer) error {
+	switch f {
+	case FileFormatSQLText:
+		return WriteInsert(pCtx, cfg, meta, tblIR, w)
+	case FileFormatCSV:
+		return WriteInsertInCsv(pCtx, cfg, meta, tblIR, w)
+	default:
+		return errors.Errorf("unknown file format")
+	}
+}


Test patch:
diff --git a/tests/null_unique_index/run.sh b/tests/null_unique_index/run.sh
index 69318ea..23e1c19 100644
--- a/tests/null_unique_index/run.sh
+++ b/tests/null_unique_index/run.sh
@@ -21,6 +21,6 @@ export DUMPLING_TEST_DATABASE=$DB_NAME
 run_dumpling -r 1
 
 data="NULL"
-cnt=$(sed "s/$data/$data\n/g" $DUMPLING_OUTPUT_DIR/$DB_NAME.t.000000001.sql | grep -c "$data") || true
+cnt=$(sed "s/$data/$data\n/g" $DUMPLING_OUTPUT_DIR/$DB_NAME.t.000000000.sql | grep -c "$data") || true
 [ $cnt = 1 ]
 
diff --git a/tests/rows/run.sh b/tests/rows/run.sh
index 18abbc5..b477320 100644
--- a/tests/rows/run.sh
+++ b/tests/rows/run.sh
@@ -54,7 +54,7 @@ if [ "$file_num" -ne 50 ]; then
   echo "obtain file number: $file_num, but expect: 50" && exit 1
 fi
 
-for i in `seq 1 10`
+for i in `seq 0 9`
 do
   r=$(printf "%02d" $i)
   for j in `seq 0 4`
diff --git a/v4/export/prepare_test.go b/v4/export/prepare_test.go
index 16c06c4..1b96c19 100644
--- a/v4/export/prepare_test.go
+++ b/v4/export/prepare_test.go
@@ -119,4 +119,11 @@ func (s *testPrepareSuite) TestConfigValidation(c *C) {
 	c.Assert(validateSpecifiedSQL(conf), ErrorMatches, "can't specify both --sql and --where at the same time. Please try to combine them into --sql")
 	conf.Where = ""
 	c.Assert(validateSpecifiedSQL(conf), IsNil)
+
+	conf.FileType = "sql"
+	c.Assert(validateFileFormat(conf), IsNil)
+	conf.FileType = "csv"
+	c.Assert(validateFileFormat(conf), IsNil)
+	conf.FileType = "rand_str"
+	c.Assert(validateFileFormat(conf), ErrorMatches, "unknown config.FileType 'rand_str'")
 }
diff --git a/v4/export/sql_test.go b/v4/export/sql_test.go
index 9766b0c..2b7f2dd 100644
--- a/v4/export/sql_test.go
+++ b/v4/export/sql_test.go
@@ -78,10 +78,10 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 		WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-	selectedField, err := buildSelectField(conn, "test", "t", false)
+	selectedField, _, err := buildSelectField(conn, "test", "t", false)
 	c.Assert(err, IsNil)
 	q := buildSelectQuery("test", "t", selectedField, "", orderByClause)
-	c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY _tidb_rowid")
+	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`").
@@ -98,7 +98,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 		WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-	selectedField, err = buildSelectField(conn, "test", "t", false)
+	selectedField, _, err = buildSelectField(conn, "test", "t", false)
 	c.Assert(err, IsNil)
 	q = buildSelectQuery("test", "t", selectedField, "", orderByClause)
 	c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY `id`")
@@ -121,7 +121,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 			WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 			WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-		selectedField, err = buildSelectField(conn, "test", "t", false)
+		selectedField, _, err = buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
 		q = buildSelectQuery("test", "t", selectedField, "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t` ORDER BY `id`", cmt)
@@ -145,7 +145,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 			WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 			WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-		selectedField, err = buildSelectField(conn, "test", "t", false)
+		selectedField, _, err = buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
 		q := buildSelectQuery("test", "t", selectedField, "", orderByClause)
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t`", cmt)
@@ -164,7 +164,7 @@ func (s *testSQLSuite) TestBuildSelectAllQuery(c *C) {
 			WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 			WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-		selectedField, err := buildSelectField(conn, "test", "t", false)
+		selectedField, _, err := buildSelectField(conn, "test", "t", false)
 		c.Assert(err, IsNil)
 		q := buildSelectQuery("test", "t", selectedField, "", "")
 		c.Assert(q, Equals, "SELECT * FROM `test`.`t`", cmt)
@@ -191,7 +191,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, "ORDER BY `_tidb_rowid`")
 
 	// _tidb_rowid is unavailable, or PKIsHandle.
 	mock.ExpectExec("SELECT _tidb_rowid from `test`.`t`").
@@ -229,7 +229,7 @@ func (s *testSQLSuite) TestBuildOrderByClause(c *C) {
 			WillReturnRows(sqlmock.NewRows([]string{"column_name"}).AddRow("id").AddRow("name"))
 		orderByClause, err := buildOrderByClause(mockConf, conn, "test", "t")
 		c.Assert(err, IsNil, cmt)
-		c.Assert(orderByClause, Equals, "ORDER BY `id`, `name`", cmt)
+		c.Assert(orderByClause, Equals, "ORDER BY `id`,`name`", cmt)
 	}
 
 	// Test table without primary key.
@@ -269,7 +269,7 @@ func (s *testSQLSuite) TestBuildSelectField(c *C) {
 		WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()).
 		WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", ""))
 
-	selectedField, err := buildSelectField(conn, "test", "t", false)
+	selectedField, _, err := buildSelectField(conn, "test", "t", false)
 	c.Assert(selectedField, Equals, "*")
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
@@ -280,7 +280,7 @@ func (s *testSQLSuite) TestBuildSelectField(c *C) {
 		WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).AddRow("id", "").
 			AddRow("name", "").AddRow("quo`te", ""))
 
-	selectedField, err = buildSelectField(conn, "test", "t", true)
+	selectedField, _, err = buildSelectField(conn, "test", "t", true)
 	c.Assert(selectedField, Equals, "`id`,`name`,`quo``te`")
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
@@ -291,7 +291,7 @@ func (s *testSQLSuite) TestBuildSelectField(c *C) {
 		WillReturnRows(sqlmock.NewRows([]string{"column_name", "extra"}).
 			AddRow("id", "").AddRow("name", "").AddRow("quo`te", "").AddRow("generated", "VIRTUAL GENERATED"))
 
-	selectedField, err = buildSelectField(conn, "test", "t", false)
+	selectedField, _, err = buildSelectField(conn, "test", "t", false)
 	c.Assert(selectedField, Equals, "`id`,`name`,`quo``te`")
 	c.Assert(err, IsNil)
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
diff --git a/v4/export/test_util.go b/v4/export/test_util.go
index 871c961..bf9bc39 100644
--- a/v4/export/test_util.go
+++ b/v4/export/test_util.go
@@ -88,6 +88,14 @@ type mockTableIR struct {
 	SQLRowIter
 }
 
+func (m *mockTableIR) ShowCreateTable() string {
+	return ""
+}
+
+func (m *mockTableIR) ShowCreateView() string {
+	return ""
+}
+
 func (m *mockTableIR) Start(ctx context.Context, conn *sql.Conn) error {
 	return nil
 }
diff --git a/v4/export/writer_test.go b/v4/export/writer_test.go
index 78fcd49..cda5fad 100644
--- a/v4/export/writer_test.go
+++ b/v4/export/writer_test.go
@@ -4,36 +4,58 @@ package export
 
 import (
 	"context"
+	"database/sql"
 	"database/sql/driver"
 	"io/ioutil"
 	"os"
 	"path"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/br/pkg/storage"
-
 	. "github.com/pingcap/check"
 )
 
 var _ = Suite(&testWriterSuite{})
 
-type testWriterSuite struct{}
+type testWriterSuite struct {
+	pool *connectionsPool
+}
 
-func createExtStore(c *C, conf *Config) storage.ExternalStorage {
-	d := &Dumper{conf: conf}
-	c.Assert(createExternalStore(d), IsNil)
-	return d.extStore
+func (s *testWriterSuite) SetUpSuite(c *C) {
+	s.pool = newMockConnectPool(c)
+}
+
+func newMockConnectPool(c *C) *connectionsPool {
+	db, _, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	conn, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	connectPool := &connectionsPool{conns: make(chan *sql.Conn, 1)}
+	connectPool.releaseConn(conn)
+	connectPool.createdConns = []*sql.Conn{conn}
+	return connectPool
+}
+
+func (s *testWriterSuite) newWriter(conf *Config, c *C) *Writer {
+	b, err := storage.ParseBackend(conf.OutputDirPath, &conf.BackendOptions)
+	c.Assert(err, IsNil)
+	extStore, err := storage.Create(context.Background(), b, false)
+	c.Assert(err, IsNil)
+	db, _, err := sqlmock.New()
+	c.Assert(err, IsNil)
+	conn, err := db.Conn(context.Background())
+	c.Assert(err, IsNil)
+	return NewWriter(0, context.Background(), conf, conn, extStore)
 }
 
 func (s *testWriterSuite) TestWriteDatabaseMeta(c *C) {
 	dir := c.MkDir()
-	ctx := context.Background()
 
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 
-	writer, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	err = writer.WriteDatabaseMeta(ctx, "test", "CREATE DATABASE `test`")
+	writer := s.newWriter(config, c)
+	err := writer.WriteDatabaseMeta("test", "CREATE DATABASE `test`")
 	c.Assert(err, IsNil)
 	p := path.Join(dir, "test-schema-create.sql")
 	_, err = os.Stat(p)
@@ -45,14 +67,12 @@ func (s *testWriterSuite) TestWriteDatabaseMeta(c *C) {
 
 func (s *testWriterSuite) TestWriteTableMeta(c *C) {
 	dir := c.MkDir()
-	ctx := context.Background()
 
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 
-	writer, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	err = writer.WriteTableMeta(ctx, "test", "t", "CREATE TABLE t (a INT)")
+	writer := s.newWriter(config, c)
+	err := writer.WriteTableMeta("test", "t", "CREATE TABLE t (a INT)")
 	c.Assert(err, IsNil)
 	p := path.Join(dir, "test.t-schema.sql")
 	_, err = os.Stat(p)
@@ -64,17 +84,15 @@ func (s *testWriterSuite) TestWriteTableMeta(c *C) {
 
 func (s *testWriterSuite) TestWriteViewMeta(c *C) {
 	dir := c.MkDir()
-	ctx := context.Background()
 
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 
-	writer, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
+	writer := s.newWriter(config, c)
 	specCmt := "/*!40101 SET NAMES binary*/;\n"
 	createTableSQL := "CREATE TABLE `v`(\n`a` int\n)ENGINE=MyISAM;\n"
 	createViewSQL := "DROP TABLE IF EXISTS `v`;\nDROP VIEW IF EXISTS `v`;\nSET @PREV_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT;\nSET @PREV_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS;\nSET @PREV_COLLATION_CONNECTION=@@COLLATION_CONNECTION;\nSET character_set_client = utf8;\nSET character_set_results = utf8;\nSET collation_connection = utf8_general_ci;\nCREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v` (`a`) AS SELECT `t`.`a` AS `a` FROM `test`.`t`;\nSET character_set_client = @PREV_CHARACTER_SET_CLIENT;\nSET character_set_results = @PREV_CHARACTER_SET_RESULTS;\nSET collation_connection = @PREV_COLLATION_CONNECTION;\n"
-	err = writer.WriteViewMeta(ctx, "test", "v", createTableSQL, createViewSQL)
+	err := writer.WriteViewMeta("test", "v", createTableSQL, createViewSQL)
 	c.Assert(err, IsNil)
 
 	p := path.Join(dir, "test.v-schema.sql")
@@ -95,14 +113,10 @@ func (s *testWriterSuite) TestWriteViewMeta(c *C) {
 func (s *testWriterSuite) TestWriteTableData(c *C) {
 	dir := c.MkDir()
 
-	ctx := context.Background()
-
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 
-	simpleWriter, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	writer := SQLWriter{SimpleWriter: simpleWriter}
+	writer := s.newWriter(config, c)
 
 	data := [][]driver.Value{
 		{"1", "male", "bob@mail.com", "020-1234", nil},
@@ -116,7 +130,7 @@ func (s *testWriterSuite) TestWriteTableData(c *C) {
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;",
 	}
 	tableIR := newMockTableIR("test", "employee", data, specCmts, colTypes)
-	err = writer.WriteTableData(ctx, tableIR)
+	err := writer.WriteTableData(tableIR, tableIR, 0)
 	c.Assert(err, IsNil)
 
 	p := path.Join(dir, "test.employee.000000000.sql")
@@ -138,8 +152,6 @@ func (s *testWriterSuite) TestWriteTableData(c *C) {
 func (s *testWriterSuite) TestWriteTableDataWithFileSize(c *C) {
 	dir := c.MkDir()
 
-	ctx := context.Background()
-
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 	config.FileSize = 50
@@ -151,9 +163,7 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSize(c *C) {
 	config.FileSize += uint64(len(specCmts[1]) + 1)
 	config.FileSize += uint64(len("INSERT INTO `employees` VALUES\n"))
 
-	simpleWriter, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	writer := SQLWriter{SimpleWriter: simpleWriter}
+	writer := s.newWriter(config, c)
 
 	data := [][]driver.Value{
 		{"1", "male", "bob@mail.com", "020-1234", nil},
@@ -163,7 +173,7 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSize(c *C) {
 	}
 	colTypes := []string{"INT", "SET", "VARCHAR", "VARCHAR", "TEXT"}
 	tableIR := newMockTableIR("test", "employee", data, specCmts, colTypes)
-	err = writer.WriteTableData(ctx, tableIR)
+	err := writer.WriteTableData(tableIR, tableIR, 0)
 	c.Assert(err, IsNil)
 
 	cases := map[string]string{
@@ -181,7 +191,7 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSize(c *C) {
 
 	for p, expected := range cases {
 		p := path.Join(dir, p)
-		_, err = os.Stat(p)
+		_, err := os.Stat(p)
 		c.Assert(err, IsNil)
 		bytes, err := ioutil.ReadFile(p)
 		c.Assert(err, IsNil)
@@ -192,8 +202,6 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSize(c *C) {
 func (s *testWriterSuite) TestWriteTableDataWithFileSizeAndRows(c *C) {
 	dir := c.MkDir()
 
-	ctx := context.Background()
-
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 	config.FileSize = 50
@@ -206,9 +214,7 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSizeAndRows(c *C) {
 	config.FileSize += uint64(len(specCmts[1]) + 1)
 	config.FileSize += uint64(len("INSERT INTO `employees` VALUES\n"))
 
-	simpleWriter, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	writer := SQLWriter{SimpleWriter: simpleWriter}
+	writer := s.newWriter(config, c)
 
 	data := [][]driver.Value{
 		{"1", "male", "bob@mail.com", "020-1234", nil},
@@ -218,7 +224,7 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSizeAndRows(c *C) {
 	}
 	colTypes := []string{"INT", "SET", "VARCHAR", "VARCHAR", "TEXT"}
 	tableIR := newMockTableIR("test", "employee", data, specCmts, colTypes)
-	err = writer.WriteTableData(ctx, tableIR)
+	err := writer.WriteTableData(tableIR, tableIR, 0)
 	c.Assert(err, IsNil)
 
 	cases := map[string]string{
@@ -247,8 +253,6 @@ func (s *testWriterSuite) TestWriteTableDataWithFileSizeAndRows(c *C) {
 func (s *testWriterSuite) TestWriteTableDataWithStatementSize(c *C) {
 	dir := c.MkDir()
 
-	ctx := context.Background()
-
 	config := DefaultConfig()
 	config.OutputDirPath = dir
 	config.StatementSize = 50
@@ -257,9 +261,7 @@ func (s *testWriterSuite) TestWriteTableDataWithStatementSize(c *C) {
 	config.OutputFileTemplate, err = ParseOutputFileTemplate("specified-name")
 	c.Assert(err, IsNil)
 
-	simpleWriter, err := NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	writer := SQLWriter{SimpleWriter: simpleWriter}
+	writer := s.newWriter(config, c)
 
 	data := [][]driver.Value{
 		{"1", "male", "bob@mail.com", "020-1234", nil},
@@ -273,7 +275,7 @@ func (s *testWriterSuite) TestWriteTableDataWithStatementSize(c *C) {
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;",
 	}
 	tableIR := newMockTableIR("te%/st", "employee", data, specCmts, colTypes)
-	err = writer.WriteTableData(ctx, tableIR)
+	err = writer.WriteTableData(tableIR, tableIR, 0)
 	c.Assert(err, IsNil)
 
 	// only with statement size
@@ -309,9 +311,7 @@ func (s *testWriterSuite) TestWriteTableDataWithStatementSize(c *C) {
 	err = os.RemoveAll(config.OutputDirPath)
 	c.Assert(err, IsNil)
 	config.OutputDirPath, err = ioutil.TempDir("", "dumpling")
-	simpleWriter, err = NewSimpleWriter(config, createExtStore(c, config))
-	c.Assert(err, IsNil)
-	writer = SQLWriter{SimpleWriter: simpleWriter}
+	writer = s.newWriter(config, c)
 
 	cases = map[string]string{
 		"000000000-employee-te%25%2Fst.sql": "/*!40101 SET NAMES binary*/;\n" +
@@ -328,7 +328,7 @@ func (s *testWriterSuite) TestWriteTableDataWithStatementSize(c *C) {
 	}
 
 	tableIR = newMockTableIR("te%/st", "employee", data, specCmts, colTypes)
-	c.Assert(writer.WriteTableData(ctx, tableIR), IsNil)
+	c.Assert(writer.WriteTableData(tableIR, tableIR, 0), IsNil)
 	c.Assert(err, IsNil)
 	for p, expected := range cases {
 		p := path.Join(config.OutputDirPath, p)
diff --git a/v4/export/writer_util_test.go b/v4/export/writer_util_test.go
index 78c6308..a3e0b1e 100644
--- a/v4/export/writer_util_test.go
+++ b/v4/export/writer_util_test.go
@@ -25,7 +25,7 @@ type testUtilSuite struct {
 	mockCfg *Config
 }
 
-func (s *testUtilSuite) SetUpSuite(c *C) {
+func (s *testUtilSuite) SetUpSuite(_ *C) {
 	s.mockCfg = &Config{
 		FileSize: UnspecifiedSize,
 	}
@@ -63,7 +63,8 @@ func (s *testUtilSuite) TestWriteInsert(c *C) {
 	tableIR := newMockTableIR("test", "employee", data, specCmts, colTypes)
 	bf := storage.NewBufferWriter()
 
-	err := WriteInsert(context.Background(), tableIR, bf, configForWriteSQL(UnspecifiedSize, UnspecifiedSize))
+	conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
+	err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected := "/*!40101 SET NAMES binary*/;\n" +
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
@@ -93,7 +94,8 @@ func (s *testUtilSuite) TestWriteInsertReturnsError(c *C) {
 	tableIR.rowErr = rowErr
 	bf := storage.NewBufferWriter()
 
-	err := WriteInsert(context.Background(), tableIR, bf, configForWriteSQL(UnspecifiedSize, UnspecifiedSize))
+	conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
+	err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, Equals, rowErr)
 	expected := "/*!40101 SET NAMES binary*/;\n" +
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
@@ -117,7 +119,8 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 
 	// test nullValue
 	opt := &csvOption{separator: []byte(","), delimiter: doubleQuotationMark, nullValue: "\\N"}
-	err := WriteInsertInCsv(context.Background(), tableIR, bf, configForWriteCSV(true, opt, UnspecifiedSize))
+	conf := configForWriteCSV(true, opt, UnspecifiedSize)
+	err := WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected := "1,\"male\",\"bob@mail.com\",\"020-1234\",\\N\n" +
 		"2,\"female\",\"sarah@mail.com\",\"020-1253\",\"healthy\"\n" +
@@ -129,7 +132,8 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	bf.Reset()
 	opt.delimiter = quotationMark
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
-	err = WriteInsertInCsv(context.Background(), tableIR, bf, configForWriteCSV(true, opt, UnspecifiedSize))
+	conf = configForWriteCSV(true, opt, UnspecifiedSize)
+	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "1,'male','bob@mail.com','020-1234',\\N\n" +
 		"2,'female','sarah@mail.com','020-1253','healthy'\n" +
@@ -141,7 +145,8 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	bf.Reset()
 	opt.separator = []byte(";")
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
-	err = WriteInsertInCsv(context.Background(), tableIR, bf, configForWriteCSV(true, opt, UnspecifiedSize))
+	conf = configForWriteCSV(true, opt, UnspecifiedSize)
+	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "1;'male';'bob@mail.com';'020-1234';\\N\n" +
 		"2;'female';'sarah@mail.com';'020-1253';'healthy'\n" +
@@ -155,7 +160,8 @@ func (s *testUtilSuite) TestWriteInsertInCsv(c *C) {
 	opt.delimiter = []byte("ma")
 	tableIR = newMockTableIR("test", "employee", data, nil, colTypes)
 	tableIR.colNames = []string{"id", "gender", "email", "phone_number", "status"}
-	err = WriteInsertInCsv(context.Background(), tableIR, bf, configForWriteCSV(false, opt, UnspecifiedSize))
+	conf = configForWriteCSV(false, opt, UnspecifiedSize)
+	err = WriteInsertInCsv(context.Background(), conf, tableIR, tableIR, bf)
 	c.Assert(err, IsNil)
 	expected = "maidma&;,?magenderma&;,?maemamailma&;,?maphone_numberma&;,?mastatusma\n" +
 		"1&;,?mamamalema&;,?mabob@mamail.comma&;,?ma020-1234ma&;,?\\N\n" +
@@ -180,7 +186,8 @@ func (s *testUtilSuite) TestSQLDataTypes(c *C) {
 		tableIR := newMockTableIR("test", "t", tableData, nil, colType)
 		bf := storage.NewBufferWriter()
 
-		err := WriteInsert(context.Background(), tableIR, bf, configForWriteSQL(UnspecifiedSize, UnspecifiedSize))
+		conf := configForWriteSQL(UnspecifiedSize, UnspecifiedSize)
+		err := WriteInsert(context.Background(), conf, tableIR, tableIR, bf)
 		c.Assert(err, IsNil)
 		lines := strings.Split(bf.String(), "\n")
 		c.Assert(len(lines), Equals, 3)