feat: Allow reordering columns of empty tables during auto-migration

crates/datastore/src/error.rs

@@ -82,8 +82,8 @@ pub enum TableError {
     ChangeColumnsError(#[from] Box<table::ChangeColumnsError>),
     #[error(transparent)]
     AddColumnsError(#[from] Box<table::AddColumnsError>),
-    #[error("Event table with ID `{0}` is not empty")]
-    EventTableNotEmpty(TableId),
+    #[error("Table with ID `{0}` attempted a reschema requiring an empty table, but it is not empty")]
+    TableNotEmpty(TableId),
     #[error(
         "Table with ID `{0}` attempted to reschema using `alter_event_table_row_type`, but it is not an event table"
     )]

crates/datastore/src/locking_tx_datastore/committed_state.rs

@@ -853,7 +853,7 @@ impl CommittedState {
                 unsafe { table.change_columns_to_unchecked(column_schemas, |_, _, _| Ok::<_, Infallible>(())) }
                     .unwrap_or_else(|e| match e {});
             }
-            ReschemaEventTable(table_id, column_schemas) => {
+            ReschemaEmptyTable(table_id, column_schemas) => {
                 let table = self.tables.get_mut(&table_id)?;
                 // SAFETY:
                 // Same argument as in `TableAlterRowType` applies,

crates/datastore/src/locking_tx_datastore/datastore.rs

@@ -327,6 +327,15 @@ impl Locking {
         tx.alter_event_table_row_type(table_id, column_schemas)
     }
 
+    pub fn alter_empty_table_row_type_mut_tx(
+        &self,
+        tx: &mut MutTxId,
+        table_id: TableId,
+        column_schemas: Vec<ColumnSchema>,
+    ) -> Result<()> {
+        tx.alter_empty_table_row_type(table_id, column_schemas)
+    }
+
     pub fn add_columns_to_table_mut_tx(
         &self,
         tx: &mut MutTxId,

crates/datastore/src/locking_tx_datastore/mut_tx.rs

@@ -1462,21 +1462,35 @@ impl MutTxId {
             return Err(TableError::ReschemaNotAnEventTable(table_id).into());
         }
 
+        self.alter_empty_table_row_type(table_id, column_schemas)
+    }
+
+    /// Change the row type of the table identified by `table_id` to `column_schemas`,
+    /// without requiring the new row type to be layout-compatible with the old
+    /// (e.g. columns may be reordered).
+    ///
+    /// This is only valid on a table with no resident rows;
+    /// errors with [`TableError::TableNotEmpty`] otherwise.
+    pub(crate) fn alter_empty_table_row_type(
+        &mut self,
+        table_id: TableId,
+        column_schemas: Vec<ColumnSchema>,
+    ) -> Result<()> {
         // Write to the table in the tx state.
         let ((tx_table, ..), (commit_table, ..)) = self.get_or_create_insert_table_mut(table_id)?;
 
         if tx_table.row_count != 0 || commit_table.row_count != 0 {
             // N.b. the delete table must also be empty, 'cause the committed table is empty.
-            return Err(TableError::EventTableNotEmpty(table_id).into());
+            return Err(TableError::TableNotEmpty(table_id).into());
         }
 
         let old_column_schemas = tx_table
             .change_columns_of_empty_table_to(column_schemas.clone())
-            .map_err(|_| TableError::EventTableNotEmpty(table_id))?;
+            .map_err(|_| TableError::TableNotEmpty(table_id))?;
 
         commit_table
             .change_columns_of_empty_table_to(column_schemas.clone())
-            .map_err(|_| TableError::EventTableNotEmpty(table_id))?;
+            .map_err(|_| TableError::TableNotEmpty(table_id))?;
 
         // Update system tables.
         // We'll simply remove all rows in `st_columns` and then add the new ones.
@@ -1487,7 +1501,7 @@ impl MutTxId {
         self.insert_st_column(&table_name, &column_schemas)?;
 
         // Remember the pending change so we can undo if necessary.
-        self.push_schema_change(PendingSchemaChange::ReschemaEventTable(table_id, old_column_schemas));
+        self.push_schema_change(PendingSchemaChange::ReschemaEmptyTable(table_id, old_column_schemas));
 
         Ok(())
     }

crates/datastore/src/locking_tx_datastore/replay.rs

@@ -930,10 +930,14 @@ impl<'cs> ReplayCommittedState<'cs> {
         let is_event = self.is_event_table_for_replay(table_id)?;
         // Update the columns and layout of the the in-memory table.
         if let Some(table) = self.tables.get_mut(&table_id) {
-            if is_event {
+            if is_event || table.row_count == 0 {
+                // Layout-incompatible reschemas (e.g. reordering the columns of an empty
+                // table, `AutoMigrateStep::ReschemaEmptyTable`) are only ever committed
+                // against a table with no resident rows, so when the table is empty at
+                // this point in the log, mirror that and skip layout-compatibility checks.
                 table
                     .change_columns_of_empty_table_to(columns)
-                    .map_err(|_| TableError::EventTableNotEmpty(table_id))?;
+                    .map_err(|_| TableError::TableNotEmpty(table_id))?;
             } else {
                 table.change_columns_to(columns).map_err(TableError::from)?;
             }

crates/datastore/src/locking_tx_datastore/tx_state.rs

@@ -123,12 +123,13 @@ pub enum PendingSchemaChange {
     /// Only non-representational row-type changes are allowed here,
     /// so existing rows in the table will be compatible with the new row type.
     TableAlterRowType(TableId, Vec<ColumnSchema>),
-    /// The row type of the event table with [`TableId`] was changed.
+    /// The row type of the empty table with [`TableId`] was changed.
     /// The old column schemas was stored.
     ///
-    /// As event tables never have rows resident across transactions or during automigrations,
-    /// we're fine to allow representational/layout-incompatible changes here.
-    ReschemaEventTable(TableId, Vec<ColumnSchema>),
+    /// The table was verified to have no resident rows at the time of the change
+    /// (event tables are rowless by construction),
+    /// so we're fine to allow representational/layout-incompatible changes here.
+    ReschemaEmptyTable(TableId, Vec<ColumnSchema>),
     /// The primary key of the table with [`TableId`] was changed.
     /// The old primary key was stored.
     TableAlterPrimaryKey(TableId, Option<ColList>),
@@ -168,7 +169,7 @@ impl MemoryUsage for PendingSchemaChange {
                 table_id.heap_usage() + sequence.heap_usage() + sequence_schema.heap_usage()
             }
             Self::SequenceAdded(table_id, sequence_id) => table_id.heap_usage() + sequence_id.heap_usage(),
-            Self::ReschemaEventTable(table_id, column_schemas) => table_id.heap_usage() + column_schemas.heap_usage(),
+            Self::ReschemaEmptyTable(table_id, column_schemas) => table_id.heap_usage() + column_schemas.heap_usage(),
         }
     }
 }

crates/engine/src/relational_db.rs

@@ -1092,6 +1092,17 @@ impl RelationalDB {
             .alter_event_table_row_type_mut_tx(tx, table_id, column_schemas)?)
     }
 
+    pub(crate) fn alter_empty_table_row_type(
+        &self,
+        tx: &mut MutTx,
+        table_id: TableId,
+        column_schemas: Vec<ColumnSchema>,
+    ) -> Result<(), DBError> {
+        Ok(self
+            .inner
+            .alter_empty_table_row_type_mut_tx(tx, table_id, column_schemas)?)
+    }
+
     pub(crate) fn add_columns_to_table_mut_tx(
         &self,
         tx: &mut MutTx,