Repository: pingcap/tidb (go)

Issue:
Issues in the Lightning error message formatting
### 🐞 Issue Description

**Problem:**  
There are two issues in the Lightning error message formatting:

1. **Grammatical mistake**:  
   The message says `"column %s are missing the default value"` — but since **"column"** is singular, the verb should be **"is"**.

2. **Missing space between strings**:  
   The message is constructed using `fmt.Sprintf(...)+` without adding a space at the end or beginning of the joined strings. This leads to incorrect concatenation, like:
   ```txt
   ...has %d columns,and data file...
   ```
   instead of:
   ```txt
   ...has %d columns, and data file...
   ```


Pull request description:
Lightning: Fix grammar and spacing in Lightning error message
<!--

Thank you for contributing to TiDB!

PR Title Format:
1. pkg [, pkg2, pkg3]: what's changed
2. *: what's changed

-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.

-->

Issue Number: close #60704 

Problem Summary:

### What changed and how does it work?

This PR improves the clarity and correctness of an error message in the Lightning component by:

- Fixing grammar:

  - "column %s are missing the default value" → "column %s is missing the default value"

- Adding necessary spaces during string concatenation to avoid words being glued together (e.g., "columns," + "and" now becomes "columns, " + "and")

These changes help ensure the error messages are clear, professional, and user-friendly.

### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test
  > - [ ] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->

Side effects

- [ ] Performance regression: Consumes more CPU
- [ ] Performance regression: Consumes more Memory
- [ ] Breaking backward compatibility

Documentation

- [ ] Affects user behaviors
- [ ] Contains syntax changes
- [ ] Contains variable changes
- [ ] Contains experimental features
- [ ] Changes MySQL compatibility

### Release note

<!-- compatibility change, improvement, bugfix, and new feature need a release note -->

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

```release-note
None
```


Patch:
diff --git a/br/metrics/grafana/br.json b/br/metrics/grafana/br.json
index 3b48947f0e..1c09319755 100644
--- a/br/metrics/grafana/br.json
+++ b/br/metrics/grafana/br.json
@@ -2917,7 +2917,7 @@
         "options": [
 
         ],
-        "query": "label_values(tikv_engine_size_bytes{k8s_cluster=\"$k8s_cluster\", tidb_cluster)",
+        "query": "label_values(tikv_engine_size_bytes{k8s_cluster=\"$k8s_cluster\"}, tidb_cluster)",
         "refresh": 2,
         "regex": "",
         "sort": 1,
