Repository: pingcap/tidb-lightning
Pull request #406 (2020-09-23 08:40:07)

mydump: support multi bytes csv delimiter and separator
<!--
Thank you for contributing to TiDB-Lightning! Please read the [CONTRIBUTING](https://github.com/pingcap/tidb-lightning/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve? <!--add issue link with summary if exists-->
partially reslove #321

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)
 - No code

Side effects

 - Possible performance regression
 - Increased code complexity

Related changes

 - Need to update the documentation
 - Need to be included in the release note

glorv: /run-all-tests
djshow832: /run-all-tests tidb=pr/20191
djshow832: /run-all-tests tidb=pr/20191

Patch:
diff --git a/lightning/config/config.go b/lightning/config/config.go
index ca46c31..27559b2 100644
--- a/lightning/config/config.go
+++ b/lightning/config/config.go
@@ -384,16 +384,12 @@ func (cfg *Config) LoadFromTOML(data []byte) error {
 func (cfg *Config) Adjust() error {
 	// Reject problematic CSV configurations.
 	csv := &cfg.Mydumper.CSV
-	if len(csv.Separator) != 1 {
-		return errors.New("invalid config: `mydumper.csv.separator` must be exactly one byte long")
+	if len(csv.Separator) == 0 {
+		return errors.New("invalid config: `mydumper.csv.separator` must not be empty")
 	}
 
-	if len(csv.Delimiter) > 1 {
-		return errors.New("invalid config: `mydumper.csv.delimiter` must be one byte long or empty")
-	}
-
-	if csv.Separator == csv.Delimiter {
-		return errors.New("invalid config: cannot use the same character for both CSV delimiter and separator")
+	if len(csv.Delimiter) > 0 && (strings.HasPrefix(csv.Separator, csv.Delimiter) || strings.HasPrefix(csv.Delimiter, csv.Separator)) {
+		return errors.New("invalid config: `mydumper.csv.separator` and `mydumper.csv.delimiter` must not be prefix of each other")
 	}
 
 	if csv.BackslashEscape {
diff --git a/lightning/mydump/bytes.go b/lightning/mydump/bytes.go
index 3ce5d3e..d110275 100644
--- a/lightning/mydump/bytes.go
+++ b/lightning/mydump/bytes.go
@@ -9,43 +9,31 @@
 
 package mydump
 
-import "unicode/utf8"
+// byteSet is a 32-byte value, where each bit represents the presence of a
+// given byte value in the set.
+type byteSet [8]uint32
 
-// asciiSet is a 32-byte value, where each bit represents the presence of a
-// given ASCII character in the set. The 128-bits of the lower 16 bytes,
-// starting with the least-significant bit of the lowest word to the
-// most-significant bit of the highest word, map to the full range of all
-// 128 ASCII characters. The 128-bits of the upper 16 bytes will be zeroed,
-// ensuring that any non-ASCII character will be reported as not in the set.
-type asciiSet [8]uint32
-
-// makeASCIISet creates a set of ASCII characters and reports whether all
-// characters in chars are ASCII.
-func makeASCIISet(chars string) (as asciiSet, ok bool) {
+// makeByteSet creates a set of byte value.
+func makeByteSet(chars []byte) (as byteSet) {
 	for i := 0; i < len(chars); i++ {
 		c := chars[i]
-		if c >= utf8.RuneSelf {
-			return as, false
-		}
 		as[c>>5] |= 1 << uint(c&31)
 	}
-	return as, true
+	return as
 }
 
 // contains reports whether c is inside the set.
-func (as *asciiSet) contains(c byte) bool {
+func (as *byteSet) contains(c byte) bool {
 	return (as[c>>5] & (1 << uint(c&31))) != 0
 }
 
-// IndexAnyAscii returns the byte index of the first occurrence in s of any of the Unicode
-// code points in chars. It returns -1 if  there is no code
-// point in common.
-func IndexAnyAscii(s []byte, as *asciiSet) int {
+// IndexAnyByte returns the byte index of the first occurrence in s of any of the byte
+// points in chars. It returns -1 if  there is no code point in common.
+func IndexAnyByte(s []byte, as *byteSet) int {
 	for i, c := range s {
 		if as.contains(c) {
 			return i
 		}
 	}
 	return -1
-
 }
diff --git a/lightning/mydump/csv_parser.go b/lightning/mydump/csv_parser.go
index 3d368c5..0cb5bff 100644
--- a/lightning/mydump/csv_parser.go
+++ b/lightning/mydump/csv_parser.go
@@ -19,6 +19,8 @@ import (
 	"strings"
 	"unicode"
 
+	"github.com/pingcap/br/pkg/utils"
+
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/types"
 
@@ -38,8 +40,8 @@ type CSVParser struct {
 	cfg       *config.CSVConfig
 	escFlavor backslashEscapeFlavor
 
-	comma            byte
-	quote            byte
+	comma            []byte
+	quote            []byte
 	quoteIndexFunc   func([]byte) int
 	unquoteIndexFunc func([]byte) int
 
@@ -66,18 +68,17 @@ func NewCSVParser(
 	ioWorkers *worker.Pool,
 	shouldParseHeader bool,
 ) *CSVParser {
-	quote := byte(0)
+	escFlavor := backslashEscapeFlavorNone
+	var quoteStopSet []byte
+	unquoteStopSet := []byte{'\r', '\n', cfg.Separator[0]}
 	if len(cfg.Delimiter) > 0 {
-		quote = cfg.Delimiter[0]
+		quoteStopSet = []byte{cfg.Delimiter[0]}
+		unquoteStopSet = append(unquoteStopSet, cfg.Delimiter[0])
 	}
-
-	escFlavor := backslashEscapeFlavorNone
-	quoteStopSet := cfg.Delimiter
-	unquoteStopSet := "\r\n" + cfg.Separator + cfg.Delimiter
 	if cfg.BackslashEscape {
 		escFlavor = backslashEscapeFlavorMySQL
-		quoteStopSet += `\`
-		unquoteStopSet += `\`
+		quoteStopSet = append(quoteStopSet, '\\')
+		unquoteStopSet = append(unquoteStopSet, '\\')
 		// we need special treatment of the NULL value \N, used by MySQL.
 		if !cfg.NotNull && cfg.Null == `\N` {
 			escFlavor = backslashEscapeFlavorMySQLWithNull
@@ -87,8 +88,8 @@ func NewCSVParser(
 	return &CSVParser{
 		blockParser:       makeBlockParser(reader, blockBufSize, ioWorkers),
 		cfg:               cfg,
-		comma:             cfg.Separator[0],
-		quote:             quote,
+		comma:             []byte(cfg.Separator),
+		quote:             []byte(cfg.Delimiter),
 		escFlavor:         escFlavor,
 		quoteIndexFunc:    makeBytesIndexFunc(quoteStopSet),
 		unquoteIndexFunc:  makeBytesIndexFunc(unquoteStopSet),
@@ -96,11 +97,11 @@ func NewCSVParser(
 	}
 }
 
-func makeBytesIndexFunc(chars string) func([]byte) int {
+func makeBytesIndexFunc(chars []byte) func([]byte) int {
 	// chars are guaranteed to be ascii str, so this call will always success
-	as, _ := makeASCIISet(chars)
+	as := makeByteSet(chars)
 	return func(s []byte) int {
-		return IndexAnyAscii(s, &as)
+		return IndexAnyByte(s, &as)
 	}
 }
 
@@ -130,21 +131,43 @@ func (parser *CSVParser) readByte() (byte, error) {
 	return b, nil
 }
 
-func (parser *CSVParser) peekByte() (byte, error) {
-	if len(parser.buf) == 0 {
+func (parser *CSVParser) readBytes(buf []byte) (int, error) {
+	cnt := 0
+	for cnt < len(buf) {
+		if len(parser.buf) == 0 {
+			if err := parser.readBlock(); err != nil {
+				return cnt, err
+			}
+		}
+		if len(parser.buf) == 0 {
+			parser.pos += int64(cnt)
+			return cnt, io.EOF
+		}
+		readCnt := utils.MinInt(len(buf)-cnt, len(parser.buf))
+		copy(buf[cnt:], parser.buf[:readCnt])
+		parser.buf = parser.buf[readCnt:]
+		cnt += readCnt
+	}
+	parser.pos += int64(cnt)
+	return cnt, nil
+}
+
+func (parser *CSVParser) peekBytes(cnt int) ([]byte, error) {
+	if len(parser.buf) < cnt {
 		if err := parser.readBlock(); err != nil {
-			return 0, err
+			return nil, err
 		}
 	}
 	if len(parser.buf) == 0 {
-		return 0, io.EOF
+		return nil, io.EOF
 	}
-	return parser.buf[0], nil
+	cnt = utils.MinInt(cnt, len(parser.buf))
+	return parser.buf[:cnt], nil
 }
 
-func (parser *CSVParser) skipByte() {
-	parser.buf = parser.buf[1:]
-	parser.pos++
+func (parser *CSVParser) skipBytes(n int) {
+	parser.buf = parser.buf[n:]
+	parser.pos += int64(n)
 }
 
 // readUntil reads the buffer until any character from the `chars` set is found.
@@ -186,6 +209,58 @@ func (parser *CSVParser) readRecord(dst []string) ([]string, error) {
 
 	isEmptyLine := true
 	whitespaceLine := true
+
+	processDefault := func(b byte) error {
+		if b == '\\' && parser.escFlavor != backslashEscapeFlavorNone {
+			if err := parser.readByteForBackslashEscape(); err != nil {
+				return err
+			}
+		} else {
+			parser.recordBuffer = append(parser.recordBuffer, b)
+		}
+		return parser.readUnquoteField()
+	}
+
+	processQuote := func(b byte) error {
+		return parser.readQuotedField()
+	}
+	if len(parser.quote) > 1 {
+		processQuote = func(b byte) error {
+			pb, err := parser.peekBytes(len(parser.quote) - 1)
+			if err != nil && errors.Cause(err) != io.EOF {
+				return err
+			}
+			if bytes.Equal(pb, parser.quote[1:]) {
+				parser.skipBytes(len(parser.quote) - 1)
+				return parser.readQuotedField()
+			}
+			return processDefault(b)
+		}
+	}
+
+	processComma := func(b byte) error {
+		parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
+		return nil
+	}
+	if len(parser.comma) > 1 {
+		processNotComma := processDefault
+		if len(parser.quote) > 0 && parser.comma[0] == parser.quote[0] {
+			processNotComma = processQuote
+		}
+		processComma = func(b byte) error {
+			pb, err := parser.peekBytes(len(parser.comma) - 1)
+			if err != nil && errors.Cause(err) != io.EOF {
+				return err
+			}
+			if bytes.Equal(pb, parser.comma[1:]) {
+				parser.skipBytes(len(parser.comma) - 1)
+				parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
+				return nil
+			}
+			return processNotComma(b)
+		}
+	}
+
 outside:
 	for {
 		firstByte, err := parser.readByte()
@@ -197,17 +272,19 @@ outside:
 			firstByte = '\n'
 		}
 
-		switch firstByte {
-		case parser.comma:
-			parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
+		switch {
+		case firstByte == parser.comma[0]:
 			whitespaceLine = false
-		case parser.quote:
-			if err := parser.readQuotedField(); err != nil {
+			if err = processComma(firstByte); err != nil {
 				return nil, err
 			}
-			whitespaceLine = false
 
-		case '\r', '\n':
+		case len(parser.quote) > 0 && firstByte == parser.quote[0]:
+			if err = processQuote(firstByte); err != nil {
+				return nil, err
+			}
+			whitespaceLine = false
+		case firstByte == '\r', firstByte == '\n':
 			// new line = end of record (ignore empty lines)
 			if isEmptyLine {
 				continue
@@ -219,22 +296,13 @@ outside:
 			}
 			parser.fieldIndexes = append(parser.fieldIndexes, len(parser.recordBuffer))
 			break outside
-
 		default:
-			if firstByte == '\\' && parser.escFlavor != backslashEscapeFlavorNone {
-				if err := parser.readByteForBackslashEscape(); err != nil {
-					return nil, err
-				}
-			} else {
-				parser.recordBuffer = append(parser.recordBuffer, firstByte)
-			}
-			if err := parser.readUnquoteField(); err != nil {
+			if err = processDefault(firstByte); err != nil {
 				return nil, err
 			}
 		}
 		isEmptyLine = false
 	}
-
 	// Create a single string and create slices out of it.
 	// This pins the memory of the fields together, but allocates once.
 	str := string(parser.recordBuffer) // Convert to string once to batch allocations
@@ -264,6 +332,25 @@ func (parser *CSVParser) readByteForBackslashEscape() error {
 }
 
 func (parser *CSVParser) readQuotedField() error {
+	processDefault := func() error {
+		// in all other cases, we've got a syntax error.
+		parser.logSyntaxError()
+		return errors.AddStack(errUnexpectedQuoteField)
+	}
+
+	processComma := func() error { return nil }
+	if len(parser.comma) > 1 {
+		processComma = func() error {
+			b, err := parser.peekBytes(len(parser.comma))
+			if err != nil && errors.Cause(err) != io.EOF {
+				return err
+			}
+			if !bytes.Equal(b, parser.comma) {
+				return processDefault()
+			}
+			return nil
+		}
+	}
 	for {
 		content, terminator, err := parser.readUntil(parser.quoteIndexFunc)
 		err = parser.replaceEOF(err, errUnterminatedQuotedField)
@@ -271,29 +358,56 @@ func (parser *CSVParser) readQuotedField() error {
 			return err
 		}
 		parser.recordBuffer = append(parser.recordBuffer, content...)
-		parser.skipByte()
-		switch terminator {
-		case parser.quote:
+		parser.skipBytes(1)
+		switch {
+		case len(parser.quote) > 0 && terminator == parser.quote[0]:
+			if len(parser.quote) > 1 {
+				b, err := parser.peekBytes(len(parser.quote) - 1)
+				if err != nil && err != io.EOF {
+					return err
+				}
+				if !bytes.Equal(b, parser.quote[1:]) {
+					parser.recordBuffer = append(parser.recordBuffer, terminator)
+					continue
+				}
+				parser.skipBytes(len(parser.quote) - 1)
+			}
 			// encountered '"' -> continue if we're seeing '""'.
-			b, err := parser.peekByte()
-			err = parser.replaceEOF(err, nil)
+			b, err := parser.peekBytes(1)
 			if err != nil {
+				if err == io.EOF {
+					err = nil
+				}
 				return err
 			}
-			switch b {
-			case parser.quote:
+			switch b[0] {
+			case parser.quote[0]:
 				// consume the double quotation mark and continue
-				parser.skipByte()
+				if len(parser.quote) > 1 {
+					b, err := parser.peekBytes(len(parser.quote))
+					if err != nil && err != io.EOF {
+						return err
+					}
+					if !bytes.Equal(b, parser.quote) {
+						if parser.quote[0] == parser.comma[0] {
+							return processComma()
+						} else {
+							return processDefault()
+						}
+					}
+				}
+				parser.skipBytes(len(parser.quote))
 				parser.recordBuffer = append(parser.recordBuffer, '"')
-			case '\r', '\n', parser.comma, 0:
+			case '\r', '\n':
 				// end the field if the next is a separator
 				return nil
+			case parser.comma[0]:
+				return processComma()
 			default:
-				// in all other cases, we've got a syntax error.
-				parser.logSyntaxError()
-				return errors.AddStack(errUnexpectedQuoteField)
+				return processDefault()
 			}
-		case '\\':
+
+		case terminator == '\\':
 			if err := parser.readByteForBackslashEscape(); err != nil {
 				return err
 			}
@@ -302,22 +416,70 @@ func (parser *CSVParser) readQuotedField() error {
 }
 
 func (parser *CSVParser) readUnquoteField() error {
+	addByte := func(b byte) {
+		// read the following byte
+		parser.recordBuffer = append(parser.recordBuffer, b)
+		parser.skipBytes(1)
+	}
+	parseQuote := func(b byte) error {
+		r, err := parser.checkBytes(parser.quote)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		if r {
+			parser.logSyntaxError()
+			return errors.AddStack(errUnexpectedQuoteField)
+		}
+		addByte(b)
+		return nil
+	}
+
+	parserNoComma := func(b byte) error {
+		addByte(b)
+		return nil
+	}
+	if len(parser.quote) > 0 && parser.comma[0] == parser.quote[0] {
+		parserNoComma = parseQuote
+	}
 	for {
 		content, terminator, err := parser.readUntil(parser.unquoteIndexFunc)
 		parser.recordBuffer = append(parser.recordBuffer, content...)
-		err = parser.replaceEOF(err, nil)
+		finished := false
 		if err != nil {
-			return err
+			if errors.Cause(err) == io.EOF {
+				finished = true
+				err = nil
+			}
+			if err != nil {
+				return err
+			}
 		}
 
-		switch terminator {
-		case '\r', '\n', parser.comma, 0:
+		switch {
+		case terminator == '\r', terminator == '\n', finished:
 			return nil
-		case parser.quote:
-			parser.logSyntaxError()
-			return errors.AddStack(errUnexpectedQuoteField)
-		case '\\':
-			parser.skipByte()
+		case terminator == parser.comma[0]:
+			r, err := parser.checkBytes(parser.comma)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if r {
+				return nil
+			}
+			if err = parserNoComma(terminator); err != nil {
+				return err
+			}
+		case len(parser.quote) > 0 && terminator == parser.quote[0]:
+			r, err := parser.checkBytes(parser.quote)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if r {
+				parser.logSyntaxError()
+				return errors.AddStack(errUnexpectedQuoteField)
+			}
+		case terminator == '\\':
+			parser.skipBytes(1)
 			if err := parser.readByteForBackslashEscape(); err != nil {
 				return err
 			}
@@ -325,6 +487,17 @@ func (parser *CSVParser) readUnquoteField() error {
 	}
 }
 
+func (parser *CSVParser) checkBytes(b []byte) (bool, error) {
+	if len(b) == 1 {
+		return true, nil
+	}
+	pb, err := parser.peekBytes(len(b))
+	if err != nil {
+		return false, err
+	}
+	return bytes.Equal(pb, b), nil
+}
+
 func (parser *CSVParser) replaceEOF(err error, replaced error) error {
 	if err == nil || errors.Cause(err) != io.EOF {
 		return err
@@ -394,16 +567,16 @@ func (parser *CSVParser) ReadColumns() error {
 	return nil
 }
 
-var newLineAsciiSet, _ = makeASCIISet("\r\n")
+var newLineAsciiSet = makeByteSet([]byte{'\r', '\n'})
 
 func indexOfNewLine(b []byte) int {
-	return IndexAnyAscii(b, &newLineAsciiSet)
+	return IndexAnyByte(b, &newLineAsciiSet)
 }
 func (parser *CSVParser) ReadUntilTokNewLine() (int64, error) {
 	_, _, err := parser.readUntil(indexOfNewLine)
 	if err != nil {
 		return 0, err
 	}
-	parser.skipByte()
+	parser.skipBytes(1)
 	return parser.pos, nil
 }
diff --git a/tidb-lightning.toml b/tidb-lightning.toml
index bbff2a1..e6932d5 100644
--- a/tidb-lightning.toml
+++ b/tidb-lightning.toml
@@ -148,9 +148,11 @@ filter = ['*.*']
 
 # CSV files are imported according to MySQL's LOAD DATA INFILE rules.
 [mydumper.csv]
-# separator between fields, should be an ASCII character.
+# separator between fields, can be one or more characters but empty. The value can
+# not be prefix of `delimiter`.
 separator = ','
-# string delimiter, can either be an ASCII character or empty string.
+# string delimiter, can either be one or more characters or empty string. If not empty,
+# the value should not be prefix of `separator`
 delimiter = '"'
 # whether the CSV files contain a header. If true, the first line will be skipped
 header = true


Test patch:
diff --git a/lightning/config/config_test.go b/lightning/config/config_test.go
index 1626cb1..d9e36df 100644
--- a/lightning/config/config_test.go
+++ b/lightning/config/config_test.go
@@ -275,14 +275,23 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 				[mydumper.csv]
 				separator = ''
 			`,
-			err: "invalid config: `mydumper.csv.separator` must be exactly one byte long",
+			err: "invalid config: `mydumper.csv.separator` must not be empty",
 		},
 		{
 			input: `
 				[mydumper.csv]
 				separator = 'hello'
+				delimiter = 'hel'
 			`,
-			err: "invalid config: `mydumper.csv.separator` must be exactly one byte long",
+			err: "invalid config: `mydumper.csv.separator` and `mydumper.csv.delimiter` must not be prefix of each other",
+		},
+		{
+			input: `
+				[mydumper.csv]
+				separator = 'hel'
+				delimiter = 'hello'
+			`,
+			err: "invalid config: `mydumper.csv.separator` and `mydumper.csv.delimiter` must not be prefix of each other",
 		},
 		{
 			input: `
@@ -297,7 +306,7 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 				[mydumper.csv]
 				separator = '，'
 			`,
-			err: "invalid config: `mydumper.csv.separator` must be exactly one byte long",
+			err: "",
 		},
 		{
 			input: `
@@ -311,7 +320,7 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 				[mydumper.csv]
 				delimiter = 'hello'
 			`,
-			err: "invalid config: `mydumper.csv.delimiter` must be one byte long or empty",
+			err: "",
 		},
 		{
 			input: `
@@ -324,9 +333,10 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 		{
 			input: `
 				[mydumper.csv]
-				delimiter = '“'
+				separator = '\s'
+				delimiter = '\d'
 			`,
-			err: "invalid config: `mydumper.csv.delimiter` must be one byte long or empty",
+			err: "",
 		},
 		{
 			input: `
@@ -334,7 +344,7 @@ func (s *configTestSuite) TestInvalidCSV(c *C) {
 				separator = '|'
 				delimiter = '|'
 			`,
-			err: "invalid config: cannot use the same character for both CSV delimiter and separator",
+			err: "invalid config: `mydumper.csv.separator` and `mydumper.csv.delimiter` must not be prefix of each other",
 		},
 		{
 			input: `
diff --git a/lightning/lightning_test.go b/lightning/lightning_test.go
index e334d6b..a45baf0 100644
--- a/lightning/lightning_test.go
+++ b/lightning/lightning_test.go
@@ -153,7 +153,7 @@ func (s *lightningServerSuite) TestRunServer(c *C) {
 	c.Assert(data["error"], Matches, "cannot parse task.*")
 	resp.Body.Close()
 
-	resp, err = http.Post(url, "application/toml", strings.NewReader("[mydumper.csv]\nseparator = 'fooo'"))
+	resp, err = http.Post(url, "application/toml", strings.NewReader("[mydumper.csv]\nseparator = 'fooo'\ndelimiter= 'foo'"))
 	c.Assert(err, IsNil)
 	c.Assert(resp.StatusCode, Equals, http.StatusBadRequest)
 	err = json.NewDecoder(resp.Body).Decode(&data)
diff --git a/lightning/mydump/csv_parser_test.go b/lightning/mydump/csv_parser_test.go
index 69dff19..144a5fa 100644
--- a/lightning/mydump/csv_parser_test.go
+++ b/lightning/mydump/csv_parser_test.go
@@ -74,12 +74,65 @@ func (s *testMydumpCSVParserSuite) runFailingTestCases(c *C, cfg *config.CSVConf
 	}
 }
 
+func tpchDatums() [][]types.Datum {
+	datums := make([][]types.Datum, 0, 3)
+	datums = append(datums, []types.Datum{
+		types.NewStringDatum("1"),
+		types.NewStringDatum("goldenrod lavender spring chocolate lace"),
+		types.NewStringDatum("Manufacturer#1"),
+		types.NewStringDatum("Brand#13"),
+		types.NewStringDatum("PROMO BURNISHED COPPER"),
+		types.NewStringDatum("7"),
+		types.NewStringDatum("JUMBO PKG"),
+		types.NewStringDatum("901.00"),
+		types.NewStringDatum("ly. slyly ironi"),
+	})
+	datums = append(datums, []types.Datum{
+		types.NewStringDatum("2"),
+		types.NewStringDatum("blush thistle blue yellow saddle"),
+		types.NewStringDatum("Manufacturer#1"),
+		types.NewStringDatum("Brand#13"),
+		types.NewStringDatum("LARGE BRUSHED BRASS"),
+		types.NewStringDatum("1"),
+		types.NewStringDatum("LG CASE"),
+		types.NewStringDatum("902.00"),
+		types.NewStringDatum("lar accounts amo"),
+	})
+	datums = append(datums, []types.Datum{
+		types.NewStringDatum("3"),
+		types.NewStringDatum("spring green yellow purple cornsilk"),
+		types.NewStringDatum("Manufacturer#4"),
+		types.NewStringDatum("Brand#42"),
+		types.NewStringDatum("STANDARD POLISHED BRASS"),
+		types.NewStringDatum("21"),
+		types.NewStringDatum("WRAP CASE"),
+		types.NewStringDatum("903.00"),
+		types.NewStringDatum("egular deposits hag"),
+	})
+
+	return datums
+}
+
+func datumsToString(datums [][]types.Datum, delimitor string, quote string, lastSep bool) string {
+	var b strings.Builder
+	for _, ds := range datums {
+		for i, d := range ds {
+			b.WriteString(quote)
+			b.WriteString(d.GetString())
+			b.WriteString(quote)
+			if lastSep || i < len(ds)-1 {
+				b.WriteString(delimitor)
+			}
+		}
+		b.WriteString("\r\n")
+	}
+	return b.String()
+}
+
 func (s *testMydumpCSVParserSuite) TestTPCH(c *C) {
-	reader := mydump.NewStringReader(
-		`1|goldenrod lavender spring chocolate lace|Manufacturer#1|Brand#13|PROMO BURNISHED COPPER|7|JUMBO PKG|901.00|ly. slyly ironi|
-2|blush thistle blue yellow saddle|Manufacturer#1|Brand#13|LARGE BRUSHED BRASS|1|LG CASE|902.00|lar accounts amo|
-3|spring green yellow purple cornsilk|Manufacturer#4|Brand#42|STANDARD POLISHED BRASS|21|WRAP CASE|903.00|egular deposits hag|
-`)
+	datums := tpchDatums()
+	input := datumsToString(datums, "|", "", true)
+	reader := mydump.NewStringReader(input)
 
 	cfg := config.CSVConfig{
 		Separator:   "|",
@@ -92,57 +145,83 @@ func (s *testMydumpCSVParserSuite) TestTPCH(c *C) {
 	c.Assert(parser.ReadRow(), IsNil)
 	c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
 		RowID: 1,
-		Row: []types.Datum{
-			types.NewStringDatum("1"),
-			types.NewStringDatum("goldenrod lavender spring chocolate lace"),
-			types.NewStringDatum("Manufacturer#1"),
-			types.NewStringDatum("Brand#13"),
-			types.NewStringDatum("PROMO BURNISHED COPPER"),
-			types.NewStringDatum("7"),
-			types.NewStringDatum("JUMBO PKG"),
-			types.NewStringDatum("901.00"),
-			types.NewStringDatum("ly. slyly ironi"),
-		},
+		Row:   datums[0],
 	})
 	c.Assert(parser, posEq, 126, 1)
 
 	c.Assert(parser.ReadRow(), IsNil)
 	c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
 		RowID: 2,
-		Row: []types.Datum{
-			types.NewStringDatum("2"),
-			types.NewStringDatum("blush thistle blue yellow saddle"),
-			types.NewStringDatum("Manufacturer#1"),
-			types.NewStringDatum("Brand#13"),
-			types.NewStringDatum("LARGE BRUSHED BRASS"),
-			types.NewStringDatum("1"),
-			types.NewStringDatum("LG CASE"),
-			types.NewStringDatum("902.00"),
-			types.NewStringDatum("lar accounts amo"),
-		},
+		Row:   datums[1],
 	})
-	c.Assert(parser, posEq, 240, 2)
+	c.Assert(parser, posEq, 241, 2)
 
 	c.Assert(parser.ReadRow(), IsNil)
 	c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
 		RowID: 3,
-		Row: []types.Datum{
-			types.NewStringDatum("3"),
-			types.NewStringDatum("spring green yellow purple cornsilk"),
-			types.NewStringDatum("Manufacturer#4"),
-			types.NewStringDatum("Brand#42"),
-			types.NewStringDatum("STANDARD POLISHED BRASS"),
-			types.NewStringDatum("21"),
-			types.NewStringDatum("WRAP CASE"),
-			types.NewStringDatum("903.00"),
-			types.NewStringDatum("egular deposits hag"),
-		},
+		Row:   datums[2],
 	})
-	c.Assert(parser, posEq, 367, 3)
+	c.Assert(parser, posEq, 369, 3)
 
 	c.Assert(errors.Cause(parser.ReadRow()), Equals, io.EOF)
 }
 
+func (s *testMydumpCSVParserSuite) TestTPCHMultiBytes(c *C) {
+	datums := tpchDatums()
+	deliAndSeps := [][2]string{
+		{",", ""},
+		{"\000", ""},
+		{"，", ""},
+		{"🤔", ""},
+		{"，", "。"},
+		{"||", ""},
+		{"|+|", ""},
+		{"##", ""},
+		{"，", "'"},
+		{"，", `"`},
+		{"🤔", `''`},
+		{"🤔", `"'`},
+		{"🤔", `"'`},
+		{"🤔", "🌚"}, // this two emoji have same prefix bytes
+		{"##", "#-"},
+		{"\\s", "\\q"},
+	}
+	for _, SepAndQuote := range deliAndSeps {
+		inputStr := datumsToString(datums, SepAndQuote[0], SepAndQuote[1], false)
+		cfg := config.CSVConfig{
+			Separator:   SepAndQuote[0],
+			Delimiter:   SepAndQuote[1],
+			TrimLastSep: false,
+		}
+
+		reader := mydump.NewStringReader(inputStr)
+		parser := mydump.NewCSVParser(&cfg, reader, config.ReadBlockSize, s.ioWorkers, false)
+		c.Assert(parser.ReadRow(), IsNil)
+		c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
+			RowID: 1,
+			Row:   datums[0],
+		})
+
+		c.Assert(parser, posEq, 117+8*len(SepAndQuote[0])+18*len(SepAndQuote[1]), 1)
+
+		c.Assert(parser.ReadRow(), IsNil)
+		c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
+			RowID: 2,
+			Row:   datums[1],
+		})
+		c.Assert(parser, posEq, 223+8*2*len(SepAndQuote[0])+18*2*len(SepAndQuote[1]), 2)
+
+		c.Assert(parser.ReadRow(), IsNil)
+		c.Assert(parser.LastRow(), DeepEquals, mydump.Row{
+			RowID: 3,
+			Row:   datums[2],
+		})
+		c.Assert(parser, posEq, 342+8*3*len(SepAndQuote[0])+18*3*len(SepAndQuote[1]), 3)
+
+		c.Assert(errors.Cause(parser.ReadRow()), Equals, io.EOF)
+	}
+}
+
 func (s *testMydumpCSVParserSuite) TestRFC4180(c *C) {
 	cfg := config.CSVConfig{
 		Separator: ",",
diff --git a/lightning/mydump/region_test.go b/lightning/mydump/region_test.go
index 18fecf1..bfb41be 100644
--- a/lightning/mydump/region_test.go
+++ b/lightning/mydump/region_test.go
@@ -15,10 +15,8 @@ package mydump_test
 
 import (
 	"context"
-	"fmt"
 	"log"
 	"os"
-	"path/filepath"
 
 	"github.com/pingcap/br/pkg/storage"
 
@@ -71,18 +69,6 @@ func (s *testMydumpRegionSuite) TestTableRegion(c *C) {
 		regions, err := MakeTableRegions(context.Background(), meta, 1, cfg, ioWorkers, loader.GetStore())
 		c.Assert(err, IsNil)
 
-		table := meta.Name
-		fmt.Printf("[%s] region count ===============> %d\n", table, len(regions))
-		for _, region := range regions {
-			fname := filepath.Base(region.FileMeta.Path)
-			fmt.Printf("[%s] rowID = %5d / rows = %5d / offset = %10d / size = %10d \n",
-				fname,
-				region.RowIDMin(),
-				region.Rows(),
-				region.Offset(),
-				region.Size())
-		}
-
 		// check - region-size vs file-size
 		var tolFileSize int64 = 0
 		for _, file := range meta.DataFiles {