Move appendable object creation outside transactions

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(),

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.
 //

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

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,49 @@ 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]
+	newNode := oldNode.Clone()
+	if oldNode.IsDEntry() {
+		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)
+	}
+	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) {

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()
 }
@@ -288,8 +299,27 @@ 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
+}
+
+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 !opts.Stats.GetAppendable() {
+		if notifier := entry.GetCatalog().getMergeNotifier(); notifier != nil {
+			notifier.OnCreateNonAppendObject(ToMergeTable(entry))
+		}
 	}
 	entry.AddEntryLocked(created)
 	return
@@ -757,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

pkg/vm/engine/tae/catalog/tableForMerge.go

@@ -90,6 +90,10 @@ func (catalog *Catalog) SetMergeNotifier(notifier MergeNotifierOnCatalog) {
 	catalog.mergeNotifier = notifier
 }
 
+func (catalog *Catalog) getMergeNotifier() MergeNotifierOnCatalog {
+	return catalog.mergeNotifier
+}
+
 func (catalog *Catalog) InitSource() iter.Seq[MergeTable] {
 	return func(yield func(MergeTable) bool) {
 		p := new(LoopProcessor)

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)

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()

pkg/vm/engine/tae/db/merge/scheduler.go

@@ -124,7 +124,6 @@ func NewMergeScheduler(
 			},
 		}
 	}
-	cata.SetMergeNotifier(sched)
 
 	return sched
 

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{