Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions lightning/pkg/errormanager/errormanager.go
Original file line number Diff line number Diff line change
Expand Up @@ -616,7 +616,7 @@ func (em *ErrorManager) ReplaceConflictKeys(
return errors.Trace(err)
}
decodedData, _, err := tables.DecodeRawRowData(encoder.SessionCtx.GetExprCtx(),
tbl.Meta(), overwrittenHandle, tbl.Cols(), overwritten)
tbl, overwrittenHandle, tbl.Cols(), overwritten)
if err != nil {
return errors.Trace(err)
}
Expand Down Expand Up @@ -796,7 +796,7 @@ func (em *ErrorManager) ReplaceConflictKeys(
return errors.Trace(err)
}
decodedData, _, err := tables.DecodeRawRowData(encoder.SessionCtx.GetExprCtx(),
tbl.Meta(), handle, tbl.Cols(), latestValue)
tbl, handle, tbl.Cols(), latestValue)
if err != nil {
return errors.Trace(err)
}
Expand Down Expand Up @@ -825,7 +825,7 @@ func (em *ErrorManager) ReplaceConflictKeys(
return errors.Trace(err)
}
decodedData, _, err := tables.DecodeRawRowData(encoder.SessionCtx.GetExprCtx(),
tbl.Meta(), handle, tbl.Cols(), rawValue)
tbl, handle, tbl.Cols(), rawValue)
if err != nil {
return errors.Trace(err)
}
Expand Down
4 changes: 2 additions & 2 deletions pkg/ddl/backfilling_dist_executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -137,9 +137,9 @@ func (s *backfillDistExecutor) newBackfillStepExecutor(

store := s.TaskRuntime.Store()
sessPool := sess.NewSessionPool(s.TaskRuntime.SysSessionPool())
// TODO getTableByTxn is using DDL ctx which is never cancelled except when shutdown.
// TODO This is using DDL ctx which is never cancelled except when shutdown.
// we should move this operation out of GetStepExecutor, and put into Init.
_, tblIface, err := getTableByTxn(ddlObj.ctx, store, jobMeta.SchemaID, jobMeta.TableID)
tblIface, err := getUserTableFromTaskStore(ddlObj.ctx, store, jobMeta)
if err != nil {
return nil, err
}
Expand Down
8 changes: 7 additions & 1 deletion pkg/ddl/backfilling_dist_scheduler.go
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ import (
"github.com/pingcap/tidb/pkg/sessionctx/vardef"
"github.com/pingcap/tidb/pkg/store/helper"
"github.com/pingcap/tidb/pkg/table"
"github.com/pingcap/tidb/pkg/table/tables"
"github.com/pingcap/tidb/pkg/util/backoff"
"github.com/pingcap/tidb/pkg/util/collate"
"github.com/tikv/client-go/v2/oracle"
"github.com/tikv/client-go/v2/tikv"
"go.uber.org/zap"
Expand Down Expand Up @@ -200,7 +202,11 @@ func getUserTableFromTaskStore(
return nil, err
}
// we don't touch table data during add-index, a fake Allocators is enough.
tbl, err := table.TableFromMeta(autoid.NewAllocators(tblInfo.SepAutoInc()), tblInfo)
tbl, err := tables.TableFromMetaWithCollate(
job.ReorgMeta.GetUseNewCollateOrDefault(collate.NewCollationEnabled()),
autoid.NewAllocators(tblInfo.SepAutoInc()),
tblInfo,
)
if err != nil {
return nil, err
}
Expand Down
7 changes: 4 additions & 3 deletions pkg/ddl/backfilling_operators.go
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ func NewAddIndexIngestPipeline(
) (*operator.AsyncPipeline, error) {
indexes := make([]table.Index, 0, len(idxInfos))
for _, idxInfo := range idxInfos {
index, err := tables.NewIndex(tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
index, err := tables.NewIndexWithCollate(tbl.UseNewCollate(), tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -169,7 +169,7 @@ func NewWriteIndexToExternalStoragePipeline(
) (*operator.AsyncPipeline, error) {
indexes := make([]table.Index, 0, len(idxInfos))
for _, idxInfo := range idxInfos {
index, err := tables.NewIndex(tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
index, err := tables.NewIndexWithCollate(tbl.UseNewCollate(), tbl.GetPhysicalID(), tbl.Meta(), idxInfo)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -914,7 +914,8 @@ func (w *indexIngestWorker) WriteChunk(rs *IndexRecordChunk) (count int, bytes i
// skip running the checker in TiDB side.
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())
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())
if err != nil || cnt == 0 {
return 0, 0, err
}
Expand Down
2 changes: 2 additions & 0 deletions pkg/ddl/backfilling_txn_executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ 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"
Expand Down Expand Up @@ -142,6 +143,7 @@ func NewReorgCopContext(
tblInfo,
allIdxInfo,
requestSource,
reorgMeta.GetUseNewCollateOrDefault(collate.NewCollationEnabled()),
)
}

Expand Down
37 changes: 28 additions & 9 deletions pkg/ddl/copr/copr_ctx.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ type CopContextBase struct {
ExprCtx exprctx.BuildContext
PushDownFlags uint64
RequestSource string
UseNewCollate bool

ColumnInfos []*model.ColumnInfo
FieldTypes []*types.FieldType
Expand Down Expand Up @@ -80,6 +81,7 @@ func NewCopContextBase(
tblInfo *model.TableInfo,
idxCols []*model.IndexColumn,
requestSource string,
useNewCollate bool,
) (*CopContextBase, error) {
var err error
usedColumnIDs := make(map[int64]struct{}, len(idxCols))
Expand Down Expand Up @@ -125,8 +127,14 @@ func NewCopContextBase(
handleIDs = []int64{extra.ID}
}

expColInfos, _, err := expression.ColumnInfos2ColumnsAndNames(exprCtx,
ast.CIStr{} /* unused */, tblInfo.Name, colInfos, tblInfo)
expColInfos, _, err := expression.ColumnInfos2ColumnsAndNamesWithCollate(
exprCtx,
ast.CIStr{}, // unused
tblInfo.Name,
colInfos,
tblInfo,
useNewCollate,
)
if err != nil {
return nil, err
}
Expand All @@ -139,6 +147,7 @@ func NewCopContextBase(
ExprCtx: exprCtx,
PushDownFlags: pushDownFlags,
RequestSource: requestSource,
UseNewCollate: useNewCollate,
ColumnInfos: colInfos,
FieldTypes: fieldTps,
ExprColumnInfos: expColInfos,
Expand All @@ -148,27 +157,36 @@ func NewCopContextBase(
}, nil
}

// NewCopContext creates a CopContext.
// NewCopContext creates a CopContext with a fixed collation mode.
func NewCopContext(
exprCtx exprctx.BuildContext,
pushDownFlags uint64,
tblInfo *model.TableInfo,
allIdxInfo []*model.IndexInfo,
requestSource string,
useNewCollate bool,
) (CopContext, error) {
if len(allIdxInfo) == 1 {
return NewCopContextSingleIndex(exprCtx, pushDownFlags, tblInfo, allIdxInfo[0], requestSource)
return NewCopContextSingleIndex(
exprCtx,
pushDownFlags,
tblInfo,
allIdxInfo[0],
requestSource,
useNewCollate,
)
}
return NewCopContextMultiIndex(exprCtx, pushDownFlags, tblInfo, allIdxInfo, requestSource)
return NewCopContextMultiIndex(exprCtx, pushDownFlags, tblInfo, allIdxInfo, requestSource, useNewCollate)
}

// NewCopContextSingleIndex creates a CopContextSingleIndex.
// NewCopContextSingleIndex creates a CopContextSingleIndex with a fixed collation mode.
func NewCopContextSingleIndex(
exprCtx exprctx.BuildContext,
pushDownFlags uint64,
tblInfo *model.TableInfo,
idxInfo *model.IndexInfo,
requestSource string,
useNewCollate bool,
) (*CopContextSingleIndex, error) {
cols := idxInfo.Columns
neededCols, err := tables.ExtractColumnsFromCondition(exprCtx, idxInfo, tblInfo, false)
Expand All @@ -178,7 +196,7 @@ func NewCopContextSingleIndex(
cols = append(cols, neededCols...)
cols = tables.DedupIndexColumns(cols)

base, err := NewCopContextBase(exprCtx, pushDownFlags, tblInfo, cols, requestSource)
base, err := NewCopContextBase(exprCtx, pushDownFlags, tblInfo, cols, requestSource, useNewCollate)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -228,13 +246,14 @@ func (c *CopContextSingleIndex) GetCondition() (expression.Expression, error) {
return expr, nil
}

// NewCopContextMultiIndex creates a CopContextMultiIndex.
// NewCopContextMultiIndex creates a CopContextMultiIndex with a fixed collation mode.
func NewCopContextMultiIndex(
exprCtx exprctx.BuildContext,
pushDownFlags uint64,
tblInfo *model.TableInfo,
allIdxInfo []*model.IndexInfo,
requestSource string,
useNewCollate bool,
) (*CopContextMultiIndex, error) {
approxColLen := 0
for _, idxInfo := range allIdxInfo {
Expand All @@ -252,7 +271,7 @@ func NewCopContextMultiIndex(
}
allIdxCols = tables.DedupIndexColumns(allIdxCols)

base, err := NewCopContextBase(exprCtx, pushDownFlags, tblInfo, allIdxCols, requestSource)
base, err := NewCopContextBase(exprCtx, pushDownFlags, tblInfo, allIdxCols, requestSource, useNewCollate)
if err != nil {
return nil, err
}
Expand Down
5 changes: 4 additions & 1 deletion pkg/ddl/copr/copr_ctx_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,10 @@ func TestNewCopContextSingleIndex(t *testing.T) {
copCtx, err := NewCopContextSingleIndex(
sctx.GetExprCtx(),
sctx.GetSessionVars().StmtCtx.PushDownFlags(),
mockTableInfo, mockIdxInfo, "",
mockTableInfo,
mockIdxInfo,
"",
false,
)
require.NoError(t, err)
base := copCtx.GetBase()
Expand Down
2 changes: 2 additions & 0 deletions pkg/ddl/executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -7501,13 +7501,15 @@ func getJobCheckInterval(action model.ActionType, i int) (time.Duration, bool) {
// NewDDLReorgMeta create a DDL ReorgMeta.
func NewDDLReorgMeta(ctx sessionctx.Context) *model.DDLReorgMeta {
tzName, tzOffset := ddlutil.GetTimeZone(ctx)
useNewCollate := collate.NewCollationEnabled()
return &model.DDLReorgMeta{
SQLMode: ctx.GetSessionVars().SQLMode,
Warnings: make(map[errors.ErrorID]*terror.Error),
WarningsCount: make(map[errors.ErrorID]int64),
Location: &model.TimeZoneLocation{Name: tzName, Offset: tzOffset},
ResourceGroupName: ctx.GetSessionVars().StmtCtx.ResourceGroupName,
Version: model.CurrentReorgMetaVersion,
UseNewCollate: &useNewCollate,
}
}

Expand Down
3 changes: 2 additions & 1 deletion pkg/ddl/export_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ import (
"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"
"github.com/pingcap/tidb/pkg/util/mock"
)

Expand Down Expand Up @@ -80,6 +81,6 @@ func ConvertRowToHandleAndIndexDatum(
c := copCtx.GetBase()
idxData := ddl.ExtractDatumByOffsets(ctx, row, copCtx.IndexColumnOutputOffsets(idxID), c.ExprColumnInfos, idxDataBuf)
handleData := ddl.ExtractDatumByOffsets(ctx, row, c.HandleOutputOffsets, c.ExprColumnInfos, handleDataBuf)
handle, err := ddl.BuildHandle(handleData, c.TableInfo, c.PrimaryKeyInfo, time.Local, errctx.StrictNoWarningContext)
handle, err := ddl.BuildHandle(collate.NewCollationEnabled(), handleData, c.TableInfo, c.PrimaryKeyInfo, time.Local, errctx.StrictNoWarningContext)
return handle, idxData, err
}
9 changes: 4 additions & 5 deletions pkg/ddl/index.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,6 @@ import (
"github.com/pingcap/tidb/pkg/util"
"github.com/pingcap/tidb/pkg/util/backoff"
"github.com/pingcap/tidb/pkg/util/chunk"
"github.com/pingcap/tidb/pkg/util/collate"
"github.com/pingcap/tidb/pkg/util/dbterror"
"github.com/pingcap/tidb/pkg/util/engine"
"github.com/pingcap/tidb/pkg/util/generatedexpr"
Expand Down Expand Up @@ -2478,7 +2477,7 @@ func (w *baseIndexWorker) getIndexRecord(idxInfo *model.IndexInfo, handle kv.Han
idxVal[j] = idxColumnVal
}

rsData := tables.TryGetHandleRestoredDataWrapper(collate.NewCollationEnabled(), w.table.Meta(), nil, w.rowMap, idxInfo)
rsData := tables.TryGetHandleRestoredDataWrapper(w.table, nil, w.rowMap, idxInfo)
idxRecord := &indexRecord{handle: handle, key: recordKey, vals: idxVal, rsData: rsData}
return idxRecord, nil
}
Expand Down Expand Up @@ -2708,6 +2707,7 @@ func writeChunk(
writeStmtBufs *variable.WriteStmtBufs,
copChunk *chunk.Chunk,
tblInfo *model.TableInfo,
useNewCollate bool,
) (rowCnt int, bytes int, err error) {
iter := chunk.NewIterator4Chunk(copChunk)
c := copCtx.GetBase()
Expand All @@ -2732,7 +2732,6 @@ func writeChunk(
}()
needRestoreForIndexes := make([]bool, len(indexes))
restore, pkNeedRestore := false, false
useNewCollate := collate.NewCollationEnabled()
if c.PrimaryKeyInfo != nil && c.TableInfo.IsCommonHandle && c.TableInfo.CommonHandleVersion != 0 {
pkNeedRestore = tables.NeedRestoredData(useNewCollate, c.PrimaryKeyInfo.Columns, c.TableInfo.Columns)
}
Expand All @@ -2753,7 +2752,7 @@ func writeChunk(
restoreDataBuf[i] = *datum.Clone()
}
}
h, err := BuildHandle(handleDataBuf, c.TableInfo, c.PrimaryKeyInfo, loc, errCtx)
h, err := BuildHandle(useNewCollate, handleDataBuf, c.TableInfo, c.PrimaryKeyInfo, loc, errCtx)
if err != nil {
return 0, totalBytes, errors.Trace(err)
}
Expand All @@ -2776,7 +2775,7 @@ func writeChunk(
idxData := idxDataBuf[:len(index.Meta().Columns)]
var rsData []types.Datum
if needRestoreForIndexes[i] {
rsData = getRestoreData(c.TableInfo, copCtx.IndexInfo(idxID), c.PrimaryKeyInfo, restoreDataBuf)
rsData = getRestoreData(useNewCollate, c.TableInfo, copCtx.IndexInfo(idxID), c.PrimaryKeyInfo, restoreDataBuf)
}
kvBytes, err := writeOneKV(ctx, writers[i], index, loc, errCtx, writeStmtBufs, idxData, rsData, h)
if err != nil {
Expand Down
11 changes: 5 additions & 6 deletions pkg/ddl/index_cop.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ import (
"github.com/pingcap/tidb/pkg/types"
"github.com/pingcap/tidb/pkg/util/chunk"
"github.com/pingcap/tidb/pkg/util/codec"
"github.com/pingcap/tidb/pkg/util/collate"
"github.com/pingcap/tidb/pkg/util/logutil"
"github.com/pingcap/tidb/pkg/util/timeutil"
"github.com/pingcap/tipb/go-tipb"
Expand Down Expand Up @@ -146,16 +145,16 @@ func completeErr(err error, idxInfo *model.IndexInfo) error {
return errors.Trace(err)
}

func getRestoreData(tblInfo *model.TableInfo, targetIdx, pkIdx *model.IndexInfo, handleDts []types.Datum) []types.Datum {
if !collate.NewCollationEnabled() || !tblInfo.IsCommonHandle || tblInfo.CommonHandleVersion == 0 {
func getRestoreData(useNewCollate bool, tblInfo *model.TableInfo, targetIdx, pkIdx *model.IndexInfo, handleDts []types.Datum) []types.Datum {
if !useNewCollate || !tblInfo.IsCommonHandle || tblInfo.CommonHandleVersion == 0 {
return nil
}
if pkIdx == nil {
return nil
}
for i, pkIdxCol := range pkIdx.Columns {
pkCol := tblInfo.Columns[pkIdxCol.Offset]
if !types.NeedRestoredData(&pkCol.FieldType) {
if !types.NeedRestoredDataWithCollate(&pkCol.FieldType, useNewCollate) {
// Since the handle data cannot be null, we can use SetNull to
// indicate that this column does not need to be restored.
handleDts[i].SetNull()
Expand Down Expand Up @@ -257,11 +256,11 @@ func ExtractDatumByOffsets(ctx expression.EvalContext, row chunk.Row, offsets []
}

// BuildHandle is exported for test.
func BuildHandle(pkDts []types.Datum, tblInfo *model.TableInfo,
func BuildHandle(useNewCollate bool, pkDts []types.Datum, tblInfo *model.TableInfo,
pkInfo *model.IndexInfo, loc *time.Location, errCtx errctx.Context) (kv.Handle, error) {
if tblInfo.IsCommonHandle {
tablecodec.TruncateIndexValues(tblInfo, pkInfo, pkDts)
handleBytes, err := codec.EncodeKey(loc, nil, pkDts...)
handleBytes, err := codec.NewEncoder(useNewCollate).EncodeKey(loc, nil, pkDts...)
err = errCtx.HandleError(err)
if err != nil {
return nil, err
Expand Down
Loading