From 67f6669207939235c2d5f10d019d8948eeeb4f04 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 8 Jul 2026 15:38:29 +0800 Subject: [PATCH 01/12] Move appendable object creation outside txn --- pkg/objectio/object_stats.go | 8 ++ pkg/vm/engine/tae/catalog/object.go | 28 ++++++ pkg/vm/engine/tae/catalog/object_list.go | 34 +++++++ pkg/vm/engine/tae/catalog/table.go | 26 ++++++ pkg/vm/engine/tae/db/replay.go | 52 ++++++++++- pkg/vm/engine/tae/db/test/db_test.go | 6 +- pkg/vm/engine/tae/db/test/replay_test.go | 3 +- pkg/vm/engine/tae/rpc/apply_table_data.go | 18 ++-- pkg/vm/engine/tae/rpc/inspect_test.go | 6 +- pkg/vm/engine/tae/tables/updates/append.go | 17 +++- pkg/vm/engine/tae/tables/updates/mvcc.go | 20 +++++ pkg/vm/engine/tae/txn/txnimpl/replaystore.go | 93 +++++++++++++------- pkg/vm/engine/tae/txn/txnimpl/table.go | 37 +++++++- pkg/vm/engine/tae/txn/txnimpl/txn_test.go | 12 ++- 14 files changed, 297 insertions(+), 63 deletions(-) diff --git a/pkg/objectio/object_stats.go b/pkg/objectio/object_stats.go index 04fed88ec9f6e..05836625c1fcb 100644 --- a/pkg/objectio/object_stats.go +++ b/pkg/objectio/object_stats.go @@ -78,6 +78,14 @@ func WithAppendable() ObjectStatsOptions { } } +func SetObjectStatsAppendable(stats *ObjectStats, appendable bool) { + if appendable { + stats[reservedOffset] |= ObjectFlag_Appendable + return + } + stats[reservedOffset] &^= ObjectFlag_Appendable +} + func NewObjectStats() *ObjectStats { return new(ObjectStats) } diff --git a/pkg/vm/engine/tae/catalog/object.go b/pkg/vm/engine/tae/catalog/object.go index be8e033ffe114..18563c124e544 100644 --- a/pkg/vm/engine/tae/catalog/object.go +++ b/pkg/vm/engine/tae/catalog/object.go @@ -386,6 +386,34 @@ func NewObjectEntry( return e } +func NewCommittedObjectEntry( + table *TableEntry, + ts types.TS, + stats objectio.ObjectStats, + dataFactory ObjectDataFactory, + isTombstone bool, +) *ObjectEntry { + e := &ObjectEntry{ + table: table, + ObjectNode: ObjectNode{ + SortHint: table.GetDB().catalog.NextObject(), + IsTombstone: isTombstone, + }, + EntryMVCCNode: EntryMVCCNode{ + CreatedAt: ts, + }, + CreateNode: txnbase.NewTxnMVCCNodeWithTS(ts), + ObjectState: ObjectState_Create_ApplyCommit, + ObjectMVCCNode: ObjectMVCCNode{ + ObjectStats: stats, + }, + } + if dataFactory != nil { + e.objData = dataFactory(e) + } + return e +} + func NewReplayObjectEntry() *ObjectEntry { e := &ObjectEntry{} return e diff --git a/pkg/vm/engine/tae/catalog/object_list.go b/pkg/vm/engine/tae/catalog/object_list.go index 62f1d55e3df9c..e2dd0d006def5 100644 --- a/pkg/vm/engine/tae/catalog/object_list.go +++ b/pkg/vm/engine/tae/catalog/object_list.go @@ -25,6 +25,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/objectio" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/txnif" + "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/txn/txnbase" "github.com/tidwall/btree" "go.uber.org/zap" ) @@ -314,6 +315,39 @@ func (l *ObjectList) DeleteAllEntries(id *objectio.ObjectId) error { return nil } +func (l *ObjectList) UpdateCreateTS(id *objectio.ObjectId, ts types.TS) (*ObjectEntry, error) { + l.Lock() + defer l.Unlock() + oldTS, ok := l.maxTs_objectID[*id] + if !ok { + return nil, moerr.GetOkExpectedEOB() + } + oldTree := l.tree.Load() + newTree := oldTree.Copy() + nodes := l.getNodesSnap(newTree, oldTS, id, true) + if len(nodes) == 0 { + return nil, moerr.GetOkExpectedEOB() + } + oldNode := nodes[0] + if oldNode.IsDEntry() { + return oldNode, nil + } + newNode := oldNode.Clone() + newNode.CreatedAt = ts + newNode.CreateNode = txnbase.NewTxnMVCCNodeWithTS(ts) + newTree.Delete(oldNode) + newTree.Set(newNode) + maxTS := newNode.CreatedAt + if maxTS.LT(&newNode.DeletedAt) { + maxTS = newNode.DeletedAt + } + l.maxTs_objectID[*id] = maxTS + if !l.tree.CompareAndSwap(oldTree, newTree) { + panic("concurrent mutation") + } + return newNode, nil +} + // WaitUntilCommitted waits for entries in txn active zone with prepareTS > ts to move to committed zone. // As CreateObject will be called in txn queue, when WaitUntilCommitted returns, all creating objects in txn active zone are invisible to ts, because they are created after ts. func (l *ObjectList) WaitUntilCommitted(ts types.TS) { diff --git a/pkg/vm/engine/tae/catalog/table.go b/pkg/vm/engine/tae/catalog/table.go index 8deee0aa3abfd..f43d0357ef894 100644 --- a/pkg/vm/engine/tae/catalog/table.go +++ b/pkg/vm/engine/tae/catalog/table.go @@ -210,6 +210,17 @@ func (entry *TableEntry) getTombstoneObjectByID(id *types.Objectid) (obj *Object return entry.tombstoneObjects.GetObjectByID(id) } +func (entry *TableEntry) UpdateObjectCreateTS(id *types.Objectid, isTombstone bool, ts types.TS) error { + obj, err := entry.getObjectList(isTombstone).UpdateCreateTS(id, ts) + if err != nil { + return err + } + if obj.GetObjectData() != nil { + obj.GetObjectData().UpdateMeta(obj) + } + return nil +} + func (entry *TableEntry) MakeTombstoneObjectIt() btree.IterG[*ObjectEntry] { return entry.tombstoneObjects.tree.Load().Iter() } @@ -295,6 +306,21 @@ func (entry *TableEntry) CreateObject( return } +func (entry *TableEntry) CreateCommittedObject( + ts types.TS, + opts *objectio.CreateObjOpt, + dataFactory ObjectDataFactory, +) (created *ObjectEntry, err error) { + entry.Lock() + defer entry.Unlock() + created = NewCommittedObjectEntry(entry, ts, *opts.Stats, dataFactory, opts.IsTombstone) + if entry.GetCatalog().mergeNotifier != nil && !opts.Stats.GetAppendable() { + entry.GetCatalog().mergeNotifier.OnCreateNonAppendObject(ToMergeTable(entry)) + } + entry.AddEntryLocked(created) + return +} + func (entry *TableEntry) MakeCommand(id uint32) (cmd txnif.TxnCmd, err error) { cmdType := IOET_WALTxnCommand_Table entry.RLock() diff --git a/pkg/vm/engine/tae/db/replay.go b/pkg/vm/engine/tae/db/replay.go index 82fd70fe530f0..ba7c1f1e98db2 100644 --- a/pkg/vm/engine/tae/db/replay.go +++ b/pkg/vm/engine/tae/db/replay.go @@ -30,6 +30,7 @@ import ( "sync" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog" + "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/txnif" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/driver" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/wal" @@ -144,6 +145,15 @@ type WalReplayer struct { maxLSN atomic.Uint64 lsn uint64 + + replayAObjectCreates map[replayAObjectCreateKey]types.TS +} + +type replayAObjectCreateKey struct { + dbID uint64 + tableID uint64 + objectID objectio.ObjectId + isTombstone bool } func newWalReplayer( @@ -152,9 +162,10 @@ func newWalReplayer( lsn uint64, ) *WalReplayer { replayer := &WalReplayer{ - db: db, - fromTS: fromTS, - lsn: lsn, + db: db, + fromTS: fromTS, + lsn: lsn, + replayAObjectCreates: make(map[replayAObjectCreateKey]types.TS), } replayer.OnTimeStamp(fromTS) return replayer @@ -184,6 +195,28 @@ func (replayer *WalReplayer) PreReplayWal() { } func (replayer *WalReplayer) postReplayWal() error { + for key, createTS := range replayer.replayAObjectCreates { + dbEntry, err := replayer.db.Catalog.GetDatabaseByID(key.dbID) + if err != nil { + if moerr.IsMoErrCode(err, moerr.OkExpectedEOB) { + continue + } + return err + } + tableEntry, err := dbEntry.GetTableEntryByID(key.tableID) + if err != nil { + if moerr.IsMoErrCode(err, moerr.OkExpectedEOB) { + continue + } + return err + } + if err = tableEntry.UpdateObjectCreateTS(&key.objectID, key.isTombstone, createTS); err != nil { + if moerr.IsMoErrCode(err, moerr.OkExpectedEOB) { + continue + } + return err + } + } processor := new(catalog.LoopProcessor) processor.ObjectFn = func(entry *catalog.ObjectEntry) (err error) { if skippedTbl[entry.GetTable().ID] { @@ -197,6 +230,19 @@ func (replayer *WalReplayer) postReplayWal() error { return replayer.db.Catalog.RecurLoop(processor) } +func (replayer *WalReplayer) RecordReplayAObjectCreate(id *common.ID, isTombstone bool, ts types.TS) { + key := replayAObjectCreateKey{ + dbID: id.DbID, + tableID: id.TableID, + objectID: *id.ObjectID(), + isTombstone: isTombstone, + } + old, ok := replayer.replayAObjectCreates[key] + if !ok || ts.LT(&old) { + replayer.replayAObjectCreates[key] = ts + } +} + func (replayer *WalReplayer) Schedule( ctx context.Context, mode driver.ReplayMode, diff --git a/pkg/vm/engine/tae/db/test/db_test.go b/pkg/vm/engine/tae/db/test/db_test.go index 225751c0c20c5..b6966226abeed 100644 --- a/pkg/vm/engine/tae/db/test/db_test.go +++ b/pkg/vm/engine/tae/db/test/db_test.go @@ -1553,13 +1553,13 @@ func TestRollback1(t *testing.T) { tableMeta := rel.GetMeta().(*catalog.TableEntry) err = tableMeta.RecurLoop(processor) assert.Nil(t, err) - assert.Equal(t, objCnt, 1) + assert.Equal(t, 1, objCnt) assert.Nil(t, txn.Rollback(context.Background())) objCnt = 0 err = tableMeta.RecurLoop(processor) assert.Nil(t, err) - assert.Equal(t, objCnt, 0) + assert.Equal(t, 1, objCnt) txn, rel = testutil.GetDefaultRelation(t, db, schema.Name) obj, err := rel.CreateObject(false) @@ -1569,7 +1569,7 @@ func TestRollback1(t *testing.T) { objCnt = 0 err = tableMeta.RecurLoop(processor) assert.Nil(t, err) - assert.Equal(t, objCnt, 1) + assert.Equal(t, 2, objCnt) txn, rel = testutil.GetDefaultRelation(t, db, schema.Name) _, err = rel.GetObject(objMeta.ID(), false) diff --git a/pkg/vm/engine/tae/db/test/replay_test.go b/pkg/vm/engine/tae/db/test/replay_test.go index 404b666f26273..2fc2ff9d619a6 100644 --- a/pkg/vm/engine/tae/db/test/replay_test.go +++ b/pkg/vm/engine/tae/db/test/replay_test.go @@ -255,8 +255,9 @@ func TestReplayCatalog3(t *testing.T) { assert.Nil(t, err) rel, err = e.GetRelationByName(schema.Name) assert.Nil(t, err) - obj, err = rel.CreateObject(false) + obj, err = rel.CreateNonAppendableObject(false, nil) assert.Nil(t, err) + testutil.MockObjectStats(t, obj) assert.Nil(t, txn.Commit(context.Background())) txn, _ = tae.StartTxn(nil) diff --git a/pkg/vm/engine/tae/rpc/apply_table_data.go b/pkg/vm/engine/tae/rpc/apply_table_data.go index a915f1b0782b1..475fe669dd6e7 100644 --- a/pkg/vm/engine/tae/rpc/apply_table_data.go +++ b/pkg/vm/engine/tae/rpc/apply_table_data.go @@ -211,18 +211,19 @@ func (a *ApplyTableDataArg) Run() (err error) { panic(fmt.Sprintf("invalid object type: %d", objTypes[i])) } stats := objectio.ObjectStats(idVec.GetBytesAt(i)) + objectio.SetObjectStatsAppendable(&stats, false) + opt := &objectio.CreateObjOpt{ + Stats: &stats, + IsTombstone: isTombstone, + } + tableEntry := a.rel.GetMeta().(*catalog.TableEntry) var obj handle.Object - if obj, err = a.rel.CreateNonAppendableObject( - isTombstone, - &objectio.CreateObjOpt{ - Stats: &stats, - IsTombstone: isTombstone, - }, - ); err != nil { + if obj, err = a.rel.CreateNonAppendableObject(isTombstone, opt); err != nil { return } + meta := obj.GetMeta().(*catalog.ObjectEntry) - schema := a.rel.GetMeta().(*catalog.TableEntry).GetLastestSchema(isTombstone) + schema := tableEntry.GetLastestSchema(isTombstone) attrs := schema.AllNames() attrs = append(attrs, objectio.TombstoneAttr_CommitTs_Attr) @@ -235,7 +236,6 @@ func (a *ApplyTableDataArg) Run() (err error) { } defer objectRelease() tnBat := containers.ToTNBatch(bat, a.mp) - meta := obj.GetMeta().(*catalog.ObjectEntry) var anodes []txnif.TxnEntry if anodes, err = meta.GetObjectData().ApplyDebugBatch(tnBat, a.txn); err != nil { return diff --git a/pkg/vm/engine/tae/rpc/inspect_test.go b/pkg/vm/engine/tae/rpc/inspect_test.go index 24078ff7355dd..f0511e97e6735 100644 --- a/pkg/vm/engine/tae/rpc/inspect_test.go +++ b/pkg/vm/engine/tae/rpc/inspect_test.go @@ -75,11 +75,7 @@ func TestMergeCommand(t *testing.T) { vector := containers.NewVector(types.T_varchar.ToType()) { id := objectio.NewObjectid() - stats := objectio.NewObjectStatsWithObjectID(&id, true, true, false) - vector.Append(stats.Marshal(), false) - - id = objectio.NewObjectid() - stats = objectio.NewObjectStatsWithObjectID(&id, false, true, false) + stats := objectio.NewObjectStatsWithObjectID(&id, false, true, false) zm := index.NewZM(types.T_int32, 0) v1 := int32(1) v2 := int32(2) diff --git a/pkg/vm/engine/tae/tables/updates/append.go b/pkg/vm/engine/tae/tables/updates/append.go index 0c087ee3b85c5..45f344eefd5f6 100644 --- a/pkg/vm/engine/tae/tables/updates/append.go +++ b/pkg/vm/engine/tae/tables/updates/append.go @@ -133,16 +133,25 @@ func (node *AppendNode) PrepareCommit() error { func (node *AppendNode) ApplyCommit(id string) error { node.mvcc.Lock() - defer node.mvcc.Unlock() if node.IsCommitted() { + node.mvcc.Unlock() panic("AppendNode | ApplyCommit | LogicErr") } node.TxnMVCCNode.ApplyCommit(id) + minCommittedTS := node.mvcc.MinCommittedAppendTSLocked() listener := node.mvcc.GetAppendListener() - if listener == nil { - return nil + var err error + if listener != nil { + err = listener(node) + } + node.mvcc.Unlock() + if !minCommittedTS.IsEmpty() && node.mvcc.meta.IsAppendable() { + if updateErr := node.mvcc.meta.GetTable().UpdateObjectCreateTS( + node.mvcc.meta.ID(), node.mvcc.meta.IsTombstone, minCommittedTS); updateErr != nil { + return updateErr + } } - return listener(node) + return err } func (node *AppendNode) ApplyRollback() (err error) { diff --git a/pkg/vm/engine/tae/tables/updates/mvcc.go b/pkg/vm/engine/tae/tables/updates/mvcc.go index 3f0ecfd48c5b8..7d2fb1bf4bf73 100644 --- a/pkg/vm/engine/tae/tables/updates/mvcc.go +++ b/pkg/vm/engine/tae/tables/updates/mvcc.go @@ -187,6 +187,26 @@ func (n *AppendMVCCHandle) GetCommitTSVecInRange(start, end types.TS, mp *mpool. return commitTSVec } +func (n *AppendMVCCHandle) MinCommittedAppendTSLocked() types.TS { + var minTS types.TS + n.appends.ForEach( + func(node *AppendNode) bool { + if !node.IsCommitted() || node.IsAborted() { + return true + } + commitTS := node.GetCommitTS() + if commitTS.IsEmpty() || commitTS.Equal(&txnif.UncommitTS) { + return true + } + if minTS.IsEmpty() || commitTS.LT(&minTS) { + minTS = commitTS + } + return true + }, + true) + return minTS +} + // it is used to get the visible max row for a txn // maxrow: is the max row that the txn can see // visible: is true if the txn can see any row diff --git a/pkg/vm/engine/tae/txn/txnimpl/replaystore.go b/pkg/vm/engine/tae/txn/txnimpl/replaystore.go index ae6f5815111da..31bc889cb4873 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/replaystore.go +++ b/pkg/vm/engine/tae/txn/txnimpl/replaystore.go @@ -18,7 +18,9 @@ import ( "context" "github.com/matrixorigin/matrixone/pkg/common/moerr" + "github.com/matrixorigin/matrixone/pkg/container/types" "github.com/matrixorigin/matrixone/pkg/logutil" + "github.com/matrixorigin/matrixone/pkg/objectio" "github.com/matrixorigin/matrixone/pkg/util/fault" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common" @@ -30,6 +32,10 @@ import ( var ErrDebugReplay = moerr.NewInternalErrorNoCtx("debug") +type replayAObjectCreateObserver interface { + RecordReplayAObjectCreate(id *common.ID, isTombstone bool, ts types.TS) +} + type replayTxnStore struct { txnbase.NoopTxnStore Cmd *txnbase.TxnCmd @@ -128,17 +134,7 @@ func (store *replayTxnStore) replayAppendData(cmd *AppendCmd, observer wal.Repla _, sarg, _ := fault.TriggerFault("replay debug log") for _, info := range cmd.Infos { id := info.GetDest() - database, err := store.catalog.GetDatabaseByID(id.DbID) - if sarg != "" { - err = ErrDebugReplay - } - if err != nil { - logutil.Infof("cmd %v\ncatalog: %v", cmd.String(), store.catalog.SimplePPString(3)) - if err != ErrDebugReplay { - panic(err) - } - } - blk, err := database.GetObjectEntryByID(id, cmd.IsTombstone) + blk, _, err := store.ensureReplayAObject(id, cmd.IsTombstone, cmd.Ts, observer) if sarg != "" { err = ErrDebugReplay } @@ -168,17 +164,7 @@ func (store *replayTxnStore) replayAppendData(cmd *AppendCmd, observer wal.Repla for _, info := range cmd.Infos { id := info.GetDest() - database, err := store.catalog.GetDatabaseByID(id.DbID) - if sarg != "" { - err = ErrDebugReplay - } - if err != nil { - logutil.Infof("cmd %v\ncatalog: %v", cmd.String(), store.catalog.SimplePPString(3)) - if err != ErrDebugReplay { - panic(err) - } - } - blk, err := database.GetObjectEntryByID(id, cmd.IsTombstone) + blk, _, err := store.ensureReplayAObject(id, cmd.IsTombstone, cmd.Ts, observer) if sarg != "" { err = ErrDebugReplay } @@ -219,18 +205,8 @@ func (store *replayTxnStore) replayDataCmds(cmd *updates.UpdateCmd, observer wal func (store *replayTxnStore) replayAppend(cmd *updates.UpdateCmd, observer wal.ReplayObserver) { appendNode := cmd.GetAppendNode() id := appendNode.GetID() - database, err := store.catalog.GetDatabaseByID(id.DbID) _, sarg, _ := fault.TriggerFault("replay debug log") - if sarg != "" { - err = ErrDebugReplay - } - if err != nil { - logutil.Infof("cmd %v\ncatalog: %v", cmd.String(), store.catalog.SimplePPString(3)) - if err != ErrDebugReplay { - panic(err) - } - } - obj, err := database.GetObjectEntryByID(id, cmd.GetAppendNode().IsTombstone()) + obj, _, err := store.ensureReplayAObject(id, appendNode.IsTombstone(), replayAppendNodeCreateTS(appendNode), observer) if sarg != "" { err = ErrDebugReplay } @@ -253,3 +229,54 @@ func (store *replayTxnStore) replayAppend(cmd *updates.UpdateCmd, observer wal.R } } } + +func replayAppendNodeCreateTS(node *updates.AppendNode) types.TS { + if txn := node.GetTxn(); txn != nil { + return txn.GetCommitTS() + } + if ts := node.GetCommitTS(); !ts.IsEmpty() { + return ts + } + return node.GetPrepare() +} + +func (store *replayTxnStore) ensureReplayAObject( + id *common.ID, + isTombstone bool, + createHint types.TS, + observer wal.ReplayObserver, +) (obj *catalog.ObjectEntry, created bool, err error) { + database, err := store.catalog.GetDatabaseByID(id.DbID) + if err != nil { + return + } + obj, err = database.GetObjectEntryByID(id, isTombstone) + if err == nil { + return + } + if !moerr.IsMoErrCode(err, moerr.OkExpectedEOB) { + return + } + table, err := database.GetTableEntryByID(id.TableID) + if err != nil { + return + } + stats := objectio.NewObjectStatsWithObjectID(id.ObjectID(), true, isTombstone, false) + var factory catalog.ObjectDataFactory + if store.catalog.DataFactory != nil { + factory = store.catalog.DataFactory.MakeObjectFactory() + } + obj, err = table.CreateCommittedObject( + createHint, + &objectio.CreateObjOpt{Stats: stats, IsTombstone: isTombstone}, + factory, + ) + if err != nil { + return + } + created = true + if recorder, ok := observer.(replayAObjectCreateObserver); ok { + recorder.RecordReplayAObjectCreate(id, isTombstone, createHint) + } + return +} diff --git a/pkg/vm/engine/tae/txn/txnimpl/table.go b/pkg/vm/engine/tae/txn/txnimpl/table.go index c8642ed032085..290428324863b 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/table.go +++ b/pkg/vm/engine/tae/txn/txnimpl/table.go @@ -48,6 +48,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/index/indexwrapper" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/wal" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/model" + "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/txn/txnbase" "go.uber.org/zap" ) @@ -715,7 +716,7 @@ func (tbl *txnTable) CreateObject(isTombstone bool) (obj handle.Object, err erro sorted, false, ) - return tbl.createObject( + return tbl.createCommittedAppendableObject( &objectio.CreateObjOpt{Stats: stats, IsTombstone: isTombstone}, ) } @@ -724,6 +725,26 @@ func (tbl *txnTable) CreateNonAppendableObject(opts *objectio.CreateObjOpt) (obj return tbl.createObject(opts) } +func (tbl *txnTable) createCommittedAppendableObject(opts *objectio.CreateObjOpt) (obj handle.Object, err error) { + if !opts.Stats.GetAppendable() { + return nil, moerr.NewInternalErrorNoCtx("CreateObject outside txn only supports appendable object") + } + baseTxn, ok := tbl.store.txn.GetBase().(*txnbase.Txn) + if !ok || baseTxn.Mgr == nil { + return nil, moerr.NewInternalErrorNoCtx("missing txn manager for appendable object create") + } + var factory catalog.ObjectDataFactory + if tbl.store.catalog.DataFactory != nil { + factory = tbl.store.catalog.DataFactory.MakeObjectFactory() + } + var meta *catalog.ObjectEntry + if meta, err = tbl.entry.CreateCommittedObject(baseTxn.Mgr.Now(), opts, factory); err != nil { + return + } + obj = newObject(tbl, meta) + return +} + func (tbl *txnTable) createObject(opts *objectio.CreateObjOpt) (obj handle.Object, err error) { var factory catalog.ObjectDataFactory if tbl.store.catalog.DataFactory != nil { @@ -1448,6 +1469,9 @@ func (tbl *txnTable) PrepareCommit() (err error) { if tbl.txnEntries.IsDeleted(idx) { continue } + if isAppendableObjectCreateEntry(node) { + panic("appendable object create entry must not be committed through txn") + } if err = node.PrepareCommit(); err != nil { if moerr.IsMoErrCode(err, moerr.ErrTxnNotFound) { var buf bytes.Buffer @@ -1476,6 +1500,9 @@ func (tbl *txnTable) PrepareCommit() (err error) { if tbl.txnEntries.IsDeleted(idx) { continue } + if isAppendableObjectCreateEntry(tbl.txnEntries.entries[idx]) { + panic("appendable object create entry must not be committed through txn") + } if err = tbl.txnEntries.entries[idx].PrepareCommit(); err != nil { break } @@ -1494,6 +1521,9 @@ func (tbl *txnTable) ApplyCommit() (err error) { if tbl.txnEntries.IsDeleted(idx) { continue } + if isAppendableObjectCreateEntry(node) { + panic("appendable object create entry must not be committed through txn") + } if err = node.ApplyCommit(tbl.store.txn.GetID()); err != nil { if moerr.IsMoErrCode(err, moerr.ErrTxnNotFound) { var buf bytes.Buffer @@ -1534,6 +1564,11 @@ func (tbl *txnTable) ApplyCommit() (err error) { return } +func isAppendableObjectCreateEntry(entry txnif.TxnEntry) bool { + obj, ok := entry.(*catalog.ObjectEntry) + return ok && obj.ObjectState == catalog.ObjectState_Create_Active && obj.IsAppendable() +} + func (tbl *txnTable) ApplyRollback() (err error) { csn := tbl.csnStart for idx, node := range tbl.txnEntries.entries { diff --git a/pkg/vm/engine/tae/txn/txnimpl/txn_test.go b/pkg/vm/engine/tae/txn/txnimpl/txn_test.go index ee3d1cfe516a3..6582e5279137d 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/txn_test.go +++ b/pkg/vm/engine/tae/txn/txnimpl/txn_test.go @@ -783,7 +783,7 @@ func TestObject1(t *testing.T) { t.Log(iobj.String()) cnt++ } - assert.Equal(t, 2, cnt) + assert.Equal(t, 1, cnt) txn3, _ := mgr.StartTxn(nil) db, _ = txn3.GetDatabase(name) @@ -830,15 +830,19 @@ func TestObject2(t *testing.T) { assert.Nil(t, err) } + err := txn1.Commit(context.Background()) + assert.Nil(t, err) + + txn2, _ := mgr.StartTxn(nil) + db, _ = txn2.GetDatabase("db") + rel, _ = db.GetRelationByName(schema.Name) it := rel.MakeObjectIt(false) cnt := 0 for it.Next() { cnt++ - // iobj := it.GetObject() } assert.Equal(t, objCnt, cnt) - // err := txn1.Commit() - // assert.Nil(t, err) + assert.Nil(t, txn2.Commit(context.Background())) t.Log(c.SimplePPString(common.PPL1)) } From 1184e0d8197b7fe6b539e690016558e984a14581 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 8 Jul 2026 15:49:51 +0800 Subject: [PATCH 02/12] fix(tae): avoid updating aobject create ts on append commit --- pkg/vm/engine/tae/tables/updates/append.go | 7 ------- pkg/vm/engine/tae/tables/updates/mvcc.go | 20 -------------------- 2 files changed, 27 deletions(-) diff --git a/pkg/vm/engine/tae/tables/updates/append.go b/pkg/vm/engine/tae/tables/updates/append.go index 45f344eefd5f6..7a6ad5c445196 100644 --- a/pkg/vm/engine/tae/tables/updates/append.go +++ b/pkg/vm/engine/tae/tables/updates/append.go @@ -138,19 +138,12 @@ func (node *AppendNode) ApplyCommit(id string) error { panic("AppendNode | ApplyCommit | LogicErr") } node.TxnMVCCNode.ApplyCommit(id) - minCommittedTS := node.mvcc.MinCommittedAppendTSLocked() listener := node.mvcc.GetAppendListener() var err error if listener != nil { err = listener(node) } node.mvcc.Unlock() - if !minCommittedTS.IsEmpty() && node.mvcc.meta.IsAppendable() { - if updateErr := node.mvcc.meta.GetTable().UpdateObjectCreateTS( - node.mvcc.meta.ID(), node.mvcc.meta.IsTombstone, minCommittedTS); updateErr != nil { - return updateErr - } - } return err } diff --git a/pkg/vm/engine/tae/tables/updates/mvcc.go b/pkg/vm/engine/tae/tables/updates/mvcc.go index 7d2fb1bf4bf73..3f0ecfd48c5b8 100644 --- a/pkg/vm/engine/tae/tables/updates/mvcc.go +++ b/pkg/vm/engine/tae/tables/updates/mvcc.go @@ -187,26 +187,6 @@ func (n *AppendMVCCHandle) GetCommitTSVecInRange(start, end types.TS, mp *mpool. return commitTSVec } -func (n *AppendMVCCHandle) MinCommittedAppendTSLocked() types.TS { - var minTS types.TS - n.appends.ForEach( - func(node *AppendNode) bool { - if !node.IsCommitted() || node.IsAborted() { - return true - } - commitTS := node.GetCommitTS() - if commitTS.IsEmpty() || commitTS.Equal(&txnif.UncommitTS) { - return true - } - if minTS.IsEmpty() || commitTS.LT(&minTS) { - minTS = commitTS - } - return true - }, - true) - return minTS -} - // it is used to get the visible max row for a txn // maxrow: is the max row that the txn can see // visible: is true if the txn can see any row From 06e4754bd788d176a2e7f356c925a10700563d83 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 8 Jul 2026 16:06:20 +0800 Subject: [PATCH 03/12] fix(tae): preserve appendable stats in apply table data --- pkg/objectio/object_stats.go | 8 -------- pkg/vm/engine/tae/iface/handle/relation.go | 1 + pkg/vm/engine/tae/iface/txnif/types.go | 1 + pkg/vm/engine/tae/rpc/apply_table_data.go | 19 ++++++++++++++++--- pkg/vm/engine/tae/txn/txnbase/handle.go | 3 +++ pkg/vm/engine/tae/txn/txnbase/store.go | 3 +++ pkg/vm/engine/tae/txn/txnimpl/relation.go | 4 ++++ pkg/vm/engine/tae/txn/txnimpl/store.go | 11 +++++++++++ pkg/vm/engine/tae/txn/txnimpl/table.go | 4 ++++ pkg/vm/engine/tae/txn/txnimpl/txndb.go | 13 +++++++++++++ 10 files changed, 56 insertions(+), 11 deletions(-) diff --git a/pkg/objectio/object_stats.go b/pkg/objectio/object_stats.go index 05836625c1fcb..04fed88ec9f6e 100644 --- a/pkg/objectio/object_stats.go +++ b/pkg/objectio/object_stats.go @@ -78,14 +78,6 @@ func WithAppendable() ObjectStatsOptions { } } -func SetObjectStatsAppendable(stats *ObjectStats, appendable bool) { - if appendable { - stats[reservedOffset] |= ObjectFlag_Appendable - return - } - stats[reservedOffset] &^= ObjectFlag_Appendable -} - func NewObjectStats() *ObjectStats { return new(ObjectStats) } diff --git a/pkg/vm/engine/tae/iface/handle/relation.go b/pkg/vm/engine/tae/iface/handle/relation.go index 36fda9bad3420..50a735f20df3e 100644 --- a/pkg/vm/engine/tae/iface/handle/relation.go +++ b/pkg/vm/engine/tae/iface/handle/relation.go @@ -54,6 +54,7 @@ type Relation interface { GetMeta() any CreateObject(bool) (Object, error) + CreateObjectWithOpt(isTombstone bool, opt *objectio.CreateObjOpt) (Object, error) CreateNonAppendableObject(isTombstone bool, opt *objectio.CreateObjOpt) (Object, error) GetObject(id *types.Objectid, isTombstone bool) (Object, error) SoftDeleteObject(id *types.Objectid, isTombstone bool) (err error) diff --git a/pkg/vm/engine/tae/iface/txnif/types.go b/pkg/vm/engine/tae/iface/txnif/types.go index d2b79ac8c3c42..58c45746af341 100644 --- a/pkg/vm/engine/tae/iface/txnif/types.go +++ b/pkg/vm/engine/tae/iface/txnif/types.go @@ -305,6 +305,7 @@ type TxnStore interface { GetObject(id *common.ID, isTombstone bool) (handle.Object, error) CreateObject(dbId, tid uint64, isTombstone bool) (handle.Object, error) + CreateObjectWithOpt(dbId, tid uint64, isTombstone bool, opt *objectio.CreateObjOpt) (handle.Object, error) CreateNonAppendableObject(dbId, tid uint64, isTombstone bool, opt *objectio.CreateObjOpt) (handle.Object, error) SoftDeleteObject(isTombstone bool, id *common.ID) error diff --git a/pkg/vm/engine/tae/rpc/apply_table_data.go b/pkg/vm/engine/tae/rpc/apply_table_data.go index 475fe669dd6e7..fb4988ebce69a 100644 --- a/pkg/vm/engine/tae/rpc/apply_table_data.go +++ b/pkg/vm/engine/tae/rpc/apply_table_data.go @@ -211,15 +211,28 @@ func (a *ApplyTableDataArg) Run() (err error) { panic(fmt.Sprintf("invalid object type: %d", objTypes[i])) } stats := objectio.ObjectStats(idVec.GetBytesAt(i)) - objectio.SetObjectStatsAppendable(&stats, false) opt := &objectio.CreateObjOpt{ Stats: &stats, IsTombstone: isTombstone, } tableEntry := a.rel.GetMeta().(*catalog.TableEntry) var obj handle.Object - if obj, err = a.rel.CreateNonAppendableObject(isTombstone, opt); err != nil { - return + if stats.GetAppendable() { + if stats.Rows() > 0 { + err = moerr.NewInternalErrorNoCtxf( + "appendable object %s has non-empty row count %d", + stats.ObjectName().String(), + stats.Rows(), + ) + return + } + if obj, err = a.rel.CreateObjectWithOpt(isTombstone, opt); err != nil { + return + } + } else { + if obj, err = a.rel.CreateNonAppendableObject(isTombstone, opt); err != nil { + return + } } meta := obj.GetMeta().(*catalog.ObjectEntry) diff --git a/pkg/vm/engine/tae/txn/txnbase/handle.go b/pkg/vm/engine/tae/txn/txnbase/handle.go index d0df9f0de63cd..8292f3d138831 100644 --- a/pkg/vm/engine/tae/txn/txnbase/handle.go +++ b/pkg/vm/engine/tae/txn/txnbase/handle.go @@ -82,6 +82,9 @@ func (rel *TxnRelation) GetObject(id *types.Objectid, isTombstone bool) (obj han } func (rel *TxnRelation) SoftDeleteObject(id *types.Objectid, isTombstone bool) (err error) { return } func (rel *TxnRelation) CreateObject(bool) (obj handle.Object, err error) { return } +func (rel *TxnRelation) CreateObjectWithOpt(bool, *objectio.CreateObjOpt) (obj handle.Object, err error) { + return +} func (rel *TxnRelation) CreateNonAppendableObject(bool, *objectio.CreateObjOpt) (obj handle.Object, err error) { return } diff --git a/pkg/vm/engine/tae/txn/txnbase/store.go b/pkg/vm/engine/tae/txn/txnbase/store.go index cc75cfcd3a308..68be3a2f81e47 100644 --- a/pkg/vm/engine/tae/txn/txnbase/store.go +++ b/pkg/vm/engine/tae/txn/txnbase/store.go @@ -106,6 +106,9 @@ func (store *NoopTxnStore) GetObject(id *common.ID, isTombstone bool) (obj handl func (store *NoopTxnStore) CreateObject(dbId, tid uint64, isTombstone bool) (obj handle.Object, err error) { return } +func (store *NoopTxnStore) CreateObjectWithOpt(dbId, tid uint64, _ bool, _ *objectio.CreateObjOpt) (obj handle.Object, err error) { + return +} func (store *NoopTxnStore) CreateNonAppendableObject(dbId, tid uint64, _ bool, _ *objectio.CreateObjOpt) (obj handle.Object, err error) { return } diff --git a/pkg/vm/engine/tae/txn/txnimpl/relation.go b/pkg/vm/engine/tae/txn/txnimpl/relation.go index c7ff4d043e627..90e8bc9883fa8 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/relation.go +++ b/pkg/vm/engine/tae/txn/txnimpl/relation.go @@ -184,6 +184,10 @@ func (h *txnRelation) CreateObject(isTombstone bool) (obj handle.Object, err err return h.Txn.GetStore().CreateObject(h.table.entry.GetDB().ID, h.table.entry.GetID(), isTombstone) } +func (h *txnRelation) CreateObjectWithOpt(isTombstone bool, opt *objectio.CreateObjOpt) (obj handle.Object, err error) { + return h.Txn.GetStore().CreateObjectWithOpt(h.table.entry.GetDB().ID, h.table.entry.GetID(), isTombstone, opt) +} + func (h *txnRelation) CreateNonAppendableObject(isTombstone bool, opt *objectio.CreateObjOpt) (obj handle.Object, err error) { if opt == nil { noid := objectio.NewObjectid() diff --git a/pkg/vm/engine/tae/txn/txnimpl/store.go b/pkg/vm/engine/tae/txn/txnimpl/store.go index c28a7329d14fd..20d5c7c1f4816 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/store.go +++ b/pkg/vm/engine/tae/txn/txnimpl/store.go @@ -706,6 +706,17 @@ func (store *txnStore) CreateObject(dbId, tid uint64, isTombstone bool) (obj han return db.CreateObject(tid, isTombstone) } +func (store *txnStore) CreateObjectWithOpt(dbId, tid uint64, isTombstone bool, opt *objectio.CreateObjOpt) (obj handle.Object, err error) { + if err = store.WantWrite("CreateObjectWithOpt"); err != nil { + return + } + var db *txnDB + if db, err = store.getOrSetDB(dbId); err != nil { + return + } + return db.CreateObjectWithOpt(tid, opt, isTombstone) +} + func (store *txnStore) CreateNonAppendableObject(dbId, tid uint64, isTombstone bool, opt *objectio.CreateObjOpt) (obj handle.Object, err error) { if err = store.WantWrite("CreateNonAppendableObject"); err != nil { return diff --git a/pkg/vm/engine/tae/txn/txnimpl/table.go b/pkg/vm/engine/tae/txn/txnimpl/table.go index 290428324863b..ce0c28736f8ea 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/table.go +++ b/pkg/vm/engine/tae/txn/txnimpl/table.go @@ -721,6 +721,10 @@ func (tbl *txnTable) CreateObject(isTombstone bool) (obj handle.Object, err erro ) } +func (tbl *txnTable) CreateObjectWithOpt(opts *objectio.CreateObjOpt) (obj handle.Object, err error) { + return tbl.createCommittedAppendableObject(opts) +} + func (tbl *txnTable) CreateNonAppendableObject(opts *objectio.CreateObjOpt) (obj handle.Object, err error) { return tbl.createObject(opts) } diff --git a/pkg/vm/engine/tae/txn/txnimpl/txndb.go b/pkg/vm/engine/tae/txn/txnimpl/txndb.go index e986f3d8e321a..ae900f446ec33 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/txndb.go +++ b/pkg/vm/engine/tae/txn/txnimpl/txndb.go @@ -344,6 +344,19 @@ func (db *txnDB) CreateObject(tid uint64, isTombstone bool) (obj handle.Object, } return table.CreateObject(isTombstone) } + +func (db *txnDB) CreateObjectWithOpt(tid uint64, opt *objectio.CreateObjOpt, isTombstone bool) (obj handle.Object, err error) { + if err = db.store.WantWrite("CreateObjectWithOpt"); err != nil { + return + } + var table *txnTable + if table, err = db.getOrSetTable(tid); err != nil { + return + } + opt.WithIsTombstone(isTombstone) + return table.CreateObjectWithOpt(opt) +} + func (db *txnDB) CreateNonAppendableObject(tid uint64, opt *objectio.CreateObjOpt, isTombstone bool) (obj handle.Object, err error) { if err = db.store.WantWrite("CreateNonAppendableObject"); err != nil { return From e732b32f9e9e0bd05908f0cb884cec4f0b0907b2 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 8 Jul 2026 17:10:01 +0800 Subject: [PATCH 04/12] fix(tae): update replay aobject create ts for delete entries --- pkg/vm/engine/tae/catalog/object_list.go | 22 +++++++++---- pkg/vm/engine/tae/catalog/table_test.go | 38 ++++++++++++++++++++++ pkg/vm/engine/tae/tables/updates/append.go | 3 +- 3 files changed, 55 insertions(+), 8 deletions(-) diff --git a/pkg/vm/engine/tae/catalog/object_list.go b/pkg/vm/engine/tae/catalog/object_list.go index e2dd0d006def5..477484c2d2e84 100644 --- a/pkg/vm/engine/tae/catalog/object_list.go +++ b/pkg/vm/engine/tae/catalog/object_list.go @@ -329,14 +329,24 @@ func (l *ObjectList) UpdateCreateTS(id *objectio.ObjectId, ts types.TS) (*Object return nil, moerr.GetOkExpectedEOB() } oldNode := nodes[0] + newNode := oldNode.Clone() if oldNode.IsDEntry() { - return oldNode, nil + newPrev := oldNode.prevVersion.Clone() + newPrev.CreatedAt = ts + newPrev.CreateNode = txnbase.NewTxnMVCCNodeWithTS(ts) + newPrev.nextVersion = newNode + newNode.CreatedAt = ts + newNode.CreateNode = txnbase.NewTxnMVCCNodeWithTS(ts) + newNode.prevVersion = newPrev + newTree.Delete(oldNode) + newTree.Set(newNode) + newTree.Set(newPrev) + } else { + newNode.CreatedAt = ts + newNode.CreateNode = txnbase.NewTxnMVCCNodeWithTS(ts) + newTree.Delete(oldNode) + newTree.Set(newNode) } - newNode := oldNode.Clone() - newNode.CreatedAt = ts - newNode.CreateNode = txnbase.NewTxnMVCCNodeWithTS(ts) - newTree.Delete(oldNode) - newTree.Set(newNode) maxTS := newNode.CreatedAt if maxTS.LT(&newNode.DeletedAt) { maxTS = newNode.DeletedAt diff --git a/pkg/vm/engine/tae/catalog/table_test.go b/pkg/vm/engine/tae/catalog/table_test.go index 28c54bbfd5ce0..315cd0e4fe738 100644 --- a/pkg/vm/engine/tae/catalog/table_test.go +++ b/pkg/vm/engine/tae/catalog/table_test.go @@ -65,6 +65,44 @@ func TestObjectList(t *testing.T) { t.Log(ll.getNodes(entry1.ID(), false)) } +func TestObjectListUpdateCreateTSWithDeleteEntry(t *testing.T) { + ll := NewObjectList(false) + nobjid := objectio.NewObjectid() + createTS := types.BuildTS(10, 0) + deleteTS := types.BuildTS(20, 0) + updatedCreateTS := types.BuildTS(5, 0) + createEntry := &ObjectEntry{ + ObjectNode: ObjectNode{SortHint: 1}, + EntryMVCCNode: EntryMVCCNode{ + CreatedAt: createTS, + }, + ObjectMVCCNode: ObjectMVCCNode{ObjectStats: *objectio.NewObjectStatsWithObjectID(&nobjid, true, false, false)}, + CreateNode: txnbase.NewTxnMVCCNodeWithTS(createTS), + ObjectState: ObjectState_Create_ApplyCommit, + } + deleteEntry := createEntry.Clone() + deleteEntry.DeletedAt = deleteTS + deleteEntry.DeleteNode = txnbase.NewTxnMVCCNodeWithTS(deleteTS) + deleteEntry.ObjectState = ObjectState_Delete_ApplyCommit + updatedCreateEntry := createEntry.Clone() + updatedCreateEntry.nextVersion = deleteEntry + deleteEntry.prevVersion = updatedCreateEntry + + ll.modify(nil, deleteEntry, updatedCreateEntry) + updated, err := ll.UpdateCreateTS(createEntry.ID(), updatedCreateTS) + require.NoError(t, err) + require.True(t, updated.IsDEntry()) + + nodes := ll.GetAllNodes(createEntry.ID()) + require.Len(t, nodes, 2) + require.Equal(t, updatedCreateTS, nodes[0].CreatedAt) + require.Equal(t, updatedCreateTS, nodes[0].CreateNode.GetPrepare()) + require.Equal(t, updatedCreateTS, nodes[1].CreatedAt) + require.Equal(t, updatedCreateTS, nodes[1].CreateNode.GetPrepare()) + require.Same(t, nodes[0].prevVersion, nodes[1]) + require.Same(t, nodes[1].nextVersion, nodes[0]) +} + func TestGetSoftdeleteObjects(t *testing.T) { db := MockDBEntryWithAccInfo(0, 0) tbl := MockTableEntryWithDB(db, 1) diff --git a/pkg/vm/engine/tae/tables/updates/append.go b/pkg/vm/engine/tae/tables/updates/append.go index 7a6ad5c445196..6901fef455067 100644 --- a/pkg/vm/engine/tae/tables/updates/append.go +++ b/pkg/vm/engine/tae/tables/updates/append.go @@ -133,8 +133,8 @@ func (node *AppendNode) PrepareCommit() error { func (node *AppendNode) ApplyCommit(id string) error { node.mvcc.Lock() + defer node.mvcc.Unlock() if node.IsCommitted() { - node.mvcc.Unlock() panic("AppendNode | ApplyCommit | LogicErr") } node.TxnMVCCNode.ApplyCommit(id) @@ -143,7 +143,6 @@ func (node *AppendNode) ApplyCommit(id string) error { if listener != nil { err = listener(node) } - node.mvcc.Unlock() return err } From 3e815e34d198870e76c14a032a57be9bd3ff28f2 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Tue, 14 Jul 2026 11:55:55 +0800 Subject: [PATCH 05/12] fix tae merge notifier race --- pkg/vm/engine/tae/catalog/catalog.go | 3 ++- pkg/vm/engine/tae/catalog/table.go | 18 ++++++++++++------ pkg/vm/engine/tae/catalog/tableForMerge.go | 10 +++++++++- 3 files changed, 23 insertions(+), 8 deletions(-) diff --git a/pkg/vm/engine/tae/catalog/catalog.go b/pkg/vm/engine/tae/catalog/catalog.go index e0a511232749f..e3583bc26d288 100644 --- a/pkg/vm/engine/tae/catalog/catalog.go +++ b/pkg/vm/engine/tae/catalog/catalog.go @@ -19,6 +19,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "time" pkgcatalog "github.com/matrixorigin/matrixone/pkg/catalog" @@ -60,7 +61,7 @@ type Catalog struct { nodesMu sync.RWMutex gcTS types.TS - mergeNotifier MergeNotifierOnCatalog + mergeNotifier atomic.Value } func MockCatalog(dataFactory DataFactory) *Catalog { diff --git a/pkg/vm/engine/tae/catalog/table.go b/pkg/vm/engine/tae/catalog/table.go index f43d0357ef894..f64d737eaf0a3 100644 --- a/pkg/vm/engine/tae/catalog/table.go +++ b/pkg/vm/engine/tae/catalog/table.go @@ -299,8 +299,10 @@ func (entry *TableEntry) CreateObject( entry.Lock() defer entry.Unlock() created = NewObjectEntry(entry, txn, *opts.Stats, dataFactory, opts.IsTombstone) - if entry.GetCatalog().mergeNotifier != nil && !opts.Stats.GetAppendable() { - entry.GetCatalog().mergeNotifier.OnCreateNonAppendObject(ToMergeTable(entry)) + if !opts.Stats.GetAppendable() { + if notifier := entry.GetCatalog().getMergeNotifier(); notifier != nil { + notifier.OnCreateNonAppendObject(ToMergeTable(entry)) + } } entry.AddEntryLocked(created) return @@ -314,8 +316,10 @@ func (entry *TableEntry) CreateCommittedObject( entry.Lock() defer entry.Unlock() created = NewCommittedObjectEntry(entry, ts, *opts.Stats, dataFactory, opts.IsTombstone) - if entry.GetCatalog().mergeNotifier != nil && !opts.Stats.GetAppendable() { - entry.GetCatalog().mergeNotifier.OnCreateNonAppendObject(ToMergeTable(entry)) + if !opts.Stats.GetAppendable() { + if notifier := entry.GetCatalog().getMergeNotifier(); notifier != nil { + notifier.OnCreateNonAppendObject(ToMergeTable(entry)) + } } entry.AddEntryLocked(created) return @@ -783,8 +787,10 @@ func (entry *TableEntry) ApplyCommit(id string) (err error) { entry.TableNode.schema.Store(schema) // create table commit - if lastestNode.DeletedAt.IsEmpty() && entry.GetCatalog().mergeNotifier != nil { - entry.GetCatalog().mergeNotifier.OnCreateTableCommit(ToMergeTable(entry)) + if lastestNode.DeletedAt.IsEmpty() { + if notifier := entry.GetCatalog().getMergeNotifier(); notifier != nil { + notifier.OnCreateTableCommit(ToMergeTable(entry)) + } } return diff --git a/pkg/vm/engine/tae/catalog/tableForMerge.go b/pkg/vm/engine/tae/catalog/tableForMerge.go index 67ca0165560a2..d7a9203936e99 100644 --- a/pkg/vm/engine/tae/catalog/tableForMerge.go +++ b/pkg/vm/engine/tae/catalog/tableForMerge.go @@ -87,7 +87,15 @@ type TNTombstoneItem struct { } func (catalog *Catalog) SetMergeNotifier(notifier MergeNotifierOnCatalog) { - catalog.mergeNotifier = notifier + catalog.mergeNotifier.Store(notifier) +} + +func (catalog *Catalog) getMergeNotifier() MergeNotifierOnCatalog { + notifier := catalog.mergeNotifier.Load() + if notifier == nil { + return nil + } + return notifier.(MergeNotifierOnCatalog) } func (catalog *Catalog) InitSource() iter.Seq[MergeTable] { From dcdab9c380dbe7c6941febaa95c4c015ec97ddf0 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Tue, 14 Jul 2026 14:43:10 +0800 Subject: [PATCH 06/12] fix empty appendable object flush --- pkg/vm/engine/disttae/change_handle.go | 2 +- .../disttae/logtailreplay/change_handle.go | 32 +++++++++++++++++++ .../engine/tae/tables/jobs/flushTableTail.go | 18 +++++++++++ pkg/vm/engine/tae/tables/updates/mvcc.go | 3 ++ 4 files changed, 54 insertions(+), 1 deletion(-) diff --git a/pkg/vm/engine/disttae/change_handle.go b/pkg/vm/engine/disttae/change_handle.go index 3f90f213adc2b..238de56d44812 100644 --- a/pkg/vm/engine/disttae/change_handle.go +++ b/pkg/vm/engine/disttae/change_handle.go @@ -434,7 +434,7 @@ func (h *PartitionChangesHandle) getNextChangeHandle(ctx context.Context) (end b if err != nil { return } - h.currentChangeHandle, err = logtailreplay.NewChangesHandlerWithCheckpointEntries( + h.currentChangeHandle, err = logtailreplay.NewChangesHandlerWithCheckpointRangeRecovery( ctx, h.tbl.tableId, h.tbl.proc.Load().GetService(), diff --git a/pkg/vm/engine/disttae/logtailreplay/change_handle.go b/pkg/vm/engine/disttae/logtailreplay/change_handle.go index 3a7c3382ebbed..87645202c15d4 100755 --- a/pkg/vm/engine/disttae/logtailreplay/change_handle.go +++ b/pkg/vm/engine/disttae/logtailreplay/change_handle.go @@ -1605,6 +1605,38 @@ func NewChangesHandlerWithCheckpointRange( ) } +// NewChangesHandlerWithCheckpointRangeRecovery rebuilds CollectChanges(start, +// end) from checkpoint metadata using range-aware object selection while +// preserving CDC/checkpoint recovery merge semantics. +func NewChangesHandlerWithCheckpointRangeRecovery( + ctx context.Context, + tid uint64, + sid string, + checkpoints []*checkpoint.CheckpointEntry, + start, end types.TS, + skipDeletes bool, + maxRow uint32, + primarySeqnum int, + mp *mpool.MPool, + fs fileservice.FileService, +) (changeHandle *ChangeHandler, err error) { + return newChangesHandlerWithCheckpointEntries( + ctx, + tid, + sid, + checkpoints, + start, + end, + skipDeletes, + maxRow, + primarySeqnum, + mp, + fs, + checkpointObjectSelectionRange, + true, + ) +} + // NewChangesHandlerWithPartitionStateRange rebuilds CollectChanges(start, end) // from the partition state visible at the range end snapshot. // diff --git a/pkg/vm/engine/tae/tables/jobs/flushTableTail.go b/pkg/vm/engine/tae/tables/jobs/flushTableTail.go index 4a242494a91a9..7aa7ae69b8532 100644 --- a/pkg/vm/engine/tae/tables/jobs/flushTableTail.go +++ b/pkg/vm/engine/tae/tables/jobs/flushTableTail.go @@ -181,6 +181,13 @@ func NewFlushTableTailTask( if !hdl.IsAppendable() { panic(fmt.Sprintf("logic err %v is nonappendable", hdl.GetID().String())) } + rows, err := obj.GetObjectData().Rows() + if err != nil { + return nil, err + } + if rows == 0 { + continue + } if obj.GetObjectData().CheckFlushTaskRetry(txn.GetStartTS()) { logutil.Info( "[FLUSH-NEED-RETRY]", @@ -211,6 +218,13 @@ func NewFlushTableTailTask( if !hdl.IsAppendable() { panic(fmt.Sprintf("logic err %v is nonappendable", hdl.GetID().String())) } + rows, err := obj.GetObjectData().Rows() + if err != nil { + return nil, err + } + if rows == 0 { + continue + } if obj.GetObjectData().CheckFlushTaskRetry(txn.GetStartTS()) { logutil.Infof("[FlushTabletail] obj %v needs retry", obj.ID().String()) return nil, txnif.ErrTxnNeedRetry @@ -506,6 +520,10 @@ func (task *flushTableTailTask) prepareAObjSortedData( if err != nil { return } + if bat == nil { + empty = true + return + } for i := range idxs { if vec := bat.Vecs[i]; vec == nil || vec.Length() == 0 { empty = true diff --git a/pkg/vm/engine/tae/tables/updates/mvcc.go b/pkg/vm/engine/tae/tables/updates/mvcc.go index 3f0ecfd48c5b8..c0128eca0b123 100644 --- a/pkg/vm/engine/tae/tables/updates/mvcc.go +++ b/pkg/vm/engine/tae/tables/updates/mvcc.go @@ -315,6 +315,9 @@ func (n *AppendMVCCHandle) PrepareCompact() bool { } func (n *AppendMVCCHandle) GetLatestAppendPrepareTSLocked() types.TS { + if n.appends == nil || n.appends.IsEmpty() { + return types.TS{} + } return n.appends.GetUpdateNodeLocked().Prepare } func (n *AppendMVCCHandle) GetMeta() *catalog.ObjectEntry { From 157f999ad1b3ec0ae3a6b48fa4ea932b560c2944 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Tue, 14 Jul 2026 15:10:59 +0800 Subject: [PATCH 07/12] set merge notifier only in write mode --- pkg/vm/engine/tae/catalog/catalog.go | 3 +-- pkg/vm/engine/tae/catalog/tableForMerge.go | 8 ++----- pkg/vm/engine/tae/db/controller.go | 28 ++++++++++++++-------- pkg/vm/engine/tae/db/merge/scheduler.go | 1 - pkg/vm/engine/tae/db/merge/simulator.go | 1 + 5 files changed, 22 insertions(+), 19 deletions(-) diff --git a/pkg/vm/engine/tae/catalog/catalog.go b/pkg/vm/engine/tae/catalog/catalog.go index e3583bc26d288..e0a511232749f 100644 --- a/pkg/vm/engine/tae/catalog/catalog.go +++ b/pkg/vm/engine/tae/catalog/catalog.go @@ -19,7 +19,6 @@ import ( "context" "fmt" "sync" - "sync/atomic" "time" pkgcatalog "github.com/matrixorigin/matrixone/pkg/catalog" @@ -61,7 +60,7 @@ type Catalog struct { nodesMu sync.RWMutex gcTS types.TS - mergeNotifier atomic.Value + mergeNotifier MergeNotifierOnCatalog } func MockCatalog(dataFactory DataFactory) *Catalog { diff --git a/pkg/vm/engine/tae/catalog/tableForMerge.go b/pkg/vm/engine/tae/catalog/tableForMerge.go index d7a9203936e99..4f7b5459056c1 100644 --- a/pkg/vm/engine/tae/catalog/tableForMerge.go +++ b/pkg/vm/engine/tae/catalog/tableForMerge.go @@ -87,15 +87,11 @@ type TNTombstoneItem struct { } func (catalog *Catalog) SetMergeNotifier(notifier MergeNotifierOnCatalog) { - catalog.mergeNotifier.Store(notifier) + catalog.mergeNotifier = notifier } func (catalog *Catalog) getMergeNotifier() MergeNotifierOnCatalog { - notifier := catalog.mergeNotifier.Load() - if notifier == nil { - return nil - } - return notifier.(MergeNotifierOnCatalog) + return catalog.mergeNotifier } func (catalog *Catalog) InitSource() iter.Seq[MergeTable] { diff --git a/pkg/vm/engine/tae/db/controller.go b/pkg/vm/engine/tae/db/controller.go index df0e99b77417a..c1129dc067d07 100644 --- a/pkg/vm/engine/tae/db/controller.go +++ b/pkg/vm/engine/tae/db/controller.go @@ -296,6 +296,7 @@ func (c *Controller) handleToReplayCmd(cmd *controlCmd) { // 1. stop the merge scheduler c.db.MergeScheduler.Stop() rollbackSteps.Add("stop merge scheduler", func() error { + c.db.Catalog.SetMergeNotifier(c.db.MergeScheduler) c.db.MergeScheduler.Start() return nil }) @@ -342,6 +343,7 @@ func (c *Controller) handleToReplayCmd(cmd *controlCmd) { // TODO: recover the previous state return } + c.db.Catalog.SetMergeNotifier(nil) // 7. wait the logtail push queue to be flushed // TODO @@ -428,17 +430,19 @@ func (c *Controller) handleToWriteCmd(cmd *controlCmd) { // 3. switch the txnmgr to write mode c.db.TxnMgr.ToWriteMode() - // 4. unfreeze the write requests - if err = c.db.TxnServer.SwitchTxnHandleStateTo(rpc2.TxnLocalHandle); err != nil { - return - } - + c.db.Catalog.SetMergeNotifier(c.db.MergeScheduler) c.db.MergeScheduler.Start() rollbackSteps.Add("stop merge scheduler", func() error { c.db.MergeScheduler.Stop() + c.db.Catalog.SetMergeNotifier(nil) return nil }) + // 4. unfreeze the write requests + if err = c.db.TxnServer.SwitchTxnHandleStateTo(rpc2.TxnLocalHandle); err != nil { + return + } + // 5. start merge scheduler|checkpoint|diskcleaner // 5.1 TODO: start the merger|checkpoint|flusher c.db.BGFlusher.Restart() // TODO: Restart with new config @@ -779,11 +783,15 @@ func (c *Controller) AssembleDB(ctx context.Context) (err error) { merge.NewTNMergeExecutor(db.Runtime), merge.NewStdClock(), ) - db.MergeScheduler.Start() - rollbackSteps.Add("stop merge scheduler", func() error { - db.MergeScheduler.Stop() - return nil - }) + if db.IsWriteMode() { + db.Catalog.SetMergeNotifier(db.MergeScheduler) + db.MergeScheduler.Start() + rollbackSteps.Add("stop merge scheduler", func() error { + db.MergeScheduler.Stop() + db.Catalog.SetMergeNotifier(nil) + return nil + }) + } // start flusher and disk cleaner db.BGFlusher.Start() diff --git a/pkg/vm/engine/tae/db/merge/scheduler.go b/pkg/vm/engine/tae/db/merge/scheduler.go index 24e7875327f62..18a1d3c8e142c 100644 --- a/pkg/vm/engine/tae/db/merge/scheduler.go +++ b/pkg/vm/engine/tae/db/merge/scheduler.go @@ -124,7 +124,6 @@ func NewMergeScheduler( }, } } - cata.SetMergeNotifier(sched) return sched diff --git a/pkg/vm/engine/tae/db/merge/simulator.go b/pkg/vm/engine/tae/db/merge/simulator.go index acc4d08d8f454..8db8472bfb72f 100644 --- a/pkg/vm/engine/tae/db/merge/simulator.go +++ b/pkg/vm/engine/tae/db/merge/simulator.go @@ -896,6 +896,7 @@ func NewSimPlayer() *SimPlayer { sexecutor, sclock, ) + scatalog.SetMergeNotifier(sched) sched.PatchTestRscController(srsc) return &SimPlayer{ From 80b1047ce06c4d216e4a9d564152df4acbe8b8ab Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 15 Jul 2026 10:36:21 +0800 Subject: [PATCH 08/12] fix tae flush tombstone transfer scan --- pkg/vm/engine/tae/tables/aobj.go | 8 +++++++ .../engine/tae/tables/jobs/flushTableTail.go | 14 ----------- pkg/vm/engine/tae/tables/table_scan.go | 24 +++++++++---------- 3 files changed, 20 insertions(+), 26 deletions(-) diff --git a/pkg/vm/engine/tae/tables/aobj.go b/pkg/vm/engine/tae/tables/aobj.go index b46a1a9ed60c5..ed975f72cf5bb 100644 --- a/pkg/vm/engine/tae/tables/aobj.go +++ b/pkg/vm/engine/tae/tables/aobj.go @@ -118,6 +118,14 @@ func (obj *aobject) PrepareCompact() bool { } return false } + rows, err := obj.Rows() + if err != nil { + logutil.Warnf("get object rows failed, obj: %v, err: %v", obj.meta.Load().ID().String(), err) + return false + } + if rows == 0 { + return false + } // see more notes in flushtabletail.go obj.freezelock.Lock() diff --git a/pkg/vm/engine/tae/tables/jobs/flushTableTail.go b/pkg/vm/engine/tae/tables/jobs/flushTableTail.go index 7aa7ae69b8532..b10144ecc257a 100644 --- a/pkg/vm/engine/tae/tables/jobs/flushTableTail.go +++ b/pkg/vm/engine/tae/tables/jobs/flushTableTail.go @@ -181,13 +181,6 @@ func NewFlushTableTailTask( if !hdl.IsAppendable() { panic(fmt.Sprintf("logic err %v is nonappendable", hdl.GetID().String())) } - rows, err := obj.GetObjectData().Rows() - if err != nil { - return nil, err - } - if rows == 0 { - continue - } if obj.GetObjectData().CheckFlushTaskRetry(txn.GetStartTS()) { logutil.Info( "[FLUSH-NEED-RETRY]", @@ -218,13 +211,6 @@ func NewFlushTableTailTask( if !hdl.IsAppendable() { panic(fmt.Sprintf("logic err %v is nonappendable", hdl.GetID().String())) } - rows, err := obj.GetObjectData().Rows() - if err != nil { - return nil, err - } - if rows == 0 { - continue - } if obj.GetObjectData().CheckFlushTaskRetry(txn.GetStartTS()) { logutil.Infof("[FlushTabletail] obj %v needs retry", obj.ID().String()) return nil, txnif.ErrTxnNeedRetry diff --git a/pkg/vm/engine/tae/tables/table_scan.go b/pkg/vm/engine/tae/tables/table_scan.go index 5b4465d448e0f..1c4056d8cda09 100644 --- a/pkg/vm/engine/tae/tables/table_scan.go +++ b/pkg/vm/engine/tae/tables/table_scan.go @@ -94,28 +94,28 @@ func TombstoneRangeScanByObject( tableEntry.WaitTombstoneObjectCommitted(end) it := tableEntry.MakeTombstoneObjectIt() defer it.Release() - earlybreak := false for ok := it.Last(); ok; ok = it.Prev() { - if earlybreak { - break - } - tombstone := it.Item() - // we only check the created version of the object. - if tombstone.HasDropIntent() { - continue - } if tombstone.IsAppendable() { if tombstone.CreatedAt.GT(&end) { // committing create object is excluded here continue } - // first committed appendable object with CreatedAt < start, stop at next round - if tombstone.CreatedAt.LT(&start) { - earlybreak = true + if tombstone.DeletedAt.Equal(&txnif.UncommitTS) { + continue + } + if tombstone.HasDropCommitted() { + deleteAt := tombstone.GetDeleteAt() + if tombstone.CreatedAt.GT(&end) || deleteAt.LT(&start) { + continue + } } } else { + // we only check the created version of the object. + if tombstone.HasDropIntent() { + continue + } if !tombstone.ObjectStats.GetCNCreated() { continue } From 5df3fc225a91d3a912b874ef95c51ffca231dd5f Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 15 Jul 2026 15:06:35 +0800 Subject: [PATCH 09/12] fix tae tombstone range scan duplicate versions --- pkg/vm/engine/tae/tables/table_scan.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/vm/engine/tae/tables/table_scan.go b/pkg/vm/engine/tae/tables/table_scan.go index 1c4056d8cda09..dc01ab6c4a123 100644 --- a/pkg/vm/engine/tae/tables/table_scan.go +++ b/pkg/vm/engine/tae/tables/table_scan.go @@ -96,6 +96,11 @@ func TombstoneRangeScanByObject( defer it.Release() for ok := it.Last(); ok; ok = it.Prev() { tombstone := it.Item() + if tombstone.IsCEntry() && tombstone.HasDCounterpart() && tombstone.GetNextVersion().HasDropCommitted() { + // The dropped counterpart owns the persisted appendable tombstone data. + // Scanning both versions duplicates the same committed delete rows. + continue + } if tombstone.IsAppendable() { if tombstone.CreatedAt.GT(&end) { From bd08a99dc44dfaa9f25cb03ca73871ea614b4470 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Thu, 16 Jul 2026 11:36:34 +0800 Subject: [PATCH 10/12] add tae appendable object create coverage --- pkg/vm/engine/tae/txn/txnimpl/txn_test.go | 225 ++++++++++++++++++++++ 1 file changed, 225 insertions(+) diff --git a/pkg/vm/engine/tae/txn/txnimpl/txn_test.go b/pkg/vm/engine/tae/txn/txnimpl/txn_test.go index 6582e5279137d..7458ba1927cf3 100644 --- a/pkg/vm/engine/tae/txn/txnimpl/txn_test.go +++ b/pkg/vm/engine/tae/txn/txnimpl/txn_test.go @@ -34,6 +34,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/entry" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/logstore/wal" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/tables" + "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/tables/updates" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/testutils" "github.com/matrixorigin/matrixone/pkg/vm/engine/tae/txn/txnbase" "github.com/panjf2000/ants/v2" @@ -989,3 +990,227 @@ func TestLogTxnState(t *testing.T) { assert.Greater(t, len(payload), 0) } } + +func TestCreateAppendableObjectWithOptions(t *testing.T) { + defer testutils.AfterTest(t)() + testutils.EnsureNoLeak(t) + + ctx := context.Background() + dir := testutils.InitTestEnv(ModuleName, t) + c, mgr, driver := initTestContext(ctx, t, dir) + defer driver.Close() + defer c.Close() + defer mgr.Stop() + + schema := catalog.MockSchema(1, 0) + txn, err := mgr.StartTxn(nil) + assert.NoError(t, err) + db, err := txn.CreateDatabase("db", "", "") + assert.NoError(t, err) + _, err = db.CreateRelation(schema) + assert.NoError(t, err) + assert.NoError(t, txn.Commit(ctx)) + + txn, err = mgr.StartTxn(nil) + assert.NoError(t, err) + db, err = txn.GetDatabase("db") + assert.NoError(t, err) + rel, err := db.GetRelationByName(schema.Name) + assert.NoError(t, err) + + id := objectio.NewObjectid() + stats := objectio.NewObjectStatsWithObjectID(&id, true, false, false) + obj, err := rel.CreateObjectWithOpt(false, &objectio.CreateObjOpt{Stats: stats}) + assert.NoError(t, err) + assert.True(t, obj.GetMeta().(*catalog.ObjectEntry).IsAppendable()) + assert.Equal(t, id, *obj.GetID()) + + store := txn.GetStore().(*txnStore) + txnDB, err := store.getOrSetDB(db.GetID()) + assert.NoError(t, err) + txnTable, err := txnDB.getOrSetTable(rel.ID()) + assert.NoError(t, err) + assert.Zero(t, txnTable.txnEntries.Len()) + assert.NoError(t, txn.Commit(ctx)) + + txn, err = mgr.StartTxn(nil) + assert.NoError(t, err) + db, err = txn.GetDatabase("db") + assert.NoError(t, err) + rel, err = db.GetRelationByName(schema.Name) + assert.NoError(t, err) + _, err = rel.GetObject(&id, false) + assert.NoError(t, err) + assert.NoError(t, txn.Commit(ctx)) +} + +func TestCreateAppendableObjectWithOptionsRejectsNonAppendable(t *testing.T) { + defer testutils.AfterTest(t)() + testutils.EnsureNoLeak(t) + + ctx := context.Background() + dir := testutils.InitTestEnv(ModuleName, t) + c, mgr, driver := initTestContext(ctx, t, dir) + defer driver.Close() + defer c.Close() + defer mgr.Stop() + + txn, err := mgr.StartTxn(nil) + assert.NoError(t, err) + db, err := txn.CreateDatabase("db", "", "") + assert.NoError(t, err) + rel, err := db.CreateRelation(catalog.MockSchema(1, 0)) + assert.NoError(t, err) + + id := objectio.NewObjectid() + stats := objectio.NewObjectStatsWithObjectID(&id, false, false, false) + _, err = rel.CreateObjectWithOpt(false, &objectio.CreateObjOpt{Stats: stats}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "only supports appendable object") + assert.NoError(t, txn.Rollback(ctx)) +} + +func TestCreateAppendableObjectWithOptionsErrors(t *testing.T) { + defer testutils.AfterTest(t)() + testutils.EnsureNoLeak(t) + + ctx := context.Background() + dir := testutils.InitTestEnv(ModuleName, t) + c, mgr, driver := initTestContext(ctx, t, dir) + defer driver.Close() + defer c.Close() + defer mgr.Stop() + + txn, err := mgr.StartTxn(nil) + assert.NoError(t, err) + db, err := txn.CreateDatabase("db", "", "") + assert.NoError(t, err) + rel, err := db.CreateRelation(catalog.MockSchema(1, 0)) + assert.NoError(t, err) + + store := txn.GetStore().(*txnStore) + newOpt := func() *objectio.CreateObjOpt { + id := objectio.NewObjectid() + return &objectio.CreateObjOpt{ + Stats: objectio.NewObjectStatsWithObjectID(&id, true, false, false), + } + } + + _, err = store.CreateObjectWithOpt(db.GetID()+1, rel.ID(), false, newOpt()) + assert.Error(t, err) + + txnDB, err := store.getOrSetDB(db.GetID()) + assert.NoError(t, err) + _, err = txnDB.CreateObjectWithOpt(rel.ID()+1, newOpt(), false) + assert.Error(t, err) + + store.isOffline = true + _, err = store.CreateObjectWithOpt(db.GetID(), rel.ID(), false, newOpt()) + assert.Error(t, err) + _, err = txnDB.CreateObjectWithOpt(rel.ID(), newOpt(), false) + assert.Error(t, err) + store.isOffline = false + assert.NoError(t, txn.Rollback(ctx)) +} + +type replayAObjectCreateRecorder struct { + created []replayAObjectCreateRecord +} + +type replayAObjectCreateRecord struct { + id common.ID + isTombstone bool + ts types.TS +} + +func (*replayAObjectCreateRecorder) OnTimeStamp(types.TS) {} + +func (r *replayAObjectCreateRecorder) RecordReplayAObjectCreate( + id *common.ID, + isTombstone bool, + ts types.TS, +) { + r.created = append(r.created, replayAObjectCreateRecord{*id, isTombstone, ts}) +} + +func TestEnsureReplayAObject(t *testing.T) { + defer testutils.AfterTest(t)() + testutils.EnsureNoLeak(t) + + ctx := context.Background() + dir := testutils.InitTestEnv(ModuleName, t) + c, mgr, driver := initTestContext(ctx, t, dir) + defer driver.Close() + defer c.Close() + defer mgr.Stop() + + txn, err := mgr.StartTxn(nil) + assert.NoError(t, err) + db, err := txn.CreateDatabase("db", "", "") + assert.NoError(t, err) + rel, err := db.CreateRelation(catalog.MockSchema(1, 0)) + assert.NoError(t, err) + assert.NoError(t, txn.Commit(ctx)) + + database, err := c.GetDatabaseByID(db.GetID()) + assert.NoError(t, err) + table, err := database.GetTableEntryByID(rel.ID()) + assert.NoError(t, err) + id := table.AsCommonID() + objectID := objectio.NewObjectid() + id.SetObjectID(&objectID) + + createTS := mgr.Now() + recorder := new(replayAObjectCreateRecorder) + store := &replayTxnStore{catalog: c} + obj, created, err := store.ensureReplayAObject(id, false, createTS, recorder) + assert.NoError(t, err) + assert.True(t, created) + assert.True(t, obj.IsAppendable()) + assert.Equal(t, createTS, obj.GetCreatedAt()) + assert.Len(t, recorder.created, 1) + assert.Equal(t, *id, recorder.created[0].id) + assert.Equal(t, createTS, recorder.created[0].ts) + + obj, created, err = store.ensureReplayAObject(id, false, mgr.Now(), recorder) + assert.NoError(t, err) + assert.False(t, created) + assert.True(t, obj.IsAppendable()) + assert.Len(t, recorder.created, 1) + + tombstoneID := table.AsCommonID() + objectID = objectio.NewObjectid() + tombstoneID.SetObjectID(&objectID) + obj, created, err = store.ensureReplayAObject(tombstoneID, true, mgr.Now(), nil) + assert.NoError(t, err) + assert.True(t, created) + assert.True(t, obj.IsTombstone) + + missingDB := *id + missingDB.DbID++ + _, _, err = store.ensureReplayAObject(&missingDB, false, mgr.Now(), recorder) + assert.Error(t, err) + + missingTable := *id + missingTable.TableID++ + _, _, err = store.ensureReplayAObject(&missingTable, false, mgr.Now(), recorder) + assert.Error(t, err) +} + +func TestReplayAppendNodeCreateTS(t *testing.T) { + prepareTS := types.BuildTS(10, 0) + commitTS := types.BuildTS(20, 0) + node := updates.NewEmptyAppendNode() + node.TxnMVCCNode.Prepare = prepareTS + node.TxnMVCCNode.End = commitTS + assert.Equal(t, commitTS, replayAppendNodeCreateTS(node)) + + txn := txnbase.NewTxn(nil, &txnbase.NoopTxnStore{}, []byte("txn"), prepareTS, prepareTS) + assert.NoError(t, txn.SetCommitTS(types.BuildTS(30, 0))) + node.TxnMVCCNode.Txn = txn + assert.Equal(t, types.BuildTS(30, 0), replayAppendNodeCreateTS(node)) + + node.TxnMVCCNode.Txn = nil + node.TxnMVCCNode.End = types.TS{} + assert.Equal(t, prepareTS, replayAppendNodeCreateTS(node)) +} From 0927bb6cb025f984f89a2d0c02532ae1b0aa23bd Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Fri, 17 Jul 2026 11:17:26 +0800 Subject: [PATCH 11/12] add flush transfer range diagnostics --- pkg/vm/engine/tae/db/test/db_test.go | 3 ++ pkg/vm/engine/tae/tables/table_scan.go | 2 +- .../tae/tables/txnentries/flushTableTail.go | 41 ++++++++++++++++--- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/pkg/vm/engine/tae/db/test/db_test.go b/pkg/vm/engine/tae/db/test/db_test.go index b6966226abeed..5c2fc5fb1ee20 100644 --- a/pkg/vm/engine/tae/db/test/db_test.go +++ b/pkg/vm/engine/tae/db/test/db_test.go @@ -11764,6 +11764,9 @@ func TestRW3(t *testing.T) { ctx := context.Background() opts := config.WithLongScanAndCKPOpts(nil) tae := testutil.NewTestEngine(ctx, ModuleName, t, opts) + defer func() { + t.Log(tae.Catalog.SimplePPString(common.PPL3)) + }() objCount := 100 schema := catalog.MockSchemaAll(1, -1) diff --git a/pkg/vm/engine/tae/tables/table_scan.go b/pkg/vm/engine/tae/tables/table_scan.go index dc01ab6c4a123..153af7d82e67b 100644 --- a/pkg/vm/engine/tae/tables/table_scan.go +++ b/pkg/vm/engine/tae/tables/table_scan.go @@ -96,7 +96,7 @@ func TombstoneRangeScanByObject( defer it.Release() for ok := it.Last(); ok; ok = it.Prev() { tombstone := it.Item() - if tombstone.IsCEntry() && tombstone.HasDCounterpart() && tombstone.GetNextVersion().HasDropCommitted() { + if tombstone.IsCEntry() && tombstone.HasDCounterpart() { // The dropped counterpart owns the persisted appendable tombstone data. // Scanning both versions duplicates the same committed delete rows. continue diff --git a/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go b/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go index 3f847b3893395..79697ff97b33c 100644 --- a/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go +++ b/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go @@ -116,7 +116,7 @@ func NewFlushTableTailEntry( time.Sleep(time.Second) } var err error - entry.transCntBeforeCommit, err = entry.collectDelsAndTransfer(ctx, entry.txn.GetStartTS(), entry.collectTs) + entry.transCntBeforeCommit, err = entry.collectDelsAndTransfer(ctx, "flush", entry.txn.GetStartTS(), entry.collectTs) if err != nil { return nil, err } @@ -195,8 +195,28 @@ func (entry *flushTableTailEntry) addTransferPages(ctx context.Context) error { // collectDelsAndTransfer collects deletes in flush process and moves them to the created obj // ATTENTION !!! (from, to] !!! func (entry *flushTableTailEntry) collectDelsAndTransfer( - ctx context.Context, from, to types.TS, + ctx context.Context, phase string, from, to types.TS, ) (transCnt int, err error) { + scanStart := from.Next() + defer func() { + fields := []zap.Field{ + zap.String("task", entry.taskName), + zap.String("phase", phase), + zap.String("range-from", from.ToString()), + zap.String("range-scan-start", scanStart.ToString()), + zap.String("range-to", to.ToString()), + zap.String("txn-start-ts", entry.txn.GetStartTS().ToString()), + zap.String("collect-ts", entry.collectTs.ToString()), + zap.String("prepare-ts", entry.txn.GetPrepareTS().ToString()), + zap.Int("aobjs", len(entry.aobjHandles)), + zap.Bool("has-created-object", entry.createdObjHandle != nil), + zap.Int("transfer-rows", transCnt), + } + if err != nil { + fields = append(fields, zap.Error(err)) + } + logutil.Info("[FLUSH-TRANSFER]", fields...) + }() if len(entry.aobjHandles) == 0 { return } @@ -226,7 +246,7 @@ func (entry *flushTableTailEntry) collectDelsAndTransfer( ctx, entry.tableEntry, *obj.ID(), - from.Next(), // NOTE HERE + scanStart, // NOTE HERE to, common.MergeAllocator, entry.rt.VectorPool.Small, @@ -285,7 +305,12 @@ func (entry *flushTableTailEntry) PrepareCommit() error { return nil } ctx := context.Background() - trans, err := entry.collectDelsAndTransfer(ctx, entry.collectTs, entry.txn.GetPrepareTS().Prev()) + txnStart := entry.txn.GetStartTS() + flushScanStart := txnStart.Next() + commitScanStart := entry.collectTs.Next() + prepareTS := entry.txn.GetPrepareTS() + preparePrev := prepareTS.Prev() + trans, err := entry.collectDelsAndTransfer(ctx, "prepare-commit", entry.collectTs, preparePrev) if err != nil { return err } @@ -294,7 +319,13 @@ func (entry *flushTableTailEntry) PrepareCommit() error { logutil.Info( "[FLUSH-PREPARE-COMMIT]", zap.String("task", entry.taskName), - zap.String("commit-ts", entry.txn.GetPrepareTS().ToString()), + zap.String("commit-ts", prepareTS.ToString()), + zap.String("flush-range-from", txnStart.ToString()), + zap.String("flush-range-scan-start", flushScanStart.ToString()), + zap.String("flush-range-to", entry.collectTs.ToString()), + zap.String("commit-range-from", entry.collectTs.ToString()), + zap.String("commit-range-scan-start", commitScanStart.ToString()), + zap.String("commit-range-to", preparePrev.ToString()), zap.Int("ablks", aconflictCnt), zap.Int("transfer-rows", totalTrans), zap.Int("in-queue-transfers", trans), From 62c7397978fdd2292277fc3dd57abfbe1ffedb70 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Fri, 17 Jul 2026 11:20:30 +0800 Subject: [PATCH 12/12] adjust flush transfer diagnostics --- .../tae/tables/txnentries/flushTableTail.go | 26 +++---------------- 1 file changed, 4 insertions(+), 22 deletions(-) diff --git a/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go b/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go index 79697ff97b33c..f6e7152fce5a4 100644 --- a/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go +++ b/pkg/vm/engine/tae/tables/txnentries/flushTableTail.go @@ -116,7 +116,7 @@ func NewFlushTableTailEntry( time.Sleep(time.Second) } var err error - entry.transCntBeforeCommit, err = entry.collectDelsAndTransfer(ctx, "flush", entry.txn.GetStartTS(), entry.collectTs) + entry.transCntBeforeCommit, err = entry.collectDelsAndTransfer(ctx, entry.txn.GetStartTS(), entry.collectTs) if err != nil { return nil, err } @@ -195,28 +195,9 @@ func (entry *flushTableTailEntry) addTransferPages(ctx context.Context) error { // collectDelsAndTransfer collects deletes in flush process and moves them to the created obj // ATTENTION !!! (from, to] !!! func (entry *flushTableTailEntry) collectDelsAndTransfer( - ctx context.Context, phase string, from, to types.TS, + ctx context.Context, from, to types.TS, ) (transCnt int, err error) { scanStart := from.Next() - defer func() { - fields := []zap.Field{ - zap.String("task", entry.taskName), - zap.String("phase", phase), - zap.String("range-from", from.ToString()), - zap.String("range-scan-start", scanStart.ToString()), - zap.String("range-to", to.ToString()), - zap.String("txn-start-ts", entry.txn.GetStartTS().ToString()), - zap.String("collect-ts", entry.collectTs.ToString()), - zap.String("prepare-ts", entry.txn.GetPrepareTS().ToString()), - zap.Int("aobjs", len(entry.aobjHandles)), - zap.Bool("has-created-object", entry.createdObjHandle != nil), - zap.Int("transfer-rows", transCnt), - } - if err != nil { - fields = append(fields, zap.Error(err)) - } - logutil.Info("[FLUSH-TRANSFER]", fields...) - }() if len(entry.aobjHandles) == 0 { return } @@ -310,7 +291,7 @@ func (entry *flushTableTailEntry) PrepareCommit() error { commitScanStart := entry.collectTs.Next() prepareTS := entry.txn.GetPrepareTS() preparePrev := prepareTS.Prev() - trans, err := entry.collectDelsAndTransfer(ctx, "prepare-commit", entry.collectTs, preparePrev) + trans, err := entry.collectDelsAndTransfer(ctx, entry.collectTs, preparePrev) if err != nil { return err } @@ -320,6 +301,7 @@ func (entry *flushTableTailEntry) PrepareCommit() error { "[FLUSH-PREPARE-COMMIT]", zap.String("task", entry.taskName), zap.String("commit-ts", prepareTS.ToString()), + zap.String("transfer-split-ts", entry.collectTs.ToString()), zap.String("flush-range-from", txnStart.ToString()), zap.String("flush-range-scan-start", flushScanStart.ToString()), zap.String("flush-range-to", entry.collectTs.ToString()),