mz_adapter/coord/
catalog_serving.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Special cases related to the "catalog serving" of Materialize
11//!
12//! Every Materialize deployment has a pre-installed [`mz_catalog_server`] cluster, which
13//! has several indexes to speed up common catalog queries. We also have a special
14//! `mz_support` role, which can be used by support teams to diagnose a deployment.
15//! For each of these use cases, we have some special restrictions we want to apply. The
16//! logic around these restrictions is defined here.
17//!
18//!
19//! [`mz_catalog_server`]: https://materialize.com/docs/sql/show-clusters/#mz_catalog_server-system-cluster
20
21use mz_expr::CollectionPlan;
22use mz_repr::GlobalId;
23use mz_repr::namespaces::is_system_schema;
24use mz_sql::catalog::SessionCatalog;
25use mz_sql::plan::{
26    ExplainPlanPlan, ExplainTimestampPlan, Explainee, ExplaineeStatement, Plan, SubscribeFrom,
27    SubscribePlan,
28};
29use smallvec::SmallVec;
30
31use crate::AdapterError;
32use crate::catalog::ConnCatalog;
33use crate::coord::TargetCluster;
34use crate::notice::AdapterNotice;
35use crate::session::Session;
36use mz_catalog::builtin::MZ_CATALOG_SERVER_CLUSTER;
37
38/// Checks whether or not we should automatically run a query on the `mz_catalog_server`
39/// cluster, as opposed to whatever the current default cluster is.
40pub fn auto_run_on_catalog_server<'a, 's, 'p>(
41    catalog: &'a ConnCatalog<'a>,
42    session: &'s Session,
43    plan: &'p Plan,
44) -> TargetCluster {
45    let inspect_subscribe = |plan: &SubscribePlan| {
46        (
47            plan.from.depends_on(),
48            match &plan.from {
49                SubscribeFrom::Id(_) => false,
50                SubscribeFrom::Query { expr, desc: _ } => expr.could_run_expensive_function(),
51            },
52        )
53    };
54
55    let (depends_on, could_run_expensive_function) = match plan {
56        Plan::Select(plan) => (
57            plan.source.depends_on(),
58            plan.source.could_run_expensive_function(),
59        ),
60        Plan::ShowColumns(plan) => (
61            plan.select_plan.source.depends_on(),
62            plan.select_plan.source.could_run_expensive_function(),
63        ),
64        Plan::Subscribe(plan) => inspect_subscribe(plan),
65        Plan::ExplainPlan(ExplainPlanPlan {
66            explainee: Explainee::Statement(ExplaineeStatement::Select { plan, .. }),
67            ..
68        }) => (
69            plan.source.depends_on(),
70            plan.source.could_run_expensive_function(),
71        ),
72        Plan::ExplainPlan(ExplainPlanPlan {
73            explainee: Explainee::Statement(ExplaineeStatement::Subscribe { plan, .. }),
74            ..
75        }) => inspect_subscribe(plan),
76        Plan::ExplainTimestamp(ExplainTimestampPlan { raw_plan, .. }) => (
77            raw_plan.depends_on(),
78            raw_plan.could_run_expensive_function(),
79        ),
80        Plan::CreateConnection(_)
81        | Plan::CreateDatabase(_)
82        | Plan::CreateSchema(_)
83        | Plan::CreateRole(_)
84        | Plan::CreateNetworkPolicy(_)
85        | Plan::CreateCluster(_)
86        | Plan::CreateClusterReplica(_)
87        | Plan::CreateSource(_)
88        | Plan::CreateSources(_)
89        | Plan::CreateSecret(_)
90        | Plan::CreateSink(_)
91        | Plan::CreateTable(_)
92        | Plan::CreateView(_)
93        | Plan::CreateMaterializedView(_)
94        | Plan::CreateIndex(_)
95        | Plan::CreateType(_)
96        | Plan::Comment(_)
97        | Plan::DiscardTemp
98        | Plan::DiscardAll
99        | Plan::DropObjects(_)
100        | Plan::DropOwned(_)
101        | Plan::EmptyQuery
102        | Plan::ShowAllVariables
103        | Plan::ShowCreate(_)
104        | Plan::ShowVariable(_)
105        | Plan::InspectShard(_)
106        | Plan::SetVariable(_)
107        | Plan::ResetVariable(_)
108        | Plan::SetTransaction(_)
109        | Plan::StartTransaction(_)
110        | Plan::CommitTransaction(_)
111        | Plan::AbortTransaction(_)
112        | Plan::CopyFrom(_)
113        | Plan::CopyTo(_)
114        | Plan::ExplainPlan(ExplainPlanPlan {
115            explainee:
116                Explainee::Statement(
117                    // Explicitly list all enum variants, to avoid bugs when somebody
118                    // adds a new variant (e.g., for `Plan::Execute`).
119                    ExplaineeStatement::CreateView { .. }
120                    | ExplaineeStatement::CreateMaterializedView { .. }
121                    | ExplaineeStatement::CreateIndex { .. },
122                ),
123            ..
124        })
125        | Plan::ExplainPlan(ExplainPlanPlan { explainee: _, .. })
126        | Plan::ExplainPushdown(_)
127        | Plan::ExplainSinkSchema(_)
128        | Plan::Insert(_)
129        | Plan::AlterNetworkPolicy(_)
130        | Plan::AlterNoop(_)
131        | Plan::AlterClusterRename(_)
132        | Plan::AlterClusterSwap(_)
133        | Plan::AlterClusterReplicaRename(_)
134        | Plan::AlterCluster(_)
135        | Plan::AlterConnection(_)
136        | Plan::AlterSource(_)
137        | Plan::AlterSetCluster(_)
138        | Plan::AlterItemRename(_)
139        | Plan::AlterRetainHistory(_)
140        | Plan::AlterSourceTimestampInterval(_)
141        | Plan::AlterSchemaRename(_)
142        | Plan::AlterSchemaSwap(_)
143        | Plan::AlterSecret(_)
144        | Plan::AlterSink(_)
145        | Plan::AlterSystemSet(_)
146        | Plan::AlterSystemReset(_)
147        | Plan::AlterSystemResetAll(_)
148        | Plan::AlterRole(_)
149        | Plan::AlterOwner(_)
150        | Plan::AlterTableAddColumn(_)
151        | Plan::AlterMaterializedViewApplyReplacement(_)
152        | Plan::Declare(_)
153        | Plan::Fetch(_)
154        | Plan::Close(_)
155        | Plan::ReadThenWrite(_)
156        | Plan::Prepare(_)
157        | Plan::Execute(_)
158        | Plan::Deallocate(_)
159        | Plan::Raise(_)
160        | Plan::GrantRole(_)
161        | Plan::RevokeRole(_)
162        | Plan::GrantPrivileges(_)
163        | Plan::RevokePrivileges(_)
164        | Plan::AlterDefaultPrivileges(_)
165        | Plan::ReassignOwned(_)
166        | Plan::ValidateConnection(_)
167        | Plan::SideEffectingFunc(_) => return TargetCluster::Active,
168    };
169
170    // Bail if the user has disabled it via the SessionVar.
171    if !session.vars().auto_route_catalog_queries() {
172        return TargetCluster::Active;
173    }
174
175    // We can't switch what cluster we're using, if the user has specified a replica.
176    if session.vars().cluster_replica().is_some() {
177        return TargetCluster::Active;
178    }
179
180    // These dependencies are just existing dataflows that are referenced in the plan.
181    let mut depends_on = depends_on
182        .into_iter()
183        .map(|gid| catalog.resolve_item_id(&gid))
184        .peekable();
185    let has_dependencies = depends_on.peek().is_some();
186
187    // Make sure we only depend on the system catalog, and nothing we depend on is a
188    // per-replica object, that requires being run a specific replica.
189    let valid_dependencies = depends_on.all(|id| {
190        let entry = catalog.state().get_entry(&id);
191        let schema = entry.name().qualifiers.schema_spec;
192
193        let system_only = catalog.state().is_system_schema_specifier(schema);
194        let non_replica = catalog.state().introspection_dependencies(id).is_empty();
195
196        system_only && non_replica
197    });
198
199    if (has_dependencies && valid_dependencies)
200        || (!has_dependencies && !could_run_expensive_function)
201    {
202        let intros_cluster = catalog
203            .state()
204            .resolve_builtin_cluster(&MZ_CATALOG_SERVER_CLUSTER);
205        tracing::debug!("Running on '{}' cluster", MZ_CATALOG_SERVER_CLUSTER.name);
206
207        // If we're running on a different cluster than the active one, notify the user.
208        if intros_cluster.name != session.vars().cluster() {
209            session.add_notice(AdapterNotice::AutoRunOnCatalogServerCluster);
210        }
211        TargetCluster::CatalogServer
212    } else {
213        TargetCluster::Active
214    }
215}
216
217/// Checks if we're currently running on the [`MZ_CATALOG_SERVER_CLUSTER`], and if so, do
218/// we depend on any objects that we're not allowed to query from the cluster.
219pub fn check_cluster_restrictions(
220    cluster: &str,
221    catalog: &impl SessionCatalog,
222    plan: &Plan,
223) -> Result<(), AdapterError> {
224    // We only impose restrictions if the current cluster is the catalog server cluster.
225    if cluster != MZ_CATALOG_SERVER_CLUSTER.name {
226        return Ok(());
227    }
228
229    // Only continue, and check restrictions, if a Plan would run some computation on the cluster.
230    //
231    // Note: We get the dependencies from the Plans themselves, because it's only after planning
232    // that we actually know what objects we'll need to reference.
233    //
234    // Note: Creating other objects like Materialized Views is prevented elsewhere. We define the
235    // 'mz_catalog_server' cluster to be "read-only", which restricts these actions.
236    let depends_on: Box<dyn Iterator<Item = GlobalId>> = match plan {
237        Plan::ReadThenWrite(plan) => Box::new(plan.selection.depends_on().into_iter()),
238        Plan::Subscribe(plan) => match plan.from {
239            SubscribeFrom::Id(id) => Box::new(std::iter::once(id)),
240            SubscribeFrom::Query { ref expr, .. } => Box::new(expr.depends_on().into_iter()),
241        },
242        Plan::Select(plan) => Box::new(plan.source.depends_on().into_iter()),
243        _ => return Ok(()),
244    };
245
246    // Collect any items that are not allowed to be run on the catalog server cluster.
247    let unallowed_dependents: SmallVec<[String; 2]> = depends_on
248        .filter_map(|id| {
249            let item = catalog.get_item_by_global_id(&id);
250            let full_name = catalog.resolve_full_name(item.name());
251
252            if !is_system_schema(&full_name.schema) {
253                Some(full_name.to_string())
254            } else {
255                None
256            }
257        })
258        .collect();
259
260    // If the query depends on unallowed items, error out.
261    if !unallowed_dependents.is_empty() {
262        Err(AdapterError::UnallowedOnCluster {
263            depends_on: unallowed_dependents,
264            cluster: MZ_CATALOG_SERVER_CLUSTER.name.to_string(),
265        })
266    } else {
267        Ok(())
268    }
269}
mz_adapter/coord/catalog_serving.rs

mz_adapter/coord/
catalog_serving.rs