diff --git a/lightning/pkg/importer/precheck_impl.go b/lightning/pkg/importer/precheck_impl.go
index 1acee2cccc..0cf3ee6534 100644
--- a/lightning/pkg/importer/precheck_impl.go
+++ b/lightning/pkg/importer/precheck_impl.go
@@ -1046,8 +1046,8 @@ func (ci *schemaCheckItem) SchemaIsValid(ctx context.Context, tableInfo *mydump.
 		// so the last several columns either can be ignored or has a default value.
 		for i := len(row); i < colCountFromTiDB; i++ {
 			if _, ok := defaultCols[core.Columns[i].Name.L]; !ok {
-				msgs = append(msgs, fmt.Sprintf("TiDB schema `%s`.`%s` has %d columns,"+
-					"and data file has %d columns, but column %s are missing the default value,"+
+				msgs = append(msgs, fmt.Sprintf("TiDB schema `%s`.`%s` has %d columns, "+
+					"and data file has %d columns, but column %s is missing the default value, "+
 					"please give column a default value to skip this check",
 					tableInfo.DB, tableInfo.Name, colCountFromTiDB, len(row), core.Columns[i].Name.L))
 			}
@@ -1067,7 +1067,7 @@ func (ci *schemaCheckItem) SchemaIsValid(ctx context.Context, tableInfo *mydump.
 			// tidb's column is ignored
 			// we need ensure this column has the default value.
 			if _, hasDefault := defaultCols[col.Name.L]; !hasDefault {
-				msgs = append(msgs, fmt.Sprintf("TiDB schema `%s`.`%s`'s column %s cannot be ignored,"+
+				msgs = append(msgs, fmt.Sprintf("TiDB schema `%s`.`%s`'s column %s cannot be ignored, "+
 					"because it doesn't have a default value, please set tables.ignoreColumns properly",
 					tableInfo.DB, tableInfo.Name, col.Name.L))
 			}
diff --git a/pkg/executor/import_into.go b/pkg/executor/import_into.go
index 170d71f772..953aaf65eb 100644
--- a/pkg/executor/import_into.go
+++ b/pkg/executor/import_into.go
@@ -276,8 +276,8 @@ func (e *ImportIntoExec) importFromSelect(ctx context.Context) error {
 			logutil.Logger(ctx).Error("close importer failed", zap.Error(err))
 		}
 	}()
-	selectedRowCh := make(chan importer.QueryRow)
-	ti.SetSelectedRowCh(selectedRowCh)
+	selectedChunkCh := make(chan importer.QueryChunk, 1)
+	ti.SetSelectedChunkCh(selectedChunkCh)
 
 	var importResult *importer.JobImportResult
 	eg, egCtx := errgroup.WithContext(ctx)
@@ -287,13 +287,14 @@ func (e *ImportIntoExec) importFromSelect(ctx context.Context) error {
 		return err
 	})
 	eg.Go(func() error {
-		defer close(selectedRowCh)
+		defer close(selectedChunkCh)
 		fields := exec.RetTypes(e.selectExec)
 		var idAllocator int64
+		chkSize := e.selectExec.InitCap()
+		maxChkSize := e.selectExec.MaxChunkSize()
 		for {
 			// rows will be consumed concurrently, we cannot use chunk pool in session ctx.
-			chk := exec.NewFirstChunk(e.selectExec)
-			iter := chunk.NewIterator4Chunk(chk)
+			chk := chunk.New(e.selectExec.RetFieldTypes(), chkSize, maxChkSize)
 			err := exec.Next(egCtx, e.selectExec, chk)
 			if err != nil {
 				return err
@@ -301,16 +302,19 @@ func (e *ImportIntoExec) importFromSelect(ctx context.Context) error {
 			if chk.NumRows() == 0 {
 				break
 			}
-			for innerChunkRow := iter.Begin(); innerChunkRow != iter.End(); innerChunkRow = iter.Next() {
-				idAllocator++
-				select {
-				case selectedRowCh <- importer.QueryRow{
-					ID:   idAllocator,
-					Data: innerChunkRow.GetDatumRow(fields),
-				}:
-				case <-egCtx.Done():
-					return egCtx.Err()
-				}
+			select {
+			case selectedChunkCh <- importer.QueryChunk{
+				Fields:      fields,
+				Chk:         chk,
+				RowIDOffset: idAllocator,
+			}:
+				idAllocator += int64(chk.NumRows())
+			case <-egCtx.Done():
+				return egCtx.Err()
+			}
+			if chkSize < maxChkSize {
+				chkSize = chkSize * 2
+				chkSize = min(chkSize, maxChkSize)
 			}
 		}
 		return nil
diff --git a/pkg/executor/importer/BUILD.bazel b/pkg/executor/importer/BUILD.bazel
index 13760bcbb9..1b1ea262c7 100644
--- a/pkg/executor/importer/BUILD.bazel
+++ b/pkg/executor/importer/BUILD.bazel
@@ -141,6 +141,7 @@ go_test(
         "//pkg/types",
         "//pkg/util",
         "//pkg/util/cdcutil",
+        "//pkg/util/chunk",
         "//pkg/util/dbterror/exeerrors",
         "//pkg/util/dbterror/plannererrors",
         "//pkg/util/etcd",
diff --git a/pkg/executor/importer/chunk_process.go b/pkg/executor/importer/chunk_process.go
index 140cfbfe2e..646c88311d 100644
--- a/pkg/executor/importer/chunk_process.go
+++ b/pkg/executor/importer/chunk_process.go
@@ -34,6 +34,7 @@ import (
 	"github.com/pingcap/tidb/pkg/tablecodec"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util"
+	"github.com/pingcap/tidb/pkg/util/chunk"
 	"github.com/pingcap/tidb/pkg/util/syncutil"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
@@ -91,27 +92,47 @@ func parserEncodeReader(parser mydump.Parser, endOffset int64, filename string)
 	}
 }
 
-// queryRowEncodeReader wraps a QueryRow channel as a encodeReaderFn.
-func queryRowEncodeReader(rowCh <-chan QueryRow) encodeReaderFn {
-	return func(ctx context.Context) (data rowToEncode, closed bool, err error) {
+type queryChunkEncodeReader struct {
+	chunkCh <-chan QueryChunk
+	currChk QueryChunk
+	cursor  int
+	numRows int
+}
+
+func (r *queryChunkEncodeReader) readRow(ctx context.Context) (data rowToEncode, closed bool, err error) {
+	if r.currChk.Chk == nil || r.cursor >= r.numRows {
 		select {
 		case <-ctx.Done():
 			err = ctx.Err()
 			return
-		case row, ok := <-rowCh:
+		case currChk, ok := <-r.chunkCh:
 			if !ok {
 				closed = true
 				return
 			}
-			data = rowToEncode{
-				row:       row.Data,
-				rowID:     row.ID,
-				endOffset: -1,
-				resetFn:   func() {},
-			}
-			return
+			r.currChk = currChk
+			r.cursor = 0
+			r.numRows = r.currChk.Chk.NumRows()
 		}
 	}
+
+	row := r.currChk.Chk.GetRow(r.cursor)
+	r.cursor++
+	data = rowToEncode{
+		row:       row.GetDatumRow(r.currChk.Fields),
+		rowID:     r.currChk.RowIDOffset + int64(r.cursor),
+		endOffset: -1,
+		resetFn:   func() {},
+	}
+	return
+}
+
+// queryRowEncodeReader wraps a queryChunkEncodeReader as a encodeReaderFn.
+func queryRowEncodeReader(chunkCh <-chan QueryChunk) encodeReaderFn {
+	reader := queryChunkEncodeReader{chunkCh: chunkCh}
+	return func(ctx context.Context) (data rowToEncode, closed bool, err error) {
+		return reader.readRow(ctx)
+	}
 }
 
 type encodedKVGroupBatch struct {
@@ -493,14 +514,15 @@ func (p *dataDeliver) summaryFields() []zap.Field {
 	}
 }
 
-// QueryRow is a row from query result.
-type QueryRow struct {
-	ID   int64
-	Data []types.Datum
+// QueryChunk is a chunk from query result.
+type QueryChunk struct {
+	Fields      []*types.FieldType
+	Chk         *chunk.Chunk
+	RowIDOffset int64
 }
 
 func newQueryChunkProcessor(
-	rowCh chan QueryRow,
+	chunkCh chan QueryChunk,
 	encoder KVEncoder,
 	keyspace []byte,
 	logger *zap.Logger,
@@ -523,7 +545,7 @@ func newQueryChunkProcessor(
 		deliver:    deliver,
 		enc: newChunkEncoder(
 			chunkName,
-			queryRowEncodeReader(rowCh),
+			queryRowEncodeReader(chunkCh),
 			-1,
 			deliver.sendEncodedData,
 			chunkLogger,
diff --git a/pkg/executor/importer/engine_process.go b/pkg/executor/importer/engine_process.go
index 8d00be701c..0c834a948f 100644
--- a/pkg/executor/importer/engine_process.go
+++ b/pkg/executor/importer/engine_process.go
@@ -106,7 +106,7 @@ func ProcessChunkWithWriter(
 		)
 	case DataSourceTypeQuery:
 		cp = newQueryChunkProcessor(
-			tableImporter.rowCh, encoder, tableImporter.GetKeySpace(), logger,
+			tableImporter.chunkCh, encoder, tableImporter.GetKeySpace(), logger,
 			tableImporter.diskQuotaLock, dataWriter, indexWriter, groupChecksum,
 		)
 	}
diff --git a/pkg/executor/importer/table_import.go b/pkg/executor/importer/table_import.go
index cfed495712..67974496c3 100644
--- a/pkg/executor/importer/table_import.go
+++ b/pkg/executor/importer/table_import.go
@@ -243,7 +243,7 @@ type TableImporter struct {
 	diskQuota       int64
 	diskQuotaLock   *syncutil.RWMutex
 
-	rowCh chan QueryRow
+	chunkCh chan QueryChunk
 }
 
 // NewTableImporterForTest creates a new table importer for test.
@@ -614,9 +614,9 @@ func (ti *TableImporter) CheckDiskQuota(ctx context.Context) {
 	}
 }
 
-// SetSelectedRowCh sets the channel to receive selected rows.
-func (ti *TableImporter) SetSelectedRowCh(ch chan QueryRow) {
-	ti.rowCh = ch
+// SetSelectedChunkCh sets the channel to receive selected rows.
+func (ti *TableImporter) SetSelectedChunkCh(ch chan QueryChunk) {
+	ti.chunkCh = ch
 }
 
 func (ti *TableImporter) closeAndCleanupEngine(engine *backend.OpenedEngine) {
diff --git a/pkg/planner/core/expression_rewriter.go b/pkg/planner/core/expression_rewriter.go
index bf299568b1..afca6b32ed 100644
--- a/pkg/planner/core/expression_rewriter.go
+++ b/pkg/planner/core/expression_rewriter.go
@@ -728,13 +728,8 @@ func (er *expressionRewriter) handleCompareSubquery(ctx context.Context, planCtx
 		er.err = err
 		return v, true
 	}
-
-	noDecorrelate := hintFlags&hint.HintFlagNoDecorrelate > 0
-	if noDecorrelate && len(coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())) == 0 {
-		b.ctx.GetSessionVars().StmtCtx.SetHintWarning(
-			"NO_DECORRELATE() is inapplicable because there are no correlated columns.")
-		noDecorrelate = false
-	}
+	corCols := coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())
+	noDecorrelate := isNoDecorrelate(planCtx, corCols, hintFlags)
 
 	// Only (a,b,c) = any (...) and (a,b,c) != all (...) can use row expression.
 	canMultiCol := (!v.All && v.Op == opcode.EQ) || (v.All && v.Op == opcode.NE)
@@ -1039,13 +1034,8 @@ func (er *expressionRewriter) handleExistSubquery(ctx context.Context, planCtx *
 		return v, true
 	}
 	np = er.popExistsSubPlan(planCtx, np)
-
-	noDecorrelate := hintFlags&hint.HintFlagNoDecorrelate > 0
-	if noDecorrelate && len(coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())) == 0 {
-		b.ctx.GetSessionVars().StmtCtx.SetHintWarning(
-			"NO_DECORRELATE() is inapplicable because there are no correlated columns.")
-		noDecorrelate = false
-	}
+	corCols := coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())
+	noDecorrelate := isNoDecorrelate(planCtx, corCols, hintFlags)
 	semiJoinRewrite := hintFlags&hint.HintFlagSemiJoinRewrite > 0
 	if semiJoinRewrite && noDecorrelate {
 		b.ctx.GetSessionVars().StmtCtx.SetHintWarning(
@@ -1212,16 +1202,11 @@ func (er *expressionRewriter) handleInSubquery(ctx context.Context, planCtx *exp
 	// If the leftKey and the rightKey have different collations, don't convert the sub-query to an inner-join
 	// since when converting we will add a distinct-agg upon the right child and this distinct-agg doesn't have the right collation.
 	// To keep it simple, we forbid this converting if they have different collations.
+	// tested by TestCollateSubQuery.
 	lt, rt := lexpr.GetType(er.sctx.GetEvalCtx()), rexpr.GetType(er.sctx.GetEvalCtx())
 	collFlag := collate.CompatibleCollate(lt.GetCollate(), rt.GetCollate())
-
-	noDecorrelate := hintFlags&hint.HintFlagNoDecorrelate > 0
 	corCols := coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())
-	if len(corCols) == 0 && noDecorrelate {
-		planCtx.builder.ctx.GetSessionVars().StmtCtx.SetHintWarning(
-			"NO_DECORRELATE() is inapplicable because there are no correlated columns.")
-		noDecorrelate = false
-	}
+	noDecorrelate := isNoDecorrelate(planCtx, corCols, hintFlags)
 
 	// If it's not the form of `not in (SUBQUERY)`,
 	// and has no correlated column from the current level plan(if the correlated column is from upper level,
@@ -1272,6 +1257,16 @@ func (er *expressionRewriter) handleInSubquery(ctx context.Context, planCtx *exp
 	return v, true
 }
 
+func isNoDecorrelate(planCtx *exprRewriterPlanCtx, corCols []*expression.CorrelatedColumn, hintFlags uint64) bool {
+	noDecorrelate := hintFlags&hint.HintFlagNoDecorrelate > 0
+	if noDecorrelate && len(corCols) == 0 {
+		planCtx.builder.ctx.GetSessionVars().StmtCtx.SetHintWarning(
+			"NO_DECORRELATE() is inapplicable because there are no correlated columns.")
+		noDecorrelate = false
+	}
+	return noDecorrelate
+}
+
 func (er *expressionRewriter) handleScalarSubquery(ctx context.Context, planCtx *exprRewriterPlanCtx, v *ast.SubqueryExpr) (ast.Node, bool) {
 	intest.AssertNotNil(planCtx)
 	ci := planCtx.builder.prepareCTECheckForSubQuery()
@@ -1282,13 +1277,8 @@ func (er *expressionRewriter) handleScalarSubquery(ctx context.Context, planCtx
 		return v, true
 	}
 	np = planCtx.builder.buildMaxOneRow(np)
-
-	noDecorrelate := hintFlags&hint.HintFlagNoDecorrelate > 0
-	if noDecorrelate && len(coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())) == 0 {
-		planCtx.builder.ctx.GetSessionVars().StmtCtx.SetHintWarning(
-			"NO_DECORRELATE() is inapplicable because there are no correlated columns.")
-		noDecorrelate = false
-	}
+	correlatedColumn := coreusage.ExtractCorColumnsBySchema4LogicalPlan(np, planCtx.plan.Schema())
+	noDecorrelate := isNoDecorrelate(planCtx, correlatedColumn, hintFlags)
 
 	if planCtx.builder.disableSubQueryPreprocessing || len(coreusage.ExtractCorrelatedCols4LogicalPlan(np)) > 0 || hasCTEConsumerInSubPlan(np) {
 		planCtx.plan = planCtx.builder.buildApplyWithJoinType(planCtx.plan, np, logicalop.LeftOuterJoin, noDecorrelate)
diff --git a/pkg/statistics/handle/globalstats/global_stats_async.go b/pkg/statistics/handle/globalstats/global_stats_async.go
index c032c451d5..f4f5bf7ad4 100644
--- a/pkg/statistics/handle/globalstats/global_stats_async.go
+++ b/pkg/statistics/handle/globalstats/global_stats_async.go
@@ -430,8 +430,8 @@ func (a *AsyncMergePartitionStats2GlobalStats) loadHistogramAndTopN(sctx session
 			hists = append(hists, h)
 			topn = append(topn, t)
 		}
-		// if hists and topn are both empty, it means that the partition is empty.
-		// we can skip it to avoid sending empty data to the channel.
+		// This only happens when there are no records for the histogram and TopN in the system tables.
+		// It may be due to the DDL event not having been processed yet (resulting in no histogram), and the table being empty (resulting in no TopNs).
 		if len(hists) == 0 && len(topn) == 0 {
 			continue
 		}
diff --git a/pkg/statistics/histogram.go b/pkg/statistics/histogram.go
index 56eb927157..64d5367f06 100644
--- a/pkg/statistics/histogram.go
+++ b/pkg/statistics/histogram.go
@@ -1443,9 +1443,9 @@ func MergePartitionHist2GlobalHist(sc *stmtctx.StatementContext, hists []*Histog
 	if expBucketNumber == 0 {
 		return nil, errors.Errorf("expBucketNumber can not be zero")
 	}
-	// The empty hists is danger to merge. we cannot get the table information from histograms
-	// The empty hists is very rare. The DDL event was not processed, and in the previous analyze,
-	// this column was not marked as "predict," resulting in it not being analyzed.
+	// This only occurs when there are no histogram records in the histogram system table.
+	// It happens only to tables whose DDL events haven’t been processed yet and that have no indexes or keys,
+	// with the predicate column feature enabled.
 	if len(hists) == 0 {
 		return nil, nil
 	}
diff --git a/pkg/util/collate/collate.go b/pkg/util/collate/collate.go
index c3471f53da..885edd27ad 100644
--- a/pkg/util/collate/collate.go
+++ b/pkg/util/collate/collate.go
@@ -104,7 +104,7 @@ func NewCollationEnabled() bool {
 func CompatibleCollate(collate1, collate2 string) bool {
 	if (collate1 == "utf8mb4_general_ci" || collate1 == "utf8_general_ci") && (collate2 == "utf8mb4_general_ci" || collate2 == "utf8_general_ci") {
 		return true
-	} else if (collate1 == "utf8mb4_bin" || collate1 == "utf8_bin" || collate1 == "latin1_bin") && (collate2 == "utf8mb4_bin" || collate2 == "utf8_bin") {
+	} else if (collate1 == "utf8mb4_bin" || collate1 == "utf8_bin" || collate1 == "latin1_bin") && (collate2 == "utf8mb4_bin" || collate2 == "utf8_bin" || collate2 == "latin1_bin") {
 		return true
 	} else if (collate1 == "utf8mb4_unicode_ci" || collate1 == "utf8_unicode_ci") && (collate2 == "utf8mb4_unicode_ci" || collate2 == "utf8_unicode_ci") {
 		return true


Test patch:
diff --git a/lightning/pkg/importer/table_import_test.go b/lightning/pkg/importer/table_import_test.go
index 04f100185e..bc3323be8a 100644
--- a/lightning/pkg/importer/table_import_test.go
+++ b/lightning/pkg/importer/table_import_test.go
@@ -1549,7 +1549,7 @@ func (s *tableRestoreSuite) TestSchemaIsValid() {
 		// we expect the check failed.
 		{
 			nil,
-			"TiDB schema `db1`.`table1` has 2 columns,and data file has 1 columns, but column colb are missing(.*)",
+			"TiDB schema `db1`.`table1` has 2 columns, and data file has 1 columns, but column colb is missing(.*)",
 			1,
 			false,
 			map[string]*checkpoints.TidbDBInfo{
diff --git a/pkg/executor/importer/importer_testkit_test.go b/pkg/executor/importer/importer_testkit_test.go
index 4d11dd15aa..7bcfdc51dd 100644
--- a/pkg/executor/importer/importer_testkit_test.go
+++ b/pkg/executor/importer/importer_testkit_test.go
@@ -38,6 +38,7 @@ import (
 	"github.com/pingcap/tidb/pkg/meta/autoid"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
 	plannercore "github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
 	"github.com/pingcap/tidb/pkg/planner/core/resolve"
@@ -46,6 +47,7 @@ import (
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
 	"github.com/pingcap/tidb/pkg/types"
+	"github.com/pingcap/tidb/pkg/util/chunk"
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/stretchr/testify/require"
@@ -331,19 +333,35 @@ func TestProcessChunkWith(t *testing.T) {
 			ti.LoadDataController.Close()
 			ti.Backend().CloseEngineMgr()
 		}()
-		rowsCh := make(chan importer.QueryRow, 3)
-		for i := 1; i <= 3; i++ {
-			rowsCh <- importer.QueryRow{
-				ID: int64(i),
-				Data: []types.Datum{
-					types.NewIntDatum(int64((i-1)*3 + 1)),
-					types.NewIntDatum(int64((i-1)*3 + 2)),
-					types.NewIntDatum(int64((i-1)*3 + 3)),
-				},
-			}
+		chkCh := make(chan importer.QueryChunk, 3)
+		fields := make([]*types.FieldType, 0, 3)
+		for i := 0; i < 3; i++ {
+			fields = append(fields, types.NewFieldType(mysql.TypeLong))
 		}
-		close(rowsCh)
-		ti.SetSelectedRowCh(rowsCh)
+		chk := chunk.New(fields, 2, 2)
+		for i := 1; i <= 2; i++ {
+			chk.AppendInt64(0, int64((i-1)*3+1))
+			chk.AppendInt64(1, int64((i-1)*3+2))
+			chk.AppendInt64(2, int64((i-1)*3+3))
+		}
+		chkCh <- importer.QueryChunk{
+			Fields:      fields,
+			Chk:         chk,
+			RowIDOffset: 0,
+		}
+		chk = chunk.New(fields, 1, 1)
+		for i := 3; i <= 3; i++ {
+			chk.AppendInt64(0, int64((i-1)*3+1))
+			chk.AppendInt64(1, int64((i-1)*3+2))
+			chk.AppendInt64(2, int64((i-1)*3+3))
+		}
+		chkCh <- importer.QueryChunk{
+			Fields:      fields,
+			Chk:         chk,
+			RowIDOffset: 2,
+		}
+		close(chkCh)
+		ti.SetSelectedChunkCh(chkCh)
 		kvWriter := mock.NewMockEngineWriter(ctrl)
 		kvWriter.EXPECT().AppendRows(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
 		checksum := verify.NewKVGroupChecksumWithKeyspace(keyspace)
diff --git a/pkg/executor/importer/table_import_testkit_test.go b/pkg/executor/importer/table_import_testkit_test.go
index 020f264a34..2459b64b71 100644
--- a/pkg/executor/importer/table_import_testkit_test.go
+++ b/pkg/executor/importer/table_import_testkit_test.go
@@ -26,12 +26,14 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/backend/local"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
 	plannercore "github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/core/resolve"
 	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util"
+	"github.com/pingcap/tidb/pkg/util/chunk"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/tikv"
 )
@@ -105,18 +107,27 @@ func TestImportFromSelectCleanup(t *testing.T) {
 		&storeHelper{kvStore: store},
 	)
 	require.NoError(t, err)
-	ch := make(chan importer.QueryRow)
-	ti.SetSelectedRowCh(ch)
+	ch := make(chan importer.QueryChunk)
+	ti.SetSelectedChunkCh(ch)
 	var wg util.WaitGroupWrapper
 	wg.Run(func() {
 		defer close(ch)
-		for i := 1; i <= 3; i++ {
-			ch <- importer.QueryRow{
-				ID: int64(i),
-				Data: []types.Datum{
-					types.NewIntDatum(int64(i)),
-				},
-			}
+		fields := make([]*types.FieldType, 0, 3)
+		fields = append(fields, types.NewFieldType(mysql.TypeLong))
+		chk := chunk.New(fields, 2, 2)
+		chk.AppendInt64(0, int64(1))
+		chk.AppendInt64(0, int64(2))
+		ch <- importer.QueryChunk{
+			Fields:      fields,
+			Chk:         chk,
+			RowIDOffset: 0,
+		}
+		chk = chunk.New(fields, 1, 1)
+		chk.AppendInt64(0, int64(3))
+		ch <- importer.QueryChunk{
+			Fields:      fields,
+			Chk:         chk,
+			RowIDOffset: 2,
 		}
 	})
 	_, err = ti.ImportSelectedRows(ctx, tk.Session())
diff --git a/pkg/planner/core/tests/subquery/BUILD.bazel b/pkg/planner/core/tests/subquery/BUILD.bazel
new file mode 100644
index 0000000000..cde70fa53c
--- /dev/null
+++ b/pkg/planner/core/tests/subquery/BUILD.bazel
@@ -0,0 +1,16 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_test")
+
+go_test(
+    name = "subquery_test",
+    timeout = "short",
+    srcs = [
+        "main_test.go",
+        "subquery_test.go",
+    ],
+    flaky = True,
+    deps = [
+        "//pkg/testkit",
+        "//pkg/testkit/testsetup",
+        "@org_uber_go_goleak//:goleak",
+    ],
+)
diff --git a/pkg/planner/core/tests/subquery/main_test.go b/pkg/planner/core/tests/subquery/main_test.go
new file mode 100644
index 0000000000..ce2c860dd7
--- /dev/null
+++ b/pkg/planner/core/tests/subquery/main_test.go
@@ -0,0 +1,34 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package subquery
+
+import (
+	"flag"
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit/testsetup"
+	"go.uber.org/goleak"
+)
+
+func TestMain(m *testing.M) {
+	testsetup.SetupForCommonTest()
+	flag.Parse()
+	opts := []goleak.Option{
+		goleak.IgnoreTopFunction("github.com/golang/glog.(*fileSink).flushDaemon"),
+		goleak.IgnoreTopFunction("github.com/bazelbuild/rules_go/go/tools/bzltestutil.RegisterTimeoutHandler.func1"),
+		goleak.IgnoreTopFunction("github.com/lestrrat-go/httprc.runFetchWorker"),
+	}
+	goleak.VerifyTestMain(m, opts...)
+}
diff --git a/pkg/planner/core/tests/subquery/subquery_test.go b/pkg/planner/core/tests/subquery/subquery_test.go
new file mode 100644
index 0000000000..def7c59548
--- /dev/null
+++ b/pkg/planner/core/tests/subquery/subquery_test.go
@@ -0,0 +1,48 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package subquery
+
+import (
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit"
+)
+
+func TestCollateSubQuery(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(id int, col varchar(100), key ix(col)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;")
+	tk.MustExec("create table t1(id varchar(100)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;")
+	samePlan := testkit.Rows(
+		"IndexHashJoin 8000.00 root  inner join, inner:IndexLookUp, outer key:Column#6, inner key:test.t.col, equal cond:eq(Column#6, test.t.col)",
+		"├─HashAgg(Build) 6400.00 root  group by:Column#12, funcs:firstrow(Column#12)->Column#6",
+		"│ └─TableReader 6400.00 root  data:HashAgg",
+		"│   └─HashAgg 6400.00 cop[tikv]  group by:cast(test.t1.id, var_string(5)), ",
+		"│     └─Selection 8000.00 cop[tikv]  not(isnull(cast(test.t1.id, var_string(5))))",
+		"│       └─TableFullScan 10000.00 cop[tikv] table:t1 keep order:false, stats:pseudo",
+		"└─IndexLookUp(Probe) 8000.00 root  ",
+		"  ├─Selection(Build) 8000.00 cop[tikv]  not(isnull(test.t.col))",
+		"  │ └─IndexRangeScan 8008.01 cop[tikv] table:t, index:ix(col) range: decided by [eq(test.t.col, Column#6)], keep order:false, stats:pseudo",
+		"  └─TableRowIDScan(Probe) 8000.00 cop[tikv] table:t keep order:false, stats:pseudo")
+	tk.MustQuery(`explain format="brief" select * from t use index(ix) where col in (select cast(id as char) from t1);`).
+		Check(samePlan)
+	tk.MustExec(`set collation_connection='utf8_bin';`)
+	tk.MustQuery(`explain format="brief" select * from t use index(ix) where col in (select cast(id as char) from t1);`).
+		Check(samePlan)
+	tk.MustExec(`set collation_connection='latin1_bin';`)
+	tk.MustQuery(`explain format="brief" select * from t use index(ix) where col in (select cast(id as char) from t1);`).
+		Check(samePlan)
+}