From 4e17335dc3255163ea219413ef72905f36efa09c Mon Sep 17 00:00:00 2001 From: Tony Du Date: Fri, 28 Aug 2026 22:59:18 -0400 Subject: [PATCH 1/3] Authorize source access before purification Fixes SQL-660. ### Motivation `CREATE TABLE ... FROM SOURCE` and `ALTER SOURCE` are purified off-thread before they are planned. The only privilege check that ran before purification was `rbac::check_usage(.., &CREATE_ITEM_USAGE)`, which requires `USAGE` on the `Secret`, `Connection` and `Type` items the statement names. These statements name neither a secret nor a connection: they reach the upstream through an existing source, whose connection comes from its own `source_desc()` rather than from the statement. So nothing was required. Purification then opens that connection with the source owner's credentials and enumerates upstream objects. A role holding no privilege on the source could therefore make Materialize dial the source's upstream, and read upstream schema, table and column names out of the resulting purification errors. SQL-655 (#38480) closed the plan-time bypass, so no rows are readable; this is the residual that fix named. `CREATE SOURCE ... FROM CONNECTION` and `CREATE SINK ... INTO` are unaffected: they name their connection, so the existing usage requirement gates them. ### Description Replace the pre-purification `check_usage` call with `rbac::check_purification`, which builds a single `RbacRequirements` for the statement and delegates to the same validation path `check_plan` uses: * the existing `CREATE_ITEM_USAGE` usage requirements, unchanged; * for `ALTER SOURCE`: ownership of the named source; * for `CREATE TABLE ... FROM SOURCE`: read privileges on the source (`SELECT` plus schema `USAGE`), required of the owner too, since an owner's `SELECT` is an ordinary revocable grant and schema `USAGE` is separate from ownership. Both mirror what planning requires later, so a statement that passes here can still be rejected by `check_plan`, never the reverse. The source is resolved from the statement rather than from `resolved_ids`. `AlterSourceStatement::source_name` is an `UnresolvedItemName`, so name resolution never records it and a `resolved_ids`-based check is silently a no-op for `ALTER SOURCE`. Resolution mirrors purification exactly, so the check gates the item that would be dialed. The resolved source id is also added to the purified statement's dependency set. It was previously absent for `ALTER SOURCE`, so a source dropped concurrently with off-thread purification passed the validity check and then panicked the coordinator on the missing catalog entry ("catalog out of sync") during planning. With the id tracked, the drop is detected and the statement is repurified, ending in a clean unknown-item error. ### Verification `test/sqllogictest/rbac_create_table_from_source.slt` gains cases that pin the ordering, not just the denial. Each uses a statement whose purification fails for a non-permission reason, so a permission error can only come from the pre-purification check: * `CREATE TABLE ... FROM SOURCE` with an unresolvable reference: purification reports whether a reference exists upstream, so without the gate this leaks reference existence. * The same statement run by the source's owner after their `SELECT` is revoked, pinning that ownership does not stand in for read privileges. * `ALTER SOURCE ... ADD SUBSOURCE` on a load generator, which purification rejects with "does not support ALTER SOURCE": an ownership error can only come from the earlier gate. Plus: an owner-succeeds case, so the rule gates the caller rather than the syntax; a schema-`USAGE` denial, pinning the other half of the read requirement; and pass-through cases for a superuser and for an RBAC-disabled deployment, each reaching purification's own error on a source they do not own, pinning that the gate filters requirements the same way `check_plan` does. --- src/adapter/src/coord/command_handler.rs | 27 ++-- src/sql/src/rbac.rs | 80 +++++++++++- .../rbac_create_table_from_source.slt | 115 ++++++++++++++++++ 3 files changed, 210 insertions(+), 12 deletions(-) diff --git a/src/adapter/src/coord/command_handler.rs b/src/adapter/src/coord/command_handler.rs index 0d29050f5c447..35fd80ae0aca6 100644 --- a/src/adapter/src/coord/command_handler.rs +++ b/src/adapter/src/coord/command_handler.rs @@ -49,7 +49,6 @@ use mz_sql::pure::{ materialized_view_option_contains_temporal, purify_create_materialized_view_options, }; use mz_sql::rbac; -use mz_sql::rbac::CREATE_ITEM_USAGE; use mz_sql::session::user::User; use mz_sql::session::vars::{ EndTransactionAction, NETWORK_POLICY, OwnedVarInput, STATEMENT_LOGGING_SAMPLE_RATE, @@ -1593,17 +1592,18 @@ impl Coordinator { task::spawn(|| format!("purify:{conn_id}"), async move { let conn_catalog = catalog.for_session(ctx.session()); - // Checks if the session is authorized to purify a statement. Usually - // authorization is checked after planning, however purification happens before - // planning, which may require the use of some connections and secrets. - if let Err(e) = rbac::check_usage( + // Authorization is usually checked after planning, but purification + // happens before planning and may use connections and secrets, so it + // gets its own check. + let source_dependency = match rbac::check_purification( &conn_catalog, ctx.session(), + &stmt, &resolved_ids, - &CREATE_ITEM_USAGE, ) { - return ctx.retire(Err(e.into())); - } + Ok(id) => id, + Err(e) => return ctx.retire(Err(e.into())), + }; let (result, cluster_id) = mz_sql::pure::purify_statement( conn_catalog, @@ -1613,7 +1613,12 @@ impl Coordinator { ) .await; let result = result.map_err(|e| e.into()); - let dependency_ids = resolved_ids.items().copied().collect(); + // The resolved source of an `ALTER SOURCE` is not in + // `resolved_ids`, so add it here: a source dropped while + // purification ran off-thread must invalidate the result + // rather than surface later as a missing catalog entry. + let mut dependency_ids: BTreeSet<_> = resolved_ids.items().copied().collect(); + dependency_ids.extend(source_dependency); let plan_validity = PlanValidity::new( &catalog, dependency_ids, @@ -1817,6 +1822,10 @@ impl Coordinator { } /// Whether the statement must be purified off of the Coordinator thread. + // Every statement listed here is authorized by `rbac::check_purification` + // before purification runs. A new variant whose purification reaches an + // upstream system through an *existing* catalog item (rather than a named + // connection) needs its own arm there, or it is gated by nothing. fn must_spawn_purification(stmt: &Statement) -> bool { // `CREATE` and `ALTER` `SOURCE` and `SINK` statements must be purified off the main // coordinator thread. diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs index cd1bec5d67f14..0a18d64ff5ccc 100644 --- a/src/sql/src/rbac.rs +++ b/src/sql/src/rbac.rs @@ -19,14 +19,14 @@ use mz_ore::str::StrExt; use mz_repr::CatalogItemId; use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem}; use mz_repr::role_id::RoleId; -use mz_sql_parser::ast::{Ident, QualifiedReplica}; +use mz_sql_parser::ast::{Ident, QualifiedReplica, RawItemName, Statement}; use tracing::debug; use crate::catalog::{ CatalogItemType, ErrorMessageObjectDescription, ObjectType, SessionCatalog, SystemObjectType, }; use crate::names::{ - CommentObjectId, ObjectId, QualifiedItemName, ResolvedDatabaseSpecifier, ResolvedIds, + Aug, CommentObjectId, ObjectId, QualifiedItemName, ResolvedDatabaseSpecifier, ResolvedIds, SchemaSpecifier, SystemObjectId, }; use crate::plan::{self, PlanKind}; @@ -398,6 +398,76 @@ pub fn check_usage( Ok(()) } +/// Authorizes a statement for purification, which runs before planning and, +/// for statements that reach an upstream system through an existing source, +/// borrows that source's own connection. The source is resolved from the +/// statement because `ALTER SOURCE` carries its target as an unresolved name +/// that `resolved_ids` never contains; resolution mirrors what purification +/// itself does, so this gates exactly the item that would be dialed. The +/// requirements are the same ones planning enforces later (ownership for +/// `ALTER SOURCE`, read for `CREATE TABLE ... FROM SOURCE`), so a statement +/// that passes here can still be rejected by [`check_plan`], never the reverse. +/// +/// Returns the resolved source's id so the caller can track it as a dependency +/// of the about-to-be-purified statement; name resolution does not record it +/// for `ALTER SOURCE`. +pub fn check_purification( + catalog: &impl SessionCatalog, + session: &dyn SessionMetadata, + stmt: &Statement, + resolved_ids: &ResolvedIds, +) -> Result, UnauthorizedError> { + // Like `check_plan`: `validate` reads the current role's membership through + // the panicking `get_role`, so the concurrent-role-drop case must be turned + // into a clean error before anything else runs. + rbac_check_preamble(catalog, session)?; + + let role_id = session.role_metadata().current_role; + let scx = crate::plan::StatementContext::new(None, catalog); + + // A name that does not resolve is left to purification to report: failing + // to resolve leaks nothing, and both purify paths bail before any + // connection is opened. + let source = match stmt { + Statement::AlterSource(stmt) => scx + .resolve_item(RawItemName::Name(stmt.source_name.clone())) + .ok() + .map(|item| (item.id(), item.item_type(), true)), + Statement::CreateTableFromSource(stmt) => scx + .get_item_by_resolved_name(&stmt.source) + .ok() + .map(|item| (item.id(), item.item_type(), false)), + // Other purified statements (`CREATE SOURCE`, `CREATE SINK`) name + // their connection, which the `item_usage` requirement below covers. + _ => None, + }; + + let mut requirements = RbacRequirements { + item_usage: &CREATE_ITEM_USAGE, + ..Default::default() + }; + let source_id = match source { + // Only a `Source` can carry a source description for purification to + // borrow a connection from; every other item type errors there before + // dialing out. + Some((id, CatalogItemType::Source, requires_ownership)) => { + if requires_ownership { + requirements.ownership = vec![ObjectId::Item(id)]; + } else { + requirements.privileges = + generate_read_privileges(catalog, iter::once(id), role_id); + } + Some(id) + } + _ => None, + }; + + let requirements = filter_requirements(catalog, session, requirements); + requirements.validate(catalog, session, resolved_ids)?; + + Ok(source_id) +} + /// Checks if a session is authorized to execute a plan. If not, an error is returned. /// /// `sql_impl_resolved_ids` contains resolved IDs discovered inside SQL-implemented function @@ -645,7 +715,9 @@ fn generate_rbac_requirements( role_id, )]; // `CREATE TABLE ... FROM SOURCE` reads the source's data, so it - // requires `SELECT` on the source. + // requires `SELECT` on the source. `check_purification` enforces + // the same read requirement before purification; keep them in sync + // or its subset invariant inverts. if let TableDataSource::DataSource { desc: DataSourceDesc::IngestionExport { ingestion_id, .. }, timeline: _, @@ -1169,6 +1241,8 @@ fn generate_rbac_requirements( ownership: vec![ObjectId::Item(*id)], ..Default::default() }, + // `check_purification` enforces the same ownership requirement before + // purification; keep them in sync or its subset invariant inverts. Plan::AlterSource(plan::AlterSourcePlan { item_id, ingestion_id: _, diff --git a/test/sqllogictest/rbac_create_table_from_source.slt b/test/sqllogictest/rbac_create_table_from_source.slt index 9459abfdf0c20..83263b0661fde 100644 --- a/test/sqllogictest/rbac_create_table_from_source.slt +++ b/test/sqllogictest/rbac_create_table_from_source.slt @@ -10,6 +10,9 @@ # `CREATE TABLE ... FROM SOURCE` reads the source's data, so it requires the # same read privileges as reading the source directly: `SELECT` on the source # and `USAGE` on its schema. +# +# Those privileges are also required *before* purification, which runs ahead of +# planning and borrows the source's own connection to contact its upstream. mode cockroach @@ -165,3 +168,115 @@ simple conn=other_team,user=other_team CREATE TABLE other_team_schema.mine FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."users"); ---- COMPLETE 0 + +# Purification runs before planning and reaches the source's upstream, so the +# read requirement is enforced before it, not only at plan time. +# +# An unresolvable reference is the discriminator: purification reports whether a +# reference exists upstream, so if it ran first this would return +# "reference to ... not found" and leak that fact to a role with no privilege on +# the source. A permission error means the check fired first. + +simple conn=mz_system,user=mz_system +REVOKE SELECT ON victim_schema.victim_auction FROM attacker; +---- +COMPLETE 0 + +simple conn=attacker,user=attacker +CREATE TABLE attacker_schema.probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist"); +---- +db error: ERROR: permission denied for SOURCE "materialize.victim_schema.victim_auction" +DETAIL: The 'attacker' role needs SELECT privileges on SOURCE "materialize.victim_schema.victim_auction" + +# `ALTER SOURCE` reaches the same purification path but carries its target as an +# unresolved name, so it is not covered by the resolved-id check and needs the +# source resolved from the statement. It mutates the source, so it requires +# ownership. +# +# `ADD SUBSOURCE` is the discriminator here: purification rejects it on a load +# generator with "does not support ALTER SOURCE", so an ownership error can only +# come from the earlier gate. `REFRESH REFERENCES` would not discriminate, since +# it purifies successfully on a load generator and plan-time ownership produces +# the identical message. + +simple conn=attacker,user=attacker +ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t; +---- +db error: ERROR: must be owner of SOURCE materialize.victim_schema.victim_auction + +# The owner is unaffected on both paths. + +simple conn=mz_system,user=mz_system +ALTER SOURCE victim_schema.victim_auction OWNER TO attacker; +---- +COMPLETE 0 + +simple conn=attacker,user=attacker +ALTER SOURCE victim_schema.victim_auction REFRESH REFERENCES; +---- +COMPLETE 0 + +# Ownership does not stand in for read privileges: an owner's `SELECT` is an +# ordinary revocable grant, and schema `USAGE` is separate from ownership, so +# the pre-purification check must require the read even from the owner. The +# attacker owns the source here but has had `SELECT` revoked; the invalid +# reference again discriminates, since a reference-not-found error would mean +# purification ran and leaked reference existence to a role planning would +# reject. + +simple conn=mz_system,user=mz_system +REVOKE SELECT ON victim_schema.victim_auction FROM attacker; +---- +COMPLETE 0 + +simple conn=attacker,user=attacker +CREATE TABLE attacker_schema.owner_probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist"); +---- +db error: ERROR: permission denied for SOURCE "materialize.victim_schema.victim_auction" +DETAIL: The 'attacker' role needs SELECT privileges on SOURCE "materialize.victim_schema.victim_auction" + +# Schema `USAGE` is the other half of the read requirement. The attacker gets +# `SELECT` back but loses `USAGE` on the schema; the denial must move to the +# schema, again ahead of purification. + +simple conn=mz_system,user=mz_system +GRANT SELECT ON victim_schema.victim_auction TO attacker; +---- +COMPLETE 0 + +simple conn=mz_system,user=mz_system +REVOKE USAGE ON SCHEMA victim_schema FROM attacker; +---- +COMPLETE 0 + +simple conn=attacker,user=attacker +CREATE TABLE attacker_schema.schema_probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist"); +---- +db error: ERROR: permission denied for SCHEMA "materialize.victim_schema" +DETAIL: The 'attacker' role needs USAGE privileges on SCHEMA "materialize.victim_schema" + +# Superusers pass through the gate: mz_system does not own the source (the +# attacker does), so reaching purification's "does not support ALTER SOURCE" +# error proves the ownership requirement was filtered, not enforced. + +simple conn=mz_system,user=mz_system +ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t; +---- +db error: ERROR: source victim_schema.victim_auction does not support ALTER SOURCE. + +# With RBAC checks disabled, a non-owner passes through the same way. + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_rbac_checks TO false; +---- +COMPLETE 0 + +simple conn=no_create,user=no_create +ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t; +---- +db error: ERROR: source victim_schema.victim_auction does not support ALTER SOURCE. + +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET enable_rbac_checks TO true; +---- +COMPLETE 0 From 337cd21cf7ac6ab089ee1262e6c3014a5a434f90 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Thu, 3 Sep 2026 15:49:12 +0200 Subject: [PATCH 2/3] sql: resolve the statement's source once, outside the rbac check `check_purification` both authorized a statement and returned the source id its caller needed for dependency tracking, so a function named "check" handed back a catalog id and the resolution rationale lived in an rbac doc comment. Move the resolution to `pure::statement_source`, next to the purification paths it has to mirror, and give it a `StatementSource` result that says how the statement uses the source rather than a bare id plus a bool. The rbac check now takes that and maps it to requirements, and the call site feeds the same value to both the check and the dependency set. No behavior change. Co-Authored-By: Claude Opus 5 (1M context) --- src/adapter/src/coord/command_handler.rs | 30 +++++----- src/sql/src/pure.rs | 49 ++++++++++++++++ src/sql/src/rbac.rs | 72 +++++++----------------- 3 files changed, 85 insertions(+), 66 deletions(-) diff --git a/src/adapter/src/coord/command_handler.rs b/src/adapter/src/coord/command_handler.rs index 35fd80ae0aca6..7d550126b2636 100644 --- a/src/adapter/src/coord/command_handler.rs +++ b/src/adapter/src/coord/command_handler.rs @@ -1592,18 +1592,19 @@ impl Coordinator { task::spawn(|| format!("purify:{conn_id}"), async move { let conn_catalog = catalog.for_session(ctx.session()); + let statement_source = mz_sql::pure::statement_source(&conn_catalog, &stmt); + // Authorization is usually checked after planning, but purification // happens before planning and may use connections and secrets, so it // gets its own check. - let source_dependency = match rbac::check_purification( + if let Err(e) = rbac::check_purification( &conn_catalog, ctx.session(), - &stmt, + statement_source, &resolved_ids, ) { - Ok(id) => id, - Err(e) => return ctx.retire(Err(e.into())), - }; + return ctx.retire(Err(e.into())); + } let (result, cluster_id) = mz_sql::pure::purify_statement( conn_catalog, @@ -1613,12 +1614,13 @@ impl Coordinator { ) .await; let result = result.map_err(|e| e.into()); - // The resolved source of an `ALTER SOURCE` is not in - // `resolved_ids`, so add it here: a source dropped while - // purification ran off-thread must invalidate the result - // rather than surface later as a missing catalog entry. + // `ALTER SOURCE` carries its target as an unresolved name, so name + // resolution never records it and `resolved_ids` does not cover it. + // Without it a source dropped while purification ran off-thread + // passes the validity check below and then surfaces as a missing + // catalog entry during planning. let mut dependency_ids: BTreeSet<_> = resolved_ids.items().copied().collect(); - dependency_ids.extend(source_dependency); + dependency_ids.extend(statement_source.map(|source| source.id())); let plan_validity = PlanValidity::new( &catalog, dependency_ids, @@ -1822,10 +1824,10 @@ impl Coordinator { } /// Whether the statement must be purified off of the Coordinator thread. - // Every statement listed here is authorized by `rbac::check_purification` - // before purification runs. A new variant whose purification reaches an - // upstream system through an *existing* catalog item (rather than a named - // connection) needs its own arm there, or it is gated by nothing. + /// + /// Every statement listed here is authorized by [`rbac::check_purification`] + /// before purification runs, against the source that + /// [`mz_sql::pure::statement_source`] resolves for it. fn must_spawn_purification(stmt: &Statement) -> bool { // `CREATE` and `ALTER` `SOURCE` and `SINK` statements must be purified off the main // coordinator thread. diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index 25c982728708a..801bb8d2b8257 100644 --- a/src/sql/src/pure.rs +++ b/src/sql/src/pure.rs @@ -273,6 +273,55 @@ pub enum PurifiedExportDetails { }, } +/// The existing source a statement drives, and how the statement uses it. +/// +/// Resolved from the statement rather than from `ResolvedIds`: `ALTER SOURCE` +/// carries its target as an `UnresolvedItemName` that name resolution never +/// records. +#[derive(Debug, Clone, Copy)] +pub enum StatementSource { + Altered(CatalogItemId), + Read(CatalogItemId), +} + +impl StatementSource { + pub fn id(&self) -> CatalogItemId { + match self { + StatementSource::Altered(id) | StatementSource::Read(id) => *id, + } + } +} + +/// Resolves the existing source that `stmt` drives, mirroring how the +/// purification paths below resolve it themselves. +/// +/// `None` means the statement drives no such source: it names its connection +/// outright (`CREATE SOURCE`, `CREATE SINK`), or the name does not resolve to a +/// source. Purification reports the latter itself, and bails on it before +/// opening any connection. A [`Statement`] variant that instead reaches +/// upstream through an existing catalog item needs an arm here, or callers see +/// no source at all. +pub fn statement_source( + catalog: &impl SessionCatalog, + stmt: &Statement, +) -> Option { + let scx = StatementContext::new(None, catalog); + match stmt { + Statement::AlterSource(stmt) => { + let item = scx + .resolve_item(RawItemName::Name(stmt.source_name.clone())) + .ok()?; + (item.item_type() == CatalogItemType::Source) + .then(|| StatementSource::Altered(item.id())) + } + Statement::CreateTableFromSource(stmt) => { + let item = scx.get_item_by_resolved_name(&stmt.source).ok()?; + (item.item_type() == CatalogItemType::Source).then(|| StatementSource::Read(item.id())) + } + _ => None, + } +} + /// Purifies a statement, removing any dependencies on external state. /// /// See the section on [purification](crate#purification) in the crate diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs index 0a18d64ff5ccc..5ab8465256550 100644 --- a/src/sql/src/rbac.rs +++ b/src/sql/src/rbac.rs @@ -19,14 +19,14 @@ use mz_ore::str::StrExt; use mz_repr::CatalogItemId; use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem}; use mz_repr::role_id::RoleId; -use mz_sql_parser::ast::{Ident, QualifiedReplica, RawItemName, Statement}; +use mz_sql_parser::ast::{Ident, QualifiedReplica}; use tracing::debug; use crate::catalog::{ CatalogItemType, ErrorMessageObjectDescription, ObjectType, SessionCatalog, SystemObjectType, }; use crate::names::{ - Aug, CommentObjectId, ObjectId, QualifiedItemName, ResolvedDatabaseSpecifier, ResolvedIds, + CommentObjectId, ObjectId, QualifiedItemName, ResolvedDatabaseSpecifier, ResolvedIds, SchemaSpecifier, SystemObjectId, }; use crate::plan::{self, PlanKind}; @@ -34,6 +34,7 @@ use crate::plan::{ DataSourceDesc, Explainee, MutationKind, Plan, SideEffectingFunc, TableDataSource, UpdatePrivilege, }; +use crate::pure::StatementSource; use crate::session::metadata::SessionMetadata; use crate::session::user::{MZ_SUPPORT_ROLE_ID, MZ_SYSTEM_ROLE_ID, SUPPORT_USER, SYSTEM_USER}; use crate::session::vars::SystemVars; @@ -398,74 +399,41 @@ pub fn check_usage( Ok(()) } -/// Authorizes a statement for purification, which runs before planning and, -/// for statements that reach an upstream system through an existing source, -/// borrows that source's own connection. The source is resolved from the -/// statement because `ALTER SOURCE` carries its target as an unresolved name -/// that `resolved_ids` never contains; resolution mirrors what purification -/// itself does, so this gates exactly the item that would be dialed. The -/// requirements are the same ones planning enforces later (ownership for -/// `ALTER SOURCE`, read for `CREATE TABLE ... FROM SOURCE`), so a statement -/// that passes here can still be rejected by [`check_plan`], never the reverse. +/// Authorizes a statement for purification, which runs before planning and so +/// escapes the [`check_plan`] gate. `source` is the existing source the +/// statement would drive, as resolved by [`crate::pure::statement_source`]. /// -/// Returns the resolved source's id so the caller can track it as a dependency -/// of the about-to-be-purified statement; name resolution does not record it -/// for `ALTER SOURCE`. +/// The requirements are the same ones planning enforces later, so a statement +/// that passes here can still be rejected by [`check_plan`], never the reverse. pub fn check_purification( catalog: &impl SessionCatalog, session: &dyn SessionMetadata, - stmt: &Statement, + source: Option, resolved_ids: &ResolvedIds, -) -> Result, UnauthorizedError> { +) -> Result<(), UnauthorizedError> { // Like `check_plan`: `validate` reads the current role's membership through // the panicking `get_role`, so the concurrent-role-drop case must be turned // into a clean error before anything else runs. rbac_check_preamble(catalog, session)?; let role_id = session.role_metadata().current_role; - let scx = crate::plan::StatementContext::new(None, catalog); - - // A name that does not resolve is left to purification to report: failing - // to resolve leaks nothing, and both purify paths bail before any - // connection is opened. - let source = match stmt { - Statement::AlterSource(stmt) => scx - .resolve_item(RawItemName::Name(stmt.source_name.clone())) - .ok() - .map(|item| (item.id(), item.item_type(), true)), - Statement::CreateTableFromSource(stmt) => scx - .get_item_by_resolved_name(&stmt.source) - .ok() - .map(|item| (item.id(), item.item_type(), false)), - // Other purified statements (`CREATE SOURCE`, `CREATE SINK`) name - // their connection, which the `item_usage` requirement below covers. - _ => None, - }; - let mut requirements = RbacRequirements { item_usage: &CREATE_ITEM_USAGE, ..Default::default() }; - let source_id = match source { - // Only a `Source` can carry a source description for purification to - // borrow a connection from; every other item type errors there before - // dialing out. - Some((id, CatalogItemType::Source, requires_ownership)) => { - if requires_ownership { - requirements.ownership = vec![ObjectId::Item(id)]; - } else { - requirements.privileges = - generate_read_privileges(catalog, iter::once(id), role_id); - } - Some(id) + match source { + Some(StatementSource::Altered(id)) => { + requirements.ownership = vec![ObjectId::Item(id)]; } - _ => None, - }; + Some(StatementSource::Read(id)) => { + requirements.privileges = generate_read_privileges(catalog, iter::once(id), role_id); + } + // Statements that name their connection are covered by `item_usage`. + None => {} + } let requirements = filter_requirements(catalog, session, requirements); - requirements.validate(catalog, session, resolved_ids)?; - - Ok(source_id) + requirements.validate(catalog, session, resolved_ids) } /// Checks if a session is authorized to execute a plan. If not, an error is returned. From c2cda8970b3ecb78398a83f9265a403810e1a8a0 Mon Sep 17 00:00:00 2001 From: Tony Du Date: Thu, 3 Sep 2026 11:29:12 -0400 Subject: [PATCH 3/3] Split the ALTER SOURCE dependency-tracking fix into its own PR Per review: it is an independent bug (coordinator panic on a concurrent drop during off-thread purification) that shares only the resolution. --- src/adapter/src/coord/command_handler.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/src/adapter/src/coord/command_handler.rs b/src/adapter/src/coord/command_handler.rs index 7d550126b2636..0ca17a4327cd1 100644 --- a/src/adapter/src/coord/command_handler.rs +++ b/src/adapter/src/coord/command_handler.rs @@ -1614,13 +1614,7 @@ impl Coordinator { ) .await; let result = result.map_err(|e| e.into()); - // `ALTER SOURCE` carries its target as an unresolved name, so name - // resolution never records it and `resolved_ids` does not cover it. - // Without it a source dropped while purification ran off-thread - // passes the validity check below and then surfaces as a missing - // catalog entry during planning. - let mut dependency_ids: BTreeSet<_> = resolved_ids.items().copied().collect(); - dependency_ids.extend(statement_source.map(|source| source.id())); + let dependency_ids = resolved_ids.items().copied().collect(); let plan_validity = PlanValidity::new( &catalog, dependency_ids,