ronantakizawa/codereview-bench huggingface dataset:
Repository: pingcap/tidb
File: pkg/executor/revoke.go (Go)

Code before:
	if len(cols) > 0 {
		for _, c := range cols {
			if tbl != nil {
				if table.FindCol(tbl.Cols(), c.Name.L) == nil {
					return infoschema.ErrColumnNotExists.GenWithStackByArgs(c.Name.L, tbl.Meta().Name.L)
				}
			}

			err = removePrivFromColumnPriv(ctx, internalSession.GetSQLExecutor(), priv.Priv, host, user, dbName, e.Level.TableName, c.Name.O)
			if err != nil {
				return err
			}

			//TODO Optimized for batch, one-shot.
		}
	} else {
		// FIXME(cbc): should not reach here?
		err = removePrivFromColumnPriv(ctx, internalSession.GetSQLExecutor(), priv.Priv, host, user, dbName, e.Level.TableName, "")
		if err != nil {
			return err
		}
	}

	if checkTablePrivTbl {
		sql := strings.Join([]string{"SELECT * FROM %n.%n WHERE User=%? AND Host=%? AND DB=%? AND Table_name=%? AND Column_priv LIKE '%%", priv.Priv.String(), "%%';"}, "")
		exists, err := recordExists(internalSession, sql, mysql.SystemDB, mysql.ColumnPrivTable, user, host, dbName, e.Level.TableName)
		if err != nil {
			return err
		}
		if !exists {
			// We should delete the row in mysql.tables_priv if there is no column privileges left.
			p := &ast.PrivElem{Priv: priv.Priv}
			if err = e.revokeTablePriv(ctx, internalSession, p, user, host, false); err != nil {
				return err
			}
		}
	}
	return nil
}


Review comment (suggestion):
I think this PR should only contain log changes. Restore the scene as much as possible is more helpful in locating problems

Code after:
	if len(cols) > 0 {
		for _, c := range cols {
			if tbl != nil {
				if table.FindCol(tbl.Cols(), c.Name.L) == nil {
					return infoschema.ErrColumnNotExists.GenWithStackByArgs(c.Name.L, tbl.Meta().Name.L)
				}
			}

			err = removePrivFromColumnPriv(ctx, internalSession.GetSQLExecutor(), priv.Priv, host, user, dbName, e.Level.TableName, c.Name.O)
			if err != nil {
				return err
			}

			//TODO Optimized for batch, one-shot.
		}
	} else {
		// FIXME(cbc): should not reach here?
		err = removePrivFromColumnPriv(ctx, internalSession.GetSQLExecutor(), priv.Priv, host, user, dbName, e.Level.TableName, "")
		if err != nil {
			return err
		}
	}

	if checkTablePrivTbl {
		sql := strings.Join([]string{"SELECT * FROM %n.%n WHERE User=%? AND Host=%? AND DB=%? AND Table_name=%? AND Column_priv LIKE '%%", priv.Priv.String(), "%%';"}, "")
		exists, err := recordExists(internalSession, sql, mysql.SystemDB, mysql.ColumnPrivTable, user, host, e.Level.DBName, e.Level.TableName)
		if err != nil {
			return err
		}
		if !exists {
			// We should delete the row in mysql.tables_priv if there is no column privileges left.
			p := &ast.PrivElem{Priv: priv.Priv}
			if err = e.revokeTablePriv(ctx, internalSession, p, user, host, false); err != nil {
				return err
			}
		}
	}
	return nil
}
Repository: pingcap/tidb
File: pkg/privilege/privileges/cache.go (Go)

Code before:
func NewHandle(sctx util.SessionPool) *Handle {
	priv := newMySQLPrivilege()
	ret := &Handle{}
	ret.sctx = sctx
	ret.priv.Store(priv)
	return ret
}

func (h *Handle) merge(data *MySQLPrivilege, userList map[string]struct{}) {
	for {
		old := h.Get()
		swapped := h.priv.CompareAndSwap(old, old.merge(data, userList))
		if swapped {
			break
		}
	}
}

// Get the MySQLPrivilege for read.
func (h *Handle) Get() *MySQLPrivilege {
	return h.priv.Load()
}

