Rollback safe system table additions

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -138,6 +138,22 @@ impl Locking {
         Replay::new(self.database_identity, committed_state, progress, error_behavior)
     }
 
+    /// Reserve the first transaction offset for the durable system-schema
+    /// bootstrap record.
+    ///
+    /// The bootstrap record describes rows already installed by
+    /// [`Self::bootstrap`], so this only advances the offset counter. Returns
+    /// `None` if any transaction has already consumed an offset.
+    pub fn reserve_system_schema_bootstrap_tx_offset(&self) -> Option<TxOffset> {
+        let mut committed_state = self.committed_state.write();
+        if committed_state.next_tx_offset == 0 {
+            committed_state.next_tx_offset = 1;
+            Some(0)
+        } else {
+            None
+        }
+    }
+
     /// Construct a new [`Locking`] datastore containing the state stored in `snapshot`.
     ///
     /// - Construct all the tables referenced by `snapshot`, computing their schemas

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -19,11 +19,11 @@ use crate::{
 use crate::{
     error::{IndexError, SequenceError, TableError},
     system_tables::{
-        with_sys_table_buf, StClientFields, StClientRow, StColumnAccessorFields, StColumnAccessorRow, StColumnFields,
-        StColumnRow, StConstraintFields, StConstraintRow, StEventTableFields, StEventTableRow, StFields as _,
-        StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow, StRowLevelSecurityFields,
-        StRowLevelSecurityRow, StScheduledFields, StScheduledRow, StSequenceFields, StSequenceRow,
-        StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow, SystemTable, ST_CLIENT_ID,
+        table_id_is_reserved, with_sys_table_buf, StClientFields, StClientRow, StColumnAccessorFields,
+        StColumnAccessorRow, StColumnFields, StColumnRow, StConstraintFields, StConstraintRow, StEventTableFields,
+        StEventTableRow, StFields as _, StIndexAccessorFields, StIndexAccessorRow, StIndexFields, StIndexRow,
+        StRowLevelSecurityFields, StRowLevelSecurityRow, StScheduledFields, StScheduledRow, StSequenceFields,
+        StSequenceRow, StTableAccessorFields, StTableAccessorRow, StTableFields, StTableRow, SystemTable, ST_CLIENT_ID,
         ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ID, ST_CONSTRAINT_ID, ST_EVENT_TABLE_ID, ST_INDEX_ACCESSOR_ID, ST_INDEX_ID,
         ST_ROW_LEVEL_SECURITY_ID, ST_SCHEDULED_ID, ST_SEQUENCE_ID, ST_TABLE_ACCESSOR_ID, ST_TABLE_ID,
     },
@@ -2042,12 +2042,24 @@ impl MutTxId {
         // Insert the sequence row into st_sequences
         // NOTE: Because st_sequences has a unique index on sequence_name, this will
         // fail if the table already exists.
+
+        // This is a hack to match the bootstrapping logic for system tables.
+        // In `bootstrap_system_tables`, the allocated value is set to `start - 1` for reserved tables,
+        // so we treat reserved tables differently here to match that behavior.
+
+        let allocated = if table_id_is_reserved(table_id) {
+            seq.start - 1
+        } else {
+            seq.start
+        };
+
         let mut sequence_row = StSequenceRow {
             sequence_id,
             sequence_name: seq.sequence_name,
             table_id,
             col_pos: seq.col_pos,
-            allocated: seq.start,
+            // allocated: seq.start,
+            allocated,
             increment: seq.increment,
             start: seq.start,
             min_value: seq.min_value,

crates/datastore/src/system_tables.rs

@@ -233,6 +233,70 @@ pub fn system_tables() -> [TableSchema; 20] {
     ]
 }
 
+/// Return the rows which describe the built-in system table schemas.
+///
+/// These rows are inserted directly into committed state by
+/// `CommittedState::bootstrap_system_tables`. Durable databases also write
+/// them into the commit log before the first ordinary transaction so replay
+/// from offset 0 can learn about system tables which are newer than the replay
+/// binary's built-in catalog.
+pub fn system_table_schema_rows() -> Vec<(TableId, ProductValue)> {
+    let schemas = system_tables();
+    let mut rows = Vec::new();
+
+    for schema in &schemas {
+        rows.push((
+            ST_TABLE_ID,
+            ProductValue::from(StTableRow {
+                table_id: schema.table_id,
+                table_name: schema.table_name.clone(),
+                table_type: StTableType::System,
+                table_access: schema.table_access,
+                table_primary_key: schema.primary_key.map(Into::into),
+            }),
+        ));
+    }
+
+    for col in schemas.iter().flat_map(|schema| schema.columns()).cloned() {
+        rows.push((ST_COLUMN_ID, ProductValue::from(StColumnRow::from(col))));
+    }
+
+    for constraint in schemas.iter().flat_map(|schema| &schema.constraints) {
+        rows.push((
+            ST_CONSTRAINT_ID,
+            ProductValue::from(StConstraintRow {
+                constraint_id: constraint.constraint_id,
+                constraint_name: constraint.constraint_name.clone(),
+                table_id: constraint.table_id,
+                constraint_data: constraint.data.clone().into(),
+            }),
+        ));
+    }
+
+    for index in schemas.iter().flat_map(|schema| &schema.indexes).cloned() {
+        rows.push((ST_INDEX_ID, ProductValue::from(StIndexRow::from(index))));
+    }
+
+    for seq in schemas.iter().flat_map(|schema| &schema.sequences) {
+        rows.push((
+            ST_SEQUENCE_ID,
+            ProductValue::from(StSequenceRow {
+                sequence_id: seq.sequence_id,
+                sequence_name: seq.sequence_name.clone(),
+                table_id: seq.table_id,
+                col_pos: seq.col_pos,
+                increment: seq.increment,
+                min_value: seq.min_value,
+                max_value: seq.max_value,
+                start: seq.start,
+                allocated: seq.start - 1,
+            }),
+        ));
+    }
+
+    rows
+}
+
 /// Types that represent the fields / columns of a system table.
 pub trait StFields: Copy + Sized {
     /// Returns the column position of the system table field.