lightning: cherry-pick some PRs

pkg/lightning/mydump/region.go

@@ -30,7 +30,11 @@ import (
  "github.com/pingcap/br/pkg/utils"
 )
 
-const tableRegionSizeWarningThreshold int64 = 1024 * 1024 * 1024
+const (
+ tableRegionSizeWarningThreshold int64 = 1024 * 1024 * 1024
+ // the increment ratio of large CSV file size threshold by `region-split-size`
+ largeCSVLowerThresholdRation = 10
+)
 
 type TableRegion struct {
  EngineID int32
@@ -268,7 +272,10 @@ func makeSourceFileRegion(
  }
  // If a csv file is overlarge, we need to split it into multiple regions.
  // Note: We can only split a csv file whose format is strict.
- if isCsvFile && dataFileSize > int64(cfg.Mydumper.MaxRegionSize) && cfg.Mydumper.StrictFormat {
+ // We increase the check threshold by 1/10 of the `max-region-size` because the source file size dumped by tools
+ // like dumpling might be slight exceed the threshold when it is equal `max-region-size`, so we can
+ // avoid split a lot of small chunks.
+ if isCsvFile && cfg.Mydumper.StrictFormat && dataFileSize > int64(cfg.Mydumper.MaxRegionSize+cfg.Mydumper.MaxRegionSize/largeCSVLowerThresholdRation) {
  _, regions, subFileSizes, err := SplitLargeFile(ctx, meta, cfg, fi, divisor, 0, ioWorkers, store)
  return regions, subFileSizes, err
  }
@@ -359,6 +366,9 @@ func SplitLargeFile(
  columns = parser.Columns()
  startOffset, _ = parser.Pos()
  endOffset = startOffset + maxRegionSize
+ if endOffset > dataFile.FileMeta.FileSize {
+ endOffset = dataFile.FileMeta.FileSize
+ }
  }
  for {
  curRowsCnt := (endOffset - startOffset) / divisor

pkg/lightning/mydump/region_test.go

@@ -332,3 +332,59 @@ func (s *testMydumpRegionSuite) TestSplitLargeFileWithCustomTerminator(c *C) {
  c.Assert(regions[i].Chunk.EndOffset, Equals, offsets[i][1])
  }
 }
+
+func (s *testMydumpRegionSuite) TestSplitLargeFileOnlyOneChunk(c *C) {
+ meta := &MDTableMeta{
+ DB: "csv",
+ Name: "large_csv_file",
+ }
+ cfg := &config.Config{
+ Mydumper: config.MydumperRuntime{
+ ReadBlockSize: config.ReadBlockSize,
+ CSV: config.CSVConfig{
+ Separator: ",",
+ Delimiter: "",
+ Header: true,
+ TrimLastSep: false,
+ NotNull: false,
+ Null: "NULL",
+ BackslashEscape: true,
+ },
+ StrictFormat: true,
+ Filter: []string{"*.*"},
+ MaxRegionSize: 15,
+ },
+ }
+
+ dir := c.MkDir()
+
+ fileName := "test.csv"
+ filePath := filepath.Join(dir, fileName)
+
+ content := []byte("field1,field2\r\n123,456\r\n")
+ err := os.WriteFile(filePath, content, 0o644)
+ c.Assert(err, IsNil)
+
+ dataFileInfo, err := os.Stat(filePath)
+ c.Assert(err, IsNil)
+ fileSize := dataFileInfo.Size()
+ fileInfo := FileInfo{FileMeta: SourceFileMeta{Path: fileName, Type: SourceTypeCSV, FileSize: fileSize}}
+ colCnt := int64(2)
+ columns := []string{"field1", "field2"}
+ prevRowIdxMax := int64(0)
+ ioWorker := worker.NewPool(context.Background(), 4, "io")
+
+ store, err := storage.NewLocalStorage(dir)
+ c.Assert(err, IsNil)
+
+ offsets := [][]int64{{14, 24}}
+
+ _, regions, _, err := SplitLargeFile(context.Background(), meta, cfg, fileInfo, colCnt, prevRowIdxMax, ioWorker, store)
+ c.Assert(err, IsNil)
+ c.Assert(regions, HasLen, len(offsets))
+ for i := range offsets {
+ c.Assert(regions[i].Chunk.Offset, Equals, offsets[i][0])
+ c.Assert(regions[i].Chunk.EndOffset, Equals, offsets[i][1])
+ c.Assert(regions[i].Chunk.Columns, DeepEquals, columns)
+ }
+}

pkg/storage/gcs.go

@@ -174,11 +174,6 @@ func (s *gcsStorage) WalkDir(ctx context.Context, opt *WalkOption, fn func(strin
  opt = &WalkOption{}
  }
 
- maxKeys := int64(1000)
- if opt.ListCount > 0 {
- maxKeys = opt.ListCount
- }
-
  prefix := path.Join(s.gcs.Prefix, opt.SubDir)
  if len(prefix) > 0 && !strings.HasSuffix(prefix, "/") {
  prefix += "/"
@@ -188,7 +183,7 @@ func (s *gcsStorage) WalkDir(ctx context.Context, opt *WalkOption, fn func(strin
  // only need each object's name and size
  query.SetAttrSelection([]string{"Name", "Size"})
  iter := s.bucket.Objects(ctx, query)
- for i := int64(0); i != maxKeys; i++ {
+ for {
  attrs, err := iter.Next()
  if err == iterator.Done {
  break

pkg/storage/gcs_test.go

@@ -4,6 +4,7 @@ package storage
 
 import (
  "context"
+ "fmt"
  "io"
  "os"
 
@@ -76,6 +77,31 @@ func (r *testStorageSuite) TestGCS(c *C) {
  c.Assert(list, Equals, "keykey1key2")
  c.Assert(totalSize, Equals, int64(42))
 
+ // test 1003 files
+ totalSize = 0
+ for i := 0; i < 1000; i += 1 {
+ err = stg.WriteFile(ctx, fmt.Sprintf("f%d", i), []byte("data"))
+ c.Assert(err, IsNil)
+ }
+ filesSet := make(map[string]struct{}, 1003)
+ err = stg.WalkDir(ctx, nil, func(name string, size int64) error {
+ filesSet[name] = struct{}{}
+ totalSize += size
+ return nil
+ })
+ c.Assert(err, IsNil)
+ c.Assert(totalSize, Equals, int64(42+4000))
+ _, ok := filesSet["key"]
+ c.Assert(ok, IsTrue)
+ _, ok = filesSet["key1"]
+ c.Assert(ok, IsTrue)
+ _, ok = filesSet["key2"]
+ c.Assert(ok, IsTrue)
+ for i := 0; i < 1000; i += 1 {
+ _, ok = filesSet[fmt.Sprintf("f%d", i)]
+ c.Assert(ok, IsTrue)
+ }
+
  efr, err := stg.Open(ctx, "key2")
  c.Assert(err, IsNil)
 

pkg/storage/s3.go

@@ -29,6 +29,7 @@ import (
  "go.uber.org/zap"
 
  berrors "github.com/pingcap/br/pkg/errors"
+ "github.com/pingcap/br/pkg/logutil"
 )
 
 const (
@@ -654,6 +655,19 @@ func (r *s3ObjectReader) Seek(offset int64, whence int) (int64, error) {
 
  if realOffset == r.pos {
  return realOffset, nil
+ } else if realOffset >= r.rangeInfo.Size {
+ // See: https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.35
+ // because s3's GetObject interface doesn't allow get a range that matches zero length data,
+ // so if the position is out of range, we need to always return io.EOF after the seek operation.
+
+ // close current read and open a new one which target offset
+ if err := r.reader.Close(); err != nil {
+ log.L().Warn("close s3 reader failed, will ignore this error", logutil.ShortError(err))
+ }
+
+ r.reader = io.NopCloser(bytes.NewReader(nil))
+ r.pos = r.rangeInfo.Size
+ return r.pos, nil
  }
 
  // if seek ahead no more than 64k, we discard these data

pkg/storage/s3_test.go

@@ -694,6 +694,15 @@ func (s *s3Suite) TestOpenSeek(c *C) {
  c.Assert(err, IsNil)
  c.Assert(n, Equals, 100)
  c.Assert(slice, DeepEquals, someRandomBytes[990100:990200])
+
+ // test seek to the file end or bigger positions
+ for _, p := range []int64{1000000, 1000001, 2000000} {
+ offset, err = reader.Seek(p, io.SeekStart)
+ c.Assert(offset, Equals, int64(1000000))
+ c.Assert(err, IsNil)
+ _, err := reader.Read(slice)
+ c.Assert(err, Equals, io.EOF)
+ }
 }
 
 type limitedBytesReader struct {