// UpdateAll loads all the users' privilege info from kv storage.
func (h *Handle) UpdateAll() error {
	logutil.BgLogger().Info("UpdateAll to refresh privilege cache")
	priv := newMySQLPrivilege()
	res, err := h.sctx.Get()
	if err != nil {
		return errors.Trace(err)
	}
	defer h.sctx.Put(res)
	exec := res.(sqlexec.SQLExecutor)

	err = priv.LoadAll(exec)
	if err != nil {
		return errors.Trace(err)
	}
	h.priv.Store(priv)
	return nil
}

// Update loads the privilege info from kv storage for the list of users.
func (h *Handle) Update(userList []string) error {
	if len(userList) > 100 {
		logutil.BgLogger().Warn("update user list is long", zap.Int("len", len(userList)))
	}

	return h.updateUsers(userList)
}


Review comment (security):
I think the log need print out all rebuild privilege items out. this log is no use for us to identify whether there is problem for build cache.

Code after:
	ret := &Handle{}
	ret.sctx = sctx
	ret.priv.Store(priv)
	return ret
}

func (h *Handle) merge(data *MySQLPrivilege, userList map[string]struct{}) {
	for {
		old := h.Get()
		swapped := h.priv.CompareAndSwap(old, old.merge(data, userList))
		if swapped {
			break
		}
	}
}

// Get the MySQLPrivilege for read.
func (h *Handle) Get() *MySQLPrivilege {
	return h.priv.Load()
}

// UpdateAll loads all the users' privilege info from kv storage.
func (h *Handle) UpdateAll() error {
	logutil.BgLogger().Info("UpdateAll to refresh privilege cache")
	priv := newMySQLPrivilege()
	res, err := h.sctx.Get()
	if err != nil {
		return errors.Trace(err)
	}
	defer h.sctx.Put(res)
	exec := res.(sqlexec.SQLExecutor)

	err = priv.LoadAll(exec)
	if err != nil {
		return errors.Trace(err)
	}
	h.priv.Store(priv)
	return nil
}

// Update loads the privilege info from kv storage for the list of users.
func (h *Handle) Update(userList []string) error {
	if len(userList) > 100 {
		logutil.BgLogger().Warn("update user list is long", zap.Int("len", len(userList)))
	}

	return h.updateUsers(userList)
}

func (h *Handle) updateUsers(userList []string) error {
	res, err := h.sctx.Get()
Repository: pingcap/tidb
File: pkg/executor/join/base_join_probe.go (Go)

Code before:
	if forOtherCondition {
		usedColumnMap := make(map[int]struct{})
		for _, colIndex := range used {
			if _, ok := usedColumnMap[colIndex]; !ok {
				srcCol := probeChk.Column(colIndex)
				dstCol := chk.Column(colIndex + collOffset)

				preAllocMemForCol(srcCol, dstCol)

				nullBitmapCapBefore := 0
				offsetCapBefore := 0
				dataCapBefore := 0
				if intest.InTest {
					nullBitmapCapBefore = dstCol.GetNullBitmapCap()
					offsetCapBefore = dstCol.GetOffsetCap()
					dataCapBefore = dstCol.GetDataCap()
				}

				for _, offsetAndLength := range j.offsetAndLengthArray {
					dstCol.AppendCellNTimes(srcCol, offsetAndLength.offset, offsetAndLength.length)
				}
				usedColumnMap[colIndex] = struct{}{}

				if intest.InTest {
					if nullBitmapCapBefore != dstCol.GetNullBitmapCap() {
						panic("Can't reserve enough memory")
					}

					if offsetCapBefore != dstCol.GetOffsetCap() {
						panic("Can't reserve enough memory")
					}

					if dataCapBefore != dstCol.GetDataCap() {
						panic("Can't reserve enough memory")
					}
				}
			}
		}
	} else {
		for index, colIndex := range used {
			srcCol := probeChk.Column(colIndex)
			dstCol := chk.Column(index + collOffset)

			preAllocMemForCol(srcCol, dstCol)

			for _, offsetAndLength := range j.offsetAndLengthArray {
				dstCol.AppendCellNTimes(srcCol, offsetAndLength.offset, offsetAndLength.length)
			}
		}
	}
}

Review comment (suggestion):
```suggestion
						panic("Don't reserve enough memory")
```

