feat(auto-increment): add transactional epoch fence (#26027)

## What type of PR is this?

- [x] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #23143

## What this PR does / why we need it:

This is the first AUTO_INCREMENT-specific prerequisite split from the
closed #25883. It does not expose `ALTER TABLE ... AUTO_INCREMENT=N`
yet.

The invariant introduced here is: a CN write carries the AUTO_INCREMENT
epoch observed when its values were allocated, and TN accepts the write
only when that epoch matches the table epoch. Once a table enters a
fenced epoch, legacy writers that cannot provide an epoch fail closed.

| Correctness requirement | Change in this PR |
| --- | --- |
| Preserve the allocator generation across the write path | Persist the
epoch in `SchemaExtra` / `plan.TableDef`, propagate it through disttae
workspace, row and object writes, flush/compaction, RPC, and TN
relations |
| Prevent stale cached values from committing after a reset | Validate
the known epoch at TN prepare/commit and require every epoch transition
to be exactly `current + 1` |
| Keep the fence effective after restart | Rebuild committed DML
watermarks during replay and keep prepared replay transactions fenced
until commit or rollback |
| Remain compatible before the feature is enabled | Treat an unknown
legacy epoch as zero only while the table epoch is zero; after the first
reset, unknown writers are rejected |

Additional guards reject epoch exhaustion before mutation and keep
ordinary schema ALTER operations from advancing the AUTO_INCREMENT
epoch.

### Explicit non-goals

- no allocator offset/cache reset API;
- no COPY/CLONE allocator reconciliation;
- no user-facing `ALTER TABLE ... AUTO_INCREMENT=N` planner or compile
entry point;
- no generic prepared/committing 2PC recovery redesign. The independent
TN route/cancellation prerequisite was merged in #25956.

### Diff size

- handwritten code and tests: 23 files, `+1283/-131`;
- generated protobuf code: 2 files, `+1565/-1079`;
- total: 25 files. Proto sources and generated files are committed
together.

### Validation

- `go test ./pkg/vm/engine/disttae ./pkg/vm/engine/tae/catalog
./pkg/vm/engine/tae/rpc ./pkg/vm/engine/tae/txn/txnimpl
./pkg/vm/engine/tae/db/test`
- focused `-race` coverage for epoch transitions and replay
commit/rollback lifecycle
- protobuf regeneration stability check
- `git diff --check`
- `mo-pr-preflight-review`: PASS at exact head
`4122a7add078c79757121a4add2d8771113c7980`
- `mo-self-review`: no open blocker

The PR is intentionally Draft while its full CI runs and while this
prerequisite is reviewed independently.

---------

Co-authored-by: ULookup <CHBulookup@outlook.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: 3 people

pkg/pb/api/api.go

@@ -76,6 +76,19 @@ func CloneExtra(info *SchemaExtra) *SchemaExtra {
 		FeatureFlag:       info.FeatureFlag,
 		IndexTables:       append([]uint64{}, info.IndexTables...),
 		ParentTableID:     info.ParentTableID,
+		AutoIncrOffset:    info.AutoIncrOffset,
+		AutoIncrEpoch:     info.AutoIncrEpoch,
+	}
+}
+
+func NewUpdateAutoIncrementReq(did, tid, offset uint64, epoch uint32) *AlterTableReq {
+	return &AlterTableReq{
+		DbId:    did,
+		TableId: tid,
+		Kind:    AlterKind_UpdateAutoIncrement,
+		Operation: &AlterTableReq_UpdateAutoIncrement{
+			UpdateAutoIncrement: &AlterTableAutoIncrement{Offset: offset, Epoch: epoch},
+		},
 	}
 }
 

pkg/pb/api/api.pb.go

