1use std::borrow::Cow;
40use std::cell::RefCell;
41use std::collections::{BTreeMap, BTreeSet};
42use std::convert::{TryFrom, TryInto};
43use std::num::NonZeroU64;
44use std::rc::Rc;
45use std::sync::{Arc, LazyLock};
46use std::{iter, mem};
47
48use itertools::Itertools;
49use mz_expr::func::variadic::{
50 ArrayCreate, ArrayIndex, Coalesce, Greatest, Least, ListCreate, ListIndex, ListSliceLinear,
51 MapBuild, RecordCreate,
52};
53use mz_expr::virtual_syntax::AlgExcept;
54use mz_expr::{
55 Eval, Id, LetRecLimit, LocalId, MapFilterProject, MirScalarExpr, REPEAT_ROW_NAME,
56 RowSetFinishing, TableFunc, func as expr_func,
57};
58use mz_ore::collections::CollectionExt;
59use mz_ore::error::ErrorExt;
60use mz_ore::id_gen::IdGen;
61use mz_ore::option::FallibleMapExt;
62use mz_ore::stack::{CheckedRecursion, RecursionGuard};
63use mz_ore::str::StrExt;
64use mz_repr::adt::char::CharLength;
65use mz_repr::adt::numeric::{NUMERIC_DATUM_MAX_PRECISION, NumericMaxScale};
66use mz_repr::adt::timestamp::TimestampPrecision;
67use mz_repr::adt::varchar::VarCharMaxLength;
68use mz_repr::namespaces::MZ_CATALOG_SCHEMA;
69use mz_repr::{
70 CatalogItemId, ColumnIndex, ColumnName, Datum, RelationDesc, RelationVersionSelector,
71 ReprColumnType, Row, RowArena, SqlColumnType, SqlRelationType, SqlScalarType,
72 UNKNOWN_COLUMN_NAME, strconv,
73};
74use mz_sql_parser::ast::display::AstDisplay;
75use mz_sql_parser::ast::visit::Visit;
76use mz_sql_parser::ast::visit_mut::{self, VisitMut};
77use mz_sql_parser::ast::{
78 AsOf, Assignment, AstInfo, CreateWebhookSourceBody, CreateWebhookSourceCheck,
79 CreateWebhookSourceHeader, CreateWebhookSourceSecret, CteBlock, DeleteStatement, Distinct,
80 Expr, Function, FunctionArgs, HomogenizingFunction, Ident, InsertSource, IsExprConstruct, Join,
81 JoinConstraint, JoinOperator, Limit, MapEntry, MutRecBlock, MutRecBlockOption,
82 MutRecBlockOptionName, OrderByExpr, Query, Select, SelectItem, SelectOption, SelectOptionName,
83 SetExpr, SetOperator, ShowStatement, SubscriptPosition, TableAlias, TableFactor,
84 TableWithJoins, UnresolvedItemName, UpdateStatement, Value, Values, WindowFrame,
85 WindowFrameBound, WindowFrameUnits, WindowSpec, visit,
86};
87use mz_sql_parser::ident;
88
89use crate::catalog::{CatalogItemType, CatalogType, SessionCatalog};
90use crate::func::{self, Func, FuncSpec, TableFuncImpl};
91use crate::names::{
92 Aug, FullItemName, PartialItemName, ResolvedDataType, ResolvedItemName, SchemaSpecifier,
93};
94use crate::plan::PlanError::InvalidWmrRecursionLimit;
95use crate::plan::error::PlanError;
96use crate::plan::hir::{
97 AbstractColumnType, AbstractExpr, AggregateExpr, AggregateFunc, AggregateWindowExpr,
98 BinaryFunc, CoercibleScalarExpr, CoercibleScalarType, ColumnOrder, ColumnRef, Hir,
99 HirRelationExpr, HirScalarExpr, JoinKind, ScalarWindowExpr, ScalarWindowFunc, UnaryFunc,
100 ValueWindowExpr, ValueWindowFunc, VariadicFunc, WindowExpr, WindowExprType,
101};
102use crate::plan::plan_utils::{self, GroupSizeHints, JoinSide};
103use crate::plan::scope::{Scope, ScopeItem, ScopeUngroupedColumn};
104use crate::plan::statement::{StatementContext, StatementDesc, show};
105use crate::plan::typeconv::{self, CastContext, plan_hypothetical_cast};
106use crate::plan::{
107 Params, PlanContext, QueryWhen, ShowCreatePlan, WebhookValidation, WebhookValidationSecret,
108 literal, transform_ast,
109};
110use crate::session::vars::ENABLE_WITH_ORDINALITY_LEGACY_FALLBACK;
111use crate::session::vars::{self, FeatureFlag};
112use crate::{ORDINALITY_COL_NAME, normalize};
113
114#[derive(Debug)]
115pub struct PlannedRootQuery<E> {
116 pub expr: E,
117 pub desc: RelationDesc,
118 pub finishing: RowSetFinishing<HirScalarExpr, HirScalarExpr>,
119 pub scope: Scope,
120}
121
122#[mz_ore::instrument(target = "compiler", level = "trace", name = "ast_to_hir")]
131pub fn plan_root_query(
132 scx: &StatementContext,
133 mut query: Query<Aug>,
134 lifetime: QueryLifetime,
135) -> Result<PlannedRootQuery<HirRelationExpr>, PlanError> {
136 transform_ast::transform(scx, &mut query)?;
137 let mut qcx = QueryContext::root(scx, lifetime);
138 let PlannedQuery {
139 mut expr,
140 scope,
141 order_by,
142 limit,
143 offset,
144 project,
145 group_size_hints,
146 } = plan_query(&mut qcx, &query)?;
147
148 let mut finishing = RowSetFinishing {
149 limit,
150 offset,
151 project,
152 order_by,
153 };
154
155 try_push_projection_order_by(&mut expr, &mut finishing.project, &mut finishing.order_by);
161
162 if lifetime.is_maintained() {
163 expr.finish_maintained(&mut finishing, group_size_hints);
164 }
165
166 let typ = qcx.relation_type(&expr);
167 let typ = SqlRelationType::new(
168 finishing
169 .project
170 .iter()
171 .map(|i| typ.column_types[*i].clone())
172 .collect(),
173 );
174 let desc = RelationDesc::new(typ, scope.column_names());
175
176 Ok(PlannedRootQuery {
177 expr,
178 desc,
179 finishing,
180 scope,
181 })
182}
183
184fn try_push_projection_order_by(
195 expr: &mut HirRelationExpr,
196 project: &mut Vec<usize>,
197 order_by: &mut Vec<ColumnOrder>,
198) -> bool {
199 let mut unproject = vec![None; expr.arity()];
200 for (out_i, in_i) in project.iter().copied().enumerate() {
201 unproject[in_i] = Some(out_i);
202 }
203 if order_by
204 .iter()
205 .all(|ob| ob.column < unproject.len() && unproject[ob.column].is_some())
206 {
207 let trivial_project = (0..project.len()).collect();
208 *expr = expr.take().project(mem::replace(project, trivial_project));
209 for ob in order_by {
210 ob.column = unproject[ob.column].unwrap();
211 }
212 true
213 } else {
214 false
215 }
216}
217
218pub fn plan_insert_query(
219 scx: &StatementContext,
220 table_name: ResolvedItemName,
221 columns: Vec<Ident>,
222 source: InsertSource<Aug>,
223 returning: Vec<SelectItem<Aug>>,
224) -> Result<
225 (
226 CatalogItemId,
227 HirRelationExpr,
228 PlannedRootQuery<Vec<HirScalarExpr>>,
229 ),
230 PlanError,
231> {
232 let mut qcx = QueryContext::root(scx, QueryLifetime::OneShot);
233 let table = scx.get_item_by_resolved_name(&table_name)?;
234
235 if table.item_type() != CatalogItemType::Table {
237 sql_bail!(
238 "cannot insert into {} '{}'",
239 table.item_type(),
240 table_name.full_name_str()
241 );
242 }
243 let desc = table
244 .relation_desc()
245 .ok_or_else(|| sql_err!("item does not have a relation description"))?;
246 let mut defaults = table
247 .writable_table_details()
248 .ok_or_else(|| {
249 sql_err!(
250 "cannot insert into non-writeable table '{}'",
251 table_name.full_name_str()
252 )
253 })?
254 .to_vec();
255
256 for default in &mut defaults {
257 transform_ast::transform(scx, default)?;
258 }
259
260 if table.id().is_system() {
261 sql_bail!(
262 "cannot insert into system table '{}'",
263 table_name.full_name_str()
264 );
265 }
266
267 let columns: Vec<_> = columns.into_iter().map(normalize::column_name).collect();
268
269 let mut source_types = Vec::with_capacity(columns.len());
271 let mut ordering = Vec::with_capacity(columns.len());
272
273 if columns.is_empty() {
274 source_types.extend(desc.iter_types().map(|x| &x.scalar_type));
277 ordering.extend(0..desc.arity());
278 } else {
279 let column_by_name: BTreeMap<&ColumnName, (usize, &SqlColumnType)> = desc
280 .iter()
281 .enumerate()
282 .map(|(idx, (name, typ))| (name, (idx, typ)))
283 .collect();
284
285 for c in &columns {
286 if let Some((idx, typ)) = column_by_name.get(c) {
287 ordering.push(*idx);
288 source_types.push(&typ.scalar_type);
289 } else {
290 sql_bail!(
291 "column {} of relation {} does not exist",
292 c.quoted(),
293 table_name.full_name_str().quoted()
294 );
295 }
296 }
297 if let Some(dup) = columns.iter().duplicates().next() {
298 sql_bail!("column {} specified more than once", dup.quoted());
299 }
300 };
301
302 let expr = match source {
304 InsertSource::Query(mut query) => {
305 transform_ast::transform(scx, &mut query)?;
306
307 match query {
308 Query {
310 body: SetExpr::Values(Values(values)),
311 ctes,
312 order_by,
313 limit: None,
314 offset: None,
315 } if ctes.is_empty() && order_by.is_empty() => {
316 let names: Vec<_> = ordering.iter().map(|i| desc.get_name(*i)).collect();
317 plan_values_insert(&qcx, &names, &source_types, &values)?
318 }
319 _ => {
320 let (expr, _scope) = plan_nested_query(&mut qcx, &query)?;
321 expr
322 }
323 }
324 }
325 InsertSource::DefaultValues => {
326 HirRelationExpr::constant(vec![vec![]], SqlRelationType::empty())
327 }
328 };
329
330 let expr_arity = expr.arity();
331
332 let max_columns = if columns.is_empty() {
335 desc.arity()
336 } else {
337 columns.len()
338 };
339 if expr_arity > max_columns {
340 sql_bail!("INSERT has more expressions than target columns");
341 }
342 if expr_arity < columns.len() {
344 sql_bail!("INSERT has more target columns than expressions");
345 }
346
347 source_types.truncate(expr_arity);
349 ordering.truncate(expr_arity);
350
351 let expr = cast_relation(&qcx, CastContext::Assignment, expr, source_types).map_err(|e| {
354 sql_err!(
355 "column {} is of type {} but expression is of type {}",
356 desc.get_name(ordering[e.column]).quoted(),
357 qcx.humanize_sql_scalar_type(&e.target_type, false),
358 qcx.humanize_sql_scalar_type(&e.source_type, false),
359 )
360 })?;
361
362 let mut map_exprs = vec![];
364 let mut project_key = Vec::with_capacity(desc.arity());
365
366 let col_to_source: BTreeMap<_, _> = ordering.iter().enumerate().map(|(a, b)| (b, a)).collect();
368
369 let column_details = desc.iter_types().zip_eq(defaults).enumerate();
370 for (col_idx, (col_typ, default)) in column_details {
371 if let Some(src_idx) = col_to_source.get(&col_idx) {
372 project_key.push(*src_idx);
373 } else {
374 let hir = plan_default_expr(scx, &default, &col_typ.scalar_type)?;
375 project_key.push(expr_arity + map_exprs.len());
376 map_exprs.push(hir);
377 }
378 }
379
380 let returning = {
381 let (scope, typ) = if let ResolvedItemName::Item {
382 full_name,
383 version: _,
384 ..
385 } = table_name
386 {
387 let scope = Scope::from_source(Some(full_name.clone().into()), desc.iter_names());
388 let typ = desc.typ().clone();
389 (scope, typ)
390 } else {
391 (Scope::empty(), SqlRelationType::empty())
392 };
393 let ecx = &ExprContext {
394 qcx: &qcx,
395 name: "RETURNING clause",
396 scope: &scope,
397 relation_type: &typ,
398 allow_aggregates: false,
399 allow_subqueries: false,
400 allow_parameters: true,
401 allow_windows: false,
402 };
403 let table_func_names = BTreeMap::new();
404 let mut output_columns = vec![];
405 let mut new_exprs = vec![];
406 let mut new_type = SqlRelationType::empty();
407 for mut si in returning {
408 transform_ast::transform(scx, &mut si)?;
409 for (select_item, column_name) in expand_select_item(ecx, &si, &table_func_names)? {
410 let expr = match &select_item {
411 ExpandedSelectItem::InputOrdinal(i) => HirScalarExpr::column(*i),
412 ExpandedSelectItem::Expr(expr) => plan_expr(ecx, expr)?.type_as_any(ecx)?,
413 };
414 output_columns.push(column_name);
415 let typ = ecx.column_type(&expr);
416 new_type.column_types.push(typ);
417 new_exprs.push(expr);
418 }
419 }
420 let desc = RelationDesc::new(new_type, output_columns);
421 let desc_arity = desc.arity();
422 PlannedRootQuery {
423 expr: new_exprs,
424 desc,
425 finishing: HirRelationExpr::trivial_row_set_finishing_hir(desc_arity),
426 scope,
427 }
428 };
429
430 Ok((
431 table.id(),
432 expr.map(map_exprs).project(project_key),
433 returning,
434 ))
435}
436
437pub fn plan_copy_item(
448 scx: &StatementContext,
449 item_name: ResolvedItemName,
450 columns: Vec<Ident>,
451) -> Result<
452 (
453 CatalogItemId,
454 RelationDesc,
455 Vec<ColumnIndex>,
456 Option<MapFilterProject>,
457 ),
458 PlanError,
459> {
460 let item = scx.get_item_by_resolved_name(&item_name)?;
461 let fullname = scx.catalog.resolve_full_name(item.name());
462 let table_desc = match item.relation_desc() {
463 Some(desc) => desc.into_owned(),
464 None => {
465 return Err(PlanError::InvalidDependency {
466 name: fullname.to_string(),
467 item_type: item.item_type().to_string(),
468 });
469 }
470 };
471 let mut ordering = Vec::with_capacity(columns.len());
472
473 let mfp = if let Some(table_defaults) = item.writable_table_details() {
483 let mut table_defaults = table_defaults.to_vec();
484
485 for default in &mut table_defaults {
486 transform_ast::transform(scx, default)?;
487 }
488
489 let source_column_names: Vec<_> = columns
491 .iter()
492 .cloned()
493 .map(normalize::column_name)
494 .collect();
495
496 let mut default_exprs = Vec::new();
497 let mut project_keys = Vec::with_capacity(table_desc.arity());
498
499 let column_details = table_desc.iter().zip_eq(table_defaults);
502 for ((col_name, col_type), col_default) in column_details {
503 let maybe_src_idx = source_column_names.iter().position(|name| name == col_name);
504 if let Some(src_idx) = maybe_src_idx {
505 project_keys.push(src_idx);
506 } else {
507 let hir = plan_default_expr(scx, &col_default, &col_type.scalar_type)?;
510 let mir = hir.lower_uncorrelated(scx.catalog.system_vars())?;
511 project_keys.push(source_column_names.len() + default_exprs.len());
512 default_exprs.push(mir);
513 }
514 }
515
516 let mfp = MapFilterProject::new(source_column_names.len())
517 .map(default_exprs)
518 .project(project_keys);
519 Some(mfp)
520 } else {
521 None
522 };
523
524 let source_desc = if columns.is_empty() {
526 let indexes = (0..table_desc.arity()).map(ColumnIndex::from_raw);
527 ordering.extend(indexes);
528
529 table_desc
531 } else {
532 let columns: Vec<_> = columns.into_iter().map(normalize::column_name).collect();
533 let column_by_name: BTreeMap<&ColumnName, (ColumnIndex, &SqlColumnType)> = table_desc
534 .iter_all()
535 .map(|(idx, name, typ)| (name, (*idx, typ)))
536 .collect();
537
538 let mut names = Vec::with_capacity(columns.len());
539 let mut source_types = Vec::with_capacity(columns.len());
540
541 for c in &columns {
542 if let Some((idx, typ)) = column_by_name.get(c) {
543 ordering.push(*idx);
544 source_types.push((*typ).clone());
545 names.push(c.clone());
546 } else {
547 sql_bail!(
548 "column {} of relation {} does not exist",
549 c.quoted(),
550 item_name.full_name_str().quoted()
551 );
552 }
553 }
554 if let Some(dup) = columns.iter().duplicates().next() {
555 sql_bail!("column {} specified more than once", dup.quoted());
556 }
557
558 RelationDesc::new(SqlRelationType::new(source_types), names)
560 };
561
562 Ok((item.id(), source_desc, ordering, mfp))
563}
564
565pub fn plan_copy_from(
569 scx: &StatementContext,
570 table_name: ResolvedItemName,
571 columns: Vec<Ident>,
572) -> Result<
573 (
574 CatalogItemId,
575 RelationDesc,
576 Vec<ColumnIndex>,
577 Option<MapFilterProject>,
578 ),
579 PlanError,
580> {
581 let table = scx.get_item_by_resolved_name(&table_name)?;
582
583 if table.item_type() != CatalogItemType::Table {
585 sql_bail!(
586 "cannot insert into {} '{}'",
587 table.item_type(),
588 table_name.full_name_str()
589 );
590 }
591
592 let _ = table.writable_table_details().ok_or_else(|| {
593 sql_err!(
594 "cannot insert into non-writeable table '{}'",
595 table_name.full_name_str()
596 )
597 })?;
598
599 if table.id().is_system() {
600 sql_bail!(
601 "cannot insert into system table '{}'",
602 table_name.full_name_str()
603 );
604 }
605 let (id, desc, ordering, mfp) = plan_copy_item(scx, table_name, columns)?;
606
607 Ok((id, desc, ordering, mfp))
608}
609
610pub fn plan_copy_from_rows(
613 pcx: &PlanContext,
614 catalog: &dyn SessionCatalog,
615 target_id: CatalogItemId,
616 target_name: String,
617 columns: Vec<ColumnIndex>,
618 rows: Vec<mz_repr::Row>,
619) -> Result<HirRelationExpr, PlanError> {
620 let scx = StatementContext::new(Some(pcx), catalog);
621
622 let table = catalog
624 .try_get_item(&target_id)
625 .ok_or_else(|| PlanError::CopyFromTargetTableDropped { target_name })?
626 .at_version(RelationVersionSelector::Latest);
627
628 let mut defaults = table
629 .writable_table_details()
630 .ok_or_else(|| sql_err!("cannot copy into non-writeable table"))?
631 .to_vec();
632
633 for default in &mut defaults {
634 transform_ast::transform(&scx, default)?;
635 }
636
637 let desc = table
638 .relation_desc()
639 .ok_or_else(|| sql_err!("item does not have a relation description"))?;
640 let column_types = columns
641 .iter()
642 .map(|x| desc.get_type(x).clone())
643 .map(|mut x| {
644 x.nullable = true;
647 x
648 })
649 .collect();
650 let typ = SqlRelationType::new(column_types);
651 let expr = HirRelationExpr::Constant {
652 rows,
653 typ: typ.clone(),
654 };
655
656 let default: Vec<_> = (0..desc.arity()).map(ColumnIndex::from_raw).collect();
662 if columns == default {
663 return Ok(expr);
664 }
665
666 let mut map_exprs = vec![];
668 let mut project_key = Vec::with_capacity(desc.arity());
669
670 let col_to_source: BTreeMap<_, _> = columns.iter().enumerate().map(|(a, b)| (b, a)).collect();
672
673 let column_details = desc.iter_all().zip_eq(defaults);
674 for ((col_idx, _col_name, col_typ), default) in column_details {
675 if let Some(src_idx) = col_to_source.get(&col_idx) {
676 project_key.push(*src_idx);
677 } else {
678 let hir = plan_default_expr(&scx, &default, &col_typ.scalar_type)?;
679 project_key.push(typ.arity() + map_exprs.len());
680 map_exprs.push(hir);
681 }
682 }
683
684 Ok(expr.map(map_exprs).project(project_key))
685}
686
687pub struct ReadThenWritePlan {
689 pub id: CatalogItemId,
690 pub selection: HirRelationExpr,
695 pub assignments: BTreeMap<usize, HirScalarExpr>,
697 pub finishing: RowSetFinishing,
698}
699
700pub fn plan_delete_query(
701 scx: &StatementContext,
702 mut delete_stmt: DeleteStatement<Aug>,
703) -> Result<ReadThenWritePlan, PlanError> {
704 transform_ast::transform(scx, &mut delete_stmt)?;
705
706 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
707 plan_mutation_query_inner(
708 qcx,
709 delete_stmt.table_name,
710 delete_stmt.alias,
711 delete_stmt.using,
712 vec![],
713 delete_stmt.selection,
714 )
715}
716
717pub fn plan_update_query(
718 scx: &StatementContext,
719 mut update_stmt: UpdateStatement<Aug>,
720) -> Result<ReadThenWritePlan, PlanError> {
721 transform_ast::transform(scx, &mut update_stmt)?;
722
723 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
724
725 plan_mutation_query_inner(
726 qcx,
727 update_stmt.table_name,
728 update_stmt.alias,
729 vec![],
730 update_stmt.assignments,
731 update_stmt.selection,
732 )
733}
734
735pub fn plan_mutation_query_inner(
736 qcx: QueryContext,
737 table_name: ResolvedItemName,
738 alias: Option<TableAlias>,
739 using: Vec<TableWithJoins<Aug>>,
740 assignments: Vec<Assignment<Aug>>,
741 selection: Option<Expr<Aug>>,
742) -> Result<ReadThenWritePlan, PlanError> {
743 let (id, version) = match table_name {
745 ResolvedItemName::Item { id, version, .. } => (id, version),
746 _ => sql_bail!("cannot mutate non-user table"),
747 };
748
749 let item = qcx.scx.get_item(&id).at_version(version);
751 if item.item_type() != CatalogItemType::Table {
752 sql_bail!(
753 "cannot mutate {} '{}'",
754 item.item_type(),
755 table_name.full_name_str()
756 );
757 }
758 let _ = item.writable_table_details().ok_or_else(|| {
759 sql_err!(
760 "cannot mutate non-writeable table '{}'",
761 table_name.full_name_str()
762 )
763 })?;
764 if id.is_system() {
765 sql_bail!(
766 "cannot mutate system table '{}'",
767 table_name.full_name_str()
768 );
769 }
770
771 let (mut get, scope) = qcx.resolve_table_name(table_name)?;
773 let scope = plan_table_alias(scope, alias.as_ref())?;
774 let desc = item.relation_desc().expect("table has desc");
775 let relation_type = qcx.relation_type(&get);
776
777 if using.is_empty() {
778 if let Some(expr) = selection {
779 let ecx = &ExprContext {
780 qcx: &qcx,
781 name: "WHERE clause",
782 scope: &scope,
783 relation_type: &relation_type,
784 allow_aggregates: false,
785 allow_subqueries: true,
786 allow_parameters: true,
787 allow_windows: false,
788 };
789 let expr = plan_expr(ecx, &expr)?.type_as(ecx, &SqlScalarType::Bool)?;
790 get = get.filter(vec![expr]);
791 }
792 } else {
793 get = handle_mutation_using_clause(&qcx, selection, using, get, scope.clone())?;
794 }
795
796 let mut sets = BTreeMap::new();
797 for Assignment { id, value } in assignments {
798 let name = normalize::column_name(id);
800 match desc.get_by_name(&name) {
801 Some((idx, typ)) => {
802 let ecx = &ExprContext {
803 qcx: &qcx,
804 name: "SET clause",
805 scope: &scope,
806 relation_type: &relation_type,
807 allow_aggregates: false,
808 allow_subqueries: false,
809 allow_parameters: true,
810 allow_windows: false,
811 };
812 let expr = plan_expr(ecx, &value)?.cast_to(
813 ecx,
814 CastContext::Assignment,
815 &typ.scalar_type,
816 )?;
817
818 if sets.insert(idx, expr).is_some() {
819 sql_bail!("column {} set twice", name)
820 }
821 }
822 None => sql_bail!("unknown column {}", name),
823 };
824 }
825
826 let finishing = RowSetFinishing {
827 order_by: vec![],
828 limit: None,
829 offset: 0,
830 project: (0..desc.arity()).collect(),
831 };
832
833 Ok(ReadThenWritePlan {
834 id,
835 selection: get,
836 finishing,
837 assignments: sets,
838 })
839}
840
841fn handle_mutation_using_clause(
853 qcx: &QueryContext,
854 selection: Option<Expr<Aug>>,
855 using: Vec<TableWithJoins<Aug>>,
856 get: HirRelationExpr,
857 outer_scope: Scope,
858) -> Result<HirRelationExpr, PlanError> {
859 let (mut using_rel_expr, using_scope) =
863 using.into_iter().try_fold(plan_join_identity(), |l, twj| {
864 let (left, left_scope) = l;
865 plan_join(
866 qcx,
867 left,
868 left_scope,
869 &Join {
870 relation: TableFactor::NestedJoin {
871 join: Box::new(twj),
872 alias: None,
873 },
874 join_operator: JoinOperator::CrossJoin,
875 },
876 )
877 })?;
878
879 if let Some(expr) = selection {
880 let on = HirScalarExpr::literal_true();
886 let joined = using_rel_expr
887 .clone()
888 .join(get.clone(), on, JoinKind::Inner);
889 let joined_scope = using_scope.product(outer_scope)?;
890 let joined_relation_type = qcx.relation_type(&joined);
891
892 let ecx = &ExprContext {
893 qcx,
894 name: "WHERE clause",
895 scope: &joined_scope,
896 relation_type: &joined_relation_type,
897 allow_aggregates: false,
898 allow_subqueries: true,
899 allow_parameters: true,
900 allow_windows: false,
901 };
902
903 let mut expr = plan_expr(ecx, &expr)?.type_as(ecx, &SqlScalarType::Bool)?;
905
906 let using_rel_arity = qcx.relation_type(&using_rel_expr).arity();
910 use mz_expr::visit::Visit;
912 expr.visit_mut_post(&mut |e| {
913 if let HirScalarExpr::Column(c, _name) = e {
914 if c.column >= using_rel_arity {
915 c.level += 1;
916 c.column -= using_rel_arity;
917 };
918 }
919 });
920
921 using_rel_expr = using_rel_expr.filter(vec![expr]);
925 } else {
926 let _joined_scope = using_scope.product(outer_scope)?;
929 }
930 Ok(get.filter(vec![using_rel_expr.exists()]))
941}
942
943#[derive(Debug)]
944pub(crate) struct CastRelationError {
945 pub(crate) column: usize,
946 pub(crate) source_type: SqlScalarType,
947 pub(crate) target_type: SqlScalarType,
948}
949
950pub(crate) fn cast_relation<'a, I>(
954 qcx: &QueryContext,
955 ccx: CastContext,
956 expr: HirRelationExpr,
957 target_types: I,
958) -> Result<HirRelationExpr, CastRelationError>
959where
960 I: IntoIterator<Item = &'a SqlScalarType>,
961{
962 let ecx = &ExprContext {
963 qcx,
964 name: "values",
965 scope: &Scope::empty(),
966 relation_type: &qcx.relation_type(&expr),
967 allow_aggregates: false,
968 allow_subqueries: true,
969 allow_parameters: true,
970 allow_windows: false,
971 };
972 let mut map_exprs = vec![];
973 let mut project_key = vec![];
974 for (i, target_typ) in target_types.into_iter().enumerate() {
975 let expr = HirScalarExpr::column(i);
976 match typeconv::plan_cast(ecx, ccx, expr.clone(), target_typ) {
980 Ok(cast_expr) => {
981 if expr == cast_expr {
982 project_key.push(i);
984 } else {
985 project_key.push(ecx.relation_type.arity() + map_exprs.len());
987 map_exprs.push(cast_expr);
988 }
989 }
990 Err(_) => {
991 return Err(CastRelationError {
992 column: i,
993 source_type: ecx.scalar_type(&expr),
994 target_type: target_typ.clone(),
995 });
996 }
997 }
998 }
999 Ok(expr.map(map_exprs).project(project_key))
1000}
1001
1002pub fn plan_as_of(
1005 scx: &StatementContext,
1006 as_of: Option<AsOf<Aug>>,
1007) -> Result<QueryWhen, PlanError> {
1008 match as_of {
1009 None => Ok(QueryWhen::Immediately),
1010 Some(as_of) => match as_of {
1011 AsOf::At(expr) => Ok(QueryWhen::AtTimestamp(plan_as_of_or_up_to(scx, expr)?)),
1012 AsOf::AtLeast(expr) => Ok(QueryWhen::AtLeastTimestamp(plan_as_of_or_up_to(scx, expr)?)),
1013 },
1014 }
1015}
1016
1017pub fn plan_as_of_or_up_to(
1027 scx: &StatementContext,
1028 mut expr: Expr<Aug>,
1029) -> Result<mz_repr::Timestamp, PlanError> {
1030 let scope = Scope::empty();
1031 let desc = RelationDesc::empty();
1032 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
1035 transform_ast::transform(scx, &mut expr)?;
1036 let ecx = &ExprContext {
1037 qcx: &qcx,
1038 name: "AS OF or UP TO",
1039 scope: &scope,
1040 relation_type: desc.typ(),
1041 allow_aggregates: false,
1042 allow_subqueries: false,
1043 allow_parameters: false,
1044 allow_windows: false,
1045 };
1046 let hir = plan_expr(ecx, &expr)?.cast_to(
1047 ecx,
1048 CastContext::Assignment,
1049 &SqlScalarType::MzTimestamp,
1050 )?;
1051 if hir.contains_unmaterializable() {
1052 bail_unsupported!("calling an unmaterializable function in AS OF or UP TO");
1053 }
1054 let timestamp = hir
1061 .into_literal_mz_timestamp()
1062 .ok_or_else(|| PlanError::InvalidAsOfUpTo)?;
1063 Ok(timestamp)
1064}
1065
1066pub fn plan_secret_as(
1068 scx: &StatementContext,
1069 mut expr: Expr<Aug>,
1070) -> Result<MirScalarExpr, PlanError> {
1071 let scope = Scope::empty();
1072 let desc = RelationDesc::empty();
1073 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
1074
1075 transform_ast::transform(scx, &mut expr)?;
1076
1077 let ecx = &ExprContext {
1078 qcx: &qcx,
1079 name: "AS",
1080 scope: &scope,
1081 relation_type: desc.typ(),
1082 allow_aggregates: false,
1083 allow_subqueries: false,
1084 allow_parameters: false,
1085 allow_windows: false,
1086 };
1087 let expr = plan_expr(ecx, &expr)?
1088 .type_as(ecx, &SqlScalarType::Bytes)?
1089 .lower_uncorrelated(scx.catalog.system_vars())?;
1090 Ok(expr)
1091}
1092
1093pub fn plan_webhook_validate_using(
1095 scx: &StatementContext,
1096 validate_using: CreateWebhookSourceCheck<Aug>,
1097) -> Result<WebhookValidation, PlanError> {
1098 let qcx = QueryContext::root(scx, QueryLifetime::Source);
1099
1100 let CreateWebhookSourceCheck {
1101 options,
1102 using: mut expr,
1103 } = validate_using;
1104
1105 let mut column_typs = vec![];
1106 let mut column_names = vec![];
1107
1108 let (bodies, headers, secrets) = options
1109 .map(|o| (o.bodies, o.headers, o.secrets))
1110 .unwrap_or_default();
1111
1112 let mut body_tuples = vec![];
1114 for CreateWebhookSourceBody { alias, use_bytes } in bodies {
1115 let scalar_type = use_bytes
1116 .then_some(SqlScalarType::Bytes)
1117 .unwrap_or(SqlScalarType::String);
1118 let name = alias
1119 .map(|a| a.into_string())
1120 .unwrap_or_else(|| "body".to_string());
1121
1122 column_typs.push(SqlColumnType {
1123 scalar_type,
1124 nullable: false,
1125 });
1126 column_names.push(name);
1127
1128 let column_idx = column_typs.len() - 1;
1130 assert_eq!(
1132 column_idx,
1133 column_names.len() - 1,
1134 "body column names and types don't match"
1135 );
1136 body_tuples.push((column_idx, use_bytes));
1137 }
1138
1139 let mut header_tuples = vec![];
1141
1142 for CreateWebhookSourceHeader { alias, use_bytes } in headers {
1143 let value_type = use_bytes
1144 .then_some(SqlScalarType::Bytes)
1145 .unwrap_or(SqlScalarType::String);
1146 let name = alias
1147 .map(|a| a.into_string())
1148 .unwrap_or_else(|| "headers".to_string());
1149
1150 column_typs.push(SqlColumnType {
1151 scalar_type: SqlScalarType::Map {
1152 value_type: Box::new(value_type),
1153 custom_id: None,
1154 },
1155 nullable: false,
1156 });
1157 column_names.push(name);
1158
1159 let column_idx = column_typs.len() - 1;
1161 assert_eq!(
1163 column_idx,
1164 column_names.len() - 1,
1165 "header column names and types don't match"
1166 );
1167 header_tuples.push((column_idx, use_bytes));
1168 }
1169
1170 let mut validation_secrets = vec![];
1172
1173 for CreateWebhookSourceSecret {
1174 secret,
1175 alias,
1176 use_bytes,
1177 } in secrets
1178 {
1179 let scalar_type = use_bytes
1181 .then_some(SqlScalarType::Bytes)
1182 .unwrap_or(SqlScalarType::String);
1183
1184 column_typs.push(SqlColumnType {
1185 scalar_type,
1186 nullable: false,
1187 });
1188 let ResolvedItemName::Item {
1189 id,
1190 full_name: FullItemName { item, .. },
1191 ..
1192 } = secret
1193 else {
1194 return Err(PlanError::InvalidSecret(Box::new(secret)));
1195 };
1196
1197 let name = if let Some(alias) = alias {
1199 alias.into_string()
1200 } else {
1201 item
1202 };
1203 column_names.push(name);
1204
1205 let column_idx = column_typs.len() - 1;
1208 assert_eq!(
1210 column_idx,
1211 column_names.len() - 1,
1212 "column names and types don't match"
1213 );
1214
1215 validation_secrets.push(WebhookValidationSecret {
1216 id,
1217 column_idx,
1218 use_bytes,
1219 });
1220 }
1221
1222 let relation_typ = SqlRelationType::new(column_typs);
1223 let desc = RelationDesc::new(relation_typ, column_names.clone());
1224 let scope = Scope::from_source(None, column_names);
1225
1226 transform_ast::transform(scx, &mut expr)?;
1227
1228 let ecx = &ExprContext {
1229 qcx: &qcx,
1230 name: "CHECK",
1231 scope: &scope,
1232 relation_type: desc.typ(),
1233 allow_aggregates: false,
1234 allow_subqueries: false,
1235 allow_parameters: false,
1236 allow_windows: false,
1237 };
1238 let expr = plan_expr(ecx, &expr)?
1239 .type_as(ecx, &SqlScalarType::Bool)?
1240 .lower_uncorrelated(scx.catalog.system_vars())?;
1241 let validation = WebhookValidation {
1242 expression: expr,
1243 relation_desc: desc,
1244 bodies: body_tuples,
1245 headers: header_tuples,
1246 secrets: validation_secrets,
1247 };
1248 Ok(validation)
1249}
1250
1251pub fn plan_default_expr(
1252 scx: &StatementContext,
1253 expr: &Expr<Aug>,
1254 target_ty: &SqlScalarType,
1255) -> Result<HirScalarExpr, PlanError> {
1256 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
1257 let ecx = &ExprContext {
1258 qcx: &qcx,
1259 name: "DEFAULT expression",
1260 scope: &Scope::empty(),
1261 relation_type: &SqlRelationType::empty(),
1262 allow_aggregates: false,
1263 allow_subqueries: false,
1264 allow_parameters: false,
1265 allow_windows: false,
1266 };
1267 let hir = plan_expr(ecx, expr)?.cast_to(ecx, CastContext::Assignment, target_ty)?;
1268 Ok(hir)
1269}
1270
1271pub fn plan_params<'a>(
1272 scx: &'a StatementContext,
1273 params: Vec<Expr<Aug>>,
1274 desc: &StatementDesc,
1275) -> Result<Params, PlanError> {
1276 if params.len() != desc.param_types.len() {
1277 sql_bail!(
1278 "expected {} params, got {}",
1279 desc.param_types.len(),
1280 params.len()
1281 );
1282 }
1283
1284 let qcx = QueryContext::root(scx, QueryLifetime::OneShot);
1285
1286 let mut datums = Row::default();
1287 let mut packer = datums.packer();
1288 let mut actual_types = Vec::new();
1289 let temp_storage = &RowArena::new();
1290 for (i, (mut expr, expected_ty)) in params.into_iter().zip_eq(&desc.param_types).enumerate() {
1291 transform_ast::transform(scx, &mut expr)?;
1292
1293 let ecx = execute_expr_context(&qcx);
1294 let ex = plan_expr(&ecx, &expr)?.type_as_any(&ecx)?;
1295 let actual_ty = ecx.scalar_type(&ex);
1296 if plan_hypothetical_cast(&ecx, *EXECUTE_CAST_CONTEXT, &actual_ty, expected_ty).is_none() {
1297 return Err(PlanError::WrongParameterType(
1298 i + 1,
1299 ecx.humanize_sql_scalar_type(expected_ty, false),
1300 ecx.humanize_sql_scalar_type(&actual_ty, false),
1301 ));
1302 }
1303 let ex = ex.lower_uncorrelated(scx.catalog.system_vars())?;
1304 let evaled = ex.eval(&[], temp_storage)?;
1305 packer.push(evaled);
1306 actual_types.push(actual_ty);
1307 }
1308 Ok(Params {
1309 datums,
1310 execute_types: actual_types,
1311 expected_types: desc.param_types.clone(),
1312 })
1313}
1314
1315static EXECUTE_CONTEXT_SCOPE: LazyLock<Scope> = LazyLock::new(Scope::empty);
1316static EXECUTE_CONTEXT_REL_TYPE: LazyLock<SqlRelationType> = LazyLock::new(SqlRelationType::empty);
1317
1318pub(crate) fn execute_expr_context<'a>(qcx: &'a QueryContext<'a>) -> ExprContext<'a> {
1320 ExprContext {
1321 qcx,
1322 name: "EXECUTE",
1323 scope: &EXECUTE_CONTEXT_SCOPE,
1324 relation_type: &EXECUTE_CONTEXT_REL_TYPE,
1325 allow_aggregates: false,
1326 allow_subqueries: false,
1327 allow_parameters: false,
1328 allow_windows: false,
1329 }
1330}
1331
1332pub(crate) static EXECUTE_CAST_CONTEXT: LazyLock<CastContext> =
1337 LazyLock::new(|| CastContext::Assignment);
1338
1339pub fn plan_index_exprs<'a>(
1340 scx: &'a StatementContext,
1341 on_desc: &RelationDesc,
1342 exprs: Vec<Expr<Aug>>,
1343) -> Result<Vec<mz_expr::MirScalarExpr>, PlanError> {
1344 let scope = Scope::from_source(None, on_desc.iter_names());
1345 let qcx = QueryContext::root(scx, QueryLifetime::Index);
1346
1347 let ecx = &ExprContext {
1348 qcx: &qcx,
1349 name: "CREATE INDEX",
1350 scope: &scope,
1351 relation_type: on_desc.typ(),
1352 allow_aggregates: false,
1353 allow_subqueries: false,
1354 allow_parameters: false,
1355 allow_windows: false,
1356 };
1357 let repr_col_types: Vec<ReprColumnType> = on_desc
1358 .typ()
1359 .column_types
1360 .iter()
1361 .map(ReprColumnType::from)
1362 .collect();
1363 let mut out = vec![];
1364 for mut expr in exprs {
1365 transform_ast::transform(scx, &mut expr)?;
1366 let expr = plan_expr_or_col_index(ecx, &expr)?;
1367 let mut expr = expr.lower_uncorrelated(scx.catalog.system_vars())?;
1368 expr.reduce(&repr_col_types);
1369 out.push(expr);
1370 }
1371 Ok(out)
1372}
1373
1374fn plan_expr_or_col_index(ecx: &ExprContext, e: &Expr<Aug>) -> Result<HirScalarExpr, PlanError> {
1375 match check_col_index(ecx.name, e, ecx.relation_type.column_types.len())? {
1376 Some(column) => Ok(HirScalarExpr::column(column)),
1377 _ => plan_expr(ecx, e)?.type_as_any(ecx),
1378 }
1379}
1380
1381fn check_col_index(name: &str, e: &Expr<Aug>, max: usize) -> Result<Option<usize>, PlanError> {
1382 match e {
1383 Expr::Value(Value::Number(n)) => {
1384 let n = n.parse::<usize>().map_err(|e| {
1385 sql_err!("unable to parse column reference in {}: {}: {}", name, n, e)
1386 })?;
1387 if n < 1 || n > max {
1388 sql_bail!(
1389 "column reference {} in {} is out of range (1 - {})",
1390 n,
1391 name,
1392 max
1393 );
1394 }
1395 Ok(Some(n - 1))
1396 }
1397 _ => Ok(None),
1398 }
1399}
1400
1401struct PlannedQuery {
1402 expr: HirRelationExpr,
1403 scope: Scope,
1404 order_by: Vec<ColumnOrder>,
1405 limit: Option<HirScalarExpr>,
1406 offset: HirScalarExpr,
1412 project: Vec<usize>,
1413 group_size_hints: GroupSizeHints,
1414}
1415
1416fn plan_query(qcx: &mut QueryContext, q: &Query<Aug>) -> Result<PlannedQuery, PlanError> {
1417 qcx.checked_recur_mut(|qcx| plan_query_inner(qcx, q))
1418}
1419
1420fn plan_query_inner(qcx: &mut QueryContext, q: &Query<Aug>) -> Result<PlannedQuery, PlanError> {
1421 let cte_bindings = plan_ctes(qcx, q)?;
1424
1425 let limit = match &q.limit {
1426 None => None,
1427 Some(Limit {
1428 quantity,
1429 with_ties: false,
1430 }) => {
1431 let ecx = &ExprContext {
1432 qcx,
1433 name: "LIMIT",
1434 scope: &Scope::empty(),
1435 relation_type: &SqlRelationType::empty(),
1436 allow_aggregates: false,
1437 allow_subqueries: true,
1438 allow_parameters: true,
1439 allow_windows: false,
1440 };
1441 let limit = plan_expr(ecx, quantity)?;
1442 let limit = limit.cast_to(ecx, CastContext::Explicit, &SqlScalarType::Int64)?;
1443
1444 let limit = if limit.is_constant() {
1445 let arena = RowArena::new();
1446 let limit = limit.lower_uncorrelated(qcx.scx.catalog.system_vars())?;
1447
1448 match limit.eval(&[], &arena)? {
1452 d @ Datum::Int64(v) if v >= 0 => {
1453 HirScalarExpr::literal(d, SqlScalarType::Int64)
1454 }
1455 d @ Datum::Null => HirScalarExpr::literal(d, SqlScalarType::Int64),
1456 Datum::Int64(_) => sql_bail!("LIMIT must not be negative"),
1457 _ => sql_bail!("constant LIMIT expression must reduce to an INT or NULL value"),
1458 }
1459 } else {
1460 qcx.scx
1462 .require_feature_flag(&vars::ENABLE_EXPRESSIONS_IN_LIMIT_SYNTAX)?;
1463 limit
1464 };
1465
1466 Some(limit)
1467 }
1468 Some(Limit {
1469 quantity: _,
1470 with_ties: true,
1471 }) => bail_unsupported!("FETCH ... WITH TIES"),
1472 };
1473
1474 let offset = match &q.offset {
1475 None => HirScalarExpr::literal(Datum::Int64(0), SqlScalarType::Int64),
1476 Some(offset) => {
1477 let ecx = &ExprContext {
1478 qcx,
1479 name: "OFFSET",
1480 scope: &Scope::empty(),
1481 relation_type: &SqlRelationType::empty(),
1482 allow_aggregates: false,
1483 allow_subqueries: false,
1484 allow_parameters: true,
1485 allow_windows: false,
1486 };
1487 let offset = plan_expr(ecx, offset)?;
1488 let offset = offset.cast_to(ecx, CastContext::Explicit, &SqlScalarType::Int64)?;
1489
1490 let offset = if offset.is_constant() {
1491 let offset_value = offset_into_value(offset)?;
1493 HirScalarExpr::literal(Datum::Int64(offset_value), SqlScalarType::Int64)
1494 } else {
1495 if !offset.contains_parameters() {
1499 return Err(PlanError::InvalidOffset(format!(
1500 "must be simplifiable to a constant, possibly after parameter binding, got {}",
1501 offset
1502 )));
1503 }
1504 offset
1505 };
1506 offset
1507 }
1508 };
1509
1510 let mut planned_query = match &q.body {
1511 SetExpr::Select(s) => {
1512 let select_option_extracted = SelectOptionExtracted::try_from(s.options.clone())?;
1514 let group_size_hints = GroupSizeHints::try_from(select_option_extracted)?;
1515
1516 let plan = plan_select_from_where(qcx, *s.clone(), q.order_by.clone())?;
1517 PlannedQuery {
1518 expr: plan.expr,
1519 scope: plan.scope,
1520 order_by: plan.order_by,
1521 project: plan.project,
1522 limit,
1523 offset,
1524 group_size_hints,
1525 }
1526 }
1527 _ => {
1528 let (expr, scope) = plan_set_expr(qcx, &q.body)?;
1529 let ecx = &ExprContext {
1530 qcx,
1531 name: "ORDER BY clause of a set expression",
1532 scope: &scope,
1533 relation_type: &qcx.relation_type(&expr),
1534 allow_aggregates: false,
1535 allow_subqueries: true,
1536 allow_parameters: true,
1537 allow_windows: false,
1538 };
1539 let output_columns: Vec<_> = scope.column_names().enumerate().collect();
1540 let (order_by, map_exprs) = plan_order_by_exprs(ecx, &q.order_by, &output_columns)?;
1541 let project = (0..ecx.relation_type.arity()).collect();
1542 PlannedQuery {
1543 expr: expr.map(map_exprs),
1544 scope,
1545 order_by,
1546 limit,
1547 project,
1548 offset,
1549 group_size_hints: GroupSizeHints::default(),
1550 }
1551 }
1552 };
1553
1554 match &q.ctes {
1556 CteBlock::Simple(_) => {
1557 for (id, value, shadowed_val) in cte_bindings.into_iter().rev() {
1558 if let Some(cte) = qcx.ctes.remove(&id) {
1559 planned_query.expr = HirRelationExpr::Let {
1560 name: cte.name,
1561 id: id.clone(),
1562 value: Box::new(value),
1563 body: Box::new(planned_query.expr),
1564 };
1565 }
1566 if let Some(shadowed_val) = shadowed_val {
1567 qcx.ctes.insert(id, shadowed_val);
1568 }
1569 }
1570 }
1571 CteBlock::MutuallyRecursive(MutRecBlock { options, ctes: _ }) => {
1572 let MutRecBlockOptionExtracted {
1573 recursion_limit,
1574 return_at_recursion_limit,
1575 error_at_recursion_limit,
1576 seen: _,
1577 } = MutRecBlockOptionExtracted::try_from(options.clone())?;
1578 let limit = match (
1579 recursion_limit,
1580 return_at_recursion_limit,
1581 error_at_recursion_limit,
1582 ) {
1583 (None, None, None) => None,
1584 (Some(max_iters), None, None) => {
1585 Some((max_iters, LetRecLimit::RETURN_AT_LIMIT_DEFAULT))
1586 }
1587 (None, Some(max_iters), None) => Some((max_iters, true)),
1588 (None, None, Some(max_iters)) => Some((max_iters, false)),
1589 _ => {
1590 return Err(InvalidWmrRecursionLimit(
1591 "More than one recursion limit given. \
1592 Please give at most one of RECURSION LIMIT, \
1593 ERROR AT RECURSION LIMIT, \
1594 RETURN AT RECURSION LIMIT."
1595 .to_owned(),
1596 ));
1597 }
1598 }
1599 .try_map(|(max_iters, return_at_limit)| {
1600 Ok::<LetRecLimit, PlanError>(LetRecLimit {
1601 max_iters: NonZeroU64::new(*max_iters).ok_or(InvalidWmrRecursionLimit(
1602 "Recursion limit has to be greater than 0.".to_owned(),
1603 ))?,
1604 return_at_limit: *return_at_limit,
1605 })
1606 })?;
1607
1608 let mut bindings = Vec::new();
1609 for (id, value, shadowed_val) in cte_bindings.into_iter() {
1610 if let Some(cte) = qcx.ctes.remove(&id) {
1611 bindings.push((cte.name, id, value, cte.desc.into_typ()));
1612 }
1613 if let Some(shadowed_val) = shadowed_val {
1614 qcx.ctes.insert(id, shadowed_val);
1615 }
1616 }
1617 if !bindings.is_empty() {
1618 planned_query.expr = HirRelationExpr::LetRec {
1619 limit,
1620 bindings,
1621 body: Box::new(planned_query.expr),
1622 }
1623 }
1624 }
1625 }
1626
1627 Ok(planned_query)
1628}
1629
1630pub(crate) fn offset_into_value(offset: HirScalarExpr) -> Result<i64, PlanError> {
1632 let offset = offset
1633 .try_into_literal_int64()
1634 .map_err(|err| PlanError::InvalidOffset(err.to_string_with_causes()))?;
1635 if offset < 0 {
1636 return Err(negative_offset_error(offset));
1637 }
1638 Ok(offset)
1639}
1640
1641pub(crate) fn negative_offset_error(offset: i64) -> PlanError {
1642 PlanError::InvalidOffset(format!("must not be negative, got {}", offset))
1643}
1644
1645generate_extracted_config!(
1646 MutRecBlockOption,
1647 (RecursionLimit, u64),
1648 (ReturnAtRecursionLimit, u64),
1649 (ErrorAtRecursionLimit, u64)
1650);
1651
1652pub fn plan_ctes(
1657 qcx: &mut QueryContext,
1658 q: &Query<Aug>,
1659) -> Result<Vec<(LocalId, HirRelationExpr, Option<CteDesc>)>, PlanError> {
1660 let mut result = Vec::new();
1662 let mut shadowed_descs = BTreeMap::new();
1665
1666 if let Some(ident) = q.ctes.bound_identifiers().duplicates().next() {
1668 sql_bail!(
1669 "WITH query name {} specified more than once",
1670 normalize::ident_ref(ident).quoted()
1671 )
1672 }
1673
1674 match &q.ctes {
1675 CteBlock::Simple(ctes) => {
1676 for cte in ctes.iter() {
1678 let cte_name = normalize::ident(cte.alias.name.clone());
1679 let (val, scope) = plan_nested_query(qcx, &cte.query)?;
1680 let typ = qcx.relation_type(&val);
1681 let mut desc = RelationDesc::new(typ, scope.column_names());
1682 plan_utils::maybe_rename_columns(
1683 format!("CTE {}", cte.alias.name),
1684 &mut desc,
1685 &cte.alias.columns,
1686 )?;
1687 let shadowed = qcx.ctes.insert(
1689 cte.id,
1690 CteDesc {
1691 name: cte_name,
1692 desc,
1693 },
1694 );
1695
1696 result.push((cte.id, val, shadowed));
1697 }
1698 }
1699 CteBlock::MutuallyRecursive(MutRecBlock { options: _, ctes }) => {
1700 for cte in ctes.iter() {
1702 let cte_name = normalize::ident(cte.name.clone());
1703 let mut desc_columns = Vec::with_capacity(cte.columns.capacity());
1704 for column in cte.columns.iter() {
1705 desc_columns.push((
1706 normalize::column_name(column.name.clone()),
1707 SqlColumnType {
1708 scalar_type: scalar_type_from_sql(qcx.scx, &column.data_type)?,
1709 nullable: true,
1710 },
1711 ));
1712 }
1713 let desc = RelationDesc::from_names_and_types(desc_columns);
1714 let shadowed = qcx.ctes.insert(
1715 cte.id,
1716 CteDesc {
1717 name: cte_name,
1718 desc,
1719 },
1720 );
1721 if let Some(shadowed) = shadowed {
1723 shadowed_descs.insert(cte.id, shadowed);
1724 }
1725 }
1726
1727 for cte in ctes.iter() {
1729 let (val, _scope) = plan_nested_query(qcx, &cte.query)?;
1730
1731 let proposed_typ = qcx.ctes[&cte.id].desc.typ();
1732
1733 if proposed_typ.column_types.iter().any(|c| !c.nullable) {
1734 sql_bail!(
1737 "[internal error]: WMR CTEs do not support NOT NULL constraints on proposed column types"
1738 );
1739 }
1740
1741 if !proposed_typ.keys.is_empty() {
1742 sql_bail!("[internal error]: WMR CTEs do not support keys");
1745 }
1746
1747 let derived_typ = qcx.relation_type(&val);
1749
1750 let type_err = |proposed_typ: &SqlRelationType, derived_typ: SqlRelationType| {
1751 let cte_name = normalize::ident(cte.name.clone());
1752 let proposed_typ = proposed_typ
1753 .column_types
1754 .iter()
1755 .map(|ty| qcx.humanize_sql_scalar_type(&ty.scalar_type, false))
1756 .collect::<Vec<_>>();
1757 let inferred_typ = derived_typ
1758 .column_types
1759 .iter()
1760 .map(|ty| qcx.humanize_sql_scalar_type(&ty.scalar_type, false))
1761 .collect::<Vec<_>>();
1762 Err(PlanError::RecursiveTypeMismatch(
1763 cte_name,
1764 proposed_typ,
1765 inferred_typ,
1766 ))
1767 };
1768
1769 if derived_typ.column_types.len() != proposed_typ.column_types.len() {
1770 return type_err(proposed_typ, derived_typ);
1771 }
1772
1773 let val = match cast_relation(
1775 qcx,
1776 CastContext::Assignment,
1781 val,
1782 proposed_typ.column_types.iter().map(|c| &c.scalar_type),
1783 ) {
1784 Ok(val) => val,
1785 Err(_) => return type_err(proposed_typ, derived_typ),
1786 };
1787
1788 result.push((cte.id, val, shadowed_descs.remove(&cte.id)));
1789 }
1790 }
1791 }
1792
1793 Ok(result)
1794}
1795
1796pub fn plan_nested_query(
1797 qcx: &mut QueryContext,
1798 q: &Query<Aug>,
1799) -> Result<(HirRelationExpr, Scope), PlanError> {
1800 let PlannedQuery {
1801 mut expr,
1802 scope,
1803 order_by,
1804 limit,
1805 offset,
1806 project,
1807 group_size_hints,
1808 } = qcx.checked_recur_mut(|qcx| plan_query(qcx, q))?;
1809 if limit.is_some()
1821 || !offset
1822 .clone()
1823 .try_into_literal_int64()
1824 .is_ok_and(|offset| offset == 0)
1825 {
1826 expr = HirRelationExpr::top_k(
1827 expr,
1828 vec![],
1829 order_by,
1830 limit,
1831 offset,
1832 group_size_hints.limit_input_group_size,
1833 );
1834 }
1835 Ok((expr.project(project), scope))
1836}
1837
1838fn plan_set_expr(
1839 qcx: &mut QueryContext,
1840 q: &SetExpr<Aug>,
1841) -> Result<(HirRelationExpr, Scope), PlanError> {
1842 match q {
1843 SetExpr::Select(select) => {
1844 let order_by_exprs = Vec::new();
1845 let plan = plan_select_from_where(qcx, *select.clone(), order_by_exprs)?;
1846 assert!(plan.order_by.is_empty());
1849 Ok((plan.expr.project(plan.project), plan.scope))
1850 }
1851 SetExpr::SetOperation {
1852 op,
1853 all,
1854 left,
1855 right,
1856 } => {
1857 let (left_expr, left_scope) = qcx.checked_recur_mut(|qcx| plan_set_expr(qcx, left))?;
1859 let (right_expr, right_scope) =
1860 qcx.checked_recur_mut(|qcx| plan_set_expr(qcx, right))?;
1861
1862 let left_type = qcx.relation_type(&left_expr);
1864 let right_type = qcx.relation_type(&right_expr);
1865 if left_type.arity() != right_type.arity() {
1866 sql_bail!(
1867 "each {} query must have the same number of columns: {} vs {}",
1868 op,
1869 left_type.arity(),
1870 right_type.arity(),
1871 );
1872 }
1873
1874 let left_ecx = &ExprContext {
1879 qcx,
1880 name: &op.to_string(),
1881 scope: &left_scope,
1882 relation_type: &left_type,
1883 allow_aggregates: false,
1884 allow_subqueries: false,
1885 allow_parameters: false,
1886 allow_windows: false,
1887 };
1888 let right_ecx = &ExprContext {
1889 qcx,
1890 name: &op.to_string(),
1891 scope: &right_scope,
1892 relation_type: &right_type,
1893 allow_aggregates: false,
1894 allow_subqueries: false,
1895 allow_parameters: false,
1896 allow_windows: false,
1897 };
1898 let mut left_casts = vec![];
1899 let mut right_casts = vec![];
1900 for (i, (left_type, right_type)) in left_type
1901 .column_types
1902 .iter()
1903 .zip_eq(right_type.column_types.iter())
1904 .enumerate()
1905 {
1906 let types = &[
1907 CoercibleScalarType::Coerced(left_type.scalar_type.clone()),
1908 CoercibleScalarType::Coerced(right_type.scalar_type.clone()),
1909 ];
1910 let target =
1911 typeconv::guess_best_common_type(&left_ecx.with_name(&op.to_string()), types)?;
1912 match typeconv::plan_cast(
1913 left_ecx,
1914 CastContext::Implicit,
1915 HirScalarExpr::column(i),
1916 &target,
1917 ) {
1918 Ok(expr) => left_casts.push(expr),
1919 Err(_) => sql_bail!(
1920 "{} types {} and {} cannot be matched",
1921 op,
1922 qcx.humanize_sql_scalar_type(&left_type.scalar_type, false),
1923 qcx.humanize_sql_scalar_type(&target, false),
1924 ),
1925 }
1926 match typeconv::plan_cast(
1927 right_ecx,
1928 CastContext::Implicit,
1929 HirScalarExpr::column(i),
1930 &target,
1931 ) {
1932 Ok(expr) => right_casts.push(expr),
1933 Err(_) => sql_bail!(
1934 "{} types {} and {} cannot be matched",
1935 op,
1936 qcx.humanize_sql_scalar_type(&target, false),
1937 qcx.humanize_sql_scalar_type(&right_type.scalar_type, false),
1938 ),
1939 }
1940 }
1941 let lhs = if left_casts
1942 .iter()
1943 .enumerate()
1944 .any(|(i, e)| e != &HirScalarExpr::column(i))
1945 {
1946 let project_key: Vec<_> = (left_type.arity()..left_type.arity() * 2).collect();
1947 left_expr.map(left_casts).project(project_key)
1948 } else {
1949 left_expr
1950 };
1951 let rhs = if right_casts
1952 .iter()
1953 .enumerate()
1954 .any(|(i, e)| e != &HirScalarExpr::column(i))
1955 {
1956 let project_key: Vec<_> = (right_type.arity()..right_type.arity() * 2).collect();
1957 right_expr.map(right_casts).project(project_key)
1958 } else {
1959 right_expr
1960 };
1961
1962 let relation_expr = match op {
1963 SetOperator::Union => {
1964 if *all {
1965 lhs.union(rhs)
1966 } else {
1967 lhs.union(rhs).distinct()
1968 }
1969 }
1970 SetOperator::Except => Hir::except(all, lhs, rhs),
1971 SetOperator::Intersect => {
1972 let (lhs, rhs) = if lhs.relation_node_count() > rhs.relation_node_count() {
1979 (rhs, lhs)
1980 } else {
1981 (lhs, rhs)
1982 };
1983 let left_clone = lhs.clone();
1991 if *all {
1992 lhs.union(left_clone.union(rhs.negate()).threshold().negate())
1993 } else {
1994 lhs.union(left_clone.union(rhs.negate()).threshold().negate())
1995 .distinct()
1996 }
1997 }
1998 };
1999 let scope = Scope::from_source(
2000 None,
2001 left_scope.column_names(),
2003 );
2004
2005 Ok((relation_expr, scope))
2006 }
2007 SetExpr::Values(Values(values)) => plan_values(qcx, values),
2008 SetExpr::Table(name) => {
2009 let (expr, scope) = qcx.resolve_table_name(name.clone())?;
2010 Ok((expr, scope))
2011 }
2012 SetExpr::Query(query) => {
2013 let (expr, scope) = plan_nested_query(qcx, query)?;
2014 Ok((expr, scope))
2015 }
2016 SetExpr::Show(stmt) => {
2017 if !qcx.lifetime.allow_show() {
2031 return Err(PlanError::ShowCommandInView);
2032 }
2033
2034 fn to_hirscope(
2037 plan: ShowCreatePlan,
2038 desc: StatementDesc,
2039 ) -> Result<(HirRelationExpr, Scope), PlanError> {
2040 let rows = vec![plan.row.iter().collect::<Vec<_>>()];
2041 let desc = desc.relation_desc.ok_or_else(|| {
2042 internal_err!("statement description missing relation descriptor")
2043 })?;
2044 let scope = Scope::from_source(None, desc.iter_names());
2045 let expr = HirRelationExpr::constant(rows, desc.into_typ());
2046 Ok((expr, scope))
2047 }
2048
2049 match stmt.clone() {
2050 ShowStatement::ShowColumns(stmt) => {
2051 show::show_columns(qcx.scx, stmt)?.plan_hir(qcx)
2052 }
2053 ShowStatement::ShowCreateConnection(stmt) => to_hirscope(
2054 show::plan_show_create_connection(qcx.scx, stmt.clone())?,
2055 show::describe_show_create_connection(qcx.scx, stmt)?,
2056 ),
2057 ShowStatement::ShowCreateCluster(stmt) => to_hirscope(
2058 show::plan_show_create_cluster(qcx.scx, stmt.clone())?,
2059 show::describe_show_create_cluster(qcx.scx, stmt)?,
2060 ),
2061 ShowStatement::ShowCreateIndex(stmt) => to_hirscope(
2062 show::plan_show_create_index(qcx.scx, stmt.clone())?,
2063 show::describe_show_create_index(qcx.scx, stmt)?,
2064 ),
2065 ShowStatement::ShowCreateSink(stmt) => to_hirscope(
2066 show::plan_show_create_sink(qcx.scx, stmt.clone())?,
2067 show::describe_show_create_sink(qcx.scx, stmt)?,
2068 ),
2069 ShowStatement::ShowCreateSource(stmt) => to_hirscope(
2070 show::plan_show_create_source(qcx.scx, stmt.clone())?,
2071 show::describe_show_create_source(qcx.scx, stmt)?,
2072 ),
2073 ShowStatement::ShowCreateTable(stmt) => to_hirscope(
2074 show::plan_show_create_table(qcx.scx, stmt.clone())?,
2075 show::describe_show_create_table(qcx.scx, stmt)?,
2076 ),
2077 ShowStatement::ShowCreateView(stmt) => to_hirscope(
2078 show::plan_show_create_view(qcx.scx, stmt.clone())?,
2079 show::describe_show_create_view(qcx.scx, stmt)?,
2080 ),
2081 ShowStatement::ShowCreateMaterializedView(stmt) => to_hirscope(
2082 show::plan_show_create_materialized_view(qcx.scx, stmt.clone())?,
2083 show::describe_show_create_materialized_view(qcx.scx, stmt)?,
2084 ),
2085 ShowStatement::ShowCreateType(stmt) => to_hirscope(
2086 show::plan_show_create_type(qcx.scx, stmt.clone())?,
2087 show::describe_show_create_type(qcx.scx, stmt)?,
2088 ),
2089 ShowStatement::ShowObjects(stmt) => {
2090 show::show_objects(qcx.scx, stmt)?.plan_hir(qcx)
2091 }
2092 ShowStatement::ShowVariable(_) => bail_unsupported!("SHOW variable in subqueries"),
2093 ShowStatement::InspectShard(_) => sql_bail!("unsupported INSPECT statement"),
2094 }
2095 }
2096 }
2097}
2098
2099fn plan_values(
2101 qcx: &QueryContext,
2102 values: &[Vec<Expr<Aug>>],
2103) -> Result<(HirRelationExpr, Scope), PlanError> {
2104 assert!(!values.is_empty());
2105
2106 let ecx = &ExprContext {
2107 qcx,
2108 name: "VALUES",
2109 scope: &Scope::empty(),
2110 relation_type: &SqlRelationType::empty(),
2111 allow_aggregates: false,
2112 allow_subqueries: true,
2113 allow_parameters: true,
2114 allow_windows: false,
2115 };
2116
2117 let ncols = values[0].len();
2118 let nrows = values.len();
2119
2120 let mut cols = vec![vec![]; ncols];
2123 for row in values {
2124 if row.len() != ncols {
2125 sql_bail!(
2126 "VALUES expression has varying number of columns: {} vs {}",
2127 row.len(),
2128 ncols
2129 );
2130 }
2131 for (i, v) in row.iter().enumerate() {
2132 cols[i].push(v);
2133 }
2134 }
2135
2136 let mut col_iters = Vec::with_capacity(ncols);
2138 let mut col_types = Vec::with_capacity(ncols);
2139 for col in &cols {
2140 let col = coerce_homogeneous_exprs(ecx, plan_exprs(ecx, col)?, None)?;
2141 let mut col_type = ecx.column_type(&col[0]);
2142 for val in &col[1..] {
2143 col_type = col_type.sql_union(&ecx.column_type(val))?; }
2145 col_types.push(col_type);
2146 col_iters.push(col.into_iter());
2147 }
2148
2149 let mut exprs = vec![];
2151 for _ in 0..nrows {
2152 for i in 0..ncols {
2153 exprs.push(col_iters[i].next().unwrap());
2154 }
2155 }
2156 let out = HirRelationExpr::CallTable {
2157 func: TableFunc::Wrap {
2158 width: ncols,
2159 types: col_types,
2160 },
2161 exprs,
2162 };
2163
2164 let mut scope = Scope::empty();
2166 for i in 0..ncols {
2167 let name = format!("column{}", i + 1);
2168 scope.items.push(ScopeItem::from_column_name(name));
2169 }
2170
2171 Ok((out, scope))
2172}
2173
2174fn plan_values_insert(
2184 qcx: &QueryContext,
2185 target_names: &[&ColumnName],
2186 target_types: &[&SqlScalarType],
2187 values: &[Vec<Expr<Aug>>],
2188) -> Result<HirRelationExpr, PlanError> {
2189 assert!(!values.is_empty());
2190
2191 if !values.iter().map(|row| row.len()).all_equal() {
2192 sql_bail!("VALUES lists must all be the same length");
2193 }
2194
2195 let ecx = &ExprContext {
2196 qcx,
2197 name: "VALUES",
2198 scope: &Scope::empty(),
2199 relation_type: &SqlRelationType::empty(),
2200 allow_aggregates: false,
2201 allow_subqueries: true,
2202 allow_parameters: true,
2203 allow_windows: false,
2204 };
2205
2206 let mut exprs = vec![];
2207 let mut types = vec![];
2208 for row in values {
2209 if row.len() > target_names.len() {
2210 sql_bail!("INSERT has more expressions than target columns");
2211 }
2212 for (column, val) in row.into_iter().enumerate() {
2213 let target_type = &target_types[column];
2214 let val = plan_expr(ecx, val)?;
2215 let val = typeconv::plan_coerce(ecx, val, target_type)?;
2216 let source_type = &ecx.scalar_type(&val);
2217 let val = match typeconv::plan_cast(ecx, CastContext::Assignment, val, target_type) {
2218 Ok(val) => val,
2219 Err(_) => sql_bail!(
2220 "column {} is of type {} but expression is of type {}",
2221 target_names[column].quoted(),
2222 qcx.humanize_sql_scalar_type(target_type, false),
2223 qcx.humanize_sql_scalar_type(source_type, false),
2224 ),
2225 };
2226 if column >= types.len() {
2227 types.push(ecx.column_type(&val));
2228 } else {
2229 types[column] = types[column].sql_union(&ecx.column_type(&val))?; }
2231 exprs.push(val);
2232 }
2233 }
2234
2235 Ok(HirRelationExpr::CallTable {
2236 func: TableFunc::Wrap {
2237 width: values[0].len(),
2238 types,
2239 },
2240 exprs,
2241 })
2242}
2243
2244fn plan_join_identity() -> (HirRelationExpr, Scope) {
2245 let typ = SqlRelationType::new(vec![]);
2246 let expr = HirRelationExpr::constant(vec![vec![]], typ);
2247 let scope = Scope::empty();
2248 (expr, scope)
2249}
2250
2251#[derive(Debug)]
2257struct SelectPlan {
2258 expr: HirRelationExpr,
2259 scope: Scope,
2260 order_by: Vec<ColumnOrder>,
2261 project: Vec<usize>,
2262}
2263
2264generate_extracted_config!(
2265 SelectOption,
2266 (ExpectedGroupSize, u64),
2267 (AggregateInputGroupSize, u64),
2268 (DistinctOnInputGroupSize, u64),
2269 (LimitInputGroupSize, u64)
2270);
2271
2272fn plan_select_from_where(
2290 qcx: &QueryContext,
2291 mut s: Select<Aug>,
2292 mut order_by_exprs: Vec<OrderByExpr<Aug>>,
2293) -> Result<SelectPlan, PlanError> {
2294 let select_option_extracted = SelectOptionExtracted::try_from(s.options.clone())?;
2301 let group_size_hints = GroupSizeHints::try_from(select_option_extracted)?;
2302
2303 let (mut relation_expr, mut from_scope) =
2305 s.from.iter().try_fold(plan_join_identity(), |l, twj| {
2306 let (left, left_scope) = l;
2307 plan_join(
2308 qcx,
2309 left,
2310 left_scope,
2311 &Join {
2312 relation: TableFactor::NestedJoin {
2313 join: Box::new(twj.clone()),
2314 alias: None,
2315 },
2316 join_operator: JoinOperator::CrossJoin,
2317 },
2318 )
2319 })?;
2320
2321 if let Some(selection) = &s.selection {
2323 let ecx = &ExprContext {
2324 qcx,
2325 name: "WHERE clause",
2326 scope: &from_scope,
2327 relation_type: &qcx.relation_type(&relation_expr),
2328 allow_aggregates: false,
2329 allow_subqueries: true,
2330 allow_parameters: true,
2331 allow_windows: false,
2332 };
2333 let expr = plan_expr(ecx, selection)
2334 .map_err(|e| sql_err!("WHERE clause error: {}", e))?
2335 .type_as(ecx, &SqlScalarType::Bool)?;
2336 relation_expr = relation_expr.filter(vec![expr]);
2337 }
2338
2339 let (aggregates, table_funcs) = {
2342 let mut visitor = AggregateTableFuncVisitor::new(qcx.scx);
2343 visitor.visit_select_mut(&mut s);
2344 for o in order_by_exprs.iter_mut() {
2345 visitor.visit_order_by_expr_mut(o);
2346 }
2347 visitor.into_result()?
2348 };
2349 let mut table_func_names: BTreeMap<String, Ident> = BTreeMap::new();
2350 if !table_funcs.is_empty() {
2351 let (expr, scope) = plan_scalar_table_funcs(
2352 qcx,
2353 table_funcs,
2354 &mut table_func_names,
2355 &relation_expr,
2356 &from_scope,
2357 )?;
2358 relation_expr = relation_expr.join(expr, HirScalarExpr::literal_true(), JoinKind::Inner);
2359 from_scope = from_scope.product(scope)?;
2360 }
2361
2362 let projection = {
2364 let ecx = &ExprContext {
2365 qcx,
2366 name: "SELECT clause",
2367 scope: &from_scope,
2368 relation_type: &qcx.relation_type(&relation_expr),
2369 allow_aggregates: true,
2370 allow_subqueries: true,
2371 allow_parameters: true,
2372 allow_windows: true,
2373 };
2374 let mut out = vec![];
2375 for si in &s.projection {
2376 if *si == SelectItem::Wildcard && s.from.is_empty() {
2377 sql_bail!("SELECT * with no tables specified is not valid");
2378 }
2379 out.extend(expand_select_item(ecx, si, &table_func_names)?);
2380 }
2381 out
2382 };
2383
2384 let (mut group_scope, select_all_mapping) = {
2388 let ecx = &ExprContext {
2390 qcx,
2391 name: "GROUP BY clause",
2392 scope: &from_scope,
2393 relation_type: &qcx.relation_type(&relation_expr),
2394 allow_aggregates: false,
2395 allow_subqueries: true,
2396 allow_parameters: true,
2397 allow_windows: false,
2398 };
2399 let mut group_key = vec![];
2400 let mut group_exprs: BTreeMap<HirScalarExpr, ScopeItem> = BTreeMap::new();
2401 let mut group_hir_exprs = vec![];
2402 let mut group_scope = Scope::empty();
2403 let mut select_all_mapping = BTreeMap::new();
2404
2405 for group_expr in &s.group_by {
2406 let (group_expr, expr) = plan_group_by_expr(ecx, group_expr, &projection)?;
2407 let new_column = group_key.len();
2408
2409 if let Some(existing_scope_item) = group_exprs.get_mut(&expr) {
2420 if let Some(group_expr) = group_expr {
2424 existing_scope_item.exprs.insert(group_expr.clone());
2425 }
2426 continue;
2427 }
2428
2429 let mut scope_item = if let HirScalarExpr::Column(
2430 ColumnRef {
2431 level: 0,
2432 column: old_column,
2433 },
2434 _name,
2435 ) = &expr
2436 {
2437 select_all_mapping.insert(*old_column, new_column);
2443 let scope_item = ecx.scope.items[*old_column].clone();
2444 scope_item
2445 } else {
2446 ScopeItem::empty()
2447 };
2448
2449 if let Some(group_expr) = group_expr.cloned() {
2450 scope_item.exprs.insert(group_expr);
2451 }
2452
2453 group_key.push(from_scope.len() + group_exprs.len());
2454 group_hir_exprs.push(expr.clone());
2455 group_exprs.insert(expr, scope_item);
2456 }
2457
2458 assert_eq!(group_hir_exprs.len(), group_exprs.len());
2459 for expr in &group_hir_exprs {
2460 if let Some(scope_item) = group_exprs.remove(expr) {
2461 group_scope.items.push(scope_item);
2462 }
2463 }
2464
2465 let ecx = &ExprContext {
2467 qcx,
2468 name: "aggregate function",
2469 scope: &from_scope,
2470 relation_type: &qcx.relation_type(&relation_expr.clone().map(group_hir_exprs.clone())),
2471 allow_aggregates: false,
2472 allow_subqueries: true,
2473 allow_parameters: true,
2474 allow_windows: false,
2475 };
2476 let mut agg_exprs = vec![];
2477 for sql_function in aggregates {
2478 if sql_function.over.is_some() {
2479 unreachable!(
2480 "Window aggregate; AggregateTableFuncVisitor explicitly filters these out"
2481 );
2482 }
2483 agg_exprs.push(plan_aggregate_common(ecx, &sql_function)?);
2484 group_scope
2485 .items
2486 .push(ScopeItem::from_expr(Expr::Function(sql_function.clone())));
2487 }
2488 if !agg_exprs.is_empty() || !group_key.is_empty() || s.having.is_some() {
2489 relation_expr = relation_expr.map(group_hir_exprs).reduce(
2491 group_key,
2492 agg_exprs,
2493 group_size_hints.aggregate_input_group_size,
2494 );
2495
2496 for i in 0..from_scope.len() {
2502 if !select_all_mapping.contains_key(&i) {
2503 let scope_item = &ecx.scope.items[i];
2504 group_scope.ungrouped_columns.push(ScopeUngroupedColumn {
2505 table_name: scope_item.table_name.clone(),
2506 column_name: scope_item.column_name.clone(),
2507 allow_unqualified_references: scope_item.allow_unqualified_references,
2508 });
2509 }
2510 }
2511
2512 (group_scope, select_all_mapping)
2513 } else {
2514 (
2516 from_scope.clone(),
2517 (0..from_scope.len()).map(|i| (i, i)).collect(),
2518 )
2519 }
2520 };
2521
2522 if let Some(ref having) = s.having {
2524 let ecx = &ExprContext {
2525 qcx,
2526 name: "HAVING clause",
2527 scope: &group_scope,
2528 relation_type: &qcx.relation_type(&relation_expr),
2529 allow_aggregates: true,
2530 allow_subqueries: true,
2531 allow_parameters: true,
2532 allow_windows: false,
2533 };
2534 let expr = plan_expr(ecx, having)?.type_as(ecx, &SqlScalarType::Bool)?;
2535 relation_expr = relation_expr.filter(vec![expr]);
2536 }
2537
2538 let window_funcs = {
2551 let mut visitor = WindowFuncCollector::default();
2552 visitor.visit_select(&s);
2556 for o in order_by_exprs.iter() {
2557 visitor.visit_order_by_expr(o);
2558 }
2559 visitor.into_result()
2560 };
2561 for window_func in window_funcs {
2562 let ecx = &ExprContext {
2563 qcx,
2564 name: "window function",
2565 scope: &group_scope,
2566 relation_type: &qcx.relation_type(&relation_expr),
2567 allow_aggregates: true,
2568 allow_subqueries: true,
2569 allow_parameters: true,
2570 allow_windows: true,
2571 };
2572 relation_expr = relation_expr.map(vec![plan_expr(ecx, &window_func)?.type_as_any(ecx)?]);
2573 group_scope.items.push(ScopeItem::from_expr(window_func));
2574 }
2575 if let Some(ref qualify) = s.qualify {
2582 let ecx = &ExprContext {
2583 qcx,
2584 name: "QUALIFY clause",
2585 scope: &group_scope,
2586 relation_type: &qcx.relation_type(&relation_expr),
2587 allow_aggregates: true,
2588 allow_subqueries: true,
2589 allow_parameters: true,
2590 allow_windows: true,
2591 };
2592 let expr = plan_expr(ecx, qualify)?.type_as(ecx, &SqlScalarType::Bool)?;
2593 relation_expr = relation_expr.filter(vec![expr]);
2594 }
2595
2596 let output_columns = {
2598 let mut new_exprs = vec![];
2599 let mut new_type = qcx.relation_type(&relation_expr);
2600 let mut output_columns = vec![];
2601 for (select_item, column_name) in &projection {
2602 let ecx = &ExprContext {
2603 qcx,
2604 name: "SELECT clause",
2605 scope: &group_scope,
2606 relation_type: &new_type,
2607 allow_aggregates: true,
2608 allow_subqueries: true,
2609 allow_parameters: true,
2610 allow_windows: true,
2611 };
2612 let expr = match select_item {
2613 ExpandedSelectItem::InputOrdinal(i) => {
2614 if let Some(column) = select_all_mapping.get(i).copied() {
2615 HirScalarExpr::column(column)
2616 } else {
2617 return Err(PlanError::ungrouped_column(&from_scope.items[*i]));
2618 }
2619 }
2620 ExpandedSelectItem::Expr(expr) => plan_expr(ecx, expr)?.type_as_any(ecx)?,
2621 };
2622 if let HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) = expr {
2623 output_columns.push((column, column_name));
2625 } else {
2626 let typ = ecx.column_type(&expr);
2633 new_type.column_types.push(typ);
2634 new_exprs.push(expr);
2635 output_columns.push((group_scope.len(), column_name));
2636 group_scope
2637 .items
2638 .push(ScopeItem::from_expr(select_item.as_expr().cloned()));
2639 }
2640 }
2641 relation_expr = relation_expr.map(new_exprs);
2642 output_columns
2643 };
2644 let mut project_key: Vec<_> = output_columns.iter().map(|(i, _name)| *i).collect();
2645
2646 let order_by = {
2648 let relation_type = qcx.relation_type(&relation_expr);
2649 let (mut order_by, mut map_exprs) = plan_order_by_exprs(
2650 &ExprContext {
2651 qcx,
2652 name: "ORDER BY clause",
2653 scope: &group_scope,
2654 relation_type: &relation_type,
2655 allow_aggregates: true,
2656 allow_subqueries: true,
2657 allow_parameters: true,
2658 allow_windows: true,
2659 },
2660 &order_by_exprs,
2661 &output_columns,
2662 )?;
2663
2664 match s.distinct {
2665 None => relation_expr = relation_expr.map(map_exprs),
2666 Some(Distinct::EntireRow) => {
2667 if relation_type.arity() == 0 {
2668 sql_bail!("SELECT DISTINCT must have at least one column");
2669 }
2670 if !try_push_projection_order_by(
2674 &mut relation_expr,
2675 &mut project_key,
2676 &mut order_by,
2677 ) {
2678 sql_bail!(
2679 "for SELECT DISTINCT, ORDER BY expressions must appear in select list"
2680 );
2681 }
2682 assert!(map_exprs.is_empty());
2683 relation_expr = relation_expr.distinct();
2684 }
2685 Some(Distinct::On(exprs)) => {
2686 let ecx = &ExprContext {
2687 qcx,
2688 name: "DISTINCT ON clause",
2689 scope: &group_scope,
2690 relation_type: &qcx.relation_type(&relation_expr),
2691 allow_aggregates: true,
2692 allow_subqueries: true,
2693 allow_parameters: true,
2694 allow_windows: true,
2695 };
2696
2697 let mut distinct_exprs = vec![];
2698 for expr in &exprs {
2699 let expr = plan_order_by_or_distinct_expr(ecx, expr, &output_columns)?;
2700 distinct_exprs.push(expr);
2701 }
2702
2703 let mut distinct_key = vec![];
2704
2705 let arity = relation_type.arity();
2715 for ord in order_by.iter().take(distinct_exprs.len()) {
2716 let mut expr = &HirScalarExpr::column(ord.column);
2719 if ord.column >= arity {
2720 expr = &map_exprs[ord.column - arity];
2721 };
2722 match distinct_exprs.iter().position(move |e| e == expr) {
2723 None => sql_bail!(
2724 "SELECT DISTINCT ON expressions must match initial ORDER BY expressions"
2725 ),
2726 Some(pos) => {
2727 distinct_exprs.remove(pos);
2728 }
2729 }
2730 distinct_key.push(ord.column);
2731 }
2732
2733 for expr in distinct_exprs {
2735 let column = match expr {
2738 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
2739 _ => {
2740 map_exprs.push(expr);
2741 arity + map_exprs.len() - 1
2742 }
2743 };
2744 distinct_key.push(column);
2745 }
2746
2747 let distinct_len = distinct_key.len();
2752 relation_expr = HirRelationExpr::top_k(
2753 relation_expr.map(map_exprs),
2754 distinct_key,
2755 order_by.iter().skip(distinct_len).cloned().collect(),
2756 Some(HirScalarExpr::literal(
2757 Datum::Int64(1),
2758 SqlScalarType::Int64,
2759 )),
2760 HirScalarExpr::literal(Datum::Int64(0), SqlScalarType::Int64),
2761 group_size_hints.distinct_on_input_group_size,
2762 );
2763 }
2764 }
2765
2766 order_by
2767 };
2768
2769 let scope = Scope::from_source(None, projection.into_iter().map(|(_expr, name)| name));
2774
2775 Ok(SelectPlan {
2776 expr: relation_expr,
2777 scope,
2778 order_by,
2779 project: project_key,
2780 })
2781}
2782
2783fn plan_scalar_table_funcs(
2784 qcx: &QueryContext,
2785 table_funcs: BTreeMap<Function<Aug>, String>,
2786 table_func_names: &mut BTreeMap<String, Ident>,
2787 relation_expr: &HirRelationExpr,
2788 from_scope: &Scope,
2789) -> Result<(HirRelationExpr, Scope), PlanError> {
2790 let rows_from_qcx = qcx.derived_context(from_scope.clone(), qcx.relation_type(relation_expr));
2791
2792 for (table_func, id) in table_funcs.iter() {
2793 table_func_names.insert(
2794 id.clone(),
2795 Ident::new_unchecked(table_func.name.full_item_name().item.clone()),
2797 );
2798 }
2799 if table_funcs.len() == 1 {
2802 let (table_func, id) = table_funcs.iter().next().unwrap();
2803 let (expr, mut scope) =
2804 plan_solitary_table_function(&rows_from_qcx, table_func, None, false)?;
2805
2806 let num_cols = scope.len();
2808 for i in 0..scope.len() {
2809 scope.items[i].table_name = Some(PartialItemName {
2810 database: None,
2811 schema: None,
2812 item: id.clone(),
2813 });
2814 scope.items[i].from_single_column_function = num_cols == 1;
2815 scope.items[i].allow_unqualified_references = false;
2816 }
2817 return Ok((expr, scope));
2818 }
2819 if table_funcs.keys().any(is_repeat_row) {
2820 bail_unsupported!(format!(
2823 "{} in a SELECT clause with multiple table functions",
2824 REPEAT_ROW_NAME
2825 ));
2826 }
2827 let (expr, mut scope, num_cols) =
2829 plan_rows_from_internal(&rows_from_qcx, table_funcs.keys(), None)?;
2830
2831 let mut i = 0;
2833 for (id, num_cols) in table_funcs.values().zip_eq(num_cols) {
2834 for _ in 0..num_cols {
2835 scope.items[i].table_name = Some(PartialItemName {
2836 database: None,
2837 schema: None,
2838 item: id.clone(),
2839 });
2840 scope.items[i].from_single_column_function = num_cols == 1;
2841 scope.items[i].allow_unqualified_references = false;
2842 i += 1;
2843 }
2844 scope.items[i].table_name = Some(PartialItemName {
2848 database: None,
2849 schema: None,
2850 item: id.clone(),
2851 });
2852 scope.items[i].is_exists_column_for_a_table_function_that_was_in_the_target_list = true;
2853 scope.items[i].allow_unqualified_references = false;
2854 i += 1;
2855 }
2856 scope.items[i].allow_unqualified_references = false;
2858 Ok((expr, scope))
2859}
2860
2861fn plan_group_by_expr<'a>(
2868 ecx: &ExprContext,
2869 group_expr: &'a Expr<Aug>,
2870 projection: &'a [(ExpandedSelectItem, ColumnName)],
2871) -> Result<(Option<&'a Expr<Aug>>, HirScalarExpr), PlanError> {
2872 let plan_projection = |column: usize| match &projection[column].0 {
2873 ExpandedSelectItem::InputOrdinal(column) => Ok((None, HirScalarExpr::column(*column))),
2874 ExpandedSelectItem::Expr(expr) => {
2875 Ok((Some(expr.as_ref()), plan_expr(ecx, expr)?.type_as_any(ecx)?))
2876 }
2877 };
2878
2879 if let Some(column) = check_col_index(ecx.name, group_expr, projection.len())? {
2882 return plan_projection(column);
2883 }
2884
2885 match group_expr {
2889 Expr::Identifier(names) => match plan_identifier(ecx, names) {
2890 Err(PlanError::UnknownColumn {
2891 table: None,
2892 column,
2893 similar,
2894 }) => {
2895 let mut iter = projection.iter().map(|(_expr, name)| name);
2898 if let Some(i) = iter.position(|n| *n == column) {
2899 if iter.any(|n| *n == column) {
2900 Err(PlanError::AmbiguousColumn(column))
2901 } else {
2902 plan_projection(i)
2903 }
2904 } else {
2905 Err(PlanError::UnknownColumn {
2908 table: None,
2909 column,
2910 similar,
2911 })
2912 }
2913 }
2914 res => Ok((Some(group_expr), res?)),
2915 },
2916 _ => Ok((
2917 Some(group_expr),
2918 plan_expr(ecx, group_expr)?.type_as_any(ecx)?,
2919 )),
2920 }
2921}
2922
2923pub(crate) fn plan_order_by_exprs(
2931 ecx: &ExprContext,
2932 order_by_exprs: &[OrderByExpr<Aug>],
2933 output_columns: &[(usize, &ColumnName)],
2934) -> Result<(Vec<ColumnOrder>, Vec<HirScalarExpr>), PlanError> {
2935 let mut order_by = vec![];
2936 let mut map_exprs = vec![];
2937 for obe in order_by_exprs {
2938 let expr = plan_order_by_or_distinct_expr(ecx, &obe.expr, output_columns)?;
2939 let column = match expr {
2942 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
2943 _ => {
2944 map_exprs.push(expr);
2945 ecx.relation_type.arity() + map_exprs.len() - 1
2946 }
2947 };
2948 order_by.push(resolve_desc_and_nulls_last(obe, column));
2949 }
2950 Ok((order_by, map_exprs))
2951}
2952
2953fn plan_order_by_or_distinct_expr(
2971 ecx: &ExprContext,
2972 expr: &Expr<Aug>,
2973 output_columns: &[(usize, &ColumnName)],
2974) -> Result<HirScalarExpr, PlanError> {
2975 if let Some(i) = check_col_index(ecx.name, expr, output_columns.len())? {
2976 return Ok(HirScalarExpr::column(output_columns[i].0));
2977 }
2978
2979 if let Expr::Identifier(names) = expr {
2980 if let [name] = &names[..] {
2981 let name = normalize::column_name(name.clone());
2982 let mut iter = output_columns.iter().filter(|(_, n)| **n == name);
2983 if let Some((i, _)) = iter.next() {
2984 match iter.next() {
2985 Some((i2, _)) if i != i2 => return Err(PlanError::AmbiguousColumn(name)),
2989 _ => return Ok(HirScalarExpr::column(*i)),
2990 }
2991 }
2992 }
2993 }
2994
2995 plan_expr(ecx, expr)?.type_as_any(ecx)
2996}
2997
2998fn plan_table_with_joins(
2999 qcx: &QueryContext,
3000 table_with_joins: &TableWithJoins<Aug>,
3001) -> Result<(HirRelationExpr, Scope), PlanError> {
3002 let (mut expr, mut scope) = plan_table_factor(qcx, &table_with_joins.relation)?;
3003 for join in &table_with_joins.joins {
3004 let (new_expr, new_scope) = plan_join(qcx, expr, scope, join)?;
3005 expr = new_expr;
3006 scope = new_scope;
3007 }
3008 Ok((expr, scope))
3009}
3010
3011fn plan_table_factor(
3012 qcx: &QueryContext,
3013 table_factor: &TableFactor<Aug>,
3014) -> Result<(HirRelationExpr, Scope), PlanError> {
3015 match table_factor {
3016 TableFactor::Table { name, alias } => {
3017 let (expr, scope) = qcx.resolve_table_name(name.clone())?;
3018 let scope = plan_table_alias(scope, alias.as_ref())?;
3019 Ok((expr, scope))
3020 }
3021
3022 TableFactor::Function {
3023 function,
3024 alias,
3025 with_ordinality,
3026 } => plan_solitary_table_function(qcx, function, alias.as_ref(), *with_ordinality),
3027
3028 TableFactor::RowsFrom {
3029 functions,
3030 alias,
3031 with_ordinality,
3032 } => plan_rows_from(qcx, functions, alias.as_ref(), *with_ordinality),
3033
3034 TableFactor::Derived {
3035 lateral,
3036 subquery,
3037 alias,
3038 } => {
3039 let mut qcx = (*qcx).clone();
3040 if !lateral {
3041 for scope in &mut qcx.outer_scopes {
3045 if scope.lateral_barrier {
3046 break;
3047 }
3048 scope.items.clear();
3049 }
3050 }
3051 qcx.outer_scopes[0].lateral_barrier = true;
3052 let (expr, scope) = plan_nested_query(&mut qcx, subquery)?;
3053 let scope = plan_table_alias(scope, alias.as_ref())?;
3054 Ok((expr, scope))
3055 }
3056
3057 TableFactor::NestedJoin { join, alias } => {
3058 let (expr, scope) = plan_table_with_joins(qcx, join)?;
3059 let scope = plan_table_alias(scope, alias.as_ref())?;
3060 Ok((expr, scope))
3061 }
3062 }
3063}
3064
3065fn plan_rows_from(
3107 qcx: &QueryContext,
3108 functions: &[Function<Aug>],
3109 alias: Option<&TableAlias>,
3110 with_ordinality: bool,
3111) -> Result<(HirRelationExpr, Scope), PlanError> {
3112 if functions.iter().any(is_repeat_row) {
3114 bail_unsupported!(format!("{} in ROWS FROM", REPEAT_ROW_NAME));
3118 }
3119
3120 if let [function] = functions {
3123 return plan_solitary_table_function(qcx, function, alias, with_ordinality);
3124 }
3125
3126 let (expr, mut scope, num_cols) = plan_rows_from_internal(
3130 qcx,
3131 functions,
3132 Some(functions[0].name.full_item_name().clone()),
3133 )?;
3134
3135 let mut columns = Vec::new();
3137 let mut offset = 0;
3138 for (idx, cols) in num_cols.into_iter().enumerate() {
3140 for i in 0..cols {
3141 columns.push(offset + i);
3142 }
3143 offset += cols + 1;
3144
3145 scope.items.remove(offset - idx - 1);
3148 }
3149
3150 if with_ordinality {
3153 columns.push(offset);
3154 } else {
3155 scope.items.pop();
3156 }
3157
3158 let expr = expr.project(columns);
3159
3160 let scope = plan_table_alias(scope, alias)?;
3161 Ok((expr, scope))
3162}
3163
3164fn is_repeat_row(f: &Function<Aug>) -> bool {
3165 f.name.full_name_str().as_str() == format!("{}.{}", MZ_CATALOG_SCHEMA, REPEAT_ROW_NAME)
3166}
3167
3168fn plan_rows_from_internal<'a>(
3191 qcx: &QueryContext,
3192 functions: impl IntoIterator<Item = &'a Function<Aug>>,
3193 table_name: Option<FullItemName>,
3194) -> Result<(HirRelationExpr, Scope, Vec<usize>), PlanError> {
3195 let mut functions = functions.into_iter();
3196 let mut num_cols = Vec::new();
3197
3198 let (mut left_expr, mut left_scope) =
3202 plan_table_function_internal(qcx, functions.next().unwrap(), true, table_name.clone())?;
3203 num_cols.push(left_scope.len() - 1);
3204 left_expr = left_expr.map(vec![HirScalarExpr::column(left_scope.len() - 1)]);
3206 left_scope
3207 .items
3208 .push(ScopeItem::from_column_name(ORDINALITY_COL_NAME));
3209
3210 for function in functions {
3211 let qcx = qcx.empty_derived_context();
3213 let (right_expr, mut right_scope) =
3214 plan_table_function_internal(&qcx, function, true, table_name.clone())?;
3215 num_cols.push(right_scope.len() - 1);
3216 let left_col = left_scope.len() - 1;
3217 let right_col = left_scope.len() + right_scope.len() - 1;
3218 let on = HirScalarExpr::call_binary(
3219 HirScalarExpr::column(left_col),
3220 HirScalarExpr::column(right_col),
3221 expr_func::Eq,
3222 );
3223 left_expr = left_expr
3224 .join(right_expr, on, JoinKind::FullOuter)
3225 .map(vec![HirScalarExpr::call_variadic(
3226 Coalesce,
3227 vec![
3228 HirScalarExpr::column(left_col),
3229 HirScalarExpr::column(right_col),
3230 ],
3231 )]);
3232
3233 left_expr = left_expr.project(
3236 (0..left_col) .chain(left_col + 1..right_col + 2) .collect(),
3239 );
3240 right_scope.items.push(left_scope.items.pop().unwrap());
3242
3243 left_scope.items.extend(right_scope.items);
3244 }
3245
3246 Ok((left_expr, left_scope, num_cols))
3247}
3248
3249fn plan_solitary_table_function(
3253 qcx: &QueryContext,
3254 function: &Function<Aug>,
3255 alias: Option<&TableAlias>,
3256 with_ordinality: bool,
3257) -> Result<(HirRelationExpr, Scope), PlanError> {
3258 let (expr, mut scope) = plan_table_function_internal(qcx, function, with_ordinality, None)?;
3259
3260 let single_column_function = scope.len() == 1 + if with_ordinality { 1 } else { 0 };
3261 if single_column_function {
3262 let item = &mut scope.items[0];
3263
3264 item.from_single_column_function = true;
3267
3268 if let Some(alias) = alias {
3283 if let ScopeItem {
3284 table_name: Some(table_name),
3285 column_name,
3286 ..
3287 } = item
3288 {
3289 if table_name.item.as_str() == column_name.as_str() {
3290 *column_name = normalize::column_name(alias.name.clone());
3291 }
3292 }
3293 }
3294 }
3295
3296 let scope = plan_table_alias(scope, alias)?;
3297 Ok((expr, scope))
3298}
3299
3300fn plan_table_function_internal(
3305 qcx: &QueryContext,
3306 Function {
3307 name,
3308 args,
3309 filter,
3310 over,
3311 distinct,
3312 }: &Function<Aug>,
3313 with_ordinality: bool,
3314 table_name: Option<FullItemName>,
3315) -> Result<(HirRelationExpr, Scope), PlanError> {
3316 if filter.is_some() {
3321 sql_bail!("FILTER is not allowed for table functions in FROM");
3322 }
3323 if over.is_some() {
3324 sql_bail!("OVER is not allowed for table functions in FROM");
3325 }
3326 if *distinct {
3327 sql_bail!("DISTINCT is not allowed for table functions in FROM");
3328 }
3329
3330 let ecx = &ExprContext {
3331 qcx,
3332 name: "table function arguments",
3333 scope: &Scope::empty(),
3334 relation_type: &SqlRelationType::empty(),
3335 allow_aggregates: false,
3336 allow_subqueries: true,
3337 allow_parameters: true,
3338 allow_windows: false,
3339 };
3340
3341 let scalar_args = match args {
3342 FunctionArgs::Star => sql_bail!("{} does not accept * as an argument", name),
3343 FunctionArgs::Args { args, order_by } => {
3344 if !order_by.is_empty() {
3345 sql_bail!(
3346 "ORDER BY specified, but {} is not an aggregate function",
3347 name
3348 );
3349 }
3350 plan_exprs(ecx, args)?
3351 }
3352 };
3353
3354 let table_name = match table_name {
3355 Some(table_name) => table_name.item,
3356 None => name.full_item_name().item.clone(),
3357 };
3358
3359 let scope_name = Some(PartialItemName {
3360 database: None,
3361 schema: None,
3362 item: table_name,
3363 });
3364
3365 let (expr, mut scope) = match resolve_func(ecx, name, args)? {
3366 Func::Table(impls) => {
3367 let tf = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3368 let scope = Scope::from_source(scope_name.clone(), tf.column_names);
3369 let expr = match tf.imp {
3370 TableFuncImpl::CallTable { mut func, exprs } => {
3371 if with_ordinality {
3372 func = TableFunc::with_ordinality(func.clone()).ok_or(
3373 PlanError::Unsupported {
3374 feature: format!("WITH ORDINALITY on {}", func),
3375 discussion_no: None,
3376 },
3377 )?;
3378 }
3379 HirRelationExpr::CallTable { func, exprs }
3380 }
3381 TableFuncImpl::Expr(expr) => {
3382 if !with_ordinality {
3383 expr
3384 } else {
3385 if qcx
3389 .scx
3390 .is_feature_flag_enabled(&ENABLE_WITH_ORDINALITY_LEGACY_FALLBACK)
3391 {
3392 tracing::error!(
3396 %name,
3397 "Using the legacy WITH ORDINALITY / ROWS FROM implementation for a table function",
3398 );
3399 expr.map(vec![HirScalarExpr::windowing(WindowExpr {
3400 func: WindowExprType::Scalar(ScalarWindowExpr {
3401 func: ScalarWindowFunc::RowNumber,
3402 order_by: vec![],
3403 }),
3404 partition_by: vec![],
3405 order_by: vec![],
3406 })])
3407 } else {
3408 bail_unsupported!(format!(
3409 "WITH ORDINALITY or ROWS FROM with {}",
3410 name
3411 ));
3412 }
3413 }
3414 }
3415 };
3416 (expr, scope)
3417 }
3418 Func::Scalar(impls) => {
3419 let expr = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3420 let output = expr.typ(
3421 &qcx.outer_relation_types,
3422 &SqlRelationType::new(vec![]),
3423 &qcx.scx.param_types.borrow(),
3424 );
3425
3426 let relation = SqlRelationType::new(vec![output]);
3427
3428 let function_ident = Ident::new(name.full_item_name().item.clone())?;
3429 let column_name = normalize::column_name(function_ident);
3430 let name = column_name.to_string();
3431
3432 let scope = Scope::from_source(scope_name.clone(), vec![column_name]);
3433
3434 let mut func = TableFunc::TabletizedScalar { relation, name };
3435 if with_ordinality {
3436 func = TableFunc::with_ordinality(func.clone()).ok_or(PlanError::Unsupported {
3437 feature: format!("WITH ORDINALITY on {}", func),
3438 discussion_no: None,
3439 })?;
3440 }
3441 (
3442 HirRelationExpr::CallTable {
3443 func,
3444 exprs: vec![expr],
3445 },
3446 scope,
3447 )
3448 }
3449 o => sql_bail!(
3450 "{} functions are not supported in functions in FROM",
3451 o.class()
3452 ),
3453 };
3454
3455 if with_ordinality {
3456 scope
3457 .items
3458 .push(ScopeItem::from_name(scope_name, "ordinality"));
3459 }
3460
3461 Ok((expr, scope))
3462}
3463
3464fn plan_table_alias(mut scope: Scope, alias: Option<&TableAlias>) -> Result<Scope, PlanError> {
3465 if let Some(TableAlias {
3466 name,
3467 columns,
3468 strict,
3469 }) = alias
3470 {
3471 if (columns.len() > scope.items.len()) || (*strict && columns.len() != scope.items.len()) {
3472 sql_bail!(
3473 "{} has {} columns available but {} columns specified",
3474 name,
3475 scope.items.len(),
3476 columns.len()
3477 );
3478 }
3479
3480 let table_name = normalize::ident(name.to_owned());
3481 for (i, item) in scope.items.iter_mut().enumerate() {
3482 item.table_name = if item.allow_unqualified_references {
3483 Some(PartialItemName {
3484 database: None,
3485 schema: None,
3486 item: table_name.clone(),
3487 })
3488 } else {
3489 None
3523 };
3524 item.column_name = columns
3525 .get(i)
3526 .map(|a| normalize::column_name(a.clone()))
3527 .unwrap_or_else(|| item.column_name.clone());
3528 }
3529 }
3530 Ok(scope)
3531}
3532
3533fn invent_column_name(
3537 ecx: &ExprContext,
3538 expr: &Expr<Aug>,
3539 table_func_names: &BTreeMap<String, Ident>,
3540) -> Result<Option<ColumnName>, PlanError> {
3541 #[derive(Debug)]
3548 enum NameQuality {
3549 Low,
3550 High,
3551 }
3552
3553 fn invent(
3554 ecx: &ExprContext,
3555 expr: &Expr<Aug>,
3556 table_func_names: &BTreeMap<String, Ident>,
3557 ) -> Result<Option<(ColumnName, NameQuality)>, PlanError> {
3558 Ok(match expr {
3559 Expr::Identifier(names) => {
3560 if let [name] = names.as_slice() {
3561 if let Some(table_func_name) = table_func_names.get(name.as_str()) {
3562 return Ok(Some((
3563 normalize::column_name(table_func_name.clone()),
3564 NameQuality::High,
3565 )));
3566 }
3567 }
3568 names
3569 .last()
3570 .map(|n| (normalize::column_name(n.clone()), NameQuality::High))
3571 }
3572 Expr::Value(v) => match v {
3573 Value::Boolean(_) => Some(("bool".into(), NameQuality::High)),
3576 Value::Interval(_) => Some(("interval".into(), NameQuality::High)),
3577 _ => None,
3578 },
3579 Expr::Function(func) => {
3580 let (schema, item) = match &func.name {
3581 ResolvedItemName::Item {
3582 qualifiers,
3583 full_name,
3584 ..
3585 } => (&qualifiers.schema_spec, full_name.item.clone()),
3586 _ => {
3589 bail_internal!("function name did not resolve to an item: {:?}", func.name)
3590 }
3591 };
3592
3593 if schema == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_internal_schema_id())
3594 || schema
3595 == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_unsafe_schema_id())
3596 {
3597 None
3598 } else {
3599 Some((item.into(), NameQuality::High))
3600 }
3601 }
3602 Expr::HomogenizingFunction { function, .. } => Some((
3603 function.to_string().to_lowercase().into(),
3604 NameQuality::High,
3605 )),
3606 Expr::NullIf { .. } => Some(("nullif".into(), NameQuality::High)),
3607 Expr::Array { .. } => Some(("array".into(), NameQuality::High)),
3608 Expr::List { .. } => Some(("list".into(), NameQuality::High)),
3609 Expr::Map { .. } | Expr::MapSubquery(_) => Some(("map".into(), NameQuality::High)),
3610 Expr::Cast { expr, data_type } => match invent(ecx, expr, table_func_names)? {
3611 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3612 _ => Some((data_type.unqualified_item_name().into(), NameQuality::Low)),
3613 },
3614 Expr::Case { else_result, .. } => {
3615 let inner = match else_result.as_ref() {
3616 Some(else_result) => invent(ecx, else_result, table_func_names)?,
3617 None => None,
3618 };
3619 match inner {
3620 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3621 _ => Some(("case".into(), NameQuality::Low)),
3622 }
3623 }
3624 Expr::FieldAccess { field, .. } => {
3625 Some((normalize::column_name(field.clone()), NameQuality::High))
3626 }
3627 Expr::Exists { .. } => Some(("exists".into(), NameQuality::High)),
3628 Expr::Subscript { expr, .. } => invent(ecx, expr, table_func_names)?,
3629 Expr::Subquery(query) | Expr::ListSubquery(query) | Expr::ArraySubquery(query) => {
3630 let Ok((_expr, scope)) = plan_nested_query(&mut ecx.derived_query_context(), query)
3638 else {
3639 return Ok(None);
3640 };
3641 scope
3642 .items
3643 .first()
3644 .map(|name| (name.column_name.clone(), NameQuality::High))
3645 }
3646 Expr::Row { .. } => Some(("row".into(), NameQuality::High)),
3647 _ => None,
3648 })
3649 }
3650
3651 Ok(invent(ecx, expr, table_func_names)?.map(|(name, _quality)| name))
3652}
3653
3654#[derive(Debug)]
3655enum ExpandedSelectItem<'a> {
3656 InputOrdinal(usize),
3657 Expr(Cow<'a, Expr<Aug>>),
3658}
3659
3660impl ExpandedSelectItem<'_> {
3661 fn as_expr(&self) -> Option<&Expr<Aug>> {
3662 match self {
3663 ExpandedSelectItem::InputOrdinal(_) => None,
3664 ExpandedSelectItem::Expr(expr) => Some(expr),
3665 }
3666 }
3667}
3668
3669fn expand_select_item<'a>(
3670 ecx: &ExprContext,
3671 s: &'a SelectItem<Aug>,
3672 table_func_names: &BTreeMap<String, Ident>,
3673) -> Result<Vec<(ExpandedSelectItem<'a>, ColumnName)>, PlanError> {
3674 match s {
3675 SelectItem::Expr {
3676 expr: Expr::QualifiedWildcard(table_name),
3677 alias: _,
3678 } => {
3679 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3680 let table_name =
3681 normalize::unresolved_item_name(UnresolvedItemName(table_name.clone()))?;
3682 let out: Vec<_> = ecx
3683 .scope
3684 .items
3685 .iter()
3686 .enumerate()
3687 .filter(|(_i, item)| item.is_from_table(&table_name))
3688 .map(|(i, item)| {
3689 let name = item.column_name.clone();
3690 (ExpandedSelectItem::InputOrdinal(i), name)
3691 })
3692 .collect();
3693 if out.is_empty() {
3694 sql_bail!("no table named '{}' in scope", table_name);
3695 }
3696 Ok(out)
3697 }
3698 SelectItem::Expr {
3699 expr: Expr::WildcardAccess(sql_expr),
3700 alias: _,
3701 } => {
3702 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3703 let expr = plan_expr(ecx, sql_expr)?.type_as_any(ecx)?;
3709 let fields = match ecx.scalar_type(&expr) {
3710 SqlScalarType::Record { fields, .. } => fields,
3711 ty => sql_bail!(
3712 "type {} is not composite",
3713 ecx.humanize_sql_scalar_type(&ty, false)
3714 ),
3715 };
3716 let mut skip_cols: BTreeSet<ColumnName> = BTreeSet::new();
3717 if let Expr::Identifier(ident) = sql_expr.as_ref() {
3718 if let [name] = ident.as_slice() {
3719 if let Ok(items) = ecx.scope.items_from_table(
3720 &[],
3721 &PartialItemName {
3722 database: None,
3723 schema: None,
3724 item: name.as_str().to_string(),
3725 },
3726 ) {
3727 for (_, item) in items {
3728 if item
3729 .is_exists_column_for_a_table_function_that_was_in_the_target_list
3730 {
3731 skip_cols.insert(item.column_name.clone());
3732 }
3733 }
3734 }
3735 }
3736 }
3737 let items = fields
3738 .iter()
3739 .filter_map(|(name, _ty)| {
3740 if skip_cols.contains(name) {
3741 None
3742 } else {
3743 let item = ExpandedSelectItem::Expr(Cow::Owned(Expr::FieldAccess {
3744 expr: sql_expr.clone(),
3745 field: name.clone().into(),
3746 }));
3747 Some((item, name.clone()))
3748 }
3749 })
3750 .collect();
3751 Ok(items)
3752 }
3753 SelectItem::Wildcard => {
3754 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3755 let items: Vec<_> = ecx
3756 .scope
3757 .items
3758 .iter()
3759 .enumerate()
3760 .filter(|(_i, item)| item.allow_unqualified_references)
3761 .map(|(i, item)| {
3762 let name = item.column_name.clone();
3763 (ExpandedSelectItem::InputOrdinal(i), name)
3764 })
3765 .collect();
3766
3767 Ok(items)
3768 }
3769 SelectItem::Expr { expr, alias } => {
3770 let name = match alias.clone().map(normalize::column_name) {
3771 Some(name) => name,
3772 None => invent_column_name(ecx, expr, table_func_names)?
3773 .unwrap_or_else(|| UNKNOWN_COLUMN_NAME.into()),
3774 };
3775 Ok(vec![(ExpandedSelectItem::Expr(Cow::Borrowed(expr)), name)])
3776 }
3777 }
3778}
3779
3780fn plan_join(
3781 left_qcx: &QueryContext,
3782 left: HirRelationExpr,
3783 left_scope: Scope,
3784 join: &Join<Aug>,
3785) -> Result<(HirRelationExpr, Scope), PlanError> {
3786 const ON_TRUE: JoinConstraint<Aug> = JoinConstraint::On(Expr::Value(Value::Boolean(true)));
3787 let (kind, constraint) = match &join.join_operator {
3788 JoinOperator::CrossJoin => (JoinKind::Inner, &ON_TRUE),
3789 JoinOperator::Inner(constraint) => (JoinKind::Inner, constraint),
3790 JoinOperator::LeftOuter(constraint) => (JoinKind::LeftOuter, constraint),
3791 JoinOperator::RightOuter(constraint) => (JoinKind::RightOuter, constraint),
3792 JoinOperator::FullOuter(constraint) => (JoinKind::FullOuter, constraint),
3793 };
3794
3795 let mut right_qcx = left_qcx.derived_context(left_scope.clone(), left_qcx.relation_type(&left));
3796 if !kind.can_be_correlated() {
3797 for item in &mut right_qcx.outer_scopes[0].items {
3798 item.error_if_referenced =
3803 Some(|table, column| PlanError::WrongJoinTypeForLateralColumn {
3804 table: table.cloned(),
3805 column: column.clone(),
3806 });
3807 }
3808 }
3809 let (right, right_scope) = plan_table_factor(&right_qcx, &join.relation)?;
3810
3811 let (expr, scope) = match constraint {
3812 JoinConstraint::On(expr) => {
3813 let product_scope = left_scope.product(right_scope)?;
3814 let ecx = &ExprContext {
3815 qcx: left_qcx,
3816 name: "ON clause",
3817 scope: &product_scope,
3818 relation_type: &SqlRelationType::new(
3819 left_qcx
3820 .relation_type(&left)
3821 .column_types
3822 .into_iter()
3823 .chain(right_qcx.relation_type(&right).column_types)
3824 .collect(),
3825 ),
3826 allow_aggregates: false,
3827 allow_subqueries: true,
3828 allow_parameters: true,
3829 allow_windows: false,
3830 };
3831 let on = plan_expr(ecx, expr)?.type_as(ecx, &SqlScalarType::Bool)?;
3832 let joined = left.join(right, on, kind);
3833 (joined, product_scope)
3834 }
3835 JoinConstraint::Using { columns, alias } => {
3836 let column_names = columns
3837 .iter()
3838 .map(|ident| normalize::column_name(ident.clone()))
3839 .collect::<Vec<_>>();
3840
3841 plan_using_constraint(
3842 &column_names,
3843 left_qcx,
3844 left,
3845 left_scope,
3846 &right_qcx,
3847 right,
3848 right_scope,
3849 kind,
3850 alias.as_ref(),
3851 )?
3852 }
3853 JoinConstraint::Natural => {
3854 *left_qcx.scx.ambiguous_columns.borrow_mut() = true;
3857 *right_qcx.scx.ambiguous_columns.borrow_mut() = true;
3858 let left_column_names = left_scope.column_names();
3859 let right_column_names: BTreeSet<_> = right_scope.column_names().collect();
3860 let column_names: Vec<_> = left_column_names
3861 .filter(|col| right_column_names.contains(col))
3862 .cloned()
3863 .collect();
3864 plan_using_constraint(
3865 &column_names,
3866 left_qcx,
3867 left,
3868 left_scope,
3869 &right_qcx,
3870 right,
3871 right_scope,
3872 kind,
3873 None,
3874 )?
3875 }
3876 };
3877 Ok((expr, scope))
3878}
3879
3880#[allow(clippy::too_many_arguments)]
3882fn plan_using_constraint(
3883 column_names: &[ColumnName],
3884 left_qcx: &QueryContext,
3885 left: HirRelationExpr,
3886 left_scope: Scope,
3887 right_qcx: &QueryContext,
3888 right: HirRelationExpr,
3889 right_scope: Scope,
3890 kind: JoinKind,
3891 alias: Option<&Ident>,
3892) -> Result<(HirRelationExpr, Scope), PlanError> {
3893 let mut both_scope = left_scope.clone().product(right_scope.clone())?;
3894
3895 let mut unique_column_names = BTreeSet::new();
3898 for c in column_names {
3899 if !unique_column_names.insert(c) {
3900 return Err(PlanError::Unsupported {
3901 feature: format!(
3902 "column name {} appears more than once in USING clause",
3903 c.quoted()
3904 ),
3905 discussion_no: None,
3906 });
3907 }
3908 }
3909
3910 let alias_item_name = alias.map(|alias| PartialItemName {
3911 database: None,
3912 schema: None,
3913 item: alias.clone().to_string(),
3914 });
3915
3916 if let Some(alias_item_name) = &alias_item_name {
3917 for partial_item_name in both_scope.table_names() {
3918 if partial_item_name.matches(alias_item_name) {
3919 sql_bail!(
3920 "table name \"{}\" specified more than once",
3921 alias_item_name
3922 )
3923 }
3924 }
3925 }
3926
3927 let ecx = &ExprContext {
3928 qcx: right_qcx,
3929 name: "USING clause",
3930 scope: &both_scope,
3931 relation_type: &SqlRelationType::new(
3932 left_qcx
3933 .relation_type(&left)
3934 .column_types
3935 .into_iter()
3936 .chain(right_qcx.relation_type(&right).column_types)
3937 .collect(),
3938 ),
3939 allow_aggregates: false,
3940 allow_subqueries: false,
3941 allow_parameters: false,
3942 allow_windows: false,
3943 };
3944
3945 let mut join_exprs = vec![];
3946 let mut map_exprs = vec![];
3947 let mut new_items = vec![];
3948 let mut join_cols = vec![];
3949 let mut hidden_cols = vec![];
3950
3951 for column_name in column_names {
3952 let (lhs, lhs_name) = left_scope.resolve_using_column(
3954 column_name,
3955 JoinSide::Left,
3956 &mut left_qcx.name_manager.borrow_mut(),
3957 )?;
3958 let (mut rhs, rhs_name) = right_scope.resolve_using_column(
3959 column_name,
3960 JoinSide::Right,
3961 &mut right_qcx.name_manager.borrow_mut(),
3962 )?;
3963
3964 rhs.column += left_scope.len();
3966
3967 let mut exprs = coerce_homogeneous_exprs(
3969 &ecx.with_name(&format!(
3970 "NATURAL/USING join column {}",
3971 column_name.quoted()
3972 )),
3973 vec![
3974 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
3975 lhs,
3976 Arc::clone(&lhs_name),
3977 )),
3978 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
3979 rhs,
3980 Arc::clone(&rhs_name),
3981 )),
3982 ],
3983 None,
3984 )?;
3985 let (expr1, expr2) = (exprs.remove(0), exprs.remove(0));
3986
3987 match kind {
3988 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
3989 join_cols.push(lhs.column);
3990 hidden_cols.push(rhs.column);
3991 }
3992 JoinKind::RightOuter => {
3993 join_cols.push(rhs.column);
3994 hidden_cols.push(lhs.column);
3995 }
3996 JoinKind::FullOuter => {
3997 join_cols.push(both_scope.items.len() + map_exprs.len());
4000 hidden_cols.push(lhs.column);
4001 hidden_cols.push(rhs.column);
4002 map_exprs.push(HirScalarExpr::call_variadic(
4003 Coalesce,
4004 vec![expr1.clone(), expr2.clone()],
4005 ));
4006 new_items.push(ScopeItem::from_column_name(column_name));
4007 }
4008 }
4009
4010 if alias_item_name.is_some() {
4015 let new_item_col = both_scope.items.len() + new_items.len();
4016 join_cols.push(new_item_col);
4017 hidden_cols.push(new_item_col);
4018
4019 new_items.push(ScopeItem::from_name(
4020 alias_item_name.clone(),
4021 column_name.clone().to_string(),
4022 ));
4023
4024 let alias_expr = match kind {
4031 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
4032 HirScalarExpr::named_column(lhs, Arc::clone(&lhs_name))
4033 }
4034 JoinKind::RightOuter => HirScalarExpr::named_column(rhs, Arc::clone(&rhs_name)),
4035 JoinKind::FullOuter => {
4036 HirScalarExpr::call_variadic(Coalesce, vec![expr1.clone(), expr2.clone()])
4037 }
4038 };
4039 map_exprs.push(alias_expr);
4040 }
4041
4042 join_exprs.push(expr1.call_binary(expr2, expr_func::Eq));
4043 }
4044 both_scope.items.extend(new_items);
4045
4046 for c in hidden_cols {
4050 both_scope.items[c].allow_unqualified_references = false;
4051 }
4052
4053 let project_key = join_cols
4055 .into_iter()
4056 .chain(0..both_scope.items.len())
4057 .unique()
4058 .collect::<Vec<_>>();
4059
4060 both_scope = both_scope.project(&project_key);
4061
4062 let on = HirScalarExpr::variadic_and(join_exprs);
4063
4064 let both = left
4065 .join(right, on, kind)
4066 .map(map_exprs)
4067 .project(project_key);
4068 Ok((both, both_scope))
4069}
4070
4071pub fn plan_expr<'a>(
4072 ecx: &'a ExprContext,
4073 e: &Expr<Aug>,
4074) -> Result<CoercibleScalarExpr, PlanError> {
4075 ecx.checked_recur(|ecx| plan_expr_inner(ecx, e))
4076}
4077
4078fn plan_expr_inner<'a>(
4079 ecx: &'a ExprContext,
4080 e: &Expr<Aug>,
4081) -> Result<CoercibleScalarExpr, PlanError> {
4082 if let Some((i, item)) = ecx.scope.resolve_expr(e) {
4083 return Ok(HirScalarExpr::named_column(
4085 i,
4086 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
4087 )
4088 .into());
4089 }
4090
4091 match e {
4092 Expr::Identifier(names) | Expr::QualifiedWildcard(names) => {
4094 Ok(plan_identifier(ecx, names)?.into())
4095 }
4096
4097 Expr::Value(val) => plan_literal(val),
4099 Expr::Parameter(n) => plan_parameter(ecx, *n),
4100 Expr::Array(exprs) => plan_array(ecx, exprs, None),
4101 Expr::List(exprs) => plan_list(ecx, exprs, None),
4102 Expr::Map(exprs) => plan_map(ecx, exprs, None),
4103 Expr::Row { exprs } => plan_row(ecx, exprs),
4104
4105 Expr::Op { op, expr1, expr2 } => {
4107 Ok(plan_op(ecx, normalize::op(op)?, expr1, expr2.as_deref())?.into())
4108 }
4109 Expr::Cast { expr, data_type } => plan_cast(ecx, expr, data_type),
4110 Expr::Function(func) => Ok(plan_function(ecx, func)?.into()),
4111
4112 Expr::Not { expr } => plan_not(ecx, expr),
4114 Expr::And { left, right } => plan_and(ecx, left, right),
4115 Expr::Or { left, right } => plan_or(ecx, left, right),
4116 Expr::IsExpr {
4117 expr,
4118 construct,
4119 negated,
4120 } => Ok(plan_is_expr(ecx, expr, construct, *negated)?.into()),
4121 Expr::Case {
4122 operand,
4123 conditions,
4124 results,
4125 else_result,
4126 } => Ok(plan_case(ecx, operand, conditions, results, else_result)?.into()),
4127 Expr::HomogenizingFunction { function, exprs } => {
4128 plan_homogenizing_function(ecx, function, exprs)
4129 }
4130 Expr::NullIf { l_expr, r_expr } => Ok(plan_case(
4131 ecx,
4132 &None,
4133 &[l_expr.clone().equals(*r_expr.clone())],
4134 &[Expr::null()],
4135 &Some(Box::new(*l_expr.clone())),
4136 )?
4137 .into()),
4138 Expr::FieldAccess { expr, field } => plan_field_access(ecx, expr, field),
4139 Expr::WildcardAccess(expr) => plan_expr(ecx, expr),
4140 Expr::Subscript { expr, positions } => plan_subscript(ecx, expr, positions),
4141 Expr::Like {
4142 expr,
4143 pattern,
4144 escape,
4145 case_insensitive,
4146 negated,
4147 } => Ok(plan_like(
4148 ecx,
4149 expr,
4150 pattern,
4151 escape.as_deref(),
4152 *case_insensitive,
4153 *negated,
4154 )?
4155 .into()),
4156
4157 Expr::InList {
4158 expr,
4159 list,
4160 negated,
4161 } => plan_in_list(ecx, expr, list, negated),
4162
4163 Expr::Exists(query) => plan_exists(ecx, query),
4165 Expr::Subquery(query) => plan_subquery(ecx, query),
4166 Expr::ListSubquery(query) => plan_list_subquery(ecx, query),
4167 Expr::MapSubquery(query) => plan_map_subquery(ecx, query),
4168 Expr::ArraySubquery(query) => plan_array_subquery(ecx, query),
4169 Expr::Collate { expr, collation } => plan_collate(ecx, expr, collation),
4170 Expr::Nested(_) => bail_internal!("Expr::Nested should have been desugared"),
4171 Expr::InSubquery { .. } => {
4172 bail_internal!("Expr::InSubquery should have been desugared")
4173 }
4174 Expr::AnyExpr { .. } => {
4175 bail_internal!("Expr::AnyExpr should have been desugared")
4176 }
4177 Expr::AllExpr { .. } => {
4178 bail_internal!("Expr::AllExpr should have been desugared")
4179 }
4180 Expr::AnySubquery { .. } => {
4181 bail_internal!("Expr::AnySubquery should have been desugared")
4182 }
4183 Expr::AllSubquery { .. } => {
4184 bail_internal!("Expr::AllSubquery should have been desugared")
4185 }
4186 Expr::Between { .. } => {
4187 bail_internal!("Expr::Between should have been desugared")
4188 }
4189 }
4190}
4191
4192fn plan_parameter(ecx: &ExprContext, n: usize) -> Result<CoercibleScalarExpr, PlanError> {
4193 if !ecx.allow_parameters {
4194 return Err(PlanError::UnknownParameter(n));
4198 }
4199 if n == 0 || n > 65536 {
4200 return Err(PlanError::UnknownParameter(n));
4201 }
4202 if ecx.param_types().borrow().contains_key(&n) {
4203 Ok(HirScalarExpr::parameter(n).into())
4204 } else {
4205 Ok(CoercibleScalarExpr::Parameter(n))
4206 }
4207}
4208
4209fn plan_row(ecx: &ExprContext, exprs: &[Expr<Aug>]) -> Result<CoercibleScalarExpr, PlanError> {
4210 let mut out = vec![];
4211 for e in exprs {
4212 out.push(plan_expr(ecx, e)?);
4213 }
4214 Ok(CoercibleScalarExpr::LiteralRecord(out))
4215}
4216
4217fn plan_cast(
4218 ecx: &ExprContext,
4219 expr: &Expr<Aug>,
4220 data_type: &ResolvedDataType,
4221) -> Result<CoercibleScalarExpr, PlanError> {
4222 let to_scalar_type = scalar_type_from_sql(ecx.qcx.scx, data_type)?;
4223 let expr = match expr {
4224 Expr::Array(exprs) => plan_array(ecx, exprs, Some(&to_scalar_type))?,
4233 Expr::List(exprs) => plan_list(ecx, exprs, Some(&to_scalar_type))?,
4234 Expr::Map(exprs) => plan_map(ecx, exprs, Some(&to_scalar_type))?,
4235 _ => plan_expr(ecx, expr)?,
4236 };
4237 let ecx = &ecx.with_name("CAST");
4238 let expr = typeconv::plan_coerce(ecx, expr, &to_scalar_type)?;
4239 let expr = typeconv::plan_cast(ecx, CastContext::Explicit, expr, &to_scalar_type)?;
4240 Ok(expr.into())
4241}
4242
4243fn plan_not(ecx: &ExprContext, expr: &Expr<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4244 let ecx = ecx.with_name("NOT argument");
4245 Ok(plan_expr(&ecx, expr)?
4246 .type_as(&ecx, &SqlScalarType::Bool)?
4247 .call_unary(UnaryFunc::Not(expr_func::Not))
4248 .into())
4249}
4250
4251fn plan_and(
4252 ecx: &ExprContext,
4253 left: &Expr<Aug>,
4254 right: &Expr<Aug>,
4255) -> Result<CoercibleScalarExpr, PlanError> {
4256 let ecx = ecx.with_name("AND argument");
4257 Ok(HirScalarExpr::variadic_and(vec![
4258 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4259 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4260 ])
4261 .into())
4262}
4263
4264fn plan_or(
4265 ecx: &ExprContext,
4266 left: &Expr<Aug>,
4267 right: &Expr<Aug>,
4268) -> Result<CoercibleScalarExpr, PlanError> {
4269 let ecx = ecx.with_name("OR argument");
4270 Ok(HirScalarExpr::variadic_or(vec![
4271 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4272 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4273 ])
4274 .into())
4275}
4276
4277fn plan_in_list(
4278 ecx: &ExprContext,
4279 lhs: &Expr<Aug>,
4280 list: &Vec<Expr<Aug>>,
4281 negated: &bool,
4282) -> Result<CoercibleScalarExpr, PlanError> {
4283 let ecx = ecx.with_name("IN list");
4284 let or = HirScalarExpr::variadic_or(
4285 list.into_iter()
4286 .map(|e| {
4287 let eq = lhs.clone().equals(e.clone());
4288 plan_expr(&ecx, &eq)?.type_as(&ecx, &SqlScalarType::Bool)
4289 })
4290 .collect::<Result<Vec<HirScalarExpr>, PlanError>>()?,
4291 );
4292 Ok(if *negated {
4293 or.call_unary(UnaryFunc::Not(expr_func::Not))
4294 } else {
4295 or
4296 }
4297 .into())
4298}
4299
4300fn plan_homogenizing_function(
4301 ecx: &ExprContext,
4302 function: &HomogenizingFunction,
4303 exprs: &[Expr<Aug>],
4304) -> Result<CoercibleScalarExpr, PlanError> {
4305 assert!(!exprs.is_empty()); let expr = HirScalarExpr::call_variadic(
4307 match function {
4308 HomogenizingFunction::Coalesce => VariadicFunc::from(Coalesce),
4309 HomogenizingFunction::Greatest => VariadicFunc::from(Greatest),
4310 HomogenizingFunction::Least => VariadicFunc::from(Least),
4311 },
4312 coerce_homogeneous_exprs(
4313 &ecx.with_name(&function.to_string().to_lowercase()),
4314 plan_exprs(ecx, exprs)?,
4315 None,
4316 )?,
4317 );
4318 Ok(expr.into())
4319}
4320
4321fn plan_field_access(
4322 ecx: &ExprContext,
4323 expr: &Expr<Aug>,
4324 field: &Ident,
4325) -> Result<CoercibleScalarExpr, PlanError> {
4326 let field = normalize::column_name(field.clone());
4327 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4328 let ty = ecx.scalar_type(&expr);
4329 let i = match &ty {
4330 SqlScalarType::Record { fields, .. } => {
4331 fields.iter().position(|(name, _ty)| *name == field)
4332 }
4333 ty => sql_bail!(
4334 "column notation applied to type {}, which is not a composite type",
4335 ecx.humanize_sql_scalar_type(ty, false)
4336 ),
4337 };
4338 match i {
4339 None => sql_bail!(
4340 "field {} not found in data type {}",
4341 field,
4342 ecx.humanize_sql_scalar_type(&ty, false)
4343 ),
4344 Some(i) => Ok(expr
4345 .call_unary(UnaryFunc::RecordGet(expr_func::RecordGet(i)))
4346 .into()),
4347 }
4348}
4349
4350fn plan_subscript(
4351 ecx: &ExprContext,
4352 expr: &Expr<Aug>,
4353 positions: &[SubscriptPosition<Aug>],
4354) -> Result<CoercibleScalarExpr, PlanError> {
4355 assert!(
4356 !positions.is_empty(),
4357 "subscript expression must contain at least one position"
4358 );
4359
4360 let ecx = &ecx.with_name("subscripting");
4361 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4362 let ty = ecx.scalar_type(&expr);
4363 match &ty {
4364 SqlScalarType::Array(..) | SqlScalarType::Int2Vector => plan_subscript_array(
4365 ecx,
4366 expr,
4367 positions,
4368 if ty == SqlScalarType::Int2Vector {
4372 1
4373 } else {
4374 0
4375 },
4376 ),
4377 SqlScalarType::Jsonb => plan_subscript_jsonb(ecx, expr, positions),
4378 SqlScalarType::List { element_type, .. } => {
4379 let elem_type_name = ecx.humanize_sql_scalar_type(element_type, false);
4381 let n_layers = ty.unwrap_list_n_layers();
4382 plan_subscript_list(ecx, expr, positions, n_layers, &elem_type_name)
4383 }
4384 ty => sql_bail!(
4385 "cannot subscript type {}",
4386 ecx.humanize_sql_scalar_type(ty, false)
4387 ),
4388 }
4389}
4390
4391fn extract_scalar_subscript_from_positions<'a>(
4395 positions: &'a [SubscriptPosition<Aug>],
4396 expr_type_name: &str,
4397) -> Result<Vec<&'a Expr<Aug>>, PlanError> {
4398 let mut scalar_subscripts = Vec::with_capacity(positions.len());
4399 for p in positions {
4400 if p.explicit_slice {
4401 sql_bail!("{} subscript does not support slices", expr_type_name);
4402 }
4403 assert!(
4404 p.end.is_none(),
4405 "index-appearing subscripts cannot have end value"
4406 );
4407 scalar_subscripts.push(p.start.as_ref().expect("has start if not slice"));
4408 }
4409 Ok(scalar_subscripts)
4410}
4411
4412fn plan_subscript_array(
4413 ecx: &ExprContext,
4414 expr: HirScalarExpr,
4415 positions: &[SubscriptPosition<Aug>],
4416 offset: i64,
4417) -> Result<CoercibleScalarExpr, PlanError> {
4418 let mut exprs = Vec::with_capacity(positions.len() + 1);
4419 exprs.push(expr);
4420
4421 let indexes = extract_scalar_subscript_from_positions(positions, "array")?;
4424
4425 for i in indexes {
4426 exprs.push(plan_expr(ecx, i)?.cast_to(
4427 ecx,
4428 CastContext::Explicit,
4429 &SqlScalarType::Int64,
4430 )?);
4431 }
4432
4433 Ok(HirScalarExpr::call_variadic(ArrayIndex { offset }, exprs).into())
4434}
4435
4436fn plan_subscript_list(
4437 ecx: &ExprContext,
4438 mut expr: HirScalarExpr,
4439 positions: &[SubscriptPosition<Aug>],
4440 mut remaining_layers: usize,
4441 elem_type_name: &str,
4442) -> Result<CoercibleScalarExpr, PlanError> {
4443 let mut i = 0;
4444
4445 while i < positions.len() {
4446 let j = positions[i..]
4448 .iter()
4449 .position(|p| p.explicit_slice)
4450 .unwrap_or(positions.len() - i);
4451 if j != 0 {
4452 let indexes = extract_scalar_subscript_from_positions(&positions[i..i + j], "")?;
4453 let (n, e) = plan_index_list(
4454 ecx,
4455 expr,
4456 indexes.as_slice(),
4457 remaining_layers,
4458 elem_type_name,
4459 )?;
4460 remaining_layers = n;
4461 expr = e;
4462 i += j;
4463 }
4464
4465 let j = positions[i..]
4467 .iter()
4468 .position(|p| !p.explicit_slice)
4469 .unwrap_or(positions.len() - i);
4470 if j != 0 {
4471 expr = plan_slice_list(
4472 ecx,
4473 expr,
4474 &positions[i..i + j],
4475 remaining_layers,
4476 elem_type_name,
4477 )?;
4478 i += j;
4479 }
4480 }
4481
4482 Ok(expr.into())
4483}
4484
4485fn plan_index_list(
4486 ecx: &ExprContext,
4487 expr: HirScalarExpr,
4488 indexes: &[&Expr<Aug>],
4489 n_layers: usize,
4490 elem_type_name: &str,
4491) -> Result<(usize, HirScalarExpr), PlanError> {
4492 let depth = indexes.len();
4493
4494 if depth > n_layers {
4495 if n_layers == 0 {
4496 sql_bail!("cannot subscript type {}", elem_type_name)
4497 } else {
4498 sql_bail!(
4499 "cannot index into {} layers; list only has {} layer{}",
4500 depth,
4501 n_layers,
4502 if n_layers == 1 { "" } else { "s" }
4503 )
4504 }
4505 }
4506
4507 let mut exprs = Vec::with_capacity(depth + 1);
4508 exprs.push(expr);
4509
4510 for i in indexes {
4511 exprs.push(plan_expr(ecx, i)?.cast_to(
4512 ecx,
4513 CastContext::Explicit,
4514 &SqlScalarType::Int64,
4515 )?);
4516 }
4517
4518 Ok((
4519 n_layers - depth,
4520 HirScalarExpr::call_variadic(ListIndex, exprs),
4521 ))
4522}
4523
4524fn plan_slice_list(
4525 ecx: &ExprContext,
4526 expr: HirScalarExpr,
4527 slices: &[SubscriptPosition<Aug>],
4528 n_layers: usize,
4529 elem_type_name: &str,
4530) -> Result<HirScalarExpr, PlanError> {
4531 if n_layers == 0 {
4532 sql_bail!("cannot subscript type {}", elem_type_name)
4533 }
4534
4535 let mut exprs = Vec::with_capacity(slices.len() + 1);
4537 exprs.push(expr);
4538 let extract_position_or_default = |position, default| -> Result<HirScalarExpr, PlanError> {
4540 Ok(match position {
4541 Some(p) => {
4542 plan_expr(ecx, p)?.cast_to(ecx, CastContext::Explicit, &SqlScalarType::Int64)?
4543 }
4544 None => HirScalarExpr::literal(Datum::Int64(default), SqlScalarType::Int64),
4545 })
4546 };
4547 for p in slices {
4548 let start = extract_position_or_default(p.start.as_ref(), 1)?;
4549 let end = extract_position_or_default(p.end.as_ref(), i64::MAX - 1)?;
4550 exprs.push(start);
4551 exprs.push(end);
4552 }
4553
4554 Ok(HirScalarExpr::call_variadic(ListSliceLinear, exprs))
4555}
4556
4557fn plan_like(
4558 ecx: &ExprContext,
4559 expr: &Expr<Aug>,
4560 pattern: &Expr<Aug>,
4561 escape: Option<&Expr<Aug>>,
4562 case_insensitive: bool,
4563 not: bool,
4564) -> Result<HirScalarExpr, PlanError> {
4565 use CastContext::Implicit;
4566 let ecx = ecx.with_name("LIKE argument");
4567 let expr = plan_expr(&ecx, expr)?;
4568 let haystack = match ecx.scalar_type(&expr) {
4569 CoercibleScalarType::Coerced(ref ty @ SqlScalarType::Char { length }) => expr
4570 .type_as(&ecx, ty)?
4571 .call_unary(UnaryFunc::PadChar(expr_func::PadChar { length })),
4572 _ => expr.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4573 };
4574 let mut pattern = plan_expr(&ecx, pattern)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?;
4575 if let Some(escape) = escape {
4576 pattern = pattern.call_binary(
4577 plan_expr(&ecx, escape)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4578 expr_func::LikeEscape,
4579 );
4580 }
4581 let func: BinaryFunc = if case_insensitive {
4582 expr_func::IsLikeMatchCaseInsensitive.into()
4583 } else {
4584 expr_func::IsLikeMatchCaseSensitive.into()
4585 };
4586 let like = haystack.call_binary(pattern, func);
4587 if not {
4588 Ok(like.call_unary(UnaryFunc::Not(expr_func::Not)))
4589 } else {
4590 Ok(like)
4591 }
4592}
4593
4594fn plan_subscript_jsonb(
4595 ecx: &ExprContext,
4596 expr: HirScalarExpr,
4597 positions: &[SubscriptPosition<Aug>],
4598) -> Result<CoercibleScalarExpr, PlanError> {
4599 use CastContext::Implicit;
4600 use SqlScalarType::{Int64, String};
4601
4602 let subscripts = extract_scalar_subscript_from_positions(positions, "jsonb")?;
4605
4606 let mut exprs = Vec::with_capacity(subscripts.len());
4607 for s in subscripts {
4608 let subscript = plan_expr(ecx, s)?;
4609 let subscript = if let Ok(subscript) = subscript.clone().cast_to(ecx, Implicit, &String) {
4610 subscript
4611 } else if let Ok(subscript) = subscript.cast_to(ecx, Implicit, &Int64) {
4612 typeconv::to_string(ecx, subscript)?
4616 } else {
4617 sql_bail!("jsonb subscript type must be coercible to integer or text");
4618 };
4619 exprs.push(subscript);
4620 }
4621
4622 let expr = expr.call_binary(
4625 HirScalarExpr::call_variadic(
4626 ArrayCreate {
4627 elem_type: SqlScalarType::String,
4628 },
4629 exprs,
4630 ),
4631 expr_func::JsonbGetPath,
4632 );
4633 Ok(expr.into())
4634}
4635
4636fn plan_exists(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4637 if !ecx.allow_subqueries {
4638 sql_bail!("{} does not allow subqueries", ecx.name)
4639 }
4640 let mut qcx = ecx.derived_query_context();
4641 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4642 Ok(expr.exists().into())
4643}
4644
4645fn plan_subquery(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4646 if !ecx.allow_subqueries {
4647 sql_bail!("{} does not allow subqueries", ecx.name)
4648 }
4649 let mut qcx = ecx.derived_query_context();
4650 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4651 let column_types = qcx.relation_type(&expr).column_types;
4652 if column_types.len() != 1 {
4653 sql_bail!(
4654 "Expected subselect to return 1 column, got {} columns",
4655 column_types.len()
4656 );
4657 }
4658 Ok(expr.select().into())
4659}
4660
4661fn plan_list_subquery(
4662 ecx: &ExprContext,
4663 query: &Query<Aug>,
4664) -> Result<CoercibleScalarExpr, PlanError> {
4665 plan_vector_like_subquery(
4666 ecx,
4667 query,
4668 |_| false,
4669 |elem_type| ListCreate { elem_type }.into(),
4670 |order_by| AggregateFunc::ListConcat { order_by },
4671 expr_func::ListListConcat.into(),
4672 |elem_type| {
4673 HirScalarExpr::literal(
4674 Datum::empty_list(),
4675 SqlScalarType::List {
4676 element_type: Box::new(elem_type),
4677 custom_id: None,
4678 },
4679 )
4680 },
4681 "list",
4682 )
4683}
4684
4685fn plan_array_subquery(
4686 ecx: &ExprContext,
4687 query: &Query<Aug>,
4688) -> Result<CoercibleScalarExpr, PlanError> {
4689 plan_vector_like_subquery(
4690 ecx,
4691 query,
4692 |elem_type| {
4693 matches!(
4694 elem_type,
4695 SqlScalarType::Char { .. }
4696 | SqlScalarType::Array { .. }
4697 | SqlScalarType::List { .. }
4698 | SqlScalarType::Map { .. }
4699 )
4700 },
4701 |elem_type| ArrayCreate { elem_type }.into(),
4702 |order_by| AggregateFunc::ArrayConcat { order_by },
4703 expr_func::ArrayArrayConcat.into(),
4704 |elem_type| {
4705 HirScalarExpr::literal(
4706 Datum::empty_array(),
4707 SqlScalarType::Array(Box::new(elem_type)),
4708 )
4709 },
4710 "[]",
4711 )
4712}
4713
4714fn plan_vector_like_subquery<F1, F2, F3, F4>(
4716 ecx: &ExprContext,
4717 query: &Query<Aug>,
4718 is_unsupported_type: F1,
4719 vector_create: F2,
4720 aggregate_concat: F3,
4721 binary_concat: BinaryFunc,
4722 empty_literal: F4,
4723 vector_type_string: &str,
4724) -> Result<CoercibleScalarExpr, PlanError>
4725where
4726 F1: Fn(&SqlScalarType) -> bool,
4727 F2: Fn(SqlScalarType) -> VariadicFunc,
4728 F3: Fn(Vec<ColumnOrder>) -> AggregateFunc,
4729 F4: Fn(SqlScalarType) -> HirScalarExpr,
4730{
4731 if !ecx.allow_subqueries {
4732 sql_bail!("{} does not allow subqueries", ecx.name)
4733 }
4734
4735 let mut qcx = ecx.derived_query_context();
4736 let mut planned_query = plan_query(&mut qcx, query)?;
4737 if planned_query.limit.is_some()
4738 || !planned_query
4739 .offset
4740 .clone()
4741 .try_into_literal_int64()
4742 .is_ok_and(|offset| offset == 0)
4743 {
4744 planned_query.expr = HirRelationExpr::top_k(
4745 planned_query.expr,
4746 vec![],
4747 planned_query.order_by.clone(),
4748 planned_query.limit,
4749 planned_query.offset,
4750 planned_query.group_size_hints.limit_input_group_size,
4751 );
4752 }
4753
4754 if planned_query.project.len() != 1 {
4755 sql_bail!(
4756 "Expected subselect to return 1 column, got {} columns",
4757 planned_query.project.len()
4758 );
4759 }
4760
4761 let project_column = *planned_query.project.get(0).unwrap();
4762 let elem_type = qcx
4763 .relation_type(&planned_query.expr)
4764 .column_types
4765 .get(project_column)
4766 .cloned()
4767 .unwrap()
4768 .scalar_type();
4769
4770 if is_unsupported_type(&elem_type) {
4771 bail_unsupported!(format!(
4772 "cannot build array from subquery because return type {}{}",
4773 ecx.humanize_sql_scalar_type(&elem_type, false),
4774 vector_type_string
4775 ));
4776 }
4777
4778 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4781 vector_create(elem_type.clone()),
4782 vec![HirScalarExpr::column(project_column)],
4783 ))
4784 .chain(
4785 planned_query
4786 .order_by
4787 .iter()
4788 .map(|co| HirScalarExpr::column(co.column)),
4789 )
4790 .collect();
4791
4792 let aggregation_projection = vec![0];
4796 let aggregation_order_by = planned_query
4797 .order_by
4798 .into_iter()
4799 .enumerate()
4800 .map(|(i, order)| ColumnOrder { column: i, ..order })
4801 .collect();
4802
4803 let reduced_expr = planned_query
4804 .expr
4805 .reduce(
4806 vec![],
4807 vec![AggregateExpr {
4808 func: aggregate_concat(aggregation_order_by),
4809 expr: Box::new(HirScalarExpr::call_variadic(
4810 RecordCreate {
4811 field_names: iter::repeat(ColumnName::from(""))
4812 .take(aggregation_exprs.len())
4813 .collect(),
4814 },
4815 aggregation_exprs,
4816 )),
4817 distinct: false,
4818 }],
4819 None,
4820 )
4821 .project(aggregation_projection);
4822
4823 Ok(reduced_expr
4825 .select()
4826 .call_binary(empty_literal(elem_type), binary_concat)
4827 .into())
4828}
4829
4830fn plan_map_subquery(
4831 ecx: &ExprContext,
4832 query: &Query<Aug>,
4833) -> Result<CoercibleScalarExpr, PlanError> {
4834 if !ecx.allow_subqueries {
4835 sql_bail!("{} does not allow subqueries", ecx.name)
4836 }
4837
4838 let mut qcx = ecx.derived_query_context();
4839 let mut query = plan_query(&mut qcx, query)?;
4840 if query.limit.is_some()
4841 || !query
4842 .offset
4843 .clone()
4844 .try_into_literal_int64()
4845 .is_ok_and(|offset| offset == 0)
4846 {
4847 query.expr = HirRelationExpr::top_k(
4848 query.expr,
4849 vec![],
4850 query.order_by.clone(),
4851 query.limit,
4852 query.offset,
4853 query.group_size_hints.limit_input_group_size,
4854 );
4855 }
4856 if query.project.len() != 2 {
4857 sql_bail!(
4858 "expected map subquery to return 2 columns, got {} columns",
4859 query.project.len()
4860 );
4861 }
4862
4863 let query_types = qcx.relation_type(&query.expr).column_types;
4864 let key_column = query.project[0];
4865 let key_type = query_types[key_column].clone().scalar_type();
4866 let value_column = query.project[1];
4867 let value_type = query_types[value_column].clone().scalar_type();
4868
4869 if key_type != SqlScalarType::String {
4870 sql_bail!("cannot build map from subquery because first column is not of type text");
4871 }
4872
4873 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4874 RecordCreate {
4875 field_names: vec![ColumnName::from("key"), ColumnName::from("value")],
4876 },
4877 vec![
4878 HirScalarExpr::column(key_column),
4879 HirScalarExpr::column(value_column),
4880 ],
4881 ))
4882 .chain(
4883 query
4884 .order_by
4885 .iter()
4886 .map(|co| HirScalarExpr::column(co.column)),
4887 )
4888 .collect();
4889
4890 let expr = query
4891 .expr
4892 .reduce(
4893 vec![],
4894 vec![AggregateExpr {
4895 func: AggregateFunc::MapAgg {
4896 order_by: query
4897 .order_by
4898 .into_iter()
4899 .enumerate()
4900 .map(|(i, order)| ColumnOrder { column: i, ..order })
4901 .collect(),
4902 value_type: value_type.clone(),
4903 },
4904 expr: Box::new(HirScalarExpr::call_variadic(
4905 RecordCreate {
4906 field_names: iter::repeat(ColumnName::from(""))
4907 .take(aggregation_exprs.len())
4908 .collect(),
4909 },
4910 aggregation_exprs,
4911 )),
4912 distinct: false,
4913 }],
4914 None,
4915 )
4916 .project(vec![0]);
4917
4918 let expr = HirScalarExpr::call_variadic(
4920 Coalesce,
4921 vec![
4922 expr.select(),
4923 HirScalarExpr::literal(
4924 Datum::empty_map(),
4925 SqlScalarType::Map {
4926 value_type: Box::new(value_type),
4927 custom_id: None,
4928 },
4929 ),
4930 ],
4931 );
4932
4933 Ok(expr.into())
4934}
4935
4936fn plan_collate(
4937 ecx: &ExprContext,
4938 expr: &Expr<Aug>,
4939 collation: &UnresolvedItemName,
4940) -> Result<CoercibleScalarExpr, PlanError> {
4941 if collation.0.len() == 2
4942 && collation.0[0] == ident!(mz_repr::namespaces::PG_CATALOG_SCHEMA)
4943 && collation.0[1] == ident!("default")
4944 {
4945 plan_expr(ecx, expr)
4946 } else {
4947 bail_unsupported!("COLLATE");
4948 }
4949}
4950
4951fn plan_exprs<E>(ecx: &ExprContext, exprs: &[E]) -> Result<Vec<CoercibleScalarExpr>, PlanError>
4958where
4959 E: std::borrow::Borrow<Expr<Aug>>,
4960{
4961 let mut out = vec![];
4962 for expr in exprs {
4963 out.push(plan_expr(ecx, expr.borrow())?);
4964 }
4965 Ok(out)
4966}
4967
4968fn plan_array(
4970 ecx: &ExprContext,
4971 exprs: &[Expr<Aug>],
4972 type_hint: Option<&SqlScalarType>,
4973) -> Result<CoercibleScalarExpr, PlanError> {
4974 let mut out = vec![];
4976 for expr in exprs {
4977 out.push(match expr {
4978 Expr::Array(exprs) => plan_array(ecx, exprs, type_hint.clone())?,
4981 _ => plan_expr(ecx, expr)?,
4982 });
4983 }
4984
4985 let type_hint = match type_hint {
4987 Some(SqlScalarType::Array(elem_type)) => {
4992 let multidimensional = out.iter().any(|e| {
4993 matches!(
4994 ecx.scalar_type(e),
4995 CoercibleScalarType::Coerced(SqlScalarType::Array(_))
4996 )
4997 });
4998 if multidimensional {
4999 type_hint
5000 } else {
5001 Some(&**elem_type)
5002 }
5003 }
5004 Some(_) => None,
5008 None => None,
5010 };
5011
5012 let (elem_type, exprs) = if exprs.is_empty() {
5014 if let Some(elem_type) = type_hint {
5015 (elem_type.clone(), vec![])
5016 } else {
5017 sql_bail!("cannot determine type of empty array");
5018 }
5019 } else {
5020 let out = coerce_homogeneous_exprs(&ecx.with_name("ARRAY"), out, type_hint)?;
5021 (ecx.scalar_type(&out[0]), out)
5022 };
5023
5024 if matches!(
5030 elem_type,
5031 SqlScalarType::Char { .. } | SqlScalarType::List { .. } | SqlScalarType::Map { .. }
5032 ) {
5033 bail_unsupported!(format!(
5034 "{}[]",
5035 ecx.humanize_sql_scalar_type(&elem_type, false)
5036 ));
5037 }
5038
5039 Ok(HirScalarExpr::call_variadic(ArrayCreate { elem_type }, exprs).into())
5040}
5041
5042fn plan_list(
5043 ecx: &ExprContext,
5044 exprs: &[Expr<Aug>],
5045 type_hint: Option<&SqlScalarType>,
5046) -> Result<CoercibleScalarExpr, PlanError> {
5047 let (elem_type, exprs) = if exprs.is_empty() {
5048 if let Some(SqlScalarType::List { element_type, .. }) = type_hint {
5049 (element_type.without_modifiers(), vec![])
5050 } else {
5051 sql_bail!("cannot determine type of empty list");
5052 }
5053 } else {
5054 let type_hint = match type_hint {
5055 Some(SqlScalarType::List { element_type, .. }) => Some(&**element_type),
5056 _ => None,
5057 };
5058
5059 let mut out = vec![];
5060 for expr in exprs {
5061 out.push(match expr {
5062 Expr::List(exprs) => plan_list(ecx, exprs, type_hint)?,
5065 _ => plan_expr(ecx, expr)?,
5066 });
5067 }
5068 let out = coerce_homogeneous_exprs(&ecx.with_name("LIST"), out, type_hint)?;
5069 (ecx.scalar_type(&out[0]).without_modifiers(), out)
5070 };
5071
5072 if matches!(elem_type, SqlScalarType::Char { .. }) {
5073 bail_unsupported!("char list");
5074 }
5075
5076 Ok(HirScalarExpr::call_variadic(ListCreate { elem_type }, exprs).into())
5077}
5078
5079fn plan_map(
5080 ecx: &ExprContext,
5081 entries: &[MapEntry<Aug>],
5082 type_hint: Option<&SqlScalarType>,
5083) -> Result<CoercibleScalarExpr, PlanError> {
5084 let (value_type, exprs) = if entries.is_empty() {
5085 if let Some(SqlScalarType::Map { value_type, .. }) = type_hint {
5086 (value_type.without_modifiers(), vec![])
5087 } else {
5088 sql_bail!("cannot determine type of empty map");
5089 }
5090 } else {
5091 let type_hint = match type_hint {
5092 Some(SqlScalarType::Map { value_type, .. }) => Some(&**value_type),
5093 _ => None,
5094 };
5095
5096 let mut keys = vec![];
5097 let mut values = vec![];
5098 for MapEntry { key, value } in entries {
5099 let key = plan_expr(ecx, key)?.type_as(ecx, &SqlScalarType::String)?;
5100 let value = match value {
5101 Expr::Map(entries) => plan_map(ecx, entries, type_hint)?,
5104 _ => plan_expr(ecx, value)?,
5105 };
5106 keys.push(key);
5107 values.push(value);
5108 }
5109 let values = coerce_homogeneous_exprs(&ecx.with_name("MAP"), values, type_hint)?;
5110 let value_type = ecx.scalar_type(&values[0]).without_modifiers();
5111 let out = itertools::interleave(keys, values).collect();
5112 (value_type, out)
5113 };
5114
5115 if matches!(value_type, SqlScalarType::Char { .. }) {
5116 bail_unsupported!("char map");
5117 }
5118
5119 let expr = HirScalarExpr::call_variadic(MapBuild { value_type }, exprs);
5120 Ok(expr.into())
5121}
5122
5123pub fn coerce_homogeneous_exprs(
5140 ecx: &ExprContext,
5141 exprs: Vec<CoercibleScalarExpr>,
5142 force_type: Option<&SqlScalarType>,
5143) -> Result<Vec<HirScalarExpr>, PlanError> {
5144 assert!(!exprs.is_empty());
5145
5146 let target_holder;
5147 let target = match force_type {
5148 Some(t) => t,
5149 None => {
5150 let types: Vec<_> = exprs.iter().map(|e| ecx.scalar_type(e)).collect();
5151 target_holder = typeconv::guess_best_common_type(ecx, &types)?;
5152 &target_holder
5153 }
5154 };
5155
5156 let mut out = Vec::new();
5158 for expr in exprs {
5159 let arg = typeconv::plan_coerce(ecx, expr, target)?;
5160 let ccx = match force_type {
5161 None => CastContext::Implicit,
5162 Some(_) => CastContext::Explicit,
5163 };
5164 match typeconv::plan_cast(ecx, ccx, arg.clone(), target) {
5165 Ok(expr) => out.push(expr),
5166 Err(_) => sql_bail!(
5167 "{} could not convert type {} to {}",
5168 ecx.name,
5169 ecx.humanize_sql_scalar_type(&ecx.scalar_type(&arg), false),
5170 ecx.humanize_sql_scalar_type(target, false),
5171 ),
5172 }
5173 }
5174 Ok(out)
5175}
5176
5177pub(crate) fn resolve_desc_and_nulls_last<T: AstInfo>(
5180 obe: &OrderByExpr<T>,
5181 column: usize,
5182) -> ColumnOrder {
5183 let desc = !obe.asc.unwrap_or(true);
5184 ColumnOrder {
5185 column,
5186 desc,
5187 nulls_last: obe.nulls_last.unwrap_or(!desc),
5190 }
5191}
5192
5193fn plan_function_order_by(
5201 ecx: &ExprContext,
5202 order_by: &[OrderByExpr<Aug>],
5203) -> Result<(Vec<HirScalarExpr>, Vec<ColumnOrder>), PlanError> {
5204 let mut order_by_exprs = vec![];
5205 let mut col_orders = vec![];
5206 {
5207 for (i, obe) in order_by.iter().enumerate() {
5208 let expr = plan_expr(ecx, &obe.expr)?.type_as_any(ecx)?;
5212 order_by_exprs.push(expr);
5213 col_orders.push(resolve_desc_and_nulls_last(obe, i));
5214 }
5215 }
5216 Ok((order_by_exprs, col_orders))
5217}
5218
5219fn humanize_or_debug(scx: &StatementContext, name: &ResolvedItemName) -> String {
5224 scx.humanize_resolved_name(name)
5225 .map(|n| n.to_string())
5226 .unwrap_or_else(|_| format!("<error when trying to humanize `{name:?}`>"))
5227}
5228
5229fn plan_aggregate_common(
5231 ecx: &ExprContext,
5232 Function::<Aug> {
5233 name,
5234 args,
5235 filter,
5236 over: _,
5237 distinct,
5238 }: &Function<Aug>,
5239) -> Result<AggregateExpr, PlanError> {
5240 let impls = match resolve_func(ecx, name, args)? {
5255 Func::Aggregate(impls) => impls,
5256 _ => bail_internal!("plan_aggregate_common called on non-aggregate function"),
5257 };
5258
5259 let (args, order_by) = match &args {
5268 FunctionArgs::Star => (vec![], vec![]),
5269 FunctionArgs::Args { args, order_by } => {
5270 if args.is_empty() {
5271 sql_bail!(
5272 "{}(*) must be used to call a parameterless aggregate function",
5273 humanize_or_debug(ecx.qcx.scx, name)
5274 );
5275 }
5276 let args = plan_exprs(ecx, args)?;
5277 (args, order_by.clone())
5278 }
5279 };
5280
5281 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &order_by)?;
5282
5283 let (mut expr, func) = func::select_impl(ecx, FuncSpec::Func(name), impls, args, col_orders)?;
5284 if let Some(filter) = &filter {
5285 let cond =
5295 plan_expr(&ecx.with_name("FILTER"), filter)?.type_as(ecx, &SqlScalarType::Bool)?;
5296 let expr_typ = ecx.scalar_type(&expr);
5297 expr = HirScalarExpr::if_then_else(
5298 cond,
5299 expr,
5300 HirScalarExpr::literal(func.identity_datum(), expr_typ),
5301 );
5302 }
5303
5304 let mut seen_outer = false;
5305 let mut seen_inner = false;
5306 #[allow(deprecated)]
5307 expr.visit_columns(0, &mut |depth, col| {
5308 if depth == 0 && col.level == 0 {
5309 seen_inner = true;
5310 } else if col.level > depth {
5311 seen_outer = true;
5312 }
5313 });
5314 if seen_outer && !seen_inner {
5315 bail_unsupported!(
5316 3720,
5317 "aggregate functions that refer exclusively to outer columns"
5318 );
5319 }
5320
5321 if func.is_order_sensitive() {
5324 let field_names = iter::repeat(ColumnName::from(""))
5325 .take(1 + order_by_exprs.len())
5326 .collect();
5327 let mut exprs = vec![expr];
5328 exprs.extend(order_by_exprs);
5329 expr = HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs);
5330 }
5331
5332 Ok(AggregateExpr {
5333 func,
5334 expr: Box::new(expr),
5335 distinct: *distinct,
5336 })
5337}
5338
5339fn plan_identifier(ecx: &ExprContext, names: &[Ident]) -> Result<HirScalarExpr, PlanError> {
5340 let mut names = names.to_vec();
5341 let Some(last) = names.pop() else {
5344 bail_internal!("empty identifier");
5345 };
5346 let col_name = normalize::column_name(last);
5347
5348 if !names.is_empty() {
5350 let table_name = normalize::unresolved_item_name(UnresolvedItemName(names))?;
5351 let (i, i_name) = ecx.scope.resolve_table_column(
5352 &ecx.qcx.outer_scopes,
5353 &table_name,
5354 &col_name,
5355 &mut ecx.qcx.name_manager.borrow_mut(),
5356 )?;
5357 return Ok(HirScalarExpr::named_column(i, i_name));
5358 }
5359
5360 let similar_names = match ecx.scope.resolve_column(
5363 &ecx.qcx.outer_scopes,
5364 &col_name,
5365 &mut ecx.qcx.name_manager.borrow_mut(),
5366 ) {
5367 Ok((i, i_name)) => {
5368 return Ok(HirScalarExpr::named_column(i, i_name));
5369 }
5370 Err(PlanError::UnknownColumn { similar, .. }) => similar,
5371 Err(e) => return Err(e),
5372 };
5373
5374 let items = ecx.scope.items_from_table(
5377 &ecx.qcx.outer_scopes,
5378 &PartialItemName {
5379 database: None,
5380 schema: None,
5381 item: col_name.as_str().to_owned(),
5382 },
5383 )?;
5384 match items.as_slice() {
5385 [] => Err(PlanError::UnknownColumn {
5387 table: None,
5388 column: col_name,
5389 similar: similar_names,
5390 }),
5391 [(column, item)] if item.from_single_column_function => Ok(HirScalarExpr::named_column(
5396 *column,
5397 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5398 )),
5399 _ => {
5402 let mut has_exists_column = None;
5403 let (exprs, field_names): (Vec<_>, Vec<_>) = items
5404 .into_iter()
5405 .filter_map(|(column, item)| {
5406 if item.is_exists_column_for_a_table_function_that_was_in_the_target_list {
5407 has_exists_column = Some(column);
5408 None
5409 } else {
5410 let expr = HirScalarExpr::named_column(
5411 column,
5412 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5413 );
5414 let name = item.column_name.clone();
5415 Some((expr, name))
5416 }
5417 })
5418 .unzip();
5419 let expr = if exprs.len() == 1 && has_exists_column.is_some() {
5421 exprs.into_element()
5422 } else {
5423 HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs)
5424 };
5425 if let Some(has_exists_column) = has_exists_column {
5426 Ok(HirScalarExpr::if_then_else(
5427 HirScalarExpr::unnamed_column(has_exists_column)
5428 .call_unary(UnaryFunc::IsNull(mz_expr::func::IsNull)),
5429 HirScalarExpr::literal_null(ecx.scalar_type(&expr)),
5430 expr,
5431 ))
5432 } else {
5433 Ok(expr)
5434 }
5435 }
5436 }
5437}
5438
5439fn plan_op(
5440 ecx: &ExprContext,
5441 op: &str,
5442 expr1: &Expr<Aug>,
5443 expr2: Option<&Expr<Aug>>,
5444) -> Result<HirScalarExpr, PlanError> {
5445 let impls = func::resolve_op(op)?;
5446 let args = match expr2 {
5447 None => plan_exprs(ecx, &[expr1])?,
5448 Some(expr2) => plan_exprs(ecx, &[expr1, expr2])?,
5449 };
5450 func::select_impl(ecx, FuncSpec::Op(op), impls, args, vec![])
5451}
5452
5453fn plan_function<'a>(
5454 ecx: &ExprContext,
5455 f @ Function {
5456 name,
5457 args,
5458 filter,
5459 over,
5460 distinct,
5461 }: &'a Function<Aug>,
5462) -> Result<HirScalarExpr, PlanError> {
5463 let impls = match resolve_func(ecx, name, args)? {
5464 Func::Table(_) => {
5465 sql_bail!(
5466 "table functions are not allowed in {} (function {})",
5467 ecx.name,
5468 name
5469 );
5470 }
5471 Func::Scalar(impls) => {
5472 if over.is_some() {
5473 sql_bail!(
5474 "OVER clause not allowed on {name}. The OVER clause can only be used with window functions (including aggregations)."
5475 );
5476 }
5477 impls
5478 }
5479 Func::ScalarWindow(impls) => {
5480 let (
5481 ignore_nulls,
5482 order_by_exprs,
5483 col_orders,
5484 _window_frame,
5485 partition_by,
5486 scalar_args,
5487 ) = plan_window_function_non_aggr(ecx, f)?;
5488
5489 if !scalar_args.is_empty() {
5493 if let ResolvedItemName::Item {
5494 full_name: FullItemName { item, .. },
5495 ..
5496 } = name
5497 {
5498 sql_bail!(
5499 "function {} has 0 parameters, but was called with {}",
5500 item,
5501 scalar_args.len()
5502 );
5503 }
5504 }
5505
5506 let func = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
5511
5512 if ignore_nulls {
5513 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5516 }
5517
5518 return Ok(HirScalarExpr::windowing(WindowExpr {
5519 func: WindowExprType::Scalar(ScalarWindowExpr {
5520 func,
5521 order_by: col_orders,
5522 }),
5523 partition_by,
5524 order_by: order_by_exprs,
5525 }));
5526 }
5527 Func::ValueWindow(impls) => {
5528 let window_plan = plan_window_function_non_aggr(ecx, f)?;
5529 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by, win_args) =
5530 window_plan;
5531
5532 let (args_encoded, func) =
5533 func::select_impl(ecx, FuncSpec::Func(name), impls, win_args, vec![])?;
5534
5535 if ignore_nulls {
5536 match func {
5537 ValueWindowFunc::Lag | ValueWindowFunc::Lead => {}
5538 _ => bail_unsupported!(IGNORE_NULLS_ERROR_MSG),
5539 }
5540 }
5541
5542 return Ok(HirScalarExpr::windowing(WindowExpr {
5543 func: WindowExprType::Value(ValueWindowExpr {
5544 func,
5545 args: Box::new(args_encoded),
5546 order_by: col_orders,
5547 window_frame,
5548 ignore_nulls, }),
5550 partition_by,
5551 order_by: order_by_exprs,
5552 }));
5553 }
5554 Func::Aggregate(_) => {
5555 if f.over.is_none() {
5556 if ecx.allow_aggregates {
5558 sql_bail!(
5561 "Internal error: encountered unplanned non-windowed aggregate function: {:?}",
5562 name,
5563 );
5564 } else {
5565 sql_bail!(
5568 "aggregate functions are not allowed in {} (function {})",
5569 ecx.name,
5570 name
5571 );
5572 }
5573 } else {
5574 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by) =
5575 plan_window_function_common(ecx, &f.name, &f.over)?;
5576
5577 match (&window_frame.start_bound, &window_frame.end_bound) {
5579 (
5580 mz_expr::WindowFrameBound::UnboundedPreceding,
5581 mz_expr::WindowFrameBound::OffsetPreceding(..),
5582 )
5583 | (
5584 mz_expr::WindowFrameBound::UnboundedPreceding,
5585 mz_expr::WindowFrameBound::OffsetFollowing(..),
5586 )
5587 | (
5588 mz_expr::WindowFrameBound::OffsetPreceding(..),
5589 mz_expr::WindowFrameBound::UnboundedFollowing,
5590 )
5591 | (
5592 mz_expr::WindowFrameBound::OffsetFollowing(..),
5593 mz_expr::WindowFrameBound::UnboundedFollowing,
5594 ) => bail_unsupported!("mixed unbounded - offset frames"),
5595 (_, _) => {} }
5597
5598 if ignore_nulls {
5599 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5603 }
5604
5605 let aggregate_expr = plan_aggregate_common(ecx, f)?;
5606
5607 if aggregate_expr.distinct {
5608 bail_unsupported!("DISTINCT in window aggregates");
5610 }
5611
5612 return Ok(HirScalarExpr::windowing(WindowExpr {
5613 func: WindowExprType::Aggregate(AggregateWindowExpr {
5614 aggregate_expr,
5615 order_by: col_orders,
5616 window_frame,
5617 }),
5618 partition_by,
5619 order_by: order_by_exprs,
5620 }));
5621 }
5622 }
5623 };
5624
5625 if over.is_some() {
5626 bail_internal!("OVER clause should have been handled by the window function path above");
5627 }
5628
5629 if *distinct {
5630 sql_bail!(
5631 "DISTINCT specified, but {} is not an aggregate function",
5632 humanize_or_debug(ecx.qcx.scx, name)
5633 );
5634 }
5635 if filter.is_some() {
5636 sql_bail!(
5637 "FILTER specified, but {} is not an aggregate function",
5638 humanize_or_debug(ecx.qcx.scx, name)
5639 );
5640 }
5641
5642 let scalar_args = match &args {
5643 FunctionArgs::Star => {
5644 sql_bail!(
5645 "* argument is invalid with non-aggregate function {}",
5646 humanize_or_debug(ecx.qcx.scx, name)
5647 )
5648 }
5649 FunctionArgs::Args { args, order_by } => {
5650 if !order_by.is_empty() {
5651 sql_bail!(
5652 "ORDER BY specified, but {} is not an aggregate function",
5653 humanize_or_debug(ecx.qcx.scx, name)
5654 );
5655 }
5656 plan_exprs(ecx, args)?
5657 }
5658 };
5659
5660 func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])
5661}
5662
5663pub const IGNORE_NULLS_ERROR_MSG: &str =
5664 "IGNORE NULLS and RESPECT NULLS options for functions other than LAG and LEAD";
5665
5666pub fn resolve_func(
5670 ecx: &ExprContext,
5671 name: &ResolvedItemName,
5672 args: &mz_sql_parser::ast::FunctionArgs<Aug>,
5673) -> Result<&'static Func, PlanError> {
5674 if let Ok(i) = ecx.qcx.scx.get_item_by_resolved_name(name) {
5675 if let Ok(f) = i.func() {
5676 return Ok(f);
5677 }
5678 }
5679
5680 let cexprs = match args {
5683 mz_sql_parser::ast::FunctionArgs::Star => vec![],
5684 mz_sql_parser::ast::FunctionArgs::Args { args, order_by } => {
5685 if !order_by.is_empty() {
5686 sql_bail!(
5687 "ORDER BY specified, but {} is not an aggregate function",
5688 name
5689 );
5690 }
5691 plan_exprs(ecx, args)?
5692 }
5693 };
5694
5695 let arg_types: Vec<_> = cexprs
5696 .into_iter()
5697 .map(|ty| match ecx.scalar_type(&ty) {
5698 CoercibleScalarType::Coerced(ty) => ecx.humanize_sql_scalar_type(&ty, false),
5699 CoercibleScalarType::Record(_) => "record".to_string(),
5700 CoercibleScalarType::Uncoerced => "unknown".to_string(),
5701 })
5702 .collect();
5703
5704 Err(PlanError::UnknownFunction {
5705 name: name.to_string(),
5706 arg_types,
5707 })
5708}
5709
5710fn plan_is_expr<'a>(
5711 ecx: &ExprContext,
5712 expr: &'a Expr<Aug>,
5713 construct: &IsExprConstruct<Aug>,
5714 not: bool,
5715) -> Result<HirScalarExpr, PlanError> {
5716 let expr_hir = plan_expr(ecx, expr)?;
5717
5718 let mut result = match construct {
5719 IsExprConstruct::Null => {
5720 expr_hir.type_as_any(ecx)?.call_is_null()
5725 }
5726 IsExprConstruct::Unknown => expr_hir.type_as(ecx, &SqlScalarType::Bool)?.call_is_null(),
5727 IsExprConstruct::True => expr_hir
5728 .type_as(ecx, &SqlScalarType::Bool)?
5729 .call_unary(UnaryFunc::IsTrue(expr_func::IsTrue)),
5730 IsExprConstruct::False => expr_hir
5731 .type_as(ecx, &SqlScalarType::Bool)?
5732 .call_unary(UnaryFunc::IsFalse(expr_func::IsFalse)),
5733 IsExprConstruct::DistinctFrom(expr2) => {
5734 let ne_ast = expr.clone().not_equals(expr2.as_ref().clone());
5745 let ne_hir = plan_expr(ecx, &ne_ast)?.type_as_any(ecx)?;
5746
5747 let expr1_hir = expr_hir.type_as_any(ecx)?;
5748 let expr2_hir = plan_expr(ecx, expr2)?.type_as_any(ecx)?;
5749
5750 let term1 = HirScalarExpr::variadic_or(vec![
5751 ne_hir,
5752 expr1_hir.clone().call_is_null(),
5753 expr2_hir.clone().call_is_null(),
5754 ]);
5755 let term2 = HirScalarExpr::variadic_or(vec![
5756 expr1_hir.call_is_null().not(),
5757 expr2_hir.call_is_null().not(),
5758 ]);
5759 term1.and(term2)
5760 }
5761 };
5762 if not {
5763 result = result.not();
5764 }
5765 Ok(result)
5766}
5767
5768fn plan_case<'a>(
5769 ecx: &ExprContext,
5770 operand: &'a Option<Box<Expr<Aug>>>,
5771 conditions: &'a [Expr<Aug>],
5772 results: &'a [Expr<Aug>],
5773 else_result: &'a Option<Box<Expr<Aug>>>,
5774) -> Result<HirScalarExpr, PlanError> {
5775 let mut cond_exprs = Vec::new();
5776 let mut result_exprs = Vec::new();
5777 for (c, r) in conditions.iter().zip_eq(results) {
5778 let c = match operand {
5779 Some(operand) => operand.clone().equals(c.clone()),
5780 None => c.clone(),
5781 };
5782 let cexpr = plan_expr(ecx, &c)?.type_as(ecx, &SqlScalarType::Bool)?;
5783 cond_exprs.push(cexpr);
5784 result_exprs.push(r);
5785 }
5786 result_exprs.push(match else_result {
5787 Some(else_result) => else_result,
5788 None => &Expr::Value(Value::Null),
5789 });
5790 let mut result_exprs = coerce_homogeneous_exprs(
5791 &ecx.with_name("CASE"),
5792 plan_exprs(ecx, &result_exprs)?,
5793 None,
5794 )?;
5795 let mut expr = result_exprs.pop().unwrap();
5796 assert_eq!(cond_exprs.len(), result_exprs.len());
5797 for (cexpr, rexpr) in cond_exprs
5798 .into_iter()
5799 .rev()
5800 .zip_eq(result_exprs.into_iter().rev())
5801 {
5802 expr = HirScalarExpr::if_then_else(cexpr, rexpr, expr);
5803 }
5804 Ok(expr)
5805}
5806
5807fn plan_literal<'a>(l: &'a Value) -> Result<CoercibleScalarExpr, PlanError> {
5808 let (datum, scalar_type) = match l {
5809 Value::Number(s) => {
5810 let d = strconv::parse_numeric(s.as_str())?;
5811 if !s.contains(&['E', '.'][..]) {
5812 if let Ok(n) = d.0.try_into() {
5814 (Datum::Int32(n), SqlScalarType::Int32)
5815 } else if let Ok(n) = d.0.try_into() {
5816 (Datum::Int64(n), SqlScalarType::Int64)
5817 } else {
5818 (
5819 Datum::Numeric(d),
5820 SqlScalarType::Numeric { max_scale: None },
5821 )
5822 }
5823 } else {
5824 (
5825 Datum::Numeric(d),
5826 SqlScalarType::Numeric { max_scale: None },
5827 )
5828 }
5829 }
5830 Value::HexString(_) => bail_unsupported!("hex string literals"),
5831 Value::Boolean(b) => match b {
5832 false => (Datum::False, SqlScalarType::Bool),
5833 true => (Datum::True, SqlScalarType::Bool),
5834 },
5835 Value::Interval(i) => {
5836 let i = literal::plan_interval(i)?;
5837 (Datum::Interval(i), SqlScalarType::Interval)
5838 }
5839 Value::String(s) => return Ok(CoercibleScalarExpr::LiteralString(s.clone())),
5840 Value::Null => return Ok(CoercibleScalarExpr::LiteralNull),
5841 };
5842 let expr = HirScalarExpr::literal(datum, scalar_type);
5843 Ok(expr.into())
5844}
5845
5846fn plan_window_function_non_aggr<'a>(
5849 ecx: &ExprContext,
5850 Function {
5851 name,
5852 args,
5853 filter,
5854 over,
5855 distinct,
5856 }: &'a Function<Aug>,
5857) -> Result<
5858 (
5859 bool,
5860 Vec<HirScalarExpr>,
5861 Vec<ColumnOrder>,
5862 mz_expr::WindowFrame,
5863 Vec<HirScalarExpr>,
5864 Vec<CoercibleScalarExpr>,
5865 ),
5866 PlanError,
5867> {
5868 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition) =
5869 plan_window_function_common(ecx, name, over)?;
5870
5871 if *distinct {
5872 sql_bail!(
5873 "DISTINCT specified, but {} is not an aggregate function",
5874 name
5875 );
5876 }
5877
5878 if filter.is_some() {
5879 bail_unsupported!("FILTER in non-aggregate window functions");
5880 }
5881
5882 let scalar_args = match &args {
5883 FunctionArgs::Star => {
5884 sql_bail!("* argument is invalid with non-aggregate function {}", name)
5885 }
5886 FunctionArgs::Args { args, order_by } => {
5887 if !order_by.is_empty() {
5888 sql_bail!(
5889 "ORDER BY specified, but {} is not an aggregate function",
5890 name
5891 );
5892 }
5893 plan_exprs(ecx, args)?
5894 }
5895 };
5896
5897 Ok((
5898 ignore_nulls,
5899 order_by_exprs,
5900 col_orders,
5901 window_frame,
5902 partition,
5903 scalar_args,
5904 ))
5905}
5906
5907fn plan_window_function_common(
5909 ecx: &ExprContext,
5910 name: &<Aug as AstInfo>::ItemName,
5911 over: &Option<WindowSpec<Aug>>,
5912) -> Result<
5913 (
5914 bool,
5915 Vec<HirScalarExpr>,
5916 Vec<ColumnOrder>,
5917 mz_expr::WindowFrame,
5918 Vec<HirScalarExpr>,
5919 ),
5920 PlanError,
5921> {
5922 if !ecx.allow_windows {
5923 sql_bail!(
5924 "window functions are not allowed in {} (function {})",
5925 ecx.name,
5926 name
5927 );
5928 }
5929
5930 let window_spec = match over.as_ref() {
5931 Some(over) => over,
5932 None => sql_bail!("window function {} requires an OVER clause", name),
5933 };
5934 if window_spec.ignore_nulls && window_spec.respect_nulls {
5935 sql_bail!("Both IGNORE NULLS and RESPECT NULLS were given.");
5936 }
5937 let window_frame = match window_spec.window_frame.as_ref() {
5938 Some(frame) => plan_window_frame(frame)?,
5939 None => mz_expr::WindowFrame::default(),
5940 };
5941 let mut partition = Vec::new();
5942 for expr in &window_spec.partition_by {
5943 partition.push(plan_expr(ecx, expr)?.type_as_any(ecx)?);
5944 }
5945
5946 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &window_spec.order_by)?;
5947
5948 Ok((
5949 window_spec.ignore_nulls,
5950 order_by_exprs,
5951 col_orders,
5952 window_frame,
5953 partition,
5954 ))
5955}
5956
5957fn plan_window_frame(
5958 WindowFrame {
5959 units,
5960 start_bound,
5961 end_bound,
5962 }: &WindowFrame,
5963) -> Result<mz_expr::WindowFrame, PlanError> {
5964 use mz_expr::WindowFrameBound::*;
5965 let units = window_frame_unit_ast_to_expr(units)?;
5966 let start_bound = window_frame_bound_ast_to_expr(start_bound);
5967 let end_bound = end_bound
5968 .as_ref()
5969 .map(window_frame_bound_ast_to_expr)
5970 .unwrap_or(CurrentRow);
5971
5972 match (&start_bound, &end_bound) {
5974 (UnboundedFollowing, _) => {
5976 sql_bail!("frame start cannot be UNBOUNDED FOLLOWING")
5977 }
5978 (_, UnboundedPreceding) => {
5980 sql_bail!("frame end cannot be UNBOUNDED PRECEDING")
5981 }
5982 (CurrentRow, OffsetPreceding(_)) => {
5984 sql_bail!("frame starting from current row cannot have preceding rows")
5985 }
5986 (OffsetFollowing(_), OffsetPreceding(_) | CurrentRow) => {
5987 sql_bail!("frame starting from following row cannot have preceding rows")
5988 }
5989 (OffsetPreceding(o1), OffsetFollowing(o2)) => {
5992 if *o1 > 1000000 || *o2 > 1000000 {
5996 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
5997 }
5998 }
5999 (OffsetPreceding(o1), OffsetPreceding(o2)) => {
6000 if *o1 > 1000000 || *o2 > 1000000 {
6001 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6002 }
6003 }
6004 (OffsetFollowing(o1), OffsetFollowing(o2)) => {
6005 if *o1 > 1000000 || *o2 > 1000000 {
6006 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6007 }
6008 }
6009 (OffsetPreceding(o), CurrentRow) => {
6010 if *o > 1000000 {
6011 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6012 }
6013 }
6014 (CurrentRow, OffsetFollowing(o)) => {
6015 if *o > 1000000 {
6016 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6017 }
6018 }
6019 (_, _) => (),
6021 }
6022
6023 if units == mz_expr::WindowFrameUnits::Range
6026 && (start_bound != UnboundedPreceding || end_bound != CurrentRow)
6027 {
6028 bail_unsupported!("RANGE in non-default window frames")
6029 }
6030
6031 let frame = mz_expr::WindowFrame {
6032 units,
6033 start_bound,
6034 end_bound,
6035 };
6036 Ok(frame)
6037}
6038
6039fn window_frame_unit_ast_to_expr(
6040 unit: &WindowFrameUnits,
6041) -> Result<mz_expr::WindowFrameUnits, PlanError> {
6042 match unit {
6043 WindowFrameUnits::Rows => Ok(mz_expr::WindowFrameUnits::Rows),
6044 WindowFrameUnits::Range => Ok(mz_expr::WindowFrameUnits::Range),
6045 WindowFrameUnits::Groups => bail_unsupported!("GROUPS in window frames"),
6046 }
6047}
6048
6049fn window_frame_bound_ast_to_expr(bound: &WindowFrameBound) -> mz_expr::WindowFrameBound {
6050 match bound {
6051 WindowFrameBound::CurrentRow => mz_expr::WindowFrameBound::CurrentRow,
6052 WindowFrameBound::Preceding(None) => mz_expr::WindowFrameBound::UnboundedPreceding,
6053 WindowFrameBound::Preceding(Some(offset)) => {
6054 mz_expr::WindowFrameBound::OffsetPreceding(*offset)
6055 }
6056 WindowFrameBound::Following(None) => mz_expr::WindowFrameBound::UnboundedFollowing,
6057 WindowFrameBound::Following(Some(offset)) => {
6058 mz_expr::WindowFrameBound::OffsetFollowing(*offset)
6059 }
6060 }
6061}
6062
6063pub fn scalar_type_from_sql(
6064 scx: &StatementContext,
6065 data_type: &ResolvedDataType,
6066) -> Result<SqlScalarType, PlanError> {
6067 match data_type {
6068 ResolvedDataType::AnonymousList(elem_type) => {
6069 let elem_type = scalar_type_from_sql(scx, elem_type)?;
6070 if matches!(elem_type, SqlScalarType::Char { .. }) {
6071 bail_unsupported!("char list");
6072 }
6073 Ok(SqlScalarType::List {
6074 element_type: Box::new(elem_type),
6075 custom_id: None,
6076 })
6077 }
6078 ResolvedDataType::AnonymousMap {
6079 key_type,
6080 value_type,
6081 } => {
6082 match scalar_type_from_sql(scx, key_type)? {
6083 SqlScalarType::String => {}
6084 other => sql_bail!(
6085 "map key type must be {}, got {}",
6086 scx.humanize_sql_scalar_type(&SqlScalarType::String, false),
6087 scx.humanize_sql_scalar_type(&other, false)
6088 ),
6089 }
6090 Ok(SqlScalarType::Map {
6091 value_type: Box::new(scalar_type_from_sql(scx, value_type)?),
6092 custom_id: None,
6093 })
6094 }
6095 ResolvedDataType::Named { id, modifiers, .. } => {
6096 scalar_type_from_catalog(scx.catalog, *id, modifiers)
6097 }
6098 ResolvedDataType::Error => bail_internal!("should have been caught in name resolution"),
6099 }
6100}
6101
6102const MAX_TYPE_NESTING_DEPTH: usize = 128;
6106
6107const MAX_TYPE_RESOLUTION_NODES: usize = 100_000;
6113
6114pub fn scalar_type_from_catalog(
6115 catalog: &dyn SessionCatalog,
6116 id: CatalogItemId,
6117 modifiers: &[i64],
6118) -> Result<SqlScalarType, PlanError> {
6119 let (depth_limit, mut budget) = type_resolution_limits(catalog);
6120 scalar_type_from_catalog_inner(catalog, id, modifiers, 0, depth_limit, &mut budget)
6121}
6122
6123fn type_resolution_limits(catalog: &dyn SessionCatalog) -> (usize, usize) {
6137 if catalog
6138 .system_vars()
6139 .unsafe_enable_unbounded_custom_type_resolution()
6140 {
6141 (usize::MAX, usize::MAX)
6142 } else {
6143 (MAX_TYPE_NESTING_DEPTH, MAX_TYPE_RESOLUTION_NODES)
6144 }
6145}
6146
6147pub struct TypeResolutionBudget {
6160 remaining: usize,
6163 depth_limit: usize,
6166}
6167
6168impl TypeResolutionBudget {
6169 pub fn for_root(catalog: &dyn SessionCatalog) -> TypeResolutionBudget {
6175 let (depth_limit, budget) = type_resolution_limits(catalog);
6176 TypeResolutionBudget {
6177 remaining: budget.saturating_sub(1),
6179 depth_limit,
6180 }
6181 }
6182
6183 pub fn resolve_child(
6187 &mut self,
6188 catalog: &dyn SessionCatalog,
6189 id: CatalogItemId,
6190 modifiers: &[i64],
6191 ) -> Result<SqlScalarType, PlanError> {
6192 scalar_type_from_catalog_inner(
6193 catalog,
6194 id,
6195 modifiers,
6196 1,
6197 self.depth_limit,
6198 &mut self.remaining,
6199 )
6200 }
6201}
6202
6203fn scalar_type_from_catalog_inner(
6204 catalog: &dyn SessionCatalog,
6205 id: CatalogItemId,
6206 modifiers: &[i64],
6207 depth: usize,
6208 depth_limit: usize,
6209 budget: &mut usize,
6210) -> Result<SqlScalarType, PlanError> {
6211 if depth > depth_limit {
6212 sql_bail!("custom type nesting depth exceeds limit of {}", depth_limit);
6213 }
6214 *budget = match budget.checked_sub(1) {
6215 Some(remaining) => remaining,
6216 None => sql_bail!("custom type is too complex to resolve"),
6217 };
6218 let entry = catalog.get_item(&id);
6219 let type_details = match entry.type_details() {
6220 Some(type_details) => type_details,
6221 None => {
6222 sql_bail!(
6225 "internal error: {} does not refer to a type",
6226 catalog.resolve_full_name(entry.name()).to_string().quoted()
6227 );
6228 }
6229 };
6230 match &type_details.typ {
6231 CatalogType::Numeric => {
6232 let mut modifiers = modifiers.iter().fuse();
6233 let precision = match modifiers.next() {
6234 Some(p) if *p < 1 || *p > i64::from(NUMERIC_DATUM_MAX_PRECISION) => {
6235 sql_bail!(
6236 "precision for type numeric must be between 1 and {}",
6237 NUMERIC_DATUM_MAX_PRECISION,
6238 );
6239 }
6240 Some(p) => Some(*p),
6241 None => None,
6242 };
6243 let scale = match modifiers.next() {
6244 Some(scale) => {
6245 if let Some(precision) = precision {
6246 if *scale > precision {
6247 sql_bail!(
6248 "scale for type numeric must be between 0 and precision {}",
6249 precision
6250 );
6251 }
6252 }
6253 Some(NumericMaxScale::try_from(*scale)?)
6254 }
6255 None => None,
6256 };
6257 if modifiers.next().is_some() {
6258 sql_bail!("type numeric supports at most two type modifiers");
6259 }
6260 Ok(SqlScalarType::Numeric { max_scale: scale })
6261 }
6262 CatalogType::Char => {
6263 let mut modifiers = modifiers.iter().fuse();
6264 let length = match modifiers.next() {
6265 Some(l) => Some(CharLength::try_from(*l)?),
6266 None => Some(CharLength::ONE),
6267 };
6268 if modifiers.next().is_some() {
6269 sql_bail!("type character supports at most one type modifier");
6270 }
6271 Ok(SqlScalarType::Char { length })
6272 }
6273 CatalogType::VarChar => {
6274 let mut modifiers = modifiers.iter().fuse();
6275 let length = match modifiers.next() {
6276 Some(l) => Some(VarCharMaxLength::try_from(*l)?),
6277 None => None,
6278 };
6279 if modifiers.next().is_some() {
6280 sql_bail!("type character varying supports at most one type modifier");
6281 }
6282 Ok(SqlScalarType::VarChar { max_length: length })
6283 }
6284 CatalogType::Timestamp => {
6285 let mut modifiers = modifiers.iter().fuse();
6286 let precision = match modifiers.next() {
6287 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6288 None => None,
6289 };
6290 if modifiers.next().is_some() {
6291 sql_bail!("type timestamp supports at most one type modifier");
6292 }
6293 Ok(SqlScalarType::Timestamp { precision })
6294 }
6295 CatalogType::TimestampTz => {
6296 let mut modifiers = modifiers.iter().fuse();
6297 let precision = match modifiers.next() {
6298 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6299 None => None,
6300 };
6301 if modifiers.next().is_some() {
6302 sql_bail!("type timestamp with time zone supports at most one type modifier");
6303 }
6304 Ok(SqlScalarType::TimestampTz { precision })
6305 }
6306 t => {
6307 if !modifiers.is_empty() {
6308 sql_bail!(
6309 "{} does not support type modifiers",
6310 catalog.resolve_full_name(entry.name()).to_string()
6311 );
6312 }
6313 match t {
6314 CatalogType::Array {
6315 element_reference: element_id,
6316 } => Ok(SqlScalarType::Array(Box::new(
6317 scalar_type_from_catalog_inner(
6318 catalog,
6319 *element_id,
6320 modifiers,
6321 depth + 1,
6322 depth_limit,
6323 budget,
6324 )?,
6325 ))),
6326 CatalogType::List {
6327 element_reference: element_id,
6328 element_modifiers,
6329 } => Ok(SqlScalarType::List {
6330 element_type: Box::new(scalar_type_from_catalog_inner(
6331 catalog,
6332 *element_id,
6333 element_modifiers,
6334 depth + 1,
6335 depth_limit,
6336 budget,
6337 )?),
6338 custom_id: Some(id),
6339 }),
6340 CatalogType::Map {
6341 key_reference: _,
6342 key_modifiers: _,
6343 value_reference: value_id,
6344 value_modifiers,
6345 } => Ok(SqlScalarType::Map {
6346 value_type: Box::new(scalar_type_from_catalog_inner(
6347 catalog,
6348 *value_id,
6349 value_modifiers,
6350 depth + 1,
6351 depth_limit,
6352 budget,
6353 )?),
6354 custom_id: Some(id),
6355 }),
6356 CatalogType::Range {
6357 element_reference: element_id,
6358 } => Ok(SqlScalarType::Range {
6359 element_type: Box::new(scalar_type_from_catalog_inner(
6360 catalog,
6361 *element_id,
6362 &[],
6363 depth + 1,
6364 depth_limit,
6365 budget,
6366 )?),
6367 }),
6368 CatalogType::Record { fields } => {
6369 let scalars: Box<[(ColumnName, SqlColumnType)]> = fields
6370 .iter()
6371 .map(|f| {
6372 let scalar_type = scalar_type_from_catalog_inner(
6373 catalog,
6374 f.type_reference,
6375 &f.type_modifiers,
6376 depth + 1,
6377 depth_limit,
6378 budget,
6379 )?;
6380 Ok((
6381 f.name.clone(),
6382 SqlColumnType {
6383 scalar_type,
6384 nullable: true,
6385 },
6386 ))
6387 })
6388 .collect::<Result<Box<_>, PlanError>>()?;
6389 Ok(SqlScalarType::Record {
6390 fields: scalars,
6391 custom_id: Some(id),
6392 })
6393 }
6394 CatalogType::AclItem => Ok(SqlScalarType::AclItem),
6395 CatalogType::Bool => Ok(SqlScalarType::Bool),
6396 CatalogType::Bytes => Ok(SqlScalarType::Bytes),
6397 CatalogType::Date => Ok(SqlScalarType::Date),
6398 CatalogType::Float32 => Ok(SqlScalarType::Float32),
6399 CatalogType::Float64 => Ok(SqlScalarType::Float64),
6400 CatalogType::Int16 => Ok(SqlScalarType::Int16),
6401 CatalogType::Int32 => Ok(SqlScalarType::Int32),
6402 CatalogType::Int64 => Ok(SqlScalarType::Int64),
6403 CatalogType::UInt16 => Ok(SqlScalarType::UInt16),
6404 CatalogType::UInt32 => Ok(SqlScalarType::UInt32),
6405 CatalogType::UInt64 => Ok(SqlScalarType::UInt64),
6406 CatalogType::MzTimestamp => Ok(SqlScalarType::MzTimestamp),
6407 CatalogType::Interval => Ok(SqlScalarType::Interval),
6408 CatalogType::Jsonb => Ok(SqlScalarType::Jsonb),
6409 CatalogType::Oid => Ok(SqlScalarType::Oid),
6410 CatalogType::PgLegacyChar => Ok(SqlScalarType::PgLegacyChar),
6411 CatalogType::PgLegacyName => Ok(SqlScalarType::PgLegacyName),
6412 CatalogType::Pseudo => {
6413 sql_bail!(
6414 "cannot reference pseudo type {}",
6415 catalog.resolve_full_name(entry.name()).to_string()
6416 )
6417 }
6418 CatalogType::RegClass => Ok(SqlScalarType::RegClass),
6419 CatalogType::RegProc => Ok(SqlScalarType::RegProc),
6420 CatalogType::RegType => Ok(SqlScalarType::RegType),
6421 CatalogType::String => Ok(SqlScalarType::String),
6422 CatalogType::Time => Ok(SqlScalarType::Time),
6423 CatalogType::Uuid => Ok(SqlScalarType::Uuid),
6424 CatalogType::Int2Vector => Ok(SqlScalarType::Int2Vector),
6425 CatalogType::MzAclItem => Ok(SqlScalarType::MzAclItem),
6426 CatalogType::Numeric => unreachable!("handled above"),
6427 CatalogType::Char => unreachable!("handled above"),
6428 CatalogType::VarChar => unreachable!("handled above"),
6429 CatalogType::Timestamp => unreachable!("handled above"),
6430 CatalogType::TimestampTz => unreachable!("handled above"),
6431 }
6432 }
6433 }
6434}
6435
6436struct AggregateTableFuncVisitor<'a> {
6439 scx: &'a StatementContext<'a>,
6440 aggs: Vec<Function<Aug>>,
6441 within_aggregate: bool,
6442 tables: BTreeMap<Function<Aug>, String>,
6443 table_disallowed_context: Vec<&'static str>,
6444 in_select_item: bool,
6445 id_gen: IdGen,
6446 err: Option<PlanError>,
6447}
6448
6449impl<'a> AggregateTableFuncVisitor<'a> {
6450 fn new(scx: &'a StatementContext<'a>) -> AggregateTableFuncVisitor<'a> {
6451 AggregateTableFuncVisitor {
6452 scx,
6453 aggs: Vec::new(),
6454 within_aggregate: false,
6455 tables: BTreeMap::new(),
6456 table_disallowed_context: Vec::new(),
6457 in_select_item: false,
6458 id_gen: Default::default(),
6459 err: None,
6460 }
6461 }
6462
6463 fn into_result(
6464 self,
6465 ) -> Result<(Vec<Function<Aug>>, BTreeMap<Function<Aug>, String>), PlanError> {
6466 match self.err {
6467 Some(err) => Err(err),
6468 None => {
6469 let mut seen = BTreeSet::new();
6472 let aggs = self
6473 .aggs
6474 .into_iter()
6475 .filter(move |agg| seen.insert(agg.clone()))
6476 .collect();
6477 Ok((aggs, self.tables))
6478 }
6479 }
6480 }
6481}
6482
6483impl<'a> VisitMut<'_, Aug> for AggregateTableFuncVisitor<'a> {
6484 fn visit_function_mut(&mut self, func: &mut Function<Aug>) {
6485 let item = match self.scx.get_item_by_resolved_name(&func.name) {
6486 Ok(i) => i,
6487 Err(_) => return,
6489 };
6490
6491 match item.func() {
6492 Ok(Func::Aggregate { .. }) if func.over.is_none() => {
6495 if self.within_aggregate {
6496 self.err = Some(sql_err!("nested aggregate functions are not allowed",));
6497 return;
6498 }
6499 self.aggs.push(func.clone());
6500 let Function {
6501 name: _,
6502 args,
6503 filter,
6504 over: _,
6505 distinct: _,
6506 } = func;
6507 if let Some(filter) = filter {
6508 self.visit_expr_mut(filter);
6509 }
6510 let old_within_aggregate = self.within_aggregate;
6511 self.within_aggregate = true;
6512 self.table_disallowed_context
6513 .push("aggregate function calls");
6514
6515 self.visit_function_args_mut(args);
6516
6517 self.within_aggregate = old_within_aggregate;
6518 self.table_disallowed_context.pop();
6519 }
6520 Ok(Func::Table { .. }) => {
6521 self.table_disallowed_context.push("other table functions");
6522 visit_mut::visit_function_mut(self, func);
6523 self.table_disallowed_context.pop();
6524 }
6525 _ => visit_mut::visit_function_mut(self, func),
6526 }
6527 }
6528
6529 fn visit_query_mut(&mut self, _query: &mut Query<Aug>) {
6530 }
6532
6533 fn visit_expr_mut(&mut self, expr: &mut Expr<Aug>) {
6534 let (disallowed_context, func) = match expr {
6535 Expr::Case { .. } => (Some("CASE"), None),
6536 Expr::HomogenizingFunction {
6537 function: HomogenizingFunction::Coalesce,
6538 ..
6539 } => (Some("COALESCE"), None),
6540 Expr::Function(func) if self.in_select_item => {
6541 let mut table_func = None;
6544 if let Ok(item) = self.scx.get_item_by_resolved_name(&func.name) {
6545 if let Ok(Func::Table { .. }) = item.func() {
6546 if let Some(context) = self.table_disallowed_context.last() {
6547 self.err = Some(sql_err!(
6548 "table functions are not allowed in {} (function {})",
6549 context,
6550 func.name
6551 ));
6552 return;
6553 }
6554 table_func = Some(func.clone());
6555 }
6556 }
6557 (None, table_func)
6560 }
6561 _ => (None, None),
6562 };
6563 if let Some(func) = func {
6564 visit_mut::visit_expr_mut(self, expr);
6566 if let Function {
6568 name: _,
6569 args: _,
6570 filter: None,
6571 over: None,
6572 distinct: false,
6573 } = &func
6574 {
6575 let unique_id = self.id_gen.allocate_id();
6577 let id = self
6578 .tables
6579 .entry(func)
6580 .or_insert_with(|| format!("table_func_{unique_id}"));
6581 *expr = Expr::Identifier(vec![Ident::new_unchecked(id.clone())]);
6584 }
6585 }
6586 if let Some(context) = disallowed_context {
6587 self.table_disallowed_context.push(context);
6588 }
6589
6590 visit_mut::visit_expr_mut(self, expr);
6591
6592 if disallowed_context.is_some() {
6593 self.table_disallowed_context.pop();
6594 }
6595 }
6596
6597 fn visit_select_item_mut(&mut self, si: &mut SelectItem<Aug>) {
6598 let old = self.in_select_item;
6599 self.in_select_item = true;
6600 visit_mut::visit_select_item_mut(self, si);
6601 self.in_select_item = old;
6602 }
6603}
6604
6605#[derive(Default)]
6606struct WindowFuncCollector {
6607 window_funcs: Vec<Expr<Aug>>,
6608}
6609
6610impl WindowFuncCollector {
6611 fn into_result(self) -> Vec<Expr<Aug>> {
6612 let mut seen = BTreeSet::new();
6614 let window_funcs_dedupped = self
6615 .window_funcs
6616 .into_iter()
6617 .filter(move |expr| seen.insert(expr.clone()))
6618 .rev()
6621 .collect();
6622 window_funcs_dedupped
6623 }
6624}
6625
6626impl Visit<'_, Aug> for WindowFuncCollector {
6627 fn visit_expr(&mut self, expr: &Expr<Aug>) {
6628 match expr {
6629 Expr::Function(func) => {
6630 if func.over.is_some() {
6631 self.window_funcs.push(expr.clone());
6632 }
6633 }
6634 _ => (),
6635 }
6636 visit::visit_expr(self, expr);
6637 }
6638
6639 fn visit_query(&mut self, _query: &Query<Aug>) {
6640 }
6642}
6643
6644#[derive(Debug, Eq, PartialEq, Copy, Clone)]
6646pub enum QueryLifetime {
6647 OneShot,
6649 Index,
6651 MaterializedView,
6653 Subscribe,
6655 View,
6657 Source,
6659}
6660
6661impl QueryLifetime {
6662 pub fn is_one_shot(&self) -> bool {
6666 let result = match self {
6667 QueryLifetime::OneShot => true,
6668 QueryLifetime::Index => false,
6669 QueryLifetime::MaterializedView => false,
6670 QueryLifetime::Subscribe => false,
6671 QueryLifetime::View => false,
6672 QueryLifetime::Source => false,
6673 };
6674 assert_eq!(!result, self.is_maintained());
6675 result
6676 }
6677
6678 pub fn is_maintained(&self) -> bool {
6681 match self {
6682 QueryLifetime::OneShot => false,
6683 QueryLifetime::Index => true,
6684 QueryLifetime::MaterializedView => true,
6685 QueryLifetime::Subscribe => true,
6686 QueryLifetime::View => true,
6687 QueryLifetime::Source => true,
6688 }
6689 }
6690
6691 pub fn allow_show(&self) -> bool {
6693 match self {
6694 QueryLifetime::OneShot => true,
6695 QueryLifetime::Index => false,
6696 QueryLifetime::MaterializedView => false,
6697 QueryLifetime::Subscribe => true, QueryLifetime::View => false,
6699 QueryLifetime::Source => false,
6700 }
6701 }
6702}
6703
6704#[derive(Debug, Clone)]
6706pub struct CteDesc {
6707 pub name: String,
6708 pub desc: RelationDesc,
6709}
6710
6711#[derive(Debug, Clone)]
6713pub struct QueryContext<'a> {
6714 pub scx: &'a StatementContext<'a>,
6716 pub lifetime: QueryLifetime,
6718 pub outer_scopes: Vec<Scope>,
6720 pub outer_relation_types: Vec<SqlRelationType>,
6722 pub ctes: BTreeMap<LocalId, CteDesc>,
6724 pub name_manager: Rc<RefCell<NameManager>>,
6726 pub recursion_guard: RecursionGuard,
6727}
6728
6729impl CheckedRecursion for QueryContext<'_> {
6730 fn recursion_guard(&self) -> &RecursionGuard {
6731 &self.recursion_guard
6732 }
6733}
6734
6735impl<'a> QueryContext<'a> {
6736 pub fn root(scx: &'a StatementContext, lifetime: QueryLifetime) -> QueryContext<'a> {
6737 QueryContext {
6738 scx,
6739 lifetime,
6740 outer_scopes: vec![],
6741 outer_relation_types: vec![],
6742 ctes: BTreeMap::new(),
6743 name_manager: Rc::new(RefCell::new(NameManager::new())),
6744 recursion_guard: RecursionGuard::with_limit(1024), }
6746 }
6747
6748 fn relation_type(&self, expr: &HirRelationExpr) -> SqlRelationType {
6749 expr.typ(&self.outer_relation_types, &self.scx.param_types.borrow())
6750 }
6751
6752 fn derived_context(&self, scope: Scope, relation_type: SqlRelationType) -> QueryContext<'a> {
6755 let ctes = self.ctes.clone();
6756 let outer_scopes = iter::once(scope).chain(self.outer_scopes.clone()).collect();
6757 let outer_relation_types = iter::once(relation_type)
6758 .chain(self.outer_relation_types.clone())
6759 .collect();
6760 let name_manager = Rc::clone(&self.name_manager);
6762
6763 QueryContext {
6764 scx: self.scx,
6765 lifetime: self.lifetime,
6766 outer_scopes,
6767 outer_relation_types,
6768 ctes,
6769 name_manager,
6770 recursion_guard: self.recursion_guard.clone(),
6771 }
6772 }
6773
6774 fn empty_derived_context(&self) -> QueryContext<'a> {
6776 let scope = Scope::empty();
6777 let ty = SqlRelationType::empty();
6778 self.derived_context(scope, ty)
6779 }
6780
6781 pub fn resolve_table_name(
6784 &self,
6785 object: ResolvedItemName,
6786 ) -> Result<(HirRelationExpr, Scope), PlanError> {
6787 match object {
6788 ResolvedItemName::Item {
6789 id,
6790 full_name,
6791 version,
6792 ..
6793 } => {
6794 let item = self.scx.get_item(&id).at_version(version);
6795 let desc = match item.relation_desc() {
6796 Some(desc) => desc.clone(),
6797 None => {
6798 return Err(PlanError::InvalidDependency {
6799 name: full_name.to_string(),
6800 item_type: item.item_type().to_string(),
6801 });
6802 }
6803 };
6804 let expr = HirRelationExpr::Get {
6805 id: Id::Global(item.global_id()),
6806 typ: desc.typ().clone(),
6807 };
6808
6809 let name = full_name.into();
6810 let scope = Scope::from_source(Some(name), desc.iter_names().cloned());
6811
6812 Ok((expr, scope))
6813 }
6814 ResolvedItemName::Cte { id, name } => {
6815 let name = name.into();
6816 let cte = self.ctes.get(&id).unwrap();
6817 let expr = HirRelationExpr::Get {
6818 id: Id::Local(id),
6819 typ: cte.desc.typ().clone(),
6820 };
6821
6822 let scope = Scope::from_source(Some(name), cte.desc.iter_names());
6823
6824 Ok((expr, scope))
6825 }
6826 ResolvedItemName::Error => bail_internal!("should have been caught in name resolution"),
6827 }
6828 }
6829
6830 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6833 self.scx.humanize_sql_scalar_type(typ, postgres_compat)
6834 }
6835}
6836
6837#[derive(Debug, Clone)]
6839pub struct ExprContext<'a> {
6840 pub qcx: &'a QueryContext<'a>,
6841 pub name: &'a str,
6843 pub scope: &'a Scope,
6846 pub relation_type: &'a SqlRelationType,
6849 pub allow_aggregates: bool,
6851 pub allow_subqueries: bool,
6853 pub allow_parameters: bool,
6855 pub allow_windows: bool,
6857}
6858
6859impl CheckedRecursion for ExprContext<'_> {
6860 fn recursion_guard(&self) -> &RecursionGuard {
6861 &self.qcx.recursion_guard
6862 }
6863}
6864
6865impl<'a> ExprContext<'a> {
6866 pub fn catalog(&self) -> &dyn SessionCatalog {
6867 self.qcx.scx.catalog
6868 }
6869
6870 pub fn with_name(&self, name: &'a str) -> ExprContext<'a> {
6871 let mut ecx = self.clone();
6872 ecx.name = name;
6873 ecx
6874 }
6875
6876 pub fn column_type<E>(&self, expr: &E) -> E::Type
6877 where
6878 E: AbstractExpr,
6879 {
6880 expr.typ(
6881 &self.qcx.outer_relation_types,
6882 self.relation_type,
6883 &self.qcx.scx.param_types.borrow(),
6884 )
6885 }
6886
6887 pub fn scalar_type<E>(&self, expr: &E) -> <E::Type as AbstractColumnType>::AbstractScalarType
6888 where
6889 E: AbstractExpr,
6890 {
6891 self.column_type(expr).scalar_type()
6892 }
6893
6894 fn derived_query_context(&self) -> QueryContext<'_> {
6895 let mut scope = self.scope.clone();
6896 scope.lateral_barrier = true;
6897 self.qcx.derived_context(scope, self.relation_type.clone())
6898 }
6899
6900 pub fn require_feature_flag(&self, flag: &'static FeatureFlag) -> Result<(), PlanError> {
6901 self.qcx.scx.require_feature_flag(flag)
6902 }
6903
6904 pub fn param_types(&self) -> &RefCell<BTreeMap<usize, SqlScalarType>> {
6905 &self.qcx.scx.param_types
6906 }
6907
6908 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6911 self.qcx.scx.humanize_sql_scalar_type(typ, postgres_compat)
6912 }
6913
6914 pub fn intern(&self, item: &ScopeItem) -> Arc<str> {
6915 self.qcx.name_manager.borrow_mut().intern_scope_item(item)
6916 }
6917}
6918
6919#[derive(Debug, Clone)]
6925pub struct NameManager(BTreeSet<Arc<str>>);
6926
6927impl NameManager {
6928 pub fn new() -> Self {
6930 Self(BTreeSet::new())
6931 }
6932
6933 fn intern<S: AsRef<str>>(&mut self, s: S) -> Arc<str> {
6936 let s = s.as_ref();
6937 if let Some(interned) = self.0.get(s) {
6938 Arc::clone(interned)
6939 } else {
6940 let interned: Arc<str> = Arc::from(s);
6941 self.0.insert(Arc::clone(&interned));
6942 interned
6943 }
6944 }
6945
6946 pub fn intern_scope_item(&mut self, item: &ScopeItem) -> Arc<str> {
6949 self.intern(item.column_name.as_str())
6965 }
6966}
6967
6968#[cfg(test)]
6969mod test {
6970 use super::*;
6971
6972 #[mz_ore::test]
6977 pub fn test_name_manager_string_interning() {
6978 let mut nm = NameManager::new();
6979
6980 let orig_hi = "hi";
6981 let hi = nm.intern(orig_hi);
6982 let hello = nm.intern("hello");
6983
6984 assert_ne!(hi.as_ptr(), hello.as_ptr());
6985
6986 let hi2 = nm.intern("hi");
6988 assert_eq!(hi.as_ptr(), hi2.as_ptr());
6989
6990 let s = format!(
6992 "{}{}",
6993 hi.chars().nth(0).unwrap(),
6994 hi2.chars().nth(1).unwrap()
6995 );
6996 assert_ne!(orig_hi.as_ptr(), s.as_ptr());
6998
6999 let hi3 = nm.intern(s);
7000 assert_eq!(hi.as_ptr(), hi3.as_ptr());
7001 }
7002}