Code after:
			if _, ok := usedColumnMap[colIndex]; !ok {
				srcCol := probeChk.Column(colIndex)
				dstCol := chk.Column(colIndex + collOffset)

				preAllocMemForCol(srcCol, dstCol)

				nullBitmapCapBefore := 0
				offsetCapBefore := 0
				dataCapBefore := 0
				if intest.InTest {
					nullBitmapCapBefore = dstCol.GetNullBitmapCap()
					offsetCapBefore = dstCol.GetOffsetCap()
					dataCapBefore = dstCol.GetDataCap()
				}

				for _, offsetAndLength := range j.offsetAndLengthArray {
					dstCol.AppendCellNTimes(srcCol, offsetAndLength.offset, offsetAndLength.length)
				}
				usedColumnMap[colIndex] = struct{}{}

				if intest.InTest {
					if nullBitmapCapBefore != dstCol.GetNullBitmapCap() {
						panic("Don't reserve enough memory")
					}

					if offsetCapBefore != dstCol.GetOffsetCap() {
						panic("Don't reserve enough memory")
					}

					if dataCapBefore != dstCol.GetDataCap() {
						panic("Don't reserve enough memory")
					}
				}
			}
		}
	} else {
		for index, colIndex := range used {
			srcCol := probeChk.Column(colIndex)
			dstCol := chk.Column(index + collOffset)

			preAllocMemForCol(srcCol, dstCol)

			for _, offsetAndLength := range j.offsetAndLengthArray {
				dstCol.AppendCellNTimes(srcCol, offsetAndLength.offset, offsetAndLength.length)
			}
		}
	}
}

func (j *baseJoinProbe) buildResultAfterOtherCondition(chk *chunk.Chunk, joinedChk *chunk.Chunk) (err error) {
	// construct the return chunk based on joinedChk and selected, there are 3 kinds of columns
Repository: pingcap/tidb
File: pkg/session/session.go (Go)

Code before:
	isNull := false
	enable := false
	var err error
	err = kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(_ context.Context, txn kv.Transaction) error {
		t := meta.NewMutator(txn)
		enable, isNull, err = t.GetMetadataLock()
		if err != nil {
			return err
		}
		if isNull {
			// Workaround for version: nightly-2022-11-07 to nightly-2022-11-17.
			enable = true
			logutil.BgLogger().Warn("metadata lock is null")
			err = t.SetMetadataLock(true)
			if err != nil {
				return err
			}
		}
		return nil
	})
	vardef.SetEnableMDL(enable)
	return err
}

// InitTiDBSchemaCacheSize initializes the tidb schema cache size.
func InitTiDBSchemaCacheSize(store kv.Storage) error {
	var (
		isNull bool
		size   uint64
		err    error
	)
	err = kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(_ context.Context, txn kv.Transaction) error {
		t := meta.NewMutator(txn)
		size, isNull, err = t.GetSchemaCacheSize()
		if err != nil {
			return errors.Trace(err)
		}
		if isNull {
			size = vardef.DefTiDBSchemaCacheSize
			return t.SetSchemaCacheSize(size)
		}
		return nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	vardef.SchemaCacheSize.Store(size)
	return nil
}

// BootstrapSession bootstrap session and domain.

Review comment (bug):
maybe move this function in later refactor PR, better make the bug fix as simple as possible

Code after:
	return nil
}

// InitMDLVariableForUpgrade initializes the metadata lock variable.
func InitMDLVariableForUpgrade(store kv.Storage) (bool, error) {
	isNull := false
	enable := false
	var err error
	err = kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(_ context.Context, txn kv.Transaction) error {
		t := meta.NewMutator(txn)
		enable, isNull, err = t.GetMetadataLock()
		if err != nil {
			return err
		}
		return nil
	})
	if isNull || !enable {
		vardef.SetEnableMDL(false)
	} else {
		vardef.SetEnableMDL(true)
	}
	return isNull, err
}

// InitMDLVariable initializes the metadata lock variable.
func InitMDLVariable(store kv.Storage) error {
	isNull := false
	enable := false
	var err error
	err = kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(_ context.Context, txn kv.Transaction) error {
		t := meta.NewMutator(txn)
		enable, isNull, err = t.GetMetadataLock()
		if err != nil {
			return err
		}
		if isNull {
			// Workaround for version: nightly-2022-11-07 to nightly-2022-11-17.
			enable = true
			logutil.BgLogger().Warn("metadata lock is null")
			err = t.SetMetadataLock(true)
			if err != nil {
				return err
			}
		}
		return nil
	})
	vardef.SetEnableMDL(enable)
	return err
}

// BootstrapSession bootstrap session and domain.
Repository: pingcap/tidb
File: pkg/lightning/backend/local/local.go (Go)

