feat(auto-increment): add transactional epoch fence

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/sql/plan/deepcopy.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/vm/engine/cmd_util/operations.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/vm/engine/disttae/cache/catalog.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/disttae/tools.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/transfer.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
 				}