From 58331a512d1734d59f482116e129f49d5108c2be Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:38:39 +0000 Subject: [PATCH 1/2] test(storage-cassandra): migrate keyspaces on a pooled session after another caller's USE (#632) Add tests/direct/keyspace_migrations.rs and register it in tests/direct_integration.rs. Its tests build their own Cassandra session with 10 connections per node and run the catalog and data migration runners on it. migrations_ignore_the_keyspace_another_caller_selected sends USE for a decoy keyspace before each runner, the way the previous runner on the same session selects its own keyspace. It runs 5 catalog and 5 data migrations, each into a new keyspace, then checks that each migrated keyspace holds every table, index and added column its runner creates and none of the other runner's tables, and that the decoy keyspace holds no table. migrations_run_on_a_session_with_no_keyspace_and_leave_none_selected runs both runners on a session that has never selected a keyspace. It checks that both succeed, that both schemas are complete, and that the session still has no keyspace selected afterwards. The cdrs_tokio driver applies USE on the connection that ran it and copies it to the session's other connections from a background task, so a runner that relies on USE can send a statement to a connection that is still in the keyspace selected before. Against runners that select their keyspace with USE, the first test finds migrations that ran in the decoy keyspace, and the second finds a keyspace left selected on the session. --- .../tests/direct/keyspace_migrations.rs | 337 ++++++++++++++++++ .../tests/direct_integration.rs | 2 + 2 files changed, 339 insertions(+) create mode 100644 crates/storage-cassandra/tests/direct/keyspace_migrations.rs diff --git a/crates/storage-cassandra/tests/direct/keyspace_migrations.rs b/crates/storage-cassandra/tests/direct/keyspace_migrations.rs new file mode 100644 index 000000000..eddd3f354 --- /dev/null +++ b/crates/storage-cassandra/tests/direct/keyspace_migrations.rs @@ -0,0 +1,337 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The migration runners create their tables, indexes and added columns in +//! the keyspace they are given, whatever keyspace the session has selected +//! (#632). +//! +//! cdrs-tokio runs a `USE` on one connection and copies it to the session's +//! other connections, on every node, from a background task. On a session +//! with more than one connection, on one node or across nodes, a statement +//! that relies on the session keyspace can run in the keyspace selected +//! before it. These tests build their own session with `POOL_SIZE` +//! connections per node, so they do not depend on the engine's pool size. + +use std::collections::BTreeSet; +use std::sync::Arc; + +use cdrs_tokio::authenticators::StaticPasswordAuthenticatorProvider; +use cdrs_tokio::cluster::NodeTcpConfigBuilder; +use cdrs_tokio::cluster::connection_pool::ConnectionPoolConfigBuilder; +use cdrs_tokio::cluster::session::{SessionBuilder, TcpSessionBuilder}; +use cdrs_tokio::load_balancing::RoundRobinLoadBalancingStrategy; +use extenddb_storage::error::StorageError; +use extenddb_storage_cassandra::CassandraSession; +use extenddb_storage_cassandra::cassandra_util::{get_column, query_rows}; +use extenddb_storage_cassandra::migrations::{run_catalog_migrations, run_data_migrations}; + +/// Connections per node, the `max_connections` default. +const POOL_SIZE: usize = 10; + +/// Catalog keyspaces, and account keyspaces, the alternating `USE` test migrates. +const ROUNDS: usize = 5; + +/// What one runner leaves in the keyspace it migrates. +struct Schema { + tables: &'static [&'static str], + indexes: &'static [&'static str], + /// `(table, column)` pairs that `ALTER TABLE` statements add. + added_columns: &'static [(&'static str, &'static str)], +} + +/// Catalog migrations V001 to V005. +const CATALOG: Schema = Schema { + tables: &[ + "access_keys", + "accounts", + "admin_users", + "backups_by_account", + "backups_by_arn", + "backups_by_table", + "continuous_backups", + "iam_group_members", + "iam_groups", + "iam_permissions_boundaries", + "iam_policies", + "iam_role_tags", + "iam_roles", + "iam_sessions", + "iam_user_tags", + "iam_users", + "idempotency_tokens_by_account", + "indexes", + "login_attempts", + "metrics", + "schema_history", + "settings", + "tables", + "tags", + ], + indexes: &["tables_table_id_idx"], + added_columns: &[ + ("backups_by_arn", "active_restores"), + ("backups_by_arn", "definition"), + ("tables", "restore_date_time"), + ("tables", "restore_heartbeat"), + ("tables", "restore_owner"), + ("tables", "restore_source_backup_arn"), + ("tables", "restore_source_table_arn"), + ("tables", "ttl_backfill_cursor"), + ], +}; + +/// Data migrations V001 and V002. +const DATA: Schema = Schema { + tables: &[ + "backup_items", + "gsi_pending", + "index_propagation_holds", + "schema_history", + "stream_records", + "stream_shards", + "transaction_ledger", + "ttl_expiration_buckets", + "ttl_expirations", + "ttl_reconcile_pending", + "ttl_repair_inflight", + "ttl_repair_inflight_partitions", + ], + indexes: &["idx_txn_started_at", "stream_shards_table_id_idx"], + added_columns: &[], +}; + +/// A session built as the engine builds it, with `POOL_SIZE` connections per node. +async fn pooled_session() -> Arc { + let config = crate::helpers::test_config(); + let mut node = NodeTcpConfigBuilder::new(); + for contact_point in &config.contact_points { + node = node.with_contact_point(contact_point.clone().into()); + } + let (Some(username), Some(password)) = (&config.username, &config.password) else { + panic!("test_config sets no credentials"); + }; + let cluster = node + .with_authenticator_provider(Arc::new(StaticPasswordAuthenticatorProvider::new( + username, password, + ))) + .build() + .await + .expect("cluster config"); + let pool = ConnectionPoolConfigBuilder::new() + .with_local_size(POOL_SIZE) + .with_remote_size(POOL_SIZE) + .build(); + let session = TcpSessionBuilder::new(RoundRobinLoadBalancingStrategy::new(), cluster) + .with_connection_pool_config(pool) + .build() + .await + .expect("pooled session"); + Arc::new(session) +} + +async fn create_keyspaces(session: &Arc, keyspaces: &[String]) { + let datacenter = crate::helpers::test_config().datacenter; + for keyspace in keyspaces { + session + .query(format!( + "CREATE KEYSPACE {keyspace} WITH replication = \ + {{'class': 'NetworkTopologyStrategy', '{datacenter}': 1}}" + )) + .await + .expect("create keyspace"); + } +} + +/// Drops `keyspaces` now. Each migrated keyspace holds 12 to 24 tables, so +/// leaving them for the harness's queued drops would grow the node's schema +/// by hundreds of tables per run. +async fn drop_keyspaces(session: &Arc, keyspaces: &[String]) { + for keyspace in keyspaces { + session + .query(format!("DROP KEYSPACE IF EXISTS {keyspace}")) + .await + .expect("drop keyspace"); + } +} + +/// The `system_schema` rows `query` returns for `keyspace`, as the values of +/// `columns` joined with a dot. +async fn schema_names( + session: &Arc, + query: &str, + keyspace: &str, + columns: &[&str], +) -> BTreeSet { + let rows = query_rows::( + session, + query, + cdrs_tokio::query_values!(keyspace), + "schema read", + ) + .await + .expect(query); + rows.iter() + .map(|row| { + columns + .iter() + .map(|column| get_column::(row, column, "schema read")) + .collect::, _>>() + .expect("schema column") + .join(".") + }) + .collect() +} + +async fn tables_in(session: &Arc, keyspace: &str) -> BTreeSet { + let query = "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?"; + schema_names(session, query, keyspace, &["table_name"]).await +} + +/// One line for each table, index or added column of `schema` that `keyspace` +/// lacks, and for each table of `other` (the other runner's schema) it holds. +async fn problems_in( + session: &Arc, + keyspace: &str, + schema: &Schema, + other: &Schema, +) -> Vec { + let tables = tables_in(session, keyspace).await; + let index_query = "SELECT index_name FROM system_schema.indexes WHERE keyspace_name = ?"; + let indexes = schema_names(session, index_query, keyspace, &["index_name"]).await; + let column_query = + "SELECT table_name, column_name FROM system_schema.columns WHERE keyspace_name = ?"; + let columns = schema_names( + session, + column_query, + keyspace, + &["table_name", "column_name"], + ) + .await; + + let mut problems = Vec::new(); + for table in schema.tables.iter().filter(|t| !tables.contains(**t)) { + problems.push(format!("{keyspace}: no table {table}")); + } + for index in schema.indexes.iter().filter(|i| !indexes.contains(**i)) { + problems.push(format!("{keyspace}: no index {index}")); + } + for (table, column) in schema.added_columns { + if !columns.contains(&format!("{table}.{column}")) { + problems.push(format!("{keyspace}: no column {table}.{column}")); + } + } + for table in other.tables.iter().filter(|t| !schema.tables.contains(t)) { + if tables.contains(*table) { + problems.push(format!( + "{keyspace}: holds {table}, which the other runner creates" + )); + } + } + problems +} + +/// Before every migration run the session selects another keyspace with +/// `USE`, as an earlier runner on the same session did before #632: +/// `extenddb init` and `extenddb migrate` run the catalog migrations and then +/// each account's data migrations on the engine's one session, and each +/// runner began with `USE` for its own keyspace. Each keyspace must end up +/// with its own schema and the other keyspace with none. Against runners +/// that rely on `USE`, some runs execute their statements in the other +/// keyspace, so the outcome varies from run to run, and `ROUNDS` runs of +/// each runner give it many chances to show. +#[tokio::test] +async fn migrations_ignore_the_keyspace_another_caller_selected() { + if crate::helpers::skip_without_cassandra() { + return; + } + let session = pooled_session().await; + let prefix = format!("m632_{}", &crate::helpers::unique_test_id()[..8]); + let decoy = format!("{prefix}_decoy"); + let rounds: Vec<[String; 2]> = (0..ROUNDS) + .map(|i| { + [ + format!("{prefix}_catalog_{i}"), + format!("{prefix}_account_{i}"), + ] + }) + .collect(); + let _drop = crate::helpers::DropKeyspacesOnDrop( + std::iter::once(decoy.clone()) + .chain(rounds.iter().flatten().cloned()) + .collect(), + ); + create_keyspaces(&session, std::slice::from_ref(&decoy)).await; + + let use_decoy = format!("USE {decoy}"); + let mut problems = Vec::new(); + for round in &rounds { + let [catalog, account] = round; + create_keyspaces(&session, round).await; + session.query(&use_decoy).await.expect("USE decoy"); + if let Err(e) = run_catalog_migrations(&session, catalog).await { + problems.push(format!("{catalog}: run_catalog_migrations: {e:?}")); + } + session.query(&use_decoy).await.expect("USE decoy"); + if let Err(e) = run_data_migrations(&session, account).await { + problems.push(format!("{account}: run_data_migrations: {e:?}")); + } + problems.extend(problems_in(&session, catalog, &CATALOG, &DATA).await); + problems.extend(problems_in(&session, account, &DATA, &CATALOG).await); + drop_keyspaces(&session, round).await; + } + for table in tables_in(&session, &decoy).await { + problems.push(format!("{decoy}: holds {table}")); + } + drop_keyspaces(&session, std::slice::from_ref(&decoy)).await; + + let bad: BTreeSet<&str> = problems + .iter() + .filter_map(|p| p.split(':').next()) + .collect(); + assert!( + problems.is_empty(), + "{} of {} keyspaces wrong after {ROUNDS} catalog and {ROUNDS} account migrations \ + on a {POOL_SIZE}-connection session:\n{}", + bad.len(), + 1 + 2 * ROUNDS, + problems.join("\n") + ); +} + +/// On a session that never selected a keyspace, any migration statement that +/// leaves its keyspace to the session fails ("No keyspace has been +/// specified"). The runners must also leave the session's keyspace unset, so +/// they move no other caller's statements. +#[tokio::test] +async fn migrations_run_on_a_session_with_no_keyspace_and_leave_none_selected() { + if crate::helpers::skip_without_cassandra() { + return; + } + let session = pooled_session().await; + assert_eq!( + session.current_keyspace(), + None, + "a new session has no keyspace" + ); + let prefix = format!("m632_{}", &crate::helpers::unique_test_id()[..8]); + let catalog = format!("{prefix}_catalog"); + let account = format!("{prefix}_account"); + let keyspaces = vec![catalog.clone(), account.clone()]; + let _drop = crate::helpers::DropKeyspacesOnDrop(keyspaces.clone()); + create_keyspaces(&session, &keyspaces).await; + + let catalog_result = run_catalog_migrations(&session, &catalog).await; + let data_result = run_data_migrations(&session, &account).await; + let mut problems = problems_in(&session, &catalog, &CATALOG, &DATA).await; + problems.extend(problems_in(&session, &account, &DATA, &CATALOG).await); + let selected = session.current_keyspace(); + drop_keyspaces(&session, &keyspaces).await; + + catalog_result.expect("catalog migrations on a session with no keyspace"); + data_result.expect("data migrations on a session with no keyspace"); + assert!(problems.is_empty(), "{}", problems.join("\n")); + assert_eq!( + selected, None, + "the migration runners selected a keyspace on the session" + ); +} diff --git a/crates/storage-cassandra/tests/direct_integration.rs b/crates/storage-cassandra/tests/direct_integration.rs index 43eb28b81..59682c6f1 100644 --- a/crates/storage-cassandra/tests/direct_integration.rs +++ b/crates/storage-cassandra/tests/direct_integration.rs @@ -31,6 +31,8 @@ mod delete_item; mod groups; #[path = "direct/index.rs"] mod index; +#[path = "direct/keyspace_migrations.rs"] +mod keyspace_migrations; #[path = "direct/metadata_engine.rs"] mod metadata_engine; #[path = "direct/policies.rs"] From 370a6b7104fe3cb226cae5cad796a43322874d8b Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:26:34 +0000 Subject: [PATCH 2/2] fix(storage-cassandra): qualify migration statements with their keyspace instead of USE (#632) The catalog and data migration runners sent USE on the shared session and then ran the unqualified statements of each migration file. The cdrs_tokio driver applies USE on the connection that ran it and copies it to the session's other connections, on every node, from a background task that can finish after the next statement is sent. On a session with more than one connection, on one node or across nodes, a statement could run in the keyspace selected before. The migration then failed with "table schema_history does not exist" or created its tables in another keyspace. The runners no longer send USE. migration_statements splits a migration file on semicolons and writes the runner's keyspace in front of the table each statement names: CREATE TABLE [IF NOT EXISTS] t, CREATE INDEX [IF NOT EXISTS] [name] ON t, ALTER TABLE [IF EXISTS] t and INSERT INTO t, after any leading comment lines. Any other statement, a table name that is quoted or already names a keyspace, and a fragment between semicolons that holds only a comment fail the file before any of its statements runs, with an error that names the file. The .cql files are unchanged. Unit tests cover each accepted form and the rejected ones. They also run every statement of every embedded migration file through migration_statements and check that the keyspace is written once, after the leading comments, in front of the table. The isolated_engine helper in tests/direct/backup_engine.rs builds a catalog through V004 statement by statement. It no longer sends USE, and it qualifies each statement through migration_statements. --- crates/storage-cassandra/src/migrations.rs | 292 ++++++++++++++++-- .../tests/direct/backup_engine.rs | 38 ++- 2 files changed, 297 insertions(+), 33 deletions(-) diff --git a/crates/storage-cassandra/src/migrations.rs b/crates/storage-cassandra/src/migrations.rs index b1bc4fe20..612e74ec7 100644 --- a/crates/storage-cassandra/src/migrations.rs +++ b/crates/storage-cassandra/src/migrations.rs @@ -30,6 +30,9 @@ //! 2. Use the next sequential version number (e.g., V003 after V002) //! 3. Add the migration to the `CATALOG_MIGRATIONS` array below //! 4. Migrations are embedded at compile time via `include_str!()` +//! 5. Name tables without a keyspace. The runner writes the keyspace it +//! migrates into each `CREATE TABLE`, `CREATE INDEX`, `ALTER TABLE` and +//! `INSERT INTO` statement, and rejects any other statement form. use cdrs_tokio::cluster::TcpConnectionManager; use cdrs_tokio::cluster::session::Session; @@ -87,19 +90,13 @@ pub async fn run_catalog_migrations( ) -> OpResult<()> { println!("--- Running catalog migrations..."); - // Set keyspace context for all subsequent queries - session - .query(format!("USE {keyspace}")) - .await - .map_err(|e| OpError::Internal(format!("Failed to USE keyspace: {e}")))?; - for (filename, sql) in CATALOG_MIGRATIONS { if is_migration_applied(session, keyspace, filename).await? { println!(" {filename} — already applied, skipping."); continue; } println!(" Applying {filename}..."); - execute_migration(session, sql).await?; + execute_migration(session, keyspace, filename, sql).await?; record_migration(session, keyspace, filename).await?; } println!(" Migrations applied."); @@ -110,19 +107,13 @@ pub async fn run_catalog_migrations( pub async fn run_data_migrations(session: &Arc, keyspace: &str) -> OpResult<()> { println!("--- Running data migrations..."); - // Set keyspace context for all subsequent queries - session - .query(format!("USE {keyspace}")) - .await - .map_err(|e| OpError::Internal(format!("Failed to USE keyspace: {e}")))?; - for (filename, sql) in DATA_MIGRATIONS { if is_migration_applied(session, keyspace, filename).await? { println!(" {filename} — already applied, skipping."); continue; } println!(" Applying {filename}..."); - execute_migration(session, sql).await?; + execute_migration(session, keyspace, filename, sql).await?; record_migration(session, keyspace, filename).await?; } println!(" Data migrations applied."); @@ -200,22 +191,137 @@ async fn is_migration_applied( Ok(applied) } -/// Execute a migration by splitting on semicolons and running each statement. -async fn execute_migration(session: &Arc, sql: &str) -> OpResult<()> { - for statement in sql.split(';').filter(|s| !s.trim().is_empty()) { - let stmt = statement.trim(); - if stmt.is_empty() { - continue; - } - +/// Execute a migration by splitting on semicolons and running each statement +/// with the table it names written as `keyspace.table`. +/// +/// The runners never select a keyspace with `USE` (#632). cdrs-tokio runs a +/// `USE` on one pooled connection and copies it to the others from a +/// background task, so a statement relying on it can run in the keyspace the +/// connection it lands on had selected before. +async fn execute_migration( + session: &Arc, + keyspace: &str, + filename: &str, + sql: &str, +) -> OpResult<()> { + for statement in migration_statements(keyspace, filename, sql)? { session - .query(stmt) + .query(statement) .await .map_err(|e| OpError::Internal(format!("Migration failed: {e}")))?; } Ok(()) } +/// The statements of migration file `filename`, split on semicolons, each +/// qualified with `keyspace` by [`qualify_statement`]. If one cannot be +/// qualified, the file fails before any of its statements runs. +#[doc(hidden)] +pub fn migration_statements(keyspace: &str, filename: &str, sql: &str) -> OpResult> { + sql.split(';') + .map(str::trim) + .filter(|statement| !statement.is_empty()) + .map(|statement| qualify_statement(keyspace, filename, statement)) + .collect() +} + +/// `statement` with the table it names written as `keyspace.table`. +/// +/// Recognises `CREATE TABLE [IF NOT EXISTS]`, +/// `CREATE INDEX [IF NOT EXISTS] [name] ON`, `ALTER TABLE [IF EXISTS]` and +/// `INSERT INTO`, each naming an unquoted table without a keyspace. Any other +/// statement is an error naming `filename`, so no statement runs in whatever +/// keyspace the session has selected. +fn qualify_statement(keyspace: &str, filename: &str, statement: &str) -> OpResult { + let body = skip_comments(statement); + if body.is_empty() { + return Err(OpError::Internal(format!( + "Migration file {filename} has a fragment between semicolons that holds only a \ + comment (a semicolon inside a comment, or a comment after the last statement): {}", + statement.lines().next().unwrap_or_default() + ))); + } + let at = table_name_offset(body).ok_or_else(|| { + OpError::Internal(format!( + "Migration file {filename} has a statement that names no table the runner can \ + qualify with a keyspace: {}", + body.lines().next().unwrap_or_default() + )) + })?; + let at = statement.len() - body.len() + at; + Ok(format!( + "{}{keyspace}.{}", + &statement[..at], + &statement[at..] + )) +} + +/// `statement` from its first character outside whitespace and `--` or `//` +/// line comments. +fn skip_comments(mut statement: &str) -> &str { + loop { + statement = statement.trim_start(); + if !statement.starts_with("--") && !statement.starts_with("//") { + return statement; + } + statement = statement.split_once('\n').map_or("", |(_, rest)| rest); + } +} + +/// The byte offset in `body` of the table name, for the statement forms +/// [`qualify_statement`] recognises. +fn table_name_offset(body: &str) -> Option { + fn is(word: Option<(usize, &str)>, keyword: &str) -> bool { + word.is_some_and(|(_, word)| word.eq_ignore_ascii_case(keyword)) + } + let mut words = words(body).peekable(); + let (first, second) = (words.next(), words.next()); + if is(first, "CREATE") && (is(second, "TABLE") || is(second, "INDEX")) { + if words + .next_if(|&(_, w)| w.eq_ignore_ascii_case("IF")) + .is_some() + && !(is(words.next(), "NOT") && is(words.next(), "EXISTS")) + { + return None; + } + // CREATE INDEX [name] ON table: the index name is optional. + if is(second, "INDEX") && !is(words.next(), "ON") && !is(words.next(), "ON") { + return None; + } + } else if is(first, "ALTER") && is(second, "TABLE") { + if words + .next_if(|&(_, w)| w.eq_ignore_ascii_case("IF")) + .is_some() + && !is(words.next(), "EXISTS") + { + return None; + } + } else if !(is(first, "INSERT") && is(second, "INTO")) { + return None; + } + let (at, table) = words.next()?; + let mut chars = table.chars(); + let unquoted = chars.next().is_some_and(|c| c.is_ascii_alphabetic()) + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_'); + unquoted.then_some(at) +} + +/// The words of `body` with their byte offsets, a word being a run of +/// characters ended by whitespace or an opening parenthesis. +fn words(body: &str) -> impl Iterator { + let mut at = 0; + std::iter::from_fn(move || { + let rest = &body[at..]; + let start = at + (rest.len() - rest.trim_start().len()); + let word = &body[start..]; + let end = word + .find(|c: char| c.is_whitespace() || c == '(') + .unwrap_or(word.len()); + at = start + end; + (end > 0).then(|| (start, &word[..end])) + }) +} + /// Record a migration in the `schema_history` table. async fn record_migration( session: &Arc, @@ -306,4 +412,144 @@ mod tests { ); assert!(super::migration_version("005_x.cql").is_err()); } + + #[test] + fn qualifies_each_statement_form_with_the_keyspace() { + for (statement, qualified) in [ + ( + "CREATE TABLE IF NOT EXISTS settings (\n key text PRIMARY KEY\n)", + "CREATE TABLE IF NOT EXISTS ks_a.settings (\n key text PRIMARY KEY\n)", + ), + ( + "create table accounts(account_id text PRIMARY KEY)", + "create table ks_a.accounts(account_id text PRIMARY KEY)", + ), + ( + "CREATE INDEX IF NOT EXISTS ON tables (table_id)", + "CREATE INDEX IF NOT EXISTS ON ks_a.tables (table_id)", + ), + ( + "CREATE INDEX ON tables (table_id)", + "CREATE INDEX ON ks_a.tables (table_id)", + ), + ( + "CREATE INDEX idx ON tables (table_id)", + "CREATE INDEX idx ON ks_a.tables (table_id)", + ), + ( + "CREATE INDEX IF NOT EXISTS idx_txn_started_at ON transaction_ledger (started_at)", + "CREATE INDEX IF NOT EXISTS idx_txn_started_at ON ks_a.transaction_ledger (started_at)", + ), + ( + "ALTER TABLE tables ADD IF NOT EXISTS ttl_backfill_cursor text", + "ALTER TABLE ks_a.tables ADD IF NOT EXISTS ttl_backfill_cursor text", + ), + ( + "ALTER TABLE IF EXISTS tables ADD IF NOT EXISTS c text", + "ALTER TABLE IF EXISTS ks_a.tables ADD IF NOT EXISTS c text", + ), + ( + "alter table if exists tables with comment = 'x'", + "alter table if exists ks_a.tables with comment = 'x'", + ), + ( + "-- Seed settings\n// two comments\nINSERT INTO settings (key) VALUES ('a')", + "-- Seed settings\n// two comments\nINSERT INTO ks_a.settings (key) VALUES ('a')", + ), + ] { + let got = + super::qualify_statement("ks_a", "V999__test.cql", statement).expect(statement); + assert_eq!(got, qualified); + } + } + + #[test] + fn rejects_statements_it_cannot_qualify() { + for statement in [ + "UPDATE settings SET value = 'x' WHERE key = 'y'", + "DROP TABLE settings", + "CREATE TABLE other.settings (key text PRIMARY KEY)", + "CREATE TABLE \"Settings\" (key text PRIMARY KEY)", + "CREATE TABLE IF EXISTS settings (key text PRIMARY KEY)", + "CREATE INDEX IF EXISTS idx ON tables (table_id)", + "ALTER TABLE IF NOT EXISTS tables ADD c text", + "CREATE INDEX idx settings (table_id)", + "-- a comment and no statement", + ] { + assert!( + super::qualify_statement("ks_a", "V999__test.cql", statement).is_err(), + "{statement}" + ); + } + } + + /// A semicolon inside a comment, or a comment after a file's last + /// statement, leaves a fragment that holds only a comment. + #[test] + fn a_fragment_of_only_a_comment_fails_naming_the_file() { + let sql = "CREATE TABLE t (k int PRIMARY KEY);\n-- a note after the last statement\n"; + let Err(super::OpError::Internal(message)) = + super::migration_statements("ks_a", "V999__note.cql", sql) + else { + panic!("a fragment that holds only a comment was accepted"); + }; + assert!( + message.contains("V999__note.cql") + && message.contains("holds only a comment") + && message.ends_with("-- a note after the last statement"), + "{message}" + ); + } + + /// Every statement of every embedded migration file, as the runner sends + /// it, is the file's statement with the keyspace written in once, in front + /// of its table: after the statement's leading comments, as a word of its + /// own after `ON` in a `CREATE INDEX`, after `INTO` in an `INSERT INTO`, + /// and after `TABLE` or `EXISTS` otherwise, and not in front of the `IF` + /// that opens an `IF EXISTS` or `IF NOT EXISTS` clause. + #[test] + fn every_migration_statement_names_the_keyspace() { + let files = super::CATALOG_MIGRATIONS + .iter() + .chain(super::DATA_MIGRATIONS); + for (filename, sql) in files { + let sent = super::migration_statements("ks_a", filename, sql).expect(filename); + let written: Vec<&str> = sql + .split(';') + .map(str::trim) + .filter(|s| !s.is_empty()) + .collect(); + assert_eq!(sent.len(), written.len(), "{filename}"); + for (sent, written) in sent.iter().zip(written) { + assert_eq!(sent.matches("ks_a.").count(), 1, "{filename}: {sent}"); + assert_eq!(sent.replacen("ks_a.", "", 1), written, "{filename}"); + let (before, after) = super::skip_comments(sent) + .split_once("ks_a.") + .unwrap_or_else(|| panic!("{filename}: keyspace in a leading comment: {sent}")); + let keyword = if before.ends_with(char::is_whitespace) { + before.split_whitespace().next_back().unwrap_or_default() + } else { + "" + }; + let table = after + .split(|c: char| c.is_whitespace() || c == '(') + .next() + .unwrap_or_default(); + let mut words = super::skip_comments(written) + .split_whitespace() + .map(str::to_ascii_uppercase); + let expected: &[&str] = match (words.next().as_deref(), words.next().as_deref()) { + (Some("CREATE"), Some("INDEX")) => &["ON"], + (Some("INSERT"), Some("INTO")) => &["INTO"], + _ => &["TABLE", "EXISTS"], + }; + assert!( + expected.iter().any(|k| keyword.eq_ignore_ascii_case(k)) + && !table.is_empty() + && !table.eq_ignore_ascii_case("IF"), + "{filename}: keyspace not in front of the table: {sent}" + ); + } + } + } } diff --git a/crates/storage-cassandra/tests/direct/backup_engine.rs b/crates/storage-cassandra/tests/direct/backup_engine.rs index 30fe919ac..3f0ef92de 100644 --- a/crates/storage-cassandra/tests/direct/backup_engine.rs +++ b/crates/storage-cassandra/tests/direct/backup_engine.rs @@ -1573,11 +1573,23 @@ async fn test_restore_of_deleted_backup_is_not_found() { /// Catalog migrations V001-V004: the catalog shape before V005 added /// `backups_by_arn.definition/active_restores` and `tables.restore_*`. -const CATALOG_THROUGH_V004: &[&str] = &[ - include_str!("../../migrations/catalog/V001__initial_schema.cql"), - include_str!("../../migrations/catalog/V002__backup_restore.cql"), - include_str!("../../migrations/catalog/V003__account_scoped_idempotency.cql"), - include_str!("../../migrations/catalog/V004__ttl_backfill_cursor.cql"), +const CATALOG_THROUGH_V004: &[(&str, &str)] = &[ + ( + "V001__initial_schema.cql", + include_str!("../../migrations/catalog/V001__initial_schema.cql"), + ), + ( + "V002__backup_restore.cql", + include_str!("../../migrations/catalog/V002__backup_restore.cql"), + ), + ( + "V003__account_scoped_idempotency.cql", + include_str!("../../migrations/catalog/V003__account_scoped_idempotency.cql"), + ), + ( + "V004__ttl_backfill_cursor.cql", + include_str!("../../migrations/catalog/V004__ttl_backfill_cursor.cql"), + ), ]; /// An engine on its own keyspace prefix (so its sweeps and workers touch no @@ -1610,11 +1622,17 @@ async fn isolated_engine( .unwrap(); } if through_v004 { - // Statement by statement, as the migration runner does, recording - // versions 1-4 in schema_history as the runner did for a 0.0.3 catalog. - session.query(format!("USE {catalog_ks}")).await.unwrap(); - for file in CATALOG_THROUGH_V004 { - for statement in file.split(';').map(str::trim).filter(|s| !s.is_empty()) { + // Statement by statement, each qualified with the catalog keyspace by + // the function the migration runner uses, recording versions 1-4 in + // schema_history as the runner did for a 0.0.3 catalog. + for (filename, sql) in CATALOG_THROUGH_V004 { + let statements = extenddb_storage_cassandra::migrations::migration_statements( + &catalog_ks, + filename, + sql, + ) + .expect(filename); + for statement in statements { session.query(statement).await.expect("V001-V004 statement"); } }