Rollback safe system table additions

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -14,7 +14,8 @@ use crate::{
     execution_context::ExecutionContext,
     system_tables::{
         read_hash_from_col, read_identity_from_col, system_table_schema, StClientRow, StModuleFields, StModuleRow,
-        StTableFields, ST_CLIENT_ID, ST_MODULE_ID, ST_TABLE_ID,
+        StTableFields, ST_CLIENT_ID, ST_COLUMN_ID, ST_CONSTRAINT_ID, ST_INDEX_ID, ST_MODULE_ID, ST_SEQUENCE_ID,
+        ST_TABLE_ID,
     },
     traits::{
         DataRow, IsolationLevel, Metadata, MutTx, MutTxDatastore, Program, RowTypeForTable, Tx, TxData, TxDatastore,
@@ -123,6 +124,41 @@ 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
+        }
+    }
+
+    /// Read the committed rows which describe built-in system table schemas.
+    ///
+    /// These are the rows installed directly by [`Self::bootstrap`] before any
+    /// transaction exists. Durable databases write these rows into the commit
+    /// log at offset 0 so replay from the beginning can reconstruct system
+    /// tables added after the replaying binary was built.
+    pub fn committed_system_table_schema_rows(&self) -> Result<Vec<(TableId, ProductValue)>> {
+        let committed_state = self.committed_state.read();
+        let mut rows = Vec::new();
+
+        for table_id in [ST_TABLE_ID, ST_COLUMN_ID, ST_CONSTRAINT_ID, ST_INDEX_ID, ST_SEQUENCE_ID] {
+            for row in committed_state.iter(table_id)? {
+                rows.push((table_id, row.to_product_value()));
+            }
+        }
+
+        Ok(rows)
+    }
+
     /// 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

@@ -2161,12 +2161,22 @@ 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.
+
+        // Match `bootstrap_system_tables` when creating built-in system sequences.
+        // Check the built-in catalog because the reserved-ID helper also includes
+        // the first user-table ID, whose sequences must start at `seq.start`.
+        let allocated = if matching_system_table_schema.is_some() {
+            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,
             increment: SequenceSchema::INCREMENT,
             start: seq.start,
             min_value: SequenceSchema::MIN_VALUE,

crates/datastore/src/system_tables.rs

@@ -234,6 +234,71 @@ pub fn system_tables() -> [TableSchema; 21] {
     ]
 }
 
+/// 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.
+#[cfg(any(test, feature = "test"))]
+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: SequenceSchema::INCREMENT,
+                min_value: SequenceSchema::MIN_VALUE,
+                max_value: SequenceSchema::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.