config: must set line terminator when use strict-format (#53444)

br/pkg/lightning/config/BUILD.bazel

@@ -41,7 +41,7 @@ go_test(
         "configlist_test.go",
     ],
     flaky = True,
-    shard_count = 45,
+    shard_count = 46,
     deps = [
         ":config",
         "//br/pkg/lightning/common",

br/pkg/lightning/config/config.go

@@ -1172,7 +1172,9 @@ func (cfg *Config) Adjust(ctx context.Context) error {
 	if err := cfg.CheckAndAdjustTiDBPort(ctx, mustHaveInternalConnections); err != nil {
 		return err
 	}
-	cfg.AdjustMydumper()
+	if err := cfg.AdjustMydumper(); err != nil {
+		return err
+	}
 	cfg.AdjustCheckPoint()
 	return cfg.CheckAndAdjustFilePath()
 }
@@ -1435,7 +1437,11 @@ func (cfg *Config) AdjustCheckPoint() {
 }
 
 // AdjustMydumper adjusts the mydumper config.
-func (cfg *Config) AdjustMydumper() {
+func (cfg *Config) AdjustMydumper() error {
+	if cfg.Mydumper.StrictFormat && len(cfg.Mydumper.CSV.Terminator) == 0 {
+		return common.ErrInvalidConfig.GenWithStack("" +
+			`mydumper.strict-format can not be used with empty mydumper.csv.terminator. Please set mydumper.csv.terminator to a non-empty value like "\r\n"`)
+	}
 	if cfg.Mydumper.BatchImportRatio < 0.0 || cfg.Mydumper.BatchImportRatio >= 1.0 {
 		cfg.Mydumper.BatchImportRatio = DefaultBatchImportRatio
 	}
@@ -1456,6 +1462,7 @@ func (cfg *Config) AdjustMydumper() {
 			ig.Columns = cols
 		}
 	}
+	return nil
 }
 
 // CheckAndAdjustSecurity checks and adjusts the security config.

br/pkg/lightning/config/config_test.go

@@ -83,6 +83,30 @@ func TestAdjustPdAddrAndPort(t *testing.T) {
 	require.Equal(t, "123.45.67.89:1234", cfg.TiDB.PdAddr)
 }
 
+func TestStrictFormat(t *testing.T) {
+	ts, host, port := startMockServer(t, http.StatusOK,
+		`{"port":4444,"advertise-address":"","path":"123.45.67.89:1234,56.78.90.12:3456"}`,
+	)
+	defer ts.Close()
+
+	cfg := config.NewConfig()
+	cfg.TiDB.Host = host
+	cfg.TiDB.StatusPort = port
+	cfg.Mydumper.SourceDir = "."
+	cfg.TikvImporter.Backend = config.BackendLocal
+	cfg.TikvImporter.SortedKVDir = "."
+	cfg.TiDB.DistSQLScanConcurrency = 1
+	cfg.Mydumper.StrictFormat = true
+
+	err := cfg.Adjust(context.Background())
+	require.ErrorContains(t, err, "mydumper.strict-format can not be used with empty mydumper.csv.terminator")
+	t.Log(err.Error())
+
+	cfg.Mydumper.CSV.Terminator = "\r\n"
+	err = cfg.Adjust(context.Background())
+	require.NoError(t, err)
+}
+
 func TestPausePDSchedulerScope(t *testing.T) {
 	ts, host, port := startMockServer(t, http.StatusOK,
 		`{"port":4444,"advertise-address":"","path":"123.45.67.89:1234,56.78.90.12:3456"}`,

br/pkg/lightning/mydump/csv_parser.go

@@ -98,7 +98,8 @@ type field struct {
 	quoted  bool
 }
 
-// NewCSVParser creates a CSV parser.
+// NewCSVParser creates a CSV parser. The ownership of the reader is transferred
+// to the parser.
 func NewCSVParser(
 	ctx context.Context,
 	cfg *config.CSVConfig,

br/pkg/lightning/mydump/region.go

@@ -442,13 +442,15 @@ func SplitLargeCSV(
 		// Create a utf8mb4 convertor to encode and decode data with the charset of CSV files.
 		charsetConvertor, err := NewCharsetConvertor(cfg.DataCharacterSet, cfg.DataInvalidCharReplace)
 		if err != nil {
+			_ = r.Close()
 			return nil, nil, err
 		}
 		parser, err := NewCSVParser(ctx, &cfg.CSV, r, cfg.ReadBlockSize, cfg.IOWorkers, true, charsetConvertor)
 		if err != nil {
 			return nil, nil, err
 		}
 		if err = parser.ReadColumns(); err != nil {
+			_ = parser.Close()
 			return nil, nil, err
 		}
 		if cfg.CSV.HeaderSchemaMatch {
@@ -459,6 +461,7 @@ func SplitLargeCSV(
 		if endOffset > dataFile.FileMeta.FileSize {
 			endOffset = dataFile.FileMeta.FileSize
 		}
+		_ = parser.Close()
 	}
 	divisor := int64(cfg.ColumnCnt)
 	for {
@@ -472,18 +475,21 @@ func SplitLargeCSV(
 			// Create a utf8mb4 convertor to encode and decode data with the charset of CSV files.
 			charsetConvertor, err := NewCharsetConvertor(cfg.DataCharacterSet, cfg.DataInvalidCharReplace)
 			if err != nil {
+				_ = r.Close()
 				return nil, nil, err
 			}
 			parser, err := NewCSVParser(ctx, &cfg.CSV, r, cfg.ReadBlockSize, cfg.IOWorkers, false, charsetConvertor)
 			if err != nil {
 				return nil, nil, err
 			}
 			if err = parser.SetPos(endOffset, 0); err != nil {
+				_ = r.Close()
 				return nil, nil, err
 			}
 			_, pos, err := parser.ReadUntilTerminator()
 			if err != nil {
 				if !errors.ErrorEqual(err, io.EOF) {
+					_ = r.Close()
 					return nil, nil, err
 				}
 				log.FromContext(ctx).Warn("file contains no terminator at end",
@@ -492,7 +498,7 @@ func SplitLargeCSV(
 				pos = dataFile.FileMeta.FileSize
 			}
 			endOffset = pos
-			parser.Close()
+			_ = parser.Close()
 		}
 		regions = append(regions,
 			&TableRegion{

br/pkg/lightning/mydump/region_test.go

@@ -480,3 +480,106 @@ func TestSplitLargeFileOnlyOneChunk(t *testing.T) {
 		require.Equal(t, columns, regions[i].Chunk.Columns)
 	}
 }
+
+func TestSplitLargeFileSeekInsideCRLF(t *testing.T) {
+	ctx := context.Background()
+	meta := &MDTableMeta{
+		DB:   "csv",
+		Name: "large_csv_seek_inside_crlf",
+	}
+
+	dir := t.TempDir()
+
+	fileName := "test.csv"
+	filePath := filepath.Join(dir, fileName)
+
+	content := []byte("1\r\n2\r\n3\r\n4\r\n")
+	err := os.WriteFile(filePath, content, 0o644)
+	require.NoError(t, err)
+
+	dataFileInfo, err := os.Stat(filePath)
+	require.NoError(t, err)
+	fileSize := dataFileInfo.Size()
+	fileInfo := FileInfo{FileMeta: SourceFileMeta{Path: fileName, Type: SourceTypeCSV, FileSize: fileSize}}
+	ioWorker := worker.NewPool(context.Background(), 4, "io")
+
+	store, err := storage.NewLocalStorage(dir)
+	require.NoError(t, err)
+
+	// if we don't set terminator, it will get the wrong result
+
+	cfg := &config.Config{
+		Mydumper: config.MydumperRuntime{
+			ReadBlockSize: config.ReadBlockSize,
+			CSV: config.CSVConfig{
+				Separator: ",",
+			},
+			StrictFormat:  true,
+			Filter:        []string{"*.*"},
+			MaxRegionSize: 2,
+		},
+	}
+
+	// in fact this is the wrong result, just to show the bug. pos mismatch with
+	// offsets. and we might read more rows than expected because we use == rather
+	// than >= to stop reading.
+	offsets := [][]int64{{0, 3}, {3, 6}, {6, 9}, {9, 12}}
+	pos := []int64{2, 5, 8, 11}
+
+	divideConfig := NewDataDivideConfig(cfg, 2, ioWorker, store, meta)
+
+	regions, _, err := SplitLargeCSV(
+		context.Background(),
+		divideConfig,
+		fileInfo,
+	)
+	require.NoError(t, err)
+	require.Len(t, regions, len(offsets))
+	for i := range offsets {
+		require.Equal(t, offsets[i][0], regions[i].Chunk.Offset)
+		require.Equal(t, offsets[i][1], regions[i].Chunk.EndOffset)
+	}
+
+	file, err := os.Open(filePath)
+	require.NoError(t, err)
+	parser, err := NewCSVParser(ctx, &cfg.Mydumper.CSV, file, 128, ioWorker, false, nil)
+	require.NoError(t, err)
+
+	for parser.ReadRow() == nil {
+		p, _ := parser.Pos()
+		require.Equal(t, pos[0], p)
+		pos = pos[1:]
+	}
+	require.NoError(t, parser.Close())
+
+	// set terminator to "\r\n"
+
+	cfg.Mydumper.CSV.Terminator = "\r\n"
+	// pos is contained in expectedOffsets
+	expectedOffsets := [][]int64{{0, 6}, {6, 12}}
+	pos = []int64{3, 6, 9, 12}
+
+	regions, _, err = SplitLargeCSV(
+		context.Background(),
+		divideConfig,
+		fileInfo,
+	)
+	require.NoError(t, err)
+	require.Len(t, regions, len(expectedOffsets))
+	for i := range expectedOffsets {
+		require.Equal(t, expectedOffsets[i][0], regions[i].Chunk.Offset)
+		require.Equal(t, expectedOffsets[i][1], regions[i].Chunk.EndOffset)
+	}
+
+	file, err = os.Open(filePath)
+	require.NoError(t, err)
+	parser, err = NewCSVParser(ctx, &cfg.Mydumper.CSV, file, 128, ioWorker, false, nil)
+	require.NoError(t, err)
+
+	for parser.ReadRow() == nil {
+		p, _ := parser.Pos()
+		require.Equal(t, pos[0], p)
+		pos = pos[1:]
+	}
+	require.NoError(t, parser.Close())
+}

br/tests/lightning_column_permutation/config.toml

@@ -1,3 +1,6 @@
 [mydumper]
 strict-format = true
 max-region-size = 200
+
+[mydumper.csv]
+terminator = "\n"

tools/check/check-bazel-prepare.sh

@@ -25,7 +25,9 @@ make bazel_prepare
 after_checksum=`find . -type f \( -name *.bazel -o -name *.bzl \) -exec md5sum {} \;| sort -k 2`
 if [ "$before_checksum" != "$after_checksum" ]
 then
-  echo "Please run \`make bazel_prepare\` to update \`.bazel\` files"
-  diff <(echo "$before_checksum") <(echo "$after_checksum")
+  echo "Please run \`make bazel_prepare\` to update \`.bazel\` files, or just apply the following git diff (run \`git apply -\` and paste following contents with ctrl-D as ending):"
+  git diff
+  echo -e "\n\nChecksum diff:"
+  diff <(echo "$before_checksum") <(echo "$after_checksum") || true
   exit 1
 fi