Code before:
	for range conn {
		eg.Go(func() error {
			for {
				select {
				case <-egCtx.Done():
					return nil
				case p, ok := <-dataAndRangeCh:
					if !ok {
						return nil
					}

					failpoint.Inject("beforeGenerateJob", nil)
					failpoint.Inject("sendDummyJob", func(_ failpoint.Value) {
						// this is used to trigger worker failure, used together
						// with WriteToTiKVNotEnoughDiskSpace
						jobToWorkerCh <- &regionJob{}
						time.Sleep(5 * time.Second)
					})
					jobs, err := local.generateJobForRange(egCtx, p.Data, p.SortedRanges, regionSplitSize, regionSplitKeys)
					if err != nil {
						if common.IsContextCanceledError(err) {
							return nil
						}
						return err
					}
					// reference the job to prevent the job from being GCed before being put into jobToWorkerCh.
					for _, job := range jobs {
						job.ref(jobWg)
					}
					for _, job := range jobs {
						select {
						case <-egCtx.Done():
							// this job is not put into jobToWorkerCh
							job.done(jobWg)
							// if the context is canceled, it means worker has error.
							return nil
						case jobToWorkerCh <- job:
						}
					}
				}
			}
		})
	}

	eg.Go(func() error {
		err := engine.LoadIngestData(egCtx, dataAndRangeCh)
		if err != nil {
			return errors.Trace(err)
		}
		close(dataAndRangeCh)
		return nil

Review comment (suggestion):
```suggestion
					// we need to increase the ref count before sending jobs to
					// jobToWorkerCh, in case some job finished quickly and decrease
					// the ref count to zero and cause the data being released.
```

Code after:
	for range conn {
		eg.Go(func() error {
			for {
				select {
				case <-egCtx.Done():
					return nil
				case p, ok := <-dataAndRangeCh:
					if !ok {
						return nil
					}

					failpoint.Inject("beforeGenerateJob", nil)
					failpoint.Inject("sendDummyJob", func(_ failpoint.Value) {
						// this is used to trigger worker failure, used together
						// with WriteToTiKVNotEnoughDiskSpace
						jobToWorkerCh <- &regionJob{}
						time.Sleep(5 * time.Second)
					})
					jobs, err := local.generateJobForRange(egCtx, p.Data, p.SortedRanges, regionSplitSize, regionSplitKeys)
					if err != nil {
						if common.IsContextCanceledError(err) {
							return nil
						}
						return err
					}
					// we need to increase the ref count before sending jobs to
					// jobToWorkerCh, in case some job finished quickly and decrease
					// the ref count to zero and cause the data being released.
					for _, job := range jobs {
						job.ref(jobWg)
					}
					for _, job := range jobs {
						select {
						case <-egCtx.Done():
							// this job is not put into jobToWorkerCh
							job.done(jobWg)
							// if the context is canceled, it means worker has error.
							return nil
						case jobToWorkerCh <- job:
						}
					}
				}
			}
		})
	}

	eg.Go(func() error {
		err := engine.LoadIngestData(egCtx, dataAndRangeCh)
		if err != nil {
			return errors.Trace(err)
		}
Repository: pingcap/tidb
File: pkg/lightning/mydump/region.go (Go)

Code before:

	//nolint: errcheck
	defer parser.Close()

	if err = parser.ReadColumns(); err != nil {
		return nil, 0, err
	}
	startOffset, _ := parser.Pos()
	return parser.Columns(), startOffset, nil
}

// SplitLargeCSV splits a large csv file into multiple regions, the size of
// each regions is specified by `config.MaxRegionSize`.
// Note: We split the file coarsely, thus the format of csv file is needed to be
// strict.
// e.g.
// - CSV file with header is invalid
// - a complete tuple split into multiple lines is invalid
func SplitLargeCSV(
	ctx context.Context,
	cfg *DataDivideConfig,
	dataFile FileInfo,
	parallel bool,
) (regions []*TableRegion, dataFileSizes []float64, err error) {
	maxRegionSize := cfg.MaxChunkSize
	regionCnt := dataFile.FileMeta.FileSize/maxRegionSize + 1
	dataFileSizes = make([]float64, 0, regionCnt)

	headerColumns, dataStart, err := getHeaderColumn(ctx, cfg, dataFile.FileMeta.Path)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	splitEndOffsets := make([]int64, 0, regionCnt)
	endOffset := dataStart + maxRegionSize
	for endOffset < dataFile.FileMeta.FileSize {
		splitEndOffsets = append(splitEndOffsets, endOffset)
		endOffset += maxRegionSize
	}

	concurrency := 1
	if parallel {
		concurrency = max(cfg.Concurrency, 1)
	}

	eg, egCtx := errgroup.WithContext(ctx)
	eg.SetLimit(concurrency)
	for i, endOffset := range splitEndOffsets {
		eg.Go(func() error {
			select {
			case <-egCtx.Done():

Review comment (suggestion):
maybe below one. if `FileSize=2*maxRegionSize`, above code cals 3, not 2
```suggestion
	regionCnt := (dataFile.FileMeta.FileSize + maxRegionSize - 1)/maxRegionSize
```

Code after:
	parser, err := openCSVParser(ctx, cfg, path, 0)
	if err != nil {
		return nil, 0, err
	}

	//nolint: errcheck
	defer parser.Close()

	if err = parser.ReadColumns(); err != nil {
		return nil, 0, err
	}
	startOffset, _ := parser.Pos()
	return parser.Columns(), startOffset, nil
}

// SplitLargeCSV splits a large csv file into multiple regions, the size of
// each regions is specified by `config.MaxRegionSize`.
// Note: We split the file coarsely, thus the format of csv file is needed to be
// strict.
// e.g.
// - CSV file with header is invalid
// - a complete tuple split into multiple lines is invalid
func SplitLargeCSV(
	ctx context.Context,
	cfg *DataDivideConfig,
	dataFile FileInfo,
	parallel bool,
) (regions []*TableRegion, dataFileSizes []float64, err error) {
	maxRegionSize := cfg.MaxChunkSize

	headerColumns, dataStart, err := getHeaderColumn(ctx, cfg, dataFile.FileMeta.Path)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	regionCnt := (dataFile.FileMeta.FileSize - dataStart + maxRegionSize - 1) / maxRegionSize
	dataFileSizes = make([]float64, 0, regionCnt)

	splitPoints := mathutil.Divide2Batches(dataFile.FileMeta.FileSize-dataStart, regionCnt)
	splitPoints = splitPoints[:len(splitPoints)-1]
	for i := range splitPoints {
		if i == 0 {
			splitPoints[i] += dataStart
		} else {
			splitPoints[i] += splitPoints[i-1]
		}
	}

	concurrency := 1
	if parallel {
		concurrency = max(cfg.Concurrency, 1)
Repository: pingcap/tidb
File: pkg/lightning/mydump/region.go (Go)

Code before:
// Note: We split the file coarsely, thus the format of csv file is needed to be
// strict.
// e.g.
// - CSV file with header is invalid
// - a complete tuple split into multiple lines is invalid
func SplitLargeCSV(
	ctx context.Context,
	cfg *DataDivideConfig,
	dataFile FileInfo,
	parallel bool,
) (regions []*TableRegion, dataFileSizes []float64, err error) {
	maxRegionSize := cfg.MaxChunkSize
	regionCnt := dataFile.FileMeta.FileSize/maxRegionSize + 1
	dataFileSizes = make([]float64, 0, regionCnt)

	headerColumns, dataStart, err := getHeaderColumn(ctx, cfg, dataFile.FileMeta.Path)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	splitEndOffsets := make([]int64, 0, regionCnt)
	endOffset := dataStart + maxRegionSize
	for endOffset < dataFile.FileMeta.FileSize {
		splitEndOffsets = append(splitEndOffsets, endOffset)
		endOffset += maxRegionSize
	}

	concurrency := 1
	if parallel {
		concurrency = max(cfg.Concurrency, 1)
	}

	eg, egCtx := errgroup.WithContext(ctx)
	eg.SetLimit(concurrency)
	for i, endOffset := range splitEndOffsets {
		eg.Go(func() error {
			select {
			case <-egCtx.Done():
				return egCtx.Err()
			default:
			}

			parser, err := openCSVParser(egCtx, cfg, dataFile.FileMeta.Path, endOffset)
			if err != nil {
				return errors.Trace(err)
			}
			//nolint: errcheck
			defer parser.Close()
			_, pos, err := parser.ReadUntilTerminator()
			if err != nil {
				if !errors.ErrorEqual(err, io.EOF) {

Review comment (suggestion):
```suggestion
	for end := dataStart + maxRegionSize; end < dataFile.FileMeta.FileSize; end += maxRegionSize {
		splitEndOffsets = append(splitEndOffsets, end)
	}
```

Code after:
}

// SplitLargeCSV splits a large csv file into multiple regions, the size of
// each regions is specified by `config.MaxRegionSize`.
// Note: We split the file coarsely, thus the format of csv file is needed to be
// strict.
// e.g.
// - CSV file with header is invalid
// - a complete tuple split into multiple lines is invalid
func SplitLargeCSV(
	ctx context.Context,
	cfg *DataDivideConfig,
	dataFile FileInfo,
	parallel bool,
) (regions []*TableRegion, dataFileSizes []float64, err error) {
	maxRegionSize := cfg.MaxChunkSize

	headerColumns, dataStart, err := getHeaderColumn(ctx, cfg, dataFile.FileMeta.Path)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	regionCnt := (dataFile.FileMeta.FileSize - dataStart + maxRegionSize - 1) / maxRegionSize
	dataFileSizes = make([]float64, 0, regionCnt)

	splitPoints := mathutil.Divide2Batches(dataFile.FileMeta.FileSize-dataStart, regionCnt)
	splitPoints = splitPoints[:len(splitPoints)-1]
	for i := range splitPoints {
		if i == 0 {
			splitPoints[i] += dataStart
		} else {
			splitPoints[i] += splitPoints[i-1]
		}
	}

	concurrency := 1
	if parallel {
		concurrency = max(cfg.Concurrency, 1)
	}

	eg, egCtx := errgroup.WithContext(ctx)
	eg.SetLimit(concurrency)
	for i, splitPoint := range splitPoints {
		eg.Go(func() error {
			select {
			case <-egCtx.Done():
				return egCtx.Err()
			default:
			}

			parser, err := openCSVParser(egCtx, cfg, dataFile.FileMeta.Path, splitPoint)
Repository: pingcap/tidb
File: pkg/lightning/mydump/region.go (Go)

Code before:

func openCSVParser(
	ctx context.Context,
	cfg *DataDivideConfig,
	path string,
	offset int64,
) (*CSVParser, error) {
	r, err := cfg.Store.Open(ctx, path, nil)
	if err != nil {
		return nil, err
	}

	// 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, err
	}

	parser, err := NewCSVParser(ctx, &cfg.CSV, r, cfg.ReadBlockSize, cfg.IOWorkers, true, charsetConvertor)
	if err != nil {
		return nil, err
	}

	if err = parser.SetPos(offset, 0); err != nil {
		_ = parser.Close()
		return nil, err
	}

	return parser, nil
}

func getHeaderColumn(
	ctx context.Context,
	cfg *DataDivideConfig,
	path string,
) ([]string, int64, error) {
	if !cfg.CSV.Header || !cfg.CSV.HeaderSchemaMatch {
		return nil, 0, nil
	}

	parser, err := openCSVParser(ctx, cfg, path, 0)
	if err != nil {
		return nil, 0, err
	}

	//nolint: errcheck
	defer parser.Close()

	if err = parser.ReadColumns(); err != nil {
		return nil, 0, err

Review comment (suggestion):
```suggestion
		_ = r.Close()
		return nil, err
```

Code after:
	}
	return []*TableRegion{region}, []float64{float64(dataFile.FileMeta.FileSize)}, nil
}

func openCSVParser(
	ctx context.Context,
	cfg *DataDivideConfig,
	path string,
	offset int64,
) (*CSVParser, error) {
	r, err := cfg.Store.Open(ctx, path, nil)
	if err != nil {
		return nil, err
	}

	// 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, err
	}

	parser, err := NewCSVParser(ctx, &cfg.CSV, r, cfg.ReadBlockSize, cfg.IOWorkers, true, charsetConvertor)
	if err != nil {
		_ = r.Close()
		return nil, err
	}

	if err = parser.SetPos(offset, 0); err != nil {
		_ = parser.Close()
		return nil, err
	}

	return parser, nil
}

func getHeaderColumn(
	ctx context.Context,
	cfg *DataDivideConfig,
	path string,
) ([]string, int64, error) {
	if !cfg.CSV.Header || !cfg.CSV.HeaderSchemaMatch {
		return nil, 0, nil
	}

	parser, err := openCSVParser(ctx, cfg, path, 0)
	if err != nil {
		return nil, 0, err
	}

	//nolint: errcheck