From 0b6a90970612afc9ef0a00c042a96c3be0c11a9e Mon Sep 17 00:00:00 2001 From: Gabor Gevay Date: Tue, 14 Jul 2026 12:55:04 +0200 Subject: [PATCH] adapter: revalidate OR REPLACE drop set in staged create finish CREATE OR REPLACE [MATERIALIZED] VIEW computes the set of items to drop (the item being replaced) at planning time, but sequencing is staged and commits that set only in the finish stage, after off-thread optimization. Most DDL cannot interleave because it queues on the coordinator DDL lock, but statements with off-thread purification (e.g. CREATE SINK) are exempt and only check that their dependencies still exist. A sink created on the replaced materialized view could therefore commit inside the window. The finish stage then dropped the old item anyway, leaving the sink with dependency edges to a missing item and panicking the coordinator catalog consistency check (SQL-521). The finish stages now recompute the replaced item dependent closure against the current catalog and abort with a retryable error if it changed since planning. This also covers the item being replaced disappearing entirely, e.g. through a concurrent ALTER MATERIALIZED VIEW APPLY REPLACEMENT, which is likewise exempt from the DDL lock. Add a create_materialized_view_optimize failpoint and a deterministic regression test in test/race-condition, wired into the Nightly pipeline. Co-Authored-By: Claude Fable 5 --- ci/nightly/pipeline.template.yml | 12 ++ src/adapter/src/coord/sequencer/inner.rs | 56 ++++++++ .../inner/create_materialized_view.rs | 18 +++ .../src/coord/sequencer/inner/create_view.rs | 6 + test/race-condition/mzcompose.py | 123 ++++++++++++++++++ 5 files changed, 215 insertions(+) diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml index 8f93e60bb08e8..5da328bea91e0 100644 --- a/ci/nightly/pipeline.template.yml +++ b/ci/nightly/pipeline.template.yml @@ -2845,6 +2845,18 @@ steps: composition: race-condition run: rotate-keys-race + - id: race-condition-create-or-replace + topics: [iceberg, kafka, mysql, postgres, sql-server] + label: "CREATE OR REPLACE MATERIALIZED VIEW vs CREATE SINK race" + depends_on: build-x86_64 + timeout_in_minutes: 15 + agents: + queue: hetzner-x86-64-4cpu-8gb + plugins: + - ./ci/plugins/mzcompose: + composition: race-condition + run: create-or-replace-race + - id: mz-deploy topics: [kafka, postgres] label: "mz-deploy" diff --git a/src/adapter/src/coord/sequencer/inner.rs b/src/adapter/src/coord/sequencer/inner.rs index dfd0b881b8797..1d1fb3ec31ca0 100644 --- a/src/adapter/src/coord/sequencer/inner.rs +++ b/src/adapter/src/coord/sequencer/inner.rs @@ -31,6 +31,7 @@ use mz_expr::{ use mz_ore::cast::CastFrom; use mz_ore::collections::{CollectionExt, HashSet}; use mz_ore::future::OreFutureExt; +use mz_ore::str::StrExt; use mz_ore::task::{self, JoinHandle, spawn}; use mz_ore::tracing::OpenTelemetryContext; use mz_ore::{assert_none, instrument}; @@ -1212,6 +1213,61 @@ impl Coordinator { } } + /// Re-validates a `CREATE OR REPLACE` plan's drop set against the current catalog. + /// + /// `drop_ids` (the item being replaced plus its dependents) is computed at planning + /// time, but `CREATE [MATERIALIZED] VIEW` sequencing is staged, so other DDL can + /// commit between planning and the finish stage. Most DDL is excluded from that + /// window by the coordinator's DDL lock, but statements with off-thread purification + /// (e.g. `CREATE SINK`, see `must_serialize_ddl`) do not take that lock and can add + /// a dependent on the item being replaced. Committing the stale drop set would leave + /// such a dependent referencing a dropped item and corrupt the catalog. + /// + /// Callers must invoke this in the same message handler that commits the ops, so + /// that no further DDL can interleave between this check and the commit. + pub(super) fn validate_or_replace_drop_ids( + &self, + session: &Session, + object_kind: &'static str, + name: &QualifiedItemName, + replace: Option, + drop_ids: &[CatalogItemId], + ) -> Result<(), AdapterError> { + let Some(replace_id) = replace else { + // Plans without `OR REPLACE` carry no drops. + return Ok(()); + }; + if self.catalog().try_get_entry(&replace_id).is_none() { + // The item being replaced was concurrently dropped (e.g. by an + // `ALTER MATERIALIZED VIEW ... APPLY REPLACEMENT`, which also does not take + // the DDL lock and removes the target's item id). + return Err(AdapterError::ConcurrentDependencyDrop { + dependency_kind: "catalog item", + dependency_id: replace_id.to_string(), + }); + } + let current: BTreeSet = self + .catalog() + .for_session(session) + .item_dependents(replace_id) + .into_iter() + .map(|id| id.unwrap_item_id()) + .collect(); + let planned: BTreeSet = drop_ids.iter().copied().collect(); + if current != planned { + let full_name = self + .catalog() + .resolve_full_name(name, Some(session.conn_id())); + return Err(AdapterError::ChangedPlan(format!( + "the set of objects that depend on {} {} changed while the statement \ + was executing", + object_kind, + full_name.to_string().quoted(), + ))); + } + Ok(()) + } + #[instrument] pub(super) async fn sequence_create_type( &mut self, diff --git a/src/adapter/src/coord/sequencer/inner/create_materialized_view.rs b/src/adapter/src/coord/sequencer/inner/create_materialized_view.rs index f27a47b07a5d1..421d28c589ae6 100644 --- a/src/adapter/src/coord/sequencer/inner/create_materialized_view.rs +++ b/src/adapter/src/coord/sequencer/inner/create_materialized_view.rs @@ -477,6 +477,12 @@ impl Coordinator { || "optimize create materialized view", move || { span.in_scope(|| { + // Lets tests pause here (off the coordinator main loop) to widen the + // window between planning, which computed `drop_ids` for + // `CREATE OR REPLACE`, and the finish stage that commits them. Used by + // test/race-condition. + fail::fail_point!("create_materialized_view_optimize"); + let mut pipeline = || -> Result<( optimize::materialized_view::LocalMirPlan, optimize::materialized_view::GlobalMirPlan, @@ -585,6 +591,7 @@ impl Coordinator { refresh_schedule, .. }, + replace, drop_ids, if_not_exists, .. @@ -597,6 +604,17 @@ impl Coordinator { .. } = stage; + // For `CREATE OR REPLACE`, re-validate the plan-time drop set. A statement not + // subject to the DDL lock (e.g. `CREATE SINK`) can have added a dependent on the + // item being replaced while the off-thread stages ran. + self.validate_or_replace_drop_ids( + ctx.session(), + "materialized view", + &name, + replace, + &drop_ids, + )?; + // Validate the replacement target, if one is given. if let Some(target_id) = replacement_target { let Some(target) = self.catalog().get_entry(&target_id).materialized_view() else { diff --git a/src/adapter/src/coord/sequencer/inner/create_view.rs b/src/adapter/src/coord/sequencer/inner/create_view.rs index 6d9348412178f..c20ddcd8fbc2b 100644 --- a/src/adapter/src/coord/sequencer/inner/create_view.rs +++ b/src/adapter/src/coord/sequencer/inner/create_view.rs @@ -392,6 +392,7 @@ impl Coordinator { column_names, temporary, }, + replace, drop_ids, if_not_exists, .. @@ -401,6 +402,11 @@ impl Coordinator { .. }: CreateViewFinish, ) -> Result>, AdapterError> { + // For `CREATE OR REPLACE`, re-validate the plan-time drop set. A statement not + // subject to the DDL lock (e.g. `CREATE SINK`) can have added a dependent on the + // item being replaced while the off-thread stages ran. + self.validate_or_replace_drop_ids(session, "view", &name, replace, &drop_ids)?; + let typ = infer_sql_type_for_catalog(&raw_expr, &optimized_expr); let ops = vec![ catalog::Op::DropObjects( diff --git a/test/race-condition/mzcompose.py b/test/race-condition/mzcompose.py index cfcc6cf68b12a..f5c39b38afc6a 100644 --- a/test/race-condition/mzcompose.py +++ b/test/race-condition/mzcompose.py @@ -1521,3 +1521,126 @@ def rotate() -> None: ) print(f"OK: ROTATE aborted with OCC conflict, SET preserved: {err_str}") + + +# Pauses CREATE MATERIALIZED VIEW in its off-thread optimize stage, between +# planning (which computes the `OR REPLACE` drop set) and the finish stage +# that commits it. Must fire off the coordinator main loop. Pausing on-loop +# would freeze the coordinator and block this test's other statements. +# Defined in src/adapter/src/coord/sequencer/inner/create_materialized_view.rs. +CREATE_MV_OPTIMIZE_FAILPOINT = "create_materialized_view_optimize" + +# Substring of the `AdapterError::ChangedPlan` message produced by +# `validate_or_replace_drop_ids` in src/adapter/src/coord/sequencer/inner.rs; +# what we look for in the CREATE OR REPLACE error to confirm the finish stage +# caught the concurrent CREATE SINK. +OR_REPLACE_CONFLICT_MESSAGE = "changed while the statement was executing" + + +def workflow_create_or_replace_race( + c: Composition, parser: WorkflowArgumentParser +) -> None: + """Regression test for SQL-521: CREATE OR REPLACE MATERIALIZED VIEW leaves + a concurrently created sink with dangling dependency edges. + + CREATE OR REPLACE MATERIALIZED VIEW computes its drop set (the item id + being replaced) at planning time, then optimizes off-thread and commits the + drop in the staged finish. Most DDL cannot interleave because it queues on + the coordinator's `serialized_ddl` lock, but CREATE SINK is exempt (it + purifies off-thread, see `must_serialize_ddl` in `command_handler.rs`) and + only checks via `PlanValidity` that its dependencies still exist. A sink on + the MV that commits inside the window is thus invisible to the plan-time + drop set, and the finish drops the old MV item anyway. The sink's `uses` + and `references` edges then point at a dropped item, and the catalog + consistency check panics the coordinator on the very transact that commits + the replacement. + + After the fix, the finish stage revalidates the drop set against the + current catalog and the CREATE OR REPLACE fails with a retryable error + instead. We force the interleaving deterministically with the + `create_materialized_view_optimize` failpoint, which pauses the replacement + in its off-thread optimize stage. + """ + parser.parse_args() + + c.up("materialized", "kafka") + + c.sql(dedent(""" + CREATE TABLE sql521_t (a int, b int); + CREATE MATERIALIZED VIEW sql521_mv AS SELECT a, b FROM sql521_t; + CREATE CONNECTION sql521_kafka_conn + TO KAFKA (BROKER 'kafka:9092', SECURITY PROTOCOL PLAINTEXT); + """)) + + # `SET failpoints` calls `fail::cfg` (see src/sql/src/session/vars/value.rs), + # which writes to the process-global registry. The pause applies to any + # session that later hits the named failpoint. + c.sql(f"SET failpoints = '{CREATE_MV_OPTIMIZE_FAILPOINT}=pause'") + + replace_err: Exception | None = None + + def replace() -> None: + nonlocal replace_err + try: + with c.sql_cursor(reuse_connection=False) as cur: + cur.execute( + b"CREATE OR REPLACE MATERIALIZED VIEW sql521_mv" + b" AS SELECT a + 1 AS a, b FROM sql521_t" + ) + except Exception as e: + replace_err = e + + t_replace = PropagatingThread(target=replace, name="replace") + t_replace.start() + + # Give the replacement time to plan (with an empty dependent set for the + # old MV) and park at the failpoint in its optimize stage. + time.sleep(1) + + # Commit a new dependent on the old MV item while the replacement is + # parked. CREATE SINK does not take the DDL lock, so it does not queue + # behind the in-flight CREATE OR REPLACE. + c.sql(dedent(""" + CREATE SINK sql521_snk FROM sql521_mv + INTO KAFKA CONNECTION sql521_kafka_conn (TOPIC 'sql521-snk') + FORMAT JSON ENVELOPE DEBEZIUM + """)) + + # Release the replacement. Its finish stage should now detect the sink and + # abort instead of dropping the old MV item out from under it. + c.sql(f"SET failpoints = '{CREATE_MV_OPTIMIZE_FAILPOINT}=off'") + + t_replace.join() + + if replace_err is None: + # The replacement committing means it dropped the old MV item while the + # sink still references it, which is the original bug. On soft-assert builds + # the coordinator panics on that very transact, so usually this shows + # up as a lost connection instead (handled below). + raise Exception( + "race reproduced: CREATE OR REPLACE succeeded despite a " + "concurrently created sink on the replaced materialized view. " + "This is SQL-521: the sink is left with dangling dependency edges." + ) + + err_str = str(replace_err) + if OR_REPLACE_CONFLICT_MESSAGE not in err_str: + raise Exception( + f"CREATE OR REPLACE failed but not with the expected conflict " + f"error (a lost connection means the coordinator panicked on the " + f"corrupted catalog): {replace_err}" + ) + + # The old MV and the sink should both have survived intact. Any DDL then + # re-runs the coordinator's consistency check (soft-assert builds), which + # would panic if the catalog had been corrupted. + rows = c.sql_query("SHOW CREATE MATERIALIZED VIEW sql521_mv") + create_sql = rows[0][1] + if "a + 1" in create_sql: + raise Exception(f"replacement was applied despite erroring:\n {create_sql}") + rows = c.sql_query("SELECT count(*) FROM mz_sinks WHERE name = 'sql521_snk'") + if rows[0][0] != 1: + raise Exception("sink sql521_snk went missing") + c.sql("CREATE TABLE sql521_probe (x int); DROP TABLE sql521_probe") + + print(f"OK: CREATE OR REPLACE aborted, sink preserved: {err_str}")