From ff614e348b1e1f5a890fcd55c286d405c5b2c5a9 Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Fri, 7 Aug 2026 23:28:08 +0200 Subject: [PATCH 1/4] feat: Allow reordering columns of empty tables during auto-migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A table with no resident rows has no data to migrate, so a layout-incompatible reschema is safe — the same justification already encoded for event tables. Reordering previously always failed with `AutoMigrateError::ReorderTable`. - Add `AutoMigratePrecheck::CheckTableEmpty`, validating emptiness before any mutations (as requested in the #4875 review), and use it for `RemoveTable` and the new step. - Add `AutoMigrateStep::ReschemaEmptyTable`; the planner emits it (plus the precheck and `DisconnectAllUsers`) for column position changes on non-event tables. Sub-objects on moved columns are removed and re-added; changed unique constraints are allowed (Remove+Add) only under this step. - Generalize `alter_event_table_row_type` into `alter_empty_table_row_type`; rename `PendingSchemaChange::ReschemaEventTable` -> `ReschemaEmptyTable` and `TableError::EventTableNotEmpty` -> `TableNotEmpty`. - Fix replay: `st_column_changed` now uses the unchecked empty-table path when the table is empty at that point in the log, so a committed reorder replays instead of failing layout checks on reopen. - Fix `change_columns_of_empty_table_to` on tables that previously contained rows: drop residual pages before `set_pages` (new `Pages::reset`). - Update the automatic-migrations docs. Co-Authored-By: Claude Fable 5 --- crates/datastore/src/error.rs | 4 +- .../locking_tx_datastore/committed_state.rs | 2 +- .../src/locking_tx_datastore/datastore.rs | 9 + .../src/locking_tx_datastore/mut_tx.rs | 22 +- .../src/locking_tx_datastore/replay.rs | 8 +- .../src/locking_tx_datastore/tx_state.rs | 11 +- crates/engine/src/relational_db.rs | 11 + crates/engine/src/update.rs | 259 +++++++++++++++++- crates/schema/src/auto_migrate.rs | 184 +++++++++---- crates/schema/src/auto_migrate/formatter.rs | 4 + .../src/auto_migrate/termcolor_formatter.rs | 9 + crates/schema/tests/ensure_same_schema.rs | 1 + crates/table/src/pages.rs | 10 + crates/table/src/table.rs | 3 + .../00200-automatic-migrations.md | 6 +- 15 files changed, 468 insertions(+), 75 deletions(-) diff --git a/crates/datastore/src/error.rs b/crates/datastore/src/error.rs index 912abca4fce..25388f26af2 100644 --- a/crates/datastore/src/error.rs +++ b/crates/datastore/src/error.rs @@ -82,8 +82,8 @@ pub enum TableError { ChangeColumnsError(#[from] Box), #[error(transparent)] AddColumnsError(#[from] Box), - #[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" )] diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs index 0ce629524c4..b40518ec666 100644 --- a/crates/datastore/src/locking_tx_datastore/committed_state.rs +++ b/crates/datastore/src/locking_tx_datastore/committed_state.rs @@ -879,7 +879,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, diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs index 7293a81fba0..549de0bc183 100644 --- a/crates/datastore/src/locking_tx_datastore/datastore.rs +++ b/crates/datastore/src/locking_tx_datastore/datastore.rs @@ -355,6 +355,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, + ) -> Result<()> { + tx.alter_empty_table_row_type(table_id, column_schemas) + } + pub fn add_columns_to_table_mut_tx( &self, tx: &mut MutTxId, diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index fec1d4804dd..cc0bf10e824 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -1587,23 +1587,37 @@ 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, + ) -> 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))?; // Remember the pending change so we can undo it if a later system-table update fails. - self.push_schema_change(PendingSchemaChange::ReschemaEventTable(table_id, old_column_schemas)); + self.push_schema_change(PendingSchemaChange::ReschemaEmptyTable(table_id, old_column_schemas)); // Update system tables. // We'll simply remove all rows in `st_columns` and then add the new ones. diff --git a/crates/datastore/src/locking_tx_datastore/replay.rs b/crates/datastore/src/locking_tx_datastore/replay.rs index c42185e2bb2..3e8c0b5e76b 100644 --- a/crates/datastore/src/locking_tx_datastore/replay.rs +++ b/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)?; } diff --git a/crates/datastore/src/locking_tx_datastore/tx_state.rs b/crates/datastore/src/locking_tx_datastore/tx_state.rs index 1d7680ae05a..81278e1488d 100644 --- a/crates/datastore/src/locking_tx_datastore/tx_state.rs +++ b/crates/datastore/src/locking_tx_datastore/tx_state.rs @@ -134,12 +134,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), - /// 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), + /// 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), /// The primary key of the table with [`TableId`] was changed. /// The old primary key was stored. TableAlterPrimaryKey(TableId, Option), @@ -190,7 +191,7 @@ impl MemoryUsage for PendingSchemaChange { + col_id.heap_usage() + alias.as_ref().map(|a| a.as_raw().heap_usage()).unwrap_or(0) } - 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(), } } } diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 933949b666b..87a23c27db3 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -1122,6 +1122,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, + ) -> 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, diff --git a/crates/engine/src/update.rs b/crates/engine/src/update.rs index 5b9e6e0fde2..1ec051e1299 100644 --- a/crates/engine/src/update.rs +++ b/crates/engine/src/update.rs @@ -257,6 +257,21 @@ fn auto_migrate_database( anyhow::bail!("Precheck failed: added sequence {sequence_name} already has values in range",); } } + spacetimedb_schema::auto_migrate::AutoMigratePrecheck::CheckTableEmpty(table_name_key) => { + let (namespace, local) = table_name_key; + let table_name = joined(namespace, local); + let table_id = stdb + .table_id_from_name_mut(tx, &table_name)? + .ok_or_else(|| anyhow::anyhow!("Precheck: table `{table_name}` not found in database"))?; + let row_count = stdb.table_row_count_mut(tx, table_id).unwrap_or(0); + if row_count > 0 { + anyhow::bail!( + "Precheck failed: table `{table_name}` contains data ({row_count} rows), \ + but this migration requires it to be empty. \ + Clear the table's rows (e.g. via a reducer) before publishing." + ); + } + } } } @@ -270,13 +285,8 @@ fn auto_migrate_database( let table_name = joined(namespace, local); let table_id = stdb.table_id_from_name_mut(tx, &table_name)?.unwrap(); - if stdb.table_row_count_mut(tx, table_id).unwrap_or(0) > 0 { - anyhow::bail!( - "Cannot remove table `{table_name}`: table contains data. \ - Clear the table's rows (e.g. via a reducer) before removing it from your schema." - ); - } - + // Emptiness was already validated by the matching `CheckTableEmpty` precheck, + // before any mutations were performed. log!(logger, "Dropping table `{table_name}`"); stdb.drop_table(tx, table_id)?; } @@ -564,6 +574,21 @@ fn auto_migrate_database( stdb.alter_event_table_row_type(tx, table_id, column_schemas)?; } + spacetimedb_schema::auto_migrate::AutoMigrateStep::ReschemaEmptyTable(table_name_key) => { + let (namespace, local) = table_name_key; + let table_name = joined(namespace, local); + let (owning_def, table_def) = plan.new.find_table(table_name_key).ok_or_else(|| { + anyhow::anyhow!("ReschemaEmptyTable: table `{table_name}` not found in new module def") + })?; + let table_id = stdb.table_id_from_name_mut(tx, &table_name).unwrap().unwrap(); + let column_schemas = column_schemas_from_defs(owning_def, &table_def.columns, table_id); + + log!(logger, "Changing column layout of empty table `{}`", table_name); + + // Emptiness was already validated by the matching `CheckTableEmpty` precheck; + // the datastore re-checks it as a backstop. + stdb.alter_empty_table_row_type(tx, table_id, column_schemas)?; + } spacetimedb_schema::auto_migrate::AutoMigrateStep::ChangeAccess(table_name_key) => { let (namespace, local) = table_name_key; let table_name = joined(namespace, local); @@ -729,6 +754,7 @@ mod test { use crate::relational_db::{ open_snapshot_repo, tests_utils::{begin_mut_tx, insert, TestDB}, + MutTx, }; use spacetimedb_datastore::locking_tx_datastore::PendingSchemaChange; use spacetimedb_datastore::system_tables::ST_EVENT_TABLE_ID; @@ -739,8 +765,11 @@ mod test { }, Identity, }; - use spacetimedb_sats::{product, raw_identifier::RawIdentifier, AlgebraicType, AlgebraicType::U64, ProductType}; - use spacetimedb_schema::auto_migrate::ponder_migrate; + use spacetimedb_primitives::ColId; + use spacetimedb_sats::{ + product, raw_identifier::RawIdentifier, AlgebraicType, AlgebraicType::U64, ProductType, ProductValue, + }; + use spacetimedb_schema::{auto_migrate::ponder_migrate, def::ModuleDef}; struct TestLogger; impl UpdateLogger for TestLogger { @@ -1481,17 +1510,223 @@ mod test { let result = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger); let err = result.err().expect("removing a non-empty table should fail"); assert!( - err.to_string().contains("table contains data"), + err.to_string().contains("contains data"), "error should mention that the table contains data, got: {err}" ); + assert_eq!(tx.pending_schema_changes(), []); + Ok(()) + } + + /// A module with a single table `points` whose columns are `(id, name)`, + /// or `(name, id)` when `swapped`, with a primary key, unique constraint, + /// and index on `id` in both cases. + fn points_module(swapped: bool) -> ModuleDef { + let mut builder = RawModuleDefV9Builder::new(); + let (product_type, id_col) = if swapped { + (ProductType::from([("name", AlgebraicType::String), ("id", U64)]), 1) + } else { + (ProductType::from([("id", U64), ("name", AlgebraicType::String)]), 0) + }; + builder + .build_table_with_new_type("points", product_type, true) + .with_unique_constraint(id_col) + .with_index(btree(id_col), "points_id_idx") + .with_primary_key(id_col) + .with_access(TableAccess::Public) + .finish(); + builder + .finish() + .try_into() + .expect("should be a valid module definition") + } + + /// Creates the tables of `module` in `stdb`, returning the [`TableId`] of `points`. + fn create_points_table(stdb: &TestDB, module: &ModuleDef) -> anyhow::Result { + let mut tx = begin_mut_tx(stdb); + for def in module.tables() { + create_table_from_def(stdb, &mut tx, module, def)?; + } + let table_id = stdb + .table_id_from_name_mut(&tx, "points")? + .expect("`points` table should exist"); + stdb.commit_tx(tx)?; + Ok(table_id) + } + + /// Asserts that the stored schema of `table_id` has exactly the column names + /// `columns`, in order, and the primary key `primary_key`. + fn assert_column_order( + stdb: &TestDB, + tx: &MutTx, + table_id: TableId, + columns: &[&str], + primary_key: Option, + ) -> anyhow::Result<()> { + let schema = stdb.schema_for_table_mut(tx, table_id)?; + let names: Vec<&str> = schema.columns().iter().map(|c| &*c.col_name).collect(); + assert_eq!(names, columns); + assert_eq!(schema.primary_key, primary_key); + Ok(()) + } + + /// Returns all rows of `table_id` as [`ProductValue`]s. + fn collect_rows(stdb: &TestDB, tx: &MutTx, table_id: TableId) -> anyhow::Result> { + Ok(stdb.iter_mut(tx, table_id)?.map(|r| r.to_product_value()).collect()) + } + + #[test] + fn reorder_empty_table_succeeds() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = points_module(false); + let new = points_module(true); + let table_id = create_points_table(&stdb, &old)?; + + // Insert a row and delete it again: a table which previously contained rows + // keeps residual (row-empty) pages, which the reschema must handle. + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "gone"])?; + stdb.commit_tx(tx)?; + let mut tx = begin_mut_tx(&stdb); + assert_eq!(stdb.delete_by_rel(&mut tx, table_id, [product![7u64, "gone"]]), 1); + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + assert!( + matches!(res, UpdateResult::RequiresClientDisconnect), + "reordering columns should disconnect clients" + ); + + // The stored schema now has the new column order, and the sub-objects + // on the moved column were re-created against its new position. + assert_column_order(&stdb, &tx, table_id, &["name", "id"], Some(ColId(1)))?; assert!( - tx.pending_schema_changes().is_empty(), - "failed migration should leave no pending schema changes: {:?}", + matches!( + tx.pending_schema_changes(), + [ + PendingSchemaChange::IndexRemoved(..), + PendingSchemaChange::ConstraintRemoved(..), + PendingSchemaChange::ReschemaEmptyTable(..), + PendingSchemaChange::IndexAdded(..), + PendingSchemaChange::ConstraintAdded(..), + PendingSchemaChange::TableAlterPrimaryKey(..), + ] + ), + "{:?}", tx.pending_schema_changes() ); + stdb.commit_tx(tx)?; + + // The table is usable with the new layout: insert a row and read it back. + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product!["p1", 42u64])?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product!["p1", 42u64]]); + stdb.commit_tx(tx)?; + + Ok(()) + } + + #[test] + fn reorder_nonempty_table_fails() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = points_module(false); + let new = points_module(true); + let table_id = create_points_table(&stdb, &old)?; + + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "p1"])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let result = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger); + let err = result.err().expect("reordering a non-empty table should fail"); + assert!( + err.to_string().contains("contains data"), + "error should mention that the table contains data, got: {err}" + ); + assert_eq!(tx.pending_schema_changes(), []); + + // The table keeps its old layout and contents. + assert_column_order(&stdb, &tx, table_id, &["id", "name"], Some(ColId(0)))?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product![7u64, "p1"]]); + + Ok(()) + } + + /// Reorders the columns of the (empty) `points` table, then replays the commitlog. + /// + /// Prior to the accompanying fix in `replay.rs` (`st_column_changed`), + /// replaying the reorder failed with a layout-compatibility error, + /// as replay used the layout-checked `change_columns_to` for non-event tables. + fn replay_reordered_table(snapshot: TakeSnapshot) -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let with_snapshot = matches!(snapshot, TakeSnapshot::BeforeAutomigration); + let stdb = with_snapshotting(with_snapshot)?; + + let old = points_module(false); + let new = points_module(true); + let table_id = create_points_table(&stdb, &old)?; + + // Insert a row and delete it again, so the commitlog contains writes to the + // table in the old layout, while the table is empty at migration time. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "gone"])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + assert_eq!(stdb.delete_by_rel(&mut tx, table_id, [product![7u64, "gone"]]), 1); + stdb.commit_tx(tx)?; + } + + if with_snapshot { + take_snapshot(&stdb)?; + } + + // Migrate, reordering `points`' columns. + { + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + assert!( + matches!(res, UpdateResult::RequiresClientDisconnect), + "reordering columns should disconnect clients" + ); + stdb.commit_tx(tx)?; + } + + // Insert a row in the new layout. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product!["p1", 42u64])?; + stdb.commit_tx(tx)?; + } + + // Replay the commitlog and verify the reordered schema and its rows survived. + let stdb = stdb.reopen()?; + let tx = begin_mut_tx(&stdb); + assert_column_order(&stdb, &tx, table_id, &["name", "id"], Some(ColId(1)))?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product!["p1", 42u64]]); + Ok(()) } + #[test] + fn replay_reordered_table_no_snapshot() -> anyhow::Result<()> { + replay_reordered_table(TakeSnapshot::None) + } + + #[test] + fn replay_reordered_table_after_snapshot() -> anyhow::Result<()> { + replay_reordered_table(TakeSnapshot::BeforeAutomigration) + } + #[test] fn add_sequence_precheck_rejects_existing_column_max_value() -> anyhow::Result<()> { let auth_ctx = AuthCtx::for_testing(); diff --git a/crates/schema/src/auto_migrate.rs b/crates/schema/src/auto_migrate.rs index 97fff428830..9febe0c80d1 100644 --- a/crates/schema/src/auto_migrate.rs +++ b/crates/schema/src/auto_migrate.rs @@ -244,6 +244,14 @@ pub enum AutoMigratePrecheck<'def> { /// Perform a check that adding a sequence is valid (the relevant column contains no values /// greater than the sequence's start value). CheckAddSequenceRangeValid(::Key<'def>), + + /// Perform a check that the table contains no rows. + /// + /// Emitted for migration steps that are only valid on an empty table, + /// such as [`AutoMigrateStep::RemoveTable`] and [`AutoMigrateStep::ReschemaEmptyTable`]. + /// The planner cannot see table contents, so the check is performed at execution time, + /// before any mutations. + CheckTableEmpty(::Key<'def>), } /// A step in an automatic migration. @@ -285,7 +293,9 @@ pub enum AutoMigrateStep<'def> { RemoveRowLevelSecurity(::Key<'def>), /// Remove an empty table and all its sub-objects (indexes, constraints, sequences). - /// Validated at execution time: fails if the table contains data. + /// + /// Only valid on an empty table; the plan will contain a matching + /// [`AutoMigratePrecheck::CheckTableEmpty`], and execution fails if the table contains data. RemoveTable(::Key<'def>), /// Change the column types of a table, in a layout compatible way. @@ -294,6 +304,13 @@ pub enum AutoMigrateStep<'def> { /// Change the column types of an event table, in a way that may not be layout-compatible. ReschemaEventTable(::Key<'def>), + /// Change the columns of a table, in a way that may not be layout-compatible + /// (e.g. reordering columns). + /// + /// Only valid on an empty table; the plan will contain a matching + /// [`AutoMigratePrecheck::CheckTableEmpty`], and execution fails if the table contains data. + ReschemaEmptyTable(::Key<'def>), + /// Add columns to a table, in a layout-INCOMPATIBLE way. /// /// This is a destructive operation that requires first running a `DisconnectAllUsers`. @@ -360,9 +377,6 @@ pub enum AutoMigrateError { #[error("Removing a column {column} from table {table} requires a manual migration")] RemoveColumn { table: Identifier, column: Identifier }, - #[error("Reordering table {table} requires a manual migration")] - ReorderTable { table: Identifier }, - #[error( "Changing the type of column {} in table {} from {:?} to {:?} requires a manual migration", .0.column, .0.table, .0.type1, .0.type2 @@ -704,6 +718,7 @@ fn auto_migrate_tables<'def>(plan: &mut AutoMigratePlan<'def>) -> Result<()> { for key in old_tables.keys() { if !new_tables.contains_key(key) { + plan.prechecks.push(AutoMigratePrecheck::CheckTableEmpty(*key)); plan.steps.push(AutoMigrateStep::RemoveTable(*key)); plan.ensure_disconnect_all_users(); } @@ -792,14 +807,14 @@ fn auto_migrate_table<'def>( let columns_ok = old .columns .iter() - .map(|old_col| -> Result> { + .map(|old_col| -> Result> { match new_col_by_name.get(&old_col.name) { None => { if is_event { // Event tables never have any resident rows, so removing a column is not a // data migration. However, changing the schema will break clients. - // `row_type_changed`, `columns_added`, `event_schema_changed` - Ok(ArrayMonoid([Any(false), Any(false), Any(true)])) + // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + Ok(ArrayMonoid([Any(false), Any(false), Any(true), Any(false)])) } else { Err(AutoMigrateError::RemoveColumn { table: old_col.table_name.clone(), @@ -830,46 +845,37 @@ fn auto_migrate_table<'def>( Err(err) } }); - // Reject reordering of existing columns (unless it's an event table). - let positions_ok = if old_col.col_id == new_col.col_id { - Ok(Any(false)) - } else if is_event { - Ok(Any(true)) - } else { - Err(AutoMigrateError::ReorderTable { - table: old_col.table_name.clone(), - } - .into()) - }; if old_col.accessor_name != new_col.accessor_name { plan.steps .push(AutoMigrateStep::ChangeColumnAccessorName(key, &old_col.name)); } - (types_ok, positions_ok) - .combine_errors() - // `row_type_changed`, `columns_added`, `event_schema_changed` - .map(|(types_changed, positions_changed)| { + // Reordering existing columns changes the row layout, which is only + // possible when the table has no resident rows. Event tables are + // rowless by construction; other tables get an emptiness precheck. + let positions_changed = Any(old_col.col_id != new_col.col_id); + types_ok + // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + .map(|types_changed| { if is_event { - ArrayMonoid([Any(false), Any(false), types_changed | positions_changed]) + ArrayMonoid([Any(false), Any(false), types_changed | positions_changed, Any(false)]) } else { - assert!(!positions_changed.0); - ArrayMonoid([types_changed, Any(false), Any(false)]) + ArrayMonoid([types_changed, Any(false), Any(false), positions_changed]) } }) } } }) - .chain(new.columns.iter().map(|new_col| -> Result> { + .chain(new.columns.iter().map(|new_col| -> Result> { if old_col_by_name.contains_key(&new_col.name) { - Ok(ArrayMonoid([Any(false), Any(false), Any(false)])) + Ok(ArrayMonoid([Any(false), Any(false), Any(false), Any(false)])) } else if is_event { // Event tables never have any resident rows, so adding a column is not a data // migration. However, changing the schema will break clients. - // `row_type_changed`, `columns_added`, `event_schema_changed` - Ok(ArrayMonoid([Any(false), Any(false), Any(true)])) + // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + Ok(ArrayMonoid([Any(false), Any(false), Any(true), Any(false)])) } else if new_col.default_value.is_some() { - // `row_type_changed`, `columns_added`, `event_schema_changed` - Ok(ArrayMonoid([Any(false), Any(true), Any(false)])) + // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + Ok(ArrayMonoid([Any(false), Any(true), Any(false), Any(false)])) } else { Err(AutoMigrateError::AddColumn { table: new_col.table_name.clone(), @@ -878,16 +884,28 @@ fn auto_migrate_table<'def>( .into()) } })) - .collect_all_errors::>(); + .collect_all_errors::>(); - let ((), (), ArrayMonoid([Any(row_type_changed), Any(columns_added), Any(event_schema_changed)])) = - (type_ok, event_ok, columns_ok).combine_errors()?; + let ( + (), + (), + ArrayMonoid([Any(row_type_changed), Any(columns_added), Any(event_schema_changed), Any(columns_reordered)]), + ) = (type_ok, event_ok, columns_ok).combine_errors()?; if event_schema_changed { // If we're rewriting an event table, there's no data migration to do. // But incompatibly changing the schema can break clients. plan.ensure_disconnect_all_users(); plan.steps.push(AutoMigrateStep::ReschemaEventTable(key)); + } else if columns_reordered { + // Reordering columns rewrites the row layout in place, which is only valid on an + // empty table. The planner cannot see table contents, so emptiness is validated + // at execution time, before any mutations. This subsumes any `ChangeColumns` or + // `AddColumns` for the same table: the reschema rebuilds the full new layout, and + // with no resident rows there is no data to migrate or default-fill. + plan.prechecks.push(AutoMigratePrecheck::CheckTableEmpty(key)); + plan.ensure_disconnect_all_users(); + plan.steps.push(AutoMigrateStep::ReschemaEmptyTable(key)); } else if columns_added { // If we're adding a column, we'll rewrite the whole table. // That makes any `ChangeColumns` moot, so we can skip it. @@ -1185,7 +1203,9 @@ fn auto_migrate_sequences<'def>( // Added or changed sequences. for (sequence_key, (table_key, new_seq)) in &new_seqs { if let Some((_, old_seq)) = old_seqs.get(sequence_key) { - // we do not need to check column ids, since in an automigrate, column ids are not changed. + // Column ids can change in an automigrate (reordering an empty table); since + // `SequenceDef` includes the column id, such a sequence diffs as changed here + // and is removed and re-added against the new column position. if *old_seq != *new_seq { plan.prechecks .push(AutoMigratePrecheck::CheckAddSequenceRangeValid(*sequence_key)); @@ -1243,14 +1263,23 @@ fn auto_migrate_constraints<'def>( } // Changed constraints. - for (constraint_key, (_, new_constraint)) in &new_constraints { + for (constraint_key, (table_key, new_constraint)) in &new_constraints { if let Some((_, old_constraint)) = old_constraints.get(constraint_key) && *old_constraint != *new_constraint { - results.push(Err(AutoMigrateError::ChangeUniqueConstraint { - constraint: old_constraint.name.clone(), + // A constraint on a reordered column keeps its name but changes its column ids. + // When the owning table is being reschema'd empty (`ReschemaEmptyTable`), + // re-adding the constraint against the new column positions is trivially valid, + // as the table contains no rows. + if plan.any_step(|step| matches!(step, AutoMigrateStep::ReschemaEmptyTable(key) if key == table_key)) { + plan.steps.push(AutoMigrateStep::RemoveConstraint(*constraint_key)); + plan.steps.push(AutoMigrateStep::AddConstraint(*constraint_key)); + } else { + results.push(Err(AutoMigrateError::ChangeUniqueConstraint { + constraint: old_constraint.name.clone(), + } + .into())); } - .into())); } } @@ -1784,11 +1813,6 @@ mod tests { } => table == &apples && column == &count ); - expect_error_matching!( - result, - AutoMigrateError::ReorderTable { table } => table == &apples - ); - expect_error_matching!( result, AutoMigrateError::ChangeColumnType(ChangeColumnTypeParts { @@ -1978,11 +2002,77 @@ mod tests { } => &index[..] == apples_id_index && old_accessor.as_ref() == Some(&accessor_old) && new_accessor.as_ref() == Some(&accessor_new) ); - // It is not currently possible to test for `ChangeUniqueConstraint`, because unique constraint names are now generated during validation, - // and are determined by their columns and table name. So it's impossible to create a unique constraint with the same name - // but different columns from an old one. + // It is not currently possible to test for `ChangeUniqueConstraint` on a table that isn't + // being reschema'd empty, because unique constraint names are now generated during validation, + // and are determined by their columns and table name. So it's impossible to create a unique constraint + // with the same name but different columns from an old one. // We've left the check in, just in case this changes in the future. } + + #[test] + fn reorder_columns_of_empty_table() { + fn points_module(swapped: bool) -> ModuleDef { + let mut builder = RawModuleDefV9Builder::new(); + let (product_type, id_col) = if swapped { + ( + ProductType::from([("name", AlgebraicType::String), ("id", AlgebraicType::U64)]), + ColId(1), + ) + } else { + ( + ProductType::from([("id", AlgebraicType::U64), ("name", AlgebraicType::String)]), + ColId(0), + ) + }; + builder + .build_table_with_new_type("Points", product_type, true) + .with_column_sequence(id_col) + .with_unique_constraint(id_col) + .with_index(btree(id_col), "id_index") + .with_primary_key(id_col) + .finish(); + builder.finish().try_into().expect("should be a valid module def") + } + + let old_def = points_module(false); + let new_def = points_module(true); + + let plan = ponder_auto_migrate(&old_def, &new_def).expect("reordering columns should plan an auto-migration"); + + let points = key("", "Points"); + let points_sequence = sub_key("", "Points_id_seq"); + let points_constraint = sub_key("", "Points_id_key"); + let points_index = sub_key("", "Points_id_idx_btree"); + + // Reordering requires the table to be empty, validated before any mutations. + // The re-added sequence also gets its usual range precheck. + assert_eq!( + &plan.prechecks[..], + &[ + AutoMigratePrecheck::CheckAddSequenceRangeValid(points_sequence), + AutoMigratePrecheck::CheckTableEmpty(points), + ], + ); + + // Sub-objects on the moved column are removed and re-added against the new position, + // around the `ReschemaEmptyTable` step. There are no `ChangeColumns`/`AddColumns` steps: + // the full-layout reschema subsumes them. + assert_eq!( + &plan.steps[..], + &[ + AutoMigrateStep::RemoveIndex(points_index), + AutoMigrateStep::RemoveConstraint(points_constraint), + AutoMigrateStep::RemoveSequence(points_sequence), + AutoMigrateStep::ReschemaEmptyTable(points), + AutoMigrateStep::AddIndex(points_index), + AutoMigrateStep::AddConstraint(points_constraint), + AutoMigrateStep::AddSequence(points_sequence), + AutoMigrateStep::ChangePrimaryKey(points), + AutoMigrateStep::DisconnectAllUsers, + ], + ); + } + #[test] fn print_empty_to_populated_schema_migration() { // Start with completely empty schema diff --git a/crates/schema/src/auto_migrate/formatter.rs b/crates/schema/src/auto_migrate/formatter.rs index 7d079f04a62..8d4cdcb5f50 100644 --- a/crates/schema/src/auto_migrate/formatter.rs +++ b/crates/schema/src/auto_migrate/formatter.rs @@ -140,6 +140,8 @@ fn format_step( // TODO(format-event-table-reschema): I (pgoldman 2026-06-10) didn't have time to meaningfully format event table reschemas, // so for now we're just printing the table name. AutoMigrateStep::ReschemaEventTable(table) => f.format_event_table_reschema(&joined(*table)), + + AutoMigrateStep::ReschemaEmptyTable(table) => f.format_empty_table_reschema(&joined(*table)), }?; Ok(()) @@ -207,6 +209,8 @@ pub trait MigrationFormatter { // TODO(format-event-table-reschema): I (pgoldman 2026-06-10) didn't have time to meaningfully format event table reschemas, // so for now we're just printing the table name. fn format_event_table_reschema(&mut self, table_name: &NamespacedIdentifier) -> io::Result<()>; + /// Format a layout-incompatible reschema of an empty (non-event) table, e.g. a column reorder. + fn format_empty_table_reschema(&mut self, table_name: &NamespacedIdentifier) -> io::Result<()>; } #[derive(Debug, Clone, PartialEq)] diff --git a/crates/schema/src/auto_migrate/termcolor_formatter.rs b/crates/schema/src/auto_migrate/termcolor_formatter.rs index 811c04b1860..1d2f3af42b6 100644 --- a/crates/schema/src/auto_migrate/termcolor_formatter.rs +++ b/crates/schema/src/auto_migrate/termcolor_formatter.rs @@ -458,6 +458,15 @@ impl MigrationFormatter for TermColorFormatter { Ok(()) } + + fn format_empty_table_reschema(&mut self, table_name: &NamespacedIdentifier) -> io::Result<()> { + self.write_action_prefix(&Action::Changed)?; + self.buffer.write_all(b" column layout of table ")?; + self.write_colored(table_name, Some(self.colors.table_name), true)?; + self.buffer.write_all(b" (requires the table to be empty)\n")?; + + Ok(()) + } } trait ActionColorExt { diff --git a/crates/schema/tests/ensure_same_schema.rs b/crates/schema/tests/ensure_same_schema.rs index b2427942e88..8efc611bd90 100644 --- a/crates/schema/tests/ensure_same_schema.rs +++ b/crates/schema/tests/ensure_same_schema.rs @@ -22,6 +22,7 @@ fn step_namespace<'a, 'def>(step: &'a AutoMigrateStep<'def>) -> Option<&'a Names | AutoMigrateStep::RemoveTable((ns, _)) | AutoMigrateStep::ChangeColumns((ns, _)) | AutoMigrateStep::ReschemaEventTable((ns, _)) + | AutoMigrateStep::ReschemaEmptyTable((ns, _)) | AutoMigrateStep::AddColumns((ns, _)) | AutoMigrateStep::AddTable((ns, _)) | AutoMigrateStep::AddSchedule((ns, _)) diff --git a/crates/table/src/pages.rs b/crates/table/src/pages.rs index b815b6415e2..31c0491a3f1 100644 --- a/crates/table/src/pages.rs +++ b/crates/table/src/pages.rs @@ -144,6 +144,16 @@ impl Pages { .collect(); } + /// Drops all pages, resetting `self` to its initial empty state. + /// + /// Unlike [`Self::clear`], which empties each page but keeps it allocated, + /// this removes the pages themselves, + /// so that [`Self::set_contents`] can be called afterwards. + pub fn reset(&mut self) { + self.pages.clear(); + self.non_full_pages.clear(); + } + /// Get a reference to fixed-len row data. /// /// Used in benchmarks. diff --git a/crates/table/src/table.rs b/crates/table/src/table.rs index 7fb63a2e429..000a9a5e78e 100644 --- a/crates/table/src/table.rs +++ b/crates/table/src/table.rs @@ -382,6 +382,9 @@ impl Table { } // Remove and drop any pages, as even though they must be empty, // they may have residual layout-derived data which conflicts with the new schema. + // A table which previously contained rows keeps its (now row-empty) pages around, + // so drop them first; `set_pages` requires that no pages are present. + self.inner.pages.reset(); // Safety: there aren't any pages here, so they cannot conflict with the schema or row layout. unsafe { self.set_pages(Vec::new(), &NullBlobStore) }; diff --git a/docs/docs/00200-core-concepts/00100-databases/00500-migrations/00200-automatic-migrations.md b/docs/docs/00200-core-concepts/00100-databases/00500-migrations/00200-automatic-migrations.md index 4a29ff7465f..857ad06eb66 100644 --- a/docs/docs/00200-core-concepts/00100-databases/00500-migrations/00200-automatic-migrations.md +++ b/docs/docs/00200-core-concepts/00100-databases/00500-migrations/00200-automatic-migrations.md @@ -33,6 +33,8 @@ These changes are allowed by automatic migration, but may cause runtime errors f - **Changing or removing reducers.** Clients attempting to call the old version of a changed reducer or a removed reducer will receive runtime errors. - **Changing tables from public to private.** Clients subscribed to a newly-private table will receive runtime errors. - **Removing `Primary Key` annotations.** Non-updated clients will still use the old primary key as a unique key in their local cache, which can result in non-deterministic behavior when updates are received. +- **Removing an empty table.** The publish fails if the table still contains rows; clear the table's rows first (e.g. via a reducer). All clients are disconnected, and non-updated clients subscribed to the removed table will receive runtime errors. +- **Reordering the columns of an empty table.** The publish fails if the table contains rows; clear the table's rows first (e.g. via a reducer). All clients are disconnected, and clients must regenerate their bindings to read the table correctly. - **Removing indexes.** This is only breaking in specific situations. The main issue occurs with subscription queries involving semijoins, such as: ```typescript @@ -48,8 +50,8 @@ These changes are allowed by automatic migration, but may cause runtime errors f The following changes cannot be performed with automatic migration and will cause the publish to fail: -- **Removing tables.** -- **Removing or modifying existing columns.** This includes changing the type, renaming, or reordering columns. +- **Removing tables that contain data.** Empty tables can be removed (see above). +- **Removing or modifying existing columns.** This includes changing the type or renaming columns. Reordering columns is only possible while the table is empty (see above). - **Adding columns without a default value.** New columns must have a default value so existing rows can be populated. - **Adding columns in the middle of a table.** New columns must be added at the end of the table definition. - **Changing whether a table is used for `scheduling`.** From 410d05b3f78a42bb8f92802b969134cf0b1b2dbd Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 22 Aug 2026 03:51:19 +0200 Subject: [PATCH 2/4] feat: Generalize the empty-table reschema to any column or constraint change Per review feedback, stop layering individual empty-table cases: removing columns, renaming columns (which diff as remove+add), layout-incompatible type changes, adding columns without a default, and changed unique constraints now all plan as CheckTableEmpty + ReschemaEmptyTable instead of erroring, exactly as column reorders already did. Emptiness is validated at execution time, before any mutations. Layout-compatible changes keep their existing data-preserving paths (ChangeColumns, AddColumns with defaults), so nothing that previously worked on populated tables changes behavior. The ReschemaEmptyTable executor needs no changes: it already rebuilds the full column set from the new module def. AutoMigrateError::{AddColumn, RemoveColumn, ChangeUniqueConstraint} have no producers left and are removed. --- crates/engine/src/update.rs | 186 +++++++++++++- crates/schema/src/auto_migrate.rs | 411 +++++++++++------------------- 2 files changed, 327 insertions(+), 270 deletions(-) diff --git a/crates/engine/src/update.rs b/crates/engine/src/update.rs index 1ec051e1299..99712c20218 100644 --- a/crates/engine/src/update.rs +++ b/crates/engine/src/update.rs @@ -1540,15 +1540,15 @@ mod test { .expect("should be a valid module definition") } - /// Creates the tables of `module` in `stdb`, returning the [`TableId`] of `points`. - fn create_points_table(stdb: &TestDB, module: &ModuleDef) -> anyhow::Result { + /// Creates the tables of `module` in `stdb`, returning the [`TableId`] of `table_name`. + fn create_table_for_module(stdb: &TestDB, module: &ModuleDef, table_name: &str) -> anyhow::Result { let mut tx = begin_mut_tx(stdb); for def in module.tables() { create_table_from_def(stdb, &mut tx, module, def)?; } let table_id = stdb - .table_id_from_name_mut(&tx, "points")? - .expect("`points` table should exist"); + .table_id_from_name_mut(&tx, table_name)? + .unwrap_or_else(|| panic!("`{table_name}` table should exist")); stdb.commit_tx(tx)?; Ok(table_id) } @@ -1581,7 +1581,7 @@ mod test { let old = points_module(false); let new = points_module(true); - let table_id = create_points_table(&stdb, &old)?; + let table_id = create_table_for_module(&stdb, &old, "points")?; // Insert a row and delete it again: a table which previously contained rows // keeps residual (row-empty) pages, which the reschema must handle. @@ -1636,7 +1636,7 @@ mod test { let old = points_module(false); let new = points_module(true); - let table_id = create_points_table(&stdb, &old)?; + let table_id = create_table_for_module(&stdb, &old, "points")?; let mut tx = begin_mut_tx(&stdb); insert(&stdb, &mut tx, table_id, &product![7u64, "p1"])?; @@ -1671,7 +1671,7 @@ mod test { let old = points_module(false); let new = points_module(true); - let table_id = create_points_table(&stdb, &old)?; + let table_id = create_table_for_module(&stdb, &old, "points")?; // Insert a row and delete it again, so the commitlog contains writes to the // table in the old layout, while the table is empty at migration time. @@ -1727,6 +1727,178 @@ mod test { replay_reordered_table(TakeSnapshot::BeforeAutomigration) } + /// A module whose `fruits` table changes shape between versions: + /// v1 = (id: U64, name: String, count: U16); v2 = (id: U64, label: String, weight: U32). + /// Relative to v1, v2 removes `count`, renames `name` to `label` (remove+add), + /// and adds `weight` without a default -- all only valid on an empty table. + fn fruits_module(v2: bool) -> ModuleDef { + let mut builder = RawModuleDefV9Builder::new(); + let product_type = if v2 { + ProductType::from([ + ("id", U64), + ("label", AlgebraicType::String), + ("weight", AlgebraicType::U32), + ]) + } else { + ProductType::from([ + ("id", U64), + ("name", AlgebraicType::String), + ("count", AlgebraicType::U16), + ]) + }; + builder + .build_table_with_new_type("fruits", product_type, true) + .with_unique_constraint(0) + .with_index(btree(0), "fruits_id_idx") + .with_primary_key(0) + .with_access(TableAccess::Public) + .finish(); + builder + .finish() + .try_into() + .expect("should be a valid module definition") + } + + #[test] + fn general_reschema_empty_table_succeeds() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = fruits_module(false); + let new = fruits_module(true); + let table_id = create_table_for_module(&stdb, &old, "fruits")?; + + // Insert a row and delete it again: a table which previously contained rows + // keeps residual (row-empty) pages, which the reschema must handle. + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "gone", 1u16])?; + stdb.commit_tx(tx)?; + let mut tx = begin_mut_tx(&stdb); + assert_eq!(stdb.delete_by_rel(&mut tx, table_id, [product![7u64, "gone", 1u16]]), 1); + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + assert!( + matches!(res, UpdateResult::RequiresClientDisconnect), + "an empty-table reschema should disconnect clients" + ); + + // The stored schema now has the new columns; the sub-objects on the + // unchanged `id` column survive, so the reschema is the only pending change. + assert_column_order(&stdb, &tx, table_id, &["id", "label", "weight"], Some(ColId(0)))?; + assert!( + matches!(tx.pending_schema_changes(), [PendingSchemaChange::ReschemaEmptyTable(..)]), + "{:?}", + tx.pending_schema_changes() + ); + stdb.commit_tx(tx)?; + + // The table is usable with the new layout: insert a row and read it back. + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![42u64, "f1", 9u32])?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product![42u64, "f1", 9u32]]); + stdb.commit_tx(tx)?; + + Ok(()) + } + + #[test] + fn general_reschema_nonempty_table_fails() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = fruits_module(false); + let new = fruits_module(true); + let table_id = create_table_for_module(&stdb, &old, "fruits")?; + + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "f1", 1u16])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let result = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger); + let err = result.err().expect("reschemaing a non-empty table should fail"); + assert!( + err.to_string().contains("contains data"), + "error should mention that the table contains data, got: {err}" + ); + assert_eq!(tx.pending_schema_changes(), []); + + // The table keeps its old layout and contents. + assert_column_order(&stdb, &tx, table_id, &["id", "name", "count"], Some(ColId(0)))?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product![7u64, "f1", 1u16]]); + + Ok(()) + } + + /// Applies the general reschema (remove + rename + add-without-default) to the + /// (empty) `fruits` table, then replays the commitlog. + fn replay_generally_reschemaed_table(snapshot: TakeSnapshot) -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let with_snapshot = matches!(snapshot, TakeSnapshot::BeforeAutomigration); + let stdb = with_snapshotting(with_snapshot)?; + + let old = fruits_module(false); + let new = fruits_module(true); + let table_id = create_table_for_module(&stdb, &old, "fruits")?; + + // Insert a row and delete it again, so the commitlog contains writes to the + // table in the old layout, while the table is empty at migration time. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64, "gone", 1u16])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + assert_eq!(stdb.delete_by_rel(&mut tx, table_id, [product![7u64, "gone", 1u16]]), 1); + stdb.commit_tx(tx)?; + } + + if with_snapshot { + take_snapshot(&stdb)?; + } + + // Migrate `fruits` to the v2 shape. + { + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + assert!( + matches!(res, UpdateResult::RequiresClientDisconnect), + "an empty-table reschema should disconnect clients" + ); + stdb.commit_tx(tx)?; + } + + // Insert a row in the new layout. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![42u64, "f1", 9u32])?; + stdb.commit_tx(tx)?; + } + + // Replay the commitlog and verify the new schema and its rows survived. + let stdb = stdb.reopen()?; + let tx = begin_mut_tx(&stdb); + assert_column_order(&stdb, &tx, table_id, &["id", "label", "weight"], Some(ColId(0)))?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product![42u64, "f1", 9u32]]); + + Ok(()) + } + + #[test] + fn replay_generally_reschemaed_table_no_snapshot() -> anyhow::Result<()> { + replay_generally_reschemaed_table(TakeSnapshot::None) + } + + #[test] + fn replay_generally_reschemaed_table_after_snapshot() -> anyhow::Result<()> { + replay_generally_reschemaed_table(TakeSnapshot::BeforeAutomigration) + } + #[test] fn add_sequence_precheck_rejects_existing_column_max_value() -> anyhow::Result<()> { let auth_ctx = AuthCtx::for_testing(); diff --git a/crates/schema/src/auto_migrate.rs b/crates/schema/src/auto_migrate.rs index 9febe0c80d1..3c528aff51c 100644 --- a/crates/schema/src/auto_migrate.rs +++ b/crates/schema/src/auto_migrate.rs @@ -219,7 +219,7 @@ pub struct AutoMigratePlan<'def> { pub steps: Vec>, } -impl AutoMigratePlan<'_> { +impl<'def> AutoMigratePlan<'def> { fn any_step(&self, f: impl Fn(&AutoMigrateStep<'_>) -> bool) -> bool { self.steps.iter().any(f) } @@ -235,6 +235,15 @@ impl AutoMigratePlan<'_> { self.steps.push(AutoMigrateStep::DisconnectAllUsers); } } + + /// Ensures that a [`AutoMigratePrecheck::CheckTableEmpty`] for `table` is present in the plan. + /// If it's already there, this is a no-op. + fn ensure_check_table_empty(&mut self, table: ::Key<'def>) { + let check = AutoMigratePrecheck::CheckTableEmpty(table); + if !self.prechecks.contains(&check) { + self.prechecks.push(check); + } + } } /// Checks that must be performed before performing an automatic migration. @@ -371,12 +380,6 @@ pub struct ChangeColumnTypeParts { /// Something that might prevent an automatic migration. #[derive(thiserror::Error, Debug, PartialEq, Eq, PartialOrd, Ord)] pub enum AutoMigrateError { - #[error("Adding a column {column} to table {table} requires a default value annotation")] - AddColumn { table: Identifier, column: Identifier }, - - #[error("Removing a column {column} from table {table} requires a manual migration")] - RemoveColumn { table: Identifier, column: Identifier }, - #[error( "Changing the type of column {} in table {} from {:?} to {:?} requires a manual migration", .0.column, .0.table, .0.type1, .0.type2 @@ -461,12 +464,6 @@ pub enum AutoMigrateError { )] ChangeWithinColumnTypeRenamedField(ChangeColumnTypeParts), - #[error("Adding a unique constraint {constraint} requires a manual migration")] - AddUniqueConstraint { constraint: RawIdentifier }, - - #[error("Changing a unique constraint {constraint} requires a manual migration")] - ChangeUniqueConstraint { constraint: RawIdentifier }, - #[error("Changing the table type of table {table} from {type1:?} to {type2:?} requires a manual migration")] ChangeTableType { table: Identifier, @@ -718,7 +715,7 @@ fn auto_migrate_tables<'def>(plan: &mut AutoMigratePlan<'def>) -> Result<()> { for key in old_tables.keys() { if !new_tables.contains_key(key) { - plan.prechecks.push(AutoMigratePrecheck::CheckTableEmpty(*key)); + plan.ensure_check_table_empty(*key); plan.steps.push(AutoMigrateStep::RemoveTable(*key)); plan.ensure_disconnect_all_users(); } @@ -813,14 +810,13 @@ fn auto_migrate_table<'def>( if is_event { // Event tables never have any resident rows, so removing a column is not a // data migration. However, changing the schema will break clients. - // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` Ok(ArrayMonoid([Any(false), Any(false), Any(true), Any(false)])) } else { - Err(AutoMigrateError::RemoveColumn { - table: old_col.table_name.clone(), - column: old_col.name.clone(), - } - .into()) + // Removing a column is a data migration on a table with rows, but is + // fine on an empty one; emptiness is validated at execution time. + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` + Ok(ArrayMonoid([Any(false), Any(false), Any(false), Any(true)])) } } Some(new_col) => { @@ -830,21 +826,19 @@ fn auto_migrate_table<'def>( let new_ty = WithTypespace::new(new_owning.typespace(), &new_col.ty) .resolve_refs() .expect("valid TableDef must have valid type refs"); - let types_ok = ensure_old_ty_upgradable_to_new( + // `Ok(changed)` = the type change (if any) is layout-compatible and valid + // on a table with resident rows; `Err` = it would require rewriting rows, + // which is only possible when the table is empty. + let (types_changed, types_incompatible) = match ensure_old_ty_upgradable_to_new( false, &|| old_col.table_name.clone(), &|| old_col.name.clone(), &old_ty, &new_ty, - ) - .or_else(|err| { - if is_event { - // Event tables have no rows, so layout-incompatible type changes are fine. - Ok(Any(true)) - } else { - Err(err) - } - }); + ) { + Ok(changed) => (changed, Any(false)), + Err(_) => (Any(false), Any(true)), + }; if old_col.accessor_name != new_col.accessor_name { plan.steps .push(AutoMigrateStep::ChangeColumnAccessorName(key, &old_col.name)); @@ -853,15 +847,22 @@ fn auto_migrate_table<'def>( // possible when the table has no resident rows. Event tables are // rowless by construction; other tables get an emptiness precheck. let positions_changed = Any(old_col.col_id != new_col.col_id); - types_ok - // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` - .map(|types_changed| { - if is_event { - ArrayMonoid([Any(false), Any(false), types_changed | positions_changed, Any(false)]) - } else { - ArrayMonoid([types_changed, Any(false), Any(false), positions_changed]) - } - }) + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` + Ok(if is_event { + ArrayMonoid([ + Any(false), + Any(false), + types_changed | types_incompatible | positions_changed, + Any(false), + ]) + } else { + ArrayMonoid([ + types_changed, + Any(false), + Any(false), + types_incompatible | positions_changed, + ]) + }) } } }) @@ -871,17 +872,16 @@ fn auto_migrate_table<'def>( } else if is_event { // Event tables never have any resident rows, so adding a column is not a data // migration. However, changing the schema will break clients. - // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` Ok(ArrayMonoid([Any(false), Any(false), Any(true), Any(false)])) } else if new_col.default_value.is_some() { - // `row_type_changed`, `columns_added`, `event_schema_changed`, `columns_reordered` + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` Ok(ArrayMonoid([Any(false), Any(true), Any(false), Any(false)])) } else { - Err(AutoMigrateError::AddColumn { - table: new_col.table_name.clone(), - column: new_col.name.clone(), - } - .into()) + // Adding a column without a default value is only possible when there are + // no resident rows to fill; emptiness is validated at execution time. + // `row_type_changed`, `columns_added`, `event_schema_changed`, `needs_empty_reschema` + Ok(ArrayMonoid([Any(false), Any(false), Any(false), Any(true)])) } })) .collect_all_errors::>(); @@ -889,7 +889,7 @@ fn auto_migrate_table<'def>( let ( (), (), - ArrayMonoid([Any(row_type_changed), Any(columns_added), Any(event_schema_changed), Any(columns_reordered)]), + ArrayMonoid([Any(row_type_changed), Any(columns_added), Any(event_schema_changed), Any(needs_empty_reschema)]), ) = (type_ok, event_ok, columns_ok).combine_errors()?; if event_schema_changed { @@ -897,13 +897,15 @@ fn auto_migrate_table<'def>( // But incompatibly changing the schema can break clients. plan.ensure_disconnect_all_users(); plan.steps.push(AutoMigrateStep::ReschemaEventTable(key)); - } else if columns_reordered { - // Reordering columns rewrites the row layout in place, which is only valid on an - // empty table. The planner cannot see table contents, so emptiness is validated - // at execution time, before any mutations. This subsumes any `ChangeColumns` or - // `AddColumns` for the same table: the reschema rebuilds the full new layout, and - // with no resident rows there is no data to migrate or default-fill. - plan.prechecks.push(AutoMigratePrecheck::CheckTableEmpty(key)); + } else if needs_empty_reschema { + // A layout-incompatible change (reordering, removing, renaming, or incompatibly + // retyping columns, or adding a column without a default) rewrites the row layout + // in place, which is only valid on an empty table. The planner cannot see table + // contents, so emptiness is validated at execution time, before any mutations. + // This subsumes any `ChangeColumns` or `AddColumns` for the same table: the + // reschema rebuilds the full new layout, and with no resident rows there is no + // data to migrate or default-fill. + plan.ensure_check_table_empty(key); plan.ensure_disconnect_all_users(); plan.steps.push(AutoMigrateStep::ReschemaEmptyTable(key)); } else if columns_added { @@ -1244,8 +1246,6 @@ fn auto_migrate_constraints<'def>( let old_constraints = constraint_map(old_module); let new_constraints = constraint_map(new_module); - let mut results: Vec> = vec![]; - // Added constraints. for (constraint_key, (table_key, _new_constraint)) in &new_constraints { if !old_constraints.contains_key(constraint_key) && !new_tables.contains(table_key) { @@ -1267,23 +1267,17 @@ fn auto_migrate_constraints<'def>( if let Some((_, old_constraint)) = old_constraints.get(constraint_key) && *old_constraint != *new_constraint { - // A constraint on a reordered column keeps its name but changes its column ids. - // When the owning table is being reschema'd empty (`ReschemaEmptyTable`), - // re-adding the constraint against the new column positions is trivially valid, - // as the table contains no rows. - if plan.any_step(|step| matches!(step, AutoMigrateStep::ReschemaEmptyTable(key) if key == table_key)) { - plan.steps.push(AutoMigrateStep::RemoveConstraint(*constraint_key)); - plan.steps.push(AutoMigrateStep::AddConstraint(*constraint_key)); - } else { - results.push(Err(AutoMigrateError::ChangeUniqueConstraint { - constraint: old_constraint.name.clone(), - } - .into())); - } + // A changed unique constraint (e.g. one on a reordered column, which keeps its + // name but changes its column ids) is removed and re-added against the new + // definition. Re-adding is trivially valid on a table with no rows; emptiness + // is validated at execution time by the precheck. + plan.ensure_check_table_empty(*table_key); + plan.steps.push(AutoMigrateStep::RemoveConstraint(*constraint_key)); + plan.steps.push(AutoMigrateStep::AddConstraint(*constraint_key)); } } - results.into_iter().collect_all_errors::>().map(|_| ()) + Ok(()) } // Because we can refer to many tables and fields on the row level-security query, we need to remove all of them, @@ -1721,7 +1715,6 @@ mod tests { .finish() .try_into() .expect("old_def should be a valid database definition"); - let resolve_old = |ty| old_def.typespace().with_type(ty).resolve_refs().unwrap(); let mut new_builder = RawModuleDefV9Builder::new(); @@ -1783,204 +1776,16 @@ mod tests { .finish() .try_into() .expect("new_def should be a valid database definition"); - let resolve_new = |ty| new_def.typespace().with_type(ty).resolve_refs().unwrap(); let result = ponder_auto_migrate(&old_def, &new_def); let apples = expect_identifier("Apples"); let _bananas = expect_identifier("Bananas"); - let weight = expect_identifier("weight"); - let count = expect_identifier("count"); - let name = expect_identifier("name"); - let sum1 = expect_identifier("sum1"); - let prod1 = expect_identifier("prod1"); - - expect_error_matching!( - result, - // This is an error because we didn't set a default value. - AutoMigrateError::AddColumn { - table, - column - } => table == &apples && column == &weight - ); - - expect_error_matching!( - result, - AutoMigrateError::RemoveColumn { - table, - column - } => table == &apples && column == &count - ); - - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnType(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &name && type1.0 == AlgebraicType::String && type2.0 == AlgebraicType::U32 - ); - - // Rename variant `foo21`. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeRenamedVariant(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == foo2_ty && type2.0 == new_foo2_ty - ); - - // foo22: U32 -> U64. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnType(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == AlgebraicType::U32 && type2.0 == AlgebraicType::U64 - ); - - // Remove variant `foo23`. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeFewerVariants(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == foo2_ty && type2.0 == new_foo2_ty - ); - - // Size of inner sum changed. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeSizeMismatch(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == foo2_ty && type2.0 == new_foo2_ty - ); - - // Align of inner sum changed. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeAlignMismatch(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == foo2_ty && type2.0 == new_foo2_ty - ); - - // Rename field `foo1`. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeRenamedField(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&foo_ty) && type2.0 == resolve_new(&new_foo_ty) - ); - - // Remove field `foo3`. - expect_error_matching!( - result, - AutoMigrateError::ChangeWithinColumnTypeFewerFields(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&foo_ty) && type2.0 == resolve_new(&new_foo_ty) - ); - - // Rename variant `bar`. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeRenamedVariant(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&sum1_ty) && type2.0 == resolve_new(&new_sum1_ty) - ); - - // Remove variant `bar`. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeFewerVariants(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&sum1_ty) && type2.0 == resolve_new(&new_sum1_ty) - ); - - // Size of outer sum changed. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeSizeMismatch(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&sum1_ty) && type2.0 == resolve_new(&new_sum1_ty) - ); - - // Align of outer sum changed. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeAlignMismatch(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &sum1 - && type1.0 == resolve_old(&sum1_ty) && type2.0 == resolve_new(&new_sum1_ty) - ); - - // Rename field `baz`. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeRenamedField(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &prod1 - && type1.0 == prod1_ty && type2.0 == new_prod1_ty - ); - - // Remove field `qux`. - expect_error_matching!( - result, - AutoMigrateError::ChangeColumnTypeFewerFields(ChangeColumnTypeParts { - table, - column, - type1, - type2 - }) => table == &apples && column == &prod1 - && type1.0 == prod1_ty && type2.0 == new_prod1_ty - ); - - // Note: `AddUniqueConstraint` is no longer an error — adding unique constraints - // to existing tables is now allowed; duplicate detection happens inside create_constraint. + // Note: column additions without defaults, column removals, incompatible column + // type changes, and column reorders are no longer plan-time errors — they are + // planned as an empty-table reschema, with emptiness validated at execution time. + // See `general_schema_changes_of_empty_table` and friends below. expect_error_matching!( result, @@ -2073,6 +1878,86 @@ mod tests { ); } + #[test] + fn general_schema_changes_of_empty_table() { + // Removing a column, incompatibly retyping a column, and adding a column without + // a default, all at once, plan as a single empty-table reschema. + let old_def: ModuleDef = { + let mut builder = RawModuleDefV9Builder::new(); + builder + .build_table_with_new_type( + "Apples", + ProductType::from([ + ("id", AlgebraicType::U64), + ("name", AlgebraicType::String), + ("count", AlgebraicType::U16), + ]), + true, + ) + .finish(); + builder.finish().try_into().expect("should be a valid module def") + }; + let new_def: ModuleDef = { + let mut builder = RawModuleDefV9Builder::new(); + builder + .build_table_with_new_type( + "Apples", + ProductType::from([ + ("id", AlgebraicType::U64), + ("name", AlgebraicType::U32), // incompatible type change + ("weight", AlgebraicType::U16), // added without a default + // `count` removed + ]), + true, + ) + .finish(); + builder.finish().try_into().expect("should be a valid module def") + }; + + let plan = ponder_auto_migrate(&old_def, &new_def) + .expect("schema changes to a (presumed empty) table should plan an auto-migration"); + + let apples = key("", "Apples"); + assert_eq!(&plan.prechecks[..], &[AutoMigratePrecheck::CheckTableEmpty(apples)]); + assert_eq!( + &plan.steps[..], + &[ + AutoMigrateStep::ReschemaEmptyTable(apples), + AutoMigrateStep::DisconnectAllUsers, + ], + ); + } + + #[test] + fn rename_column_of_empty_table() { + // A renamed column diffs as remove+add, which plans as an empty-table reschema. + let build = |col_name: &'static str| -> ModuleDef { + let mut builder = RawModuleDefV9Builder::new(); + builder + .build_table_with_new_type( + "Points", + ProductType::from([("id", AlgebraicType::U64), (col_name, AlgebraicType::String)]), + true, + ) + .finish(); + builder.finish().try_into().expect("should be a valid module def") + }; + let old_def = build("name"); + let new_def = build("label"); + + let plan = ponder_auto_migrate(&old_def, &new_def).expect("renaming a column of an empty table should plan"); + + let points = key("", "Points"); + assert_eq!(&plan.prechecks[..], &[AutoMigratePrecheck::CheckTableEmpty(points)]); + assert_eq!( + &plan.steps[..], + &[ + AutoMigrateStep::ReschemaEmptyTable(points), + AutoMigrateStep::DisconnectAllUsers, + ], + ); + } + #[test] fn print_empty_to_populated_schema_migration() { // Start with completely empty schema From 4a96c05f6b169567135e7c070a39a87f492312a2 Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 22 Aug 2026 03:53:14 +0200 Subject: [PATCH 3/4] refactor: Drop the now-unproduced column-type migration diagnostics With the table diff routing layout-incompatible type changes to the empty-table reschema, the detailed AutoMigrateError::ChangeColumnType* variants have no producers left: the view diff already discards the upgradability result. Remove the 14 variants and ChangeColumnTypeParts, and simplify ensure_old_ty_upgradable_to_new to return Option (None = the change requires an empty table). The compatibility rules themselves are unchanged. --- crates/schema/src/auto_migrate.rs | 276 ++++++------------------------ 1 file changed, 54 insertions(+), 222 deletions(-) diff --git a/crates/schema/src/auto_migrate.rs b/crates/schema/src/auto_migrate.rs index 3c528aff51c..121cc0fb0ea 100644 --- a/crates/schema/src/auto_migrate.rs +++ b/crates/schema/src/auto_migrate.rs @@ -2,7 +2,6 @@ use core::{cmp::Ordering, ops::BitOr}; use crate::{ def::*, - error::PrettyAlgebraicType, identifier::{Identifier, NamespacePath}, }; use formatter::format_plan; @@ -369,101 +368,9 @@ pub enum AutoMigrateStep<'def> { DisconnectAllUsers, } -#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct ChangeColumnTypeParts { - pub table: Identifier, - pub column: Identifier, - pub type1: PrettyAlgebraicType, - pub type2: PrettyAlgebraicType, -} - /// Something that might prevent an automatic migration. #[derive(thiserror::Error, Debug, PartialEq, Eq, PartialOrd, Ord)] pub enum AutoMigrateError { - #[error( - "Changing the type of column {} in table {} from {:?} to {:?} requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnType(ChangeColumnTypeParts), - - #[error( - "Changing a type within column {} in table {} from {:?} to {:?} requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnType(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with fewer variants, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeFewerVariants(ChangeColumnTypeParts), - - #[error( - "Changing a type within column {} in table {} from {:?} to {:?}, with fewer variants, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeFewerVariants(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with a renamed variant, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeRenamedVariant(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with a renamed variant, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeRenamedVariant(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, requires a manual migration, due to size mismatch", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeSizeMismatch(ChangeColumnTypeParts), - - #[error( - "Changing a type within column {} in table {} from {:?} to {:?}, requires a manual migration, due to size mismatch", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeSizeMismatch(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, requires a manual migration, due to alignment mismatch", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeAlignMismatch(ChangeColumnTypeParts), - - #[error( - "Changing a type within column {} in table {} from {:?} to {:?}, requires a manual migration, due to alignment mismatch", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeAlignMismatch(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with fewer fields, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeFewerFields(ChangeColumnTypeParts), - - #[error( - "Changing a type within column {} in table {} from {:?} to {:?}, with fewer fields, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeFewerFields(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with a renamed field, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeColumnTypeRenamedField(ChangeColumnTypeParts), - - #[error( - "Changing the type of column {} in table {} from {:?} to {:?}, with a renamed field, requires a manual migration", - .0.column, .0.table, .0.type1, .0.type2 - )] - ChangeWithinColumnTypeRenamedField(ChangeColumnTypeParts), - #[error("Changing the table type of table {table} from {type1:?} to {type2:?} requires a manual migration")] ChangeTableType { table: Identifier, @@ -618,9 +525,6 @@ fn auto_migrate_view<'def>( return Any(true); } ensure_old_ty_upgradable_to_new( - false, - &|| old_col.view_name.clone(), - &|| old_col.name.clone(), &WithTypespace::new(old_owning.typespace(), &old_col.ty) .resolve_refs() .expect("valid ViewDefs must have valid type refs"), @@ -653,9 +557,6 @@ fn auto_migrate_view<'def>( return Any(true); } ensure_old_ty_upgradable_to_new( - false, - &|| old_col.view_name.clone(), - &|| old_col.name.clone(), &WithTypespace::new(old_owning.typespace(), &old_col.ty) .resolve_refs() .expect("valid ViewDefs must have valid type refs"), @@ -826,18 +727,12 @@ fn auto_migrate_table<'def>( let new_ty = WithTypespace::new(new_owning.typespace(), &new_col.ty) .resolve_refs() .expect("valid TableDef must have valid type refs"); - // `Ok(changed)` = the type change (if any) is layout-compatible and valid - // on a table with resident rows; `Err` = it would require rewriting rows, - // which is only possible when the table is empty. - let (types_changed, types_incompatible) = match ensure_old_ty_upgradable_to_new( - false, - &|| old_col.table_name.clone(), - &|| old_col.name.clone(), - &old_ty, - &new_ty, - ) { - Ok(changed) => (changed, Any(false)), - Err(_) => (Any(false), Any(true)), + // `Some(changed)` = the type change (if any) is layout-compatible and + // valid on a table with resident rows; `None` = it would require rewriting + // rows, which is only possible when the table is empty. + let (types_changed, types_incompatible) = match ensure_old_ty_upgradable_to_new(&old_ty, &new_ty) { + Some(changed) => (changed, Any(false)), + None => (Any(false), Any(true)), }; if old_col.accessor_name != new_col.accessor_name { plan.steps @@ -970,26 +865,14 @@ where } } -fn ensure_old_ty_upgradable_to_new( - within: bool, - old_container_name: &impl Fn() -> Identifier, - old_column_name: &impl Fn() -> Identifier, - old_ty: &AlgebraicType, - new_ty: &AlgebraicType, -) -> Result { - use AutoMigrateError::*; - // Ensures an `old_ty` within `old` is upgradable to `new_ty`. - let ensure = - |(old_ty, new_ty)| ensure_old_ty_upgradable_to_new(true, old_container_name, old_column_name, old_ty, new_ty); - - // Returns a `ChangeColumnTypeParts` error using the current `old_ty` and `new_ty`. - let parts_for_error = || ChangeColumnTypeParts { - table: old_container_name(), - column: old_column_name(), - type1: old_ty.clone().into(), - type2: new_ty.clone().into(), - }; - +/// Is `old_ty` upgradable to `new_ty` without rewriting existing rows? +/// +/// Returns `Some(Any(changed))` when the change (if any) is layout-compatible +/// and valid on a table with resident rows +/// (`changed` is whether the types differ at all, e.g. a sum type gained variants), +/// and `None` when upgrading would require rewriting rows, +/// which is only possible when the table is empty. +fn ensure_old_ty_upgradable_to_new(old_ty: &AlgebraicType, new_ty: &AlgebraicType) -> Option { match (old_ty, new_ty) { // For sums, we allow the variants in `old_ty` to be a prefix of `new_ty`. (AlgebraicType::Sum(old_ty), AlgebraicType::Sum(new_ty)) => { @@ -997,113 +880,62 @@ fn ensure_old_ty_upgradable_to_new( let new_vars = &*new_ty.variants; // The number of variants in `new_ty` cannot decrease. - let var_lens_ok = match old_vars.len().cmp(&new_vars.len()) { - Ordering::Less => Ok(Any(true)), - Ordering::Equal => Ok(Any(false)), - Ordering::Greater if within => Err(ChangeWithinColumnTypeFewerVariants(parts_for_error()).into()), - Ordering::Greater => Err(ChangeColumnTypeFewerVariants(parts_for_error()).into()), + let len_changed = match old_vars.len().cmp(&new_vars.len()) { + Ordering::Less => Any(true), + Ordering::Equal => Any(false), + Ordering::Greater => return None, }; - // The variants in `old_ty` must be upgradable to those in `old_ty`. + // The variants in `old_ty` must be upgradable to those in `new_ty`, + // and their names must not change. // Strict equality is *not* imposed in the prefix! - let prefix_ok = old_vars - .iter() - .zip(new_vars) - .map(|(o, n)| { - // Ensure type compatibility. - let res_ty = ensure((&o.algebraic_type, &n.algebraic_type)); - // Ensure name doesn't change. - let res_name = if o.name() == n.name() { - Ok(()) - } else if within { - Err(ChangeWithinColumnTypeRenamedVariant(parts_for_error()).into()) - } else { - Err(ChangeColumnTypeRenamedVariant(parts_for_error()).into()) - }; - (res_ty, res_name).combine_errors().map(|(c, ())| c) - }) - .collect_all_errors::(); + let mut prefix_changed = Any(false); + for (o, n) in old_vars.iter().zip(new_vars) { + if o.name() != n.name() { + return None; + } + prefix_changed = + prefix_changed | ensure_old_ty_upgradable_to_new(&o.algebraic_type, &n.algebraic_type)?; + } // The old and the new sum types must have matching layout sizes and alignments. - let old_ty = SumTypeLayout::from(old_ty.clone()); - let new_ty = SumTypeLayout::from(new_ty.clone()); - let old_layout = old_ty.layout(); - let new_layout = new_ty.layout(); - let size_ok = if old_layout.size == new_layout.size { - Ok(()) - } else if within { - Err(ChangeWithinColumnTypeSizeMismatch(parts_for_error()).into()) - } else { - Err(ChangeColumnTypeSizeMismatch(parts_for_error()).into()) - }; - let align_ok = if old_layout.align == new_layout.align { - Ok(()) - } else if within { - Err(ChangeWithinColumnTypeAlignMismatch(parts_for_error()).into()) - } else { - Err(ChangeColumnTypeAlignMismatch(parts_for_error()).into()) - }; + let old_layout_ty = SumTypeLayout::from(old_ty.clone()); + let new_layout_ty = SumTypeLayout::from(new_ty.clone()); + let old_layout = old_layout_ty.layout(); + let new_layout = new_layout_ty.layout(); + if old_layout.size != new_layout.size || old_layout.align != new_layout.align { + return None; + } - let (len_changed, prefix_changed, ..) = (var_lens_ok, prefix_ok, size_ok, align_ok).combine_errors()?; - Ok(len_changed | prefix_changed) + Some(len_changed | prefix_changed) } // For products, // we need to check each field's upgradability due to sums, - // and there must be as many fields. - // Note that we don't care about field names. + // there must be as many fields, and their names must not change. (AlgebraicType::Product(old_ty), AlgebraicType::Product(new_ty)) => { - // The number of variants in `new_ty` cannot decrease. - let len_eq_ok = if old_ty.len() == new_ty.len() { - Ok(()) - } else { - Err(if within { - ChangeWithinColumnTypeFewerFields(parts_for_error()) - } else { - ChangeColumnTypeFewerFields(parts_for_error()) - } - .into()) - }; - - // The fields in `old_ty` must be upgradable to those in `old_ty`. - let fields_ok = old_ty - .iter() - .zip(new_ty.iter()) - .map(|(o, n)| { - // Ensure type compatibility. - let res_ty = ensure((&o.algebraic_type, &n.algebraic_type)); - // Ensure name doesn't change. - let res_name = if o.name() == n.name() { - Ok(()) - } else if within { - Err(ChangeWithinColumnTypeRenamedField(parts_for_error()).into()) - } else { - Err(ChangeColumnTypeRenamedField(parts_for_error()).into()) - }; - (res_ty, res_name).combine_errors().map(|(c, ())| c) - }) - .collect_all_errors::(); + if old_ty.len() != new_ty.len() { + return None; + } - (len_eq_ok, fields_ok).combine_errors().map(|(_, x)| x) + let mut changed = Any(false); + for (o, n) in old_ty.iter().zip(new_ty.iter()) { + if o.name() != n.name() { + return None; + } + changed = changed | ensure_old_ty_upgradable_to_new(&o.algebraic_type, &n.algebraic_type)?; + } + Some(changed) } - // For arrays, we need to check each field's upgradability due to sums. - (AlgebraicType::Array(old_ty), AlgebraicType::Array(new_ty)) => ensure_old_ty_upgradable_to_new( - true, - old_container_name, - old_column_name, - &old_ty.elem_ty, - &new_ty.elem_ty, - ), + // For arrays, we need to check the element type's upgradability due to sums. + (AlgebraicType::Array(old_ty), AlgebraicType::Array(new_ty)) => { + ensure_old_ty_upgradable_to_new(&old_ty.elem_ty, &new_ty.elem_ty) + } // We only have the simple cases left, and there, no change is good change. - (old_ty, new_ty) if old_ty == new_ty => Ok(Any(false)), - _ => Err(if within { - ChangeWithinColumnType(parts_for_error()) - } else { - ChangeColumnType(parts_for_error()) - } - .into()), + (old_ty, new_ty) if old_ty == new_ty => Some(Any(false)), + _ => None, } } From 8f11e8741d571da7b2de336c7ae058a7cae5208e Mon Sep 17 00:00:00 2001 From: Ludv1g Date: Sat, 22 Aug 2026 03:55:30 +0200 Subject: [PATCH 4/4] feat: Allow toggling the event flag of empty tables during auto-migration Ports the is_event toggle from #4875 (superseding that PR) onto the CheckTableEmpty precheck requested in its review: the planner emits a new ChangeEventFlag step gated by CheckTableEmpty + DisconnectAllUsers, with emptiness validated before any mutations instead of mid-step. When the flag flips together with column changes, the columns are planned with the non-event (empty-reschema) machinery, since the table is verified empty anyway. Datastore: MutTxId::alter_table_event_flag flips the schema flag on the tx and commit tables, inserts/deletes the st_event_table row, and records PendingSchemaChange::TableAlterEventFlag for rollback. Idempotent flips are a no-op; as a backstop it errors with TableNotEmpty on resident rows. Replay: st_event_table inserts/deletes flip is_event on the referenced table's cached schema, so rows logged after a flip replay with the correct event-ness. A delete which empties a table is persisted as a truncation, so an is_event = false flip which empties st_event_table reaches replay as Truncate(st_event_table): replay_truncate un-marks the referenced tables. Covered by the replay_event_flag_flip_* tests, which fail without it. --- .../locking_tx_datastore/committed_state.rs | 6 + .../src/locking_tx_datastore/datastore.rs | 196 +++++++++++++++++- .../src/locking_tx_datastore/mut_tx.rs | 53 ++++- .../src/locking_tx_datastore/replay.rs | 55 +++++ .../src/locking_tx_datastore/tx_state.rs | 5 + crates/engine/src/relational_db.rs | 4 + crates/engine/src/update.rs | 168 +++++++++++++++ crates/schema/src/auto_migrate.rs | 88 ++++---- crates/schema/src/auto_migrate/formatter.rs | 7 + .../src/auto_migrate/termcolor_formatter.rs | 14 ++ crates/schema/tests/ensure_same_schema.rs | 1 + 11 files changed, 548 insertions(+), 49 deletions(-) diff --git a/crates/datastore/src/locking_tx_datastore/committed_state.rs b/crates/datastore/src/locking_tx_datastore/committed_state.rs index b40518ec666..a68bf4ce3e8 100644 --- a/crates/datastore/src/locking_tx_datastore/committed_state.rs +++ b/crates/datastore/src/locking_tx_datastore/committed_state.rs @@ -856,6 +856,12 @@ impl CommittedState { let table = self.tables.get_mut(&table_id)?; table.with_mut_schema(|s| s.table_access = access); } + // A table's `is_event` flag was changed. Change back to the old one. + TableAlterEventFlag(table_id, old_is_event) => { + let table = self.tables.get_mut(&table_id)?; + assert_eq!(table.row_count, 0); + table.with_mut_schema(|s| s.is_event = old_is_event); + } // A table's primary key was changed. Change back to the old one. TableAlterPrimaryKey(table_id, old_pk) => { let table = self.tables.get_mut(&table_id)?; diff --git a/crates/datastore/src/locking_tx_datastore/datastore.rs b/crates/datastore/src/locking_tx_datastore/datastore.rs index 549de0bc183..7934a3a0a9f 100644 --- a/crates/datastore/src/locking_tx_datastore/datastore.rs +++ b/crates/datastore/src/locking_tx_datastore/datastore.rs @@ -296,6 +296,14 @@ impl Locking { tx.alter_table_access(table_id, access) } + pub fn alter_table_event_flag_mut_tx(&self, tx: &mut MutTxId, name: &str, is_event: bool) -> Result<()> { + let table_id = self + .table_id_from_name_mut_tx(tx, name)? + .ok_or_else(|| TableError::NotFound(name.into()))?; + + tx.alter_table_event_flag(table_id, is_event) + } + pub fn alter_table_primary_key_mut_tx( &self, tx: &mut MutTxId, @@ -1090,12 +1098,12 @@ pub(crate) mod tests { use crate::locking_tx_datastore::tx_state::PendingSchemaChange; use crate::system_tables::{ system_tables, StColumnRow, StConnectionCredentialsFields, StConstraintData, StConstraintFields, - StConstraintRow, StEventTableFields, StIndexAlgorithm, StIndexFields, StIndexRow, StRowLevelSecurityFields, - StScheduledFields, StSequenceFields, StSequenceRow, StTableRow, StVarFields, StViewArgFields, StViewFields, - ST_CLIENT_ID, ST_CLIENT_NAME, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_NAME, ST_COLUMN_ID, ST_COLUMN_NAME, - ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID, ST_CONSTRAINT_NAME, - ST_EVENT_TABLE_ID, ST_EVENT_TABLE_NAME, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_NAME, ST_INDEX_ID, - ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE, ST_ROW_LEVEL_SECURITY_ID, + StConstraintRow, StEventTableFields, StEventTableRow, StIndexAlgorithm, StIndexFields, StIndexRow, + StRowLevelSecurityFields, StScheduledFields, StSequenceFields, StSequenceRow, StTableRow, StVarFields, + StViewArgFields, StViewFields, ST_CLIENT_ID, ST_CLIENT_NAME, ST_COLUMN_ACCESSOR_ID, ST_COLUMN_ACCESSOR_NAME, + ST_COLUMN_ID, ST_COLUMN_NAME, ST_CONNECTION_CREDENTIALS_ID, ST_CONNECTION_CREDENTIALS_NAME, ST_CONSTRAINT_ID, + ST_CONSTRAINT_NAME, ST_EVENT_TABLE_ID, ST_EVENT_TABLE_NAME, ST_INDEX_ACCESSOR_ID, ST_INDEX_ACCESSOR_NAME, + ST_INDEX_ID, ST_INDEX_NAME, ST_MODULE_NAME, ST_RESERVED_SEQUENCE_RANGE, ST_ROW_LEVEL_SECURITY_ID, ST_ROW_LEVEL_SECURITY_NAME, ST_SCHEDULED_ID, ST_SCHEDULED_NAME, ST_SEQUENCE_ID, ST_SEQUENCE_NAME, ST_TABLE_ACCESSOR_ID, ST_TABLE_ACCESSOR_NAME, ST_TABLE_NAME, ST_VAR_ID, ST_VAR_NAME, ST_VIEW_ARG_ID, ST_VIEW_ARG_NAME, ST_VIEW_COLUMN_ID, ST_VIEW_COLUMN_NAME, ST_VIEW_ID, ST_VIEW_NAME, ST_VIEW_PARAM_ID, @@ -3260,6 +3268,182 @@ pub(crate) mod tests { Ok(()) } + /// Asserts that the live schema's `is_event` flag for `table_id` equals `expected`. + fn assert_is_event_state(tx: &MutTxId, table_id: TableId, expected: bool) { + let actual = tx + .get_schema(table_id) + .map(|s| s.is_event) + .expect("schema should exist"); + assert_eq!(actual, expected, "expected table {table_id:?} is_event={expected}"); + } + + /// Returns whether `st_event_table` contains a row referencing `table_id`. + fn st_event_table_has_row(datastore: &Locking, tx: &MutTxId, table_id: TableId) -> bool { + datastore + .iter_by_col_eq_mut_tx(tx, ST_EVENT_TABLE_ID, StEventTableFields::TableId, &table_id.into()) + .expect("st_event_table lookup should succeed") + .next() + .is_some() + } + + /// Asserts that `tx.pending_schema_changes()` contains exactly one + /// `TableAlterEventFlag` change for `table_id` recording the old value + /// (i.e. the value just before we altered to `state`). + fn check_table_event_flag_altered(tx: &MutTxId, table_id: TableId, state: bool) { + assert_eq!( + tx.pending_schema_changes(), + [PendingSchemaChange::TableAlterEventFlag(table_id, !state)] + ); + } + + #[test] + fn test_alter_table_event_flag_non_event_to_event() -> ResultTest<()> { + // Create a non-event table. + let (datastore, tx, table_id) = setup_table()?; + commit(&datastore, tx)?; + + // Flip `is_event` from `false` to `true`. + let mut tx = begin_mut_tx(&datastore); + assert_is_event_state(&tx, table_id, false); + assert!( + !st_event_table_has_row(&datastore, &tx, table_id), + "fresh non-event table must not have a row in `st_event_table`" + ); + + tx.alter_table_event_flag(table_id, true)?; + check_table_event_flag_altered(&tx, table_id, true); + assert_is_event_state(&tx, table_id, true); + assert!( + st_event_table_has_row(&datastore, &tx, table_id), + "after flipping to event, `st_event_table` should have the row" + ); + + let tx_data = commit(&datastore, tx)?; + // Flipping to event inserts one row into `st_event_table` + // and does not touch the user table's row data. + let expected_row = ProductValue::from(StEventTableRow { table_id }); + assert_eq!(tx_data.inserts_for_table(ST_EVENT_TABLE_ID), Some(&[expected_row][..]),); + assert_eq!(tx_data.inserts_for_table(table_id), None); + assert_eq!(tx_data.deletes_for_table(table_id), None); + + // After commit, the schema should reflect the flipped flag + // and `st_event_table` should contain the row. + let tx = begin_mut_tx(&datastore); + assert_is_event_state(&tx, table_id, true); + assert!( + st_event_table_has_row(&datastore, &tx, table_id), + "after commit, `st_event_table` should have the row" + ); + Ok(()) + } + + #[test] + fn test_alter_table_event_flag_event_to_non_event() -> ResultTest<()> { + // Create an event table. + let (datastore, tx, table_id) = setup_event_table()?; + commit(&datastore, tx)?; + + // Sanity check: `st_event_table` should have the row. + let mut tx = begin_mut_tx(&datastore); + assert_is_event_state(&tx, table_id, true); + assert!( + st_event_table_has_row(&datastore, &tx, table_id), + "event table should have a row in `st_event_table`" + ); + + // Flip `is_event` from `true` to `false`. + tx.alter_table_event_flag(table_id, false)?; + check_table_event_flag_altered(&tx, table_id, false); + assert_is_event_state(&tx, table_id, false); + assert!( + !st_event_table_has_row(&datastore, &tx, table_id), + "after flipping to non-event, `st_event_table` should not have the row" + ); + + let tx_data = commit(&datastore, tx)?; + // Flipping away from event deletes one row from `st_event_table` + // and does not touch the user table's row data. + let expected_row = ProductValue::from(StEventTableRow { table_id }); + assert_eq!(tx_data.deletes_for_table(ST_EVENT_TABLE_ID), Some(&[expected_row][..]),); + assert_eq!(tx_data.inserts_for_table(table_id), None); + assert_eq!(tx_data.deletes_for_table(table_id), None); + + // After commit, the schema should reflect the flipped flag + // and `st_event_table` should NOT contain the row. + let tx = begin_mut_tx(&datastore); + assert_is_event_state(&tx, table_id, false); + assert!( + !st_event_table_has_row(&datastore, &tx, table_id), + "after commit, `st_event_table` should not have the row" + ); + Ok(()) + } + + #[test] + fn test_alter_table_event_flag_rollback_reverts_live_state_and_st_event_table() -> ResultTest<()> { + // Create a non-event table. + let (datastore, tx, table_id) = setup_table()?; + commit(&datastore, tx)?; + + // Start a new tx, flip, check pending change, then rollback. + let mut tx = begin_mut_tx(&datastore); + assert!(!st_event_table_has_row(&datastore, &tx, table_id)); + + tx.alter_table_event_flag(table_id, true)?; + check_table_event_flag_altered(&tx, table_id, true); + // The in-tx view must reflect the flip. + assert_is_event_state(&tx, table_id, true); + assert!( + st_event_table_has_row(&datastore, &tx, table_id), + "after flipping within the tx, `st_event_table` should have the row" + ); + let _ = datastore.rollback_mut_tx(tx); + + // After rollback, the schema and `st_event_table` should be back to pre-state. + let tx = begin_mut_tx(&datastore); + assert_eq!(tx.pending_schema_changes(), []); + assert_is_event_state(&tx, table_id, false); + assert!( + !st_event_table_has_row(&datastore, &tx, table_id), + "rollback should revert the `st_event_table` row" + ); + Ok(()) + } + + #[test] + fn test_alter_table_event_flag_idempotent_no_pending_change() -> ResultTest<()> { + let (datastore, tx, table_id) = setup_table()?; + commit(&datastore, tx)?; + + let mut tx = begin_mut_tx(&datastore); + tx.alter_table_event_flag(table_id, false)?; + assert_eq!(tx.pending_schema_changes(), []); + Ok(()) + } + + #[test] + fn test_alter_table_event_flag_rejects_non_empty_table() -> ResultTest<()> { + let (datastore, tx, table_id) = setup_table()?; + commit(&datastore, tx)?; + + // Insert a committed row. + let mut tx = begin_mut_tx(&datastore); + insert(&datastore, &mut tx, table_id, &u32_str_u32(1, "row", 1))?; + commit(&datastore, tx)?; + + let mut tx = begin_mut_tx(&datastore); + let err = tx + .alter_table_event_flag(table_id, true) + .expect_err("flipping `is_event` on a non-empty table should fail"); + assert!( + matches!(err, DatastoreError::Table(TableError::TableNotEmpty(id)) if id == table_id), + "unexpected error: {err:?}" + ); + assert_eq!(tx.pending_schema_changes(), []); + assert_is_event_state(&tx, table_id, false); + Ok(()) + } + #[test] fn test_alter_table_row_type_rejects_some_bad_changes() -> ResultTest<()> { let datastore = get_datastore()?; diff --git a/crates/datastore/src/locking_tx_datastore/mut_tx.rs b/crates/datastore/src/locking_tx_datastore/mut_tx.rs index cc0bf10e824..b272cc84a58 100644 --- a/crates/datastore/src/locking_tx_datastore/mut_tx.rs +++ b/crates/datastore/src/locking_tx_datastore/mut_tx.rs @@ -956,8 +956,7 @@ impl MutTxId { // Insert into st_event_table if this is an event table. if is_event { - let row = StEventTableRow { table_id }; - self.insert_via_serialize_bsatn(ST_EVENT_TABLE_ID, &row)?; + self.insert_st_event_table_row(table_id)?; } // Create the indexes for the table. @@ -1387,6 +1386,56 @@ impl MutTxId { Ok(()) } + /// Change the `is_event` flag of the table identified by `table_id`. + /// + /// Updates both the in-memory schema and the `st_event_table` system table. + /// This is only valid on a table with no resident rows (the committed-state + /// semantics of the table flip); errors with [`TableError::TableNotEmpty`] otherwise. + /// The flip is a breaking change for subscribed clients, + /// so callers must arrange a `DisconnectAllUsers`. + pub(crate) fn alter_table_event_flag(&mut self, table_id: TableId, is_event: bool) -> Result<()> { + // Write to the table in the tx state (and clone into commit state). + let ((tx_table, ..), (commit_table, ..)) = self.get_or_create_insert_table_mut(table_id)?; + let old_is_event = tx_table.get_schema().is_event; + if old_is_event == is_event { + // Idempotent no-op; do not record a pending change or it would confuse rollback. + return Ok(()); + } + 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::TableNotEmpty(table_id).into()); + } + tx_table.with_mut_schema_and_clone(commit_table, |s| s.is_event = is_event); + + // Remember the pending change so we can undo it if a later system-table update fails. + self.push_schema_change(PendingSchemaChange::TableAlterEventFlag(table_id, old_is_event)); + + // Update `st_event_table`. + if is_event { + self.insert_st_event_table_row(table_id)?; + } else { + self.delete_st_event_table_row(table_id)?; + } + + Ok(()) + } + + /// Inserts a row into `st_event_table` marking `table_id` as an event table. + fn insert_st_event_table_row(&mut self, table_id: TableId) -> Result<()> { + let row = StEventTableRow { table_id }; + self.insert_via_serialize_bsatn(ST_EVENT_TABLE_ID, &row)?; + Ok(()) + } + + /// Drops the row in `st_event_table` for this `table_id`. + fn delete_st_event_table_row(&mut self, table_id: TableId) -> Result<()> { + self.delete_col_eq( + ST_EVENT_TABLE_ID, + StEventTableFields::TableId.col_id(), + &table_id.into(), + ) + } + /// Change the primary key of the table identified by `table_id`. /// /// Updates both the in-memory schema and the `st_table` system table. diff --git a/crates/datastore/src/locking_tx_datastore/replay.rs b/crates/datastore/src/locking_tx_datastore/replay.rs index 3e8c0b5e76b..b383c9f22a4 100644 --- a/crates/datastore/src/locking_tx_datastore/replay.rs +++ b/crates/datastore/src/locking_tx_datastore/replay.rs @@ -820,6 +820,14 @@ impl<'cs> ReplayCommittedState<'cs> { self.st_column_changed(referenced_table_id)?; } + if table_id == ST_EVENT_TABLE_ID { + // An `st_event_table` row was inserted; flip `is_event = true` + // on the referenced table's cached schema. + // The `table_id` is the first (and only) field in `StEventTableRow`. + let referenced_table_id = Self::read_table_id(row); + self.reschema_table_for_st_event_table_update(referenced_table_id, true); + } + Ok(()) } @@ -1019,9 +1027,35 @@ impl<'cs> ReplayCommittedState<'cs> { } } + if table_id == ST_EVENT_TABLE_ID { + // An `st_event_table` row was deleted; flip `is_event = false` + // on the referenced table's cached schema. + // The `table_id` is the first (and only) field in `StEventTableRow`. + // If the referenced table was dropped in this transaction, its in-memory + // structure is already gone and there is nothing to update. + let referenced_table_id = Self::read_table_id(row); + if !self.replay_table_dropped.contains(&referenced_table_id) { + self.reschema_table_for_st_event_table_update(referenced_table_id, false); + } + } + Ok(()) } + /// Update the in-memory table structure's `is_event` flag in response to + /// replay of an `st_event_table` mutation. + fn reschema_table_for_st_event_table_update(&mut self, table_id: TableId, is_event: bool) { + // We only need to update if we've already constructed the in-memory table structure. + // If we haven't yet, then `get_table_and_blob_store_or_create` will see the correct + // schema (via a live `st_event_table` lookup) when it eventually runs. + if let Ok((table, ..)) = self.get_table_and_blob_store_mut(table_id) { + assert_eq!(table.row_count, 0); + table.with_mut_schema(|schema| { + schema.is_event = is_event; + }); + } + } + fn is_event_table_for_replay(&self, table_id: TableId) -> Result { match self.find_st_event_table_row(table_id) { Ok(_) => Ok(true), @@ -1042,6 +1076,27 @@ impl<'cs> ReplayCommittedState<'cs> { return Ok(()); } + if table_id == ST_EVENT_TABLE_ID { + // A delete which empties a table is persisted as a truncation, + // so an `is_event = false` flip which empties `st_event_table` + // reaches replay as a truncation of `st_event_table`. + // Flip `is_event = false` on every table its rows reference. + let unmarked: Vec = self + .table_scan(ST_EVENT_TABLE_ID) + .expect("`st_event_table` should exist when replaying its truncation") + .map(|row_ref| { + row_ref + .read_col::(StEventTableFields::TableId) + .expect("`st_event_table` row should conform to `st_event_table` schema") + }) + .collect(); + for referenced_table_id in unmarked { + if !self.replay_table_dropped.contains(&referenced_table_id) { + self.reschema_table_for_st_event_table_update(referenced_table_id, false); + } + } + } + // Get the table for mutation. let (table, blob_store, ..) = self.get_table_and_blob_store_mut(table_id)?; diff --git a/crates/datastore/src/locking_tx_datastore/tx_state.rs b/crates/datastore/src/locking_tx_datastore/tx_state.rs index 81278e1488d..a100bdbc9cb 100644 --- a/crates/datastore/src/locking_tx_datastore/tx_state.rs +++ b/crates/datastore/src/locking_tx_datastore/tx_state.rs @@ -129,6 +129,10 @@ pub enum PendingSchemaChange { /// The access of the table with [`TableId`] was changed. /// The old access was stored. TableAlterAccess(TableId, StAccess), + /// The `is_event` flag of the table with [`TableId`] was changed. + /// The old value is stored. + /// The table was verified to have no resident rows at the time of the change. + TableAlterEventFlag(TableId, bool), /// The row type of the table with [`TableId`] was changed. /// The old column schemas was stored. /// Only non-representational row-type changes are allowed here, @@ -171,6 +175,7 @@ impl MemoryUsage for PendingSchemaChange { Self::TableRemoved(table_id, table) => table_id.heap_usage() + table.heap_usage(), Self::TableAdded(table_id) => table_id.heap_usage(), Self::TableAlterAccess(table_id, st_access) => table_id.heap_usage() + st_access.heap_usage(), + Self::TableAlterEventFlag(table_id, old_is_event) => table_id.heap_usage() + old_is_event.heap_usage(), Self::TableAlterRowType(table_id, column_schemas) => table_id.heap_usage() + column_schemas.heap_usage(), Self::TableAlterPrimaryKey(table_id, pk) => table_id.heap_usage() + pk.heap_usage(), Self::ConstraintRemoved(table_id, constraint_schema, index_ids) => { diff --git a/crates/engine/src/relational_db.rs b/crates/engine/src/relational_db.rs index 87a23c27db3..cb05c73593d 100644 --- a/crates/engine/src/relational_db.rs +++ b/crates/engine/src/relational_db.rs @@ -1063,6 +1063,10 @@ impl RelationalDB { Ok(self.inner.alter_table_access_mut_tx(tx, name, access)?) } + pub(crate) fn alter_table_event_flag(&self, tx: &mut MutTx, name: &str, is_event: bool) -> Result<(), DBError> { + Ok(self.inner.alter_table_event_flag_mut_tx(tx, name, is_event)?) + } + pub(crate) fn alter_table_primary_key( &self, tx: &mut MutTx, diff --git a/crates/engine/src/update.rs b/crates/engine/src/update.rs index 99712c20218..6b1d9b45e91 100644 --- a/crates/engine/src/update.rs +++ b/crates/engine/src/update.rs @@ -606,6 +606,21 @@ fn auto_migrate_database( }; stdb.alter_table_access(tx, &table_name, access.into())?; } + spacetimedb_schema::auto_migrate::AutoMigrateStep::ChangeEventFlag(table_name_key) => { + let (namespace, local) = table_name_key; + let table_name = joined(namespace, local); + let (_owning_def, table_def) = plan.new.find_table(table_name_key).ok_or_else(|| { + anyhow::anyhow!("ChangeEventFlag: table `{table_name}` not found in new module def") + })?; + log!( + logger, + "Changing `event` flag on table `{table_name}` to `{}`", + table_def.is_event + ); + // Emptiness was already validated by the matching `CheckTableEmpty` precheck; + // the datastore re-checks it as a backstop. + stdb.alter_table_event_flag(tx, &table_name, table_def.is_event)?; + } spacetimedb_schema::auto_migrate::AutoMigrateStep::ChangePrimaryKey(table_name_key) => { let (namespace, local) = table_name_key; let table_name = joined(namespace, local); @@ -1899,6 +1914,159 @@ mod test { replay_generally_reschemaed_table(TakeSnapshot::BeforeAutomigration) } + /// A single-table module whose `events` table's `is_event` flag is `is_event`. + fn eventable_module(is_event: bool) -> ModuleDef { + let mut builder = RawModuleDefV10Builder::new(); + builder + .build_table_with_new_type("events", ProductType::from([("id", U64)]), true) + .with_event(is_event) + .with_access(TableAccess::Public) + .finish(); + builder + .finish() + .try_into() + .expect("should be a valid module definition") + } + + fn assert_is_event(stdb: &TestDB, tx: &MutTx, table_id: TableId, expected: bool) -> anyhow::Result<()> { + let schema = stdb.schema_for_table_mut(tx, table_id)?; + assert_eq!(schema.is_event, expected); + Ok(()) + } + + #[test] + fn change_event_flag_empty_table_succeeds() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = eventable_module(false); + let new = eventable_module(true); + let table_id = create_table_for_module(&stdb, &old, "events")?; + + let mut tx = begin_mut_tx(&stdb); + assert_is_event(&stdb, &tx, table_id, false)?; + + let plan = ponder_migrate(&old, &new)?; + let res = update_database(&stdb, &mut tx, auth_ctx.clone(), plan, &TestLogger)?; + assert!( + matches!(res, UpdateResult::RequiresClientDisconnect), + "flipping the `event` flag should disconnect clients" + ); + assert_is_event(&stdb, &tx, table_id, true)?; + assert_eq!( + tx.pending_schema_changes(), + [PendingSchemaChange::TableAlterEventFlag(table_id, false)] + ); + stdb.commit_tx(tx)?; + + // And flip back. + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&new, &old)?; + let res = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + assert!(matches!(res, UpdateResult::RequiresClientDisconnect)); + assert_is_event(&stdb, &tx, table_id, false)?; + stdb.commit_tx(tx)?; + + Ok(()) + } + + #[test] + fn change_event_flag_nonempty_table_fails() -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let stdb = TestDB::durable()?; + + let old = eventable_module(false); + let new = eventable_module(true); + let table_id = create_table_for_module(&stdb, &old, "events")?; + + // Insert a row in a separate tx so the pre-flip table state is committed. + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![42u64])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&old, &new)?; + let err = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger) + .err() + .expect("flipping `is_event` on a non-empty table should fail"); + assert!( + err.to_string().contains("contains data"), + "error should mention that the table contains data, got: {err}" + ); + assert_is_event(&stdb, &tx, table_id, false)?; + assert_eq!(tx.pending_schema_changes(), []); + Ok(()) + } + + /// Flips `events` from non-event to event and back to non-event across commits, + /// inserting rows before the first flip (deleted) and after the second, + /// then replays the commitlog and verifies flag and rows. + fn replay_event_flag_flip(snapshot: TakeSnapshot) -> anyhow::Result<()> { + let auth_ctx = AuthCtx::for_testing(); + let with_snapshot = matches!(snapshot, TakeSnapshot::BeforeAutomigration); + let stdb = with_snapshotting(with_snapshot)?; + + let non_event = eventable_module(false); + let event = eventable_module(true); + let table_id = create_table_for_module(&stdb, &non_event, "events")?; + + // Commitlog contains writes to the table while it was a regular table. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![7u64])?; + stdb.commit_tx(tx)?; + + let mut tx = begin_mut_tx(&stdb); + assert_eq!(stdb.delete_by_rel(&mut tx, table_id, [product![7u64]]), 1); + stdb.commit_tx(tx)?; + } + + if with_snapshot { + take_snapshot(&stdb)?; + } + + // Flip to event... + { + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&non_event, &event)?; + let _ = update_database(&stdb, &mut tx, auth_ctx.clone(), plan, &TestLogger)?; + stdb.commit_tx(tx)?; + } + + // ...and back to non-event. + { + let mut tx = begin_mut_tx(&stdb); + let plan = ponder_migrate(&event, &non_event)?; + let _ = update_database(&stdb, &mut tx, auth_ctx, plan, &TestLogger)?; + stdb.commit_tx(tx)?; + } + + // Insert a row now that the table is a regular table again. + { + let mut tx = begin_mut_tx(&stdb); + insert(&stdb, &mut tx, table_id, &product![42u64])?; + stdb.commit_tx(tx)?; + } + + // Replay the commitlog and verify the flag and the rows survived. + let stdb = stdb.reopen()?; + let tx = begin_mut_tx(&stdb); + assert_is_event(&stdb, &tx, table_id, false)?; + assert_eq!(collect_rows(&stdb, &tx, table_id)?, [product![42u64]]); + + Ok(()) + } + + #[test] + fn replay_event_flag_flip_no_snapshot() -> anyhow::Result<()> { + replay_event_flag_flip(TakeSnapshot::None) + } + + #[test] + fn replay_event_flag_flip_after_snapshot() -> anyhow::Result<()> { + replay_event_flag_flip(TakeSnapshot::BeforeAutomigration) + } + #[test] fn add_sequence_precheck_rejects_existing_column_max_value() -> anyhow::Result<()> { let auth_ctx = AuthCtx::for_testing(); diff --git a/crates/schema/src/auto_migrate.rs b/crates/schema/src/auto_migrate.rs index 121cc0fb0ea..83583d570cd 100644 --- a/crates/schema/src/auto_migrate.rs +++ b/crates/schema/src/auto_migrate.rs @@ -354,6 +354,14 @@ pub enum AutoMigrateStep<'def> { /// Change the access of a table or view. ChangeAccess(::Key<'def>), + /// Toggle the `is_event` flag of a table. + /// + /// Only valid on an empty table; the plan will contain a matching + /// [`AutoMigratePrecheck::CheckTableEmpty`], and execution fails if the table contains data. + /// This is a breaking change for subscribed clients (the committed-state semantics of the + /// table flip), so this step is always accompanied by a `DisconnectAllUsers`. + ChangeEventFlag(::Key<'def>), + /// Change the primary key of a table. /// /// This updates the `table_primary_key` field in `st_table` to match the new module definition. @@ -378,9 +386,6 @@ pub enum AutoMigrateError { type2: TableType, }, - #[error("Changing the event flag of table {table} requires a manual migration")] - ChangeTableEventFlag { table: Identifier }, - #[error( "Changing the accessor name on index {index} from {old_accessor:?} to {new_accessor:?} requires a manual migration" )] @@ -668,17 +673,19 @@ fn auto_migrate_table<'def>( } .into()) }; - let event_ok: Result<()> = if old.is_event == new.is_event { - Ok(()) - } else { - Err(AutoMigrateError::ChangeTableEventFlag { - table: old.name.clone(), - } - .into()) - }; + if old.is_event != new.is_event { + // Flipping `is_event` changes the committed-state semantics of the table, which is + // only possible when there are no committed rows; emptiness is validated at + // execution time. The flip is observable to subscribers, so clients must reconnect. + plan.ensure_check_table_empty(key); + plan.ensure_disconnect_all_users(); + plan.steps.push(AutoMigrateStep::ChangeEventFlag(key)); + } - // Combined with our validation of `event_ok`, `old.is_event` is sufficient to identify this as an event table. - let is_event = old.is_event; + // Plan column changes with the event-table machinery only when the table is an event + // table both before and after. When the flag is flipping, the table is verified empty + // at execution time anyway, so column changes ride the empty-reschema machinery. + let is_event = old.is_event && new.is_event; if old.table_access != new.table_access { plan.steps.push(AutoMigrateStep::ChangeAccess(key)); @@ -782,10 +789,9 @@ fn auto_migrate_table<'def>( .collect_all_errors::>(); let ( - (), (), ArrayMonoid([Any(row_type_changed), Any(columns_added), Any(event_schema_changed), Any(needs_empty_reschema)]), - ) = (type_ok, event_ok, columns_ok).combine_errors()?; + ) = (type_ok, columns_ok).combine_errors()?; if event_schema_changed { // If we're rewriting an event table, there's no data migration to do. @@ -2595,32 +2601,32 @@ mod tests { } #[test] - fn test_change_event_flag_rejected() { - // non-event → event - let old = create_v10_module_def(|builder| { - builder - .build_table_with_new_type("Events", ProductType::from([("id", AlgebraicType::U64)]), true) - .finish(); - }); - let new = create_v10_module_def(|builder| { - builder - .build_table_with_new_type("events", ProductType::from([("id", AlgebraicType::U64)]), true) - .with_event(true) - .finish(); - }); - - let result = ponder_auto_migrate(&old, &new); - expect_error_matching!( - result, - AutoMigrateError::ChangeTableEventFlag { table } => &table[..] == "events" - ); - - // event → non-event (reverse direction) - let result = ponder_auto_migrate(&new, &old); - expect_error_matching!( - result, - AutoMigrateError::ChangeTableEventFlag { table } => &table[..] == "events" - ); + fn test_change_event_flag_produces_step() { + let build = |is_event: bool| { + create_v10_module_def(|builder| { + builder + .build_table_with_new_type("events", ProductType::from([("id", AlgebraicType::U64)]), true) + .with_event(is_event) + .finish(); + }) + }; + let assert_flip = |old_is_event: bool, new_is_event: bool| { + let old = build(old_is_event); + let new = build(new_is_event); + let events = key("", "events"); + let plan = ponder_auto_migrate(&old, &new).expect("toggling `is_event` on an empty table should plan"); + // The flip is only valid on an empty table, validated before any mutations. + assert_eq!(&plan.prechecks[..], &[AutoMigratePrecheck::CheckTableEmpty(events)]); + assert_eq!( + &plan.steps[..], + &[ + AutoMigrateStep::ChangeEventFlag(events), + AutoMigrateStep::DisconnectAllUsers, + ], + ); + }; + assert_flip(false, true); + assert_flip(true, false); } #[test] diff --git a/crates/schema/src/auto_migrate/formatter.rs b/crates/schema/src/auto_migrate/formatter.rs index 8d4cdcb5f50..dcb915fd067 100644 --- a/crates/schema/src/auto_migrate/formatter.rs +++ b/crates/schema/src/auto_migrate/formatter.rs @@ -97,6 +97,12 @@ fn format_step( let access_info = extract_access_change_info(*table, plan)?; f.format_change_access(&access_info) } + AutoMigrateStep::ChangeEventFlag(table) => { + let name = joined(*table); + let not_found = || FormattingErrors::TableNotFound { table: (&*name).into() }; + let (_, new_table) = plan.new.find_table(*table).ok_or_else(not_found)?; + f.format_change_event_flag(&name, new_table.is_event) + } AutoMigrateStep::ChangePrimaryKey(table) => { let name = joined(*table); let not_found = || FormattingErrors::TableNotFound { table: (&*name).into() }; @@ -189,6 +195,7 @@ pub trait MigrationFormatter { fn format_constraint(&mut self, constraint_info: &ConstraintInfo, action: Action) -> io::Result<()>; fn format_sequence(&mut self, sequence_info: &SequenceInfo, action: Action) -> io::Result<()>; fn format_change_access(&mut self, access_info: &AccessChangeInfo) -> io::Result<()>; + fn format_change_event_flag(&mut self, table_name: &NamespacedIdentifier, new_is_event: bool) -> io::Result<()>; fn format_change_primary_key( &mut self, table_name: &NamespacedIdentifier, diff --git a/crates/schema/src/auto_migrate/termcolor_formatter.rs b/crates/schema/src/auto_migrate/termcolor_formatter.rs index 1d2f3af42b6..7f928cd72e4 100644 --- a/crates/schema/src/auto_migrate/termcolor_formatter.rs +++ b/crates/schema/src/auto_migrate/termcolor_formatter.rs @@ -325,6 +325,20 @@ impl MigrationFormatter for TermColorFormatter { self.buffer.write_all(b")\n") } + fn format_change_event_flag(&mut self, table_name: &NamespacedIdentifier, new_is_event: bool) -> io::Result<()> { + let direction = if new_is_event { + "non-event \u{2192} event" + } else { + "event \u{2192} non-event" + }; + self.write_action_prefix(&Action::Changed)?; + self.buffer.write_all(b" event flag for table ")?; + self.write_colored(table_name, Some(self.colors.table_name), true)?; + self.buffer.write_all(b" (")?; + self.write_colored(direction, Some(self.colors.access), false)?; + self.buffer.write_all(b")\n") + } + fn format_change_primary_key( &mut self, table_name: &NamespacedIdentifier, diff --git a/crates/schema/tests/ensure_same_schema.rs b/crates/schema/tests/ensure_same_schema.rs index 8efc611bd90..1734b0735ab 100644 --- a/crates/schema/tests/ensure_same_schema.rs +++ b/crates/schema/tests/ensure_same_schema.rs @@ -28,6 +28,7 @@ fn step_namespace<'a, 'def>(step: &'a AutoMigrateStep<'def>) -> Option<&'a Names | AutoMigrateStep::AddSchedule((ns, _)) | AutoMigrateStep::AddView((ns, _)) | AutoMigrateStep::ChangeAccess((ns, _)) + | AutoMigrateStep::ChangeEventFlag((ns, _)) | AutoMigrateStep::ChangePrimaryKey((ns, _)) | AutoMigrateStep::UpdateView((ns, _)) | AutoMigrateStep::RemoveIndex((ns, _))