lightning: split large csv file if possible (pingcap#272)

* lightning: split large csv file if possible * gofmt * gofmt * unit test * add unit test * tiny change * tiny refine * fix ci * remove useless code * fix ci * fix ci * address comments * go fmt for all * address comment * correct the estimateChunkCount Co-authored-by: kennytm <kennytm@gmail.com>
overvenus · Mar 12, 2020 · 3c8f4d7 · 3c8f4d7
1 parent 093ce9f
commit 3c8f4d7
Show file tree

Hide file tree

Showing 10 changed files with 4,867 additions and 4,384 deletions.
diff --git a/lightning/config/config.go b/lightning/config/config.go
@@ -133,6 +133,7 @@ type CSVConfig struct {
 	NotNull         bool   `toml:"not-null" json:"not-null"`
 	Null            string `toml:"null" json:"null"`
 	BackslashEscape bool   `toml:"backslash-escape" json:"backslash-escape"`
+	MaxRegionSize   int64  `toml:"max-region-size" json:"max-region-size"`
 }
 
 type MydumperRuntime struct {
@@ -144,6 +145,8 @@ type MydumperRuntime struct {
 	CharacterSet     string    `toml:"character-set" json:"character-set"`
 	CSV              CSVConfig `toml:"csv" json:"csv"`
 	CaseSensitive    bool      `toml:"case-sensitive" json:"case-sensitive"`
+	StrictFormat     bool      `toml:"strict-format" json:"strict-format"`
+	MaxRegionSize    int64     `toml:"max-region-size" json:"max-region-size"`
 }
 
 type TikvImporter struct {
@@ -244,6 +247,8 @@ func NewConfig() *Config {
 				BackslashEscape: true,
 				TrimLastSep:     false,
 			},
+			StrictFormat:  false,
+			MaxRegionSize: MaxRegionSize,
 		},
 		TikvImporter: TikvImporter{
 			Backend:     BackendImporter,

diff --git a/lightning/config/const.go b/lightning/config/const.go
@@ -21,6 +21,7 @@ const (
 	// mydumper
 	ReadBlockSize int64 = 64 * _K
 	MinRegionSize int64 = 256 * _M
+	MaxRegionSize int64 = 256 * _M
 
 	BufferSizeScale = 5
 

diff --git a/lightning/mydump/csv/split_large_file.csv b/lightning/mydump/csv/split_large_file.csv
@@ -0,0 +1,4 @@
+1,1,2
+2,2,1
+3,2,2
+4,2,2
diff --git a/lightning/mydump/csv_parser.go b/lightning/mydump/csv_parser.go
@@ -150,3 +150,25 @@ func (parser *CSVParser) ReadRow() error {
 		}
 	}
 }
+
+func (parser *CSVParser) ReadUntilTokNewLine() (pos int64, err error) {
+	hasField := false
+	for {
+		tok, _, err := parser.lex()
+		switch errors.Cause(err) {
+		case nil:
+		case io.EOF:
+			if hasField {
+				tok = csvTokNewLine
+				break
+			}
+			fallthrough
+		default:
+			return parser.pos, errors.Trace(err)
+		}
+		hasField = true
+		if tok == csvTokNewLine {
+			return parser.pos, nil
+		}
+	}
+}