Repository: pingcap/dumpling
Pull request #50 (2020-03-16 04:01:04)

optimize dumpling's effiency
When dumping 2.9G data from a single mysql table, dumpling need more than 1min10s now.
In this we use `bytes.Buffer`, write pipeline, `sync.Pool` to save the memory copy time. Current dumpling need 28s to do this job.
```
real	0m27.489s
user	0m26.615s
sys	0m10.495s
```

codecov-io: # [Codecov](https://codecov.io/gh/pingcap/dumpling/pull/50?src=pr&el=h1) Report
> Merging [#50](https://codecov.io/gh/pingcap/dumpling/pull/50?src=pr&el=desc) into [master](https://codecov.io/gh/pingcap/dumpling/commit/67fc4be734c5de0ce8412d57d3f18711b268ff00&el=desc) will **not change** coverage by `%`.
> The diff coverage is `n/a`.


```diff
@@           Coverage Diff           @@
##           master      #50   +/-   ##
=======================================
  Coverage   59.60%   59.60%           
=======================================
  Files          16       16           
  Lines        1369     1369           
=======================================
  Hits          816      816           
  Misses        493      493           
  Partials       60       60           
```


Patch:
diff --git a/v4/export/ir.go b/v4/export/ir.go
index b1607db..db7a0c4 100644
--- a/v4/export/ir.go
+++ b/v4/export/ir.go
@@ -1,6 +1,9 @@
 package export
 
-import "database/sql"
+import (
+	"bytes"
+	"database/sql"
+)
 
 // TableDataIR is table data intermediate representation.
 type TableDataIR interface {
@@ -18,7 +21,8 @@ type TableDataIR interface {
 
 // SQLRowIter is the iterator on a collection of sql.Row.
 type SQLRowIter interface {
-	Next(RowReceiver) error
+	Decode(RowReceiver) error
+	Next()
 	HasNext() bool
 	HasNextSQLRowIter() bool
 	NextSQLRowIter() SQLRowIter
@@ -32,7 +36,7 @@ type RowReceiverStringer interface {
 }
 
 type Stringer interface {
-	ToString(bool) string
+	WriteToBuffer(*bytes.Buffer, bool)
 }
 
 type RowReceiver interface {
diff --git a/v4/export/ir_impl.go b/v4/export/ir_impl.go
index 1839a84..c244bd4 100644
--- a/v4/export/ir_impl.go
+++ b/v4/export/ir_impl.go
@@ -34,10 +34,12 @@ func (iter *rowIter) Close() error {
 	return iter.rows.Close()
 }
 
-func (iter *rowIter) Next(row RowReceiver) error {
-	err := decodeFromRows(iter.rows, iter.args, row)
+func (iter *rowIter) Decode(row RowReceiver) error {
+	return decodeFromRows(iter.rows, iter.args, row)
+}
+
+func (iter *rowIter) Next() {
 	iter.hasNext = iter.rows.Next()
-	return err
 }
 
 func (iter *rowIter) HasNext() bool {
@@ -65,18 +67,21 @@ func (c *fileRowIter) Close() error {
 	return c.rowIter.Close()
 }
 
-func (c *fileRowIter) Next(row RowReceiver) error {
-	err := c.rowIter.Next(row)
+func (c *fileRowIter) Decode(row RowReceiver) error {
+	err := c.rowIter.Decode(row)
 	if err != nil {
 		return err
 	}
-
 	size := row.ReportSize()
 	c.currentFileSize += size
 	c.currentStatementSize += size
 	return nil
 }
 
+func (c *fileRowIter) Next() {
+	c.rowIter.Next()
+}
+
 func (c *fileRowIter) HasNext() bool {
 	if c.fileSizeLimit != UnspecifiedSize && c.currentFileSize >= c.fileSizeLimit {
 		return false
diff --git a/v4/export/sql_type.go b/v4/export/sql_type.go
index 7b9c17a..2afa2df 100644
--- a/v4/export/sql_type.go
+++ b/v4/export/sql_type.go
@@ -4,11 +4,15 @@ import (
 	"bytes"
 	"database/sql"
 	"fmt"
-	"strings"
 )
 
 var colTypeRowReceiverMap = map[string]func() RowReceiverStringer{}
 
+var nullValue = "NULL"
+var quotationMark byte = '\''
+var quotationMarkNotQuote = []byte{quotationMark}
+var quotationMarkQuote = []byte{quotationMark, quotationMark}
+
 func init() {
 	for _, s := range dataTypeString {
 		colTypeRowReceiverMap[s] = SQLTypeStringMaker
@@ -42,12 +46,12 @@ var dataTypeBin = []string{
 	"BIT",
 }
 
-func escape(s string, escapeBackslash bool) string {
+func escape(s []byte, bf *bytes.Buffer, escapeBackslash bool) {
 	if !escapeBackslash {
-		return strings.ReplaceAll(s, "'", "''")
+		bf.Write(bytes.ReplaceAll(s, quotationMarkNotQuote, quotationMarkQuote))
+		return
 	}
 	var (
-		bf     bytes.Buffer
 		escape byte
 		last   = 0
 	)
@@ -79,23 +83,17 @@ func escape(s string, escapeBackslash bool) string {
 		}
 
 		if escape != 0 {
-			if last == 0 {
-				bf.Grow(2 * len(s))
-			}
-			bf.WriteString(s[last:i])
+			bf.Write(s[last:i])
 			bf.WriteByte('\\')
 			bf.WriteByte(escape)
 			last = i + 1
 		}
 	}
 	if last == 0 {
-		return s
-	}
-	if last < len(s) {
-		bf.WriteString(s[last:])
+		bf.Write(s)
+	} else if last < len(s) {
+		bf.Write(s[last:])
 	}
-	defer bf.Reset()
-	return bf.String()
 }
 
 func SQLTypeStringMaker() RowReceiverStringer {
@@ -126,9 +124,7 @@ type RowReceiverArr []RowReceiverStringer
 
 func (r RowReceiverArr) BindAddress(args []interface{}) {
 	for i := range args {
-		var singleAddr [1]interface{}
-		r[i].BindAddress(singleAddr[:])
-		args[i] = singleAddr[0]
+		r[i].BindAddress(args[i : i+1])
 	}
 }
 func (r RowReceiverArr) ReportSize() uint64 {
@@ -138,62 +134,65 @@ func (r RowReceiverArr) ReportSize() uint64 {
 	}
 	return sum
 }
-func (r RowReceiverArr) ToString(escapeBackslash bool) string {
-	var sb strings.Builder
-	sb.WriteString("(")
+
+func (r RowReceiverArr) WriteToBuffer(bf *bytes.Buffer, escapeBackslash bool) {
+	bf.WriteByte('(')
 	for i, receiver := range r {
-		sb.WriteString(receiver.ToString(escapeBackslash))
+		receiver.WriteToBuffer(bf, escapeBackslash)
 		if i != len(r)-1 {
-			sb.WriteString(", ")
+			bf.WriteByte(',')
 		}
 	}
-	sb.WriteString(")")
-	return sb.String()
+	bf.WriteByte(')')
 }
 
 type SQLTypeNumber struct {
 	SQLTypeString
 }
 
-func (s SQLTypeNumber) ToString(bool) string {
-	if s.Valid {
-		return s.String
+func (s SQLTypeNumber) WriteToBuffer(bf *bytes.Buffer, _ bool) {
+	if s.RawBytes != nil {
+		bf.Write(s.RawBytes)
 	} else {
-		return "NULL"
+		bf.WriteString(nullValue)
 	}
 }
 
 type SQLTypeString struct {
-	sql.NullString
+	sql.RawBytes
 }
 
 func (s *SQLTypeString) BindAddress(arg []interface{}) {
-	arg[0] = s
+	arg[0] = &s.RawBytes
 }
 func (s *SQLTypeString) ReportSize() uint64 {
-	if s.Valid {
-		return uint64(len(s.String))
+	if s.RawBytes != nil {
+		return uint64(len(s.RawBytes))
 	}
-	return uint64(len("NULL"))
+	return uint64(len(nullValue))
 }
-func (s *SQLTypeString) ToString(escapeBackslash bool) string {
-	if s.Valid {
-		return fmt.Sprintf(`'%s'`, escape(s.String, escapeBackslash))
+
+func (s *SQLTypeString) WriteToBuffer(bf *bytes.Buffer, escapeBackslash bool) {
+	if s.RawBytes != nil {
+		bf.WriteByte(quotationMark)
+		escape(s.RawBytes, bf, escapeBackslash)
+		bf.WriteByte(quotationMark)
 	} else {
-		return "NULL"
+		bf.WriteString(nullValue)
 	}
 }
 
 type SQLTypeBytes struct {
-	bytes []byte
+	sql.RawBytes
 }
 
 func (s *SQLTypeBytes) BindAddress(arg []interface{}) {
-	arg[0] = &s.bytes
+	arg[0] = &s.RawBytes
 }
 func (s *SQLTypeBytes) ReportSize() uint64 {
-	return uint64(len(s.bytes))
+	return uint64(len(s.RawBytes))
 }
-func (s *SQLTypeBytes) ToString(bool) string {
-	return fmt.Sprintf("x'%x'", s.bytes)
+
+func (s *SQLTypeBytes) WriteToBuffer(bf *bytes.Buffer, _ bool) {
+	fmt.Fprintf(bf, "x'%x'", s.RawBytes)
 }
diff --git a/v4/export/writer.go b/v4/export/writer.go
index 15d49b1..7cf10f6 100644
--- a/v4/export/writer.go
+++ b/v4/export/writer.go
@@ -56,15 +56,14 @@ func (f *SimpleWriter) WriteTableData(ctx context.Context, ir TableDataIR) error
 
 	for {
 		filePath := path.Join(f.cfg.OutputDirPath, fileName)
-		fileWriter, tearDown := buildLazyFileWriter(filePath)
-		intWriter := &InterceptStringWriter{StringWriter: fileWriter}
-		err := WriteInsert(chunksIter, intWriter)
+		fileWriter, tearDown := buildInterceptFileWriter(filePath)
+		err := WriteInsert(chunksIter, fileWriter)
 		tearDown()
 		if err != nil {
 			return err
 		}
 
-		if !intWriter.SomethingIsWritten {
+		if w, ok := fileWriter.(*InterceptFileWriter); ok && !w.SomethingIsWritten {
 			break
 		}
 
diff --git a/v4/export/writer_util.go b/v4/export/writer_util.go
index 608a143..479974c 100644
--- a/v4/export/writer_util.go
+++ b/v4/export/writer_util.go
@@ -2,6 +2,8 @@ package export
 
 import (
 	"bufio"
+	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"os"
@@ -12,6 +14,61 @@ import (
 	"go.uber.org/zap"
 )
 
+const lengthLimit = 1048576
+
+var pool = sync.Pool{New: func() interface{} {
+	return &bytes.Buffer{}
+}}
+
+type writerPipe struct {
+	input  chan []byte
+	closed chan struct{}
+	errCh  chan error
+
+	w io.Writer
+}
+
+func newWriterPipe(w io.Writer) *writerPipe {
+	return &writerPipe{
+		input:  make(chan []byte, 8),
+		closed: make(chan struct{}),
+		errCh:  make(chan error, 1),
+		w:      w,
+	}
+}
+
+func (b *writerPipe) Run(ctx context.Context) {
+	defer close(b.closed)
+	var errOccurs bool
+	for {
+		select {
+		case s, ok := <-b.input:
+			if !ok {
+				return
+			}
+			if errOccurs {
+				continue
+			}
+			err := writeBytes(b.w, s)
+			if err != nil {
+				errOccurs = true
+				b.errCh <- err
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (b *writerPipe) Error() error {
+	select {
+	case err := <-b.errCh:
+		return err
+	default:
+		return nil
+	}
+}
+
 func WriteMeta(meta MetaIR, w io.StringWriter) error {
 	log.Zap().Debug("start dumping meta data", zap.String("target", meta.TargetName()))
 
@@ -30,24 +87,41 @@ func WriteMeta(meta MetaIR, w io.StringWriter) error {
 	return nil
 }
 
-func WriteInsert(tblIR TableDataIR, w io.StringWriter) error {
+func WriteInsert(tblIR TableDataIR, w io.Writer) error {
 	fileRowIter := tblIR.Rows()
 	if !fileRowIter.HasNext() {
 		return nil
 	}
 
+	bf := pool.Get().(*bytes.Buffer)
+	bf.Grow(lengthLimit)
+
+	wp := newWriterPipe(w)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		wp.Run(ctx)
+		wg.Done()
+	}()
+	defer func() {
+		cancel()
+		wg.Wait()
+	}()
+
 	specCmtIter := tblIR.SpecialComments()
 	for specCmtIter.HasNext() {
-		if err := write(w, fmt.Sprintf("%s\n", specCmtIter.Next())); err != nil {
-			return err
-		}
+		bf.WriteString(specCmtIter.Next())
+		bf.WriteByte('\n')
 	}
 
 	var (
-		insertStatementPrefix = fmt.Sprintf("INSERT INTO %s VALUES\n", wrapBackTicks(tblIR.TableName()))
+		insertStatementPrefix string
 		row                   = MakeRowReceiver(tblIR.ColumnTypes())
 		counter               = 0
 		escapeBackSlash       = tblIR.EscapeBackSlash()
+		err                   error
 	)
 
 	selectedField := tblIR.SelectedField()
@@ -55,48 +129,79 @@ func WriteInsert(tblIR TableDataIR, w io.StringWriter) error {
 	if selectedField != "" {
 		insertStatementPrefix = fmt.Sprintf("INSERT INTO %s %s VALUES\n",
 			wrapBackTicks(tblIR.TableName()), selectedField)
+	} else {
+		insertStatementPrefix = fmt.Sprintf("INSERT INTO %s VALUES\n",
+			wrapBackTicks(tblIR.TableName()))
 	}
 
 	for fileRowIter.HasNextSQLRowIter() {
-		if err := write(w, insertStatementPrefix); err != nil {
-			return err
-		}
+		bf.WriteString(insertStatementPrefix)
 
 		fileRowIter = fileRowIter.NextSQLRowIter()
 		for fileRowIter.HasNext() {
-			if err := fileRowIter.Next(row); err != nil {
+			if err = fileRowIter.Decode(row); err != nil {
 				log.Zap().Error("scanning from sql.Row failed", zap.Error(err))
 				return err
 			}
 
-			if err := write(w, row.ToString(escapeBackSlash)); err != nil {
-				return err
-			}
+			row.WriteToBuffer(bf, escapeBackSlash)
 			counter += 1
 
-			var splitter string
+			if bf.Len() >= lengthLimit {
+				wp.input <- bf.Bytes()
+				bf.Reset()
+			}
+
+			fileRowIter.Next()
 			if fileRowIter.HasNext() {
-				splitter = ","
+				bf.WriteString(",\n")
 			} else {
-				splitter = ";"
+				bf.WriteString(";\n")
 			}
-			if err := write(w, fmt.Sprintf("%s\n", splitter)); err != nil {
+
+			if err = wp.Error(); err != nil {
 				return err
 			}
 		}
 	}
-
 	log.Zap().Debug("dumping table",
 		zap.String("table", tblIR.TableName()),
 		zap.Int("record counts", counter))
-	return nil
+	if bf.Len() > 0 {
+		wp.input <- bf.Bytes()
+		bf.Reset()
+	}
+	close(wp.input)
+	<-wp.closed
+	pool.Put(bf)
+	return wp.Error()
 }
 
 func write(writer io.StringWriter, str string) error {
 	_, err := writer.WriteString(str)
 	if err != nil {
+		// str might be very long, only output the first 200 chars
+		outputLength := len(str)
+		if outputLength >= 200 {
+			outputLength = 200
+		}
+		log.Zap().Error("writing failed",
+			zap.String("string", str[:outputLength]),
+			zap.Error(err))
+	}
+	return err
+}
+
+func writeBytes(writer io.Writer, p []byte) error {
+	_, err := writer.Write(p)
+	if err != nil {
+		// str might be very long, only output the first 200 chars
+		outputLength := len(p)
+		if outputLength >= 200 {
+			outputLength = 200
+		}
 		log.Zap().Error("writing failed",
-			zap.String("string", str),
+			zap.ByteString("string", p[:outputLength]),
 			zap.Error(err))
 	}
 	return err
@@ -125,10 +230,9 @@ func buildFileWriter(path string) (io.StringWriter, func(), error) {
 	return buf, tearDownRoutine, nil
 }
 
-func buildLazyFileWriter(path string) (io.StringWriter, func()) {
+func buildInterceptFileWriter(path string) (io.Writer, func()) {
 	var file *os.File
-	var buf *bufio.Writer
-	lazyStringWriter := &LazyStringWriter{}
+	fileWriter := &InterceptFileWriter{}
 	initRoutine := func() error {
 		f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0755)
 		file = f
@@ -138,25 +242,23 @@ func buildLazyFileWriter(path string) (io.StringWriter, func()) {
 				zap.Error(err))
 		}
 		log.Zap().Debug("opened file", zap.String("path", path))
-		buf = bufio.NewWriter(file)
-		lazyStringWriter.StringWriter = buf
+		fileWriter.Writer = file
 		return err
 	}
-	lazyStringWriter.initRoutine = initRoutine
+	fileWriter.initRoutine = initRoutine
 
 	tearDownRoutine := func() {
 		if file == nil {
 			return
 		}
 		log.Zap().Debug("tear down lazy file writer...")
-		_ = buf.Flush()
 		err := file.Close()
 		if err == nil {
 			return
 		}
 		log.Zap().Error("close file failed", zap.String("path", path))
 	}
-	return lazyStringWriter, tearDownRoutine
+	return fileWriter, tearDownRoutine
 }
 
 type LazyStringWriter struct {
@@ -174,18 +276,26 @@ func (l *LazyStringWriter) WriteString(str string) (int, error) {
 	return l.StringWriter.WriteString(str)
 }
 
-// InterceptStringWriter is an interceptor of io.StringWriter,
+// InterceptFileWriter is an interceptor of os.File,
 // tracking whether a StringWriter has written something.
-type InterceptStringWriter struct {
-	io.StringWriter
+type InterceptFileWriter struct {
+	io.Writer
+	sync.Once
+	initRoutine func() error
+	err         error
+
 	SomethingIsWritten bool
 }
 
-func (w *InterceptStringWriter) WriteString(str string) (int, error) {
-	if len(str) > 0 {
+func (w *InterceptFileWriter) Write(p []byte) (int, error) {
+	w.Do(func() { w.err = w.initRoutine() })
+	if len(p) > 0 {
 		w.SomethingIsWritten = true
 	}
-	return w.StringWriter.WriteString(str)
+	if w.err != nil {
+		return 0, fmt.Errorf("open file error: %s", w.err.Error())
+	}
+	return w.Writer.Write(p)
 }
 
 func wrapBackTicks(identifier string) string {


Test patch:
diff --git a/tests/primary_key/result/pk_case_0.sql b/tests/primary_key/result/pk_case_0.sql
index 60fd653..6c8c7fb 100644
--- a/tests/primary_key/result/pk_case_0.sql
+++ b/tests/primary_key/result/pk_case_0.sql
@@ -1,13 +1,13 @@
 /*!40101 SET NAMES binary*/;
 INSERT INTO `pk_case_0` VALUES
-(0, 10),
-(1, 9),
-(2, 8),
-(3, 7),
-(4, 6),
-(5, 5),
-(6, 4),
-(7, 3),
-(8, 2),
-(9, 1),
-(10, 0);
+(0,10),
+(1,9),
+(2,8),
+(3,7),
+(4,6),
+(5,5),
+(6,4),
+(7,3),
+(8,2),
+(9,1),
+(10,0);
diff --git a/tests/primary_key/result/pk_case_1.sql b/tests/primary_key/result/pk_case_1.sql
index d940c19..4a671cc 100644
--- a/tests/primary_key/result/pk_case_1.sql
+++ b/tests/primary_key/result/pk_case_1.sql
@@ -1,13 +1,13 @@
 /*!40101 SET NAMES binary*/;
 INSERT INTO `pk_case_1` VALUES
-(0, 10),
-(1, 9),
-(2, 8),
-(3, 7),
-(4, 6),
-(5, 5),
-(6, 4),
-(7, 3),
-(8, 2),
-(9, 1),
-(10, 0);
+(0,10),
+(1,9),
+(2,8),
+(3,7),
+(4,6),
+(5,5),
+(6,4),
+(7,3),
+(8,2),
+(9,1),
+(10,0);
diff --git a/tests/primary_key/result/pk_case_2.sql b/tests/primary_key/result/pk_case_2.sql
index 2364bb3..af9494d 100644
--- a/tests/primary_key/result/pk_case_2.sql
+++ b/tests/primary_key/result/pk_case_2.sql
@@ -1,13 +1,13 @@
 /*!40101 SET NAMES binary*/;
 INSERT INTO `pk_case_2` VALUES
-(0, 10),
-(1, 9),
-(2, 8),
-(3, 7),
-(4, 6),
-(5, 5),
-(6, 4),
-(7, 3),
-(8, 2),
-(9, 1),
-(10, 0);
+(0,10),
+(1,9),
+(2,8),
+(3,7),
+(4,6),
+(5,5),
+(6,4),
+(7,3),
+(8,2),
+(9,1),
+(10,0);
diff --git a/tests/primary_key/result/pk_case_3.sql b/tests/primary_key/result/pk_case_3.sql
index 42c6cb6..0698d1a 100644
--- a/tests/primary_key/result/pk_case_3.sql
+++ b/tests/primary_key/result/pk_case_3.sql
@@ -1,13 +1,13 @@
 /*!40101 SET NAMES binary*/;
 INSERT INTO `pk_case_3` VALUES
-(6, 4),
-(4, 6),
-(8, 2),
-(3, 7),
-(1, 9),
-(2, 8),
-(5, 5),
-(10, 0),
-(0, 10),
-(9, 1),
-(7, 3);
+(6,4),
+(4,6),
+(8,2),
+(3,7),
+(1,9),
+(2,8),
+(5,5),
+(10,0),
+(0,10),
+(9,1),
+(7,3);
diff --git a/v4/export/dump_test.go b/v4/export/dump_test.go
index 8278f43..7163f85 100644
--- a/v4/export/dump_test.go
+++ b/v4/export/dump_test.go
@@ -106,10 +106,12 @@ func (s *testDumpSuite) TestDumpTable(c *C) {
 	rowIter := tbDataRes.Rows()
 	c.Assert(rowIter.HasNext(), IsTrue)
 	receiver := newSimpleRowReceiver(1)
-	c.Assert(rowIter.Next(receiver), IsNil)
+	c.Assert(rowIter.Decode(receiver), IsNil)
 	c.Assert(receiver.data[0], Equals, "1")
-	c.Assert(rowIter.Next(receiver), IsNil)
+	rowIter.Next()
+	c.Assert(rowIter.Decode(receiver), IsNil)
 	c.Assert(receiver.data[0], Equals, "2")
+	rowIter.Next()
 	c.Assert(mock.ExpectationsWereMet(), IsNil)
 }
 
@@ -154,8 +156,9 @@ func (s *testDumpSuite) TestDumpTableWhereClause(c *C) {
 
 	for i := 4; i < 9; i++ {
 		c.Assert(rowIter.HasNext(), IsTrue)
-		c.Assert(rowIter.Next(receiver), IsNil)
+		c.Assert(rowIter.Decode(receiver), IsNil)
 		c.Assert(receiver.data[0], Equals, strconv.Itoa(i))
+		rowIter.Next()
 	}
 	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 9fd3f35..4a981b8 100644
--- a/v4/export/ir_impl_test.go
+++ b/v4/export/ir_impl_test.go
@@ -51,14 +51,17 @@ func (s *testIRImplSuite) TestRowIter(c *C) {
 		c.Assert(iter.HasNext(), IsTrue)
 	}
 	res := newSimpleRowReceiver(1)
-	c.Assert(iter.Next(res), IsNil)
+	c.Assert(iter.Decode(res), IsNil)
 	c.Assert(res.data, DeepEquals, []string{"1"})
+	iter.Next()
 	c.Assert(iter.HasNext(), IsTrue)
 	c.Assert(iter.HasNext(), IsTrue)
-	c.Assert(iter.Next(res), IsNil)
+	c.Assert(iter.Decode(res), IsNil)
 	c.Assert(res.data, DeepEquals, []string{"2"})
+	iter.Next()
 	c.Assert(iter.HasNext(), IsTrue)
-	c.Assert(iter.Next(res), IsNil)
+	c.Assert(iter.Decode(res), IsNil)
+	iter.Next()
 	c.Assert(res.data, DeepEquals, []string{"3"})
 	c.Assert(iter.HasNext(), IsFalse)
 }
@@ -105,7 +108,8 @@ func (s *testIRImplSuite) TestChunkRowIter(c *C) {
 		c.Assert(ok, IsTrue)
 
 		for sqlRowIter.HasNext() {
-			c.Assert(sqlRowIter.Next(res), IsNil)
+			c.Assert(sqlRowIter.Decode(res), IsNil)
+			sqlRowIter.Next()
 			resSize = append(resSize, []uint64{fileRowIter.currentFileSize, fileRowIter.currentStatementSize})
 		}
 	}
@@ -114,5 +118,6 @@ func (s *testIRImplSuite) TestChunkRowIter(c *C) {
 	c.Assert(sqlRowIter.HasNextSQLRowIter(), IsFalse)
 	c.Assert(sqlRowIter.HasNext(), IsFalse)
 	rows.Close()
-	c.Assert(sqlRowIter.Next(res), NotNil)
+	c.Assert(sqlRowIter.Decode(res), NotNil)
+	sqlRowIter.Next()
 }
diff --git a/v4/export/sql_type_test.go b/v4/export/sql_type_test.go
index 68ba9ff..3fd5fff 100644
--- a/v4/export/sql_type_test.go
+++ b/v4/export/sql_type_test.go
@@ -1,15 +1,23 @@
 package export
 
-import . "github.com/pingcap/check"
+import (
+	"bytes"
+
+	. "github.com/pingcap/check"
+)
 
 var _ = Suite(&testSqlByteSuite{})
 
 type testSqlByteSuite struct{}
 
 func (s *testSqlByteSuite) TestEscape(c *C) {
-	str := `MWQeWw""'\rNmtGxzGp`
+	var bf bytes.Buffer
+	str := []byte(`MWQeWw""'\rNmtGxzGp`)
 	expectStrBackslash := `MWQeWw\"\"\'\\rNmtGxzGp`
 	expectStrWithoutBackslash := `MWQeWw""''\rNmtGxzGp`
-	c.Assert(escape(str, true), Equals, expectStrBackslash)
-	c.Assert(escape(str, false), Equals, expectStrWithoutBackslash)
+	escape(str, &bf, true)
+	c.Assert(bf.String(), Equals, expectStrBackslash)
+	bf.Reset()
+	escape(str, &bf, false)
+	c.Assert(bf.String(), Equals, expectStrWithoutBackslash)
 }
diff --git a/v4/export/writer_test.go b/v4/export/writer_test.go
index 3e6b8d3..3e7524d 100644
--- a/v4/export/writer_test.go
+++ b/v4/export/writer_test.go
@@ -93,10 +93,10 @@ func (s *testDumpSuite) TestWriteTableData(c *C) {
 	expected := "/*!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" +
-		"(2, 'female', 'sarah@mail.com', '020-1253', 'healthy'),\n" +
-		"(3, 'male', 'john@mail.com', '020-1256', 'healthy'),\n" +
-		"(4, 'female', 'sarah@mail.com', '020-1235', 'healthy');\n"
+		"(1,'male','bob@mail.com','020-1234',NULL),\n" +
+		"(2,'female','sarah@mail.com','020-1253','healthy'),\n" +
+		"(3,'male','john@mail.com','020-1256','healthy'),\n" +
+		"(4,'female','sarah@mail.com','020-1235','healthy');\n"
 	c.Assert(string(bytes), Equals, expected)
 }
 
@@ -132,13 +132,13 @@ func (s *testDumpSuite) TestWriteTableDataWithFileSize(c *C) {
 		"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" +
-			"(2, 'female', 'sarah@mail.com', '020-1253', 'healthy');\n",
+			"(1,'male','bob@mail.com','020-1234',NULL),\n" +
+			"(2,'female','sarah@mail.com','020-1253','healthy');\n",
 		"test.employee.1.sql": "/*!40101 SET NAMES binary*/;\n" +
 			"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
 			"INSERT INTO `employee` VALUES\n" +
-			"(3, 'male', 'john@mail.com', '020-1256', 'healthy'),\n" +
-			"(4, 'female', 'sarah@mail.com', '020-1235', 'healthy');\n",
+			"(3,'male','john@mail.com','020-1256','healthy'),\n" +
+			"(4,'female','sarah@mail.com','020-1235','healthy');\n",
 	}
 
 	for p, expected := range cases {
@@ -184,11 +184,11 @@ func (s *testDumpSuite) TestWriteTableDataWithStatementSize(c *C) {
 		"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" +
-			"(2, 'female', 'sarah@mail.com', '020-1253', 'healthy');\n" +
+			"(1,'male','bob@mail.com','020-1234',NULL),\n" +
+			"(2,'female','sarah@mail.com','020-1253','healthy');\n" +
 			"INSERT INTO `employee` VALUES\n" +
-			"(3, 'male', 'john@mail.com', '020-1256', 'healthy'),\n" +
-			"(4, 'female', 'sarah@mail.com', '020-1235', 'healthy');\n",
+			"(3,'male','john@mail.com','020-1256','healthy'),\n" +
+			"(4,'female','sarah@mail.com','020-1235','healthy');\n",
 	}
 
 	for p, expected := range cases {
@@ -212,14 +212,14 @@ func (s *testDumpSuite) TestWriteTableDataWithStatementSize(c *C) {
 		"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" +
-			"(2, 'female', 'sarah@mail.com', '020-1253', 'healthy');\n" +
+			"(1,'male','bob@mail.com','020-1234',NULL),\n" +
+			"(2,'female','sarah@mail.com','020-1253','healthy');\n" +
 			"INSERT INTO `employee` VALUES\n" +
-			"(3, 'male', 'john@mail.com', '020-1256', 'healthy');\n",
+			"(3,'male','john@mail.com','020-1256','healthy');\n",
 		"test.employee.1.sql": "/*!40101 SET NAMES binary*/;\n" +
 			"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;\n" +
 			"INSERT INTO `employee` VALUES\n" +
-			"(4, 'female', 'sarah@mail.com', '020-1235', 'healthy');\n",
+			"(4,'female','sarah@mail.com','020-1235','healthy');\n",
 	}
 
 	tableIR = newMockTableIR("test", "employee", data, specCmts, colTypes)
diff --git a/v4/export/writer_util_test.go b/v4/export/writer_util_test.go
index 281171d..2c0b04e 100644
--- a/v4/export/writer_util_test.go
+++ b/v4/export/writer_util_test.go
@@ -1,11 +1,13 @@
 package export
 
 import (
+	"bytes"
 	"fmt"
 	"strings"
 	"testing"
 
 	"database/sql/driver"
+
 	. "github.com/pingcap/check"
 )
 
@@ -55,18 +57,18 @@ func (s *testUtilSuite) TestWriteInsert(c *C) {
 		"/*!40014 SET FOREIGN_KEY_CHECKS=0*/;",
 	}
 	tableIR := newMockTableIR("test", "employee", data, specCmts, colTypes)
-	strCollector := &mockStringCollector{}
+	bf := &bytes.Buffer{}
 
-	err := WriteInsert(tableIR, strCollector)
+	err := WriteInsert(tableIR, bf)
 	c.Assert(err, IsNil)
 	expected := "/*!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" +
-		"(2, 'female', 'sarah@mail.com', '020-1253', 'healthy'),\n" +
-		"(3, 'male', 'john@mail.com', '020-1256', 'healthy'),\n" +
-		"(4, 'female', 'sarah@mail.com', '020-1235', 'healthy');\n"
-	c.Assert(strCollector.buf, Equals, expected)
+		"(1,'male','bob@mail.com','020-1234',NULL),\n" +
+		"(2,'female','sarah@mail.com','020-1253','healthy'),\n" +
+		"(3,'male','john@mail.com','020-1256','healthy'),\n" +
+		"(4,'female','sarah@mail.com','020-1235','healthy');\n"
+	c.Assert(bf.String(), Equals, expected)
 }
 
 func (s *testUtilSuite) TestSQLDataTypes(c *C) {
@@ -82,11 +84,11 @@ func (s *testUtilSuite) TestSQLDataTypes(c *C) {
 		tableData := [][]driver.Value{{origin}}
 		colType := []string{sqlType}
 		tableIR := newMockTableIR("test", "t", tableData, nil, colType)
-		strCollector := &mockStringCollector{}
+		bf := &bytes.Buffer{}
 
-		err := WriteInsert(tableIR, strCollector)
+		err := WriteInsert(tableIR, bf)
 		c.Assert(err, IsNil)
-		lines := strings.Split(strCollector.buf, "\n")
+		lines := strings.Split(bf.String(), "\n")
 		c.Assert(len(lines), Equals, 3)
 		c.Assert(lines[1], Equals, fmt.Sprintf("(%s);", result))
 	}