Repository: pingcap/tidb (go)

Issue:
BR Grafana Backup & Import Query expression error
## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

There is a query expression error in the tidb_cluster variables option of the Grafana Backup & Import dashboard settings.

### 2. What did you expect to see? (Required)

label_values(tikv_engine_size_bytes{k8s_cluster="$k8s_cluster"}, tidb_cluster)

### 3. What did you see instead (Required)

label_values(tikv_engine_size_bytes{k8s_cluster="$k8s_cluster", tidb_cluster)

### 4. What is your TiDB version? (Required)
v8.1.1、v8.5.1

<!-- Paste the output of SELECT tidb_version() -->

--------------------------------+
| tidb_version()                                                                                                                                                                                                                                |
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Release Version: v8.1.1
Edition: Community
Git Commit Hash: a7df4f9845d5d6a590c5d45dad0dcc9f21aa8765
Git Branch: HEAD
UTC Build Time: 2024-08-22 05:50:03
GoVersion: go1.21.13
Race Enabled: false
Check Table Before Drop: false
Store: tikv |
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

Pull request description:
br: fix a PromQL syntax error
<!--

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 #60212 

Problem Summary:

fix a promql syntax issue in backup&restore panel. As a workaround, please goto tikv panel to check br related metrics.

### What changed and how does it work?

### Check List

Tests <!-- At least one of them must be included. -->

- [ ] Unit test
- [ ] Integration test
- [X] 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/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/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/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)
+}