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::ShowCreateMetricSink(stmt) => to_hirscope(
2070 show::plan_show_create_metric_sink(qcx.scx, stmt.clone())?,
2071 show::describe_show_create_metric_sink(qcx.scx, stmt)?,
2072 ),
2073 ShowStatement::ShowCreateSource(stmt) => to_hirscope(
2074 show::plan_show_create_source(qcx.scx, stmt.clone())?,
2075 show::describe_show_create_source(qcx.scx, stmt)?,
2076 ),
2077 ShowStatement::ShowCreateTable(stmt) => to_hirscope(
2078 show::plan_show_create_table(qcx.scx, stmt.clone())?,
2079 show::describe_show_create_table(qcx.scx, stmt)?,
2080 ),
2081 ShowStatement::ShowCreateView(stmt) => to_hirscope(
2082 show::plan_show_create_view(qcx.scx, stmt.clone())?,
2083 show::describe_show_create_view(qcx.scx, stmt)?,
2084 ),
2085 ShowStatement::ShowCreateMaterializedView(stmt) => to_hirscope(
2086 show::plan_show_create_materialized_view(qcx.scx, stmt.clone())?,
2087 show::describe_show_create_materialized_view(qcx.scx, stmt)?,
2088 ),
2089 ShowStatement::ShowCreateType(stmt) => to_hirscope(
2090 show::plan_show_create_type(qcx.scx, stmt.clone())?,
2091 show::describe_show_create_type(qcx.scx, stmt)?,
2092 ),
2093 ShowStatement::ShowObjects(stmt) => {
2094 show::show_objects(qcx.scx, stmt)?.plan_hir(qcx)
2095 }
2096 ShowStatement::ShowVariable(_) => bail_unsupported!("SHOW variable in subqueries"),
2097 ShowStatement::InspectShard(_) => sql_bail!("unsupported INSPECT statement"),
2098 }
2099 }
2100 }
2101}
2102
2103fn plan_values(
2105 qcx: &QueryContext,
2106 values: &[Vec<Expr<Aug>>],
2107) -> Result<(HirRelationExpr, Scope), PlanError> {
2108 assert!(!values.is_empty());
2109
2110 let ecx = &ExprContext {
2111 qcx,
2112 name: "VALUES",
2113 scope: &Scope::empty(),
2114 relation_type: &SqlRelationType::empty(),
2115 allow_aggregates: false,
2116 allow_subqueries: true,
2117 allow_parameters: true,
2118 allow_windows: false,
2119 };
2120
2121 let ncols = values[0].len();
2122 let nrows = values.len();
2123
2124 let mut cols = vec![vec![]; ncols];
2127 for row in values {
2128 if row.len() != ncols {
2129 sql_bail!(
2130 "VALUES expression has varying number of columns: {} vs {}",
2131 row.len(),
2132 ncols
2133 );
2134 }
2135 for (i, v) in row.iter().enumerate() {
2136 cols[i].push(v);
2137 }
2138 }
2139
2140 let mut col_iters = Vec::with_capacity(ncols);
2142 let mut col_types = Vec::with_capacity(ncols);
2143 for col in &cols {
2144 let col = coerce_homogeneous_exprs(ecx, plan_exprs(ecx, col)?, None)?;
2145 let mut col_type = ecx.column_type(&col[0]);
2146 for val in &col[1..] {
2147 col_type = col_type.sql_union(&ecx.column_type(val))?; }
2149 col_types.push(col_type);
2150 col_iters.push(col.into_iter());
2151 }
2152
2153 let mut exprs = vec![];
2155 for _ in 0..nrows {
2156 for i in 0..ncols {
2157 exprs.push(col_iters[i].next().unwrap());
2158 }
2159 }
2160 let out = HirRelationExpr::CallTable {
2161 func: TableFunc::Wrap {
2162 width: ncols,
2163 types: col_types,
2164 },
2165 exprs,
2166 };
2167
2168 let mut scope = Scope::empty();
2170 for i in 0..ncols {
2171 let name = format!("column{}", i + 1);
2172 scope.items.push(ScopeItem::from_column_name(name));
2173 }
2174
2175 Ok((out, scope))
2176}
2177
2178fn plan_values_insert(
2188 qcx: &QueryContext,
2189 target_names: &[&ColumnName],
2190 target_types: &[&SqlScalarType],
2191 values: &[Vec<Expr<Aug>>],
2192) -> Result<HirRelationExpr, PlanError> {
2193 assert!(!values.is_empty());
2194
2195 if !values.iter().map(|row| row.len()).all_equal() {
2196 sql_bail!("VALUES lists must all be the same length");
2197 }
2198
2199 let ecx = &ExprContext {
2200 qcx,
2201 name: "VALUES",
2202 scope: &Scope::empty(),
2203 relation_type: &SqlRelationType::empty(),
2204 allow_aggregates: false,
2205 allow_subqueries: true,
2206 allow_parameters: true,
2207 allow_windows: false,
2208 };
2209
2210 let mut exprs = vec![];
2211 let mut types = vec![];
2212 for row in values {
2213 if row.len() > target_names.len() {
2214 sql_bail!("INSERT has more expressions than target columns");
2215 }
2216 for (column, val) in row.into_iter().enumerate() {
2217 let target_type = &target_types[column];
2218 let val = plan_expr(ecx, val)?;
2219 let val = typeconv::plan_coerce(ecx, val, target_type)?;
2220 let source_type = &ecx.scalar_type(&val);
2221 let val = match typeconv::plan_cast(ecx, CastContext::Assignment, val, target_type) {
2222 Ok(val) => val,
2223 Err(_) => sql_bail!(
2224 "column {} is of type {} but expression is of type {}",
2225 target_names[column].quoted(),
2226 qcx.humanize_sql_scalar_type(target_type, false),
2227 qcx.humanize_sql_scalar_type(source_type, false),
2228 ),
2229 };
2230 if column >= types.len() {
2231 types.push(ecx.column_type(&val));
2232 } else {
2233 types[column] = types[column].sql_union(&ecx.column_type(&val))?; }
2235 exprs.push(val);
2236 }
2237 }
2238
2239 Ok(HirRelationExpr::CallTable {
2240 func: TableFunc::Wrap {
2241 width: values[0].len(),
2242 types,
2243 },
2244 exprs,
2245 })
2246}
2247
2248fn plan_join_identity() -> (HirRelationExpr, Scope) {
2249 let typ = SqlRelationType::new(vec![]);
2250 let expr = HirRelationExpr::constant(vec![vec![]], typ);
2251 let scope = Scope::empty();
2252 (expr, scope)
2253}
2254
2255#[derive(Debug)]
2261struct SelectPlan {
2262 expr: HirRelationExpr,
2263 scope: Scope,
2264 order_by: Vec<ColumnOrder>,
2265 project: Vec<usize>,
2266}
2267
2268generate_extracted_config!(
2269 SelectOption,
2270 (ExpectedGroupSize, u64),
2271 (AggregateInputGroupSize, u64),
2272 (DistinctOnInputGroupSize, u64),
2273 (LimitInputGroupSize, u64)
2274);
2275
2276fn plan_select_from_where(
2294 qcx: &QueryContext,
2295 mut s: Select<Aug>,
2296 mut order_by_exprs: Vec<OrderByExpr<Aug>>,
2297) -> Result<SelectPlan, PlanError> {
2298 let select_option_extracted = SelectOptionExtracted::try_from(s.options.clone())?;
2305 let group_size_hints = GroupSizeHints::try_from(select_option_extracted)?;
2306
2307 let (mut relation_expr, mut from_scope) =
2309 s.from.iter().try_fold(plan_join_identity(), |l, twj| {
2310 let (left, left_scope) = l;
2311 plan_join(
2312 qcx,
2313 left,
2314 left_scope,
2315 &Join {
2316 relation: TableFactor::NestedJoin {
2317 join: Box::new(twj.clone()),
2318 alias: None,
2319 },
2320 join_operator: JoinOperator::CrossJoin,
2321 },
2322 )
2323 })?;
2324
2325 if let Some(selection) = &s.selection {
2327 let ecx = &ExprContext {
2328 qcx,
2329 name: "WHERE clause",
2330 scope: &from_scope,
2331 relation_type: &qcx.relation_type(&relation_expr),
2332 allow_aggregates: false,
2333 allow_subqueries: true,
2334 allow_parameters: true,
2335 allow_windows: false,
2336 };
2337 let expr = plan_expr(ecx, selection)
2338 .map_err(|e| sql_err!("WHERE clause error: {}", e))?
2339 .type_as(ecx, &SqlScalarType::Bool)?;
2340 relation_expr = relation_expr.filter(vec![expr]);
2341 }
2342
2343 let (aggregates, table_funcs) = {
2346 let mut visitor = AggregateTableFuncVisitor::new(qcx.scx);
2347 visitor.visit_select_mut(&mut s);
2348 for o in order_by_exprs.iter_mut() {
2349 visitor.visit_order_by_expr_mut(o);
2350 }
2351 visitor.into_result()?
2352 };
2353 let mut table_func_names: BTreeMap<String, Ident> = BTreeMap::new();
2354 let pre_table_funcs_arity = from_scope.len();
2361 let mut pre_table_funcs_relation = None;
2362 let mut table_funcs_deferred = false;
2363 if !table_funcs.is_empty() {
2364 let (expr, scope) = plan_scalar_table_funcs(
2365 qcx,
2366 &table_funcs,
2367 &mut table_func_names,
2368 &relation_expr,
2369 &from_scope,
2370 )?;
2371 if !aggregates.is_empty() || !s.group_by.is_empty() || s.having.is_some() {
2372 pre_table_funcs_relation = Some(relation_expr.clone());
2373 }
2374 relation_expr = relation_expr.join(expr, HirScalarExpr::literal_true(), JoinKind::Inner);
2375 from_scope = from_scope.product(scope)?;
2376 }
2377
2378 let projection = {
2380 let ecx = &ExprContext {
2381 qcx,
2382 name: "SELECT clause",
2383 scope: &from_scope,
2384 relation_type: &qcx.relation_type(&relation_expr),
2385 allow_aggregates: true,
2386 allow_subqueries: true,
2387 allow_parameters: true,
2388 allow_windows: true,
2389 };
2390 let mut out = vec![];
2391 for si in &s.projection {
2392 if *si == SelectItem::Wildcard && s.from.is_empty() {
2393 sql_bail!("SELECT * with no tables specified is not valid");
2394 }
2395 out.extend(expand_select_item(ecx, si, &table_func_names)?);
2396 }
2397 out
2398 };
2399
2400 let (mut group_scope, select_all_mapping) = {
2404 let ecx = &ExprContext {
2406 qcx,
2407 name: "GROUP BY clause",
2408 scope: &from_scope,
2409 relation_type: &qcx.relation_type(&relation_expr),
2410 allow_aggregates: false,
2411 allow_subqueries: true,
2412 allow_parameters: true,
2413 allow_windows: false,
2414 };
2415 let mut group_key = vec![];
2416 let mut group_exprs: BTreeMap<HirScalarExpr, ScopeItem> = BTreeMap::new();
2417 let mut group_hir_exprs = vec![];
2418 let mut group_scope = Scope::empty();
2419 let mut select_all_mapping = BTreeMap::new();
2420
2421 for group_expr in &s.group_by {
2422 let (group_expr, expr) = plan_group_by_expr(ecx, group_expr, &projection)?;
2423 let new_column = group_key.len();
2424
2425 if let Some(existing_scope_item) = group_exprs.get_mut(&expr) {
2436 if let Some(group_expr) = group_expr {
2440 existing_scope_item.exprs.insert(group_expr.clone());
2441 }
2442 continue;
2443 }
2444
2445 let mut scope_item = if let HirScalarExpr::Column(
2446 ColumnRef {
2447 level: 0,
2448 column: old_column,
2449 },
2450 _name,
2451 ) = &expr
2452 {
2453 select_all_mapping.insert(*old_column, new_column);
2459 let scope_item = ecx.scope.items[*old_column].clone();
2460 scope_item
2461 } else {
2462 ScopeItem::empty()
2463 };
2464
2465 if let Some(group_expr) = group_expr.cloned() {
2466 scope_item.exprs.insert(group_expr);
2467 }
2468
2469 group_key.push(from_scope.len() + group_exprs.len());
2470 group_hir_exprs.push(expr.clone());
2471 group_exprs.insert(expr, scope_item);
2472 }
2473
2474 assert_eq!(group_hir_exprs.len(), group_exprs.len());
2475 for expr in &group_hir_exprs {
2476 if let Some(scope_item) = group_exprs.remove(expr) {
2477 group_scope.items.push(scope_item);
2478 }
2479 }
2480
2481 let ecx = &ExprContext {
2483 qcx,
2484 name: "aggregate function",
2485 scope: &from_scope,
2486 relation_type: &qcx.relation_type(&relation_expr.clone().map(group_hir_exprs.clone())),
2487 allow_aggregates: false,
2488 allow_subqueries: true,
2489 allow_parameters: true,
2490 allow_windows: false,
2491 };
2492 let mut agg_exprs = vec![];
2493 for sql_function in aggregates {
2494 if sql_function.over.is_some() {
2495 unreachable!(
2496 "Window aggregate; AggregateTableFuncVisitor explicitly filters these out"
2497 );
2498 }
2499 agg_exprs.push(plan_aggregate_common(ecx, &sql_function)?);
2500 group_scope
2501 .items
2502 .push(ScopeItem::from_expr(Expr::Function(sql_function.clone())));
2503 }
2504 if !agg_exprs.is_empty() || !group_key.is_empty() || s.having.is_some() {
2505 if let Some(pre_relation_expr) = pre_table_funcs_relation.take() {
2509 let mut references_table_funcs = false;
2510 let mut check = |column: usize| {
2511 if column >= pre_table_funcs_arity {
2512 references_table_funcs = true;
2513 }
2514 };
2515 for expr in &group_hir_exprs {
2516 expr.visit_columns_referring_to_root_level(&mut check);
2517 }
2518 for agg_expr in &agg_exprs {
2519 agg_expr
2520 .expr
2521 .visit_columns_referring_to_root_level(&mut check);
2522 }
2523 if !references_table_funcs {
2524 relation_expr = pre_relation_expr;
2525 for (i, key) in group_key.iter_mut().enumerate() {
2528 *key = pre_table_funcs_arity + i;
2529 }
2530 table_funcs_deferred = true;
2531 }
2532 }
2533
2534 relation_expr = relation_expr.map(group_hir_exprs).reduce(
2536 group_key,
2537 agg_exprs,
2538 group_size_hints.aggregate_input_group_size,
2539 );
2540
2541 let ungrouped_arity = if table_funcs_deferred {
2549 pre_table_funcs_arity
2550 } else {
2551 from_scope.len()
2552 };
2553 for i in 0..ungrouped_arity {
2554 if !select_all_mapping.contains_key(&i) {
2555 let scope_item = &ecx.scope.items[i];
2556 group_scope.ungrouped_columns.push(ScopeUngroupedColumn {
2557 table_name: scope_item.table_name.clone(),
2558 column_name: scope_item.column_name.clone(),
2559 allow_unqualified_references: scope_item.allow_unqualified_references,
2560 });
2561 }
2562 }
2563
2564 (group_scope, select_all_mapping)
2565 } else {
2566 (
2568 from_scope.clone(),
2569 (0..from_scope.len()).map(|i| (i, i)).collect(),
2570 )
2571 }
2572 };
2573
2574 if let Some(ref having) = s.having {
2576 let ecx = &ExprContext {
2577 qcx,
2578 name: "HAVING clause",
2579 scope: &group_scope,
2580 relation_type: &qcx.relation_type(&relation_expr),
2581 allow_aggregates: true,
2582 allow_subqueries: true,
2583 allow_parameters: true,
2584 allow_windows: false,
2585 };
2586 let expr = plan_expr(ecx, having)?.type_as(ecx, &SqlScalarType::Bool)?;
2587 relation_expr = relation_expr.filter(vec![expr]);
2588 }
2589
2590 let window_funcs = {
2603 let mut visitor = WindowFuncCollector::default();
2604 visitor.visit_select(&s);
2608 for o in order_by_exprs.iter() {
2609 visitor.visit_order_by_expr(o);
2610 }
2611 visitor.into_result()
2612 };
2613 for window_func in window_funcs {
2614 let ecx = &ExprContext {
2615 qcx,
2616 name: "window function",
2617 scope: &group_scope,
2618 relation_type: &qcx.relation_type(&relation_expr),
2619 allow_aggregates: true,
2620 allow_subqueries: true,
2621 allow_parameters: true,
2622 allow_windows: true,
2623 };
2624 relation_expr = relation_expr.map(vec![plan_expr(ecx, &window_func)?.type_as_any(ecx)?]);
2625 group_scope.items.push(ScopeItem::from_expr(window_func));
2626 }
2627 if let Some(ref qualify) = s.qualify {
2634 let ecx = &ExprContext {
2635 qcx,
2636 name: "QUALIFY clause",
2637 scope: &group_scope,
2638 relation_type: &qcx.relation_type(&relation_expr),
2639 allow_aggregates: true,
2640 allow_subqueries: true,
2641 allow_parameters: true,
2642 allow_windows: true,
2643 };
2644 let expr = plan_expr(ecx, qualify)?.type_as(ecx, &SqlScalarType::Bool)?;
2645 relation_expr = relation_expr.filter(vec![expr]);
2646 }
2647
2648 if table_funcs_deferred {
2652 let (expr, scope) = plan_scalar_table_funcs(
2653 qcx,
2654 &table_funcs,
2655 &mut table_func_names,
2656 &relation_expr,
2657 &group_scope,
2658 )?;
2659 relation_expr = relation_expr.join(expr, HirScalarExpr::literal_true(), JoinKind::Inner);
2660 let ungrouped_columns = mem::take(&mut group_scope.ungrouped_columns);
2663 group_scope = group_scope.product(scope)?;
2664 group_scope.ungrouped_columns = ungrouped_columns;
2665 }
2666
2667 let output_columns = {
2669 let mut new_exprs = vec![];
2670 let mut new_type = qcx.relation_type(&relation_expr);
2671 let mut output_columns = vec![];
2672 for (select_item, column_name) in &projection {
2673 let ecx = &ExprContext {
2674 qcx,
2675 name: "SELECT clause",
2676 scope: &group_scope,
2677 relation_type: &new_type,
2678 allow_aggregates: true,
2679 allow_subqueries: true,
2680 allow_parameters: true,
2681 allow_windows: true,
2682 };
2683 let expr = match select_item {
2684 ExpandedSelectItem::InputOrdinal(i) => {
2685 if let Some(column) = select_all_mapping.get(i).copied() {
2686 HirScalarExpr::column(column)
2687 } else {
2688 return Err(PlanError::ungrouped_column(&from_scope.items[*i]));
2689 }
2690 }
2691 ExpandedSelectItem::Expr(expr) => plan_expr(ecx, expr)?.type_as_any(ecx)?,
2692 };
2693 if let HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) = expr {
2694 output_columns.push((column, column_name));
2696 } else {
2697 let typ = ecx.column_type(&expr);
2704 new_type.column_types.push(typ);
2705 new_exprs.push(expr);
2706 output_columns.push((group_scope.len(), column_name));
2707 group_scope
2708 .items
2709 .push(ScopeItem::from_expr(select_item.as_expr().cloned()));
2710 }
2711 }
2712 relation_expr = relation_expr.map(new_exprs);
2713 output_columns
2714 };
2715 let mut project_key: Vec<_> = output_columns.iter().map(|(i, _name)| *i).collect();
2716
2717 let order_by = {
2719 let relation_type = qcx.relation_type(&relation_expr);
2720 let (mut order_by, mut map_exprs) = plan_order_by_exprs(
2721 &ExprContext {
2722 qcx,
2723 name: "ORDER BY clause",
2724 scope: &group_scope,
2725 relation_type: &relation_type,
2726 allow_aggregates: true,
2727 allow_subqueries: true,
2728 allow_parameters: true,
2729 allow_windows: true,
2730 },
2731 &order_by_exprs,
2732 &output_columns,
2733 )?;
2734
2735 match s.distinct {
2736 None => relation_expr = relation_expr.map(map_exprs),
2737 Some(Distinct::EntireRow) => {
2738 if relation_type.arity() == 0 {
2739 sql_bail!("SELECT DISTINCT must have at least one column");
2740 }
2741 if !try_push_projection_order_by(
2745 &mut relation_expr,
2746 &mut project_key,
2747 &mut order_by,
2748 ) {
2749 sql_bail!(
2750 "for SELECT DISTINCT, ORDER BY expressions must appear in select list"
2751 );
2752 }
2753 assert!(map_exprs.is_empty());
2754 relation_expr = relation_expr.distinct();
2755 }
2756 Some(Distinct::On(exprs)) => {
2757 if table_funcs_deferred && !order_by_exprs.is_empty() {
2765 bail_unsupported!(
2766 "SELECT list table function with DISTINCT ON and ORDER BY over an aggregation"
2767 );
2768 }
2769
2770 let ecx = &ExprContext {
2771 qcx,
2772 name: "DISTINCT ON clause",
2773 scope: &group_scope,
2774 relation_type: &qcx.relation_type(&relation_expr),
2775 allow_aggregates: true,
2776 allow_subqueries: true,
2777 allow_parameters: true,
2778 allow_windows: true,
2779 };
2780
2781 let mut distinct_exprs = vec![];
2782 for expr in &exprs {
2783 let expr = plan_order_by_or_distinct_expr(ecx, expr, &output_columns)?;
2784 distinct_exprs.push(expr);
2785 }
2786
2787 let mut distinct_key = vec![];
2788
2789 let arity = relation_type.arity();
2799 for ord in order_by.iter().take(distinct_exprs.len()) {
2800 let mut expr = &HirScalarExpr::column(ord.column);
2803 if ord.column >= arity {
2804 expr = &map_exprs[ord.column - arity];
2805 };
2806 match distinct_exprs.iter().position(move |e| e == expr) {
2807 None => sql_bail!(
2808 "SELECT DISTINCT ON expressions must match initial ORDER BY expressions"
2809 ),
2810 Some(pos) => {
2811 distinct_exprs.remove(pos);
2812 }
2813 }
2814 distinct_key.push(ord.column);
2815 }
2816
2817 for expr in distinct_exprs {
2819 let column = match expr {
2822 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
2823 _ => {
2824 map_exprs.push(expr);
2825 arity + map_exprs.len() - 1
2826 }
2827 };
2828 distinct_key.push(column);
2829 }
2830
2831 let distinct_len = distinct_key.len();
2836 relation_expr = HirRelationExpr::top_k(
2837 relation_expr.map(map_exprs),
2838 distinct_key,
2839 order_by.iter().skip(distinct_len).cloned().collect(),
2840 Some(HirScalarExpr::literal(
2841 Datum::Int64(1),
2842 SqlScalarType::Int64,
2843 )),
2844 HirScalarExpr::literal(Datum::Int64(0), SqlScalarType::Int64),
2845 group_size_hints.distinct_on_input_group_size,
2846 );
2847 }
2848 }
2849
2850 order_by
2851 };
2852
2853 let scope = Scope::from_source(None, projection.into_iter().map(|(_expr, name)| name));
2858
2859 Ok(SelectPlan {
2860 expr: relation_expr,
2861 scope,
2862 order_by,
2863 project: project_key,
2864 })
2865}
2866
2867fn plan_scalar_table_funcs(
2868 qcx: &QueryContext,
2869 table_funcs: &BTreeMap<Function<Aug>, String>,
2870 table_func_names: &mut BTreeMap<String, Ident>,
2871 relation_expr: &HirRelationExpr,
2872 from_scope: &Scope,
2873) -> Result<(HirRelationExpr, Scope), PlanError> {
2874 let rows_from_qcx = qcx.derived_context(from_scope.clone(), qcx.relation_type(relation_expr));
2875
2876 for (table_func, id) in table_funcs.iter() {
2877 table_func_names.insert(
2878 id.clone(),
2879 Ident::new_unchecked(table_func.name.full_item_name().item.clone()),
2881 );
2882 }
2883 if table_funcs.len() == 1 {
2886 let (table_func, id) = table_funcs.iter().next().unwrap();
2887 let (expr, mut scope) =
2888 plan_solitary_table_function(&rows_from_qcx, table_func, None, false)?;
2889
2890 let num_cols = scope.len();
2892 for i in 0..scope.len() {
2893 scope.items[i].table_name = Some(PartialItemName {
2894 database: None,
2895 schema: None,
2896 item: id.clone(),
2897 });
2898 scope.items[i].from_single_column_function = num_cols == 1;
2899 scope.items[i].allow_unqualified_references = false;
2900 }
2901 return Ok((expr, scope));
2902 }
2903 if table_funcs.keys().any(is_repeat_row) {
2904 bail_unsupported!(format!(
2907 "{} in a SELECT clause with multiple table functions",
2908 REPEAT_ROW_NAME
2909 ));
2910 }
2911 let (expr, mut scope, num_cols) =
2913 plan_rows_from_internal(&rows_from_qcx, table_funcs.keys(), None)?;
2914
2915 let mut i = 0;
2917 for (id, num_cols) in table_funcs.values().zip_eq(num_cols) {
2918 for _ in 0..num_cols {
2919 scope.items[i].table_name = Some(PartialItemName {
2920 database: None,
2921 schema: None,
2922 item: id.clone(),
2923 });
2924 scope.items[i].from_single_column_function = num_cols == 1;
2925 scope.items[i].allow_unqualified_references = false;
2926 i += 1;
2927 }
2928 scope.items[i].table_name = Some(PartialItemName {
2932 database: None,
2933 schema: None,
2934 item: id.clone(),
2935 });
2936 scope.items[i].is_exists_column_for_a_table_function_that_was_in_the_target_list = true;
2937 scope.items[i].allow_unqualified_references = false;
2938 i += 1;
2939 }
2940 scope.items[i].allow_unqualified_references = false;
2942 Ok((expr, scope))
2943}
2944
2945fn plan_group_by_expr<'a>(
2952 ecx: &ExprContext,
2953 group_expr: &'a Expr<Aug>,
2954 projection: &'a [(ExpandedSelectItem, ColumnName)],
2955) -> Result<(Option<&'a Expr<Aug>>, HirScalarExpr), PlanError> {
2956 let plan_projection = |column: usize| match &projection[column].0 {
2957 ExpandedSelectItem::InputOrdinal(column) => Ok((None, HirScalarExpr::column(*column))),
2958 ExpandedSelectItem::Expr(expr) => {
2959 Ok((Some(expr.as_ref()), plan_expr(ecx, expr)?.type_as_any(ecx)?))
2960 }
2961 };
2962
2963 if let Some(column) = check_col_index(ecx.name, group_expr, projection.len())? {
2966 return plan_projection(column);
2967 }
2968
2969 match group_expr {
2973 Expr::Identifier(names) => match plan_identifier(ecx, names) {
2974 Err(PlanError::UnknownColumn {
2975 table: None,
2976 column,
2977 similar,
2978 }) => {
2979 let mut iter = projection.iter().map(|(_expr, name)| name);
2982 if let Some(i) = iter.position(|n| *n == column) {
2983 if iter.any(|n| *n == column) {
2984 Err(PlanError::AmbiguousColumn(column))
2985 } else {
2986 plan_projection(i)
2987 }
2988 } else {
2989 Err(PlanError::UnknownColumn {
2992 table: None,
2993 column,
2994 similar,
2995 })
2996 }
2997 }
2998 res => Ok((Some(group_expr), res?)),
2999 },
3000 _ => Ok((
3001 Some(group_expr),
3002 plan_expr(ecx, group_expr)?.type_as_any(ecx)?,
3003 )),
3004 }
3005}
3006
3007pub(crate) fn plan_order_by_exprs(
3015 ecx: &ExprContext,
3016 order_by_exprs: &[OrderByExpr<Aug>],
3017 output_columns: &[(usize, &ColumnName)],
3018) -> Result<(Vec<ColumnOrder>, Vec<HirScalarExpr>), PlanError> {
3019 let mut order_by = vec![];
3020 let mut map_exprs = vec![];
3021 for obe in order_by_exprs {
3022 let expr = plan_order_by_or_distinct_expr(ecx, &obe.expr, output_columns)?;
3023 let column = match expr {
3026 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
3027 _ => {
3028 map_exprs.push(expr);
3029 ecx.relation_type.arity() + map_exprs.len() - 1
3030 }
3031 };
3032 order_by.push(resolve_desc_and_nulls_last(obe, column));
3033 }
3034 Ok((order_by, map_exprs))
3035}
3036
3037fn plan_order_by_or_distinct_expr(
3055 ecx: &ExprContext,
3056 expr: &Expr<Aug>,
3057 output_columns: &[(usize, &ColumnName)],
3058) -> Result<HirScalarExpr, PlanError> {
3059 if let Some(i) = check_col_index(ecx.name, expr, output_columns.len())? {
3060 return Ok(HirScalarExpr::column(output_columns[i].0));
3061 }
3062
3063 if let Expr::Identifier(names) = expr {
3064 if let [name] = &names[..] {
3065 let name = normalize::column_name(name.clone());
3066 let mut iter = output_columns.iter().filter(|(_, n)| **n == name);
3067 if let Some((i, _)) = iter.next() {
3068 match iter.next() {
3069 Some((i2, _)) if i != i2 => return Err(PlanError::AmbiguousColumn(name)),
3073 _ => return Ok(HirScalarExpr::column(*i)),
3074 }
3075 }
3076 }
3077 }
3078
3079 plan_expr(ecx, expr)?.type_as_any(ecx)
3080}
3081
3082fn plan_table_with_joins(
3083 qcx: &QueryContext,
3084 table_with_joins: &TableWithJoins<Aug>,
3085) -> Result<(HirRelationExpr, Scope), PlanError> {
3086 let (mut expr, mut scope) = plan_table_factor(qcx, &table_with_joins.relation)?;
3087 for join in &table_with_joins.joins {
3088 let (new_expr, new_scope) = plan_join(qcx, expr, scope, join)?;
3089 expr = new_expr;
3090 scope = new_scope;
3091 }
3092 Ok((expr, scope))
3093}
3094
3095fn plan_table_factor(
3096 qcx: &QueryContext,
3097 table_factor: &TableFactor<Aug>,
3098) -> Result<(HirRelationExpr, Scope), PlanError> {
3099 match table_factor {
3100 TableFactor::Table { name, alias } => {
3101 let (expr, scope) = qcx.resolve_table_name(name.clone())?;
3102 let scope = plan_table_alias(scope, alias.as_ref())?;
3103 Ok((expr, scope))
3104 }
3105
3106 TableFactor::Function {
3107 function,
3108 alias,
3109 with_ordinality,
3110 } => plan_solitary_table_function(qcx, function, alias.as_ref(), *with_ordinality),
3111
3112 TableFactor::RowsFrom {
3113 functions,
3114 alias,
3115 with_ordinality,
3116 } => plan_rows_from(qcx, functions, alias.as_ref(), *with_ordinality),
3117
3118 TableFactor::Derived {
3119 lateral,
3120 subquery,
3121 alias,
3122 } => {
3123 let mut qcx = (*qcx).clone();
3124 if !lateral {
3125 for scope in &mut qcx.outer_scopes {
3129 if scope.lateral_barrier {
3130 break;
3131 }
3132 scope.items.clear();
3133 }
3134 }
3135 qcx.outer_scopes[0].lateral_barrier = true;
3136 let (expr, scope) = plan_nested_query(&mut qcx, subquery)?;
3137 let scope = plan_table_alias(scope, alias.as_ref())?;
3138 Ok((expr, scope))
3139 }
3140
3141 TableFactor::NestedJoin { join, alias } => {
3142 let (expr, scope) = plan_table_with_joins(qcx, join)?;
3143 let scope = plan_table_alias(scope, alias.as_ref())?;
3144 Ok((expr, scope))
3145 }
3146 }
3147}
3148
3149fn plan_rows_from(
3191 qcx: &QueryContext,
3192 functions: &[Function<Aug>],
3193 alias: Option<&TableAlias>,
3194 with_ordinality: bool,
3195) -> Result<(HirRelationExpr, Scope), PlanError> {
3196 if functions.iter().any(is_repeat_row) {
3198 bail_unsupported!(format!("{} in ROWS FROM", REPEAT_ROW_NAME));
3202 }
3203
3204 if let [function] = functions {
3207 return plan_solitary_table_function(qcx, function, alias, with_ordinality);
3208 }
3209
3210 let (expr, mut scope, num_cols) = plan_rows_from_internal(
3214 qcx,
3215 functions,
3216 Some(functions[0].name.full_item_name().clone()),
3217 )?;
3218
3219 let mut columns = Vec::new();
3221 let mut offset = 0;
3222 for (idx, cols) in num_cols.into_iter().enumerate() {
3224 for i in 0..cols {
3225 columns.push(offset + i);
3226 }
3227 offset += cols + 1;
3228
3229 scope.items.remove(offset - idx - 1);
3232 }
3233
3234 if with_ordinality {
3237 columns.push(offset);
3238 } else {
3239 scope.items.pop();
3240 }
3241
3242 let expr = expr.project(columns);
3243
3244 let scope = plan_table_alias(scope, alias)?;
3245 Ok((expr, scope))
3246}
3247
3248fn is_repeat_row(f: &Function<Aug>) -> bool {
3249 f.name.full_name_str().as_str() == format!("{}.{}", MZ_CATALOG_SCHEMA, REPEAT_ROW_NAME)
3250}
3251
3252fn plan_rows_from_internal<'a>(
3275 qcx: &QueryContext,
3276 functions: impl IntoIterator<Item = &'a Function<Aug>>,
3277 table_name: Option<FullItemName>,
3278) -> Result<(HirRelationExpr, Scope, Vec<usize>), PlanError> {
3279 let mut functions = functions.into_iter();
3280 let mut num_cols = Vec::new();
3281
3282 let (mut left_expr, mut left_scope) =
3286 plan_table_function_internal(qcx, functions.next().unwrap(), true, table_name.clone())?;
3287 num_cols.push(left_scope.len() - 1);
3288 left_expr = left_expr.map(vec![HirScalarExpr::column(left_scope.len() - 1)]);
3290 left_scope
3291 .items
3292 .push(ScopeItem::from_column_name(ORDINALITY_COL_NAME));
3293
3294 for function in functions {
3295 let qcx = qcx.empty_derived_context();
3297 let (right_expr, mut right_scope) =
3298 plan_table_function_internal(&qcx, function, true, table_name.clone())?;
3299 num_cols.push(right_scope.len() - 1);
3300 let left_col = left_scope.len() - 1;
3301 let right_col = left_scope.len() + right_scope.len() - 1;
3302 let on = HirScalarExpr::call_binary(
3303 HirScalarExpr::column(left_col),
3304 HirScalarExpr::column(right_col),
3305 expr_func::Eq,
3306 );
3307 left_expr = left_expr
3308 .join(right_expr, on, JoinKind::FullOuter)
3309 .map(vec![HirScalarExpr::call_variadic(
3310 Coalesce,
3311 vec![
3312 HirScalarExpr::column(left_col),
3313 HirScalarExpr::column(right_col),
3314 ],
3315 )]);
3316
3317 left_expr = left_expr.project(
3320 (0..left_col) .chain(left_col + 1..right_col + 2) .collect(),
3323 );
3324 right_scope.items.push(left_scope.items.pop().unwrap());
3326
3327 left_scope.items.extend(right_scope.items);
3328 }
3329
3330 Ok((left_expr, left_scope, num_cols))
3331}
3332
3333fn plan_solitary_table_function(
3337 qcx: &QueryContext,
3338 function: &Function<Aug>,
3339 alias: Option<&TableAlias>,
3340 with_ordinality: bool,
3341) -> Result<(HirRelationExpr, Scope), PlanError> {
3342 let (expr, mut scope) = plan_table_function_internal(qcx, function, with_ordinality, None)?;
3343
3344 let single_column_function = scope.len() == 1 + if with_ordinality { 1 } else { 0 };
3345 if single_column_function {
3346 let item = &mut scope.items[0];
3347
3348 item.from_single_column_function = true;
3351
3352 if let Some(alias) = alias {
3367 if let ScopeItem {
3368 table_name: Some(table_name),
3369 column_name,
3370 ..
3371 } = item
3372 {
3373 if table_name.item.as_str() == column_name.as_str() {
3374 *column_name = normalize::column_name(alias.name.clone());
3375 }
3376 }
3377 }
3378 }
3379
3380 let scope = plan_table_alias(scope, alias)?;
3381 Ok((expr, scope))
3382}
3383
3384fn plan_table_function_internal(
3389 qcx: &QueryContext,
3390 Function {
3391 name,
3392 args,
3393 filter,
3394 over,
3395 distinct,
3396 }: &Function<Aug>,
3397 with_ordinality: bool,
3398 table_name: Option<FullItemName>,
3399) -> Result<(HirRelationExpr, Scope), PlanError> {
3400 if filter.is_some() {
3405 sql_bail!("FILTER is not allowed for table functions in FROM");
3406 }
3407 if over.is_some() {
3408 sql_bail!("OVER is not allowed for table functions in FROM");
3409 }
3410 if *distinct {
3411 sql_bail!("DISTINCT is not allowed for table functions in FROM");
3412 }
3413
3414 let ecx = &ExprContext {
3415 qcx,
3416 name: "table function arguments",
3417 scope: &Scope::empty(),
3418 relation_type: &SqlRelationType::empty(),
3419 allow_aggregates: false,
3420 allow_subqueries: true,
3421 allow_parameters: true,
3422 allow_windows: false,
3423 };
3424
3425 let scalar_args = match args {
3426 FunctionArgs::Star => sql_bail!("{} does not accept * as an argument", name),
3427 FunctionArgs::Args { args, order_by } => {
3428 if !order_by.is_empty() {
3429 sql_bail!(
3430 "ORDER BY specified, but {} is not an aggregate function",
3431 name
3432 );
3433 }
3434 plan_exprs(ecx, args)?
3435 }
3436 };
3437
3438 let table_name = match table_name {
3439 Some(table_name) => table_name.item,
3440 None => name.full_item_name().item.clone(),
3441 };
3442
3443 let scope_name = Some(PartialItemName {
3444 database: None,
3445 schema: None,
3446 item: table_name,
3447 });
3448
3449 let (expr, mut scope) = match resolve_func(ecx, name, args)? {
3450 Func::Table(impls) => {
3451 let tf = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3452 let scope = Scope::from_source(scope_name.clone(), tf.column_names);
3453 let expr = match tf.imp {
3454 TableFuncImpl::CallTable { mut func, exprs } => {
3455 if with_ordinality {
3456 func = TableFunc::with_ordinality(func.clone()).ok_or(
3457 PlanError::Unsupported {
3458 feature: format!("WITH ORDINALITY on {}", func),
3459 discussion_no: None,
3460 },
3461 )?;
3462 }
3463 HirRelationExpr::CallTable { func, exprs }
3464 }
3465 TableFuncImpl::Expr(expr) => {
3466 if !with_ordinality {
3467 expr
3468 } else {
3469 if qcx
3473 .scx
3474 .is_feature_flag_enabled(&ENABLE_WITH_ORDINALITY_LEGACY_FALLBACK)
3475 {
3476 tracing::error!(
3480 %name,
3481 "Using the legacy WITH ORDINALITY / ROWS FROM implementation for a table function",
3482 );
3483 expr.map(vec![HirScalarExpr::windowing(WindowExpr {
3484 func: WindowExprType::Scalar(ScalarWindowExpr {
3485 func: ScalarWindowFunc::RowNumber,
3486 order_by: vec![],
3487 }),
3488 partition_by: vec![],
3489 order_by: vec![],
3490 })])
3491 } else {
3492 bail_unsupported!(format!(
3493 "WITH ORDINALITY or ROWS FROM with {}",
3494 name
3495 ));
3496 }
3497 }
3498 }
3499 };
3500 (expr, scope)
3501 }
3502 Func::Scalar(impls) => {
3503 let expr = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3504 let output = expr.typ(
3505 &qcx.outer_relation_types,
3506 &SqlRelationType::new(vec![]),
3507 &qcx.scx.param_types.borrow(),
3508 );
3509
3510 let relation = SqlRelationType::new(vec![output]);
3511
3512 let function_ident = Ident::new(name.full_item_name().item.clone())?;
3513 let column_name = normalize::column_name(function_ident);
3514 let name = column_name.to_string();
3515
3516 let scope = Scope::from_source(scope_name.clone(), vec![column_name]);
3517
3518 let mut func = TableFunc::TabletizedScalar { relation, name };
3519 if with_ordinality {
3520 func = TableFunc::with_ordinality(func.clone()).ok_or(PlanError::Unsupported {
3521 feature: format!("WITH ORDINALITY on {}", func),
3522 discussion_no: None,
3523 })?;
3524 }
3525 (
3526 HirRelationExpr::CallTable {
3527 func,
3528 exprs: vec![expr],
3529 },
3530 scope,
3531 )
3532 }
3533 o => sql_bail!(
3534 "{} functions are not supported in functions in FROM",
3535 o.class()
3536 ),
3537 };
3538
3539 if with_ordinality {
3540 scope
3541 .items
3542 .push(ScopeItem::from_name(scope_name, "ordinality"));
3543 }
3544
3545 Ok((expr, scope))
3546}
3547
3548fn plan_table_alias(mut scope: Scope, alias: Option<&TableAlias>) -> Result<Scope, PlanError> {
3549 if let Some(TableAlias {
3550 name,
3551 columns,
3552 strict,
3553 }) = alias
3554 {
3555 if (columns.len() > scope.items.len()) || (*strict && columns.len() != scope.items.len()) {
3556 sql_bail!(
3557 "{} has {} columns available but {} columns specified",
3558 name,
3559 scope.items.len(),
3560 columns.len()
3561 );
3562 }
3563
3564 let table_name = normalize::ident(name.to_owned());
3565 for (i, item) in scope.items.iter_mut().enumerate() {
3566 item.table_name = if item.allow_unqualified_references {
3567 Some(PartialItemName {
3568 database: None,
3569 schema: None,
3570 item: table_name.clone(),
3571 })
3572 } else {
3573 None
3607 };
3608 item.column_name = columns
3609 .get(i)
3610 .map(|a| normalize::column_name(a.clone()))
3611 .unwrap_or_else(|| item.column_name.clone());
3612 }
3613 }
3614 Ok(scope)
3615}
3616
3617fn invent_column_name(
3621 ecx: &ExprContext,
3622 expr: &Expr<Aug>,
3623 table_func_names: &BTreeMap<String, Ident>,
3624) -> Result<Option<ColumnName>, PlanError> {
3625 #[derive(Debug)]
3632 enum NameQuality {
3633 Low,
3634 High,
3635 }
3636
3637 fn invent(
3638 ecx: &ExprContext,
3639 expr: &Expr<Aug>,
3640 table_func_names: &BTreeMap<String, Ident>,
3641 ) -> Result<Option<(ColumnName, NameQuality)>, PlanError> {
3642 Ok(match expr {
3643 Expr::Identifier(names) => {
3644 if let [name] = names.as_slice() {
3645 if let Some(table_func_name) = table_func_names.get(name.as_str()) {
3646 return Ok(Some((
3647 normalize::column_name(table_func_name.clone()),
3648 NameQuality::High,
3649 )));
3650 }
3651 }
3652 names
3653 .last()
3654 .map(|n| (normalize::column_name(n.clone()), NameQuality::High))
3655 }
3656 Expr::Value(v) => match v {
3657 Value::Boolean(_) => Some(("bool".into(), NameQuality::High)),
3660 Value::Interval(_) => Some(("interval".into(), NameQuality::High)),
3661 _ => None,
3662 },
3663 Expr::Function(func) => {
3664 let (schema, item) = match &func.name {
3665 ResolvedItemName::Item {
3666 qualifiers,
3667 full_name,
3668 ..
3669 } => (&qualifiers.schema_spec, full_name.item.clone()),
3670 _ => {
3673 bail_internal!("function name did not resolve to an item: {:?}", func.name)
3674 }
3675 };
3676
3677 if schema == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_internal_schema_id())
3678 || schema
3679 == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_unsafe_schema_id())
3680 {
3681 None
3682 } else {
3683 Some((item.into(), NameQuality::High))
3684 }
3685 }
3686 Expr::HomogenizingFunction { function, .. } => Some((
3687 function.to_string().to_lowercase().into(),
3688 NameQuality::High,
3689 )),
3690 Expr::NullIf { .. } => Some(("nullif".into(), NameQuality::High)),
3691 Expr::Array { .. } => Some(("array".into(), NameQuality::High)),
3692 Expr::List { .. } => Some(("list".into(), NameQuality::High)),
3693 Expr::Map { .. } | Expr::MapSubquery(_) => Some(("map".into(), NameQuality::High)),
3694 Expr::Cast { expr, data_type } => match invent(ecx, expr, table_func_names)? {
3695 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3696 _ => Some((data_type.unqualified_item_name().into(), NameQuality::Low)),
3697 },
3698 Expr::Case { else_result, .. } => {
3699 let inner = match else_result.as_ref() {
3700 Some(else_result) => invent(ecx, else_result, table_func_names)?,
3701 None => None,
3702 };
3703 match inner {
3704 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3705 _ => Some(("case".into(), NameQuality::Low)),
3706 }
3707 }
3708 Expr::FieldAccess { field, .. } => {
3709 Some((normalize::column_name(field.clone()), NameQuality::High))
3710 }
3711 Expr::Exists { .. } => Some(("exists".into(), NameQuality::High)),
3712 Expr::Subscript { expr, .. } => invent(ecx, expr, table_func_names)?,
3713 Expr::Subquery(query) | Expr::ListSubquery(query) | Expr::ArraySubquery(query) => {
3714 let Ok((_expr, scope)) = plan_nested_query(&mut ecx.derived_query_context(), query)
3722 else {
3723 return Ok(None);
3724 };
3725 scope
3726 .items
3727 .first()
3728 .map(|name| (name.column_name.clone(), NameQuality::High))
3729 }
3730 Expr::Row { .. } => Some(("row".into(), NameQuality::High)),
3731 _ => None,
3732 })
3733 }
3734
3735 Ok(invent(ecx, expr, table_func_names)?.map(|(name, _quality)| name))
3736}
3737
3738#[derive(Debug)]
3739enum ExpandedSelectItem<'a> {
3740 InputOrdinal(usize),
3741 Expr(Cow<'a, Expr<Aug>>),
3742}
3743
3744impl ExpandedSelectItem<'_> {
3745 fn as_expr(&self) -> Option<&Expr<Aug>> {
3746 match self {
3747 ExpandedSelectItem::InputOrdinal(_) => None,
3748 ExpandedSelectItem::Expr(expr) => Some(expr),
3749 }
3750 }
3751}
3752
3753fn expand_select_item<'a>(
3754 ecx: &ExprContext,
3755 s: &'a SelectItem<Aug>,
3756 table_func_names: &BTreeMap<String, Ident>,
3757) -> Result<Vec<(ExpandedSelectItem<'a>, ColumnName)>, PlanError> {
3758 match s {
3759 SelectItem::Expr {
3760 expr: Expr::QualifiedWildcard(table_name),
3761 alias: _,
3762 } => {
3763 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3764 let table_name =
3765 normalize::unresolved_item_name(UnresolvedItemName(table_name.clone()))?;
3766 let out: Vec<_> = ecx
3767 .scope
3768 .items
3769 .iter()
3770 .enumerate()
3771 .filter(|(_i, item)| item.is_from_table(&table_name))
3772 .map(|(i, item)| {
3773 let name = item.column_name.clone();
3774 (ExpandedSelectItem::InputOrdinal(i), name)
3775 })
3776 .collect();
3777 if out.is_empty() {
3778 sql_bail!("no table named '{}' in scope", table_name);
3779 }
3780 Ok(out)
3781 }
3782 SelectItem::Expr {
3783 expr: Expr::WildcardAccess(sql_expr),
3784 alias: _,
3785 } => {
3786 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3787 let expr = plan_expr(ecx, sql_expr)?.type_as_any(ecx)?;
3793 let fields = match ecx.scalar_type(&expr) {
3794 SqlScalarType::Record { fields, .. } => fields,
3795 ty => sql_bail!(
3796 "type {} is not composite",
3797 ecx.humanize_sql_scalar_type(&ty, false)
3798 ),
3799 };
3800 let mut skip_cols: BTreeSet<ColumnName> = BTreeSet::new();
3801 if let Expr::Identifier(ident) = sql_expr.as_ref() {
3802 if let [name] = ident.as_slice() {
3803 if let Ok(items) = ecx.scope.items_from_table(
3804 &[],
3805 &PartialItemName {
3806 database: None,
3807 schema: None,
3808 item: name.as_str().to_string(),
3809 },
3810 ) {
3811 for (_, item) in items {
3812 if item
3813 .is_exists_column_for_a_table_function_that_was_in_the_target_list
3814 {
3815 skip_cols.insert(item.column_name.clone());
3816 }
3817 }
3818 }
3819 }
3820 }
3821 let items = fields
3822 .iter()
3823 .filter_map(|(name, _ty)| {
3824 if skip_cols.contains(name) {
3825 None
3826 } else {
3827 let item = ExpandedSelectItem::Expr(Cow::Owned(Expr::FieldAccess {
3828 expr: sql_expr.clone(),
3829 field: name.clone().into(),
3830 }));
3831 Some((item, name.clone()))
3832 }
3833 })
3834 .collect();
3835 Ok(items)
3836 }
3837 SelectItem::Wildcard => {
3838 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3839 let items: Vec<_> = ecx
3840 .scope
3841 .items
3842 .iter()
3843 .enumerate()
3844 .filter(|(_i, item)| item.allow_unqualified_references)
3845 .map(|(i, item)| {
3846 let name = item.column_name.clone();
3847 (ExpandedSelectItem::InputOrdinal(i), name)
3848 })
3849 .collect();
3850
3851 Ok(items)
3852 }
3853 SelectItem::Expr { expr, alias } => {
3854 let name = match alias.clone().map(normalize::column_name) {
3855 Some(name) => name,
3856 None => invent_column_name(ecx, expr, table_func_names)?
3857 .unwrap_or_else(|| UNKNOWN_COLUMN_NAME.into()),
3858 };
3859 Ok(vec![(ExpandedSelectItem::Expr(Cow::Borrowed(expr)), name)])
3860 }
3861 }
3862}
3863
3864fn plan_join(
3865 left_qcx: &QueryContext,
3866 left: HirRelationExpr,
3867 left_scope: Scope,
3868 join: &Join<Aug>,
3869) -> Result<(HirRelationExpr, Scope), PlanError> {
3870 const ON_TRUE: JoinConstraint<Aug> = JoinConstraint::On(Expr::Value(Value::Boolean(true)));
3871 let (kind, constraint) = match &join.join_operator {
3872 JoinOperator::CrossJoin => (JoinKind::Inner, &ON_TRUE),
3873 JoinOperator::Inner(constraint) => (JoinKind::Inner, constraint),
3874 JoinOperator::LeftOuter(constraint) => (JoinKind::LeftOuter, constraint),
3875 JoinOperator::RightOuter(constraint) => (JoinKind::RightOuter, constraint),
3876 JoinOperator::FullOuter(constraint) => (JoinKind::FullOuter, constraint),
3877 };
3878
3879 let mut right_qcx = left_qcx.derived_context(left_scope.clone(), left_qcx.relation_type(&left));
3880 if !kind.can_be_correlated() {
3881 for item in &mut right_qcx.outer_scopes[0].items {
3882 item.error_if_referenced =
3887 Some(|table, column| PlanError::WrongJoinTypeForLateralColumn {
3888 table: table.cloned(),
3889 column: column.clone(),
3890 });
3891 }
3892 }
3893 let (right, right_scope) = plan_table_factor(&right_qcx, &join.relation)?;
3894
3895 let (expr, scope) = match constraint {
3896 JoinConstraint::On(expr) => {
3897 let product_scope = left_scope.product(right_scope)?;
3898 let ecx = &ExprContext {
3899 qcx: left_qcx,
3900 name: "ON clause",
3901 scope: &product_scope,
3902 relation_type: &SqlRelationType::new(
3903 left_qcx
3904 .relation_type(&left)
3905 .column_types
3906 .into_iter()
3907 .chain(right_qcx.relation_type(&right).column_types)
3908 .collect(),
3909 ),
3910 allow_aggregates: false,
3911 allow_subqueries: true,
3912 allow_parameters: true,
3913 allow_windows: false,
3914 };
3915 let on = plan_expr(ecx, expr)?.type_as(ecx, &SqlScalarType::Bool)?;
3916 let joined = left.join(right, on, kind);
3917 (joined, product_scope)
3918 }
3919 JoinConstraint::Using { columns, alias } => {
3920 let column_names = columns
3921 .iter()
3922 .map(|ident| normalize::column_name(ident.clone()))
3923 .collect::<Vec<_>>();
3924
3925 plan_using_constraint(
3926 &column_names,
3927 left_qcx,
3928 left,
3929 left_scope,
3930 &right_qcx,
3931 right,
3932 right_scope,
3933 kind,
3934 alias.as_ref(),
3935 )?
3936 }
3937 JoinConstraint::Natural => {
3938 *left_qcx.scx.ambiguous_columns.borrow_mut() = true;
3941 *right_qcx.scx.ambiguous_columns.borrow_mut() = true;
3942 let left_column_names = left_scope.column_names();
3943 let right_column_names: BTreeSet<_> = right_scope.column_names().collect();
3944 let column_names: Vec<_> = left_column_names
3945 .filter(|col| right_column_names.contains(col))
3946 .cloned()
3947 .collect();
3948 plan_using_constraint(
3949 &column_names,
3950 left_qcx,
3951 left,
3952 left_scope,
3953 &right_qcx,
3954 right,
3955 right_scope,
3956 kind,
3957 None,
3958 )?
3959 }
3960 };
3961 Ok((expr, scope))
3962}
3963
3964#[allow(clippy::too_many_arguments)]
3966fn plan_using_constraint(
3967 column_names: &[ColumnName],
3968 left_qcx: &QueryContext,
3969 left: HirRelationExpr,
3970 left_scope: Scope,
3971 right_qcx: &QueryContext,
3972 right: HirRelationExpr,
3973 right_scope: Scope,
3974 kind: JoinKind,
3975 alias: Option<&Ident>,
3976) -> Result<(HirRelationExpr, Scope), PlanError> {
3977 let mut both_scope = left_scope.clone().product(right_scope.clone())?;
3978
3979 let mut unique_column_names = BTreeSet::new();
3982 for c in column_names {
3983 if !unique_column_names.insert(c) {
3984 return Err(PlanError::Unsupported {
3985 feature: format!(
3986 "column name {} appears more than once in USING clause",
3987 c.quoted()
3988 ),
3989 discussion_no: None,
3990 });
3991 }
3992 }
3993
3994 let alias_item_name = alias.map(|alias| PartialItemName {
3995 database: None,
3996 schema: None,
3997 item: alias.clone().to_string(),
3998 });
3999
4000 if let Some(alias_item_name) = &alias_item_name {
4001 for partial_item_name in both_scope.table_names() {
4002 if partial_item_name.matches(alias_item_name) {
4003 sql_bail!(
4004 "table name \"{}\" specified more than once",
4005 alias_item_name
4006 )
4007 }
4008 }
4009 }
4010
4011 let ecx = &ExprContext {
4012 qcx: right_qcx,
4013 name: "USING clause",
4014 scope: &both_scope,
4015 relation_type: &SqlRelationType::new(
4016 left_qcx
4017 .relation_type(&left)
4018 .column_types
4019 .into_iter()
4020 .chain(right_qcx.relation_type(&right).column_types)
4021 .collect(),
4022 ),
4023 allow_aggregates: false,
4024 allow_subqueries: false,
4025 allow_parameters: false,
4026 allow_windows: false,
4027 };
4028
4029 let mut join_exprs = vec![];
4030 let mut map_exprs = vec![];
4031 let mut new_items = vec![];
4032 let mut join_cols = vec![];
4033 let mut hidden_cols = vec![];
4034
4035 for column_name in column_names {
4036 let (lhs, lhs_name) = left_scope.resolve_using_column(
4038 column_name,
4039 JoinSide::Left,
4040 &mut left_qcx.name_manager.borrow_mut(),
4041 )?;
4042 let (mut rhs, rhs_name) = right_scope.resolve_using_column(
4043 column_name,
4044 JoinSide::Right,
4045 &mut right_qcx.name_manager.borrow_mut(),
4046 )?;
4047
4048 rhs.column += left_scope.len();
4050
4051 let mut exprs = coerce_homogeneous_exprs(
4053 &ecx.with_name(&format!(
4054 "NATURAL/USING join column {}",
4055 column_name.quoted()
4056 )),
4057 vec![
4058 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
4059 lhs,
4060 Arc::clone(&lhs_name),
4061 )),
4062 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
4063 rhs,
4064 Arc::clone(&rhs_name),
4065 )),
4066 ],
4067 None,
4068 )?;
4069 let (expr1, expr2) = (exprs.remove(0), exprs.remove(0));
4070
4071 match kind {
4072 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
4073 join_cols.push(lhs.column);
4074 hidden_cols.push(rhs.column);
4075 }
4076 JoinKind::RightOuter => {
4077 join_cols.push(rhs.column);
4078 hidden_cols.push(lhs.column);
4079 }
4080 JoinKind::FullOuter => {
4081 join_cols.push(both_scope.items.len() + map_exprs.len());
4084 hidden_cols.push(lhs.column);
4085 hidden_cols.push(rhs.column);
4086 map_exprs.push(HirScalarExpr::call_variadic(
4087 Coalesce,
4088 vec![expr1.clone(), expr2.clone()],
4089 ));
4090 new_items.push(ScopeItem::from_column_name(column_name));
4091 }
4092 }
4093
4094 if alias_item_name.is_some() {
4099 let new_item_col = both_scope.items.len() + new_items.len();
4100 join_cols.push(new_item_col);
4101 hidden_cols.push(new_item_col);
4102
4103 new_items.push(ScopeItem::from_name(
4104 alias_item_name.clone(),
4105 column_name.clone().to_string(),
4106 ));
4107
4108 let alias_expr = match kind {
4115 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
4116 HirScalarExpr::named_column(lhs, Arc::clone(&lhs_name))
4117 }
4118 JoinKind::RightOuter => HirScalarExpr::named_column(rhs, Arc::clone(&rhs_name)),
4119 JoinKind::FullOuter => {
4120 HirScalarExpr::call_variadic(Coalesce, vec![expr1.clone(), expr2.clone()])
4121 }
4122 };
4123 map_exprs.push(alias_expr);
4124 }
4125
4126 join_exprs.push(expr1.call_binary(expr2, expr_func::Eq));
4127 }
4128 both_scope.items.extend(new_items);
4129
4130 for c in hidden_cols {
4134 both_scope.items[c].allow_unqualified_references = false;
4135 }
4136
4137 let project_key = join_cols
4139 .into_iter()
4140 .chain(0..both_scope.items.len())
4141 .unique()
4142 .collect::<Vec<_>>();
4143
4144 both_scope = both_scope.project(&project_key);
4145
4146 let on = HirScalarExpr::variadic_and(join_exprs);
4147
4148 let both = left
4149 .join(right, on, kind)
4150 .map(map_exprs)
4151 .project(project_key);
4152 Ok((both, both_scope))
4153}
4154
4155pub fn plan_expr<'a>(
4156 ecx: &'a ExprContext,
4157 e: &Expr<Aug>,
4158) -> Result<CoercibleScalarExpr, PlanError> {
4159 ecx.checked_recur(|ecx| plan_expr_inner(ecx, e))
4160}
4161
4162fn plan_expr_inner<'a>(
4163 ecx: &'a ExprContext,
4164 e: &Expr<Aug>,
4165) -> Result<CoercibleScalarExpr, PlanError> {
4166 if let Some((i, item)) = ecx.scope.resolve_expr(e) {
4167 return Ok(HirScalarExpr::named_column(
4169 i,
4170 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
4171 )
4172 .into());
4173 }
4174
4175 match e {
4176 Expr::Identifier(names) | Expr::QualifiedWildcard(names) => {
4178 Ok(plan_identifier(ecx, names)?.into())
4179 }
4180
4181 Expr::Value(val) => plan_literal(val),
4183 Expr::Parameter(n) => plan_parameter(ecx, *n),
4184 Expr::Array(exprs) => plan_array(ecx, exprs, None),
4185 Expr::List(exprs) => plan_list(ecx, exprs, None),
4186 Expr::Map(exprs) => plan_map(ecx, exprs, None),
4187 Expr::Row { exprs } => plan_row(ecx, exprs),
4188
4189 Expr::Op { op, expr1, expr2 } => {
4191 Ok(plan_op(ecx, normalize::op(op)?, expr1, expr2.as_deref())?.into())
4192 }
4193 Expr::Cast { expr, data_type } => plan_cast(ecx, expr, data_type),
4194 Expr::Function(func) => Ok(plan_function(ecx, func)?.into()),
4195
4196 Expr::Not { expr } => plan_not(ecx, expr),
4198 Expr::And { left, right } => plan_and(ecx, left, right),
4199 Expr::Or { left, right } => plan_or(ecx, left, right),
4200 Expr::IsExpr {
4201 expr,
4202 construct,
4203 negated,
4204 } => Ok(plan_is_expr(ecx, expr, construct, *negated)?.into()),
4205 Expr::Case {
4206 operand,
4207 conditions,
4208 results,
4209 else_result,
4210 } => Ok(plan_case(ecx, operand, conditions, results, else_result)?.into()),
4211 Expr::HomogenizingFunction { function, exprs } => {
4212 plan_homogenizing_function(ecx, function, exprs)
4213 }
4214 Expr::NullIf { l_expr, r_expr } => Ok(plan_case(
4215 ecx,
4216 &None,
4217 &[l_expr.clone().equals(*r_expr.clone())],
4218 &[Expr::null()],
4219 &Some(Box::new(*l_expr.clone())),
4220 )?
4221 .into()),
4222 Expr::FieldAccess { expr, field } => plan_field_access(ecx, expr, field),
4223 Expr::WildcardAccess(expr) => plan_expr(ecx, expr),
4224 Expr::Subscript { expr, positions } => plan_subscript(ecx, expr, positions),
4225 Expr::Like {
4226 expr,
4227 pattern,
4228 escape,
4229 case_insensitive,
4230 negated,
4231 } => Ok(plan_like(
4232 ecx,
4233 expr,
4234 pattern,
4235 escape.as_deref(),
4236 *case_insensitive,
4237 *negated,
4238 )?
4239 .into()),
4240
4241 Expr::InList {
4242 expr,
4243 list,
4244 negated,
4245 } => plan_in_list(ecx, expr, list, negated),
4246
4247 Expr::Exists(query) => plan_exists(ecx, query),
4249 Expr::Subquery(query) => plan_subquery(ecx, query),
4250 Expr::ListSubquery(query) => plan_list_subquery(ecx, query),
4251 Expr::MapSubquery(query) => plan_map_subquery(ecx, query),
4252 Expr::ArraySubquery(query) => plan_array_subquery(ecx, query),
4253 Expr::Collate { expr, collation } => plan_collate(ecx, expr, collation),
4254 Expr::Nested(_) => bail_internal!("Expr::Nested should have been desugared"),
4255 Expr::InSubquery { .. } => {
4256 bail_internal!("Expr::InSubquery should have been desugared")
4257 }
4258 Expr::AnyExpr { .. } => {
4259 bail_internal!("Expr::AnyExpr should have been desugared")
4260 }
4261 Expr::AllExpr { .. } => {
4262 bail_internal!("Expr::AllExpr should have been desugared")
4263 }
4264 Expr::AnySubquery { .. } => {
4265 bail_internal!("Expr::AnySubquery should have been desugared")
4266 }
4267 Expr::AllSubquery { .. } => {
4268 bail_internal!("Expr::AllSubquery should have been desugared")
4269 }
4270 Expr::Between { .. } => {
4271 bail_internal!("Expr::Between should have been desugared")
4272 }
4273 }
4274}
4275
4276fn plan_parameter(ecx: &ExprContext, n: usize) -> Result<CoercibleScalarExpr, PlanError> {
4277 if !ecx.allow_parameters {
4278 return Err(PlanError::UnknownParameter(n));
4282 }
4283 if n == 0 || n > 65536 {
4284 return Err(PlanError::UnknownParameter(n));
4285 }
4286 if ecx.param_types().borrow().contains_key(&n) {
4287 Ok(HirScalarExpr::parameter(n).into())
4288 } else {
4289 Ok(CoercibleScalarExpr::Parameter(n))
4290 }
4291}
4292
4293fn plan_row(ecx: &ExprContext, exprs: &[Expr<Aug>]) -> Result<CoercibleScalarExpr, PlanError> {
4294 let mut out = vec![];
4295 for e in exprs {
4296 out.push(plan_expr(ecx, e)?);
4297 }
4298 Ok(CoercibleScalarExpr::LiteralRecord(out))
4299}
4300
4301fn plan_cast(
4302 ecx: &ExprContext,
4303 expr: &Expr<Aug>,
4304 data_type: &ResolvedDataType,
4305) -> Result<CoercibleScalarExpr, PlanError> {
4306 let to_scalar_type = scalar_type_from_sql(ecx.qcx.scx, data_type)?;
4307 let expr = match expr {
4308 Expr::Array(exprs) => plan_array(ecx, exprs, Some(&to_scalar_type))?,
4317 Expr::List(exprs) => plan_list(ecx, exprs, Some(&to_scalar_type))?,
4318 Expr::Map(exprs) => plan_map(ecx, exprs, Some(&to_scalar_type))?,
4319 _ => plan_expr(ecx, expr)?,
4320 };
4321 let ecx = &ecx.with_name("CAST");
4322 let expr = typeconv::plan_coerce(ecx, expr, &to_scalar_type)?;
4323 let expr = typeconv::plan_cast(ecx, CastContext::Explicit, expr, &to_scalar_type)?;
4324 Ok(expr.into())
4325}
4326
4327fn plan_not(ecx: &ExprContext, expr: &Expr<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4328 let ecx = ecx.with_name("NOT argument");
4329 Ok(plan_expr(&ecx, expr)?
4330 .type_as(&ecx, &SqlScalarType::Bool)?
4331 .call_unary(UnaryFunc::Not(expr_func::Not))
4332 .into())
4333}
4334
4335fn plan_and(
4336 ecx: &ExprContext,
4337 left: &Expr<Aug>,
4338 right: &Expr<Aug>,
4339) -> Result<CoercibleScalarExpr, PlanError> {
4340 let ecx = ecx.with_name("AND argument");
4341 Ok(HirScalarExpr::variadic_and(vec![
4342 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4343 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4344 ])
4345 .into())
4346}
4347
4348fn plan_or(
4349 ecx: &ExprContext,
4350 left: &Expr<Aug>,
4351 right: &Expr<Aug>,
4352) -> Result<CoercibleScalarExpr, PlanError> {
4353 let ecx = ecx.with_name("OR argument");
4354 Ok(HirScalarExpr::variadic_or(vec![
4355 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4356 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4357 ])
4358 .into())
4359}
4360
4361fn plan_in_list(
4362 ecx: &ExprContext,
4363 lhs: &Expr<Aug>,
4364 list: &Vec<Expr<Aug>>,
4365 negated: &bool,
4366) -> Result<CoercibleScalarExpr, PlanError> {
4367 let ecx = ecx.with_name("IN list");
4368 let or = HirScalarExpr::variadic_or(
4369 list.into_iter()
4370 .map(|e| {
4371 let eq = lhs.clone().equals(e.clone());
4372 plan_expr(&ecx, &eq)?.type_as(&ecx, &SqlScalarType::Bool)
4373 })
4374 .collect::<Result<Vec<HirScalarExpr>, PlanError>>()?,
4375 );
4376 Ok(if *negated {
4377 or.call_unary(UnaryFunc::Not(expr_func::Not))
4378 } else {
4379 or
4380 }
4381 .into())
4382}
4383
4384fn plan_homogenizing_function(
4385 ecx: &ExprContext,
4386 function: &HomogenizingFunction,
4387 exprs: &[Expr<Aug>],
4388) -> Result<CoercibleScalarExpr, PlanError> {
4389 assert!(!exprs.is_empty()); let expr = HirScalarExpr::call_variadic(
4391 match function {
4392 HomogenizingFunction::Coalesce => VariadicFunc::from(Coalesce),
4393 HomogenizingFunction::Greatest => VariadicFunc::from(Greatest),
4394 HomogenizingFunction::Least => VariadicFunc::from(Least),
4395 },
4396 coerce_homogeneous_exprs(
4397 &ecx.with_name(&function.to_string().to_lowercase()),
4398 plan_exprs(ecx, exprs)?,
4399 None,
4400 )?,
4401 );
4402 Ok(expr.into())
4403}
4404
4405fn plan_field_access(
4406 ecx: &ExprContext,
4407 expr: &Expr<Aug>,
4408 field: &Ident,
4409) -> Result<CoercibleScalarExpr, PlanError> {
4410 let field = normalize::column_name(field.clone());
4411 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4412 let ty = ecx.scalar_type(&expr);
4413 let i = match &ty {
4414 SqlScalarType::Record { fields, .. } => {
4415 fields.iter().position(|(name, _ty)| *name == field)
4416 }
4417 ty => sql_bail!(
4418 "column notation applied to type {}, which is not a composite type",
4419 ecx.humanize_sql_scalar_type(ty, false)
4420 ),
4421 };
4422 match i {
4423 None => sql_bail!(
4424 "field {} not found in data type {}",
4425 field,
4426 ecx.humanize_sql_scalar_type(&ty, false)
4427 ),
4428 Some(i) => Ok(expr
4429 .call_unary(UnaryFunc::RecordGet(expr_func::RecordGet(i)))
4430 .into()),
4431 }
4432}
4433
4434fn plan_subscript(
4435 ecx: &ExprContext,
4436 expr: &Expr<Aug>,
4437 positions: &[SubscriptPosition<Aug>],
4438) -> Result<CoercibleScalarExpr, PlanError> {
4439 assert!(
4440 !positions.is_empty(),
4441 "subscript expression must contain at least one position"
4442 );
4443
4444 let ecx = &ecx.with_name("subscripting");
4445 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4446 let ty = ecx.scalar_type(&expr);
4447 match &ty {
4448 SqlScalarType::Array(..) | SqlScalarType::Int2Vector => plan_subscript_array(
4449 ecx,
4450 expr,
4451 positions,
4452 if ty == SqlScalarType::Int2Vector {
4456 1
4457 } else {
4458 0
4459 },
4460 ),
4461 SqlScalarType::Jsonb => plan_subscript_jsonb(ecx, expr, positions),
4462 SqlScalarType::List { element_type, .. } => {
4463 let elem_type_name = ecx.humanize_sql_scalar_type(element_type, false);
4465 let n_layers = ty.unwrap_list_n_layers();
4466 plan_subscript_list(ecx, expr, positions, n_layers, &elem_type_name)
4467 }
4468 ty => sql_bail!(
4469 "cannot subscript type {}",
4470 ecx.humanize_sql_scalar_type(ty, false)
4471 ),
4472 }
4473}
4474
4475fn extract_scalar_subscript_from_positions<'a>(
4479 positions: &'a [SubscriptPosition<Aug>],
4480 expr_type_name: &str,
4481) -> Result<Vec<&'a Expr<Aug>>, PlanError> {
4482 let mut scalar_subscripts = Vec::with_capacity(positions.len());
4483 for p in positions {
4484 if p.explicit_slice {
4485 sql_bail!("{} subscript does not support slices", expr_type_name);
4486 }
4487 assert!(
4488 p.end.is_none(),
4489 "index-appearing subscripts cannot have end value"
4490 );
4491 scalar_subscripts.push(p.start.as_ref().expect("has start if not slice"));
4492 }
4493 Ok(scalar_subscripts)
4494}
4495
4496fn plan_subscript_array(
4497 ecx: &ExprContext,
4498 expr: HirScalarExpr,
4499 positions: &[SubscriptPosition<Aug>],
4500 offset: i64,
4501) -> Result<CoercibleScalarExpr, PlanError> {
4502 let mut exprs = Vec::with_capacity(positions.len() + 1);
4503 exprs.push(expr);
4504
4505 let indexes = extract_scalar_subscript_from_positions(positions, "array")?;
4508
4509 for i in indexes {
4510 exprs.push(plan_expr(ecx, i)?.cast_to(
4511 ecx,
4512 CastContext::Explicit,
4513 &SqlScalarType::Int64,
4514 )?);
4515 }
4516
4517 Ok(HirScalarExpr::call_variadic(ArrayIndex { offset }, exprs).into())
4518}
4519
4520fn plan_subscript_list(
4521 ecx: &ExprContext,
4522 mut expr: HirScalarExpr,
4523 positions: &[SubscriptPosition<Aug>],
4524 mut remaining_layers: usize,
4525 elem_type_name: &str,
4526) -> Result<CoercibleScalarExpr, PlanError> {
4527 let mut i = 0;
4528
4529 while i < positions.len() {
4530 let j = positions[i..]
4532 .iter()
4533 .position(|p| p.explicit_slice)
4534 .unwrap_or(positions.len() - i);
4535 if j != 0 {
4536 let indexes = extract_scalar_subscript_from_positions(&positions[i..i + j], "")?;
4537 let (n, e) = plan_index_list(
4538 ecx,
4539 expr,
4540 indexes.as_slice(),
4541 remaining_layers,
4542 elem_type_name,
4543 )?;
4544 remaining_layers = n;
4545 expr = e;
4546 i += j;
4547 }
4548
4549 let j = positions[i..]
4551 .iter()
4552 .position(|p| !p.explicit_slice)
4553 .unwrap_or(positions.len() - i);
4554 if j != 0 {
4555 expr = plan_slice_list(
4556 ecx,
4557 expr,
4558 &positions[i..i + j],
4559 remaining_layers,
4560 elem_type_name,
4561 )?;
4562 i += j;
4563 }
4564 }
4565
4566 Ok(expr.into())
4567}
4568
4569fn plan_index_list(
4570 ecx: &ExprContext,
4571 expr: HirScalarExpr,
4572 indexes: &[&Expr<Aug>],
4573 n_layers: usize,
4574 elem_type_name: &str,
4575) -> Result<(usize, HirScalarExpr), PlanError> {
4576 let depth = indexes.len();
4577
4578 if depth > n_layers {
4579 if n_layers == 0 {
4580 sql_bail!("cannot subscript type {}", elem_type_name)
4581 } else {
4582 sql_bail!(
4583 "cannot index into {} layers; list only has {} layer{}",
4584 depth,
4585 n_layers,
4586 if n_layers == 1 { "" } else { "s" }
4587 )
4588 }
4589 }
4590
4591 let mut exprs = Vec::with_capacity(depth + 1);
4592 exprs.push(expr);
4593
4594 for i in indexes {
4595 exprs.push(plan_expr(ecx, i)?.cast_to(
4596 ecx,
4597 CastContext::Explicit,
4598 &SqlScalarType::Int64,
4599 )?);
4600 }
4601
4602 Ok((
4603 n_layers - depth,
4604 HirScalarExpr::call_variadic(ListIndex, exprs),
4605 ))
4606}
4607
4608fn plan_slice_list(
4609 ecx: &ExprContext,
4610 expr: HirScalarExpr,
4611 slices: &[SubscriptPosition<Aug>],
4612 n_layers: usize,
4613 elem_type_name: &str,
4614) -> Result<HirScalarExpr, PlanError> {
4615 if n_layers == 0 {
4616 sql_bail!("cannot subscript type {}", elem_type_name)
4617 }
4618
4619 let mut exprs = Vec::with_capacity(slices.len() + 1);
4621 exprs.push(expr);
4622 let extract_position_or_default = |position, default| -> Result<HirScalarExpr, PlanError> {
4624 Ok(match position {
4625 Some(p) => {
4626 plan_expr(ecx, p)?.cast_to(ecx, CastContext::Explicit, &SqlScalarType::Int64)?
4627 }
4628 None => HirScalarExpr::literal(Datum::Int64(default), SqlScalarType::Int64),
4629 })
4630 };
4631 for p in slices {
4632 let start = extract_position_or_default(p.start.as_ref(), 1)?;
4633 let end = extract_position_or_default(p.end.as_ref(), i64::MAX - 1)?;
4634 exprs.push(start);
4635 exprs.push(end);
4636 }
4637
4638 Ok(HirScalarExpr::call_variadic(ListSliceLinear, exprs))
4639}
4640
4641fn plan_like(
4642 ecx: &ExprContext,
4643 expr: &Expr<Aug>,
4644 pattern: &Expr<Aug>,
4645 escape: Option<&Expr<Aug>>,
4646 case_insensitive: bool,
4647 not: bool,
4648) -> Result<HirScalarExpr, PlanError> {
4649 use CastContext::Implicit;
4650 let ecx = ecx.with_name("LIKE argument");
4651 let expr = plan_expr(&ecx, expr)?;
4652 let haystack = match ecx.scalar_type(&expr) {
4653 CoercibleScalarType::Coerced(ref ty @ SqlScalarType::Char { length }) => expr
4654 .type_as(&ecx, ty)?
4655 .call_unary(UnaryFunc::PadChar(expr_func::PadChar { length })),
4656 _ => expr.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4657 };
4658 let mut pattern = plan_expr(&ecx, pattern)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?;
4659 if let Some(escape) = escape {
4660 pattern = pattern.call_binary(
4661 plan_expr(&ecx, escape)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4662 expr_func::LikeEscape,
4663 );
4664 }
4665 let func: BinaryFunc = if case_insensitive {
4666 expr_func::IsLikeMatchCaseInsensitive.into()
4667 } else {
4668 expr_func::IsLikeMatchCaseSensitive.into()
4669 };
4670 let like = haystack.call_binary(pattern, func);
4671 if not {
4672 Ok(like.call_unary(UnaryFunc::Not(expr_func::Not)))
4673 } else {
4674 Ok(like)
4675 }
4676}
4677
4678fn plan_subscript_jsonb(
4679 ecx: &ExprContext,
4680 expr: HirScalarExpr,
4681 positions: &[SubscriptPosition<Aug>],
4682) -> Result<CoercibleScalarExpr, PlanError> {
4683 use CastContext::Implicit;
4684 use SqlScalarType::{Int64, String};
4685
4686 let subscripts = extract_scalar_subscript_from_positions(positions, "jsonb")?;
4689
4690 let mut exprs = Vec::with_capacity(subscripts.len());
4691 for s in subscripts {
4692 let subscript = plan_expr(ecx, s)?;
4693 let subscript = if let Ok(subscript) = subscript.clone().cast_to(ecx, Implicit, &String) {
4694 subscript
4695 } else if let Ok(subscript) = subscript.cast_to(ecx, Implicit, &Int64) {
4696 typeconv::to_string(ecx, subscript)?
4700 } else {
4701 sql_bail!("jsonb subscript type must be coercible to integer or text");
4702 };
4703 exprs.push(subscript);
4704 }
4705
4706 let expr = expr.call_binary(
4709 HirScalarExpr::call_variadic(
4710 ArrayCreate {
4711 elem_type: SqlScalarType::String,
4712 },
4713 exprs,
4714 ),
4715 expr_func::JsonbGetPath,
4716 );
4717 Ok(expr.into())
4718}
4719
4720fn plan_exists(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4721 if !ecx.allow_subqueries {
4722 sql_bail!("{} does not allow subqueries", ecx.name)
4723 }
4724 let mut qcx = ecx.derived_query_context();
4725 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4726 Ok(expr.exists().into())
4727}
4728
4729fn plan_subquery(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4730 if !ecx.allow_subqueries {
4731 sql_bail!("{} does not allow subqueries", ecx.name)
4732 }
4733 let mut qcx = ecx.derived_query_context();
4734 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4735 let column_types = qcx.relation_type(&expr).column_types;
4736 if column_types.len() != 1 {
4737 sql_bail!(
4738 "Expected subselect to return 1 column, got {} columns",
4739 column_types.len()
4740 );
4741 }
4742 Ok(expr.select().into())
4743}
4744
4745fn plan_list_subquery(
4746 ecx: &ExprContext,
4747 query: &Query<Aug>,
4748) -> Result<CoercibleScalarExpr, PlanError> {
4749 plan_vector_like_subquery(
4750 ecx,
4751 query,
4752 |_| false,
4753 |elem_type| ListCreate { elem_type }.into(),
4754 |order_by| AggregateFunc::ListConcat { order_by },
4755 expr_func::ListListConcat.into(),
4756 |elem_type| {
4757 HirScalarExpr::literal(
4758 Datum::empty_list(),
4759 SqlScalarType::List {
4760 element_type: Box::new(elem_type),
4761 custom_id: None,
4762 },
4763 )
4764 },
4765 "list",
4766 )
4767}
4768
4769fn plan_array_subquery(
4770 ecx: &ExprContext,
4771 query: &Query<Aug>,
4772) -> Result<CoercibleScalarExpr, PlanError> {
4773 plan_vector_like_subquery(
4774 ecx,
4775 query,
4776 |elem_type| {
4777 matches!(
4778 elem_type,
4779 SqlScalarType::Char { .. }
4780 | SqlScalarType::Array { .. }
4781 | SqlScalarType::List { .. }
4782 | SqlScalarType::Map { .. }
4783 )
4784 },
4785 |elem_type| ArrayCreate { elem_type }.into(),
4786 |order_by| AggregateFunc::ArrayConcat { order_by },
4787 expr_func::ArrayArrayConcat.into(),
4788 |elem_type| {
4789 HirScalarExpr::literal(
4790 Datum::empty_array(),
4791 SqlScalarType::Array(Box::new(elem_type)),
4792 )
4793 },
4794 "[]",
4795 )
4796}
4797
4798fn plan_vector_like_subquery<F1, F2, F3, F4>(
4800 ecx: &ExprContext,
4801 query: &Query<Aug>,
4802 is_unsupported_type: F1,
4803 vector_create: F2,
4804 aggregate_concat: F3,
4805 binary_concat: BinaryFunc,
4806 empty_literal: F4,
4807 vector_type_string: &str,
4808) -> Result<CoercibleScalarExpr, PlanError>
4809where
4810 F1: Fn(&SqlScalarType) -> bool,
4811 F2: Fn(SqlScalarType) -> VariadicFunc,
4812 F3: Fn(Vec<ColumnOrder>) -> AggregateFunc,
4813 F4: Fn(SqlScalarType) -> HirScalarExpr,
4814{
4815 if !ecx.allow_subqueries {
4816 sql_bail!("{} does not allow subqueries", ecx.name)
4817 }
4818
4819 let mut qcx = ecx.derived_query_context();
4820 let mut planned_query = plan_query(&mut qcx, query)?;
4821 if planned_query.limit.is_some()
4822 || !planned_query
4823 .offset
4824 .clone()
4825 .try_into_literal_int64()
4826 .is_ok_and(|offset| offset == 0)
4827 {
4828 planned_query.expr = HirRelationExpr::top_k(
4829 planned_query.expr,
4830 vec![],
4831 planned_query.order_by.clone(),
4832 planned_query.limit,
4833 planned_query.offset,
4834 planned_query.group_size_hints.limit_input_group_size,
4835 );
4836 }
4837
4838 if planned_query.project.len() != 1 {
4839 sql_bail!(
4840 "Expected subselect to return 1 column, got {} columns",
4841 planned_query.project.len()
4842 );
4843 }
4844
4845 let project_column = *planned_query.project.get(0).unwrap();
4846 let elem_type = qcx
4847 .relation_type(&planned_query.expr)
4848 .column_types
4849 .get(project_column)
4850 .cloned()
4851 .unwrap()
4852 .scalar_type();
4853
4854 if is_unsupported_type(&elem_type) {
4855 bail_unsupported!(format!(
4856 "cannot build array from subquery because return type {}{}",
4857 ecx.humanize_sql_scalar_type(&elem_type, false),
4858 vector_type_string
4859 ));
4860 }
4861
4862 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4865 vector_create(elem_type.clone()),
4866 vec![HirScalarExpr::column(project_column)],
4867 ))
4868 .chain(
4869 planned_query
4870 .order_by
4871 .iter()
4872 .map(|co| HirScalarExpr::column(co.column)),
4873 )
4874 .collect();
4875
4876 let aggregation_projection = vec![0];
4880 let aggregation_order_by = planned_query
4881 .order_by
4882 .into_iter()
4883 .enumerate()
4884 .map(|(i, order)| ColumnOrder { column: i, ..order })
4885 .collect();
4886
4887 let reduced_expr = planned_query
4888 .expr
4889 .reduce(
4890 vec![],
4891 vec![AggregateExpr {
4892 func: aggregate_concat(aggregation_order_by),
4893 expr: Box::new(HirScalarExpr::call_variadic(
4894 RecordCreate {
4895 field_names: iter::repeat(ColumnName::from(""))
4896 .take(aggregation_exprs.len())
4897 .collect(),
4898 },
4899 aggregation_exprs,
4900 )),
4901 distinct: false,
4902 }],
4903 None,
4904 )
4905 .project(aggregation_projection);
4906
4907 Ok(reduced_expr
4909 .select()
4910 .call_binary(empty_literal(elem_type), binary_concat)
4911 .into())
4912}
4913
4914fn plan_map_subquery(
4915 ecx: &ExprContext,
4916 query: &Query<Aug>,
4917) -> Result<CoercibleScalarExpr, PlanError> {
4918 if !ecx.allow_subqueries {
4919 sql_bail!("{} does not allow subqueries", ecx.name)
4920 }
4921
4922 let mut qcx = ecx.derived_query_context();
4923 let mut query = plan_query(&mut qcx, query)?;
4924 if query.limit.is_some()
4925 || !query
4926 .offset
4927 .clone()
4928 .try_into_literal_int64()
4929 .is_ok_and(|offset| offset == 0)
4930 {
4931 query.expr = HirRelationExpr::top_k(
4932 query.expr,
4933 vec![],
4934 query.order_by.clone(),
4935 query.limit,
4936 query.offset,
4937 query.group_size_hints.limit_input_group_size,
4938 );
4939 }
4940 if query.project.len() != 2 {
4941 sql_bail!(
4942 "expected map subquery to return 2 columns, got {} columns",
4943 query.project.len()
4944 );
4945 }
4946
4947 let query_types = qcx.relation_type(&query.expr).column_types;
4948 let key_column = query.project[0];
4949 let key_type = query_types[key_column].clone().scalar_type();
4950 let value_column = query.project[1];
4951 let value_type = query_types[value_column].clone().scalar_type();
4952
4953 if key_type != SqlScalarType::String {
4954 sql_bail!("cannot build map from subquery because first column is not of type text");
4955 }
4956
4957 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4958 RecordCreate {
4959 field_names: vec![ColumnName::from("key"), ColumnName::from("value")],
4960 },
4961 vec![
4962 HirScalarExpr::column(key_column),
4963 HirScalarExpr::column(value_column),
4964 ],
4965 ))
4966 .chain(
4967 query
4968 .order_by
4969 .iter()
4970 .map(|co| HirScalarExpr::column(co.column)),
4971 )
4972 .collect();
4973
4974 let expr = query
4975 .expr
4976 .reduce(
4977 vec![],
4978 vec![AggregateExpr {
4979 func: AggregateFunc::MapAgg {
4980 order_by: query
4981 .order_by
4982 .into_iter()
4983 .enumerate()
4984 .map(|(i, order)| ColumnOrder { column: i, ..order })
4985 .collect(),
4986 value_type: value_type.clone(),
4987 },
4988 expr: Box::new(HirScalarExpr::call_variadic(
4989 RecordCreate {
4990 field_names: iter::repeat(ColumnName::from(""))
4991 .take(aggregation_exprs.len())
4992 .collect(),
4993 },
4994 aggregation_exprs,
4995 )),
4996 distinct: false,
4997 }],
4998 None,
4999 )
5000 .project(vec![0]);
5001
5002 let expr = HirScalarExpr::call_variadic(
5004 Coalesce,
5005 vec![
5006 expr.select(),
5007 HirScalarExpr::literal(
5008 Datum::empty_map(),
5009 SqlScalarType::Map {
5010 value_type: Box::new(value_type),
5011 custom_id: None,
5012 },
5013 ),
5014 ],
5015 );
5016
5017 Ok(expr.into())
5018}
5019
5020fn plan_collate(
5021 ecx: &ExprContext,
5022 expr: &Expr<Aug>,
5023 collation: &UnresolvedItemName,
5024) -> Result<CoercibleScalarExpr, PlanError> {
5025 if collation.0.len() == 2
5026 && collation.0[0] == ident!(mz_repr::namespaces::PG_CATALOG_SCHEMA)
5027 && collation.0[1] == ident!("default")
5028 {
5029 plan_expr(ecx, expr)
5030 } else {
5031 bail_unsupported!("COLLATE");
5032 }
5033}
5034
5035fn plan_exprs<E>(ecx: &ExprContext, exprs: &[E]) -> Result<Vec<CoercibleScalarExpr>, PlanError>
5042where
5043 E: std::borrow::Borrow<Expr<Aug>>,
5044{
5045 let mut out = vec![];
5046 for expr in exprs {
5047 out.push(plan_expr(ecx, expr.borrow())?);
5048 }
5049 Ok(out)
5050}
5051
5052fn plan_array(
5054 ecx: &ExprContext,
5055 exprs: &[Expr<Aug>],
5056 type_hint: Option<&SqlScalarType>,
5057) -> Result<CoercibleScalarExpr, PlanError> {
5058 let mut out = vec![];
5060 for expr in exprs {
5061 out.push(match expr {
5062 Expr::Array(exprs) => plan_array(ecx, exprs, type_hint.clone())?,
5065 _ => plan_expr(ecx, expr)?,
5066 });
5067 }
5068
5069 let type_hint = match type_hint {
5071 Some(SqlScalarType::Array(elem_type)) => {
5076 let multidimensional = out.iter().any(|e| {
5077 matches!(
5078 ecx.scalar_type(e),
5079 CoercibleScalarType::Coerced(SqlScalarType::Array(_))
5080 )
5081 });
5082 if multidimensional {
5083 type_hint
5084 } else {
5085 Some(&**elem_type)
5086 }
5087 }
5088 Some(_) => None,
5092 None => None,
5094 };
5095
5096 let (elem_type, exprs) = if exprs.is_empty() {
5098 if let Some(elem_type) = type_hint {
5099 (elem_type.clone(), vec![])
5100 } else {
5101 sql_bail!("cannot determine type of empty array");
5102 }
5103 } else {
5104 let out = coerce_homogeneous_exprs(&ecx.with_name("ARRAY"), out, type_hint)?;
5105 (ecx.scalar_type(&out[0]), out)
5106 };
5107
5108 if matches!(
5114 elem_type,
5115 SqlScalarType::Char { .. } | SqlScalarType::List { .. } | SqlScalarType::Map { .. }
5116 ) {
5117 bail_unsupported!(format!(
5118 "{}[]",
5119 ecx.humanize_sql_scalar_type(&elem_type, false)
5120 ));
5121 }
5122
5123 Ok(HirScalarExpr::call_variadic(ArrayCreate { elem_type }, exprs).into())
5124}
5125
5126fn plan_list(
5127 ecx: &ExprContext,
5128 exprs: &[Expr<Aug>],
5129 type_hint: Option<&SqlScalarType>,
5130) -> Result<CoercibleScalarExpr, PlanError> {
5131 let (elem_type, exprs) = if exprs.is_empty() {
5132 if let Some(SqlScalarType::List { element_type, .. }) = type_hint {
5133 (element_type.without_modifiers(), vec![])
5134 } else {
5135 sql_bail!("cannot determine type of empty list");
5136 }
5137 } else {
5138 let type_hint = match type_hint {
5139 Some(SqlScalarType::List { element_type, .. }) => Some(&**element_type),
5140 _ => None,
5141 };
5142
5143 let mut out = vec![];
5144 for expr in exprs {
5145 out.push(match expr {
5146 Expr::List(exprs) => plan_list(ecx, exprs, type_hint)?,
5149 _ => plan_expr(ecx, expr)?,
5150 });
5151 }
5152 let out = coerce_homogeneous_exprs(&ecx.with_name("LIST"), out, type_hint)?;
5153 (ecx.scalar_type(&out[0]).without_modifiers(), out)
5154 };
5155
5156 if matches!(elem_type, SqlScalarType::Char { .. }) {
5157 bail_unsupported!("char list");
5158 }
5159
5160 Ok(HirScalarExpr::call_variadic(ListCreate { elem_type }, exprs).into())
5161}
5162
5163fn plan_map(
5164 ecx: &ExprContext,
5165 entries: &[MapEntry<Aug>],
5166 type_hint: Option<&SqlScalarType>,
5167) -> Result<CoercibleScalarExpr, PlanError> {
5168 let (value_type, exprs) = if entries.is_empty() {
5169 if let Some(SqlScalarType::Map { value_type, .. }) = type_hint {
5170 (value_type.without_modifiers(), vec![])
5171 } else {
5172 sql_bail!("cannot determine type of empty map");
5173 }
5174 } else {
5175 let type_hint = match type_hint {
5176 Some(SqlScalarType::Map { value_type, .. }) => Some(&**value_type),
5177 _ => None,
5178 };
5179
5180 let mut keys = vec![];
5181 let mut values = vec![];
5182 for MapEntry { key, value } in entries {
5183 let key = plan_expr(ecx, key)?.type_as(ecx, &SqlScalarType::String)?;
5184 let value = match value {
5185 Expr::Map(entries) => plan_map(ecx, entries, type_hint)?,
5188 _ => plan_expr(ecx, value)?,
5189 };
5190 keys.push(key);
5191 values.push(value);
5192 }
5193 let values = coerce_homogeneous_exprs(&ecx.with_name("MAP"), values, type_hint)?;
5194 let value_type = ecx.scalar_type(&values[0]).without_modifiers();
5195 let out = itertools::interleave(keys, values).collect();
5196 (value_type, out)
5197 };
5198
5199 if matches!(value_type, SqlScalarType::Char { .. }) {
5200 bail_unsupported!("char map");
5201 }
5202
5203 let expr = HirScalarExpr::call_variadic(MapBuild { value_type }, exprs);
5204 Ok(expr.into())
5205}
5206
5207pub fn coerce_homogeneous_exprs(
5224 ecx: &ExprContext,
5225 exprs: Vec<CoercibleScalarExpr>,
5226 force_type: Option<&SqlScalarType>,
5227) -> Result<Vec<HirScalarExpr>, PlanError> {
5228 assert!(!exprs.is_empty());
5229
5230 let target_holder;
5231 let target = match force_type {
5232 Some(t) => t,
5233 None => {
5234 let types: Vec<_> = exprs.iter().map(|e| ecx.scalar_type(e)).collect();
5235 target_holder = typeconv::guess_best_common_type(ecx, &types)?;
5236 &target_holder
5237 }
5238 };
5239
5240 let mut out = Vec::new();
5242 for expr in exprs {
5243 let arg = typeconv::plan_coerce(ecx, expr, target)?;
5244 let ccx = match force_type {
5245 None => CastContext::Implicit,
5246 Some(_) => CastContext::Explicit,
5247 };
5248 match typeconv::plan_cast(ecx, ccx, arg.clone(), target) {
5249 Ok(expr) => out.push(expr),
5250 Err(_) => sql_bail!(
5251 "{} could not convert type {} to {}",
5252 ecx.name,
5253 ecx.humanize_sql_scalar_type(&ecx.scalar_type(&arg), false),
5254 ecx.humanize_sql_scalar_type(target, false),
5255 ),
5256 }
5257 }
5258 Ok(out)
5259}
5260
5261pub(crate) fn resolve_desc_and_nulls_last<T: AstInfo>(
5264 obe: &OrderByExpr<T>,
5265 column: usize,
5266) -> ColumnOrder {
5267 let desc = !obe.asc.unwrap_or(true);
5268 ColumnOrder {
5269 column,
5270 desc,
5271 nulls_last: obe.nulls_last.unwrap_or(!desc),
5274 }
5275}
5276
5277fn plan_function_order_by(
5285 ecx: &ExprContext,
5286 order_by: &[OrderByExpr<Aug>],
5287) -> Result<(Vec<HirScalarExpr>, Vec<ColumnOrder>), PlanError> {
5288 let mut order_by_exprs = vec![];
5289 let mut col_orders = vec![];
5290 {
5291 for (i, obe) in order_by.iter().enumerate() {
5292 let expr = plan_expr(ecx, &obe.expr)?.type_as_any(ecx)?;
5296 order_by_exprs.push(expr);
5297 col_orders.push(resolve_desc_and_nulls_last(obe, i));
5298 }
5299 }
5300 Ok((order_by_exprs, col_orders))
5301}
5302
5303fn humanize_or_debug(scx: &StatementContext, name: &ResolvedItemName) -> String {
5308 scx.humanize_resolved_name(name)
5309 .map(|n| n.to_string())
5310 .unwrap_or_else(|_| format!("<error when trying to humanize `{name:?}`>"))
5311}
5312
5313fn plan_aggregate_common(
5315 ecx: &ExprContext,
5316 Function::<Aug> {
5317 name,
5318 args,
5319 filter,
5320 over: _,
5321 distinct,
5322 }: &Function<Aug>,
5323) -> Result<AggregateExpr, PlanError> {
5324 let impls = match resolve_func(ecx, name, args)? {
5339 Func::Aggregate(impls) => impls,
5340 _ => bail_internal!("plan_aggregate_common called on non-aggregate function"),
5341 };
5342
5343 let (args, order_by) = match &args {
5352 FunctionArgs::Star => (vec![], vec![]),
5353 FunctionArgs::Args { args, order_by } => {
5354 if args.is_empty() {
5355 sql_bail!(
5356 "{}(*) must be used to call a parameterless aggregate function",
5357 humanize_or_debug(ecx.qcx.scx, name)
5358 );
5359 }
5360 let args = plan_exprs(ecx, args)?;
5361 (args, order_by.clone())
5362 }
5363 };
5364
5365 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &order_by)?;
5366
5367 let (mut expr, func) = func::select_impl(ecx, FuncSpec::Func(name), impls, args, col_orders)?;
5368 if let Some(filter) = &filter {
5369 let cond =
5379 plan_expr(&ecx.with_name("FILTER"), filter)?.type_as(ecx, &SqlScalarType::Bool)?;
5380 let expr_typ = ecx.scalar_type(&expr);
5381 expr = HirScalarExpr::if_then_else(
5382 cond,
5383 expr,
5384 HirScalarExpr::literal(func.identity_datum(), expr_typ),
5385 );
5386 }
5387
5388 let mut seen_outer = false;
5389 let mut seen_inner = false;
5390 #[allow(deprecated)]
5391 expr.visit_columns(0, &mut |depth, col| {
5392 if depth == 0 && col.level == 0 {
5393 seen_inner = true;
5394 } else if col.level > depth {
5395 seen_outer = true;
5396 }
5397 });
5398 if seen_outer && !seen_inner {
5399 bail_unsupported!(
5400 3720,
5401 "aggregate functions that refer exclusively to outer columns"
5402 );
5403 }
5404
5405 if func.is_order_sensitive() {
5408 let field_names = iter::repeat(ColumnName::from(""))
5409 .take(1 + order_by_exprs.len())
5410 .collect();
5411 let mut exprs = vec![expr];
5412 exprs.extend(order_by_exprs);
5413 expr = HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs);
5414 }
5415
5416 Ok(AggregateExpr {
5417 func,
5418 expr: Box::new(expr),
5419 distinct: *distinct,
5420 })
5421}
5422
5423fn plan_identifier(ecx: &ExprContext, names: &[Ident]) -> Result<HirScalarExpr, PlanError> {
5424 let mut names = names.to_vec();
5425 let Some(last) = names.pop() else {
5428 bail_internal!("empty identifier");
5429 };
5430 let col_name = normalize::column_name(last);
5431
5432 if !names.is_empty() {
5434 let table_name = normalize::unresolved_item_name(UnresolvedItemName(names))?;
5435 let (i, i_name) = ecx.scope.resolve_table_column(
5436 &ecx.qcx.outer_scopes,
5437 &table_name,
5438 &col_name,
5439 &mut ecx.qcx.name_manager.borrow_mut(),
5440 )?;
5441 return Ok(HirScalarExpr::named_column(i, i_name));
5442 }
5443
5444 let similar_names = match ecx.scope.resolve_column(
5447 &ecx.qcx.outer_scopes,
5448 &col_name,
5449 &mut ecx.qcx.name_manager.borrow_mut(),
5450 ) {
5451 Ok((i, i_name)) => {
5452 return Ok(HirScalarExpr::named_column(i, i_name));
5453 }
5454 Err(PlanError::UnknownColumn { similar, .. }) => similar,
5455 Err(e) => return Err(e),
5456 };
5457
5458 let items = ecx.scope.items_from_table(
5461 &ecx.qcx.outer_scopes,
5462 &PartialItemName {
5463 database: None,
5464 schema: None,
5465 item: col_name.as_str().to_owned(),
5466 },
5467 )?;
5468 match items.as_slice() {
5469 [] => Err(PlanError::UnknownColumn {
5471 table: None,
5472 column: col_name,
5473 similar: similar_names,
5474 }),
5475 [(column, item)] if item.from_single_column_function => Ok(HirScalarExpr::named_column(
5480 *column,
5481 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5482 )),
5483 _ => {
5486 let mut has_exists_column = None;
5487 let (exprs, field_names): (Vec<_>, Vec<_>) = items
5488 .into_iter()
5489 .filter_map(|(column, item)| {
5490 if item.is_exists_column_for_a_table_function_that_was_in_the_target_list {
5491 has_exists_column = Some(column);
5492 None
5493 } else {
5494 let expr = HirScalarExpr::named_column(
5495 column,
5496 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5497 );
5498 let name = item.column_name.clone();
5499 Some((expr, name))
5500 }
5501 })
5502 .unzip();
5503 let expr = if exprs.len() == 1 && has_exists_column.is_some() {
5505 exprs.into_element()
5506 } else {
5507 HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs)
5508 };
5509 if let Some(has_exists_column) = has_exists_column {
5510 Ok(HirScalarExpr::if_then_else(
5511 HirScalarExpr::unnamed_column(has_exists_column)
5512 .call_unary(UnaryFunc::IsNull(mz_expr::func::IsNull)),
5513 HirScalarExpr::literal_null(ecx.scalar_type(&expr)),
5514 expr,
5515 ))
5516 } else {
5517 Ok(expr)
5518 }
5519 }
5520 }
5521}
5522
5523fn plan_op(
5524 ecx: &ExprContext,
5525 op: &str,
5526 expr1: &Expr<Aug>,
5527 expr2: Option<&Expr<Aug>>,
5528) -> Result<HirScalarExpr, PlanError> {
5529 let impls = func::resolve_op(op)?;
5530 let args = match expr2 {
5531 None => plan_exprs(ecx, &[expr1])?,
5532 Some(expr2) => plan_exprs(ecx, &[expr1, expr2])?,
5533 };
5534 func::select_impl(ecx, FuncSpec::Op(op), impls, args, vec![])
5535}
5536
5537fn plan_function<'a>(
5538 ecx: &ExprContext,
5539 f @ Function {
5540 name,
5541 args,
5542 filter,
5543 over,
5544 distinct,
5545 }: &'a Function<Aug>,
5546) -> Result<HirScalarExpr, PlanError> {
5547 let impls = match resolve_func(ecx, name, args)? {
5548 Func::Table(_) => {
5549 sql_bail!(
5550 "table functions are not allowed in {} (function {})",
5551 ecx.name,
5552 name
5553 );
5554 }
5555 Func::Scalar(impls) => {
5556 if over.is_some() {
5557 sql_bail!(
5558 "OVER clause not allowed on {name}. The OVER clause can only be used with window functions (including aggregations)."
5559 );
5560 }
5561 impls
5562 }
5563 Func::ScalarWindow(impls) => {
5564 let (
5565 ignore_nulls,
5566 order_by_exprs,
5567 col_orders,
5568 _window_frame,
5569 partition_by,
5570 scalar_args,
5571 ) = plan_window_function_non_aggr(ecx, f)?;
5572
5573 if !scalar_args.is_empty() {
5577 if let ResolvedItemName::Item {
5578 full_name: FullItemName { item, .. },
5579 ..
5580 } = name
5581 {
5582 sql_bail!(
5583 "function {} has 0 parameters, but was called with {}",
5584 item,
5585 scalar_args.len()
5586 );
5587 }
5588 }
5589
5590 let func = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
5595
5596 if ignore_nulls {
5597 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5600 }
5601
5602 return Ok(HirScalarExpr::windowing(WindowExpr {
5603 func: WindowExprType::Scalar(ScalarWindowExpr {
5604 func,
5605 order_by: col_orders,
5606 }),
5607 partition_by,
5608 order_by: order_by_exprs,
5609 }));
5610 }
5611 Func::ValueWindow(impls) => {
5612 let window_plan = plan_window_function_non_aggr(ecx, f)?;
5613 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by, win_args) =
5614 window_plan;
5615
5616 let (args_encoded, func) =
5617 func::select_impl(ecx, FuncSpec::Func(name), impls, win_args, vec![])?;
5618
5619 if ignore_nulls {
5620 match func {
5621 ValueWindowFunc::Lag | ValueWindowFunc::Lead => {}
5622 _ => bail_unsupported!(IGNORE_NULLS_ERROR_MSG),
5623 }
5624 }
5625
5626 return Ok(HirScalarExpr::windowing(WindowExpr {
5627 func: WindowExprType::Value(ValueWindowExpr {
5628 func,
5629 args: Box::new(args_encoded),
5630 order_by: col_orders,
5631 window_frame,
5632 ignore_nulls, }),
5634 partition_by,
5635 order_by: order_by_exprs,
5636 }));
5637 }
5638 Func::Aggregate(_) => {
5639 if f.over.is_none() {
5640 if ecx.allow_aggregates {
5642 sql_bail!(
5645 "Internal error: encountered unplanned non-windowed aggregate function: {:?}",
5646 name,
5647 );
5648 } else {
5649 sql_bail!(
5652 "aggregate functions are not allowed in {} (function {})",
5653 ecx.name,
5654 name
5655 );
5656 }
5657 } else {
5658 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by) =
5659 plan_window_function_common(ecx, &f.name, &f.over)?;
5660
5661 match (&window_frame.start_bound, &window_frame.end_bound) {
5663 (
5664 mz_expr::WindowFrameBound::UnboundedPreceding,
5665 mz_expr::WindowFrameBound::OffsetPreceding(..),
5666 )
5667 | (
5668 mz_expr::WindowFrameBound::UnboundedPreceding,
5669 mz_expr::WindowFrameBound::OffsetFollowing(..),
5670 )
5671 | (
5672 mz_expr::WindowFrameBound::OffsetPreceding(..),
5673 mz_expr::WindowFrameBound::UnboundedFollowing,
5674 )
5675 | (
5676 mz_expr::WindowFrameBound::OffsetFollowing(..),
5677 mz_expr::WindowFrameBound::UnboundedFollowing,
5678 ) => bail_unsupported!("mixed unbounded - offset frames"),
5679 (_, _) => {} }
5681
5682 if ignore_nulls {
5683 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5687 }
5688
5689 let aggregate_expr = plan_aggregate_common(ecx, f)?;
5690
5691 if aggregate_expr.distinct {
5692 bail_unsupported!("DISTINCT in window aggregates");
5694 }
5695
5696 return Ok(HirScalarExpr::windowing(WindowExpr {
5697 func: WindowExprType::Aggregate(AggregateWindowExpr {
5698 aggregate_expr,
5699 order_by: col_orders,
5700 window_frame,
5701 }),
5702 partition_by,
5703 order_by: order_by_exprs,
5704 }));
5705 }
5706 }
5707 };
5708
5709 if over.is_some() {
5710 bail_internal!("OVER clause should have been handled by the window function path above");
5711 }
5712
5713 if *distinct {
5714 sql_bail!(
5715 "DISTINCT specified, but {} is not an aggregate function",
5716 humanize_or_debug(ecx.qcx.scx, name)
5717 );
5718 }
5719 if filter.is_some() {
5720 sql_bail!(
5721 "FILTER specified, but {} is not an aggregate function",
5722 humanize_or_debug(ecx.qcx.scx, name)
5723 );
5724 }
5725
5726 let scalar_args = match &args {
5727 FunctionArgs::Star => {
5728 sql_bail!(
5729 "* argument is invalid with non-aggregate function {}",
5730 humanize_or_debug(ecx.qcx.scx, name)
5731 )
5732 }
5733 FunctionArgs::Args { args, order_by } => {
5734 if !order_by.is_empty() {
5735 sql_bail!(
5736 "ORDER BY specified, but {} is not an aggregate function",
5737 humanize_or_debug(ecx.qcx.scx, name)
5738 );
5739 }
5740 plan_exprs(ecx, args)?
5741 }
5742 };
5743
5744 func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])
5745}
5746
5747pub const IGNORE_NULLS_ERROR_MSG: &str =
5748 "IGNORE NULLS and RESPECT NULLS options for functions other than LAG and LEAD";
5749
5750pub fn resolve_func(
5754 ecx: &ExprContext,
5755 name: &ResolvedItemName,
5756 args: &mz_sql_parser::ast::FunctionArgs<Aug>,
5757) -> Result<&'static Func, PlanError> {
5758 if let Ok(i) = ecx.qcx.scx.get_item_by_resolved_name(name) {
5759 if let Ok(f) = i.func() {
5760 return Ok(f);
5761 }
5762 }
5763
5764 let cexprs = match args {
5767 mz_sql_parser::ast::FunctionArgs::Star => vec![],
5768 mz_sql_parser::ast::FunctionArgs::Args { args, order_by } => {
5769 if !order_by.is_empty() {
5770 sql_bail!(
5771 "ORDER BY specified, but {} is not an aggregate function",
5772 name
5773 );
5774 }
5775 plan_exprs(ecx, args)?
5776 }
5777 };
5778
5779 let arg_types: Vec<_> = cexprs
5780 .into_iter()
5781 .map(|ty| match ecx.scalar_type(&ty) {
5782 CoercibleScalarType::Coerced(ty) => ecx.humanize_sql_scalar_type(&ty, false),
5783 CoercibleScalarType::Record(_) => "record".to_string(),
5784 CoercibleScalarType::Uncoerced => "unknown".to_string(),
5785 })
5786 .collect();
5787
5788 Err(PlanError::UnknownFunction {
5789 name: name.to_string(),
5790 arg_types,
5791 })
5792}
5793
5794fn plan_is_expr<'a>(
5795 ecx: &ExprContext,
5796 expr: &'a Expr<Aug>,
5797 construct: &IsExprConstruct<Aug>,
5798 not: bool,
5799) -> Result<HirScalarExpr, PlanError> {
5800 let expr_hir = plan_expr(ecx, expr)?;
5801
5802 let mut result = match construct {
5803 IsExprConstruct::Null => {
5804 expr_hir.type_as_any(ecx)?.call_is_null()
5809 }
5810 IsExprConstruct::Unknown => expr_hir.type_as(ecx, &SqlScalarType::Bool)?.call_is_null(),
5811 IsExprConstruct::True => expr_hir
5812 .type_as(ecx, &SqlScalarType::Bool)?
5813 .call_unary(UnaryFunc::IsTrue(expr_func::IsTrue)),
5814 IsExprConstruct::False => expr_hir
5815 .type_as(ecx, &SqlScalarType::Bool)?
5816 .call_unary(UnaryFunc::IsFalse(expr_func::IsFalse)),
5817 IsExprConstruct::DistinctFrom(expr2) => {
5818 let ne_ast = expr.clone().not_equals(expr2.as_ref().clone());
5829 let ne_hir = plan_expr(ecx, &ne_ast)?.type_as_any(ecx)?;
5830
5831 let expr1_hir = expr_hir.type_as_any(ecx)?;
5832 let expr2_hir = plan_expr(ecx, expr2)?.type_as_any(ecx)?;
5833
5834 let term1 = HirScalarExpr::variadic_or(vec![
5835 ne_hir,
5836 expr1_hir.clone().call_is_null(),
5837 expr2_hir.clone().call_is_null(),
5838 ]);
5839 let term2 = HirScalarExpr::variadic_or(vec![
5840 expr1_hir.call_is_null().not(),
5841 expr2_hir.call_is_null().not(),
5842 ]);
5843 term1.and(term2)
5844 }
5845 };
5846 if not {
5847 result = result.not();
5848 }
5849 Ok(result)
5850}
5851
5852fn plan_case<'a>(
5853 ecx: &ExprContext,
5854 operand: &'a Option<Box<Expr<Aug>>>,
5855 conditions: &'a [Expr<Aug>],
5856 results: &'a [Expr<Aug>],
5857 else_result: &'a Option<Box<Expr<Aug>>>,
5858) -> Result<HirScalarExpr, PlanError> {
5859 let mut cond_exprs = Vec::new();
5860 let mut result_exprs = Vec::new();
5861 for (c, r) in conditions.iter().zip_eq(results) {
5862 let c = match operand {
5863 Some(operand) => operand.clone().equals(c.clone()),
5864 None => c.clone(),
5865 };
5866 let cexpr = plan_expr(ecx, &c)?.type_as(ecx, &SqlScalarType::Bool)?;
5867 cond_exprs.push(cexpr);
5868 result_exprs.push(r);
5869 }
5870 result_exprs.push(match else_result {
5871 Some(else_result) => else_result,
5872 None => &Expr::Value(Value::Null),
5873 });
5874 let mut result_exprs = coerce_homogeneous_exprs(
5875 &ecx.with_name("CASE"),
5876 plan_exprs(ecx, &result_exprs)?,
5877 None,
5878 )?;
5879 let mut expr = result_exprs.pop().unwrap();
5880 assert_eq!(cond_exprs.len(), result_exprs.len());
5881 for (cexpr, rexpr) in cond_exprs
5882 .into_iter()
5883 .rev()
5884 .zip_eq(result_exprs.into_iter().rev())
5885 {
5886 expr = HirScalarExpr::if_then_else(cexpr, rexpr, expr);
5887 }
5888 Ok(expr)
5889}
5890
5891fn plan_literal<'a>(l: &'a Value) -> Result<CoercibleScalarExpr, PlanError> {
5892 let (datum, scalar_type) = match l {
5893 Value::Number(s) => {
5894 let d = strconv::parse_numeric(s.as_str())?;
5895 if !s.contains(&['E', '.'][..]) {
5896 if let Ok(n) = d.0.try_into() {
5898 (Datum::Int32(n), SqlScalarType::Int32)
5899 } else if let Ok(n) = d.0.try_into() {
5900 (Datum::Int64(n), SqlScalarType::Int64)
5901 } else {
5902 (
5903 Datum::Numeric(d),
5904 SqlScalarType::Numeric { max_scale: None },
5905 )
5906 }
5907 } else {
5908 (
5909 Datum::Numeric(d),
5910 SqlScalarType::Numeric { max_scale: None },
5911 )
5912 }
5913 }
5914 Value::HexString(_) => bail_unsupported!("hex string literals"),
5915 Value::Boolean(b) => match b {
5916 false => (Datum::False, SqlScalarType::Bool),
5917 true => (Datum::True, SqlScalarType::Bool),
5918 },
5919 Value::Interval(i) => {
5920 let i = literal::plan_interval(i)?;
5921 (Datum::Interval(i), SqlScalarType::Interval)
5922 }
5923 Value::String(s) => return Ok(CoercibleScalarExpr::LiteralString(s.clone())),
5924 Value::Null => return Ok(CoercibleScalarExpr::LiteralNull),
5925 };
5926 let expr = HirScalarExpr::literal(datum, scalar_type);
5927 Ok(expr.into())
5928}
5929
5930fn plan_window_function_non_aggr<'a>(
5933 ecx: &ExprContext,
5934 Function {
5935 name,
5936 args,
5937 filter,
5938 over,
5939 distinct,
5940 }: &'a Function<Aug>,
5941) -> Result<
5942 (
5943 bool,
5944 Vec<HirScalarExpr>,
5945 Vec<ColumnOrder>,
5946 mz_expr::WindowFrame,
5947 Vec<HirScalarExpr>,
5948 Vec<CoercibleScalarExpr>,
5949 ),
5950 PlanError,
5951> {
5952 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition) =
5953 plan_window_function_common(ecx, name, over)?;
5954
5955 if *distinct {
5956 sql_bail!(
5957 "DISTINCT specified, but {} is not an aggregate function",
5958 name
5959 );
5960 }
5961
5962 if filter.is_some() {
5963 bail_unsupported!("FILTER in non-aggregate window functions");
5964 }
5965
5966 let scalar_args = match &args {
5967 FunctionArgs::Star => {
5968 sql_bail!("* argument is invalid with non-aggregate function {}", name)
5969 }
5970 FunctionArgs::Args { args, order_by } => {
5971 if !order_by.is_empty() {
5972 sql_bail!(
5973 "ORDER BY specified, but {} is not an aggregate function",
5974 name
5975 );
5976 }
5977 plan_exprs(ecx, args)?
5978 }
5979 };
5980
5981 Ok((
5982 ignore_nulls,
5983 order_by_exprs,
5984 col_orders,
5985 window_frame,
5986 partition,
5987 scalar_args,
5988 ))
5989}
5990
5991fn plan_window_function_common(
5993 ecx: &ExprContext,
5994 name: &<Aug as AstInfo>::ItemName,
5995 over: &Option<WindowSpec<Aug>>,
5996) -> Result<
5997 (
5998 bool,
5999 Vec<HirScalarExpr>,
6000 Vec<ColumnOrder>,
6001 mz_expr::WindowFrame,
6002 Vec<HirScalarExpr>,
6003 ),
6004 PlanError,
6005> {
6006 if !ecx.allow_windows {
6007 sql_bail!(
6008 "window functions are not allowed in {} (function {})",
6009 ecx.name,
6010 name
6011 );
6012 }
6013
6014 let window_spec = match over.as_ref() {
6015 Some(over) => over,
6016 None => sql_bail!("window function {} requires an OVER clause", name),
6017 };
6018 if window_spec.ignore_nulls && window_spec.respect_nulls {
6019 sql_bail!("Both IGNORE NULLS and RESPECT NULLS were given.");
6020 }
6021 let window_frame = match window_spec.window_frame.as_ref() {
6022 Some(frame) => plan_window_frame(frame)?,
6023 None => mz_expr::WindowFrame::default(),
6024 };
6025 let mut partition = Vec::new();
6026 for expr in &window_spec.partition_by {
6027 partition.push(plan_expr(ecx, expr)?.type_as_any(ecx)?);
6028 }
6029
6030 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &window_spec.order_by)?;
6031
6032 Ok((
6033 window_spec.ignore_nulls,
6034 order_by_exprs,
6035 col_orders,
6036 window_frame,
6037 partition,
6038 ))
6039}
6040
6041fn plan_window_frame(
6042 WindowFrame {
6043 units,
6044 start_bound,
6045 end_bound,
6046 }: &WindowFrame,
6047) -> Result<mz_expr::WindowFrame, PlanError> {
6048 use mz_expr::WindowFrameBound::*;
6049 let units = window_frame_unit_ast_to_expr(units)?;
6050 let start_bound = window_frame_bound_ast_to_expr(start_bound);
6051 let end_bound = end_bound
6052 .as_ref()
6053 .map(window_frame_bound_ast_to_expr)
6054 .unwrap_or(CurrentRow);
6055
6056 match (&start_bound, &end_bound) {
6058 (UnboundedFollowing, _) => {
6060 sql_bail!("frame start cannot be UNBOUNDED FOLLOWING")
6061 }
6062 (_, UnboundedPreceding) => {
6064 sql_bail!("frame end cannot be UNBOUNDED PRECEDING")
6065 }
6066 (CurrentRow, OffsetPreceding(_)) => {
6068 sql_bail!("frame starting from current row cannot have preceding rows")
6069 }
6070 (OffsetFollowing(_), OffsetPreceding(_) | CurrentRow) => {
6071 sql_bail!("frame starting from following row cannot have preceding rows")
6072 }
6073 (OffsetPreceding(o1), OffsetFollowing(o2)) => {
6076 if *o1 > 1000000 || *o2 > 1000000 {
6080 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6081 }
6082 }
6083 (OffsetPreceding(o1), OffsetPreceding(o2)) => {
6084 if *o1 > 1000000 || *o2 > 1000000 {
6085 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6086 }
6087 }
6088 (OffsetFollowing(o1), OffsetFollowing(o2)) => {
6089 if *o1 > 1000000 || *o2 > 1000000 {
6090 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6091 }
6092 }
6093 (OffsetPreceding(o), CurrentRow) => {
6094 if *o > 1000000 {
6095 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6096 }
6097 }
6098 (CurrentRow, OffsetFollowing(o)) => {
6099 if *o > 1000000 {
6100 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6101 }
6102 }
6103 (_, _) => (),
6105 }
6106
6107 if units == mz_expr::WindowFrameUnits::Range
6110 && (start_bound != UnboundedPreceding || end_bound != CurrentRow)
6111 {
6112 bail_unsupported!("RANGE in non-default window frames")
6113 }
6114
6115 let frame = mz_expr::WindowFrame {
6116 units,
6117 start_bound,
6118 end_bound,
6119 };
6120 Ok(frame)
6121}
6122
6123fn window_frame_unit_ast_to_expr(
6124 unit: &WindowFrameUnits,
6125) -> Result<mz_expr::WindowFrameUnits, PlanError> {
6126 match unit {
6127 WindowFrameUnits::Rows => Ok(mz_expr::WindowFrameUnits::Rows),
6128 WindowFrameUnits::Range => Ok(mz_expr::WindowFrameUnits::Range),
6129 WindowFrameUnits::Groups => bail_unsupported!("GROUPS in window frames"),
6130 }
6131}
6132
6133fn window_frame_bound_ast_to_expr(bound: &WindowFrameBound) -> mz_expr::WindowFrameBound {
6134 match bound {
6135 WindowFrameBound::CurrentRow => mz_expr::WindowFrameBound::CurrentRow,
6136 WindowFrameBound::Preceding(None) => mz_expr::WindowFrameBound::UnboundedPreceding,
6137 WindowFrameBound::Preceding(Some(offset)) => {
6138 mz_expr::WindowFrameBound::OffsetPreceding(*offset)
6139 }
6140 WindowFrameBound::Following(None) => mz_expr::WindowFrameBound::UnboundedFollowing,
6141 WindowFrameBound::Following(Some(offset)) => {
6142 mz_expr::WindowFrameBound::OffsetFollowing(*offset)
6143 }
6144 }
6145}
6146
6147pub fn scalar_type_from_sql(
6148 scx: &StatementContext,
6149 data_type: &ResolvedDataType,
6150) -> Result<SqlScalarType, PlanError> {
6151 match data_type {
6152 ResolvedDataType::AnonymousList(elem_type) => {
6153 let elem_type = scalar_type_from_sql(scx, elem_type)?;
6154 if matches!(elem_type, SqlScalarType::Char { .. }) {
6155 bail_unsupported!("char list");
6156 }
6157 Ok(SqlScalarType::List {
6158 element_type: Box::new(elem_type),
6159 custom_id: None,
6160 })
6161 }
6162 ResolvedDataType::AnonymousMap {
6163 key_type,
6164 value_type,
6165 } => {
6166 match scalar_type_from_sql(scx, key_type)? {
6167 SqlScalarType::String => {}
6168 other => sql_bail!(
6169 "map key type must be {}, got {}",
6170 scx.humanize_sql_scalar_type(&SqlScalarType::String, false),
6171 scx.humanize_sql_scalar_type(&other, false)
6172 ),
6173 }
6174 Ok(SqlScalarType::Map {
6175 value_type: Box::new(scalar_type_from_sql(scx, value_type)?),
6176 custom_id: None,
6177 })
6178 }
6179 ResolvedDataType::Named { id, modifiers, .. } => {
6180 scalar_type_from_catalog(scx.catalog, *id, modifiers)
6181 }
6182 ResolvedDataType::Error => bail_internal!("should have been caught in name resolution"),
6183 }
6184}
6185
6186const MAX_TYPE_NESTING_DEPTH: usize = 128;
6190
6191const MAX_TYPE_RESOLUTION_NODES: usize = 100_000;
6197
6198pub fn scalar_type_from_catalog(
6199 catalog: &dyn SessionCatalog,
6200 id: CatalogItemId,
6201 modifiers: &[i64],
6202) -> Result<SqlScalarType, PlanError> {
6203 let (depth_limit, mut budget) = type_resolution_limits(catalog);
6204 scalar_type_from_catalog_inner(catalog, id, modifiers, 0, depth_limit, &mut budget)
6205}
6206
6207fn type_resolution_limits(catalog: &dyn SessionCatalog) -> (usize, usize) {
6221 if catalog
6222 .system_vars()
6223 .unsafe_enable_unbounded_custom_type_resolution()
6224 {
6225 (usize::MAX, usize::MAX)
6226 } else {
6227 (MAX_TYPE_NESTING_DEPTH, MAX_TYPE_RESOLUTION_NODES)
6228 }
6229}
6230
6231pub struct TypeResolutionBudget {
6244 remaining: usize,
6247 depth_limit: usize,
6250}
6251
6252impl TypeResolutionBudget {
6253 pub fn for_root(catalog: &dyn SessionCatalog) -> TypeResolutionBudget {
6259 let (depth_limit, budget) = type_resolution_limits(catalog);
6260 TypeResolutionBudget {
6261 remaining: budget.saturating_sub(1),
6263 depth_limit,
6264 }
6265 }
6266
6267 pub fn resolve_child(
6271 &mut self,
6272 catalog: &dyn SessionCatalog,
6273 id: CatalogItemId,
6274 modifiers: &[i64],
6275 ) -> Result<SqlScalarType, PlanError> {
6276 scalar_type_from_catalog_inner(
6277 catalog,
6278 id,
6279 modifiers,
6280 1,
6281 self.depth_limit,
6282 &mut self.remaining,
6283 )
6284 }
6285}
6286
6287fn scalar_type_from_catalog_inner(
6288 catalog: &dyn SessionCatalog,
6289 id: CatalogItemId,
6290 modifiers: &[i64],
6291 depth: usize,
6292 depth_limit: usize,
6293 budget: &mut usize,
6294) -> Result<SqlScalarType, PlanError> {
6295 if depth > depth_limit {
6296 sql_bail!("custom type nesting depth exceeds limit of {}", depth_limit);
6297 }
6298 *budget = match budget.checked_sub(1) {
6299 Some(remaining) => remaining,
6300 None => sql_bail!("custom type is too complex to resolve"),
6301 };
6302 let entry = catalog.get_item(&id);
6303 let type_details = match entry.type_details() {
6304 Some(type_details) => type_details,
6305 None => {
6306 sql_bail!(
6309 "internal error: {} does not refer to a type",
6310 catalog.resolve_full_name(entry.name()).to_string().quoted()
6311 );
6312 }
6313 };
6314 match &type_details.typ {
6315 CatalogType::Numeric => {
6316 let mut modifiers = modifiers.iter().fuse();
6317 let precision = match modifiers.next() {
6318 Some(p) if *p < 1 || *p > i64::from(NUMERIC_DATUM_MAX_PRECISION) => {
6319 sql_bail!(
6320 "precision for type numeric must be between 1 and {}",
6321 NUMERIC_DATUM_MAX_PRECISION,
6322 );
6323 }
6324 Some(p) => Some(*p),
6325 None => None,
6326 };
6327 let scale = match modifiers.next() {
6328 Some(scale) => {
6329 if let Some(precision) = precision {
6330 if *scale > precision {
6331 sql_bail!(
6332 "scale for type numeric must be between 0 and precision {}",
6333 precision
6334 );
6335 }
6336 }
6337 Some(NumericMaxScale::try_from(*scale)?)
6338 }
6339 None => None,
6340 };
6341 if modifiers.next().is_some() {
6342 sql_bail!("type numeric supports at most two type modifiers");
6343 }
6344 Ok(SqlScalarType::Numeric { max_scale: scale })
6345 }
6346 CatalogType::Char => {
6347 let mut modifiers = modifiers.iter().fuse();
6348 let length = match modifiers.next() {
6349 Some(l) => Some(CharLength::try_from(*l)?),
6350 None => Some(CharLength::ONE),
6351 };
6352 if modifiers.next().is_some() {
6353 sql_bail!("type character supports at most one type modifier");
6354 }
6355 Ok(SqlScalarType::Char { length })
6356 }
6357 CatalogType::VarChar => {
6358 let mut modifiers = modifiers.iter().fuse();
6359 let length = match modifiers.next() {
6360 Some(l) => Some(VarCharMaxLength::try_from(*l)?),
6361 None => None,
6362 };
6363 if modifiers.next().is_some() {
6364 sql_bail!("type character varying supports at most one type modifier");
6365 }
6366 Ok(SqlScalarType::VarChar { max_length: length })
6367 }
6368 CatalogType::Timestamp => {
6369 let mut modifiers = modifiers.iter().fuse();
6370 let precision = match modifiers.next() {
6371 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6372 None => None,
6373 };
6374 if modifiers.next().is_some() {
6375 sql_bail!("type timestamp supports at most one type modifier");
6376 }
6377 Ok(SqlScalarType::Timestamp { precision })
6378 }
6379 CatalogType::TimestampTz => {
6380 let mut modifiers = modifiers.iter().fuse();
6381 let precision = match modifiers.next() {
6382 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6383 None => None,
6384 };
6385 if modifiers.next().is_some() {
6386 sql_bail!("type timestamp with time zone supports at most one type modifier");
6387 }
6388 Ok(SqlScalarType::TimestampTz { precision })
6389 }
6390 t => {
6391 if !modifiers.is_empty() {
6392 sql_bail!(
6393 "{} does not support type modifiers",
6394 catalog.resolve_full_name(entry.name()).to_string()
6395 );
6396 }
6397 match t {
6398 CatalogType::Array {
6399 element_reference: element_id,
6400 } => Ok(SqlScalarType::Array(Box::new(
6401 scalar_type_from_catalog_inner(
6402 catalog,
6403 *element_id,
6404 modifiers,
6405 depth + 1,
6406 depth_limit,
6407 budget,
6408 )?,
6409 ))),
6410 CatalogType::List {
6411 element_reference: element_id,
6412 element_modifiers,
6413 } => Ok(SqlScalarType::List {
6414 element_type: Box::new(scalar_type_from_catalog_inner(
6415 catalog,
6416 *element_id,
6417 element_modifiers,
6418 depth + 1,
6419 depth_limit,
6420 budget,
6421 )?),
6422 custom_id: Some(id),
6423 }),
6424 CatalogType::Map {
6425 key_reference: _,
6426 key_modifiers: _,
6427 value_reference: value_id,
6428 value_modifiers,
6429 } => Ok(SqlScalarType::Map {
6430 value_type: Box::new(scalar_type_from_catalog_inner(
6431 catalog,
6432 *value_id,
6433 value_modifiers,
6434 depth + 1,
6435 depth_limit,
6436 budget,
6437 )?),
6438 custom_id: Some(id),
6439 }),
6440 CatalogType::Range {
6441 element_reference: element_id,
6442 } => Ok(SqlScalarType::Range {
6443 element_type: Box::new(scalar_type_from_catalog_inner(
6444 catalog,
6445 *element_id,
6446 &[],
6447 depth + 1,
6448 depth_limit,
6449 budget,
6450 )?),
6451 }),
6452 CatalogType::Record { fields } => {
6453 let scalars: Box<[(ColumnName, SqlColumnType)]> = fields
6454 .iter()
6455 .map(|f| {
6456 let scalar_type = scalar_type_from_catalog_inner(
6457 catalog,
6458 f.type_reference,
6459 &f.type_modifiers,
6460 depth + 1,
6461 depth_limit,
6462 budget,
6463 )?;
6464 Ok((
6465 f.name.clone(),
6466 SqlColumnType {
6467 scalar_type,
6468 nullable: true,
6469 },
6470 ))
6471 })
6472 .collect::<Result<Box<_>, PlanError>>()?;
6473 Ok(SqlScalarType::Record {
6474 fields: scalars,
6475 custom_id: Some(id),
6476 })
6477 }
6478 CatalogType::AclItem => Ok(SqlScalarType::AclItem),
6479 CatalogType::Bool => Ok(SqlScalarType::Bool),
6480 CatalogType::Bytes => Ok(SqlScalarType::Bytes),
6481 CatalogType::Date => Ok(SqlScalarType::Date),
6482 CatalogType::Float32 => Ok(SqlScalarType::Float32),
6483 CatalogType::Float64 => Ok(SqlScalarType::Float64),
6484 CatalogType::Int16 => Ok(SqlScalarType::Int16),
6485 CatalogType::Int32 => Ok(SqlScalarType::Int32),
6486 CatalogType::Int64 => Ok(SqlScalarType::Int64),
6487 CatalogType::UInt16 => Ok(SqlScalarType::UInt16),
6488 CatalogType::UInt32 => Ok(SqlScalarType::UInt32),
6489 CatalogType::UInt64 => Ok(SqlScalarType::UInt64),
6490 CatalogType::MzTimestamp => Ok(SqlScalarType::MzTimestamp),
6491 CatalogType::Interval => Ok(SqlScalarType::Interval),
6492 CatalogType::Jsonb => Ok(SqlScalarType::Jsonb),
6493 CatalogType::Oid => Ok(SqlScalarType::Oid),
6494 CatalogType::PgLegacyChar => Ok(SqlScalarType::PgLegacyChar),
6495 CatalogType::PgLegacyName => Ok(SqlScalarType::PgLegacyName),
6496 CatalogType::Pseudo => {
6497 sql_bail!(
6498 "cannot reference pseudo type {}",
6499 catalog.resolve_full_name(entry.name()).to_string()
6500 )
6501 }
6502 CatalogType::RegClass => Ok(SqlScalarType::RegClass),
6503 CatalogType::RegProc => Ok(SqlScalarType::RegProc),
6504 CatalogType::RegType => Ok(SqlScalarType::RegType),
6505 CatalogType::String => Ok(SqlScalarType::String),
6506 CatalogType::Time => Ok(SqlScalarType::Time),
6507 CatalogType::Uuid => Ok(SqlScalarType::Uuid),
6508 CatalogType::Int2Vector => Ok(SqlScalarType::Int2Vector),
6509 CatalogType::MzAclItem => Ok(SqlScalarType::MzAclItem),
6510 CatalogType::Numeric => unreachable!("handled above"),
6511 CatalogType::Char => unreachable!("handled above"),
6512 CatalogType::VarChar => unreachable!("handled above"),
6513 CatalogType::Timestamp => unreachable!("handled above"),
6514 CatalogType::TimestampTz => unreachable!("handled above"),
6515 }
6516 }
6517 }
6518}
6519
6520struct AggregateTableFuncVisitor<'a> {
6523 scx: &'a StatementContext<'a>,
6524 aggs: Vec<Function<Aug>>,
6525 within_aggregate: bool,
6526 tables: BTreeMap<Function<Aug>, String>,
6527 table_disallowed_context: Vec<&'static str>,
6528 in_select_item: bool,
6529 id_gen: IdGen,
6530 err: Option<PlanError>,
6531}
6532
6533impl<'a> AggregateTableFuncVisitor<'a> {
6534 fn new(scx: &'a StatementContext<'a>) -> AggregateTableFuncVisitor<'a> {
6535 AggregateTableFuncVisitor {
6536 scx,
6537 aggs: Vec::new(),
6538 within_aggregate: false,
6539 tables: BTreeMap::new(),
6540 table_disallowed_context: Vec::new(),
6541 in_select_item: false,
6542 id_gen: Default::default(),
6543 err: None,
6544 }
6545 }
6546
6547 fn into_result(
6548 self,
6549 ) -> Result<(Vec<Function<Aug>>, BTreeMap<Function<Aug>, String>), PlanError> {
6550 match self.err {
6551 Some(err) => Err(err),
6552 None => {
6553 let mut seen = BTreeSet::new();
6556 let aggs = self
6557 .aggs
6558 .into_iter()
6559 .filter(move |agg| seen.insert(agg.clone()))
6560 .collect();
6561 Ok((aggs, self.tables))
6562 }
6563 }
6564 }
6565}
6566
6567impl<'a> VisitMut<'_, Aug> for AggregateTableFuncVisitor<'a> {
6568 fn visit_function_mut(&mut self, func: &mut Function<Aug>) {
6569 let item = match self.scx.get_item_by_resolved_name(&func.name) {
6570 Ok(i) => i,
6571 Err(_) => return,
6573 };
6574
6575 match item.func() {
6576 Ok(Func::Aggregate { .. }) if func.over.is_none() => {
6579 if self.within_aggregate {
6580 self.err = Some(sql_err!("nested aggregate functions are not allowed",));
6581 return;
6582 }
6583 self.aggs.push(func.clone());
6584 let Function {
6585 name: _,
6586 args,
6587 filter,
6588 over: _,
6589 distinct: _,
6590 } = func;
6591 if let Some(filter) = filter {
6592 self.visit_expr_mut(filter);
6593 }
6594 let old_within_aggregate = self.within_aggregate;
6595 self.within_aggregate = true;
6596 self.table_disallowed_context
6597 .push("aggregate function calls");
6598
6599 self.visit_function_args_mut(args);
6600
6601 self.within_aggregate = old_within_aggregate;
6602 self.table_disallowed_context.pop();
6603 }
6604 Ok(Func::Table { .. }) => {
6605 self.table_disallowed_context.push("other table functions");
6606 visit_mut::visit_function_mut(self, func);
6607 self.table_disallowed_context.pop();
6608 }
6609 _ => visit_mut::visit_function_mut(self, func),
6610 }
6611 }
6612
6613 fn visit_query_mut(&mut self, _query: &mut Query<Aug>) {
6614 }
6616
6617 fn visit_expr_mut(&mut self, expr: &mut Expr<Aug>) {
6618 let (disallowed_context, func) = match expr {
6619 Expr::Case { .. } => (Some("CASE"), None),
6620 Expr::HomogenizingFunction {
6621 function: HomogenizingFunction::Coalesce,
6622 ..
6623 } => (Some("COALESCE"), None),
6624 Expr::Function(func) if self.in_select_item => {
6625 let mut table_func = None;
6628 if let Ok(item) = self.scx.get_item_by_resolved_name(&func.name) {
6629 if let Ok(Func::Table { .. }) = item.func() {
6630 if let Some(context) = self.table_disallowed_context.last() {
6631 self.err = Some(sql_err!(
6632 "table functions are not allowed in {} (function {})",
6633 context,
6634 func.name
6635 ));
6636 return;
6637 }
6638 table_func = Some(func.clone());
6639 }
6640 }
6641 (None, table_func)
6644 }
6645 _ => (None, None),
6646 };
6647 if let Some(func) = func {
6648 visit_mut::visit_expr_mut(self, expr);
6650 if let Function {
6652 name: _,
6653 args: _,
6654 filter: None,
6655 over: None,
6656 distinct: false,
6657 } = &func
6658 {
6659 let unique_id = self.id_gen.allocate_id();
6661 let id = self
6662 .tables
6663 .entry(func)
6664 .or_insert_with(|| format!("table_func_{unique_id}"));
6665 *expr = Expr::Identifier(vec![Ident::new_unchecked(id.clone())]);
6668 }
6669 }
6670 if let Some(context) = disallowed_context {
6671 self.table_disallowed_context.push(context);
6672 }
6673
6674 visit_mut::visit_expr_mut(self, expr);
6675
6676 if disallowed_context.is_some() {
6677 self.table_disallowed_context.pop();
6678 }
6679 }
6680
6681 fn visit_select_item_mut(&mut self, si: &mut SelectItem<Aug>) {
6682 let old = self.in_select_item;
6683 self.in_select_item = true;
6684 visit_mut::visit_select_item_mut(self, si);
6685 self.in_select_item = old;
6686 }
6687}
6688
6689#[derive(Default)]
6690struct WindowFuncCollector {
6691 window_funcs: Vec<Expr<Aug>>,
6692}
6693
6694impl WindowFuncCollector {
6695 fn into_result(self) -> Vec<Expr<Aug>> {
6696 let mut seen = BTreeSet::new();
6698 let window_funcs_dedupped = self
6699 .window_funcs
6700 .into_iter()
6701 .filter(move |expr| seen.insert(expr.clone()))
6702 .rev()
6705 .collect();
6706 window_funcs_dedupped
6707 }
6708}
6709
6710impl Visit<'_, Aug> for WindowFuncCollector {
6711 fn visit_expr(&mut self, expr: &Expr<Aug>) {
6712 match expr {
6713 Expr::Function(func) => {
6714 if func.over.is_some() {
6715 self.window_funcs.push(expr.clone());
6716 }
6717 }
6718 _ => (),
6719 }
6720 visit::visit_expr(self, expr);
6721 }
6722
6723 fn visit_query(&mut self, _query: &Query<Aug>) {
6724 }
6726}
6727
6728#[derive(Debug, Eq, PartialEq, Copy, Clone)]
6730pub enum QueryLifetime {
6731 OneShot,
6733 Index,
6735 MaterializedView,
6737 Subscribe,
6739 View,
6741 Source,
6743}
6744
6745impl QueryLifetime {
6746 pub fn is_one_shot(&self) -> bool {
6750 let result = match self {
6751 QueryLifetime::OneShot => true,
6752 QueryLifetime::Index => false,
6753 QueryLifetime::MaterializedView => false,
6754 QueryLifetime::Subscribe => false,
6755 QueryLifetime::View => false,
6756 QueryLifetime::Source => false,
6757 };
6758 assert_eq!(!result, self.is_maintained());
6759 result
6760 }
6761
6762 pub fn is_maintained(&self) -> bool {
6765 match self {
6766 QueryLifetime::OneShot => false,
6767 QueryLifetime::Index => true,
6768 QueryLifetime::MaterializedView => true,
6769 QueryLifetime::Subscribe => true,
6770 QueryLifetime::View => true,
6771 QueryLifetime::Source => true,
6772 }
6773 }
6774
6775 pub fn allow_show(&self) -> bool {
6777 match self {
6778 QueryLifetime::OneShot => true,
6779 QueryLifetime::Index => false,
6780 QueryLifetime::MaterializedView => false,
6781 QueryLifetime::Subscribe => true, QueryLifetime::View => false,
6783 QueryLifetime::Source => false,
6784 }
6785 }
6786}
6787
6788#[derive(Debug, Clone)]
6790pub struct CteDesc {
6791 pub name: String,
6792 pub desc: RelationDesc,
6793}
6794
6795#[derive(Debug, Clone)]
6797pub struct QueryContext<'a> {
6798 pub scx: &'a StatementContext<'a>,
6800 pub lifetime: QueryLifetime,
6802 pub outer_scopes: Vec<Scope>,
6804 pub outer_relation_types: Vec<SqlRelationType>,
6806 pub ctes: BTreeMap<LocalId, CteDesc>,
6808 pub name_manager: Rc<RefCell<NameManager>>,
6810 pub recursion_guard: RecursionGuard,
6811}
6812
6813impl CheckedRecursion for QueryContext<'_> {
6814 fn recursion_guard(&self) -> &RecursionGuard {
6815 &self.recursion_guard
6816 }
6817}
6818
6819impl<'a> QueryContext<'a> {
6820 pub fn root(scx: &'a StatementContext, lifetime: QueryLifetime) -> QueryContext<'a> {
6821 QueryContext {
6822 scx,
6823 lifetime,
6824 outer_scopes: vec![],
6825 outer_relation_types: vec![],
6826 ctes: BTreeMap::new(),
6827 name_manager: Rc::new(RefCell::new(NameManager::new())),
6828 recursion_guard: RecursionGuard::with_limit(1024), }
6830 }
6831
6832 fn relation_type(&self, expr: &HirRelationExpr) -> SqlRelationType {
6833 expr.typ(&self.outer_relation_types, &self.scx.param_types.borrow())
6834 }
6835
6836 fn derived_context(&self, scope: Scope, relation_type: SqlRelationType) -> QueryContext<'a> {
6839 let ctes = self.ctes.clone();
6840 let outer_scopes = iter::once(scope).chain(self.outer_scopes.clone()).collect();
6841 let outer_relation_types = iter::once(relation_type)
6842 .chain(self.outer_relation_types.clone())
6843 .collect();
6844 let name_manager = Rc::clone(&self.name_manager);
6846
6847 QueryContext {
6848 scx: self.scx,
6849 lifetime: self.lifetime,
6850 outer_scopes,
6851 outer_relation_types,
6852 ctes,
6853 name_manager,
6854 recursion_guard: self.recursion_guard.clone(),
6855 }
6856 }
6857
6858 fn empty_derived_context(&self) -> QueryContext<'a> {
6860 let scope = Scope::empty();
6861 let ty = SqlRelationType::empty();
6862 self.derived_context(scope, ty)
6863 }
6864
6865 pub fn resolve_table_name(
6868 &self,
6869 object: ResolvedItemName,
6870 ) -> Result<(HirRelationExpr, Scope), PlanError> {
6871 match object {
6872 ResolvedItemName::Item {
6873 id,
6874 full_name,
6875 version,
6876 ..
6877 } => {
6878 let item = self.scx.get_item(&id).at_version(version);
6879 let desc = match item.relation_desc() {
6880 Some(desc) => desc.clone(),
6881 None => {
6882 return Err(PlanError::InvalidDependency {
6883 name: full_name.to_string(),
6884 item_type: item.item_type().to_string(),
6885 });
6886 }
6887 };
6888 let expr = HirRelationExpr::Get {
6889 id: Id::Global(item.global_id()),
6890 typ: desc.typ().clone(),
6891 };
6892
6893 let name = full_name.into();
6894 let scope = Scope::from_source(Some(name), desc.iter_names().cloned());
6895
6896 Ok((expr, scope))
6897 }
6898 ResolvedItemName::Cte { id, name } => {
6899 let name = name.into();
6900 let cte = self.ctes.get(&id).unwrap();
6901 let expr = HirRelationExpr::Get {
6902 id: Id::Local(id),
6903 typ: cte.desc.typ().clone(),
6904 };
6905
6906 let scope = Scope::from_source(Some(name), cte.desc.iter_names());
6907
6908 Ok((expr, scope))
6909 }
6910 ResolvedItemName::Error => bail_internal!("should have been caught in name resolution"),
6911 }
6912 }
6913
6914 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6917 self.scx.humanize_sql_scalar_type(typ, postgres_compat)
6918 }
6919}
6920
6921#[derive(Debug, Clone)]
6923pub struct ExprContext<'a> {
6924 pub qcx: &'a QueryContext<'a>,
6925 pub name: &'a str,
6927 pub scope: &'a Scope,
6930 pub relation_type: &'a SqlRelationType,
6933 pub allow_aggregates: bool,
6935 pub allow_subqueries: bool,
6937 pub allow_parameters: bool,
6939 pub allow_windows: bool,
6941}
6942
6943impl CheckedRecursion for ExprContext<'_> {
6944 fn recursion_guard(&self) -> &RecursionGuard {
6945 &self.qcx.recursion_guard
6946 }
6947}
6948
6949impl<'a> ExprContext<'a> {
6950 pub fn catalog(&self) -> &dyn SessionCatalog {
6951 self.qcx.scx.catalog
6952 }
6953
6954 pub fn with_name(&self, name: &'a str) -> ExprContext<'a> {
6955 let mut ecx = self.clone();
6956 ecx.name = name;
6957 ecx
6958 }
6959
6960 pub fn column_type<E>(&self, expr: &E) -> E::Type
6961 where
6962 E: AbstractExpr,
6963 {
6964 expr.typ(
6965 &self.qcx.outer_relation_types,
6966 self.relation_type,
6967 &self.qcx.scx.param_types.borrow(),
6968 )
6969 }
6970
6971 pub fn scalar_type<E>(&self, expr: &E) -> <E::Type as AbstractColumnType>::AbstractScalarType
6972 where
6973 E: AbstractExpr,
6974 {
6975 self.column_type(expr).scalar_type()
6976 }
6977
6978 fn derived_query_context(&self) -> QueryContext<'_> {
6979 let mut scope = self.scope.clone();
6980 scope.lateral_barrier = true;
6981 self.qcx.derived_context(scope, self.relation_type.clone())
6982 }
6983
6984 pub fn require_feature_flag(&self, flag: &'static FeatureFlag) -> Result<(), PlanError> {
6985 self.qcx.scx.require_feature_flag(flag)
6986 }
6987
6988 pub fn param_types(&self) -> &RefCell<BTreeMap<usize, SqlScalarType>> {
6989 &self.qcx.scx.param_types
6990 }
6991
6992 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6995 self.qcx.scx.humanize_sql_scalar_type(typ, postgres_compat)
6996 }
6997
6998 pub fn intern(&self, item: &ScopeItem) -> Arc<str> {
6999 self.qcx.name_manager.borrow_mut().intern_scope_item(item)
7000 }
7001}
7002
7003#[derive(Debug, Clone)]
7009pub struct NameManager(BTreeSet<Arc<str>>);
7010
7011impl NameManager {
7012 pub fn new() -> Self {
7014 Self(BTreeSet::new())
7015 }
7016
7017 fn intern<S: AsRef<str>>(&mut self, s: S) -> Arc<str> {
7020 let s = s.as_ref();
7021 if let Some(interned) = self.0.get(s) {
7022 Arc::clone(interned)
7023 } else {
7024 let interned: Arc<str> = Arc::from(s);
7025 self.0.insert(Arc::clone(&interned));
7026 interned
7027 }
7028 }
7029
7030 pub fn intern_scope_item(&mut self, item: &ScopeItem) -> Arc<str> {
7033 self.intern(item.column_name.as_str())
7049 }
7050}
7051
7052#[cfg(test)]
7053mod test {
7054 use super::*;
7055
7056 #[mz_ore::test]
7061 pub fn test_name_manager_string_interning() {
7062 let mut nm = NameManager::new();
7063
7064 let orig_hi = "hi";
7065 let hi = nm.intern(orig_hi);
7066 let hello = nm.intern("hello");
7067
7068 assert_ne!(hi.as_ptr(), hello.as_ptr());
7069
7070 let hi2 = nm.intern("hi");
7072 assert_eq!(hi.as_ptr(), hi2.as_ptr());
7073
7074 let s = format!(
7076 "{}{}",
7077 hi.chars().nth(0).unwrap(),
7078 hi2.chars().nth(1).unwrap()
7079 );
7080 assert_ne!(orig_hi.as_ptr(), s.as_ptr());
7082
7083 let hi3 = nm.intern(s);
7084 assert_eq!(hi.as_ptr(), hi3.as_ptr());
7085 }
7086}