*: honor task collation across DXF encoding and expression paths

pkg/ddl/backfilling_dist_scheduler.go

@@ -208,15 +208,7 @@ func getUserTableFromTaskStore(
 	})
 	useNewCollate := job.ReorgMeta.GetUseNewCollateOrDefault(defaultUseNewCollate)
 	failpoint.InjectCall("afterResolveUserTableNewCollateForBackfillStep", job, defaultUseNewCollate, useNewCollate)
-	tbl, err := tables.TableFromMetaWithCollate(
-		useNewCollate,
-		autoid.NewAllocators(tblInfo.SepAutoInc()),
-		tblInfo,
-	)
-	if err != nil {
-		return nil, err
-	}
-	return tbl, nil
+	return tables.TableFromMetaWithCollate(useNewCollate, autoid.NewAllocators(tblInfo.SepAutoInc()), tblInfo)
 }
 
 // GetNextStep implements scheduler.Extension interface.

pkg/ddl/backfilling_operators.go

@@ -46,7 +46,6 @@ import (
 	"github.com/pingcap/tidb/pkg/resourcemanager/util"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/table"
-	"github.com/pingcap/tidb/pkg/table/tables"
 	"github.com/pingcap/tidb/pkg/tablecodec"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 	contextutil "github.com/pingcap/tidb/pkg/util/context"
@@ -109,13 +108,9 @@ func NewAddIndexIngestPipeline(
 	concurrency int,
 	collector execute.Collector,
 ) (*operator.AsyncPipeline, error) {
-	indexes := make([]table.Index, 0, len(idxInfos))
-	for _, idxInfo := range idxInfos {
-		index, err := tables.NewIndexWithCollate(tbl.UseNewCollate(), tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
-		if err != nil {
-			return nil, err
-		}
-		indexes = append(indexes, index)
+	indexes, err := indexesForBackfill(tbl, idxInfos)
+	if err != nil {
+		return nil, err
 	}
 	reqSrc := getDDLRequestSource(model.ActionAddIndex)
 	copCtx, err := NewReorgCopContext(reorgMeta, tbl.Meta(), idxInfos, reqSrc)
@@ -167,13 +162,9 @@ func NewWriteIndexToExternalStoragePipeline(
 	collector execute.Collector,
 	tikvCodec tikv.Codec,
 ) (*operator.AsyncPipeline, error) {
-	indexes := make([]table.Index, 0, len(idxInfos))
-	for _, idxInfo := range idxInfos {
-		index, err := tables.NewIndexWithCollate(tbl.UseNewCollate(), tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
-		if err != nil {
-			return nil, err
-		}
-		indexes = append(indexes, index)
+	indexes, err := indexesForBackfill(tbl, idxInfos)
+	if err != nil {
+		return nil, err
 	}
 	reqSrc := getDDLRequestSource(model.ActionAddIndex)
 	copCtx, err := NewReorgCopContext(reorgMeta, tbl.Meta(), idxInfos, reqSrc)
@@ -219,6 +210,23 @@ func NewWriteIndexToExternalStoragePipeline(
 	), nil
 }
 
+func indexesForBackfill(tbl table.PhysicalTable, idxInfos []*model.IndexInfo) ([]table.Index, error) {
+	indexesByID := make(map[int64]table.Index, len(tbl.Indices()))
+	for _, idx := range tbl.Indices() {
+		indexesByID[idx.Meta().ID] = idx
+	}
+
+	indexes := make([]table.Index, 0, len(idxInfos))
+	for _, idxInfo := range idxInfos {
+		idx, ok := indexesByID[idxInfo.ID]
+		if !ok {
+			return nil, errors.Errorf("index ID %d not found in physical table %d", idxInfo.ID, tbl.GetPhysicalID())
+		}
+		indexes = append(indexes, idx)
+	}
+	return indexes, nil
+}
+
 func createChunkPool(copCtx copr.CopContext, reorgMeta *model.DDLReorgMeta) *sync.Pool {
 	return &sync.Pool{
 		New: func() any {
@@ -915,7 +923,7 @@ func (w *indexIngestWorker) WriteChunk(rs *IndexRecordChunk) (count int, bytes i
 		indexConditionCheckers = nil
 	}
 	cnt, kvBytes, err := writeChunk(w.ctx, w.writers, w.indexes, indexConditionCheckers, w.copCtx,
-		sc.TimeZone(), sc.ErrCtx(), vars.GetWriteStmtBufs(), rs.Chunk, w.tbl.Meta(), w.tbl.UseNewCollate())
+		sc.TimeZone(), sc.ErrCtx(), vars.GetWriteStmtBufs(), rs.Chunk, w.tbl.Meta())
 	if err != nil || cnt == 0 {
 		return 0, 0, err
 	}

pkg/ddl/backfilling_test.go

@@ -26,16 +26,20 @@ import (
 	"github.com/pingcap/tidb/pkg/ddl/ingest"
 	distsqlctx "github.com/pingcap/tidb/pkg/distsql/context"
 	"github.com/pingcap/tidb/pkg/errctx"
+	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/expression/exprstatic"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/resourcemanager/pool/workerpool"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/table"
 	"github.com/pingcap/tidb/pkg/types"
+	"github.com/pingcap/tidb/pkg/util/chunk"
+	"github.com/pingcap/tidb/pkg/util/collate"
 	contextutil "github.com/pingcap/tidb/pkg/util/context"
 	"github.com/pingcap/tidb/pkg/util/deeptest"
 	"github.com/pingcap/tidb/pkg/util/mock"
@@ -71,6 +75,85 @@ func TestIndexInfoNotFoundIsNonRetryable(t *testing.T) {
 	require.False(t, (&backfillDistExecutor{}).IsRetryableError(err))
 }
 
+func TestBuildIndexConditionCheckerUsesFixedCollation(t *testing.T) {
+	origin := collate.NewCollationEnabled()
+	collate.SetNewCollationEnabledForTest(true)
+	defer collate.SetNewCollationEnabledForTest(origin)
+
+	originBuildSimpleExpr := expression.BuildSimpleExpr
+	defer func() {
+		expression.BuildSimpleExpr = originBuildSimpleExpr
+	}()
+	expression.BuildSimpleExpr = func(ctx expression.BuildContext, _ ast.ExprNode, opts ...expression.BuildOption) (expression.Expression, error) {
+		var options expression.BuildOptions
+		for _, opt := range opts {
+			opt(&options)
+		}
+		if options.InputSchema == nil {
+			return expression.NewOne(), nil
+		}
+		constantTp := types.NewFieldTypeWithCollation(mysql.TypeVarchar, "utf8mb4_general_ci", 16)
+		return expression.NewFunction(
+			ctx,
+			ast.EQ,
+			types.NewFieldType(mysql.TypeTiny),
+			options.InputSchema.Columns[0],
+			&expression.Constant{Value: types.NewDatum("A"), RetType: constantTp},
+		)
+	}
+
+	colTp := types.NewFieldTypeWithCollation(mysql.TypeVarchar, "utf8mb4_general_ci", 16)
+	colInfo := &model.ColumnInfo{
+		ID:        1,
+		Offset:    0,
+		Name:      ast.NewCIStr("c0"),
+		FieldType: *colTp,
+		State:     model.StatePublic,
+	}
+	idxInfo := &model.IndexInfo{
+		ID:                  1,
+		Name:                ast.NewCIStr("idx"),
+		Columns:             []*model.IndexColumn{{Name: colInfo.Name, Offset: colInfo.Offset}},
+		State:               model.StatePublic,
+		ConditionExprString: "c0 = 'A'",
+	}
+	tblInfo := &model.TableInfo{
+		Name:    ast.NewCIStr("t"),
+		Columns: []*model.ColumnInfo{colInfo},
+		Indices: []*model.IndexInfo{idxInfo},
+	}
+
+	sctx := mock.NewContext()
+	exprCtx := sctx.ExprContext.IntoStatic()
+	copCtx, err := copr.NewCopContextSingleIndex(
+		exprCtx.Apply(exprstatic.WithNewCollationEnabled(false)),
+		sctx.GetSessionVars().StmtCtx.PushDownFlags(),
+		tblInfo,
+		idxInfo,
+		"",
+	)
+	require.NoError(t, err)
+	checker, err := buildIndexConditionChecker(copCtx, tblInfo, idxInfo)
+	require.NoError(t, err)
+	matched, err := checker(chunk.MutRowFromValues("a").ToRow())
+	require.NoError(t, err)
+	require.False(t, matched)
+
+	copCtx, err = copr.NewCopContextSingleIndex(
+		exprCtx.Apply(exprstatic.WithNewCollationEnabled(true)),
+		sctx.GetSessionVars().StmtCtx.PushDownFlags(),
+		tblInfo,
+		idxInfo,
+		"",
+	)
+	require.NoError(t, err)
+	checker, err = buildIndexConditionChecker(copCtx, tblInfo, idxInfo)
+	require.NoError(t, err)
+	matched, err = checker(chunk.MutRowFromValues("a").ToRow())
+	require.NoError(t, err)
+	require.True(t, matched)
+}
+
 func TestPickBackfillType(t *testing.T) {
 	ingest.LitDiskRoot = ingest.NewDiskRootImpl(t.TempDir())
 	ingest.LitMemRoot = ingest.NewMemRootImpl(math.MaxInt64)
@@ -194,6 +277,7 @@ func assertStaticExprContextEqual(t *testing.T, sctx sessionctx.Context, exprCtx
 		f.check(exprCtx)
 		ignoreFields = append(ignoreFields, "$.exprCtxState."+f.field)
 	}
+	ignoreFields = append(ignoreFields, "$.exprCtxState.newCollationEnabled")
 	deeptest.AssertDeepClonedEqual(t, expected, exprCtx, deeptest.WithIgnorePath(ignoreFields))
 
 	// check EvalContext
@@ -227,6 +311,10 @@ func newMockReorgSessCtx(store kv.Storage) sessionctx.Context {
 // compatible with newMockReorgSessCtx(nil).GetExprCtx() to make it safe to replace `mock.Context` usage.
 // After refactor, the TestReorgExprContext can be removed.
 func TestReorgExprContext(t *testing.T) {
+	origin := collate.NewCollationEnabled()
+	collate.SetNewCollationEnabledForTest(true)
+	defer collate.SetNewCollationEnabledForTest(origin)
+
 	// test default expr context
 	store := &mockStorage{client: &mock.Client{}}
 	sctx := newMockReorgSessCtx(store)
@@ -238,26 +326,51 @@ func TestReorgExprContext(t *testing.T) {
 	defaultTypeCtx := evalCtx.TypeCtx()
 	defaultErrCtx := evalCtx.ErrCtx()
 
+	oldCollation := false
+	newCollation := true
+
 	// test expr context from DDLReorgMeta
-	for _, reorg := range []model.DDLReorgMeta{
+	for _, testCase := range []struct {
+		reorg                 model.DDLReorgMeta
+		expectedUseNewCollate bool
+	}{
+		{
+			reorg: model.DDLReorgMeta{
+				SQLMode:           mysql.ModeStrictTransTables | mysql.ModeAllowInvalidDates,
+				Location:          &model.TimeZoneLocation{Name: "Asia/Tokyo"},
+				ReorgTp:           model.ReorgTypeIngest,
+				ResourceGroupName: "rg1",
+				UseNewCollate:     &oldCollation,
+			},
+			expectedUseNewCollate: false,
+		},
 		{
-			SQLMode:           mysql.ModeStrictTransTables | mysql.ModeAllowInvalidDates,
-			Location:          &model.TimeZoneLocation{Name: "Asia/Tokyo"},
-			ReorgTp:           model.ReorgTypeIngest,
-			ResourceGroupName: "rg1",
+			reorg: model.DDLReorgMeta{
+				SQLMode: mysql.ModeAllowInvalidDates,
+				// should load location from system value when reorg.Location is nil
+				Location:          nil,
+				ReorgTp:           model.ReorgTypeTxnMerge,
+				ResourceGroupName: "rg2",
+				UseNewCollate:     &newCollation,
+			},
+			expectedUseNewCollate: true,
 		},
 		{
-			SQLMode: mysql.ModeAllowInvalidDates,
-			// should load location from system value when reorg.Location is nil
-			Location:          nil,
-			ReorgTp:           model.ReorgTypeTxnMerge,
-			ResourceGroupName: "rg2",
+			reorg: model.DDLReorgMeta{
+				SQLMode:           mysql.ModeAllowInvalidDates,
+				Location:          nil,
+				ReorgTp:           model.ReorgTypeTxnMerge,
+				ResourceGroupName: "rg3",
+			},
+			expectedUseNewCollate: true,
 		},
 	} {
+		reorg := testCase.reorg
 		sctx = newMockReorgSessCtx(store)
 		require.NoError(t, initSessCtx(sctx, &reorg))
 		ctx, err := newReorgExprCtxWithReorgMeta(&reorg, sctx.GetSessionVars().StmtCtx.WarnHandler)
 		require.NoError(t, err)
+		require.Equal(t, testCase.expectedUseNewCollate, ctx.NewCollationEnabled())
 		assertStaticExprContextEqual(t, sctx, ctx, ctx.GetStaticEvalCtx().GetWarnHandler())
 		evalCtx := ctx.GetEvalCtx()
 		tc, ec := evalCtx.TypeCtx(), evalCtx.ErrCtx()

pkg/ddl/backfilling_txn_executor.go

@@ -33,7 +33,6 @@ import (
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
 	"github.com/pingcap/tidb/pkg/table"
-	"github.com/pingcap/tidb/pkg/util/collate"
 	contextutil "github.com/pingcap/tidb/pkg/util/context"
 	"github.com/pingcap/tidb/pkg/util/execdetails"
 	"github.com/pingcap/tidb/pkg/util/intest"
@@ -143,7 +142,6 @@ func NewReorgCopContext(
 		tblInfo,
 		allIdxInfo,
 		requestSource,
-		reorgMeta.GetUseNewCollateOrDefault(collate.NewCollationEnabled()),
 	)
 }
 

pkg/ddl/column.go

@@ -43,8 +43,6 @@ import (
 	"github.com/pingcap/tidb/pkg/table"
 	"github.com/pingcap/tidb/pkg/tablecodec"
 	"github.com/pingcap/tidb/pkg/types"
-	"github.com/pingcap/tidb/pkg/util/codec"
-	"github.com/pingcap/tidb/pkg/util/collate"
 	contextutil "github.com/pingcap/tidb/pkg/util/context"
 	"github.com/pingcap/tidb/pkg/util/dbterror"
 	"github.com/pingcap/tidb/pkg/util/intest"
@@ -812,8 +810,7 @@ func (w *updateColumnWorker) getRowRecord(handle kv.Handle, recordKey []byte, ra
 	if w.checksumNeeded {
 		checksum = rowcodec.RawChecksum{Handle: handle}
 	}
-	enc := codec.NewEncoder(collate.NewCollationEnabled())
-	newRowVal, err := tablecodec.EncodeRow(enc, sysTZ, newRow, newColumnIDs, nil, nil, checksum, rd)
+	newRowVal, err := tablecodec.EncodeRow(sysTZ, newRow, newColumnIDs, nil, nil, checksum, rd)
 	err = ec.HandleError(err)
 	if err != nil {
 		return errors.Trace(err)

pkg/ddl/copr/BUILD.bazel

@@ -23,14 +23,15 @@ go_test(
     srcs = ["copr_ctx_test.go"],
     embed = [":copr"],
     flaky = True,
-    shard_count = 3,
+    shard_count = 4,
     deps = [
         "//pkg/expression",
         "//pkg/expression/exprstatic",
         "//pkg/meta/model",
         "//pkg/parser/ast",
         "//pkg/parser/mysql",
         "//pkg/types",
+        "//pkg/util/collate",
         "//pkg/util/mock",
         "@com_github_stretchr_testify//require",
     ],