adapter: revalidate OR REPLACE drop set in staged create finish

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"

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<CatalogItemId>,
+        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<CatalogItemId> = self
+            .catalog()
+            .for_session(session)
+            .item_dependents(replace_id)
+            .into_iter()
+            .map(|id| id.unwrap_item_id())
+            .collect();
+        let planned: BTreeSet<CatalogItemId> = 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,

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 {

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<StageResult<Box<CreateViewStage>>, 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(

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}")