1use std::cell::RefCell;
15use std::collections::{BTreeMap, BTreeSet};
16
17use mz_repr::namespaces::is_system_schema;
18use mz_repr::{
19 CatalogItemId, ColumnIndex, RelationDesc, RelationVersionSelector, SqlColumnType, SqlScalarType,
20};
21use mz_sql_parser::ast::{
22 ColumnDef, ColumnName, ConnectionDefaultAwsPrivatelink, CreateMaterializedViewStatement,
23 RawItemName, ShowStatement, StatementKind, TableConstraint, UnresolvedDatabaseName,
24 UnresolvedSchemaName,
25};
26use mz_storage_types::connections::inline::ReferencedConnection;
27use mz_storage_types::connections::{AwsPrivatelink, Connection, SshTunnel, Tunnel};
28
29use crate::ast::{Ident, Statement, UnresolvedItemName};
30use crate::catalog::{
31 CatalogCluster, CatalogCollectionItem, CatalogDatabase, CatalogItem, CatalogItemType,
32 CatalogSchema, ObjectType, SessionCatalog, SystemObjectType,
33};
34use crate::names::{
35 self, Aug, DatabaseId, FullItemName, ItemQualifiers, ObjectId, PartialItemName,
36 QualifiedItemName, RawDatabaseSpecifier, ResolvedColumnReference, ResolvedDataType,
37 ResolvedDatabaseSpecifier, ResolvedIds, ResolvedItemName, ResolvedSchemaName, SchemaSpecifier,
38 SystemObjectId,
39};
40use crate::normalize;
41use crate::plan::error::PlanError;
42use crate::plan::{Params, Plan, PlanContext, PlanKind, query, with_options};
43use crate::session::vars::FeatureFlag;
44
45mod acl;
46pub(crate) mod ddl;
47mod dml;
48mod raise;
49mod scl;
50pub(crate) mod show;
51mod tcl;
52mod validate;
53
54use crate::session::vars;
55pub(crate) use ddl::PgConfigOptionExtracted;
56use mz_controller_types::ClusterId;
57use mz_pgrepr::oid::{FIRST_MATERIALIZE_OID, FIRST_USER_OID};
58use mz_repr::role_id::RoleId;
59
60#[derive(Debug, Clone, Eq, PartialEq)]
62pub struct StatementDesc {
63 pub relation_desc: Option<RelationDesc>,
66 pub param_types: Vec<SqlScalarType>,
68 pub is_copy: bool,
70}
71
72impl StatementDesc {
73 pub fn new(relation_desc: Option<RelationDesc>) -> Self {
74 StatementDesc {
75 relation_desc,
76 param_types: vec![],
77 is_copy: false,
78 }
79 }
80
81 pub fn arity(&self) -> usize {
84 self.relation_desc
85 .as_ref()
86 .map(|desc| desc.typ().column_types.len())
87 .unwrap_or(0)
88 }
89
90 fn with_params(mut self, param_types: Vec<SqlScalarType>) -> Self {
91 self.param_types = param_types;
92 self
93 }
94
95 fn with_is_copy(mut self) -> Self {
96 self.is_copy = true;
97 self
98 }
99}
100
101pub fn describe(
105 pcx: &PlanContext,
106 catalog: &dyn SessionCatalog,
107 stmt: Statement<Aug>,
108 param_types_in: &[Option<SqlScalarType>],
109) -> Result<StatementDesc, PlanError> {
110 let mut param_types = BTreeMap::new();
111 for (i, ty) in param_types_in.iter().enumerate() {
112 if let Some(ty) = ty {
113 param_types.insert(i + 1, ty.clone());
114 }
115 }
116
117 let scx = StatementContext {
118 pcx: Some(pcx),
119 catalog,
120 param_types: RefCell::new(param_types),
121 ambiguous_columns: RefCell::new(false),
122 };
123
124 let desc = match stmt {
125 Statement::AlterCluster(stmt) => ddl::describe_alter_cluster_set_options(&scx, stmt)?,
127 Statement::AlterConnection(stmt) => ddl::describe_alter_connection(&scx, stmt)?,
128 Statement::AlterIndex(stmt) => ddl::describe_alter_index_options(&scx, stmt)?,
129 Statement::AlterObjectRename(stmt) => ddl::describe_alter_object_rename(&scx, stmt)?,
130 Statement::AlterObjectSwap(stmt) => ddl::describe_alter_object_swap(&scx, stmt)?,
131 Statement::AlterRetainHistory(stmt) => ddl::describe_alter_retain_history(&scx, stmt)?,
132 Statement::AlterRole(stmt) => ddl::describe_alter_role(&scx, stmt)?,
133 Statement::AlterSecret(stmt) => ddl::describe_alter_secret_options(&scx, stmt)?,
134 Statement::AlterSetCluster(stmt) => ddl::describe_alter_set_cluster(&scx, stmt)?,
135 Statement::AlterSink(stmt) => ddl::describe_alter_sink(&scx, stmt)?,
136 Statement::AlterSource(stmt) => ddl::describe_alter_source(&scx, stmt)?,
137 Statement::AlterSystemSet(stmt) => ddl::describe_alter_system_set(&scx, stmt)?,
138 Statement::AlterSystemReset(stmt) => ddl::describe_alter_system_reset(&scx, stmt)?,
139 Statement::AlterSystemResetAll(stmt) => ddl::describe_alter_system_reset_all(&scx, stmt)?,
140 Statement::AlterTableAddColumn(stmt) => ddl::describe_alter_table_add_column(&scx, stmt)?,
141 Statement::AlterNetworkPolicy(stmt) => ddl::describe_alter_network_policy(&scx, stmt)?,
142 Statement::Comment(stmt) => ddl::describe_comment(&scx, stmt)?,
143 Statement::CreateCluster(stmt) => ddl::describe_create_cluster(&scx, stmt)?,
144 Statement::CreateClusterReplica(stmt) => ddl::describe_create_cluster_replica(&scx, stmt)?,
145 Statement::CreateConnection(stmt) => ddl::describe_create_connection(&scx, stmt)?,
146 Statement::CreateDatabase(stmt) => ddl::describe_create_database(&scx, stmt)?,
147 Statement::CreateIndex(stmt) => ddl::describe_create_index(&scx, stmt)?,
148 Statement::CreateRole(stmt) => ddl::describe_create_role(&scx, stmt)?,
149 Statement::CreateSchema(stmt) => ddl::describe_create_schema(&scx, stmt)?,
150 Statement::CreateSecret(stmt) => ddl::describe_create_secret(&scx, stmt)?,
151 Statement::CreateSink(stmt) => ddl::describe_create_sink(&scx, stmt)?,
152 Statement::CreateWebhookSource(stmt) => ddl::describe_create_webhook_source(&scx, stmt)?,
153 Statement::CreateSource(stmt) => ddl::describe_create_source(&scx, stmt)?,
154 Statement::CreateSubsource(stmt) => ddl::describe_create_subsource(&scx, stmt)?,
155 Statement::CreateTable(stmt) => ddl::describe_create_table(&scx, stmt)?,
156 Statement::CreateTableFromSource(stmt) => {
157 ddl::describe_create_table_from_source(&scx, stmt)?
158 }
159 Statement::CreateType(stmt) => ddl::describe_create_type(&scx, stmt)?,
160 Statement::CreateView(stmt) => ddl::describe_create_view(&scx, stmt)?,
161 Statement::CreateMaterializedView(stmt) => {
162 ddl::describe_create_materialized_view(&scx, stmt)?
163 }
164 Statement::CreateContinualTask(stmt) => ddl::describe_create_continual_task(&scx, stmt)?,
165 Statement::CreateNetworkPolicy(stmt) => ddl::describe_create_network_policy(&scx, stmt)?,
166 Statement::DropObjects(stmt) => ddl::describe_drop_objects(&scx, stmt)?,
167 Statement::DropOwned(stmt) => ddl::describe_drop_owned(&scx, stmt)?,
168
169 Statement::AlterOwner(stmt) => acl::describe_alter_owner(&scx, stmt)?,
171 Statement::GrantRole(stmt) => acl::describe_grant_role(&scx, stmt)?,
172 Statement::RevokeRole(stmt) => acl::describe_revoke_role(&scx, stmt)?,
173 Statement::GrantPrivileges(stmt) => acl::describe_grant_privileges(&scx, stmt)?,
174 Statement::RevokePrivileges(stmt) => acl::describe_revoke_privileges(&scx, stmt)?,
175 Statement::AlterDefaultPrivileges(stmt) => {
176 acl::describe_alter_default_privileges(&scx, stmt)?
177 }
178 Statement::ReassignOwned(stmt) => acl::describe_reassign_owned(&scx, stmt)?,
179
180 Statement::Show(ShowStatement::ShowColumns(stmt)) => {
182 show::show_columns(&scx, stmt)?.describe()?
183 }
184 Statement::Show(ShowStatement::ShowCreateConnection(stmt)) => {
185 show::describe_show_create_connection(&scx, stmt)?
186 }
187 Statement::Show(ShowStatement::ShowCreateCluster(stmt)) => {
188 show::describe_show_create_cluster(&scx, stmt)?
189 }
190 Statement::Show(ShowStatement::ShowCreateIndex(stmt)) => {
191 show::describe_show_create_index(&scx, stmt)?
192 }
193 Statement::Show(ShowStatement::ShowCreateSink(stmt)) => {
194 show::describe_show_create_sink(&scx, stmt)?
195 }
196 Statement::Show(ShowStatement::ShowCreateSource(stmt)) => {
197 show::describe_show_create_source(&scx, stmt)?
198 }
199 Statement::Show(ShowStatement::ShowCreateTable(stmt)) => {
200 show::describe_show_create_table(&scx, stmt)?
201 }
202 Statement::Show(ShowStatement::ShowCreateView(stmt)) => {
203 show::describe_show_create_view(&scx, stmt)?
204 }
205 Statement::Show(ShowStatement::ShowCreateMaterializedView(stmt)) => {
206 show::describe_show_create_materialized_view(&scx, stmt)?
207 }
208 Statement::Show(ShowStatement::ShowCreateType(stmt)) => {
209 show::describe_show_create_type(&scx, stmt)?
210 }
211 Statement::Show(ShowStatement::ShowObjects(stmt)) => {
212 show::show_objects(&scx, stmt)?.describe()?
213 }
214
215 Statement::Close(stmt) => scl::describe_close(&scx, stmt)?,
217 Statement::Deallocate(stmt) => scl::describe_deallocate(&scx, stmt)?,
218 Statement::Declare(stmt) => scl::describe_declare(&scx, stmt, param_types_in)?,
219 Statement::Discard(stmt) => scl::describe_discard(&scx, stmt)?,
220 Statement::Execute(stmt) => scl::describe_execute(&scx, stmt)?,
221 Statement::Fetch(stmt) => scl::describe_fetch(&scx, stmt)?,
222 Statement::Prepare(stmt) => scl::describe_prepare(&scx, stmt)?,
223 Statement::ResetVariable(stmt) => scl::describe_reset_variable(&scx, stmt)?,
224 Statement::SetVariable(stmt) => scl::describe_set_variable(&scx, stmt)?,
225 Statement::Show(ShowStatement::ShowVariable(stmt)) => {
226 scl::describe_show_variable(&scx, stmt)?
227 }
228
229 Statement::Copy(stmt) => dml::describe_copy(&scx, stmt)?,
231 Statement::Delete(stmt) => dml::describe_delete(&scx, stmt)?,
232 Statement::ExplainPlan(stmt) => dml::describe_explain_plan(&scx, stmt)?,
233 Statement::ExplainPushdown(stmt) => dml::describe_explain_pushdown(&scx, stmt)?,
234 Statement::ExplainAnalyzeObject(stmt) => dml::describe_explain_analyze_object(&scx, stmt)?,
235 Statement::ExplainAnalyzeCluster(stmt) => {
236 dml::describe_explain_analyze_cluster(&scx, stmt)?
237 }
238 Statement::ExplainTimestamp(stmt) => dml::describe_explain_timestamp(&scx, stmt)?,
239 Statement::ExplainSinkSchema(stmt) => dml::describe_explain_schema(&scx, stmt)?,
240 Statement::Insert(stmt) => dml::describe_insert(&scx, stmt)?,
241 Statement::Select(stmt) => dml::describe_select(&scx, stmt)?,
242 Statement::Subscribe(stmt) => dml::describe_subscribe(&scx, stmt)?,
243 Statement::Update(stmt) => dml::describe_update(&scx, stmt)?,
244
245 Statement::Commit(stmt) => tcl::describe_commit(&scx, stmt)?,
247 Statement::Rollback(stmt) => tcl::describe_rollback(&scx, stmt)?,
248 Statement::SetTransaction(stmt) => tcl::describe_set_transaction(&scx, stmt)?,
249 Statement::StartTransaction(stmt) => tcl::describe_start_transaction(&scx, stmt)?,
250
251 Statement::Raise(stmt) => raise::describe_raise(&scx, stmt)?,
253 Statement::Show(ShowStatement::InspectShard(stmt)) => {
254 scl::describe_inspect_shard(&scx, stmt)?
255 }
256 Statement::ValidateConnection(stmt) => validate::describe_validate_connection(&scx, stmt)?,
257 };
258
259 let desc = desc.with_params(scx.finalize_param_types()?);
260 Ok(desc)
261}
262
263#[mz_ore::instrument(level = "debug")]
280pub fn plan(
281 pcx: Option<&PlanContext>,
282 catalog: &dyn SessionCatalog,
283 stmt: Statement<Aug>,
284 params: &Params,
285 resolved_ids: &ResolvedIds,
286) -> Result<Plan, PlanError> {
287 let param_types = params
288 .expected_types
292 .iter()
293 .enumerate()
294 .map(|(i, ty)| (i + 1, ty.clone()))
295 .collect();
296
297 let kind: StatementKind = (&stmt).into();
298 let permitted_plans = Plan::generated_from(&kind);
299
300 let scx = &mut StatementContext {
301 pcx,
302 catalog,
303 param_types: RefCell::new(param_types),
304 ambiguous_columns: RefCell::new(false),
305 };
306
307 if resolved_ids
308 .items()
309 .filter_map(|id| catalog.try_get_item(id))
311 .any(|item| {
312 item.func().is_ok()
313 && item.name().qualifiers.schema_spec
314 == SchemaSpecifier::Id(catalog.get_mz_unsafe_schema_id())
315 })
316 {
317 scx.require_feature_flag(&vars::UNSAFE_ENABLE_UNSAFE_FUNCTIONS)?;
318 }
319
320 let plan = match stmt {
321 Statement::AlterCluster(stmt) => ddl::plan_alter_cluster(scx, stmt),
323 Statement::AlterConnection(stmt) => ddl::plan_alter_connection(scx, stmt),
324 Statement::AlterIndex(stmt) => ddl::plan_alter_index_options(scx, stmt),
325 Statement::AlterObjectRename(stmt) => ddl::plan_alter_object_rename(scx, stmt),
326 Statement::AlterObjectSwap(stmt) => ddl::plan_alter_object_swap(scx, stmt),
327 Statement::AlterRetainHistory(stmt) => ddl::plan_alter_retain_history(scx, stmt),
328 Statement::AlterRole(stmt) => ddl::plan_alter_role(scx, stmt),
329 Statement::AlterSecret(stmt) => ddl::plan_alter_secret(scx, stmt),
330 Statement::AlterSetCluster(stmt) => ddl::plan_alter_item_set_cluster(scx, stmt),
331 Statement::AlterSink(stmt) => ddl::plan_alter_sink(scx, stmt),
332 Statement::AlterSource(stmt) => ddl::plan_alter_source(scx, stmt),
333 Statement::AlterSystemSet(stmt) => ddl::plan_alter_system_set(scx, stmt),
334 Statement::AlterSystemReset(stmt) => ddl::plan_alter_system_reset(scx, stmt),
335 Statement::AlterSystemResetAll(stmt) => ddl::plan_alter_system_reset_all(scx, stmt),
336 Statement::AlterTableAddColumn(stmt) => ddl::plan_alter_table_add_column(scx, stmt),
337 Statement::AlterNetworkPolicy(stmt) => ddl::plan_alter_network_policy(scx, stmt),
338 Statement::Comment(stmt) => ddl::plan_comment(scx, stmt),
339 Statement::CreateCluster(stmt) => ddl::plan_create_cluster(scx, stmt),
340 Statement::CreateClusterReplica(stmt) => ddl::plan_create_cluster_replica(scx, stmt),
341 Statement::CreateConnection(stmt) => ddl::plan_create_connection(scx, stmt),
342 Statement::CreateDatabase(stmt) => ddl::plan_create_database(scx, stmt),
343 Statement::CreateIndex(stmt) => ddl::plan_create_index(scx, stmt),
344 Statement::CreateRole(stmt) => ddl::plan_create_role(scx, stmt),
345 Statement::CreateSchema(stmt) => ddl::plan_create_schema(scx, stmt),
346 Statement::CreateSecret(stmt) => ddl::plan_create_secret(scx, stmt),
347 Statement::CreateSink(stmt) => ddl::plan_create_sink(scx, stmt),
348 Statement::CreateWebhookSource(stmt) => ddl::plan_create_webhook_source(scx, stmt),
349 Statement::CreateSource(stmt) => ddl::plan_create_source(scx, stmt),
350 Statement::CreateSubsource(stmt) => ddl::plan_create_subsource(scx, stmt),
351 Statement::CreateTable(stmt) => ddl::plan_create_table(scx, stmt),
352 Statement::CreateTableFromSource(stmt) => ddl::plan_create_table_from_source(scx, stmt),
353 Statement::CreateType(stmt) => ddl::plan_create_type(scx, stmt),
354 Statement::CreateView(stmt) => ddl::plan_create_view(scx, stmt),
355 Statement::CreateMaterializedView(stmt) => ddl::plan_create_materialized_view(scx, stmt),
356 Statement::CreateContinualTask(stmt) => ddl::plan_create_continual_task(scx, stmt),
357 Statement::CreateNetworkPolicy(stmt) => ddl::plan_create_network_policy(scx, stmt),
358 Statement::DropObjects(stmt) => ddl::plan_drop_objects(scx, stmt),
359 Statement::DropOwned(stmt) => ddl::plan_drop_owned(scx, stmt),
360
361 Statement::AlterOwner(stmt) => acl::plan_alter_owner(scx, stmt),
363 Statement::GrantRole(stmt) => acl::plan_grant_role(scx, stmt),
364 Statement::RevokeRole(stmt) => acl::plan_revoke_role(scx, stmt),
365 Statement::GrantPrivileges(stmt) => acl::plan_grant_privileges(scx, stmt),
366 Statement::RevokePrivileges(stmt) => acl::plan_revoke_privileges(scx, stmt),
367 Statement::AlterDefaultPrivileges(stmt) => acl::plan_alter_default_privileges(scx, stmt),
368 Statement::ReassignOwned(stmt) => acl::plan_reassign_owned(scx, stmt),
369
370 Statement::Copy(stmt) => dml::plan_copy(scx, stmt),
372 Statement::Delete(stmt) => dml::plan_delete(scx, stmt, params),
373 Statement::ExplainPlan(stmt) => dml::plan_explain_plan(scx, stmt, params),
374 Statement::ExplainPushdown(stmt) => dml::plan_explain_pushdown(scx, stmt, params),
375 Statement::ExplainAnalyzeObject(stmt) => {
376 dml::plan_explain_analyze_object(scx, stmt, params)
377 }
378 Statement::ExplainAnalyzeCluster(stmt) => {
379 dml::plan_explain_analyze_cluster(scx, stmt, params)
380 }
381 Statement::ExplainTimestamp(stmt) => dml::plan_explain_timestamp(scx, stmt),
382 Statement::ExplainSinkSchema(stmt) => dml::plan_explain_schema(scx, stmt),
383 Statement::Insert(stmt) => dml::plan_insert(scx, stmt, params),
384 Statement::Select(stmt) => dml::plan_select(scx, stmt, params, None),
385 Statement::Subscribe(stmt) => dml::plan_subscribe(scx, stmt, params, None),
386 Statement::Update(stmt) => dml::plan_update(scx, stmt, params),
387
388 Statement::Show(ShowStatement::ShowColumns(stmt)) => show::show_columns(scx, stmt)?.plan(),
390 Statement::Show(ShowStatement::ShowCreateConnection(stmt)) => {
391 show::plan_show_create_connection(scx, stmt).map(Plan::ShowCreate)
392 }
393 Statement::Show(ShowStatement::ShowCreateCluster(stmt)) => {
394 show::plan_show_create_cluster(scx, stmt).map(Plan::ShowCreate)
395 }
396 Statement::Show(ShowStatement::ShowCreateIndex(stmt)) => {
397 show::plan_show_create_index(scx, stmt).map(Plan::ShowCreate)
398 }
399 Statement::Show(ShowStatement::ShowCreateSink(stmt)) => {
400 show::plan_show_create_sink(scx, stmt).map(Plan::ShowCreate)
401 }
402 Statement::Show(ShowStatement::ShowCreateSource(stmt)) => {
403 show::plan_show_create_source(scx, stmt).map(Plan::ShowCreate)
404 }
405 Statement::Show(ShowStatement::ShowCreateTable(stmt)) => {
406 show::plan_show_create_table(scx, stmt).map(Plan::ShowCreate)
407 }
408 Statement::Show(ShowStatement::ShowCreateView(stmt)) => {
409 show::plan_show_create_view(scx, stmt).map(Plan::ShowCreate)
410 }
411 Statement::Show(ShowStatement::ShowCreateMaterializedView(stmt)) => {
412 show::plan_show_create_materialized_view(scx, stmt).map(Plan::ShowCreate)
413 }
414 Statement::Show(ShowStatement::ShowCreateType(stmt)) => {
415 show::plan_show_create_type(scx, stmt).map(Plan::ShowCreate)
416 }
417 Statement::Show(ShowStatement::ShowObjects(stmt)) => show::show_objects(scx, stmt)?.plan(),
418
419 Statement::Close(stmt) => scl::plan_close(scx, stmt),
421 Statement::Deallocate(stmt) => scl::plan_deallocate(scx, stmt),
422 Statement::Declare(stmt) => scl::plan_declare(scx, stmt, params),
423 Statement::Discard(stmt) => scl::plan_discard(scx, stmt),
424 Statement::Execute(stmt) => scl::plan_execute(scx, stmt),
425 Statement::Fetch(stmt) => scl::plan_fetch(scx, stmt),
426 Statement::Prepare(stmt) => scl::plan_prepare(scx, stmt),
427 Statement::ResetVariable(stmt) => scl::plan_reset_variable(scx, stmt),
428 Statement::SetVariable(stmt) => scl::plan_set_variable(scx, stmt),
429 Statement::Show(ShowStatement::ShowVariable(stmt)) => scl::plan_show_variable(scx, stmt),
430
431 Statement::Commit(stmt) => tcl::plan_commit(scx, stmt),
433 Statement::Rollback(stmt) => tcl::plan_rollback(scx, stmt),
434 Statement::SetTransaction(stmt) => tcl::plan_set_transaction(scx, stmt),
435 Statement::StartTransaction(stmt) => tcl::plan_start_transaction(scx, stmt),
436
437 Statement::Raise(stmt) => raise::plan_raise(scx, stmt),
439 Statement::Show(ShowStatement::InspectShard(stmt)) => scl::plan_inspect_shard(scx, stmt),
440 Statement::ValidateConnection(stmt) => validate::plan_validate_connection(scx, stmt),
441 };
442
443 if let Ok(plan) = &plan {
444 mz_ore::soft_assert_no_log!(
445 permitted_plans.contains(&PlanKind::from(plan)),
446 "plan {:?}, permitted plans {:?}",
447 plan,
448 permitted_plans
449 );
450 }
451
452 plan
453}
454
455pub fn plan_copy_from(
456 pcx: &PlanContext,
457 catalog: &dyn SessionCatalog,
458 target_id: CatalogItemId,
459 target_name: String,
460 columns: Vec<ColumnIndex>,
461 rows: Vec<mz_repr::Row>,
462) -> Result<super::HirRelationExpr, PlanError> {
463 query::plan_copy_from_rows(pcx, catalog, target_id, target_name, columns, rows)
464}
465
466impl PartialEq<ObjectType> for CatalogItemType {
473 fn eq(&self, other: &ObjectType) -> bool {
474 match (self, other) {
475 (CatalogItemType::Source, ObjectType::Source)
476 | (CatalogItemType::Table, ObjectType::Table)
477 | (CatalogItemType::Sink, ObjectType::Sink)
478 | (CatalogItemType::View, ObjectType::View)
479 | (CatalogItemType::MaterializedView, ObjectType::MaterializedView)
480 | (CatalogItemType::Index, ObjectType::Index)
481 | (CatalogItemType::Type, ObjectType::Type)
482 | (CatalogItemType::Secret, ObjectType::Secret)
483 | (CatalogItemType::Connection, ObjectType::Connection) => true,
484 (_, _) => false,
485 }
486 }
487}
488
489impl PartialEq<CatalogItemType> for ObjectType {
490 fn eq(&self, other: &CatalogItemType) -> bool {
491 other == self
492 }
493}
494
495#[derive(Debug, Clone)]
497pub struct StatementContext<'a> {
498 pcx: Option<&'a PlanContext>,
503 pub catalog: &'a dyn SessionCatalog,
504 pub param_types: RefCell<BTreeMap<usize, SqlScalarType>>,
507 pub ambiguous_columns: RefCell<bool>,
510}
511
512impl<'a> StatementContext<'a> {
513 pub fn new(
514 pcx: Option<&'a PlanContext>,
515 catalog: &'a dyn SessionCatalog,
516 ) -> StatementContext<'a> {
517 StatementContext {
518 pcx,
519 catalog,
520 param_types: Default::default(),
521 ambiguous_columns: RefCell::new(false),
522 }
523 }
524
525 pub fn current_schemas(&self) -> &[(ResolvedDatabaseSpecifier, SchemaSpecifier)] {
527 self.catalog.search_path()
528 }
529
530 pub fn current_schema(&self) -> Option<&(ResolvedDatabaseSpecifier, SchemaSpecifier)> {
533 self.current_schemas().into_iter().next()
534 }
535
536 pub fn pcx(&self) -> Result<&PlanContext, PlanError> {
537 self.pcx.ok_or_else(|| sql_err!("no plan context"))
538 }
539
540 pub fn allocate_full_name(&self, name: PartialItemName) -> Result<FullItemName, PlanError> {
541 let (database, schema): (RawDatabaseSpecifier, String) = match (name.database, name.schema)
542 {
543 (None, None) => {
544 let Some((database, schema)) = self.current_schema() else {
545 return Err(PlanError::InvalidSchemaName);
546 };
547 let schema = self.get_schema(database, schema);
548 let database = match schema.database() {
549 ResolvedDatabaseSpecifier::Ambient => RawDatabaseSpecifier::Ambient,
550 ResolvedDatabaseSpecifier::Id(id) => {
551 RawDatabaseSpecifier::Name(self.catalog.get_database(id).name().to_string())
552 }
553 };
554 (database, schema.name().schema.clone())
555 }
556 (None, Some(schema)) => {
557 if is_system_schema(&schema) {
558 (RawDatabaseSpecifier::Ambient, schema)
559 } else {
560 match self.catalog.active_database_name() {
561 Some(name) => (RawDatabaseSpecifier::Name(name.to_string()), schema),
562 None => {
563 sql_bail!(
564 "no database specified for non-system schema and no active database"
565 )
566 }
567 }
568 }
569 }
570 (Some(_database), None) => {
571 sql_bail!("unreachable: specified the database but no schema")
574 }
575 (Some(database), Some(schema)) => (RawDatabaseSpecifier::Name(database), schema),
576 };
577 let item = name.item;
578 Ok(FullItemName {
579 database,
580 schema,
581 item,
582 })
583 }
584
585 pub fn allocate_qualified_name(
586 &self,
587 name: PartialItemName,
588 ) -> Result<QualifiedItemName, PlanError> {
589 let full_name = self.allocate_full_name(name)?;
590 let database_spec = match full_name.database {
591 RawDatabaseSpecifier::Ambient => ResolvedDatabaseSpecifier::Ambient,
592 RawDatabaseSpecifier::Name(name) => ResolvedDatabaseSpecifier::Id(
593 self.resolve_database(&UnresolvedDatabaseName(Ident::new(name)?))?
594 .id(),
595 ),
596 };
597 let schema_spec = self
598 .resolve_schema_in_database(&database_spec, &Ident::new(full_name.schema)?)?
599 .id()
600 .clone();
601 Ok(QualifiedItemName {
602 qualifiers: ItemQualifiers {
603 database_spec,
604 schema_spec,
605 },
606 item: full_name.item,
607 })
608 }
609
610 pub fn allocate_temporary_full_name(&self, name: PartialItemName) -> FullItemName {
611 FullItemName {
612 database: RawDatabaseSpecifier::Ambient,
613 schema: name
614 .schema
615 .unwrap_or_else(|| mz_repr::namespaces::MZ_TEMP_SCHEMA.to_owned()),
616 item: name.item,
617 }
618 }
619
620 pub fn allocate_temporary_qualified_name(
621 &self,
622 name: PartialItemName,
623 ) -> Result<QualifiedItemName, PlanError> {
624 if let Some(name) = name.schema {
625 if name
626 != self
627 .get_schema(
628 &ResolvedDatabaseSpecifier::Ambient,
629 &SchemaSpecifier::Temporary,
630 )
631 .name()
632 .schema
633 {
634 return Err(PlanError::InvalidTemporarySchema);
635 }
636 }
637
638 Ok(QualifiedItemName {
639 qualifiers: ItemQualifiers {
640 database_spec: ResolvedDatabaseSpecifier::Ambient,
641 schema_spec: SchemaSpecifier::Temporary,
642 },
643 item: name.item,
644 })
645 }
646
647 pub fn allocate_resolved_item_name(
650 &self,
651 id: CatalogItemId,
652 name: UnresolvedItemName,
653 ) -> Result<ResolvedItemName, PlanError> {
654 let partial = normalize::unresolved_item_name(name)?;
655 let qualified = self.allocate_qualified_name(partial.clone())?;
656 let full_name = self.allocate_full_name(partial)?;
657 Ok(ResolvedItemName::Item {
658 id,
659 qualifiers: qualified.qualifiers,
660 full_name,
661 print_id: true,
662 version: RelationVersionSelector::Latest,
663 })
664 }
665
666 pub fn active_database(&self) -> Option<&DatabaseId> {
667 self.catalog.active_database()
668 }
669
670 pub fn resolve_optional_schema(
671 &self,
672 schema_name: &Option<ResolvedSchemaName>,
673 ) -> Result<SchemaSpecifier, PlanError> {
674 match schema_name {
675 Some(ResolvedSchemaName::Schema { schema_spec, .. }) => Ok(schema_spec.clone()),
676 None => self.resolve_active_schema().map(|spec| spec.clone()),
677 Some(ResolvedSchemaName::Error) => {
678 unreachable!("should have been handled by name resolution")
679 }
680 }
681 }
682
683 pub fn resolve_active_schema(&self) -> Result<&SchemaSpecifier, PlanError> {
684 match self.current_schema() {
685 Some((_db, schema)) => Ok(schema),
686 None => Err(PlanError::InvalidSchemaName),
687 }
688 }
689
690 pub fn get_cluster(&self, id: &ClusterId) -> &dyn CatalogCluster<'_> {
691 self.catalog.get_cluster(*id)
692 }
693
694 pub fn resolve_database(
695 &self,
696 name: &UnresolvedDatabaseName,
697 ) -> Result<&dyn CatalogDatabase, PlanError> {
698 let name = normalize::ident_ref(&name.0);
699 Ok(self.catalog.resolve_database(name)?)
700 }
701
702 pub fn get_database(&self, id: &DatabaseId) -> &dyn CatalogDatabase {
703 self.catalog.get_database(id)
704 }
705
706 pub fn resolve_schema_in_database(
707 &self,
708 database_spec: &ResolvedDatabaseSpecifier,
709 schema: &Ident,
710 ) -> Result<&dyn CatalogSchema, PlanError> {
711 let schema = normalize::ident_ref(schema);
712 Ok(self
713 .catalog
714 .resolve_schema_in_database(database_spec, schema)?)
715 }
716
717 pub fn resolve_schema(
718 &self,
719 name: UnresolvedSchemaName,
720 ) -> Result<&dyn CatalogSchema, PlanError> {
721 let name = normalize::unresolved_schema_name(name)?;
722 Ok(self
723 .catalog
724 .resolve_schema(name.database.as_deref(), &name.schema)?)
725 }
726
727 pub fn get_schema(
728 &self,
729 database_spec: &ResolvedDatabaseSpecifier,
730 schema_spec: &SchemaSpecifier,
731 ) -> &dyn CatalogSchema {
732 self.catalog.get_schema(database_spec, schema_spec)
733 }
734
735 pub fn resolve_item(&self, name: RawItemName) -> Result<&dyn CatalogItem, PlanError> {
736 match name {
737 RawItemName::Name(name) => {
738 let name = normalize::unresolved_item_name(name)?;
739 Ok(self.catalog.resolve_item(&name)?)
740 }
741 RawItemName::Id(id, _, _) => {
742 let gid = id.parse()?;
743 Ok(self.catalog.get_item(&gid))
744 }
745 }
746 }
747
748 pub fn get_item(&self, id: &CatalogItemId) -> &dyn CatalogItem {
749 self.catalog.get_item(id)
750 }
751
752 pub fn get_item_by_resolved_name(
753 &self,
754 name: &ResolvedItemName,
755 ) -> Result<Box<dyn CatalogCollectionItem + '_>, PlanError> {
756 match name {
757 ResolvedItemName::Item { id, version, .. } => {
758 Ok(self.get_item(id).at_version(*version))
759 }
760 ResolvedItemName::Cte { .. } => sql_bail!("non-user item"),
761 ResolvedItemName::ContinualTask { .. } => sql_bail!("non-user item"),
762 ResolvedItemName::Error => unreachable!("should have been caught in name resolution"),
763 }
764 }
765
766 pub fn get_column_by_resolved_name(
767 &self,
768 name: &ColumnName<Aug>,
769 ) -> Result<(Box<dyn CatalogCollectionItem + '_>, usize), PlanError> {
770 match (&name.relation, &name.column) {
771 (
772 ResolvedItemName::Item { id, version, .. },
773 ResolvedColumnReference::Column { index, .. },
774 ) => {
775 let item = self.get_item(id).at_version(*version);
776 Ok((item, *index))
777 }
778 _ => unreachable!(
779 "get_column_by_resolved_name errors should have been caught in name resolution"
780 ),
781 }
782 }
783
784 pub fn resolve_function(
785 &self,
786 name: UnresolvedItemName,
787 ) -> Result<&dyn CatalogItem, PlanError> {
788 let name = normalize::unresolved_item_name(name)?;
789 Ok(self.catalog.resolve_function(&name)?)
790 }
791
792 pub fn resolve_cluster(
793 &self,
794 name: Option<&Ident>,
795 ) -> Result<&dyn CatalogCluster<'_>, PlanError> {
796 let name = name.map(|name| name.as_str());
797 Ok(self.catalog.resolve_cluster(name)?)
798 }
799
800 pub fn resolve_type(&self, mut ty: mz_pgrepr::Type) -> Result<ResolvedDataType, PlanError> {
801 match &mut ty {
807 mz_pgrepr::Type::Interval { constraints } => *constraints = None,
808 mz_pgrepr::Type::Time { precision } => *precision = None,
809 mz_pgrepr::Type::TimeTz { precision } => *precision = None,
810 mz_pgrepr::Type::Timestamp { precision } => *precision = None,
811 mz_pgrepr::Type::TimestampTz { precision } => *precision = None,
812 _ => (),
813 }
814 let mut ty = if ty.oid() >= FIRST_USER_OID {
820 sql_bail!("internal error, unexpected user type: {ty:?} ");
821 } else if ty.oid() < FIRST_MATERIALIZE_OID {
822 format!("pg_catalog.{}", ty)
823 } else {
824 format!("mz_catalog.{}", ty)
826 };
827 if ty == "pg_catalog.json" {
831 ty = "pg_catalog.jsonb".into();
832 }
833 let data_type = mz_sql_parser::parser::parse_data_type(&ty)?;
834 let (data_type, _) = names::resolve(self.catalog, data_type)?;
835 Ok(data_type)
836 }
837
838 pub fn get_object_type(&self, id: &ObjectId) -> ObjectType {
839 self.catalog.get_object_type(id)
840 }
841
842 pub fn get_system_object_type(&self, id: &SystemObjectId) -> SystemObjectType {
843 match id {
844 SystemObjectId::Object(id) => SystemObjectType::Object(self.get_object_type(id)),
845 SystemObjectId::System => SystemObjectType::System,
846 }
847 }
848
849 pub fn require_feature_flag(&self, flag: &'static FeatureFlag) -> Result<(), PlanError> {
851 flag.require(self.catalog.system_vars())?;
852 Ok(())
853 }
854
855 pub fn is_feature_flag_enabled(&self, flag: &'static FeatureFlag) -> bool {
857 self.require_feature_flag(flag).is_ok()
858 }
859
860 pub fn finalize_param_types(self) -> Result<Vec<SqlScalarType>, PlanError> {
861 let param_types = self.param_types.into_inner();
862 let mut out = vec![];
863 for (i, (n, typ)) in param_types.into_iter().enumerate() {
864 if n != i + 1 {
865 sql_bail!("unable to infer type for parameter ${}", i + 1);
866 }
867 out.push(typ);
868 }
869 Ok(out)
870 }
871
872 pub fn humanize_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
875 self.catalog.humanize_scalar_type(typ, postgres_compat)
876 }
877
878 pub fn humanize_column_type(&self, typ: &SqlColumnType, postgres_compat: bool) -> String {
881 self.catalog.humanize_column_type(typ, postgres_compat)
882 }
883
884 pub(crate) fn build_tunnel_definition(
885 &self,
886 ssh_tunnel: Option<with_options::Object>,
887 aws_privatelink: Option<ConnectionDefaultAwsPrivatelink<Aug>>,
888 ) -> Result<Tunnel<ReferencedConnection>, PlanError> {
889 match (ssh_tunnel, aws_privatelink) {
890 (None, None) => Ok(Tunnel::Direct),
891 (Some(ssh_tunnel), None) => {
892 let id = CatalogItemId::from(ssh_tunnel);
893 let ssh_tunnel = self.catalog.get_item(&id);
894 match ssh_tunnel.connection()? {
895 Connection::Ssh(_connection) => Ok(Tunnel::Ssh(SshTunnel {
896 connection_id: id,
897 connection: id,
898 })),
899 _ => sql_bail!("{} is not an SSH connection", ssh_tunnel.name().item),
900 }
901 }
902 (None, Some(aws_privatelink)) => {
903 let id = aws_privatelink.connection.item_id().clone();
904 let entry = self.catalog.get_item(&id);
905 match entry.connection()? {
906 Connection::AwsPrivatelink(_) => Ok(Tunnel::AwsPrivatelink(AwsPrivatelink {
907 connection_id: id,
908 availability_zone: None,
910 port: aws_privatelink.port,
912 })),
913 _ => sql_bail!("{} is not an AWS PRIVATELINK connection", entry.name().item),
914 }
915 }
916 (Some(_), Some(_)) => {
917 sql_bail!("cannot specify both SSH TUNNEL and AWS PRIVATELINK");
918 }
919 }
920 }
921
922 pub fn relation_desc_into_table_defs(
923 &self,
924 desc: &RelationDesc,
925 ) -> Result<(Vec<ColumnDef<Aug>>, Vec<TableConstraint<Aug>>), PlanError> {
926 let mut columns = vec![];
927 let mut null_cols = BTreeSet::new();
928 for (column_name, column_type) in desc.iter() {
929 let name = Ident::new(column_name.as_str().to_owned())?;
930
931 let ty = mz_pgrepr::Type::from(&column_type.scalar_type);
932 let data_type = self.resolve_type(ty)?;
933
934 let options = if !column_type.nullable {
935 null_cols.insert(columns.len());
936 vec![mz_sql_parser::ast::ColumnOptionDef {
937 name: None,
938 option: mz_sql_parser::ast::ColumnOption::NotNull,
939 }]
940 } else {
941 vec![]
942 };
943
944 columns.push(ColumnDef {
945 name,
946 data_type,
947 collation: None,
948 options,
949 });
950 }
951
952 let mut table_constraints = vec![];
953 for key in desc.typ().keys.iter() {
954 let mut col_names = vec![];
955 for col_idx in key {
956 if !null_cols.contains(col_idx) {
957 sql_bail!(
960 "[internal error] key columns must be NOT NULL when generating table constraints"
961 );
962 }
963 col_names.push(columns[*col_idx].name.clone());
964 }
965 table_constraints.push(TableConstraint::Unique {
966 name: None,
967 columns: col_names,
968 is_primary: false,
969 nulls_not_distinct: false,
970 });
971 }
972
973 Ok((columns, table_constraints))
974 }
975
976 pub fn get_owner_id(&self, id: &ObjectId) -> Option<RoleId> {
977 self.catalog.get_owner_id(id)
978 }
979
980 pub fn humanize_resolved_name(
981 &self,
982 name: &ResolvedItemName,
983 ) -> Result<PartialItemName, PlanError> {
984 let item = self.get_item_by_resolved_name(name)?;
985 Ok(self.catalog.minimal_qualification(item.name()))
986 }
987
988 pub fn dangerous_resolve_name(&self, name: Vec<&str>) -> ResolvedItemName {
991 tracing::trace!("dangerous_resolve_name {:?}", name);
992 let name: Vec<_> = name.into_iter().map(Ident::new_unchecked).collect();
994 let name = UnresolvedItemName::qualified(&name);
995 let entry = match self.resolve_item(RawItemName::Name(name.clone())) {
996 Ok(entry) => entry,
997 Err(_) => self
998 .resolve_function(name.clone())
999 .expect("name referred to an existing object"),
1000 };
1001
1002 let partial = normalize::unresolved_item_name(name).unwrap();
1003 let full_name = self.allocate_full_name(partial).unwrap();
1004
1005 ResolvedItemName::Item {
1006 id: entry.id(),
1007 qualifiers: entry.name().qualifiers.clone(),
1008 full_name,
1009 print_id: true,
1010 version: RelationVersionSelector::Latest,
1011 }
1012 }
1013}
1014
1015pub fn resolve_cluster_for_materialized_view<'a>(
1016 catalog: &'a dyn SessionCatalog,
1017 stmt: &CreateMaterializedViewStatement<Aug>,
1018) -> Result<ClusterId, PlanError> {
1019 Ok(match &stmt.in_cluster {
1020 None => catalog.resolve_cluster(None)?.id(),
1021 Some(in_cluster) => in_cluster.id,
1022 })
1023}
1024
1025#[derive(Debug, Clone, Copy)]
1027pub enum StatementClassification {
1028 ACL,
1029 DDL,
1030 DML,
1031 Other,
1032 SCL,
1033 Show,
1034 TCL,
1035}
1036
1037impl StatementClassification {
1038 pub fn is_ddl(&self) -> bool {
1039 matches!(self, StatementClassification::DDL)
1040 }
1041}
1042
1043impl<T: mz_sql_parser::ast::AstInfo> From<&Statement<T>> for StatementClassification {
1044 fn from(value: &Statement<T>) -> Self {
1045 use StatementClassification::*;
1046
1047 match value {
1048 Statement::AlterCluster(_) => DDL,
1050 Statement::AlterConnection(_) => DDL,
1051 Statement::AlterIndex(_) => DDL,
1052 Statement::AlterObjectRename(_) => DDL,
1053 Statement::AlterObjectSwap(_) => DDL,
1054 Statement::AlterNetworkPolicy(_) => DDL,
1055 Statement::AlterRetainHistory(_) => DDL,
1056 Statement::AlterRole(_) => DDL,
1057 Statement::AlterSecret(_) => DDL,
1058 Statement::AlterSetCluster(_) => DDL,
1059 Statement::AlterSink(_) => DDL,
1060 Statement::AlterSource(_) => DDL,
1061 Statement::AlterSystemSet(_) => DDL,
1062 Statement::AlterSystemReset(_) => DDL,
1063 Statement::AlterSystemResetAll(_) => DDL,
1064 Statement::AlterTableAddColumn(_) => DDL,
1065 Statement::Comment(_) => DDL,
1066 Statement::CreateCluster(_) => DDL,
1067 Statement::CreateClusterReplica(_) => DDL,
1068 Statement::CreateConnection(_) => DDL,
1069 Statement::CreateContinualTask(_) => DDL,
1070 Statement::CreateDatabase(_) => DDL,
1071 Statement::CreateIndex(_) => DDL,
1072 Statement::CreateRole(_) => DDL,
1073 Statement::CreateSchema(_) => DDL,
1074 Statement::CreateSecret(_) => DDL,
1075 Statement::CreateSink(_) => DDL,
1076 Statement::CreateWebhookSource(_) => DDL,
1077 Statement::CreateSource(_) => DDL,
1078 Statement::CreateSubsource(_) => DDL,
1079 Statement::CreateTable(_) => DDL,
1080 Statement::CreateTableFromSource(_) => DDL,
1081 Statement::CreateType(_) => DDL,
1082 Statement::CreateView(_) => DDL,
1083 Statement::CreateMaterializedView(_) => DDL,
1084 Statement::CreateNetworkPolicy(_) => DDL,
1085 Statement::DropObjects(_) => DDL,
1086 Statement::DropOwned(_) => DDL,
1087
1088 Statement::AlterOwner(_) => ACL,
1090 Statement::GrantRole(_) => ACL,
1091 Statement::RevokeRole(_) => ACL,
1092 Statement::GrantPrivileges(_) => ACL,
1093 Statement::RevokePrivileges(_) => ACL,
1094 Statement::AlterDefaultPrivileges(_) => ACL,
1095 Statement::ReassignOwned(_) => ACL,
1096
1097 Statement::Copy(_) => DML,
1099 Statement::Delete(_) => DML,
1100 Statement::ExplainPlan(_) => DML,
1101 Statement::ExplainPushdown(_) => DML,
1102 Statement::ExplainAnalyzeObject(_) => DML,
1103 Statement::ExplainAnalyzeCluster(_) => DML,
1104 Statement::ExplainTimestamp(_) => DML,
1105 Statement::ExplainSinkSchema(_) => DML,
1106 Statement::Insert(_) => DML,
1107 Statement::Select(_) => DML,
1108 Statement::Subscribe(_) => DML,
1109 Statement::Update(_) => DML,
1110
1111 Statement::Show(ShowStatement::ShowColumns(_)) => Show,
1113 Statement::Show(ShowStatement::ShowCreateConnection(_)) => Show,
1114 Statement::Show(ShowStatement::ShowCreateCluster(_)) => Show,
1115 Statement::Show(ShowStatement::ShowCreateIndex(_)) => Show,
1116 Statement::Show(ShowStatement::ShowCreateSink(_)) => Show,
1117 Statement::Show(ShowStatement::ShowCreateSource(_)) => Show,
1118 Statement::Show(ShowStatement::ShowCreateTable(_)) => Show,
1119 Statement::Show(ShowStatement::ShowCreateView(_)) => Show,
1120 Statement::Show(ShowStatement::ShowCreateMaterializedView(_)) => Show,
1121 Statement::Show(ShowStatement::ShowCreateType(_)) => Show,
1122 Statement::Show(ShowStatement::ShowObjects(_)) => Show,
1123
1124 Statement::Close(_) => SCL,
1126 Statement::Deallocate(_) => SCL,
1127 Statement::Declare(_) => SCL,
1128 Statement::Discard(_) => SCL,
1129 Statement::Execute(_) => SCL,
1130 Statement::Fetch(_) => SCL,
1131 Statement::Prepare(_) => SCL,
1132 Statement::ResetVariable(_) => SCL,
1133 Statement::SetVariable(_) => SCL,
1134 Statement::Show(ShowStatement::ShowVariable(_)) => SCL,
1135
1136 Statement::Commit(_) => TCL,
1138 Statement::Rollback(_) => TCL,
1139 Statement::SetTransaction(_) => TCL,
1140 Statement::StartTransaction(_) => TCL,
1141
1142 Statement::Raise(_) => Other,
1144 Statement::Show(ShowStatement::InspectShard(_)) => Other,
1145 Statement::ValidateConnection(_) => Other,
1146 }
1147 }
1148}