diff --git a/pkg/ddl/column_change_test.go b/pkg/ddl/column_change_test.go index b147cbbf9dd34..9a4779058a52f 100644 --- a/pkg/ddl/column_change_test.go +++ b/pkg/ddl/column_change_test.go @@ -253,7 +253,7 @@ func checkAddWriteOnly(ctx sessionctx.Context, deleteOnlyTable, writeOnlyTable t if err != nil { return errors.Trace(err) } - err = writeOnlyTable.UpdateRecord(context.Background(), ctx.GetTableCtx(), h, types.MakeDatums(1, 2, 3), types.MakeDatums(2, 2, 3), touchedSlice(writeOnlyTable)) + err = writeOnlyTable.UpdateRecord(ctx.GetTableCtx(), h, types.MakeDatums(1, 2, 3), types.MakeDatums(2, 2, 3), touchedSlice(writeOnlyTable)) if err != nil { return errors.Trace(err) } @@ -317,7 +317,7 @@ func checkAddPublic(sctx sessionctx.Context, writeOnlyTable, publicTable table.T return errors.Errorf("%v", oldRow) } newRow := types.MakeDatums(3, 4, oldRow[2].GetValue()) - err = writeOnlyTable.UpdateRecord(context.Background(), sctx.GetTableCtx(), h, oldRow, newRow, touchedSlice(writeOnlyTable)) + err = writeOnlyTable.UpdateRecord(sctx.GetTableCtx(), h, oldRow, newRow, touchedSlice(writeOnlyTable)) if err != nil { return errors.Trace(err) } diff --git a/pkg/ddl/index_change_test.go b/pkg/ddl/index_change_test.go index 2d8f262462ce4..34ad661cecccf 100644 --- a/pkg/ddl/index_change_test.go +++ b/pkg/ddl/index_change_test.go @@ -173,7 +173,7 @@ func checkAddWriteOnlyForAddIndex(ctx sessionctx.Context, delOnlyTbl, writeOnlyT } // WriteOnlyTable: update t set c2 = 1 where c1 = 4 and c2 = 4 - err = writeOnlyTbl.UpdateRecord(context.Background(), ctx.GetTableCtx(), kv.IntHandle(4), types.MakeDatums(4, 4), types.MakeDatums(4, 1), touchedSlice(writeOnlyTbl)) + err = writeOnlyTbl.UpdateRecord(ctx.GetTableCtx(), kv.IntHandle(4), types.MakeDatums(4, 4), types.MakeDatums(4, 1), touchedSlice(writeOnlyTbl)) if err != nil { return errors.Trace(err) } @@ -183,7 +183,7 @@ func checkAddWriteOnlyForAddIndex(ctx sessionctx.Context, delOnlyTbl, writeOnlyT } // DeleteOnlyTable: update t set c2 = 3 where c1 = 4 and c2 = 1 - err = delOnlyTbl.UpdateRecord(context.Background(), ctx.GetTableCtx(), kv.IntHandle(4), types.MakeDatums(4, 1), types.MakeDatums(4, 3), touchedSlice(writeOnlyTbl)) + err = delOnlyTbl.UpdateRecord(ctx.GetTableCtx(), kv.IntHandle(4), types.MakeDatums(4, 1), types.MakeDatums(4, 3), touchedSlice(writeOnlyTbl)) if err != nil { return errors.Trace(err) } @@ -250,7 +250,7 @@ func checkAddPublicForAddIndex(ctx sessionctx.Context, writeTbl, publicTbl table } // WriteOnlyTable: update t set c2 = 5 where c1 = 7 and c2 = 7 - err = writeTbl.UpdateRecord(context.Background(), ctx.GetTableCtx(), kv.IntHandle(7), types.MakeDatums(7, 7), types.MakeDatums(7, 5), touchedSlice(writeTbl)) + err = writeTbl.UpdateRecord(ctx.GetTableCtx(), kv.IntHandle(7), types.MakeDatums(7, 7), types.MakeDatums(7, 5), touchedSlice(writeTbl)) if err != nil { return errors.Trace(err) } @@ -328,7 +328,7 @@ func checkDropWriteOnly(ctx sessionctx.Context, publicTbl, writeTbl table.Table) } // WriteOnlyTable update t set c2 = 7 where c1 = 8 and c2 = 8 - err = writeTbl.UpdateRecord(context.Background(), ctx.GetTableCtx(), kv.IntHandle(8), types.MakeDatums(8, 8), types.MakeDatums(8, 7), touchedSlice(writeTbl)) + err = writeTbl.UpdateRecord(ctx.GetTableCtx(), kv.IntHandle(8), types.MakeDatums(8, 8), types.MakeDatums(8, 7), touchedSlice(writeTbl)) if err != nil { return errors.Trace(err) } @@ -383,7 +383,7 @@ func checkDropDeleteOnly(ctx sessionctx.Context, writeTbl, delTbl table.Table) e } // DeleteOnlyTable update t set c2 = 10 where c1 = 9 - err = delTbl.UpdateRecord(context.Background(), ctx.GetTableCtx(), kv.IntHandle(9), types.MakeDatums(9, 9), types.MakeDatums(9, 10), touchedSlice(delTbl)) + err = delTbl.UpdateRecord(ctx.GetTableCtx(), kv.IntHandle(9), types.MakeDatums(9, 9), types.MakeDatums(9, 10), touchedSlice(delTbl)) if err != nil { return errors.Trace(err) } diff --git a/pkg/executor/write.go b/pkg/executor/write.go index 98d8518bd4215..aa71e48bdab51 100644 --- a/pkg/executor/write.go +++ b/pkg/executor/write.go @@ -194,7 +194,7 @@ func updateRecord( } } else { // Update record to new value and update index. - if err := t.UpdateRecord(ctx, sctx.GetTableCtx(), h, oldData, newData, modified); err != nil { + if err := t.UpdateRecord(sctx.GetTableCtx(), h, oldData, newData, modified, table.WithCtx(ctx)); err != nil { if terr, ok := errors.Cause(err).(*terror.Error); ok && (terr.Code() == errno.ErrNoPartitionForGivenValue || terr.Code() == errno.ErrRowDoesNotMatchGivenPartitionSet) { ec := sctx.GetSessionVars().StmtCtx.ErrCtx() return false, ec.HandleError(err) diff --git a/pkg/infoschema/tables.go b/pkg/infoschema/tables.go index 89e5f088fac40..144c0ae9cc2a4 100644 --- a/pkg/infoschema/tables.go +++ b/pkg/infoschema/tables.go @@ -2436,7 +2436,7 @@ func (it *infoschemaTable) RemoveRecord(ctx table.MutateContext, h kv.Handle, r } // UpdateRecord implements table.Table UpdateRecord interface. -func (it *infoschemaTable) UpdateRecord(gctx context.Context, ctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool) error { +func (it *infoschemaTable) UpdateRecord(ctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { return table.ErrUnsupportedOp } @@ -2529,7 +2529,7 @@ func (vt *VirtualTable) RemoveRecord(ctx table.MutateContext, h kv.Handle, r []t } // UpdateRecord implements table.Table UpdateRecord interface. -func (vt *VirtualTable) UpdateRecord(ctx context.Context, sctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool) error { +func (vt *VirtualTable) UpdateRecord(ctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { return table.ErrUnsupportedOp } diff --git a/pkg/table/BUILD.bazel b/pkg/table/BUILD.bazel index 33b766c024e52..90d494dd7ad75 100644 --- a/pkg/table/BUILD.bazel +++ b/pkg/table/BUILD.bazel @@ -53,7 +53,7 @@ go_test( embed = [":table"], flaky = True, race = "on", - shard_count = 9, + shard_count = 10, deps = [ "//pkg/errctx", "//pkg/errno", diff --git a/pkg/table/index.go b/pkg/table/index.go index d451a22db1eca..9c523f7bf9606 100644 --- a/pkg/table/index.go +++ b/pkg/table/index.go @@ -15,7 +15,6 @@ package table import ( - "context" "time" "github.com/pingcap/tidb/pkg/errctx" @@ -32,24 +31,37 @@ type IndexIterator interface { // CreateIdxOpt contains the options will be used when creating an index. type CreateIdxOpt struct { - Ctx context.Context - Untouched bool // If true, the index key/value is no need to commit. + commonMutateOpt IgnoreAssertion bool FromBackFill bool } +// NewCreateIdxOpt creates a new CreateIdxOpt. +func NewCreateIdxOpt(opts ...CreateIdxOption) *CreateIdxOpt { + opt := &CreateIdxOpt{} + for _, o := range opts { + o.ApplyCreateIdxOpt(opt) + } + return opt +} + +// CreateIdxOption is defined for the Create() method of the Index interface. +type CreateIdxOption interface { + ApplyCreateIdxOpt(*CreateIdxOpt) +} + // CreateIdxOptFunc is defined for the Create() method of Index interface. // Here is a blog post about how to use this pattern: // https://dave.cheney.net/2014/10/17/functional-options-for-friendly-apis type CreateIdxOptFunc func(*CreateIdxOpt) -// IndexIsUntouched uses to indicate the index kv is untouched. -var IndexIsUntouched CreateIdxOptFunc = func(opt *CreateIdxOpt) { - opt.Untouched = true +// ApplyCreateIdxOpt implements the CreateIdxOption interface. +func (f CreateIdxOptFunc) ApplyCreateIdxOpt(opt *CreateIdxOpt) { + f(opt) } // WithIgnoreAssertion uses to indicate the process can ignore assertion. -var WithIgnoreAssertion = func(opt *CreateIdxOpt) { +var WithIgnoreAssertion CreateIdxOptFunc = func(opt *CreateIdxOpt) { opt.IgnoreAssertion = true } @@ -57,18 +69,10 @@ var WithIgnoreAssertion = func(opt *CreateIdxOpt) { // In the backfill-merge process, the index KVs from DML will be redirected to // the temp index. On the other hand, the index KVs from DDL backfill worker should // never be redirected to the temp index. -var FromBackfill = func(opt *CreateIdxOpt) { +var FromBackfill CreateIdxOptFunc = func(opt *CreateIdxOpt) { opt.FromBackFill = true } -// WithCtx returns a CreateIdxFunc. -// This option is used to pass context.Context. -func WithCtx(ctx context.Context) CreateIdxOptFunc { - return func(opt *CreateIdxOpt) { - opt.Ctx = ctx - } -} - // Index is the interface for index data on KV store. type Index interface { // Meta returns IndexInfo. @@ -76,7 +80,7 @@ type Index interface { // TableMeta returns TableInfo TableMeta() *model.TableInfo // Create supports insert into statement. - Create(ctx MutateContext, txn kv.Transaction, indexedValues []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...CreateIdxOptFunc) (kv.Handle, error) + Create(ctx MutateContext, txn kv.Transaction, indexedValues []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...CreateIdxOption) (kv.Handle, error) // Delete supports delete from statement. Delete(ctx MutateContext, txn kv.Transaction, indexedValues []types.Datum, h kv.Handle) error // GenIndexKVIter generate index key and value for multi-valued index, use iterator to reduce the memory allocation. diff --git a/pkg/table/table.go b/pkg/table/table.go index 6e4eb24a26c92..c971d751014c9 100644 --- a/pkg/table/table.go +++ b/pkg/table/table.go @@ -119,38 +119,106 @@ var ( // RecordIterFunc is used for low-level record iteration. type RecordIterFunc func(h kv.Handle, rec []types.Datum, cols []*Column) (more bool, err error) +// commonMutateOpt is the common options for mutating a table. +type commonMutateOpt struct { + Ctx context.Context +} + // AddRecordOpt contains the options will be used when adding a record. type AddRecordOpt struct { - CreateIdxOpt + commonMutateOpt IsUpdate bool ReserveAutoID int } +// NewAddRecordOpt creates a new AddRecordOpt with options. +func NewAddRecordOpt(opts ...AddRecordOption) *AddRecordOpt { + opt := &AddRecordOpt{} + for _, o := range opts { + o.ApplyAddRecordOpt(opt) + } + return opt +} + +// GetCreateIdxOpt creates a CreateIdxOpt. +func (opt *AddRecordOpt) GetCreateIdxOpt() *CreateIdxOpt { + return &CreateIdxOpt{commonMutateOpt: opt.commonMutateOpt} +} + // AddRecordOption is defined for the AddRecord() method of the Table interface. type AddRecordOption interface { - ApplyOn(*AddRecordOpt) + ApplyAddRecordOpt(*AddRecordOpt) +} + +// UpdateRecordOpt contains the options will be used when updating a record. +type UpdateRecordOpt struct { + commonMutateOpt +} + +// NewUpdateRecordOpt creates a new UpdateRecordOpt with options. +func NewUpdateRecordOpt(opts ...UpdateRecordOption) *UpdateRecordOpt { + opt := &UpdateRecordOpt{} + for _, o := range opts { + o.ApplyUpdateRecordOpt(opt) + } + return opt +} + +// GetAddRecordOpt creates a AddRecordOpt. +func (opt *UpdateRecordOpt) GetAddRecordOpt() *AddRecordOpt { + return &AddRecordOpt{commonMutateOpt: opt.commonMutateOpt} +} + +// GetCreateIdxOpt creates a CreateIdxOpt. +func (opt *UpdateRecordOpt) GetCreateIdxOpt() *CreateIdxOpt { + return &CreateIdxOpt{commonMutateOpt: opt.commonMutateOpt} +} + +// UpdateRecordOption is defined for the UpdateRecord() method of the Table interface. +type UpdateRecordOption interface { + ApplyUpdateRecordOpt(*UpdateRecordOpt) +} + +// CommonMutateOptFunc is a function to provide common options for mutating a table. +type CommonMutateOptFunc func(*commonMutateOpt) + +// ApplyAddRecordOpt implements the AddRecordOption interface. +func (f CommonMutateOptFunc) ApplyAddRecordOpt(opt *AddRecordOpt) { + f(&opt.commonMutateOpt) +} + +// ApplyUpdateRecordOpt implements the UpdateRecordOption interface. +func (f CommonMutateOptFunc) ApplyUpdateRecordOpt(opt *UpdateRecordOpt) { + f(&opt.commonMutateOpt) +} + +// ApplyCreateIdxOpt implements the CreateIdxOption interface. +func (f CommonMutateOptFunc) ApplyCreateIdxOpt(opt *CreateIdxOpt) { + f(&opt.commonMutateOpt) +} + +// WithCtx returns a CommonMutateOptFunc. +// This option is used to pass context.Context. +func WithCtx(ctx context.Context) CommonMutateOptFunc { + return func(opt *commonMutateOpt) { + opt.Ctx = ctx + } } // WithReserveAutoIDHint tells the AddRecord operation to reserve a batch of auto ID in the stmtctx. type WithReserveAutoIDHint int -// ApplyOn implements the AddRecordOption interface. -func (n WithReserveAutoIDHint) ApplyOn(opt *AddRecordOpt) { +// ApplyAddRecordOpt implements the AddRecordOption interface. +func (n WithReserveAutoIDHint) ApplyAddRecordOpt(opt *AddRecordOpt) { opt.ReserveAutoID = int(n) } -// ApplyOn implements the AddRecordOption interface, so any CreateIdxOptFunc -// can be passed as the optional argument to the table.AddRecord method. -func (f CreateIdxOptFunc) ApplyOn(opt *AddRecordOpt) { - f(&opt.CreateIdxOpt) -} - // IsUpdate is a defined value for AddRecordOptFunc. var IsUpdate AddRecordOption = isUpdate{} type isUpdate struct{} -func (i isUpdate) ApplyOn(opt *AddRecordOpt) { +func (i isUpdate) ApplyAddRecordOpt(opt *AddRecordOpt) { opt.IsUpdate = true } @@ -202,7 +270,7 @@ type Table interface { AddRecord(ctx MutateContext, r []types.Datum, opts ...AddRecordOption) (recordID kv.Handle, err error) // UpdateRecord updates a row which should contain only writable columns. - UpdateRecord(gctx context.Context, ctx MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool) error + UpdateRecord(ctx MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool, opts ...UpdateRecordOption) error // RemoveRecord removes a row in the table. RemoveRecord(ctx MutateContext, h kv.Handle, r []types.Datum) error diff --git a/pkg/table/table_test.go b/pkg/table/table_test.go index cfdcdd8e3dafc..a0dc976eb2e2f 100644 --- a/pkg/table/table_test.go +++ b/pkg/table/table_test.go @@ -15,6 +15,7 @@ package table import ( + "context" "testing" mysql "github.com/pingcap/tidb/pkg/errno" @@ -41,3 +42,39 @@ func TestErrorCode(t *testing.T) { require.Equal(t, mysql.ErrNoPartitionForGivenValue, int(terror.ToSQLError(ErrNoPartitionForGivenValue).Code)) require.Equal(t, mysql.ErrLockOrActiveTransaction, int(terror.ToSQLError(ErrLockOrActiveTransaction).Code)) } + +func TestOptions(t *testing.T) { + ctx := context.WithValue(context.Background(), "test", "test") + // NewAddRecordOpt without option + addOpt := NewAddRecordOpt() + require.Equal(t, AddRecordOpt{}, *addOpt) + require.Equal(t, CreateIdxOpt{}, *(addOpt.GetCreateIdxOpt())) + // NewAddRecordOpt with options + addOpt = NewAddRecordOpt(WithCtx(ctx), IsUpdate, WithReserveAutoIDHint(12)) + require.Equal(t, AddRecordOpt{ + commonMutateOpt: commonMutateOpt{Ctx: ctx}, + IsUpdate: true, + ReserveAutoID: 12, + }, *addOpt) + require.Equal(t, CreateIdxOpt{commonMutateOpt: commonMutateOpt{Ctx: ctx}}, *(addOpt.GetCreateIdxOpt())) + // NewUpdateRecordOpt without option + updateOpt := NewUpdateRecordOpt() + require.Equal(t, UpdateRecordOpt{}, *updateOpt) + require.Equal(t, AddRecordOpt{}, *(updateOpt.GetAddRecordOpt())) + require.Equal(t, CreateIdxOpt{}, *(updateOpt.GetCreateIdxOpt())) + // NewUpdateRecordOpt with options + updateOpt = NewUpdateRecordOpt(WithCtx(ctx)) + require.Equal(t, UpdateRecordOpt{commonMutateOpt: commonMutateOpt{Ctx: ctx}}, *updateOpt) + require.Equal(t, AddRecordOpt{commonMutateOpt: commonMutateOpt{Ctx: ctx}}, *(updateOpt.GetAddRecordOpt())) + require.Equal(t, CreateIdxOpt{commonMutateOpt: commonMutateOpt{Ctx: ctx}}, *(updateOpt.GetCreateIdxOpt())) + // NewCreateIdxOpt without option + createIdxOpt := NewCreateIdxOpt() + require.Equal(t, CreateIdxOpt{}, *createIdxOpt) + // NewCreateIdxOpt with options + createIdxOpt = NewCreateIdxOpt(WithCtx(ctx), WithIgnoreAssertion, FromBackfill) + require.Equal(t, CreateIdxOpt{ + commonMutateOpt: commonMutateOpt{Ctx: ctx}, + IgnoreAssertion: true, + FromBackFill: true, + }, *createIdxOpt) +} diff --git a/pkg/table/tables/BUILD.bazel b/pkg/table/tables/BUILD.bazel index 2e653027c6c09..86c641ae56ba7 100644 --- a/pkg/table/tables/BUILD.bazel +++ b/pkg/table/tables/BUILD.bazel @@ -43,6 +43,7 @@ go_library( "//pkg/util/dbterror", "//pkg/util/generatedexpr", "//pkg/util/hack", + "//pkg/util/intest", "//pkg/util/logutil", "//pkg/util/ranger", "//pkg/util/rowcodec", diff --git a/pkg/table/tables/bench_test.go b/pkg/table/tables/bench_test.go index 3131a066fcb6f..67bb98d2a888f 100644 --- a/pkg/table/tables/bench_test.go +++ b/pkg/table/tables/bench_test.go @@ -23,6 +23,7 @@ import ( "github.com/pingcap/tidb/pkg/parser/model" "github.com/pingcap/tidb/pkg/sessionctx/variable" "github.com/pingcap/tidb/pkg/sessiontxn" + "github.com/pingcap/tidb/pkg/table" "github.com/pingcap/tidb/pkg/testkit" "github.com/pingcap/tidb/pkg/types" _ "github.com/pingcap/tidb/pkg/util/context" @@ -198,7 +199,7 @@ func BenchmarkUpdateRecordInPipelinedDML(b *testing.B) { for j := 0; j < batchSize; j++ { // Update record handle := kv.IntHandle(j) - err := tb.UpdateRecord(context.TODO(), se.GetTableCtx(), handle, records[j], newData[j], touched) + err := tb.UpdateRecord(se.GetTableCtx(), handle, records[j], newData[j], touched, table.WithCtx(context.TODO())) if err != nil { b.Fatal(err) } diff --git a/pkg/table/tables/cache.go b/pkg/table/tables/cache.go index 515a2a781f5bc..1051ee8a995f6 100644 --- a/pkg/table/tables/cache.go +++ b/pkg/table/tables/cache.go @@ -254,13 +254,13 @@ func txnCtxAddCachedTable(sctx table.MutateContext, tid int64, handle *cachedTab } // UpdateRecord implements table.Table -func (c *cachedTable) UpdateRecord(ctx context.Context, sctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool) error { +func (c *cachedTable) UpdateRecord(ctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { // Prevent furthur writing when the table is already too large. if atomic.LoadInt64(&c.totalSize) > cachedTableSizeLimit { return table.ErrOptOnCacheTable.GenWithStackByArgs("table too large") } - txnCtxAddCachedTable(sctx, c.Meta().ID, c) - return c.TableCommon.UpdateRecord(ctx, sctx, h, oldData, newData, touched) + txnCtxAddCachedTable(ctx, c.Meta().ID, c) + return c.TableCommon.UpdateRecord(ctx, h, oldData, newData, touched, opts...) } // RemoveRecord implements table.Table RemoveRecord interface. diff --git a/pkg/table/tables/index.go b/pkg/table/tables/index.go index 0f9eaf6ebddc3..f2eacd72838a5 100644 --- a/pkg/table/tables/index.go +++ b/pkg/table/tables/index.go @@ -158,15 +158,15 @@ out: // Create creates a new entry in the kvIndex data. // If the index is unique and there is an existing entry with the same key, // Create will return the existing entry's handle as the first return value, ErrKeyExists as the second return value. -func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValue []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...table.CreateIdxOptFunc) (kv.Handle, error) { +func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValue []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...table.CreateIdxOption) (kv.Handle, error) { + opt := table.NewCreateIdxOpt(opts...) + return c.create(sctx, txn, indexedValue, h, handleRestoreData, false, opt) +} + +func (c *index) create(sctx table.MutateContext, txn kv.Transaction, indexedValue []types.Datum, h kv.Handle, handleRestoreData []types.Datum, untouched bool, opt *table.CreateIdxOpt) (kv.Handle, error) { if c.Meta().Unique { txn.CacheTableInfo(c.phyTblID, c.tblInfo) } - var opt table.CreateIdxOpt - for _, fn := range opts { - fn(&opt) - } - indexedValues := c.getIndexedValue(indexedValue) ctx := opt.Ctx if ctx != nil { @@ -203,10 +203,10 @@ func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValu if txn.IsPipelined() { // For pipelined DML, disable the untouched optimization to avoid extra RPCs for MemBuffer.Get(). // TODO: optimize this. - opt.Untouched = false + untouched = false } - if opt.Untouched { + if untouched { txn, err1 := sctx.Txn(true) if err1 != nil { return nil, err1 @@ -228,7 +228,7 @@ func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValu return nil, err } if keyFlags.HasPresumeKeyNotExists() { - opt.Untouched = false + untouched = false } } } @@ -240,7 +240,7 @@ func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValu c.needRestoredData = NeedRestoredData(c.idxInfo.Columns, c.tblInfo.Columns) }) idxVal, err := tablecodec.GenIndexValuePortal(loc, c.tblInfo, c.idxInfo, - c.needRestoredData, distinct, opt.Untouched, value, h, c.phyTblID, handleRestoreData, nil) + c.needRestoredData, distinct, untouched, value, h, c.phyTblID, handleRestoreData, nil) err = ec.HandleError(err) if err != nil { return nil, err @@ -248,9 +248,9 @@ func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValu opt.IgnoreAssertion = opt.IgnoreAssertion || c.idxInfo.State != model.StatePublic - if !distinct || skipCheck || opt.Untouched { + if !distinct || skipCheck || untouched { val := idxVal - if opt.Untouched && (keyIsTempIdxKey || len(tempKey) > 0) { + if untouched && (keyIsTempIdxKey || len(tempKey) > 0) { // Untouched key-values never occur in the storage and the temp index is not public. // It is unnecessary to write the untouched temp index key-values. continue @@ -271,7 +271,7 @@ func (c *index) Create(sctx table.MutateContext, txn kv.Transaction, indexedValu return nil, err } } - if !opt.IgnoreAssertion && (!opt.Untouched) { + if !opt.IgnoreAssertion && (!untouched) { if sctx.GetSessionVars().LazyCheckKeyNotExists() && !txn.IsPessimistic() { err = txn.SetAssertion(key, kv.SetAssertUnknown) } else { diff --git a/pkg/table/tables/partition.go b/pkg/table/tables/partition.go index 3f94e2b9379ee..e2a5e002928d7 100644 --- a/pkg/table/tables/partition.go +++ b/pkg/table/tables/partition.go @@ -16,7 +16,6 @@ package tables import ( "bytes" - "context" stderr "errors" "fmt" "hash/crc32" @@ -1491,6 +1490,11 @@ func (t *partitionedTable) locateHashPartition(ctx expression.EvalContext, partE // GetPartition returns a Table, which is actually a partition. func (t *partitionedTable) GetPartition(pid int64) table.PhysicalTable { + return t.getPartition(pid) +} + +// getPartition returns a Table, which is actually a partition. +func (t *partitionedTable) getPartition(pid int64) *partition { // Attention, can't simply use `return t.partitions[pid]` here. // Because A nil of type *partition is a kind of `table.PhysicalTable` part, ok := t.partitions[pid] @@ -1594,6 +1598,7 @@ func (t *partitionedTable) AddRecord(ctx table.MutateContext, r []types.Datum, o } func partitionedTableAddRecord(ctx table.MutateContext, t *partitionedTable, r []types.Datum, partitionSelection map[int64]struct{}, opts []table.AddRecordOption) (recordID kv.Handle, err error) { + opt := table.NewAddRecordOpt(opts...) pid, err := t.locatePartition(ctx.GetExprCtx().GetEvalCtx(), r) if err != nil { return nil, errors.Trace(err) @@ -1615,8 +1620,8 @@ func partitionedTableAddRecord(ctx table.MutateContext, t *partitionedTable, r [ return nil, errors.WithStack(err) } } - tbl := t.GetPartition(pid) - recordID, err = tbl.AddRecord(ctx, r, opts...) + tbl := t.getPartition(pid) + recordID, err = tbl.addRecord(ctx, r, opt) if err != nil { return } @@ -1629,8 +1634,8 @@ func partitionedTableAddRecord(ctx table.MutateContext, t *partitionedTable, r [ if err != nil { return nil, errors.Trace(err) } - tbl = t.GetPartition(pid) - recordID, err = tbl.AddRecord(ctx, r, opts...) + tbl = t.getPartition(pid) + recordID, err = tbl.addRecord(ctx, r, opt) if err != nil { return } @@ -1712,15 +1717,16 @@ func (t *partitionedTable) GetAllPartitionIDs() []int64 { // UpdateRecord implements table.Table UpdateRecord interface. // `touched` means which columns are really modified, used for secondary indices. // Length of `oldData` and `newData` equals to length of `t.WritableCols()`. -func (t *partitionedTable) UpdateRecord(ctx context.Context, sctx table.MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool) error { - return partitionedTableUpdateRecord(ctx, sctx, t, h, currData, newData, touched, nil) +func (t *partitionedTable) UpdateRecord(ctx table.MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { + return partitionedTableUpdateRecord(ctx, t, h, currData, newData, touched, nil, opts...) } -func (t *partitionTableWithGivenSets) UpdateRecord(ctx context.Context, sctx table.MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool) error { - return partitionedTableUpdateRecord(ctx, sctx, t.partitionedTable, h, currData, newData, touched, t.givenSetPartitions) +func (t *partitionTableWithGivenSets) UpdateRecord(ctx table.MutateContext, h kv.Handle, currData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { + return partitionedTableUpdateRecord(ctx, t.partitionedTable, h, currData, newData, touched, t.givenSetPartitions, opts...) } -func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, t *partitionedTable, h kv.Handle, currData, newData []types.Datum, touched []bool, partitionSelection map[int64]struct{}) error { +func partitionedTableUpdateRecord(ctx table.MutateContext, t *partitionedTable, h kv.Handle, currData, newData []types.Datum, touched []bool, partitionSelection map[int64]struct{}, opts ...table.UpdateRecordOption) error { + opt := table.NewUpdateRecordOpt(opts...) ectx := ctx.GetExprCtx() from, err := t.locatePartition(ectx.GetEvalCtx(), currData) if err != nil { @@ -1767,7 +1773,7 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, return errors.Trace(err) } - _, err = t.GetPartition(to).AddRecord(ctx, newData) + _, err = t.getPartition(to).addRecord(ctx, newData, opt.GetAddRecordOpt()) if err != nil { return errors.Trace(err) } @@ -1789,7 +1795,7 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, } if newTo == newFrom && newTo != 0 { // Update needs to be done in StateDeleteOnly as well - err = t.GetPartition(newTo).UpdateRecord(gctx, ctx, h, currData, newData, touched) + err = t.getPartition(newTo).updateRecord(ctx, h, currData, newData, touched, opt) if err != nil { return errors.Trace(err) } @@ -1798,14 +1804,14 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, } if newFrom != 0 { - err = t.GetPartition(newFrom).RemoveRecord(ctx, h, currData) + err = t.getPartition(newFrom).RemoveRecord(ctx, h, currData) // TODO: Can this happen? When the data is not yet backfilled? if err != nil { return errors.Trace(err) } } if newTo != 0 && t.Meta().GetPartitionInfo().DDLState != model.StateDeleteOnly { - _, err = t.GetPartition(newTo).AddRecord(ctx, newData) + _, err = t.getPartition(newTo).addRecord(ctx, newData, opt.GetAddRecordOpt()) if err != nil { return errors.Trace(err) } @@ -1813,8 +1819,8 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, memBuffer.Release(sh) return nil } - tbl := t.GetPartition(to) - err = tbl.UpdateRecord(gctx, ctx, h, currData, newData, touched) + tbl := t.getPartition(to) + err = tbl.updateRecord(ctx, h, currData, newData, touched, opt) if err != nil { return errors.Trace(err) } @@ -1830,11 +1836,11 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, return errors.Trace(err) } if newTo == newFrom { - tbl = t.GetPartition(newTo) + tbl = t.getPartition(newTo) if t.Meta().Partition.DDLState == model.StateDeleteOnly { err = tbl.RemoveRecord(ctx, h, currData) } else { - err = tbl.UpdateRecord(gctx, ctx, h, currData, newData, touched) + err = tbl.updateRecord(ctx, h, currData, newData, touched, opt) } if err != nil { return errors.Trace(err) @@ -1842,14 +1848,14 @@ func partitionedTableUpdateRecord(gctx context.Context, ctx table.MutateContext, memBuffer.Release(sh) return nil } - tbl = t.GetPartition(newFrom) + tbl = t.getPartition(newFrom) err = tbl.RemoveRecord(ctx, h, currData) if err != nil { return errors.Trace(err) } if t.Meta().GetPartitionInfo().DDLState != model.StateDeleteOnly { - tbl = t.GetPartition(newTo) - _, err = tbl.AddRecord(ctx, newData) + tbl = t.getPartition(newTo) + _, err = tbl.addRecord(ctx, newData, opt.GetAddRecordOpt()) if err != nil { return errors.Trace(err) } diff --git a/pkg/table/tables/tables.go b/pkg/table/tables/tables.go index 83f58f727c844..5b29b08a773cd 100644 --- a/pkg/table/tables/tables.go +++ b/pkg/table/tables/tables.go @@ -49,6 +49,7 @@ import ( "github.com/pingcap/tidb/pkg/util/codec" "github.com/pingcap/tidb/pkg/util/collate" "github.com/pingcap/tidb/pkg/util/generatedexpr" + "github.com/pingcap/tidb/pkg/util/intest" "github.com/pingcap/tidb/pkg/util/logutil" "github.com/pingcap/tidb/pkg/util/stringutil" "github.com/pingcap/tidb/pkg/util/tableutil" @@ -272,11 +273,25 @@ func initTableIndices(t *TableCommon) error { // Use partition ID for index, because TableCommon may be table or partition. idx := NewIndex(t.physicalTableID, tblInfo, idxInfo) + intest.AssertFunc(func() bool { + // `TableCommon.indices` is type of `[]table.Index` to implement interface method `Table.Indices`. + // However, we have an assumption that the specific type of each element in it should always be `*index`. + // We have this assumption because some codes access the inner method of `*index`, + // and they use `asIndex` to cast `table.Index` to `*index`. + _, ok := idx.(*index) + intest.Assert(ok, "index should be type of `*index`") + return true + }) t.indices = append(t.indices, idx) } return nil } +// asIndex casts a table.Index to *index which is the actual type of index in TableCommon. +func asIndex(idx table.Index) *index { + return idx.(*index) +} + func initTableCommonWithIndices(t *TableCommon, tblInfo *model.TableInfo, physicalTableID int64, cols []*table.Column, allocs autoid.Allocators, constraints []*table.Constraint) error { initTableCommon(t, tblInfo, physicalTableID, cols, allocs, constraints) return initTableIndices(t) @@ -425,7 +440,12 @@ func (t *TableCommon) shouldAssert(level variable.AssertionLevel) bool { // UpdateRecord implements table.Table UpdateRecord interface. // `touched` means which columns are really modified, used for secondary indices. // Length of `oldData` and `newData` equals to length of `t.WritableCols()`. -func (t *TableCommon) UpdateRecord(ctx context.Context, sctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool) error { +func (t *TableCommon) UpdateRecord(ctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool, opts ...table.UpdateRecordOption) error { + opt := table.NewUpdateRecordOpt(opts...) + return t.updateRecord(ctx, h, oldData, newData, touched, opt) +} + +func (t *TableCommon) updateRecord(sctx table.MutateContext, h kv.Handle, oldData, newData []types.Datum, touched []bool, opt *table.UpdateRecordOpt) error { txn, err := sctx.Txn(true) if err != nil { return err @@ -512,7 +532,7 @@ func (t *TableCommon) UpdateRecord(ctx context.Context, sctx table.MutateContext } } // rebuild index - err = t.rebuildIndices(sctx, txn, h, touched, oldData, newData, table.WithCtx(ctx)) + err = t.rebuildIndices(sctx, txn, h, touched, oldData, newData, opt.GetCreateIdxOpt()) if err != nil { return err } @@ -587,7 +607,7 @@ func (t *TableCommon) UpdateRecord(ctx context.Context, sctx table.MutateContext return nil } -func (t *TableCommon) rebuildIndices(ctx table.MutateContext, txn kv.Transaction, h kv.Handle, touched []bool, oldData []types.Datum, newData []types.Datum, opts ...table.CreateIdxOptFunc) error { +func (t *TableCommon) rebuildIndices(ctx table.MutateContext, txn kv.Transaction, h kv.Handle, touched []bool, oldData []types.Datum, newData []types.Datum, opt *table.CreateIdxOpt) error { for _, idx := range t.deletableIndices() { if t.meta.IsCommonHandle && idx.Meta().Primary { continue @@ -634,7 +654,7 @@ func (t *TableCommon) rebuildIndices(ctx table.MutateContext, txn kv.Transaction if err != nil { return err } - if err := t.buildIndexForRow(ctx, h, newVs, newData, idx, txn, untouched, opts...); err != nil { + if err := t.buildIndexForRow(ctx, h, newVs, newData, asIndex(idx), txn, untouched, opt); err != nil { return err } } @@ -653,22 +673,6 @@ func FindPrimaryIndex(tblInfo *model.TableInfo) *model.IndexInfo { return pkIdx } -// CommonAddRecordCtx is used in `AddRecord` to avoid memory malloc for some temp slices. -// This is useful in lightning parse row data to key-values pairs. This can gain upto 5% performance -// improvement in lightning's local mode. -type CommonAddRecordCtx struct { - colIDs []int64 - row []types.Datum -} - -// commonAddRecordKey is used as key in `sessionctx.Context.Value(key)` -type commonAddRecordKey struct{} - -// String implement `stringer.String` for CommonAddRecordKey -func (c commonAddRecordKey) String() string { - return "_common_add_record_context_key" -} - // TryGetCommonPkColumnIds get the IDs of primary key column if the table has common handle. func TryGetCommonPkColumnIds(tbl *model.TableInfo) []int64 { if !tbl.IsCommonHandle { @@ -734,17 +738,16 @@ func checkTempTableSize(tmpTable tbctx.TemporaryTableHandler, sizeLimit int64) e // AddRecord implements table.Table AddRecord interface. func (t *TableCommon) AddRecord(sctx table.MutateContext, r []types.Datum, opts ...table.AddRecordOption) (recordID kv.Handle, err error) { + // TODO: optimize the allocation (and calculation) of opt. + opt := table.NewAddRecordOpt(opts...) + return t.addRecord(sctx, r, opt) +} + +func (t *TableCommon) addRecord(sctx table.MutateContext, r []types.Datum, opt *table.AddRecordOpt) (recordID kv.Handle, err error) { txn, err := sctx.Txn(true) if err != nil { return nil, err } - - // TODO: optimize the allocation (and calculation) of opt. - var opt table.AddRecordOpt - for _, fn := range opts { - fn.ApplyOn(&opt) - } - if m := t.Meta(); m.TempTableType != model.TempTableNone { if tmpTable, sizeLimit, ok := addTemporaryTable(sctx, m); ok { if err = checkTempTableSize(tmpTable, sizeLimit); err != nil { @@ -948,17 +951,8 @@ func (t *TableCommon) AddRecord(sctx table.MutateContext, r []types.Datum, opts return nil, err } - var createIdxOpts []table.CreateIdxOptFunc - if len(opts) > 0 { - createIdxOpts = make([]table.CreateIdxOptFunc, 0, len(opts)) - for _, fn := range opts { - if raw, ok := fn.(table.CreateIdxOptFunc); ok { - createIdxOpts = append(createIdxOpts, raw) - } - } - } // Insert new entries into indices. - h, err := t.addIndices(sctx, recordID, r, txn, createIdxOpts) + h, err := t.addIndices(sctx, recordID, r, txn, opt.GetCreateIdxOpt()) if err != nil { return h, err } @@ -1016,7 +1010,7 @@ func genIndexKeyStrs(colVals []types.Datum) ([]string, error) { } // addIndices adds data into indices. If any key is duplicated, returns the original handle. -func (t *TableCommon) addIndices(sctx table.MutateContext, recordID kv.Handle, r []types.Datum, txn kv.Transaction, opts []table.CreateIdxOptFunc) (kv.Handle, error) { +func (t *TableCommon) addIndices(sctx table.MutateContext, recordID kv.Handle, r []types.Datum, txn kv.Transaction, opt *table.CreateIdxOpt) (kv.Handle, error) { writeBufs := sctx.GetMutateBuffers().GetWriteStmtBufs() indexVals := writeBufs.IndexValsBuf skipCheck := sctx.GetSessionVars().StmtCtx.BatchCheck @@ -1046,7 +1040,7 @@ func (t *TableCommon) addIndices(sctx table.MutateContext, recordID kv.Handle, r dupErr = kv.GenKeyExistsErr(colStrVals, fmt.Sprintf("%s.%s", v.TableMeta().Name.String(), v.Meta().Name.String())) } rsData := TryGetHandleRestoredDataWrapper(t.meta, r, nil, v.Meta()) - if dupHandle, err := v.Create(sctx, txn, indexVals, recordID, rsData, opts...); err != nil { + if dupHandle, err := asIndex(v).create(sctx, txn, indexVals, recordID, rsData, false, opt); err != nil { if kv.ErrKeyExists.Equal(err) { return dupHandle, dupErr } @@ -1427,14 +1421,9 @@ func (t *TableCommon) removeRowIndex(ctx table.MutateContext, h kv.Handle, vals } // buildIndexForRow implements table.Table BuildIndexForRow interface. -func (t *TableCommon) buildIndexForRow(ctx table.MutateContext, h kv.Handle, vals []types.Datum, newData []types.Datum, idx table.Index, txn kv.Transaction, untouched bool, popts ...table.CreateIdxOptFunc) error { - var opts []table.CreateIdxOptFunc - opts = append(opts, popts...) - if untouched { - opts = append(opts, table.IndexIsUntouched) - } +func (t *TableCommon) buildIndexForRow(ctx table.MutateContext, h kv.Handle, vals []types.Datum, newData []types.Datum, idx *index, txn kv.Transaction, untouched bool, opt *table.CreateIdxOpt) error { rsData := TryGetHandleRestoredDataWrapper(t.meta, newData, nil, idx.Meta()) - if _, err := idx.Create(ctx, txn, vals, h, rsData, opts...); err != nil { + if _, err := idx.create(ctx, txn, vals, h, rsData, untouched, opt); err != nil { if kv.ErrKeyExists.Equal(err) { // Make error message consistent with MySQL. tablecodec.TruncateIndexValues(t.meta, idx.Meta(), vals) diff --git a/pkg/table/tables/tables_test.go b/pkg/table/tables/tables_test.go index 78151ac9d3628..9c9801d35921d 100644 --- a/pkg/table/tables/tables_test.go +++ b/pkg/table/tables/tables_test.go @@ -111,7 +111,7 @@ func TestBasic(t *testing.T) { _, err = tb.AddRecord(ctx.GetTableCtx(), types.MakeDatums(2, "abc")) require.Error(t, err) - require.Nil(t, tb.UpdateRecord(context.Background(), ctx.GetTableCtx(), rid, types.MakeDatums(1, "abc"), types.MakeDatums(1, "cba"), []bool{false, true})) + require.Nil(t, tb.UpdateRecord(ctx.GetTableCtx(), rid, types.MakeDatums(1, "abc"), types.MakeDatums(1, "cba"), []bool{false, true})) err = tables.IterRecords(tb, ctx, tb.Cols(), func(_ kv.Handle, data []types.Datum, cols []*table.Column) (bool, error) { return true, nil