@@ -507,6 +507,7 @@ func DeepCopyTableDef(table *plan.TableDef, withCols bool) *plan.TableDef {
 		TableLockType:  table.TableLockType,
 		IsTemporary:    table.IsTemporary,
 		AutoIncrOffset: table.AutoIncrOffset,
+		AutoIncrEpoch:  table.AutoIncrEpoch,
 		DbName:         table.DbName,
 		DbId:           table.DbId,
 		FeatureFlag:    table.FeatureFlag,

pkg/pb/plan/plan.pb.go

@@ -162,8 +162,13 @@ type WriteReq struct {
 	TableID      uint64
 	DatabaseName string
 	TableName    string
-	Schema       *catalog2.Schema
-	Batch        *batch.Batch
+	// AutoIncrEpoch is the allocator epoch used by CN to plan the write.
+	// AutoIncrEpochKnown distinguishes a valid initial zero epoch from an
+	// old CN that did not send the dependency.
+	AutoIncrEpoch      uint32
+	AutoIncrEpochKnown bool
+	Schema             *catalog2.Schema
+	Batch              *batch.Batch
 	//[IncrementalDedup|FullSkipWorkspaceDedup|FullDedup], default is IncrementalDedup.
 	//If incremental-dedup in dn.toml is false, IncrementalDedup will be treated as FullSkipWorkspaceDedup.
 	//IncrementalDedup do not check uniqueness of PK before txn's snapshot TS.

pkg/sql/plan/deepcopy.go

@@ -853,24 +853,26 @@ func getTableDef(tblItem *TableItem, coldefs []engine.TableDef) (*plan.TableDef,
 	}
 
 	return &plan.TableDef{
-		TblId:         tblItem.Id,
-		Name:          tblItem.Name,
-		DbName:        tblItem.DatabaseName,
-		Cols:          cols,
-		Name2ColIndex: name2index,
-		Defs:          defs,
-		TableType:     TableType,
-		Createsql:     Createsql,
-		Pkey:          primarykey,
-		ViewSql:       viewSql,
-		Fkeys:         foreignKeys,
-		RefChildTbls:  refChildTbls,
-		ClusterBy:     clusterByDef,
-		Indexes:       indexes,
-		Version:       tblItem.Version,
-		DbId:          tblItem.DatabaseId,
-		Partition:     partition,
-		FeatureFlag:   tblItem.ExtraInfo.GetFeatureFlag(),
-		LogicalId:     tblItem.LogicalId,
+		TblId:          tblItem.Id,
+		Name:           tblItem.Name,
+		DbName:         tblItem.DatabaseName,
+		Cols:           cols,
+		Name2ColIndex:  name2index,
+		Defs:           defs,
+		TableType:      TableType,
+		Createsql:      Createsql,
+		Pkey:           primarykey,
+		ViewSql:        viewSql,
+		Fkeys:          foreignKeys,
+		RefChildTbls:   refChildTbls,
+		ClusterBy:      clusterByDef,
+		Indexes:        indexes,
+		Version:        tblItem.Version,
+		DbId:           tblItem.DatabaseId,
+		Partition:      partition,
+		FeatureFlag:    tblItem.ExtraInfo.GetFeatureFlag(),
+		AutoIncrOffset: tblItem.ExtraInfo.GetAutoIncrOffset(),
+		AutoIncrEpoch:  tblItem.ExtraInfo.GetAutoIncrEpoch(),
+		LogicalId:      tblItem.LogicalId,
 	}, tableDef
 }

pkg/vm/engine/cmd_util/operations.go

@@ -173,14 +173,16 @@ func toPBEntry(e Entry) (*api.Entry, error) {
 		return nil, err
 	}
 	return &api.Entry{
-		Bat:          bat,
-		EntryType:    typ,
-		TableId:      e.tableId,
-		DatabaseId:   e.databaseId,
-		TableName:    e.tableName,
-		DatabaseName: e.databaseName,
-		FileName:     e.fileName,
-		PkCheckByTn:  int32(e.pkChkByTN),
+		Bat:                bat,
+		EntryType:          typ,
+		TableId:            e.tableId,
+		DatabaseId:         e.databaseId,
+		TableName:          e.tableName,
+		DatabaseName:       e.databaseName,
+		FileName:           e.fileName,
+		PkCheckByTn:        int32(e.pkChkByTN),
+		AutoIncrEpoch:      e.autoIncrEpoch,
+		AutoIncrEpochKnown: e.autoIncrEpochKnown,
 	}, nil
 }
 

pkg/vm/engine/disttae/cache/catalog.go

@@ -148,11 +148,12 @@ func transferTombstoneObjects(
 
 			if bat.RowCount() > 0 {
 				fileName := slist[0].ObjectName().String()
-				if err = txn.WriteFileLocked(
+				if err = txn.writeFileLockedWithAutoIncrEpoch(
 					DELETE,
 					tbl.accountId, tbl.db.databaseId, tbl.tableId,
 					tbl.db.databaseName, tbl.tableName, fileName,
 					bat, txn.tnStores[0],
+					tbl.extraInfo.AutoIncrEpoch,
 				); err != nil {
 					return err
 				}