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 left_clone = lhs.clone();
1978 if *all {
1979 lhs.union(left_clone.union(rhs.negate()).threshold().negate())
1980 } else {
1981 lhs.union(left_clone.union(rhs.negate()).threshold().negate())
1982 .distinct()
1983 }
1984 }
1985 };
1986 let scope = Scope::from_source(
1987 None,
1988 left_scope.column_names(),
1990 );
1991
1992 Ok((relation_expr, scope))
1993 }
1994 SetExpr::Values(Values(values)) => plan_values(qcx, values),
1995 SetExpr::Table(name) => {
1996 let (expr, scope) = qcx.resolve_table_name(name.clone())?;
1997 Ok((expr, scope))
1998 }
1999 SetExpr::Query(query) => {
2000 let (expr, scope) = plan_nested_query(qcx, query)?;
2001 Ok((expr, scope))
2002 }
2003 SetExpr::Show(stmt) => {
2004 if !qcx.lifetime.allow_show() {
2018 return Err(PlanError::ShowCommandInView);
2019 }
2020
2021 fn to_hirscope(
2024 plan: ShowCreatePlan,
2025 desc: StatementDesc,
2026 ) -> Result<(HirRelationExpr, Scope), PlanError> {
2027 let rows = vec![plan.row.iter().collect::<Vec<_>>()];
2028 let desc = desc.relation_desc.ok_or_else(|| {
2029 internal_err!("statement description missing relation descriptor")
2030 })?;
2031 let scope = Scope::from_source(None, desc.iter_names());
2032 let expr = HirRelationExpr::constant(rows, desc.into_typ());
2033 Ok((expr, scope))
2034 }
2035
2036 match stmt.clone() {
2037 ShowStatement::ShowColumns(stmt) => {
2038 show::show_columns(qcx.scx, stmt)?.plan_hir(qcx)
2039 }
2040 ShowStatement::ShowCreateConnection(stmt) => to_hirscope(
2041 show::plan_show_create_connection(qcx.scx, stmt.clone())?,
2042 show::describe_show_create_connection(qcx.scx, stmt)?,
2043 ),
2044 ShowStatement::ShowCreateCluster(stmt) => to_hirscope(
2045 show::plan_show_create_cluster(qcx.scx, stmt.clone())?,
2046 show::describe_show_create_cluster(qcx.scx, stmt)?,
2047 ),
2048 ShowStatement::ShowCreateIndex(stmt) => to_hirscope(
2049 show::plan_show_create_index(qcx.scx, stmt.clone())?,
2050 show::describe_show_create_index(qcx.scx, stmt)?,
2051 ),
2052 ShowStatement::ShowCreateSink(stmt) => to_hirscope(
2053 show::plan_show_create_sink(qcx.scx, stmt.clone())?,
2054 show::describe_show_create_sink(qcx.scx, stmt)?,
2055 ),
2056 ShowStatement::ShowCreateSource(stmt) => to_hirscope(
2057 show::plan_show_create_source(qcx.scx, stmt.clone())?,
2058 show::describe_show_create_source(qcx.scx, stmt)?,
2059 ),
2060 ShowStatement::ShowCreateTable(stmt) => to_hirscope(
2061 show::plan_show_create_table(qcx.scx, stmt.clone())?,
2062 show::describe_show_create_table(qcx.scx, stmt)?,
2063 ),
2064 ShowStatement::ShowCreateView(stmt) => to_hirscope(
2065 show::plan_show_create_view(qcx.scx, stmt.clone())?,
2066 show::describe_show_create_view(qcx.scx, stmt)?,
2067 ),
2068 ShowStatement::ShowCreateMaterializedView(stmt) => to_hirscope(
2069 show::plan_show_create_materialized_view(qcx.scx, stmt.clone())?,
2070 show::describe_show_create_materialized_view(qcx.scx, stmt)?,
2071 ),
2072 ShowStatement::ShowCreateType(stmt) => to_hirscope(
2073 show::plan_show_create_type(qcx.scx, stmt.clone())?,
2074 show::describe_show_create_type(qcx.scx, stmt)?,
2075 ),
2076 ShowStatement::ShowObjects(stmt) => {
2077 show::show_objects(qcx.scx, stmt)?.plan_hir(qcx)
2078 }
2079 ShowStatement::ShowVariable(_) => bail_unsupported!("SHOW variable in subqueries"),
2080 ShowStatement::InspectShard(_) => sql_bail!("unsupported INSPECT statement"),
2081 }
2082 }
2083 }
2084}
2085
2086fn plan_values(
2088 qcx: &QueryContext,
2089 values: &[Vec<Expr<Aug>>],
2090) -> Result<(HirRelationExpr, Scope), PlanError> {
2091 assert!(!values.is_empty());
2092
2093 let ecx = &ExprContext {
2094 qcx,
2095 name: "VALUES",
2096 scope: &Scope::empty(),
2097 relation_type: &SqlRelationType::empty(),
2098 allow_aggregates: false,
2099 allow_subqueries: true,
2100 allow_parameters: true,
2101 allow_windows: false,
2102 };
2103
2104 let ncols = values[0].len();
2105 let nrows = values.len();
2106
2107 let mut cols = vec![vec![]; ncols];
2110 for row in values {
2111 if row.len() != ncols {
2112 sql_bail!(
2113 "VALUES expression has varying number of columns: {} vs {}",
2114 row.len(),
2115 ncols
2116 );
2117 }
2118 for (i, v) in row.iter().enumerate() {
2119 cols[i].push(v);
2120 }
2121 }
2122
2123 let mut col_iters = Vec::with_capacity(ncols);
2125 let mut col_types = Vec::with_capacity(ncols);
2126 for col in &cols {
2127 let col = coerce_homogeneous_exprs(ecx, plan_exprs(ecx, col)?, None)?;
2128 let mut col_type = ecx.column_type(&col[0]);
2129 for val in &col[1..] {
2130 col_type = col_type.sql_union(&ecx.column_type(val))?; }
2132 col_types.push(col_type);
2133 col_iters.push(col.into_iter());
2134 }
2135
2136 let mut exprs = vec![];
2138 for _ in 0..nrows {
2139 for i in 0..ncols {
2140 exprs.push(col_iters[i].next().unwrap());
2141 }
2142 }
2143 let out = HirRelationExpr::CallTable {
2144 func: TableFunc::Wrap {
2145 width: ncols,
2146 types: col_types,
2147 },
2148 exprs,
2149 };
2150
2151 let mut scope = Scope::empty();
2153 for i in 0..ncols {
2154 let name = format!("column{}", i + 1);
2155 scope.items.push(ScopeItem::from_column_name(name));
2156 }
2157
2158 Ok((out, scope))
2159}
2160
2161fn plan_values_insert(
2171 qcx: &QueryContext,
2172 target_names: &[&ColumnName],
2173 target_types: &[&SqlScalarType],
2174 values: &[Vec<Expr<Aug>>],
2175) -> Result<HirRelationExpr, PlanError> {
2176 assert!(!values.is_empty());
2177
2178 if !values.iter().map(|row| row.len()).all_equal() {
2179 sql_bail!("VALUES lists must all be the same length");
2180 }
2181
2182 let ecx = &ExprContext {
2183 qcx,
2184 name: "VALUES",
2185 scope: &Scope::empty(),
2186 relation_type: &SqlRelationType::empty(),
2187 allow_aggregates: false,
2188 allow_subqueries: true,
2189 allow_parameters: true,
2190 allow_windows: false,
2191 };
2192
2193 let mut exprs = vec![];
2194 let mut types = vec![];
2195 for row in values {
2196 if row.len() > target_names.len() {
2197 sql_bail!("INSERT has more expressions than target columns");
2198 }
2199 for (column, val) in row.into_iter().enumerate() {
2200 let target_type = &target_types[column];
2201 let val = plan_expr(ecx, val)?;
2202 let val = typeconv::plan_coerce(ecx, val, target_type)?;
2203 let source_type = &ecx.scalar_type(&val);
2204 let val = match typeconv::plan_cast(ecx, CastContext::Assignment, val, target_type) {
2205 Ok(val) => val,
2206 Err(_) => sql_bail!(
2207 "column {} is of type {} but expression is of type {}",
2208 target_names[column].quoted(),
2209 qcx.humanize_sql_scalar_type(target_type, false),
2210 qcx.humanize_sql_scalar_type(source_type, false),
2211 ),
2212 };
2213 if column >= types.len() {
2214 types.push(ecx.column_type(&val));
2215 } else {
2216 types[column] = types[column].sql_union(&ecx.column_type(&val))?; }
2218 exprs.push(val);
2219 }
2220 }
2221
2222 Ok(HirRelationExpr::CallTable {
2223 func: TableFunc::Wrap {
2224 width: values[0].len(),
2225 types,
2226 },
2227 exprs,
2228 })
2229}
2230
2231fn plan_join_identity() -> (HirRelationExpr, Scope) {
2232 let typ = SqlRelationType::new(vec![]);
2233 let expr = HirRelationExpr::constant(vec![vec![]], typ);
2234 let scope = Scope::empty();
2235 (expr, scope)
2236}
2237
2238#[derive(Debug)]
2244struct SelectPlan {
2245 expr: HirRelationExpr,
2246 scope: Scope,
2247 order_by: Vec<ColumnOrder>,
2248 project: Vec<usize>,
2249}
2250
2251generate_extracted_config!(
2252 SelectOption,
2253 (ExpectedGroupSize, u64),
2254 (AggregateInputGroupSize, u64),
2255 (DistinctOnInputGroupSize, u64),
2256 (LimitInputGroupSize, u64)
2257);
2258
2259fn plan_select_from_where(
2277 qcx: &QueryContext,
2278 mut s: Select<Aug>,
2279 mut order_by_exprs: Vec<OrderByExpr<Aug>>,
2280) -> Result<SelectPlan, PlanError> {
2281 let select_option_extracted = SelectOptionExtracted::try_from(s.options.clone())?;
2288 let group_size_hints = GroupSizeHints::try_from(select_option_extracted)?;
2289
2290 let (mut relation_expr, mut from_scope) =
2292 s.from.iter().try_fold(plan_join_identity(), |l, twj| {
2293 let (left, left_scope) = l;
2294 plan_join(
2295 qcx,
2296 left,
2297 left_scope,
2298 &Join {
2299 relation: TableFactor::NestedJoin {
2300 join: Box::new(twj.clone()),
2301 alias: None,
2302 },
2303 join_operator: JoinOperator::CrossJoin,
2304 },
2305 )
2306 })?;
2307
2308 if let Some(selection) = &s.selection {
2310 let ecx = &ExprContext {
2311 qcx,
2312 name: "WHERE clause",
2313 scope: &from_scope,
2314 relation_type: &qcx.relation_type(&relation_expr),
2315 allow_aggregates: false,
2316 allow_subqueries: true,
2317 allow_parameters: true,
2318 allow_windows: false,
2319 };
2320 let expr = plan_expr(ecx, selection)
2321 .map_err(|e| sql_err!("WHERE clause error: {}", e))?
2322 .type_as(ecx, &SqlScalarType::Bool)?;
2323 relation_expr = relation_expr.filter(vec![expr]);
2324 }
2325
2326 let (aggregates, table_funcs) = {
2329 let mut visitor = AggregateTableFuncVisitor::new(qcx.scx);
2330 visitor.visit_select_mut(&mut s);
2331 for o in order_by_exprs.iter_mut() {
2332 visitor.visit_order_by_expr_mut(o);
2333 }
2334 visitor.into_result()?
2335 };
2336 let mut table_func_names: BTreeMap<String, Ident> = BTreeMap::new();
2337 if !table_funcs.is_empty() {
2338 let (expr, scope) = plan_scalar_table_funcs(
2339 qcx,
2340 table_funcs,
2341 &mut table_func_names,
2342 &relation_expr,
2343 &from_scope,
2344 )?;
2345 relation_expr = relation_expr.join(expr, HirScalarExpr::literal_true(), JoinKind::Inner);
2346 from_scope = from_scope.product(scope)?;
2347 }
2348
2349 let projection = {
2351 let ecx = &ExprContext {
2352 qcx,
2353 name: "SELECT clause",
2354 scope: &from_scope,
2355 relation_type: &qcx.relation_type(&relation_expr),
2356 allow_aggregates: true,
2357 allow_subqueries: true,
2358 allow_parameters: true,
2359 allow_windows: true,
2360 };
2361 let mut out = vec![];
2362 for si in &s.projection {
2363 if *si == SelectItem::Wildcard && s.from.is_empty() {
2364 sql_bail!("SELECT * with no tables specified is not valid");
2365 }
2366 out.extend(expand_select_item(ecx, si, &table_func_names)?);
2367 }
2368 out
2369 };
2370
2371 let (mut group_scope, select_all_mapping) = {
2375 let ecx = &ExprContext {
2377 qcx,
2378 name: "GROUP BY clause",
2379 scope: &from_scope,
2380 relation_type: &qcx.relation_type(&relation_expr),
2381 allow_aggregates: false,
2382 allow_subqueries: true,
2383 allow_parameters: true,
2384 allow_windows: false,
2385 };
2386 let mut group_key = vec![];
2387 let mut group_exprs: BTreeMap<HirScalarExpr, ScopeItem> = BTreeMap::new();
2388 let mut group_hir_exprs = vec![];
2389 let mut group_scope = Scope::empty();
2390 let mut select_all_mapping = BTreeMap::new();
2391
2392 for group_expr in &s.group_by {
2393 let (group_expr, expr) = plan_group_by_expr(ecx, group_expr, &projection)?;
2394 let new_column = group_key.len();
2395
2396 if let Some(existing_scope_item) = group_exprs.get_mut(&expr) {
2407 if let Some(group_expr) = group_expr {
2411 existing_scope_item.exprs.insert(group_expr.clone());
2412 }
2413 continue;
2414 }
2415
2416 let mut scope_item = if let HirScalarExpr::Column(
2417 ColumnRef {
2418 level: 0,
2419 column: old_column,
2420 },
2421 _name,
2422 ) = &expr
2423 {
2424 select_all_mapping.insert(*old_column, new_column);
2430 let scope_item = ecx.scope.items[*old_column].clone();
2431 scope_item
2432 } else {
2433 ScopeItem::empty()
2434 };
2435
2436 if let Some(group_expr) = group_expr.cloned() {
2437 scope_item.exprs.insert(group_expr);
2438 }
2439
2440 group_key.push(from_scope.len() + group_exprs.len());
2441 group_hir_exprs.push(expr.clone());
2442 group_exprs.insert(expr, scope_item);
2443 }
2444
2445 assert_eq!(group_hir_exprs.len(), group_exprs.len());
2446 for expr in &group_hir_exprs {
2447 if let Some(scope_item) = group_exprs.remove(expr) {
2448 group_scope.items.push(scope_item);
2449 }
2450 }
2451
2452 let ecx = &ExprContext {
2454 qcx,
2455 name: "aggregate function",
2456 scope: &from_scope,
2457 relation_type: &qcx.relation_type(&relation_expr.clone().map(group_hir_exprs.clone())),
2458 allow_aggregates: false,
2459 allow_subqueries: true,
2460 allow_parameters: true,
2461 allow_windows: false,
2462 };
2463 let mut agg_exprs = vec![];
2464 for sql_function in aggregates {
2465 if sql_function.over.is_some() {
2466 unreachable!(
2467 "Window aggregate; AggregateTableFuncVisitor explicitly filters these out"
2468 );
2469 }
2470 agg_exprs.push(plan_aggregate_common(ecx, &sql_function)?);
2471 group_scope
2472 .items
2473 .push(ScopeItem::from_expr(Expr::Function(sql_function.clone())));
2474 }
2475 if !agg_exprs.is_empty() || !group_key.is_empty() || s.having.is_some() {
2476 relation_expr = relation_expr.map(group_hir_exprs).reduce(
2478 group_key,
2479 agg_exprs,
2480 group_size_hints.aggregate_input_group_size,
2481 );
2482
2483 for i in 0..from_scope.len() {
2489 if !select_all_mapping.contains_key(&i) {
2490 let scope_item = &ecx.scope.items[i];
2491 group_scope.ungrouped_columns.push(ScopeUngroupedColumn {
2492 table_name: scope_item.table_name.clone(),
2493 column_name: scope_item.column_name.clone(),
2494 allow_unqualified_references: scope_item.allow_unqualified_references,
2495 });
2496 }
2497 }
2498
2499 (group_scope, select_all_mapping)
2500 } else {
2501 (
2503 from_scope.clone(),
2504 (0..from_scope.len()).map(|i| (i, i)).collect(),
2505 )
2506 }
2507 };
2508
2509 if let Some(ref having) = s.having {
2511 let ecx = &ExprContext {
2512 qcx,
2513 name: "HAVING clause",
2514 scope: &group_scope,
2515 relation_type: &qcx.relation_type(&relation_expr),
2516 allow_aggregates: true,
2517 allow_subqueries: true,
2518 allow_parameters: true,
2519 allow_windows: false,
2520 };
2521 let expr = plan_expr(ecx, having)?.type_as(ecx, &SqlScalarType::Bool)?;
2522 relation_expr = relation_expr.filter(vec![expr]);
2523 }
2524
2525 let window_funcs = {
2538 let mut visitor = WindowFuncCollector::default();
2539 visitor.visit_select(&s);
2543 for o in order_by_exprs.iter() {
2544 visitor.visit_order_by_expr(o);
2545 }
2546 visitor.into_result()
2547 };
2548 for window_func in window_funcs {
2549 let ecx = &ExprContext {
2550 qcx,
2551 name: "window function",
2552 scope: &group_scope,
2553 relation_type: &qcx.relation_type(&relation_expr),
2554 allow_aggregates: true,
2555 allow_subqueries: true,
2556 allow_parameters: true,
2557 allow_windows: true,
2558 };
2559 relation_expr = relation_expr.map(vec![plan_expr(ecx, &window_func)?.type_as_any(ecx)?]);
2560 group_scope.items.push(ScopeItem::from_expr(window_func));
2561 }
2562 if let Some(ref qualify) = s.qualify {
2569 let ecx = &ExprContext {
2570 qcx,
2571 name: "QUALIFY clause",
2572 scope: &group_scope,
2573 relation_type: &qcx.relation_type(&relation_expr),
2574 allow_aggregates: true,
2575 allow_subqueries: true,
2576 allow_parameters: true,
2577 allow_windows: true,
2578 };
2579 let expr = plan_expr(ecx, qualify)?.type_as(ecx, &SqlScalarType::Bool)?;
2580 relation_expr = relation_expr.filter(vec![expr]);
2581 }
2582
2583 let output_columns = {
2585 let mut new_exprs = vec![];
2586 let mut new_type = qcx.relation_type(&relation_expr);
2587 let mut output_columns = vec![];
2588 for (select_item, column_name) in &projection {
2589 let ecx = &ExprContext {
2590 qcx,
2591 name: "SELECT clause",
2592 scope: &group_scope,
2593 relation_type: &new_type,
2594 allow_aggregates: true,
2595 allow_subqueries: true,
2596 allow_parameters: true,
2597 allow_windows: true,
2598 };
2599 let expr = match select_item {
2600 ExpandedSelectItem::InputOrdinal(i) => {
2601 if let Some(column) = select_all_mapping.get(i).copied() {
2602 HirScalarExpr::column(column)
2603 } else {
2604 return Err(PlanError::ungrouped_column(&from_scope.items[*i]));
2605 }
2606 }
2607 ExpandedSelectItem::Expr(expr) => plan_expr(ecx, expr)?.type_as_any(ecx)?,
2608 };
2609 if let HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) = expr {
2610 output_columns.push((column, column_name));
2612 } else {
2613 let typ = ecx.column_type(&expr);
2620 new_type.column_types.push(typ);
2621 new_exprs.push(expr);
2622 output_columns.push((group_scope.len(), column_name));
2623 group_scope
2624 .items
2625 .push(ScopeItem::from_expr(select_item.as_expr().cloned()));
2626 }
2627 }
2628 relation_expr = relation_expr.map(new_exprs);
2629 output_columns
2630 };
2631 let mut project_key: Vec<_> = output_columns.iter().map(|(i, _name)| *i).collect();
2632
2633 let order_by = {
2635 let relation_type = qcx.relation_type(&relation_expr);
2636 let (mut order_by, mut map_exprs) = plan_order_by_exprs(
2637 &ExprContext {
2638 qcx,
2639 name: "ORDER BY clause",
2640 scope: &group_scope,
2641 relation_type: &relation_type,
2642 allow_aggregates: true,
2643 allow_subqueries: true,
2644 allow_parameters: true,
2645 allow_windows: true,
2646 },
2647 &order_by_exprs,
2648 &output_columns,
2649 )?;
2650
2651 match s.distinct {
2652 None => relation_expr = relation_expr.map(map_exprs),
2653 Some(Distinct::EntireRow) => {
2654 if relation_type.arity() == 0 {
2655 sql_bail!("SELECT DISTINCT must have at least one column");
2656 }
2657 if !try_push_projection_order_by(
2661 &mut relation_expr,
2662 &mut project_key,
2663 &mut order_by,
2664 ) {
2665 sql_bail!(
2666 "for SELECT DISTINCT, ORDER BY expressions must appear in select list"
2667 );
2668 }
2669 assert!(map_exprs.is_empty());
2670 relation_expr = relation_expr.distinct();
2671 }
2672 Some(Distinct::On(exprs)) => {
2673 let ecx = &ExprContext {
2674 qcx,
2675 name: "DISTINCT ON clause",
2676 scope: &group_scope,
2677 relation_type: &qcx.relation_type(&relation_expr),
2678 allow_aggregates: true,
2679 allow_subqueries: true,
2680 allow_parameters: true,
2681 allow_windows: true,
2682 };
2683
2684 let mut distinct_exprs = vec![];
2685 for expr in &exprs {
2686 let expr = plan_order_by_or_distinct_expr(ecx, expr, &output_columns)?;
2687 distinct_exprs.push(expr);
2688 }
2689
2690 let mut distinct_key = vec![];
2691
2692 let arity = relation_type.arity();
2702 for ord in order_by.iter().take(distinct_exprs.len()) {
2703 let mut expr = &HirScalarExpr::column(ord.column);
2706 if ord.column >= arity {
2707 expr = &map_exprs[ord.column - arity];
2708 };
2709 match distinct_exprs.iter().position(move |e| e == expr) {
2710 None => sql_bail!(
2711 "SELECT DISTINCT ON expressions must match initial ORDER BY expressions"
2712 ),
2713 Some(pos) => {
2714 distinct_exprs.remove(pos);
2715 }
2716 }
2717 distinct_key.push(ord.column);
2718 }
2719
2720 for expr in distinct_exprs {
2722 let column = match expr {
2725 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
2726 _ => {
2727 map_exprs.push(expr);
2728 arity + map_exprs.len() - 1
2729 }
2730 };
2731 distinct_key.push(column);
2732 }
2733
2734 let distinct_len = distinct_key.len();
2739 relation_expr = HirRelationExpr::top_k(
2740 relation_expr.map(map_exprs),
2741 distinct_key,
2742 order_by.iter().skip(distinct_len).cloned().collect(),
2743 Some(HirScalarExpr::literal(
2744 Datum::Int64(1),
2745 SqlScalarType::Int64,
2746 )),
2747 HirScalarExpr::literal(Datum::Int64(0), SqlScalarType::Int64),
2748 group_size_hints.distinct_on_input_group_size,
2749 );
2750 }
2751 }
2752
2753 order_by
2754 };
2755
2756 let scope = Scope::from_source(None, projection.into_iter().map(|(_expr, name)| name));
2761
2762 Ok(SelectPlan {
2763 expr: relation_expr,
2764 scope,
2765 order_by,
2766 project: project_key,
2767 })
2768}
2769
2770fn plan_scalar_table_funcs(
2771 qcx: &QueryContext,
2772 table_funcs: BTreeMap<Function<Aug>, String>,
2773 table_func_names: &mut BTreeMap<String, Ident>,
2774 relation_expr: &HirRelationExpr,
2775 from_scope: &Scope,
2776) -> Result<(HirRelationExpr, Scope), PlanError> {
2777 let rows_from_qcx = qcx.derived_context(from_scope.clone(), qcx.relation_type(relation_expr));
2778
2779 for (table_func, id) in table_funcs.iter() {
2780 table_func_names.insert(
2781 id.clone(),
2782 Ident::new_unchecked(table_func.name.full_item_name().item.clone()),
2784 );
2785 }
2786 if table_funcs.len() == 1 {
2789 let (table_func, id) = table_funcs.iter().next().unwrap();
2790 let (expr, mut scope) =
2791 plan_solitary_table_function(&rows_from_qcx, table_func, None, false)?;
2792
2793 let num_cols = scope.len();
2795 for i in 0..scope.len() {
2796 scope.items[i].table_name = Some(PartialItemName {
2797 database: None,
2798 schema: None,
2799 item: id.clone(),
2800 });
2801 scope.items[i].from_single_column_function = num_cols == 1;
2802 scope.items[i].allow_unqualified_references = false;
2803 }
2804 return Ok((expr, scope));
2805 }
2806 if table_funcs.keys().any(is_repeat_row) {
2807 bail_unsupported!(format!(
2810 "{} in a SELECT clause with multiple table functions",
2811 REPEAT_ROW_NAME
2812 ));
2813 }
2814 let (expr, mut scope, num_cols) =
2816 plan_rows_from_internal(&rows_from_qcx, table_funcs.keys(), None)?;
2817
2818 let mut i = 0;
2820 for (id, num_cols) in table_funcs.values().zip_eq(num_cols) {
2821 for _ in 0..num_cols {
2822 scope.items[i].table_name = Some(PartialItemName {
2823 database: None,
2824 schema: None,
2825 item: id.clone(),
2826 });
2827 scope.items[i].from_single_column_function = num_cols == 1;
2828 scope.items[i].allow_unqualified_references = false;
2829 i += 1;
2830 }
2831 scope.items[i].table_name = Some(PartialItemName {
2835 database: None,
2836 schema: None,
2837 item: id.clone(),
2838 });
2839 scope.items[i].is_exists_column_for_a_table_function_that_was_in_the_target_list = true;
2840 scope.items[i].allow_unqualified_references = false;
2841 i += 1;
2842 }
2843 scope.items[i].allow_unqualified_references = false;
2845 Ok((expr, scope))
2846}
2847
2848fn plan_group_by_expr<'a>(
2855 ecx: &ExprContext,
2856 group_expr: &'a Expr<Aug>,
2857 projection: &'a [(ExpandedSelectItem, ColumnName)],
2858) -> Result<(Option<&'a Expr<Aug>>, HirScalarExpr), PlanError> {
2859 let plan_projection = |column: usize| match &projection[column].0 {
2860 ExpandedSelectItem::InputOrdinal(column) => Ok((None, HirScalarExpr::column(*column))),
2861 ExpandedSelectItem::Expr(expr) => {
2862 Ok((Some(expr.as_ref()), plan_expr(ecx, expr)?.type_as_any(ecx)?))
2863 }
2864 };
2865
2866 if let Some(column) = check_col_index(ecx.name, group_expr, projection.len())? {
2869 return plan_projection(column);
2870 }
2871
2872 match group_expr {
2876 Expr::Identifier(names) => match plan_identifier(ecx, names) {
2877 Err(PlanError::UnknownColumn {
2878 table: None,
2879 column,
2880 similar,
2881 }) => {
2882 let mut iter = projection.iter().map(|(_expr, name)| name);
2885 if let Some(i) = iter.position(|n| *n == column) {
2886 if iter.any(|n| *n == column) {
2887 Err(PlanError::AmbiguousColumn(column))
2888 } else {
2889 plan_projection(i)
2890 }
2891 } else {
2892 Err(PlanError::UnknownColumn {
2895 table: None,
2896 column,
2897 similar,
2898 })
2899 }
2900 }
2901 res => Ok((Some(group_expr), res?)),
2902 },
2903 _ => Ok((
2904 Some(group_expr),
2905 plan_expr(ecx, group_expr)?.type_as_any(ecx)?,
2906 )),
2907 }
2908}
2909
2910pub(crate) fn plan_order_by_exprs(
2918 ecx: &ExprContext,
2919 order_by_exprs: &[OrderByExpr<Aug>],
2920 output_columns: &[(usize, &ColumnName)],
2921) -> Result<(Vec<ColumnOrder>, Vec<HirScalarExpr>), PlanError> {
2922 let mut order_by = vec![];
2923 let mut map_exprs = vec![];
2924 for obe in order_by_exprs {
2925 let expr = plan_order_by_or_distinct_expr(ecx, &obe.expr, output_columns)?;
2926 let column = match expr {
2929 HirScalarExpr::Column(ColumnRef { level: 0, column }, _name) => column,
2930 _ => {
2931 map_exprs.push(expr);
2932 ecx.relation_type.arity() + map_exprs.len() - 1
2933 }
2934 };
2935 order_by.push(resolve_desc_and_nulls_last(obe, column));
2936 }
2937 Ok((order_by, map_exprs))
2938}
2939
2940fn plan_order_by_or_distinct_expr(
2958 ecx: &ExprContext,
2959 expr: &Expr<Aug>,
2960 output_columns: &[(usize, &ColumnName)],
2961) -> Result<HirScalarExpr, PlanError> {
2962 if let Some(i) = check_col_index(ecx.name, expr, output_columns.len())? {
2963 return Ok(HirScalarExpr::column(output_columns[i].0));
2964 }
2965
2966 if let Expr::Identifier(names) = expr {
2967 if let [name] = &names[..] {
2968 let name = normalize::column_name(name.clone());
2969 let mut iter = output_columns.iter().filter(|(_, n)| **n == name);
2970 if let Some((i, _)) = iter.next() {
2971 match iter.next() {
2972 Some((i2, _)) if i != i2 => return Err(PlanError::AmbiguousColumn(name)),
2976 _ => return Ok(HirScalarExpr::column(*i)),
2977 }
2978 }
2979 }
2980 }
2981
2982 plan_expr(ecx, expr)?.type_as_any(ecx)
2983}
2984
2985fn plan_table_with_joins(
2986 qcx: &QueryContext,
2987 table_with_joins: &TableWithJoins<Aug>,
2988) -> Result<(HirRelationExpr, Scope), PlanError> {
2989 let (mut expr, mut scope) = plan_table_factor(qcx, &table_with_joins.relation)?;
2990 for join in &table_with_joins.joins {
2991 let (new_expr, new_scope) = plan_join(qcx, expr, scope, join)?;
2992 expr = new_expr;
2993 scope = new_scope;
2994 }
2995 Ok((expr, scope))
2996}
2997
2998fn plan_table_factor(
2999 qcx: &QueryContext,
3000 table_factor: &TableFactor<Aug>,
3001) -> Result<(HirRelationExpr, Scope), PlanError> {
3002 match table_factor {
3003 TableFactor::Table { name, alias } => {
3004 let (expr, scope) = qcx.resolve_table_name(name.clone())?;
3005 let scope = plan_table_alias(scope, alias.as_ref())?;
3006 Ok((expr, scope))
3007 }
3008
3009 TableFactor::Function {
3010 function,
3011 alias,
3012 with_ordinality,
3013 } => plan_solitary_table_function(qcx, function, alias.as_ref(), *with_ordinality),
3014
3015 TableFactor::RowsFrom {
3016 functions,
3017 alias,
3018 with_ordinality,
3019 } => plan_rows_from(qcx, functions, alias.as_ref(), *with_ordinality),
3020
3021 TableFactor::Derived {
3022 lateral,
3023 subquery,
3024 alias,
3025 } => {
3026 let mut qcx = (*qcx).clone();
3027 if !lateral {
3028 for scope in &mut qcx.outer_scopes {
3032 if scope.lateral_barrier {
3033 break;
3034 }
3035 scope.items.clear();
3036 }
3037 }
3038 qcx.outer_scopes[0].lateral_barrier = true;
3039 let (expr, scope) = plan_nested_query(&mut qcx, subquery)?;
3040 let scope = plan_table_alias(scope, alias.as_ref())?;
3041 Ok((expr, scope))
3042 }
3043
3044 TableFactor::NestedJoin { join, alias } => {
3045 let (expr, scope) = plan_table_with_joins(qcx, join)?;
3046 let scope = plan_table_alias(scope, alias.as_ref())?;
3047 Ok((expr, scope))
3048 }
3049 }
3050}
3051
3052fn plan_rows_from(
3094 qcx: &QueryContext,
3095 functions: &[Function<Aug>],
3096 alias: Option<&TableAlias>,
3097 with_ordinality: bool,
3098) -> Result<(HirRelationExpr, Scope), PlanError> {
3099 if functions.iter().any(is_repeat_row) {
3101 bail_unsupported!(format!("{} in ROWS FROM", REPEAT_ROW_NAME));
3105 }
3106
3107 if let [function] = functions {
3110 return plan_solitary_table_function(qcx, function, alias, with_ordinality);
3111 }
3112
3113 let (expr, mut scope, num_cols) = plan_rows_from_internal(
3117 qcx,
3118 functions,
3119 Some(functions[0].name.full_item_name().clone()),
3120 )?;
3121
3122 let mut columns = Vec::new();
3124 let mut offset = 0;
3125 for (idx, cols) in num_cols.into_iter().enumerate() {
3127 for i in 0..cols {
3128 columns.push(offset + i);
3129 }
3130 offset += cols + 1;
3131
3132 scope.items.remove(offset - idx - 1);
3135 }
3136
3137 if with_ordinality {
3140 columns.push(offset);
3141 } else {
3142 scope.items.pop();
3143 }
3144
3145 let expr = expr.project(columns);
3146
3147 let scope = plan_table_alias(scope, alias)?;
3148 Ok((expr, scope))
3149}
3150
3151fn is_repeat_row(f: &Function<Aug>) -> bool {
3152 f.name.full_name_str().as_str() == format!("{}.{}", MZ_CATALOG_SCHEMA, REPEAT_ROW_NAME)
3153}
3154
3155fn plan_rows_from_internal<'a>(
3178 qcx: &QueryContext,
3179 functions: impl IntoIterator<Item = &'a Function<Aug>>,
3180 table_name: Option<FullItemName>,
3181) -> Result<(HirRelationExpr, Scope, Vec<usize>), PlanError> {
3182 let mut functions = functions.into_iter();
3183 let mut num_cols = Vec::new();
3184
3185 let (mut left_expr, mut left_scope) =
3189 plan_table_function_internal(qcx, functions.next().unwrap(), true, table_name.clone())?;
3190 num_cols.push(left_scope.len() - 1);
3191 left_expr = left_expr.map(vec![HirScalarExpr::column(left_scope.len() - 1)]);
3193 left_scope
3194 .items
3195 .push(ScopeItem::from_column_name(ORDINALITY_COL_NAME));
3196
3197 for function in functions {
3198 let qcx = qcx.empty_derived_context();
3200 let (right_expr, mut right_scope) =
3201 plan_table_function_internal(&qcx, function, true, table_name.clone())?;
3202 num_cols.push(right_scope.len() - 1);
3203 let left_col = left_scope.len() - 1;
3204 let right_col = left_scope.len() + right_scope.len() - 1;
3205 let on = HirScalarExpr::call_binary(
3206 HirScalarExpr::column(left_col),
3207 HirScalarExpr::column(right_col),
3208 expr_func::Eq,
3209 );
3210 left_expr = left_expr
3211 .join(right_expr, on, JoinKind::FullOuter)
3212 .map(vec![HirScalarExpr::call_variadic(
3213 Coalesce,
3214 vec![
3215 HirScalarExpr::column(left_col),
3216 HirScalarExpr::column(right_col),
3217 ],
3218 )]);
3219
3220 left_expr = left_expr.project(
3223 (0..left_col) .chain(left_col + 1..right_col + 2) .collect(),
3226 );
3227 right_scope.items.push(left_scope.items.pop().unwrap());
3229
3230 left_scope.items.extend(right_scope.items);
3231 }
3232
3233 Ok((left_expr, left_scope, num_cols))
3234}
3235
3236fn plan_solitary_table_function(
3240 qcx: &QueryContext,
3241 function: &Function<Aug>,
3242 alias: Option<&TableAlias>,
3243 with_ordinality: bool,
3244) -> Result<(HirRelationExpr, Scope), PlanError> {
3245 let (expr, mut scope) = plan_table_function_internal(qcx, function, with_ordinality, None)?;
3246
3247 let single_column_function = scope.len() == 1 + if with_ordinality { 1 } else { 0 };
3248 if single_column_function {
3249 let item = &mut scope.items[0];
3250
3251 item.from_single_column_function = true;
3254
3255 if let Some(alias) = alias {
3270 if let ScopeItem {
3271 table_name: Some(table_name),
3272 column_name,
3273 ..
3274 } = item
3275 {
3276 if table_name.item.as_str() == column_name.as_str() {
3277 *column_name = normalize::column_name(alias.name.clone());
3278 }
3279 }
3280 }
3281 }
3282
3283 let scope = plan_table_alias(scope, alias)?;
3284 Ok((expr, scope))
3285}
3286
3287fn plan_table_function_internal(
3292 qcx: &QueryContext,
3293 Function {
3294 name,
3295 args,
3296 filter,
3297 over,
3298 distinct,
3299 }: &Function<Aug>,
3300 with_ordinality: bool,
3301 table_name: Option<FullItemName>,
3302) -> Result<(HirRelationExpr, Scope), PlanError> {
3303 if filter.is_some() {
3308 sql_bail!("FILTER is not allowed for table functions in FROM");
3309 }
3310 if over.is_some() {
3311 sql_bail!("OVER is not allowed for table functions in FROM");
3312 }
3313 if *distinct {
3314 sql_bail!("DISTINCT is not allowed for table functions in FROM");
3315 }
3316
3317 let ecx = &ExprContext {
3318 qcx,
3319 name: "table function arguments",
3320 scope: &Scope::empty(),
3321 relation_type: &SqlRelationType::empty(),
3322 allow_aggregates: false,
3323 allow_subqueries: true,
3324 allow_parameters: true,
3325 allow_windows: false,
3326 };
3327
3328 let scalar_args = match args {
3329 FunctionArgs::Star => sql_bail!("{} does not accept * as an argument", name),
3330 FunctionArgs::Args { args, order_by } => {
3331 if !order_by.is_empty() {
3332 sql_bail!(
3333 "ORDER BY specified, but {} is not an aggregate function",
3334 name
3335 );
3336 }
3337 plan_exprs(ecx, args)?
3338 }
3339 };
3340
3341 let table_name = match table_name {
3342 Some(table_name) => table_name.item,
3343 None => name.full_item_name().item.clone(),
3344 };
3345
3346 let scope_name = Some(PartialItemName {
3347 database: None,
3348 schema: None,
3349 item: table_name,
3350 });
3351
3352 let (expr, mut scope) = match resolve_func(ecx, name, args)? {
3353 Func::Table(impls) => {
3354 let tf = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3355 let scope = Scope::from_source(scope_name.clone(), tf.column_names);
3356 let expr = match tf.imp {
3357 TableFuncImpl::CallTable { mut func, exprs } => {
3358 if with_ordinality {
3359 func = TableFunc::with_ordinality(func.clone()).ok_or(
3360 PlanError::Unsupported {
3361 feature: format!("WITH ORDINALITY on {}", func),
3362 discussion_no: None,
3363 },
3364 )?;
3365 }
3366 HirRelationExpr::CallTable { func, exprs }
3367 }
3368 TableFuncImpl::Expr(expr) => {
3369 if !with_ordinality {
3370 expr
3371 } else {
3372 if qcx
3376 .scx
3377 .is_feature_flag_enabled(&ENABLE_WITH_ORDINALITY_LEGACY_FALLBACK)
3378 {
3379 tracing::error!(
3383 %name,
3384 "Using the legacy WITH ORDINALITY / ROWS FROM implementation for a table function",
3385 );
3386 expr.map(vec![HirScalarExpr::windowing(WindowExpr {
3387 func: WindowExprType::Scalar(ScalarWindowExpr {
3388 func: ScalarWindowFunc::RowNumber,
3389 order_by: vec![],
3390 }),
3391 partition_by: vec![],
3392 order_by: vec![],
3393 })])
3394 } else {
3395 bail_unsupported!(format!(
3396 "WITH ORDINALITY or ROWS FROM with {}",
3397 name
3398 ));
3399 }
3400 }
3401 }
3402 };
3403 (expr, scope)
3404 }
3405 Func::Scalar(impls) => {
3406 let expr = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
3407 let output = expr.typ(
3408 &qcx.outer_relation_types,
3409 &SqlRelationType::new(vec![]),
3410 &qcx.scx.param_types.borrow(),
3411 );
3412
3413 let relation = SqlRelationType::new(vec![output]);
3414
3415 let function_ident = Ident::new(name.full_item_name().item.clone())?;
3416 let column_name = normalize::column_name(function_ident);
3417 let name = column_name.to_string();
3418
3419 let scope = Scope::from_source(scope_name.clone(), vec![column_name]);
3420
3421 let mut func = TableFunc::TabletizedScalar { relation, name };
3422 if with_ordinality {
3423 func = TableFunc::with_ordinality(func.clone()).ok_or(PlanError::Unsupported {
3424 feature: format!("WITH ORDINALITY on {}", func),
3425 discussion_no: None,
3426 })?;
3427 }
3428 (
3429 HirRelationExpr::CallTable {
3430 func,
3431 exprs: vec![expr],
3432 },
3433 scope,
3434 )
3435 }
3436 o => sql_bail!(
3437 "{} functions are not supported in functions in FROM",
3438 o.class()
3439 ),
3440 };
3441
3442 if with_ordinality {
3443 scope
3444 .items
3445 .push(ScopeItem::from_name(scope_name, "ordinality"));
3446 }
3447
3448 Ok((expr, scope))
3449}
3450
3451fn plan_table_alias(mut scope: Scope, alias: Option<&TableAlias>) -> Result<Scope, PlanError> {
3452 if let Some(TableAlias {
3453 name,
3454 columns,
3455 strict,
3456 }) = alias
3457 {
3458 if (columns.len() > scope.items.len()) || (*strict && columns.len() != scope.items.len()) {
3459 sql_bail!(
3460 "{} has {} columns available but {} columns specified",
3461 name,
3462 scope.items.len(),
3463 columns.len()
3464 );
3465 }
3466
3467 let table_name = normalize::ident(name.to_owned());
3468 for (i, item) in scope.items.iter_mut().enumerate() {
3469 item.table_name = if item.allow_unqualified_references {
3470 Some(PartialItemName {
3471 database: None,
3472 schema: None,
3473 item: table_name.clone(),
3474 })
3475 } else {
3476 None
3510 };
3511 item.column_name = columns
3512 .get(i)
3513 .map(|a| normalize::column_name(a.clone()))
3514 .unwrap_or_else(|| item.column_name.clone());
3515 }
3516 }
3517 Ok(scope)
3518}
3519
3520fn invent_column_name(
3524 ecx: &ExprContext,
3525 expr: &Expr<Aug>,
3526 table_func_names: &BTreeMap<String, Ident>,
3527) -> Result<Option<ColumnName>, PlanError> {
3528 #[derive(Debug)]
3535 enum NameQuality {
3536 Low,
3537 High,
3538 }
3539
3540 fn invent(
3541 ecx: &ExprContext,
3542 expr: &Expr<Aug>,
3543 table_func_names: &BTreeMap<String, Ident>,
3544 ) -> Result<Option<(ColumnName, NameQuality)>, PlanError> {
3545 Ok(match expr {
3546 Expr::Identifier(names) => {
3547 if let [name] = names.as_slice() {
3548 if let Some(table_func_name) = table_func_names.get(name.as_str()) {
3549 return Ok(Some((
3550 normalize::column_name(table_func_name.clone()),
3551 NameQuality::High,
3552 )));
3553 }
3554 }
3555 names
3556 .last()
3557 .map(|n| (normalize::column_name(n.clone()), NameQuality::High))
3558 }
3559 Expr::Value(v) => match v {
3560 Value::Boolean(_) => Some(("bool".into(), NameQuality::High)),
3563 Value::Interval(_) => Some(("interval".into(), NameQuality::High)),
3564 _ => None,
3565 },
3566 Expr::Function(func) => {
3567 let (schema, item) = match &func.name {
3568 ResolvedItemName::Item {
3569 qualifiers,
3570 full_name,
3571 ..
3572 } => (&qualifiers.schema_spec, full_name.item.clone()),
3573 _ => {
3576 bail_internal!("function name did not resolve to an item: {:?}", func.name)
3577 }
3578 };
3579
3580 if schema == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_internal_schema_id())
3581 || schema
3582 == &SchemaSpecifier::from(ecx.qcx.scx.catalog.get_mz_unsafe_schema_id())
3583 {
3584 None
3585 } else {
3586 Some((item.into(), NameQuality::High))
3587 }
3588 }
3589 Expr::HomogenizingFunction { function, .. } => Some((
3590 function.to_string().to_lowercase().into(),
3591 NameQuality::High,
3592 )),
3593 Expr::NullIf { .. } => Some(("nullif".into(), NameQuality::High)),
3594 Expr::Array { .. } => Some(("array".into(), NameQuality::High)),
3595 Expr::List { .. } => Some(("list".into(), NameQuality::High)),
3596 Expr::Map { .. } | Expr::MapSubquery(_) => Some(("map".into(), NameQuality::High)),
3597 Expr::Cast { expr, data_type } => match invent(ecx, expr, table_func_names)? {
3598 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3599 _ => Some((data_type.unqualified_item_name().into(), NameQuality::Low)),
3600 },
3601 Expr::Case { else_result, .. } => {
3602 let inner = match else_result.as_ref() {
3603 Some(else_result) => invent(ecx, else_result, table_func_names)?,
3604 None => None,
3605 };
3606 match inner {
3607 Some((name, NameQuality::High)) => Some((name, NameQuality::High)),
3608 _ => Some(("case".into(), NameQuality::Low)),
3609 }
3610 }
3611 Expr::FieldAccess { field, .. } => {
3612 Some((normalize::column_name(field.clone()), NameQuality::High))
3613 }
3614 Expr::Exists { .. } => Some(("exists".into(), NameQuality::High)),
3615 Expr::Subscript { expr, .. } => invent(ecx, expr, table_func_names)?,
3616 Expr::Subquery(query) | Expr::ListSubquery(query) | Expr::ArraySubquery(query) => {
3617 let Ok((_expr, scope)) = plan_nested_query(&mut ecx.derived_query_context(), query)
3625 else {
3626 return Ok(None);
3627 };
3628 scope
3629 .items
3630 .first()
3631 .map(|name| (name.column_name.clone(), NameQuality::High))
3632 }
3633 Expr::Row { .. } => Some(("row".into(), NameQuality::High)),
3634 _ => None,
3635 })
3636 }
3637
3638 Ok(invent(ecx, expr, table_func_names)?.map(|(name, _quality)| name))
3639}
3640
3641#[derive(Debug)]
3642enum ExpandedSelectItem<'a> {
3643 InputOrdinal(usize),
3644 Expr(Cow<'a, Expr<Aug>>),
3645}
3646
3647impl ExpandedSelectItem<'_> {
3648 fn as_expr(&self) -> Option<&Expr<Aug>> {
3649 match self {
3650 ExpandedSelectItem::InputOrdinal(_) => None,
3651 ExpandedSelectItem::Expr(expr) => Some(expr),
3652 }
3653 }
3654}
3655
3656fn expand_select_item<'a>(
3657 ecx: &ExprContext,
3658 s: &'a SelectItem<Aug>,
3659 table_func_names: &BTreeMap<String, Ident>,
3660) -> Result<Vec<(ExpandedSelectItem<'a>, ColumnName)>, PlanError> {
3661 match s {
3662 SelectItem::Expr {
3663 expr: Expr::QualifiedWildcard(table_name),
3664 alias: _,
3665 } => {
3666 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3667 let table_name =
3668 normalize::unresolved_item_name(UnresolvedItemName(table_name.clone()))?;
3669 let out: Vec<_> = ecx
3670 .scope
3671 .items
3672 .iter()
3673 .enumerate()
3674 .filter(|(_i, item)| item.is_from_table(&table_name))
3675 .map(|(i, item)| {
3676 let name = item.column_name.clone();
3677 (ExpandedSelectItem::InputOrdinal(i), name)
3678 })
3679 .collect();
3680 if out.is_empty() {
3681 sql_bail!("no table named '{}' in scope", table_name);
3682 }
3683 Ok(out)
3684 }
3685 SelectItem::Expr {
3686 expr: Expr::WildcardAccess(sql_expr),
3687 alias: _,
3688 } => {
3689 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3690 let expr = plan_expr(ecx, sql_expr)?.type_as_any(ecx)?;
3696 let fields = match ecx.scalar_type(&expr) {
3697 SqlScalarType::Record { fields, .. } => fields,
3698 ty => sql_bail!(
3699 "type {} is not composite",
3700 ecx.humanize_sql_scalar_type(&ty, false)
3701 ),
3702 };
3703 let mut skip_cols: BTreeSet<ColumnName> = BTreeSet::new();
3704 if let Expr::Identifier(ident) = sql_expr.as_ref() {
3705 if let [name] = ident.as_slice() {
3706 if let Ok(items) = ecx.scope.items_from_table(
3707 &[],
3708 &PartialItemName {
3709 database: None,
3710 schema: None,
3711 item: name.as_str().to_string(),
3712 },
3713 ) {
3714 for (_, item) in items {
3715 if item
3716 .is_exists_column_for_a_table_function_that_was_in_the_target_list
3717 {
3718 skip_cols.insert(item.column_name.clone());
3719 }
3720 }
3721 }
3722 }
3723 }
3724 let items = fields
3725 .iter()
3726 .filter_map(|(name, _ty)| {
3727 if skip_cols.contains(name) {
3728 None
3729 } else {
3730 let item = ExpandedSelectItem::Expr(Cow::Owned(Expr::FieldAccess {
3731 expr: sql_expr.clone(),
3732 field: name.clone().into(),
3733 }));
3734 Some((item, name.clone()))
3735 }
3736 })
3737 .collect();
3738 Ok(items)
3739 }
3740 SelectItem::Wildcard => {
3741 *ecx.qcx.scx.ambiguous_columns.borrow_mut() = true;
3742 let items: Vec<_> = ecx
3743 .scope
3744 .items
3745 .iter()
3746 .enumerate()
3747 .filter(|(_i, item)| item.allow_unqualified_references)
3748 .map(|(i, item)| {
3749 let name = item.column_name.clone();
3750 (ExpandedSelectItem::InputOrdinal(i), name)
3751 })
3752 .collect();
3753
3754 Ok(items)
3755 }
3756 SelectItem::Expr { expr, alias } => {
3757 let name = match alias.clone().map(normalize::column_name) {
3758 Some(name) => name,
3759 None => invent_column_name(ecx, expr, table_func_names)?
3760 .unwrap_or_else(|| UNKNOWN_COLUMN_NAME.into()),
3761 };
3762 Ok(vec![(ExpandedSelectItem::Expr(Cow::Borrowed(expr)), name)])
3763 }
3764 }
3765}
3766
3767fn plan_join(
3768 left_qcx: &QueryContext,
3769 left: HirRelationExpr,
3770 left_scope: Scope,
3771 join: &Join<Aug>,
3772) -> Result<(HirRelationExpr, Scope), PlanError> {
3773 const ON_TRUE: JoinConstraint<Aug> = JoinConstraint::On(Expr::Value(Value::Boolean(true)));
3774 let (kind, constraint) = match &join.join_operator {
3775 JoinOperator::CrossJoin => (JoinKind::Inner, &ON_TRUE),
3776 JoinOperator::Inner(constraint) => (JoinKind::Inner, constraint),
3777 JoinOperator::LeftOuter(constraint) => (JoinKind::LeftOuter, constraint),
3778 JoinOperator::RightOuter(constraint) => (JoinKind::RightOuter, constraint),
3779 JoinOperator::FullOuter(constraint) => (JoinKind::FullOuter, constraint),
3780 };
3781
3782 let mut right_qcx = left_qcx.derived_context(left_scope.clone(), left_qcx.relation_type(&left));
3783 if !kind.can_be_correlated() {
3784 for item in &mut right_qcx.outer_scopes[0].items {
3785 item.error_if_referenced =
3790 Some(|table, column| PlanError::WrongJoinTypeForLateralColumn {
3791 table: table.cloned(),
3792 column: column.clone(),
3793 });
3794 }
3795 }
3796 let (right, right_scope) = plan_table_factor(&right_qcx, &join.relation)?;
3797
3798 let (expr, scope) = match constraint {
3799 JoinConstraint::On(expr) => {
3800 let product_scope = left_scope.product(right_scope)?;
3801 let ecx = &ExprContext {
3802 qcx: left_qcx,
3803 name: "ON clause",
3804 scope: &product_scope,
3805 relation_type: &SqlRelationType::new(
3806 left_qcx
3807 .relation_type(&left)
3808 .column_types
3809 .into_iter()
3810 .chain(right_qcx.relation_type(&right).column_types)
3811 .collect(),
3812 ),
3813 allow_aggregates: false,
3814 allow_subqueries: true,
3815 allow_parameters: true,
3816 allow_windows: false,
3817 };
3818 let on = plan_expr(ecx, expr)?.type_as(ecx, &SqlScalarType::Bool)?;
3819 let joined = left.join(right, on, kind);
3820 (joined, product_scope)
3821 }
3822 JoinConstraint::Using { columns, alias } => {
3823 let column_names = columns
3824 .iter()
3825 .map(|ident| normalize::column_name(ident.clone()))
3826 .collect::<Vec<_>>();
3827
3828 plan_using_constraint(
3829 &column_names,
3830 left_qcx,
3831 left,
3832 left_scope,
3833 &right_qcx,
3834 right,
3835 right_scope,
3836 kind,
3837 alias.as_ref(),
3838 )?
3839 }
3840 JoinConstraint::Natural => {
3841 *left_qcx.scx.ambiguous_columns.borrow_mut() = true;
3844 *right_qcx.scx.ambiguous_columns.borrow_mut() = true;
3845 let left_column_names = left_scope.column_names();
3846 let right_column_names: BTreeSet<_> = right_scope.column_names().collect();
3847 let column_names: Vec<_> = left_column_names
3848 .filter(|col| right_column_names.contains(col))
3849 .cloned()
3850 .collect();
3851 plan_using_constraint(
3852 &column_names,
3853 left_qcx,
3854 left,
3855 left_scope,
3856 &right_qcx,
3857 right,
3858 right_scope,
3859 kind,
3860 None,
3861 )?
3862 }
3863 };
3864 Ok((expr, scope))
3865}
3866
3867#[allow(clippy::too_many_arguments)]
3869fn plan_using_constraint(
3870 column_names: &[ColumnName],
3871 left_qcx: &QueryContext,
3872 left: HirRelationExpr,
3873 left_scope: Scope,
3874 right_qcx: &QueryContext,
3875 right: HirRelationExpr,
3876 right_scope: Scope,
3877 kind: JoinKind,
3878 alias: Option<&Ident>,
3879) -> Result<(HirRelationExpr, Scope), PlanError> {
3880 let mut both_scope = left_scope.clone().product(right_scope.clone())?;
3881
3882 let mut unique_column_names = BTreeSet::new();
3885 for c in column_names {
3886 if !unique_column_names.insert(c) {
3887 return Err(PlanError::Unsupported {
3888 feature: format!(
3889 "column name {} appears more than once in USING clause",
3890 c.quoted()
3891 ),
3892 discussion_no: None,
3893 });
3894 }
3895 }
3896
3897 let alias_item_name = alias.map(|alias| PartialItemName {
3898 database: None,
3899 schema: None,
3900 item: alias.clone().to_string(),
3901 });
3902
3903 if let Some(alias_item_name) = &alias_item_name {
3904 for partial_item_name in both_scope.table_names() {
3905 if partial_item_name.matches(alias_item_name) {
3906 sql_bail!(
3907 "table name \"{}\" specified more than once",
3908 alias_item_name
3909 )
3910 }
3911 }
3912 }
3913
3914 let ecx = &ExprContext {
3915 qcx: right_qcx,
3916 name: "USING clause",
3917 scope: &both_scope,
3918 relation_type: &SqlRelationType::new(
3919 left_qcx
3920 .relation_type(&left)
3921 .column_types
3922 .into_iter()
3923 .chain(right_qcx.relation_type(&right).column_types)
3924 .collect(),
3925 ),
3926 allow_aggregates: false,
3927 allow_subqueries: false,
3928 allow_parameters: false,
3929 allow_windows: false,
3930 };
3931
3932 let mut join_exprs = vec![];
3933 let mut map_exprs = vec![];
3934 let mut new_items = vec![];
3935 let mut join_cols = vec![];
3936 let mut hidden_cols = vec![];
3937
3938 for column_name in column_names {
3939 let (lhs, lhs_name) = left_scope.resolve_using_column(
3941 column_name,
3942 JoinSide::Left,
3943 &mut left_qcx.name_manager.borrow_mut(),
3944 )?;
3945 let (mut rhs, rhs_name) = right_scope.resolve_using_column(
3946 column_name,
3947 JoinSide::Right,
3948 &mut right_qcx.name_manager.borrow_mut(),
3949 )?;
3950
3951 rhs.column += left_scope.len();
3953
3954 let mut exprs = coerce_homogeneous_exprs(
3956 &ecx.with_name(&format!(
3957 "NATURAL/USING join column {}",
3958 column_name.quoted()
3959 )),
3960 vec![
3961 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
3962 lhs,
3963 Arc::clone(&lhs_name),
3964 )),
3965 CoercibleScalarExpr::Coerced(HirScalarExpr::named_column(
3966 rhs,
3967 Arc::clone(&rhs_name),
3968 )),
3969 ],
3970 None,
3971 )?;
3972 let (expr1, expr2) = (exprs.remove(0), exprs.remove(0));
3973
3974 match kind {
3975 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
3976 join_cols.push(lhs.column);
3977 hidden_cols.push(rhs.column);
3978 }
3979 JoinKind::RightOuter => {
3980 join_cols.push(rhs.column);
3981 hidden_cols.push(lhs.column);
3982 }
3983 JoinKind::FullOuter => {
3984 join_cols.push(both_scope.items.len() + map_exprs.len());
3987 hidden_cols.push(lhs.column);
3988 hidden_cols.push(rhs.column);
3989 map_exprs.push(HirScalarExpr::call_variadic(
3990 Coalesce,
3991 vec![expr1.clone(), expr2.clone()],
3992 ));
3993 new_items.push(ScopeItem::from_column_name(column_name));
3994 }
3995 }
3996
3997 if alias_item_name.is_some() {
4002 let new_item_col = both_scope.items.len() + new_items.len();
4003 join_cols.push(new_item_col);
4004 hidden_cols.push(new_item_col);
4005
4006 new_items.push(ScopeItem::from_name(
4007 alias_item_name.clone(),
4008 column_name.clone().to_string(),
4009 ));
4010
4011 let alias_expr = match kind {
4018 JoinKind::LeftOuter { .. } | JoinKind::Inner { .. } => {
4019 HirScalarExpr::named_column(lhs, Arc::clone(&lhs_name))
4020 }
4021 JoinKind::RightOuter => HirScalarExpr::named_column(rhs, Arc::clone(&rhs_name)),
4022 JoinKind::FullOuter => {
4023 HirScalarExpr::call_variadic(Coalesce, vec![expr1.clone(), expr2.clone()])
4024 }
4025 };
4026 map_exprs.push(alias_expr);
4027 }
4028
4029 join_exprs.push(expr1.call_binary(expr2, expr_func::Eq));
4030 }
4031 both_scope.items.extend(new_items);
4032
4033 for c in hidden_cols {
4037 both_scope.items[c].allow_unqualified_references = false;
4038 }
4039
4040 let project_key = join_cols
4042 .into_iter()
4043 .chain(0..both_scope.items.len())
4044 .unique()
4045 .collect::<Vec<_>>();
4046
4047 both_scope = both_scope.project(&project_key);
4048
4049 let on = HirScalarExpr::variadic_and(join_exprs);
4050
4051 let both = left
4052 .join(right, on, kind)
4053 .map(map_exprs)
4054 .project(project_key);
4055 Ok((both, both_scope))
4056}
4057
4058pub fn plan_expr<'a>(
4059 ecx: &'a ExprContext,
4060 e: &Expr<Aug>,
4061) -> Result<CoercibleScalarExpr, PlanError> {
4062 ecx.checked_recur(|ecx| plan_expr_inner(ecx, e))
4063}
4064
4065fn plan_expr_inner<'a>(
4066 ecx: &'a ExprContext,
4067 e: &Expr<Aug>,
4068) -> Result<CoercibleScalarExpr, PlanError> {
4069 if let Some((i, item)) = ecx.scope.resolve_expr(e) {
4070 return Ok(HirScalarExpr::named_column(
4072 i,
4073 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
4074 )
4075 .into());
4076 }
4077
4078 match e {
4079 Expr::Identifier(names) | Expr::QualifiedWildcard(names) => {
4081 Ok(plan_identifier(ecx, names)?.into())
4082 }
4083
4084 Expr::Value(val) => plan_literal(val),
4086 Expr::Parameter(n) => plan_parameter(ecx, *n),
4087 Expr::Array(exprs) => plan_array(ecx, exprs, None),
4088 Expr::List(exprs) => plan_list(ecx, exprs, None),
4089 Expr::Map(exprs) => plan_map(ecx, exprs, None),
4090 Expr::Row { exprs } => plan_row(ecx, exprs),
4091
4092 Expr::Op { op, expr1, expr2 } => {
4094 Ok(plan_op(ecx, normalize::op(op)?, expr1, expr2.as_deref())?.into())
4095 }
4096 Expr::Cast { expr, data_type } => plan_cast(ecx, expr, data_type),
4097 Expr::Function(func) => Ok(plan_function(ecx, func)?.into()),
4098
4099 Expr::Not { expr } => plan_not(ecx, expr),
4101 Expr::And { left, right } => plan_and(ecx, left, right),
4102 Expr::Or { left, right } => plan_or(ecx, left, right),
4103 Expr::IsExpr {
4104 expr,
4105 construct,
4106 negated,
4107 } => Ok(plan_is_expr(ecx, expr, construct, *negated)?.into()),
4108 Expr::Case {
4109 operand,
4110 conditions,
4111 results,
4112 else_result,
4113 } => Ok(plan_case(ecx, operand, conditions, results, else_result)?.into()),
4114 Expr::HomogenizingFunction { function, exprs } => {
4115 plan_homogenizing_function(ecx, function, exprs)
4116 }
4117 Expr::NullIf { l_expr, r_expr } => Ok(plan_case(
4118 ecx,
4119 &None,
4120 &[l_expr.clone().equals(*r_expr.clone())],
4121 &[Expr::null()],
4122 &Some(Box::new(*l_expr.clone())),
4123 )?
4124 .into()),
4125 Expr::FieldAccess { expr, field } => plan_field_access(ecx, expr, field),
4126 Expr::WildcardAccess(expr) => plan_expr(ecx, expr),
4127 Expr::Subscript { expr, positions } => plan_subscript(ecx, expr, positions),
4128 Expr::Like {
4129 expr,
4130 pattern,
4131 escape,
4132 case_insensitive,
4133 negated,
4134 } => Ok(plan_like(
4135 ecx,
4136 expr,
4137 pattern,
4138 escape.as_deref(),
4139 *case_insensitive,
4140 *negated,
4141 )?
4142 .into()),
4143
4144 Expr::InList {
4145 expr,
4146 list,
4147 negated,
4148 } => plan_in_list(ecx, expr, list, negated),
4149
4150 Expr::Exists(query) => plan_exists(ecx, query),
4152 Expr::Subquery(query) => plan_subquery(ecx, query),
4153 Expr::ListSubquery(query) => plan_list_subquery(ecx, query),
4154 Expr::MapSubquery(query) => plan_map_subquery(ecx, query),
4155 Expr::ArraySubquery(query) => plan_array_subquery(ecx, query),
4156 Expr::Collate { expr, collation } => plan_collate(ecx, expr, collation),
4157 Expr::Nested(_) => bail_internal!("Expr::Nested should have been desugared"),
4158 Expr::InSubquery { .. } => {
4159 bail_internal!("Expr::InSubquery should have been desugared")
4160 }
4161 Expr::AnyExpr { .. } => {
4162 bail_internal!("Expr::AnyExpr should have been desugared")
4163 }
4164 Expr::AllExpr { .. } => {
4165 bail_internal!("Expr::AllExpr should have been desugared")
4166 }
4167 Expr::AnySubquery { .. } => {
4168 bail_internal!("Expr::AnySubquery should have been desugared")
4169 }
4170 Expr::AllSubquery { .. } => {
4171 bail_internal!("Expr::AllSubquery should have been desugared")
4172 }
4173 Expr::Between { .. } => {
4174 bail_internal!("Expr::Between should have been desugared")
4175 }
4176 }
4177}
4178
4179fn plan_parameter(ecx: &ExprContext, n: usize) -> Result<CoercibleScalarExpr, PlanError> {
4180 if !ecx.allow_parameters {
4181 return Err(PlanError::UnknownParameter(n));
4185 }
4186 if n == 0 || n > 65536 {
4187 return Err(PlanError::UnknownParameter(n));
4188 }
4189 if ecx.param_types().borrow().contains_key(&n) {
4190 Ok(HirScalarExpr::parameter(n).into())
4191 } else {
4192 Ok(CoercibleScalarExpr::Parameter(n))
4193 }
4194}
4195
4196fn plan_row(ecx: &ExprContext, exprs: &[Expr<Aug>]) -> Result<CoercibleScalarExpr, PlanError> {
4197 let mut out = vec![];
4198 for e in exprs {
4199 out.push(plan_expr(ecx, e)?);
4200 }
4201 Ok(CoercibleScalarExpr::LiteralRecord(out))
4202}
4203
4204fn plan_cast(
4205 ecx: &ExprContext,
4206 expr: &Expr<Aug>,
4207 data_type: &ResolvedDataType,
4208) -> Result<CoercibleScalarExpr, PlanError> {
4209 let to_scalar_type = scalar_type_from_sql(ecx.qcx.scx, data_type)?;
4210 let expr = match expr {
4211 Expr::Array(exprs) => plan_array(ecx, exprs, Some(&to_scalar_type))?,
4220 Expr::List(exprs) => plan_list(ecx, exprs, Some(&to_scalar_type))?,
4221 Expr::Map(exprs) => plan_map(ecx, exprs, Some(&to_scalar_type))?,
4222 _ => plan_expr(ecx, expr)?,
4223 };
4224 let ecx = &ecx.with_name("CAST");
4225 let expr = typeconv::plan_coerce(ecx, expr, &to_scalar_type)?;
4226 let expr = typeconv::plan_cast(ecx, CastContext::Explicit, expr, &to_scalar_type)?;
4227 Ok(expr.into())
4228}
4229
4230fn plan_not(ecx: &ExprContext, expr: &Expr<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4231 let ecx = ecx.with_name("NOT argument");
4232 Ok(plan_expr(&ecx, expr)?
4233 .type_as(&ecx, &SqlScalarType::Bool)?
4234 .call_unary(UnaryFunc::Not(expr_func::Not))
4235 .into())
4236}
4237
4238fn plan_and(
4239 ecx: &ExprContext,
4240 left: &Expr<Aug>,
4241 right: &Expr<Aug>,
4242) -> Result<CoercibleScalarExpr, PlanError> {
4243 let ecx = ecx.with_name("AND argument");
4244 Ok(HirScalarExpr::variadic_and(vec![
4245 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4246 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4247 ])
4248 .into())
4249}
4250
4251fn plan_or(
4252 ecx: &ExprContext,
4253 left: &Expr<Aug>,
4254 right: &Expr<Aug>,
4255) -> Result<CoercibleScalarExpr, PlanError> {
4256 let ecx = ecx.with_name("OR argument");
4257 Ok(HirScalarExpr::variadic_or(vec![
4258 plan_expr(&ecx, left)?.type_as(&ecx, &SqlScalarType::Bool)?,
4259 plan_expr(&ecx, right)?.type_as(&ecx, &SqlScalarType::Bool)?,
4260 ])
4261 .into())
4262}
4263
4264fn plan_in_list(
4265 ecx: &ExprContext,
4266 lhs: &Expr<Aug>,
4267 list: &Vec<Expr<Aug>>,
4268 negated: &bool,
4269) -> Result<CoercibleScalarExpr, PlanError> {
4270 let ecx = ecx.with_name("IN list");
4271 let or = HirScalarExpr::variadic_or(
4272 list.into_iter()
4273 .map(|e| {
4274 let eq = lhs.clone().equals(e.clone());
4275 plan_expr(&ecx, &eq)?.type_as(&ecx, &SqlScalarType::Bool)
4276 })
4277 .collect::<Result<Vec<HirScalarExpr>, PlanError>>()?,
4278 );
4279 Ok(if *negated {
4280 or.call_unary(UnaryFunc::Not(expr_func::Not))
4281 } else {
4282 or
4283 }
4284 .into())
4285}
4286
4287fn plan_homogenizing_function(
4288 ecx: &ExprContext,
4289 function: &HomogenizingFunction,
4290 exprs: &[Expr<Aug>],
4291) -> Result<CoercibleScalarExpr, PlanError> {
4292 assert!(!exprs.is_empty()); let expr = HirScalarExpr::call_variadic(
4294 match function {
4295 HomogenizingFunction::Coalesce => VariadicFunc::from(Coalesce),
4296 HomogenizingFunction::Greatest => VariadicFunc::from(Greatest),
4297 HomogenizingFunction::Least => VariadicFunc::from(Least),
4298 },
4299 coerce_homogeneous_exprs(
4300 &ecx.with_name(&function.to_string().to_lowercase()),
4301 plan_exprs(ecx, exprs)?,
4302 None,
4303 )?,
4304 );
4305 Ok(expr.into())
4306}
4307
4308fn plan_field_access(
4309 ecx: &ExprContext,
4310 expr: &Expr<Aug>,
4311 field: &Ident,
4312) -> Result<CoercibleScalarExpr, PlanError> {
4313 let field = normalize::column_name(field.clone());
4314 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4315 let ty = ecx.scalar_type(&expr);
4316 let i = match &ty {
4317 SqlScalarType::Record { fields, .. } => {
4318 fields.iter().position(|(name, _ty)| *name == field)
4319 }
4320 ty => sql_bail!(
4321 "column notation applied to type {}, which is not a composite type",
4322 ecx.humanize_sql_scalar_type(ty, false)
4323 ),
4324 };
4325 match i {
4326 None => sql_bail!(
4327 "field {} not found in data type {}",
4328 field,
4329 ecx.humanize_sql_scalar_type(&ty, false)
4330 ),
4331 Some(i) => Ok(expr
4332 .call_unary(UnaryFunc::RecordGet(expr_func::RecordGet(i)))
4333 .into()),
4334 }
4335}
4336
4337fn plan_subscript(
4338 ecx: &ExprContext,
4339 expr: &Expr<Aug>,
4340 positions: &[SubscriptPosition<Aug>],
4341) -> Result<CoercibleScalarExpr, PlanError> {
4342 assert!(
4343 !positions.is_empty(),
4344 "subscript expression must contain at least one position"
4345 );
4346
4347 let ecx = &ecx.with_name("subscripting");
4348 let expr = plan_expr(ecx, expr)?.type_as_any(ecx)?;
4349 let ty = ecx.scalar_type(&expr);
4350 match &ty {
4351 SqlScalarType::Array(..) | SqlScalarType::Int2Vector => plan_subscript_array(
4352 ecx,
4353 expr,
4354 positions,
4355 if ty == SqlScalarType::Int2Vector {
4359 1
4360 } else {
4361 0
4362 },
4363 ),
4364 SqlScalarType::Jsonb => plan_subscript_jsonb(ecx, expr, positions),
4365 SqlScalarType::List { element_type, .. } => {
4366 let elem_type_name = ecx.humanize_sql_scalar_type(element_type, false);
4368 let n_layers = ty.unwrap_list_n_layers();
4369 plan_subscript_list(ecx, expr, positions, n_layers, &elem_type_name)
4370 }
4371 ty => sql_bail!(
4372 "cannot subscript type {}",
4373 ecx.humanize_sql_scalar_type(ty, false)
4374 ),
4375 }
4376}
4377
4378fn extract_scalar_subscript_from_positions<'a>(
4382 positions: &'a [SubscriptPosition<Aug>],
4383 expr_type_name: &str,
4384) -> Result<Vec<&'a Expr<Aug>>, PlanError> {
4385 let mut scalar_subscripts = Vec::with_capacity(positions.len());
4386 for p in positions {
4387 if p.explicit_slice {
4388 sql_bail!("{} subscript does not support slices", expr_type_name);
4389 }
4390 assert!(
4391 p.end.is_none(),
4392 "index-appearing subscripts cannot have end value"
4393 );
4394 scalar_subscripts.push(p.start.as_ref().expect("has start if not slice"));
4395 }
4396 Ok(scalar_subscripts)
4397}
4398
4399fn plan_subscript_array(
4400 ecx: &ExprContext,
4401 expr: HirScalarExpr,
4402 positions: &[SubscriptPosition<Aug>],
4403 offset: i64,
4404) -> Result<CoercibleScalarExpr, PlanError> {
4405 let mut exprs = Vec::with_capacity(positions.len() + 1);
4406 exprs.push(expr);
4407
4408 let indexes = extract_scalar_subscript_from_positions(positions, "array")?;
4411
4412 for i in indexes {
4413 exprs.push(plan_expr(ecx, i)?.cast_to(
4414 ecx,
4415 CastContext::Explicit,
4416 &SqlScalarType::Int64,
4417 )?);
4418 }
4419
4420 Ok(HirScalarExpr::call_variadic(ArrayIndex { offset }, exprs).into())
4421}
4422
4423fn plan_subscript_list(
4424 ecx: &ExprContext,
4425 mut expr: HirScalarExpr,
4426 positions: &[SubscriptPosition<Aug>],
4427 mut remaining_layers: usize,
4428 elem_type_name: &str,
4429) -> Result<CoercibleScalarExpr, PlanError> {
4430 let mut i = 0;
4431
4432 while i < positions.len() {
4433 let j = positions[i..]
4435 .iter()
4436 .position(|p| p.explicit_slice)
4437 .unwrap_or(positions.len() - i);
4438 if j != 0 {
4439 let indexes = extract_scalar_subscript_from_positions(&positions[i..i + j], "")?;
4440 let (n, e) = plan_index_list(
4441 ecx,
4442 expr,
4443 indexes.as_slice(),
4444 remaining_layers,
4445 elem_type_name,
4446 )?;
4447 remaining_layers = n;
4448 expr = e;
4449 i += j;
4450 }
4451
4452 let j = positions[i..]
4454 .iter()
4455 .position(|p| !p.explicit_slice)
4456 .unwrap_or(positions.len() - i);
4457 if j != 0 {
4458 expr = plan_slice_list(
4459 ecx,
4460 expr,
4461 &positions[i..i + j],
4462 remaining_layers,
4463 elem_type_name,
4464 )?;
4465 i += j;
4466 }
4467 }
4468
4469 Ok(expr.into())
4470}
4471
4472fn plan_index_list(
4473 ecx: &ExprContext,
4474 expr: HirScalarExpr,
4475 indexes: &[&Expr<Aug>],
4476 n_layers: usize,
4477 elem_type_name: &str,
4478) -> Result<(usize, HirScalarExpr), PlanError> {
4479 let depth = indexes.len();
4480
4481 if depth > n_layers {
4482 if n_layers == 0 {
4483 sql_bail!("cannot subscript type {}", elem_type_name)
4484 } else {
4485 sql_bail!(
4486 "cannot index into {} layers; list only has {} layer{}",
4487 depth,
4488 n_layers,
4489 if n_layers == 1 { "" } else { "s" }
4490 )
4491 }
4492 }
4493
4494 let mut exprs = Vec::with_capacity(depth + 1);
4495 exprs.push(expr);
4496
4497 for i in indexes {
4498 exprs.push(plan_expr(ecx, i)?.cast_to(
4499 ecx,
4500 CastContext::Explicit,
4501 &SqlScalarType::Int64,
4502 )?);
4503 }
4504
4505 Ok((
4506 n_layers - depth,
4507 HirScalarExpr::call_variadic(ListIndex, exprs),
4508 ))
4509}
4510
4511fn plan_slice_list(
4512 ecx: &ExprContext,
4513 expr: HirScalarExpr,
4514 slices: &[SubscriptPosition<Aug>],
4515 n_layers: usize,
4516 elem_type_name: &str,
4517) -> Result<HirScalarExpr, PlanError> {
4518 if n_layers == 0 {
4519 sql_bail!("cannot subscript type {}", elem_type_name)
4520 }
4521
4522 let mut exprs = Vec::with_capacity(slices.len() + 1);
4524 exprs.push(expr);
4525 let extract_position_or_default = |position, default| -> Result<HirScalarExpr, PlanError> {
4527 Ok(match position {
4528 Some(p) => {
4529 plan_expr(ecx, p)?.cast_to(ecx, CastContext::Explicit, &SqlScalarType::Int64)?
4530 }
4531 None => HirScalarExpr::literal(Datum::Int64(default), SqlScalarType::Int64),
4532 })
4533 };
4534 for p in slices {
4535 let start = extract_position_or_default(p.start.as_ref(), 1)?;
4536 let end = extract_position_or_default(p.end.as_ref(), i64::MAX - 1)?;
4537 exprs.push(start);
4538 exprs.push(end);
4539 }
4540
4541 Ok(HirScalarExpr::call_variadic(ListSliceLinear, exprs))
4542}
4543
4544fn plan_like(
4545 ecx: &ExprContext,
4546 expr: &Expr<Aug>,
4547 pattern: &Expr<Aug>,
4548 escape: Option<&Expr<Aug>>,
4549 case_insensitive: bool,
4550 not: bool,
4551) -> Result<HirScalarExpr, PlanError> {
4552 use CastContext::Implicit;
4553 let ecx = ecx.with_name("LIKE argument");
4554 let expr = plan_expr(&ecx, expr)?;
4555 let haystack = match ecx.scalar_type(&expr) {
4556 CoercibleScalarType::Coerced(ref ty @ SqlScalarType::Char { length }) => expr
4557 .type_as(&ecx, ty)?
4558 .call_unary(UnaryFunc::PadChar(expr_func::PadChar { length })),
4559 _ => expr.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4560 };
4561 let mut pattern = plan_expr(&ecx, pattern)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?;
4562 if let Some(escape) = escape {
4563 pattern = pattern.call_binary(
4564 plan_expr(&ecx, escape)?.cast_to(&ecx, Implicit, &SqlScalarType::String)?,
4565 expr_func::LikeEscape,
4566 );
4567 }
4568 let func: BinaryFunc = if case_insensitive {
4569 expr_func::IsLikeMatchCaseInsensitive.into()
4570 } else {
4571 expr_func::IsLikeMatchCaseSensitive.into()
4572 };
4573 let like = haystack.call_binary(pattern, func);
4574 if not {
4575 Ok(like.call_unary(UnaryFunc::Not(expr_func::Not)))
4576 } else {
4577 Ok(like)
4578 }
4579}
4580
4581fn plan_subscript_jsonb(
4582 ecx: &ExprContext,
4583 expr: HirScalarExpr,
4584 positions: &[SubscriptPosition<Aug>],
4585) -> Result<CoercibleScalarExpr, PlanError> {
4586 use CastContext::Implicit;
4587 use SqlScalarType::{Int64, String};
4588
4589 let subscripts = extract_scalar_subscript_from_positions(positions, "jsonb")?;
4592
4593 let mut exprs = Vec::with_capacity(subscripts.len());
4594 for s in subscripts {
4595 let subscript = plan_expr(ecx, s)?;
4596 let subscript = if let Ok(subscript) = subscript.clone().cast_to(ecx, Implicit, &String) {
4597 subscript
4598 } else if let Ok(subscript) = subscript.cast_to(ecx, Implicit, &Int64) {
4599 typeconv::to_string(ecx, subscript)?
4603 } else {
4604 sql_bail!("jsonb subscript type must be coercible to integer or text");
4605 };
4606 exprs.push(subscript);
4607 }
4608
4609 let expr = expr.call_binary(
4612 HirScalarExpr::call_variadic(
4613 ArrayCreate {
4614 elem_type: SqlScalarType::String,
4615 },
4616 exprs,
4617 ),
4618 expr_func::JsonbGetPath,
4619 );
4620 Ok(expr.into())
4621}
4622
4623fn plan_exists(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4624 if !ecx.allow_subqueries {
4625 sql_bail!("{} does not allow subqueries", ecx.name)
4626 }
4627 let mut qcx = ecx.derived_query_context();
4628 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4629 Ok(expr.exists().into())
4630}
4631
4632fn plan_subquery(ecx: &ExprContext, query: &Query<Aug>) -> Result<CoercibleScalarExpr, PlanError> {
4633 if !ecx.allow_subqueries {
4634 sql_bail!("{} does not allow subqueries", ecx.name)
4635 }
4636 let mut qcx = ecx.derived_query_context();
4637 let (expr, _scope) = plan_nested_query(&mut qcx, query)?;
4638 let column_types = qcx.relation_type(&expr).column_types;
4639 if column_types.len() != 1 {
4640 sql_bail!(
4641 "Expected subselect to return 1 column, got {} columns",
4642 column_types.len()
4643 );
4644 }
4645 Ok(expr.select().into())
4646}
4647
4648fn plan_list_subquery(
4649 ecx: &ExprContext,
4650 query: &Query<Aug>,
4651) -> Result<CoercibleScalarExpr, PlanError> {
4652 plan_vector_like_subquery(
4653 ecx,
4654 query,
4655 |_| false,
4656 |elem_type| ListCreate { elem_type }.into(),
4657 |order_by| AggregateFunc::ListConcat { order_by },
4658 expr_func::ListListConcat.into(),
4659 |elem_type| {
4660 HirScalarExpr::literal(
4661 Datum::empty_list(),
4662 SqlScalarType::List {
4663 element_type: Box::new(elem_type),
4664 custom_id: None,
4665 },
4666 )
4667 },
4668 "list",
4669 )
4670}
4671
4672fn plan_array_subquery(
4673 ecx: &ExprContext,
4674 query: &Query<Aug>,
4675) -> Result<CoercibleScalarExpr, PlanError> {
4676 plan_vector_like_subquery(
4677 ecx,
4678 query,
4679 |elem_type| {
4680 matches!(
4681 elem_type,
4682 SqlScalarType::Char { .. }
4683 | SqlScalarType::Array { .. }
4684 | SqlScalarType::List { .. }
4685 | SqlScalarType::Map { .. }
4686 )
4687 },
4688 |elem_type| ArrayCreate { elem_type }.into(),
4689 |order_by| AggregateFunc::ArrayConcat { order_by },
4690 expr_func::ArrayArrayConcat.into(),
4691 |elem_type| {
4692 HirScalarExpr::literal(
4693 Datum::empty_array(),
4694 SqlScalarType::Array(Box::new(elem_type)),
4695 )
4696 },
4697 "[]",
4698 )
4699}
4700
4701fn plan_vector_like_subquery<F1, F2, F3, F4>(
4703 ecx: &ExprContext,
4704 query: &Query<Aug>,
4705 is_unsupported_type: F1,
4706 vector_create: F2,
4707 aggregate_concat: F3,
4708 binary_concat: BinaryFunc,
4709 empty_literal: F4,
4710 vector_type_string: &str,
4711) -> Result<CoercibleScalarExpr, PlanError>
4712where
4713 F1: Fn(&SqlScalarType) -> bool,
4714 F2: Fn(SqlScalarType) -> VariadicFunc,
4715 F3: Fn(Vec<ColumnOrder>) -> AggregateFunc,
4716 F4: Fn(SqlScalarType) -> HirScalarExpr,
4717{
4718 if !ecx.allow_subqueries {
4719 sql_bail!("{} does not allow subqueries", ecx.name)
4720 }
4721
4722 let mut qcx = ecx.derived_query_context();
4723 let mut planned_query = plan_query(&mut qcx, query)?;
4724 if planned_query.limit.is_some()
4725 || !planned_query
4726 .offset
4727 .clone()
4728 .try_into_literal_int64()
4729 .is_ok_and(|offset| offset == 0)
4730 {
4731 planned_query.expr = HirRelationExpr::top_k(
4732 planned_query.expr,
4733 vec![],
4734 planned_query.order_by.clone(),
4735 planned_query.limit,
4736 planned_query.offset,
4737 planned_query.group_size_hints.limit_input_group_size,
4738 );
4739 }
4740
4741 if planned_query.project.len() != 1 {
4742 sql_bail!(
4743 "Expected subselect to return 1 column, got {} columns",
4744 planned_query.project.len()
4745 );
4746 }
4747
4748 let project_column = *planned_query.project.get(0).unwrap();
4749 let elem_type = qcx
4750 .relation_type(&planned_query.expr)
4751 .column_types
4752 .get(project_column)
4753 .cloned()
4754 .unwrap()
4755 .scalar_type();
4756
4757 if is_unsupported_type(&elem_type) {
4758 bail_unsupported!(format!(
4759 "cannot build array from subquery because return type {}{}",
4760 ecx.humanize_sql_scalar_type(&elem_type, false),
4761 vector_type_string
4762 ));
4763 }
4764
4765 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4768 vector_create(elem_type.clone()),
4769 vec![HirScalarExpr::column(project_column)],
4770 ))
4771 .chain(
4772 planned_query
4773 .order_by
4774 .iter()
4775 .map(|co| HirScalarExpr::column(co.column)),
4776 )
4777 .collect();
4778
4779 let aggregation_projection = vec![0];
4783 let aggregation_order_by = planned_query
4784 .order_by
4785 .into_iter()
4786 .enumerate()
4787 .map(|(i, order)| ColumnOrder { column: i, ..order })
4788 .collect();
4789
4790 let reduced_expr = planned_query
4791 .expr
4792 .reduce(
4793 vec![],
4794 vec![AggregateExpr {
4795 func: aggregate_concat(aggregation_order_by),
4796 expr: Box::new(HirScalarExpr::call_variadic(
4797 RecordCreate {
4798 field_names: iter::repeat(ColumnName::from(""))
4799 .take(aggregation_exprs.len())
4800 .collect(),
4801 },
4802 aggregation_exprs,
4803 )),
4804 distinct: false,
4805 }],
4806 None,
4807 )
4808 .project(aggregation_projection);
4809
4810 Ok(reduced_expr
4812 .select()
4813 .call_binary(empty_literal(elem_type), binary_concat)
4814 .into())
4815}
4816
4817fn plan_map_subquery(
4818 ecx: &ExprContext,
4819 query: &Query<Aug>,
4820) -> Result<CoercibleScalarExpr, PlanError> {
4821 if !ecx.allow_subqueries {
4822 sql_bail!("{} does not allow subqueries", ecx.name)
4823 }
4824
4825 let mut qcx = ecx.derived_query_context();
4826 let mut query = plan_query(&mut qcx, query)?;
4827 if query.limit.is_some()
4828 || !query
4829 .offset
4830 .clone()
4831 .try_into_literal_int64()
4832 .is_ok_and(|offset| offset == 0)
4833 {
4834 query.expr = HirRelationExpr::top_k(
4835 query.expr,
4836 vec![],
4837 query.order_by.clone(),
4838 query.limit,
4839 query.offset,
4840 query.group_size_hints.limit_input_group_size,
4841 );
4842 }
4843 if query.project.len() != 2 {
4844 sql_bail!(
4845 "expected map subquery to return 2 columns, got {} columns",
4846 query.project.len()
4847 );
4848 }
4849
4850 let query_types = qcx.relation_type(&query.expr).column_types;
4851 let key_column = query.project[0];
4852 let key_type = query_types[key_column].clone().scalar_type();
4853 let value_column = query.project[1];
4854 let value_type = query_types[value_column].clone().scalar_type();
4855
4856 if key_type != SqlScalarType::String {
4857 sql_bail!("cannot build map from subquery because first column is not of type text");
4858 }
4859
4860 let aggregation_exprs: Vec<_> = iter::once(HirScalarExpr::call_variadic(
4861 RecordCreate {
4862 field_names: vec![ColumnName::from("key"), ColumnName::from("value")],
4863 },
4864 vec![
4865 HirScalarExpr::column(key_column),
4866 HirScalarExpr::column(value_column),
4867 ],
4868 ))
4869 .chain(
4870 query
4871 .order_by
4872 .iter()
4873 .map(|co| HirScalarExpr::column(co.column)),
4874 )
4875 .collect();
4876
4877 let expr = query
4878 .expr
4879 .reduce(
4880 vec![],
4881 vec![AggregateExpr {
4882 func: AggregateFunc::MapAgg {
4883 order_by: query
4884 .order_by
4885 .into_iter()
4886 .enumerate()
4887 .map(|(i, order)| ColumnOrder { column: i, ..order })
4888 .collect(),
4889 value_type: value_type.clone(),
4890 },
4891 expr: Box::new(HirScalarExpr::call_variadic(
4892 RecordCreate {
4893 field_names: iter::repeat(ColumnName::from(""))
4894 .take(aggregation_exprs.len())
4895 .collect(),
4896 },
4897 aggregation_exprs,
4898 )),
4899 distinct: false,
4900 }],
4901 None,
4902 )
4903 .project(vec![0]);
4904
4905 let expr = HirScalarExpr::call_variadic(
4907 Coalesce,
4908 vec![
4909 expr.select(),
4910 HirScalarExpr::literal(
4911 Datum::empty_map(),
4912 SqlScalarType::Map {
4913 value_type: Box::new(value_type),
4914 custom_id: None,
4915 },
4916 ),
4917 ],
4918 );
4919
4920 Ok(expr.into())
4921}
4922
4923fn plan_collate(
4924 ecx: &ExprContext,
4925 expr: &Expr<Aug>,
4926 collation: &UnresolvedItemName,
4927) -> Result<CoercibleScalarExpr, PlanError> {
4928 if collation.0.len() == 2
4929 && collation.0[0] == ident!(mz_repr::namespaces::PG_CATALOG_SCHEMA)
4930 && collation.0[1] == ident!("default")
4931 {
4932 plan_expr(ecx, expr)
4933 } else {
4934 bail_unsupported!("COLLATE");
4935 }
4936}
4937
4938fn plan_exprs<E>(ecx: &ExprContext, exprs: &[E]) -> Result<Vec<CoercibleScalarExpr>, PlanError>
4945where
4946 E: std::borrow::Borrow<Expr<Aug>>,
4947{
4948 let mut out = vec![];
4949 for expr in exprs {
4950 out.push(plan_expr(ecx, expr.borrow())?);
4951 }
4952 Ok(out)
4953}
4954
4955fn plan_array(
4957 ecx: &ExprContext,
4958 exprs: &[Expr<Aug>],
4959 type_hint: Option<&SqlScalarType>,
4960) -> Result<CoercibleScalarExpr, PlanError> {
4961 let mut out = vec![];
4963 for expr in exprs {
4964 out.push(match expr {
4965 Expr::Array(exprs) => plan_array(ecx, exprs, type_hint.clone())?,
4968 _ => plan_expr(ecx, expr)?,
4969 });
4970 }
4971
4972 let type_hint = match type_hint {
4974 Some(SqlScalarType::Array(elem_type)) => {
4979 let multidimensional = out.iter().any(|e| {
4980 matches!(
4981 ecx.scalar_type(e),
4982 CoercibleScalarType::Coerced(SqlScalarType::Array(_))
4983 )
4984 });
4985 if multidimensional {
4986 type_hint
4987 } else {
4988 Some(&**elem_type)
4989 }
4990 }
4991 Some(_) => None,
4995 None => None,
4997 };
4998
4999 let (elem_type, exprs) = if exprs.is_empty() {
5001 if let Some(elem_type) = type_hint {
5002 (elem_type.clone(), vec![])
5003 } else {
5004 sql_bail!("cannot determine type of empty array");
5005 }
5006 } else {
5007 let out = coerce_homogeneous_exprs(&ecx.with_name("ARRAY"), out, type_hint)?;
5008 (ecx.scalar_type(&out[0]), out)
5009 };
5010
5011 if matches!(
5017 elem_type,
5018 SqlScalarType::Char { .. } | SqlScalarType::List { .. } | SqlScalarType::Map { .. }
5019 ) {
5020 bail_unsupported!(format!(
5021 "{}[]",
5022 ecx.humanize_sql_scalar_type(&elem_type, false)
5023 ));
5024 }
5025
5026 Ok(HirScalarExpr::call_variadic(ArrayCreate { elem_type }, exprs).into())
5027}
5028
5029fn plan_list(
5030 ecx: &ExprContext,
5031 exprs: &[Expr<Aug>],
5032 type_hint: Option<&SqlScalarType>,
5033) -> Result<CoercibleScalarExpr, PlanError> {
5034 let (elem_type, exprs) = if exprs.is_empty() {
5035 if let Some(SqlScalarType::List { element_type, .. }) = type_hint {
5036 (element_type.without_modifiers(), vec![])
5037 } else {
5038 sql_bail!("cannot determine type of empty list");
5039 }
5040 } else {
5041 let type_hint = match type_hint {
5042 Some(SqlScalarType::List { element_type, .. }) => Some(&**element_type),
5043 _ => None,
5044 };
5045
5046 let mut out = vec![];
5047 for expr in exprs {
5048 out.push(match expr {
5049 Expr::List(exprs) => plan_list(ecx, exprs, type_hint)?,
5052 _ => plan_expr(ecx, expr)?,
5053 });
5054 }
5055 let out = coerce_homogeneous_exprs(&ecx.with_name("LIST"), out, type_hint)?;
5056 (ecx.scalar_type(&out[0]).without_modifiers(), out)
5057 };
5058
5059 if matches!(elem_type, SqlScalarType::Char { .. }) {
5060 bail_unsupported!("char list");
5061 }
5062
5063 Ok(HirScalarExpr::call_variadic(ListCreate { elem_type }, exprs).into())
5064}
5065
5066fn plan_map(
5067 ecx: &ExprContext,
5068 entries: &[MapEntry<Aug>],
5069 type_hint: Option<&SqlScalarType>,
5070) -> Result<CoercibleScalarExpr, PlanError> {
5071 let (value_type, exprs) = if entries.is_empty() {
5072 if let Some(SqlScalarType::Map { value_type, .. }) = type_hint {
5073 (value_type.without_modifiers(), vec![])
5074 } else {
5075 sql_bail!("cannot determine type of empty map");
5076 }
5077 } else {
5078 let type_hint = match type_hint {
5079 Some(SqlScalarType::Map { value_type, .. }) => Some(&**value_type),
5080 _ => None,
5081 };
5082
5083 let mut keys = vec![];
5084 let mut values = vec![];
5085 for MapEntry { key, value } in entries {
5086 let key = plan_expr(ecx, key)?.type_as(ecx, &SqlScalarType::String)?;
5087 let value = match value {
5088 Expr::Map(entries) => plan_map(ecx, entries, type_hint)?,
5091 _ => plan_expr(ecx, value)?,
5092 };
5093 keys.push(key);
5094 values.push(value);
5095 }
5096 let values = coerce_homogeneous_exprs(&ecx.with_name("MAP"), values, type_hint)?;
5097 let value_type = ecx.scalar_type(&values[0]).without_modifiers();
5098 let out = itertools::interleave(keys, values).collect();
5099 (value_type, out)
5100 };
5101
5102 if matches!(value_type, SqlScalarType::Char { .. }) {
5103 bail_unsupported!("char map");
5104 }
5105
5106 let expr = HirScalarExpr::call_variadic(MapBuild { value_type }, exprs);
5107 Ok(expr.into())
5108}
5109
5110pub fn coerce_homogeneous_exprs(
5127 ecx: &ExprContext,
5128 exprs: Vec<CoercibleScalarExpr>,
5129 force_type: Option<&SqlScalarType>,
5130) -> Result<Vec<HirScalarExpr>, PlanError> {
5131 assert!(!exprs.is_empty());
5132
5133 let target_holder;
5134 let target = match force_type {
5135 Some(t) => t,
5136 None => {
5137 let types: Vec<_> = exprs.iter().map(|e| ecx.scalar_type(e)).collect();
5138 target_holder = typeconv::guess_best_common_type(ecx, &types)?;
5139 &target_holder
5140 }
5141 };
5142
5143 let mut out = Vec::new();
5145 for expr in exprs {
5146 let arg = typeconv::plan_coerce(ecx, expr, target)?;
5147 let ccx = match force_type {
5148 None => CastContext::Implicit,
5149 Some(_) => CastContext::Explicit,
5150 };
5151 match typeconv::plan_cast(ecx, ccx, arg.clone(), target) {
5152 Ok(expr) => out.push(expr),
5153 Err(_) => sql_bail!(
5154 "{} could not convert type {} to {}",
5155 ecx.name,
5156 ecx.humanize_sql_scalar_type(&ecx.scalar_type(&arg), false),
5157 ecx.humanize_sql_scalar_type(target, false),
5158 ),
5159 }
5160 }
5161 Ok(out)
5162}
5163
5164pub(crate) fn resolve_desc_and_nulls_last<T: AstInfo>(
5167 obe: &OrderByExpr<T>,
5168 column: usize,
5169) -> ColumnOrder {
5170 let desc = !obe.asc.unwrap_or(true);
5171 ColumnOrder {
5172 column,
5173 desc,
5174 nulls_last: obe.nulls_last.unwrap_or(!desc),
5177 }
5178}
5179
5180fn plan_function_order_by(
5188 ecx: &ExprContext,
5189 order_by: &[OrderByExpr<Aug>],
5190) -> Result<(Vec<HirScalarExpr>, Vec<ColumnOrder>), PlanError> {
5191 let mut order_by_exprs = vec![];
5192 let mut col_orders = vec![];
5193 {
5194 for (i, obe) in order_by.iter().enumerate() {
5195 let expr = plan_expr(ecx, &obe.expr)?.type_as_any(ecx)?;
5199 order_by_exprs.push(expr);
5200 col_orders.push(resolve_desc_and_nulls_last(obe, i));
5201 }
5202 }
5203 Ok((order_by_exprs, col_orders))
5204}
5205
5206fn humanize_or_debug(scx: &StatementContext, name: &ResolvedItemName) -> String {
5211 scx.humanize_resolved_name(name)
5212 .map(|n| n.to_string())
5213 .unwrap_or_else(|_| format!("<error when trying to humanize `{name:?}`>"))
5214}
5215
5216fn plan_aggregate_common(
5218 ecx: &ExprContext,
5219 Function::<Aug> {
5220 name,
5221 args,
5222 filter,
5223 over: _,
5224 distinct,
5225 }: &Function<Aug>,
5226) -> Result<AggregateExpr, PlanError> {
5227 let impls = match resolve_func(ecx, name, args)? {
5242 Func::Aggregate(impls) => impls,
5243 _ => bail_internal!("plan_aggregate_common called on non-aggregate function"),
5244 };
5245
5246 let (args, order_by) = match &args {
5255 FunctionArgs::Star => (vec![], vec![]),
5256 FunctionArgs::Args { args, order_by } => {
5257 if args.is_empty() {
5258 sql_bail!(
5259 "{}(*) must be used to call a parameterless aggregate function",
5260 humanize_or_debug(ecx.qcx.scx, name)
5261 );
5262 }
5263 let args = plan_exprs(ecx, args)?;
5264 (args, order_by.clone())
5265 }
5266 };
5267
5268 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &order_by)?;
5269
5270 let (mut expr, func) = func::select_impl(ecx, FuncSpec::Func(name), impls, args, col_orders)?;
5271 if let Some(filter) = &filter {
5272 let cond =
5282 plan_expr(&ecx.with_name("FILTER"), filter)?.type_as(ecx, &SqlScalarType::Bool)?;
5283 let expr_typ = ecx.scalar_type(&expr);
5284 expr = HirScalarExpr::if_then_else(
5285 cond,
5286 expr,
5287 HirScalarExpr::literal(func.identity_datum(), expr_typ),
5288 );
5289 }
5290
5291 let mut seen_outer = false;
5292 let mut seen_inner = false;
5293 #[allow(deprecated)]
5294 expr.visit_columns(0, &mut |depth, col| {
5295 if depth == 0 && col.level == 0 {
5296 seen_inner = true;
5297 } else if col.level > depth {
5298 seen_outer = true;
5299 }
5300 });
5301 if seen_outer && !seen_inner {
5302 bail_unsupported!(
5303 3720,
5304 "aggregate functions that refer exclusively to outer columns"
5305 );
5306 }
5307
5308 if func.is_order_sensitive() {
5311 let field_names = iter::repeat(ColumnName::from(""))
5312 .take(1 + order_by_exprs.len())
5313 .collect();
5314 let mut exprs = vec![expr];
5315 exprs.extend(order_by_exprs);
5316 expr = HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs);
5317 }
5318
5319 Ok(AggregateExpr {
5320 func,
5321 expr: Box::new(expr),
5322 distinct: *distinct,
5323 })
5324}
5325
5326fn plan_identifier(ecx: &ExprContext, names: &[Ident]) -> Result<HirScalarExpr, PlanError> {
5327 let mut names = names.to_vec();
5328 let Some(last) = names.pop() else {
5331 bail_internal!("empty identifier");
5332 };
5333 let col_name = normalize::column_name(last);
5334
5335 if !names.is_empty() {
5337 let table_name = normalize::unresolved_item_name(UnresolvedItemName(names))?;
5338 let (i, i_name) = ecx.scope.resolve_table_column(
5339 &ecx.qcx.outer_scopes,
5340 &table_name,
5341 &col_name,
5342 &mut ecx.qcx.name_manager.borrow_mut(),
5343 )?;
5344 return Ok(HirScalarExpr::named_column(i, i_name));
5345 }
5346
5347 let similar_names = match ecx.scope.resolve_column(
5350 &ecx.qcx.outer_scopes,
5351 &col_name,
5352 &mut ecx.qcx.name_manager.borrow_mut(),
5353 ) {
5354 Ok((i, i_name)) => {
5355 return Ok(HirScalarExpr::named_column(i, i_name));
5356 }
5357 Err(PlanError::UnknownColumn { similar, .. }) => similar,
5358 Err(e) => return Err(e),
5359 };
5360
5361 let items = ecx.scope.items_from_table(
5364 &ecx.qcx.outer_scopes,
5365 &PartialItemName {
5366 database: None,
5367 schema: None,
5368 item: col_name.as_str().to_owned(),
5369 },
5370 )?;
5371 match items.as_slice() {
5372 [] => Err(PlanError::UnknownColumn {
5374 table: None,
5375 column: col_name,
5376 similar: similar_names,
5377 }),
5378 [(column, item)] if item.from_single_column_function => Ok(HirScalarExpr::named_column(
5383 *column,
5384 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5385 )),
5386 _ => {
5389 let mut has_exists_column = None;
5390 let (exprs, field_names): (Vec<_>, Vec<_>) = items
5391 .into_iter()
5392 .filter_map(|(column, item)| {
5393 if item.is_exists_column_for_a_table_function_that_was_in_the_target_list {
5394 has_exists_column = Some(column);
5395 None
5396 } else {
5397 let expr = HirScalarExpr::named_column(
5398 column,
5399 ecx.qcx.name_manager.borrow_mut().intern_scope_item(item),
5400 );
5401 let name = item.column_name.clone();
5402 Some((expr, name))
5403 }
5404 })
5405 .unzip();
5406 let expr = if exprs.len() == 1 && has_exists_column.is_some() {
5408 exprs.into_element()
5409 } else {
5410 HirScalarExpr::call_variadic(RecordCreate { field_names }, exprs)
5411 };
5412 if let Some(has_exists_column) = has_exists_column {
5413 Ok(HirScalarExpr::if_then_else(
5414 HirScalarExpr::unnamed_column(has_exists_column)
5415 .call_unary(UnaryFunc::IsNull(mz_expr::func::IsNull)),
5416 HirScalarExpr::literal_null(ecx.scalar_type(&expr)),
5417 expr,
5418 ))
5419 } else {
5420 Ok(expr)
5421 }
5422 }
5423 }
5424}
5425
5426fn plan_op(
5427 ecx: &ExprContext,
5428 op: &str,
5429 expr1: &Expr<Aug>,
5430 expr2: Option<&Expr<Aug>>,
5431) -> Result<HirScalarExpr, PlanError> {
5432 let impls = func::resolve_op(op)?;
5433 let args = match expr2 {
5434 None => plan_exprs(ecx, &[expr1])?,
5435 Some(expr2) => plan_exprs(ecx, &[expr1, expr2])?,
5436 };
5437 func::select_impl(ecx, FuncSpec::Op(op), impls, args, vec![])
5438}
5439
5440fn plan_function<'a>(
5441 ecx: &ExprContext,
5442 f @ Function {
5443 name,
5444 args,
5445 filter,
5446 over,
5447 distinct,
5448 }: &'a Function<Aug>,
5449) -> Result<HirScalarExpr, PlanError> {
5450 let impls = match resolve_func(ecx, name, args)? {
5451 Func::Table(_) => {
5452 sql_bail!(
5453 "table functions are not allowed in {} (function {})",
5454 ecx.name,
5455 name
5456 );
5457 }
5458 Func::Scalar(impls) => {
5459 if over.is_some() {
5460 sql_bail!(
5461 "OVER clause not allowed on {name}. The OVER clause can only be used with window functions (including aggregations)."
5462 );
5463 }
5464 impls
5465 }
5466 Func::ScalarWindow(impls) => {
5467 let (
5468 ignore_nulls,
5469 order_by_exprs,
5470 col_orders,
5471 _window_frame,
5472 partition_by,
5473 scalar_args,
5474 ) = plan_window_function_non_aggr(ecx, f)?;
5475
5476 if !scalar_args.is_empty() {
5480 if let ResolvedItemName::Item {
5481 full_name: FullItemName { item, .. },
5482 ..
5483 } = name
5484 {
5485 sql_bail!(
5486 "function {} has 0 parameters, but was called with {}",
5487 item,
5488 scalar_args.len()
5489 );
5490 }
5491 }
5492
5493 let func = func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])?;
5498
5499 if ignore_nulls {
5500 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5503 }
5504
5505 return Ok(HirScalarExpr::windowing(WindowExpr {
5506 func: WindowExprType::Scalar(ScalarWindowExpr {
5507 func,
5508 order_by: col_orders,
5509 }),
5510 partition_by,
5511 order_by: order_by_exprs,
5512 }));
5513 }
5514 Func::ValueWindow(impls) => {
5515 let window_plan = plan_window_function_non_aggr(ecx, f)?;
5516 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by, win_args) =
5517 window_plan;
5518
5519 let (args_encoded, func) =
5520 func::select_impl(ecx, FuncSpec::Func(name), impls, win_args, vec![])?;
5521
5522 if ignore_nulls {
5523 match func {
5524 ValueWindowFunc::Lag | ValueWindowFunc::Lead => {}
5525 _ => bail_unsupported!(IGNORE_NULLS_ERROR_MSG),
5526 }
5527 }
5528
5529 return Ok(HirScalarExpr::windowing(WindowExpr {
5530 func: WindowExprType::Value(ValueWindowExpr {
5531 func,
5532 args: Box::new(args_encoded),
5533 order_by: col_orders,
5534 window_frame,
5535 ignore_nulls, }),
5537 partition_by,
5538 order_by: order_by_exprs,
5539 }));
5540 }
5541 Func::Aggregate(_) => {
5542 if f.over.is_none() {
5543 if ecx.allow_aggregates {
5545 sql_bail!(
5548 "Internal error: encountered unplanned non-windowed aggregate function: {:?}",
5549 name,
5550 );
5551 } else {
5552 sql_bail!(
5555 "aggregate functions are not allowed in {} (function {})",
5556 ecx.name,
5557 name
5558 );
5559 }
5560 } else {
5561 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition_by) =
5562 plan_window_function_common(ecx, &f.name, &f.over)?;
5563
5564 match (&window_frame.start_bound, &window_frame.end_bound) {
5566 (
5567 mz_expr::WindowFrameBound::UnboundedPreceding,
5568 mz_expr::WindowFrameBound::OffsetPreceding(..),
5569 )
5570 | (
5571 mz_expr::WindowFrameBound::UnboundedPreceding,
5572 mz_expr::WindowFrameBound::OffsetFollowing(..),
5573 )
5574 | (
5575 mz_expr::WindowFrameBound::OffsetPreceding(..),
5576 mz_expr::WindowFrameBound::UnboundedFollowing,
5577 )
5578 | (
5579 mz_expr::WindowFrameBound::OffsetFollowing(..),
5580 mz_expr::WindowFrameBound::UnboundedFollowing,
5581 ) => bail_unsupported!("mixed unbounded - offset frames"),
5582 (_, _) => {} }
5584
5585 if ignore_nulls {
5586 bail_unsupported!(IGNORE_NULLS_ERROR_MSG);
5590 }
5591
5592 let aggregate_expr = plan_aggregate_common(ecx, f)?;
5593
5594 if aggregate_expr.distinct {
5595 bail_unsupported!("DISTINCT in window aggregates");
5597 }
5598
5599 return Ok(HirScalarExpr::windowing(WindowExpr {
5600 func: WindowExprType::Aggregate(AggregateWindowExpr {
5601 aggregate_expr,
5602 order_by: col_orders,
5603 window_frame,
5604 }),
5605 partition_by,
5606 order_by: order_by_exprs,
5607 }));
5608 }
5609 }
5610 };
5611
5612 if over.is_some() {
5613 bail_internal!("OVER clause should have been handled by the window function path above");
5614 }
5615
5616 if *distinct {
5617 sql_bail!(
5618 "DISTINCT specified, but {} is not an aggregate function",
5619 humanize_or_debug(ecx.qcx.scx, name)
5620 );
5621 }
5622 if filter.is_some() {
5623 sql_bail!(
5624 "FILTER specified, but {} is not an aggregate function",
5625 humanize_or_debug(ecx.qcx.scx, name)
5626 );
5627 }
5628
5629 let scalar_args = match &args {
5630 FunctionArgs::Star => {
5631 sql_bail!(
5632 "* argument is invalid with non-aggregate function {}",
5633 humanize_or_debug(ecx.qcx.scx, name)
5634 )
5635 }
5636 FunctionArgs::Args { args, order_by } => {
5637 if !order_by.is_empty() {
5638 sql_bail!(
5639 "ORDER BY specified, but {} is not an aggregate function",
5640 humanize_or_debug(ecx.qcx.scx, name)
5641 );
5642 }
5643 plan_exprs(ecx, args)?
5644 }
5645 };
5646
5647 func::select_impl(ecx, FuncSpec::Func(name), impls, scalar_args, vec![])
5648}
5649
5650pub const IGNORE_NULLS_ERROR_MSG: &str =
5651 "IGNORE NULLS and RESPECT NULLS options for functions other than LAG and LEAD";
5652
5653pub fn resolve_func(
5657 ecx: &ExprContext,
5658 name: &ResolvedItemName,
5659 args: &mz_sql_parser::ast::FunctionArgs<Aug>,
5660) -> Result<&'static Func, PlanError> {
5661 if let Ok(i) = ecx.qcx.scx.get_item_by_resolved_name(name) {
5662 if let Ok(f) = i.func() {
5663 return Ok(f);
5664 }
5665 }
5666
5667 let cexprs = match args {
5670 mz_sql_parser::ast::FunctionArgs::Star => vec![],
5671 mz_sql_parser::ast::FunctionArgs::Args { args, order_by } => {
5672 if !order_by.is_empty() {
5673 sql_bail!(
5674 "ORDER BY specified, but {} is not an aggregate function",
5675 name
5676 );
5677 }
5678 plan_exprs(ecx, args)?
5679 }
5680 };
5681
5682 let arg_types: Vec<_> = cexprs
5683 .into_iter()
5684 .map(|ty| match ecx.scalar_type(&ty) {
5685 CoercibleScalarType::Coerced(ty) => ecx.humanize_sql_scalar_type(&ty, false),
5686 CoercibleScalarType::Record(_) => "record".to_string(),
5687 CoercibleScalarType::Uncoerced => "unknown".to_string(),
5688 })
5689 .collect();
5690
5691 Err(PlanError::UnknownFunction {
5692 name: name.to_string(),
5693 arg_types,
5694 })
5695}
5696
5697fn plan_is_expr<'a>(
5698 ecx: &ExprContext,
5699 expr: &'a Expr<Aug>,
5700 construct: &IsExprConstruct<Aug>,
5701 not: bool,
5702) -> Result<HirScalarExpr, PlanError> {
5703 let expr_hir = plan_expr(ecx, expr)?;
5704
5705 let mut result = match construct {
5706 IsExprConstruct::Null => {
5707 expr_hir.type_as_any(ecx)?.call_is_null()
5712 }
5713 IsExprConstruct::Unknown => expr_hir.type_as(ecx, &SqlScalarType::Bool)?.call_is_null(),
5714 IsExprConstruct::True => expr_hir
5715 .type_as(ecx, &SqlScalarType::Bool)?
5716 .call_unary(UnaryFunc::IsTrue(expr_func::IsTrue)),
5717 IsExprConstruct::False => expr_hir
5718 .type_as(ecx, &SqlScalarType::Bool)?
5719 .call_unary(UnaryFunc::IsFalse(expr_func::IsFalse)),
5720 IsExprConstruct::DistinctFrom(expr2) => {
5721 let ne_ast = expr.clone().not_equals(expr2.as_ref().clone());
5732 let ne_hir = plan_expr(ecx, &ne_ast)?.type_as_any(ecx)?;
5733
5734 let expr1_hir = expr_hir.type_as_any(ecx)?;
5735 let expr2_hir = plan_expr(ecx, expr2)?.type_as_any(ecx)?;
5736
5737 let term1 = HirScalarExpr::variadic_or(vec![
5738 ne_hir,
5739 expr1_hir.clone().call_is_null(),
5740 expr2_hir.clone().call_is_null(),
5741 ]);
5742 let term2 = HirScalarExpr::variadic_or(vec![
5743 expr1_hir.call_is_null().not(),
5744 expr2_hir.call_is_null().not(),
5745 ]);
5746 term1.and(term2)
5747 }
5748 };
5749 if not {
5750 result = result.not();
5751 }
5752 Ok(result)
5753}
5754
5755fn plan_case<'a>(
5756 ecx: &ExprContext,
5757 operand: &'a Option<Box<Expr<Aug>>>,
5758 conditions: &'a [Expr<Aug>],
5759 results: &'a [Expr<Aug>],
5760 else_result: &'a Option<Box<Expr<Aug>>>,
5761) -> Result<HirScalarExpr, PlanError> {
5762 let mut cond_exprs = Vec::new();
5763 let mut result_exprs = Vec::new();
5764 for (c, r) in conditions.iter().zip_eq(results) {
5765 let c = match operand {
5766 Some(operand) => operand.clone().equals(c.clone()),
5767 None => c.clone(),
5768 };
5769 let cexpr = plan_expr(ecx, &c)?.type_as(ecx, &SqlScalarType::Bool)?;
5770 cond_exprs.push(cexpr);
5771 result_exprs.push(r);
5772 }
5773 result_exprs.push(match else_result {
5774 Some(else_result) => else_result,
5775 None => &Expr::Value(Value::Null),
5776 });
5777 let mut result_exprs = coerce_homogeneous_exprs(
5778 &ecx.with_name("CASE"),
5779 plan_exprs(ecx, &result_exprs)?,
5780 None,
5781 )?;
5782 let mut expr = result_exprs.pop().unwrap();
5783 assert_eq!(cond_exprs.len(), result_exprs.len());
5784 for (cexpr, rexpr) in cond_exprs
5785 .into_iter()
5786 .rev()
5787 .zip_eq(result_exprs.into_iter().rev())
5788 {
5789 expr = HirScalarExpr::if_then_else(cexpr, rexpr, expr);
5790 }
5791 Ok(expr)
5792}
5793
5794fn plan_literal<'a>(l: &'a Value) -> Result<CoercibleScalarExpr, PlanError> {
5795 let (datum, scalar_type) = match l {
5796 Value::Number(s) => {
5797 let d = strconv::parse_numeric(s.as_str())?;
5798 if !s.contains(&['E', '.'][..]) {
5799 if let Ok(n) = d.0.try_into() {
5801 (Datum::Int32(n), SqlScalarType::Int32)
5802 } else if let Ok(n) = d.0.try_into() {
5803 (Datum::Int64(n), SqlScalarType::Int64)
5804 } else {
5805 (
5806 Datum::Numeric(d),
5807 SqlScalarType::Numeric { max_scale: None },
5808 )
5809 }
5810 } else {
5811 (
5812 Datum::Numeric(d),
5813 SqlScalarType::Numeric { max_scale: None },
5814 )
5815 }
5816 }
5817 Value::HexString(_) => bail_unsupported!("hex string literals"),
5818 Value::Boolean(b) => match b {
5819 false => (Datum::False, SqlScalarType::Bool),
5820 true => (Datum::True, SqlScalarType::Bool),
5821 },
5822 Value::Interval(i) => {
5823 let i = literal::plan_interval(i)?;
5824 (Datum::Interval(i), SqlScalarType::Interval)
5825 }
5826 Value::String(s) => return Ok(CoercibleScalarExpr::LiteralString(s.clone())),
5827 Value::Null => return Ok(CoercibleScalarExpr::LiteralNull),
5828 };
5829 let expr = HirScalarExpr::literal(datum, scalar_type);
5830 Ok(expr.into())
5831}
5832
5833fn plan_window_function_non_aggr<'a>(
5836 ecx: &ExprContext,
5837 Function {
5838 name,
5839 args,
5840 filter,
5841 over,
5842 distinct,
5843 }: &'a Function<Aug>,
5844) -> Result<
5845 (
5846 bool,
5847 Vec<HirScalarExpr>,
5848 Vec<ColumnOrder>,
5849 mz_expr::WindowFrame,
5850 Vec<HirScalarExpr>,
5851 Vec<CoercibleScalarExpr>,
5852 ),
5853 PlanError,
5854> {
5855 let (ignore_nulls, order_by_exprs, col_orders, window_frame, partition) =
5856 plan_window_function_common(ecx, name, over)?;
5857
5858 if *distinct {
5859 sql_bail!(
5860 "DISTINCT specified, but {} is not an aggregate function",
5861 name
5862 );
5863 }
5864
5865 if filter.is_some() {
5866 bail_unsupported!("FILTER in non-aggregate window functions");
5867 }
5868
5869 let scalar_args = match &args {
5870 FunctionArgs::Star => {
5871 sql_bail!("* argument is invalid with non-aggregate function {}", name)
5872 }
5873 FunctionArgs::Args { args, order_by } => {
5874 if !order_by.is_empty() {
5875 sql_bail!(
5876 "ORDER BY specified, but {} is not an aggregate function",
5877 name
5878 );
5879 }
5880 plan_exprs(ecx, args)?
5881 }
5882 };
5883
5884 Ok((
5885 ignore_nulls,
5886 order_by_exprs,
5887 col_orders,
5888 window_frame,
5889 partition,
5890 scalar_args,
5891 ))
5892}
5893
5894fn plan_window_function_common(
5896 ecx: &ExprContext,
5897 name: &<Aug as AstInfo>::ItemName,
5898 over: &Option<WindowSpec<Aug>>,
5899) -> Result<
5900 (
5901 bool,
5902 Vec<HirScalarExpr>,
5903 Vec<ColumnOrder>,
5904 mz_expr::WindowFrame,
5905 Vec<HirScalarExpr>,
5906 ),
5907 PlanError,
5908> {
5909 if !ecx.allow_windows {
5910 sql_bail!(
5911 "window functions are not allowed in {} (function {})",
5912 ecx.name,
5913 name
5914 );
5915 }
5916
5917 let window_spec = match over.as_ref() {
5918 Some(over) => over,
5919 None => sql_bail!("window function {} requires an OVER clause", name),
5920 };
5921 if window_spec.ignore_nulls && window_spec.respect_nulls {
5922 sql_bail!("Both IGNORE NULLS and RESPECT NULLS were given.");
5923 }
5924 let window_frame = match window_spec.window_frame.as_ref() {
5925 Some(frame) => plan_window_frame(frame)?,
5926 None => mz_expr::WindowFrame::default(),
5927 };
5928 let mut partition = Vec::new();
5929 for expr in &window_spec.partition_by {
5930 partition.push(plan_expr(ecx, expr)?.type_as_any(ecx)?);
5931 }
5932
5933 let (order_by_exprs, col_orders) = plan_function_order_by(ecx, &window_spec.order_by)?;
5934
5935 Ok((
5936 window_spec.ignore_nulls,
5937 order_by_exprs,
5938 col_orders,
5939 window_frame,
5940 partition,
5941 ))
5942}
5943
5944fn plan_window_frame(
5945 WindowFrame {
5946 units,
5947 start_bound,
5948 end_bound,
5949 }: &WindowFrame,
5950) -> Result<mz_expr::WindowFrame, PlanError> {
5951 use mz_expr::WindowFrameBound::*;
5952 let units = window_frame_unit_ast_to_expr(units)?;
5953 let start_bound = window_frame_bound_ast_to_expr(start_bound);
5954 let end_bound = end_bound
5955 .as_ref()
5956 .map(window_frame_bound_ast_to_expr)
5957 .unwrap_or(CurrentRow);
5958
5959 match (&start_bound, &end_bound) {
5961 (UnboundedFollowing, _) => {
5963 sql_bail!("frame start cannot be UNBOUNDED FOLLOWING")
5964 }
5965 (_, UnboundedPreceding) => {
5967 sql_bail!("frame end cannot be UNBOUNDED PRECEDING")
5968 }
5969 (CurrentRow, OffsetPreceding(_)) => {
5971 sql_bail!("frame starting from current row cannot have preceding rows")
5972 }
5973 (OffsetFollowing(_), OffsetPreceding(_) | CurrentRow) => {
5974 sql_bail!("frame starting from following row cannot have preceding rows")
5975 }
5976 (OffsetPreceding(o1), OffsetFollowing(o2)) => {
5979 if *o1 > 1000000 || *o2 > 1000000 {
5983 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
5984 }
5985 }
5986 (OffsetPreceding(o1), OffsetPreceding(o2)) => {
5987 if *o1 > 1000000 || *o2 > 1000000 {
5988 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
5989 }
5990 }
5991 (OffsetFollowing(o1), OffsetFollowing(o2)) => {
5992 if *o1 > 1000000 || *o2 > 1000000 {
5993 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
5994 }
5995 }
5996 (OffsetPreceding(o), CurrentRow) => {
5997 if *o > 1000000 {
5998 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
5999 }
6000 }
6001 (CurrentRow, OffsetFollowing(o)) => {
6002 if *o > 1000000 {
6003 sql_bail!("Window frame offsets greater than 1000000 are currently not supported")
6004 }
6005 }
6006 (_, _) => (),
6008 }
6009
6010 if units == mz_expr::WindowFrameUnits::Range
6013 && (start_bound != UnboundedPreceding || end_bound != CurrentRow)
6014 {
6015 bail_unsupported!("RANGE in non-default window frames")
6016 }
6017
6018 let frame = mz_expr::WindowFrame {
6019 units,
6020 start_bound,
6021 end_bound,
6022 };
6023 Ok(frame)
6024}
6025
6026fn window_frame_unit_ast_to_expr(
6027 unit: &WindowFrameUnits,
6028) -> Result<mz_expr::WindowFrameUnits, PlanError> {
6029 match unit {
6030 WindowFrameUnits::Rows => Ok(mz_expr::WindowFrameUnits::Rows),
6031 WindowFrameUnits::Range => Ok(mz_expr::WindowFrameUnits::Range),
6032 WindowFrameUnits::Groups => bail_unsupported!("GROUPS in window frames"),
6033 }
6034}
6035
6036fn window_frame_bound_ast_to_expr(bound: &WindowFrameBound) -> mz_expr::WindowFrameBound {
6037 match bound {
6038 WindowFrameBound::CurrentRow => mz_expr::WindowFrameBound::CurrentRow,
6039 WindowFrameBound::Preceding(None) => mz_expr::WindowFrameBound::UnboundedPreceding,
6040 WindowFrameBound::Preceding(Some(offset)) => {
6041 mz_expr::WindowFrameBound::OffsetPreceding(*offset)
6042 }
6043 WindowFrameBound::Following(None) => mz_expr::WindowFrameBound::UnboundedFollowing,
6044 WindowFrameBound::Following(Some(offset)) => {
6045 mz_expr::WindowFrameBound::OffsetFollowing(*offset)
6046 }
6047 }
6048}
6049
6050pub fn scalar_type_from_sql(
6051 scx: &StatementContext,
6052 data_type: &ResolvedDataType,
6053) -> Result<SqlScalarType, PlanError> {
6054 match data_type {
6055 ResolvedDataType::AnonymousList(elem_type) => {
6056 let elem_type = scalar_type_from_sql(scx, elem_type)?;
6057 if matches!(elem_type, SqlScalarType::Char { .. }) {
6058 bail_unsupported!("char list");
6059 }
6060 Ok(SqlScalarType::List {
6061 element_type: Box::new(elem_type),
6062 custom_id: None,
6063 })
6064 }
6065 ResolvedDataType::AnonymousMap {
6066 key_type,
6067 value_type,
6068 } => {
6069 match scalar_type_from_sql(scx, key_type)? {
6070 SqlScalarType::String => {}
6071 other => sql_bail!(
6072 "map key type must be {}, got {}",
6073 scx.humanize_sql_scalar_type(&SqlScalarType::String, false),
6074 scx.humanize_sql_scalar_type(&other, false)
6075 ),
6076 }
6077 Ok(SqlScalarType::Map {
6078 value_type: Box::new(scalar_type_from_sql(scx, value_type)?),
6079 custom_id: None,
6080 })
6081 }
6082 ResolvedDataType::Named { id, modifiers, .. } => {
6083 scalar_type_from_catalog(scx.catalog, *id, modifiers)
6084 }
6085 ResolvedDataType::Error => bail_internal!("should have been caught in name resolution"),
6086 }
6087}
6088
6089const MAX_TYPE_NESTING_DEPTH: usize = 128;
6093
6094const MAX_TYPE_RESOLUTION_NODES: usize = 100_000;
6100
6101pub fn scalar_type_from_catalog(
6102 catalog: &dyn SessionCatalog,
6103 id: CatalogItemId,
6104 modifiers: &[i64],
6105) -> Result<SqlScalarType, PlanError> {
6106 let (depth_limit, mut budget) = type_resolution_limits(catalog);
6107 scalar_type_from_catalog_inner(catalog, id, modifiers, 0, depth_limit, &mut budget)
6108}
6109
6110fn type_resolution_limits(catalog: &dyn SessionCatalog) -> (usize, usize) {
6124 if catalog
6125 .system_vars()
6126 .unsafe_enable_unbounded_custom_type_resolution()
6127 {
6128 (usize::MAX, usize::MAX)
6129 } else {
6130 (MAX_TYPE_NESTING_DEPTH, MAX_TYPE_RESOLUTION_NODES)
6131 }
6132}
6133
6134pub struct TypeResolutionBudget {
6147 remaining: usize,
6150 depth_limit: usize,
6153}
6154
6155impl TypeResolutionBudget {
6156 pub fn for_root(catalog: &dyn SessionCatalog) -> TypeResolutionBudget {
6162 let (depth_limit, budget) = type_resolution_limits(catalog);
6163 TypeResolutionBudget {
6164 remaining: budget.saturating_sub(1),
6166 depth_limit,
6167 }
6168 }
6169
6170 pub fn resolve_child(
6174 &mut self,
6175 catalog: &dyn SessionCatalog,
6176 id: CatalogItemId,
6177 modifiers: &[i64],
6178 ) -> Result<SqlScalarType, PlanError> {
6179 scalar_type_from_catalog_inner(
6180 catalog,
6181 id,
6182 modifiers,
6183 1,
6184 self.depth_limit,
6185 &mut self.remaining,
6186 )
6187 }
6188}
6189
6190fn scalar_type_from_catalog_inner(
6191 catalog: &dyn SessionCatalog,
6192 id: CatalogItemId,
6193 modifiers: &[i64],
6194 depth: usize,
6195 depth_limit: usize,
6196 budget: &mut usize,
6197) -> Result<SqlScalarType, PlanError> {
6198 if depth > depth_limit {
6199 sql_bail!("custom type nesting depth exceeds limit of {}", depth_limit);
6200 }
6201 *budget = match budget.checked_sub(1) {
6202 Some(remaining) => remaining,
6203 None => sql_bail!("custom type is too complex to resolve"),
6204 };
6205 let entry = catalog.get_item(&id);
6206 let type_details = match entry.type_details() {
6207 Some(type_details) => type_details,
6208 None => {
6209 sql_bail!(
6212 "internal error: {} does not refer to a type",
6213 catalog.resolve_full_name(entry.name()).to_string().quoted()
6214 );
6215 }
6216 };
6217 match &type_details.typ {
6218 CatalogType::Numeric => {
6219 let mut modifiers = modifiers.iter().fuse();
6220 let precision = match modifiers.next() {
6221 Some(p) if *p < 1 || *p > i64::from(NUMERIC_DATUM_MAX_PRECISION) => {
6222 sql_bail!(
6223 "precision for type numeric must be between 1 and {}",
6224 NUMERIC_DATUM_MAX_PRECISION,
6225 );
6226 }
6227 Some(p) => Some(*p),
6228 None => None,
6229 };
6230 let scale = match modifiers.next() {
6231 Some(scale) => {
6232 if let Some(precision) = precision {
6233 if *scale > precision {
6234 sql_bail!(
6235 "scale for type numeric must be between 0 and precision {}",
6236 precision
6237 );
6238 }
6239 }
6240 Some(NumericMaxScale::try_from(*scale)?)
6241 }
6242 None => None,
6243 };
6244 if modifiers.next().is_some() {
6245 sql_bail!("type numeric supports at most two type modifiers");
6246 }
6247 Ok(SqlScalarType::Numeric { max_scale: scale })
6248 }
6249 CatalogType::Char => {
6250 let mut modifiers = modifiers.iter().fuse();
6251 let length = match modifiers.next() {
6252 Some(l) => Some(CharLength::try_from(*l)?),
6253 None => Some(CharLength::ONE),
6254 };
6255 if modifiers.next().is_some() {
6256 sql_bail!("type character supports at most one type modifier");
6257 }
6258 Ok(SqlScalarType::Char { length })
6259 }
6260 CatalogType::VarChar => {
6261 let mut modifiers = modifiers.iter().fuse();
6262 let length = match modifiers.next() {
6263 Some(l) => Some(VarCharMaxLength::try_from(*l)?),
6264 None => None,
6265 };
6266 if modifiers.next().is_some() {
6267 sql_bail!("type character varying supports at most one type modifier");
6268 }
6269 Ok(SqlScalarType::VarChar { max_length: length })
6270 }
6271 CatalogType::Timestamp => {
6272 let mut modifiers = modifiers.iter().fuse();
6273 let precision = match modifiers.next() {
6274 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6275 None => None,
6276 };
6277 if modifiers.next().is_some() {
6278 sql_bail!("type timestamp supports at most one type modifier");
6279 }
6280 Ok(SqlScalarType::Timestamp { precision })
6281 }
6282 CatalogType::TimestampTz => {
6283 let mut modifiers = modifiers.iter().fuse();
6284 let precision = match modifiers.next() {
6285 Some(p) => Some(TimestampPrecision::try_from(*p)?),
6286 None => None,
6287 };
6288 if modifiers.next().is_some() {
6289 sql_bail!("type timestamp with time zone supports at most one type modifier");
6290 }
6291 Ok(SqlScalarType::TimestampTz { precision })
6292 }
6293 t => {
6294 if !modifiers.is_empty() {
6295 sql_bail!(
6296 "{} does not support type modifiers",
6297 catalog.resolve_full_name(entry.name()).to_string()
6298 );
6299 }
6300 match t {
6301 CatalogType::Array {
6302 element_reference: element_id,
6303 } => Ok(SqlScalarType::Array(Box::new(
6304 scalar_type_from_catalog_inner(
6305 catalog,
6306 *element_id,
6307 modifiers,
6308 depth + 1,
6309 depth_limit,
6310 budget,
6311 )?,
6312 ))),
6313 CatalogType::List {
6314 element_reference: element_id,
6315 element_modifiers,
6316 } => Ok(SqlScalarType::List {
6317 element_type: Box::new(scalar_type_from_catalog_inner(
6318 catalog,
6319 *element_id,
6320 element_modifiers,
6321 depth + 1,
6322 depth_limit,
6323 budget,
6324 )?),
6325 custom_id: Some(id),
6326 }),
6327 CatalogType::Map {
6328 key_reference: _,
6329 key_modifiers: _,
6330 value_reference: value_id,
6331 value_modifiers,
6332 } => Ok(SqlScalarType::Map {
6333 value_type: Box::new(scalar_type_from_catalog_inner(
6334 catalog,
6335 *value_id,
6336 value_modifiers,
6337 depth + 1,
6338 depth_limit,
6339 budget,
6340 )?),
6341 custom_id: Some(id),
6342 }),
6343 CatalogType::Range {
6344 element_reference: element_id,
6345 } => Ok(SqlScalarType::Range {
6346 element_type: Box::new(scalar_type_from_catalog_inner(
6347 catalog,
6348 *element_id,
6349 &[],
6350 depth + 1,
6351 depth_limit,
6352 budget,
6353 )?),
6354 }),
6355 CatalogType::Record { fields } => {
6356 let scalars: Box<[(ColumnName, SqlColumnType)]> = fields
6357 .iter()
6358 .map(|f| {
6359 let scalar_type = scalar_type_from_catalog_inner(
6360 catalog,
6361 f.type_reference,
6362 &f.type_modifiers,
6363 depth + 1,
6364 depth_limit,
6365 budget,
6366 )?;
6367 Ok((
6368 f.name.clone(),
6369 SqlColumnType {
6370 scalar_type,
6371 nullable: true,
6372 },
6373 ))
6374 })
6375 .collect::<Result<Box<_>, PlanError>>()?;
6376 Ok(SqlScalarType::Record {
6377 fields: scalars,
6378 custom_id: Some(id),
6379 })
6380 }
6381 CatalogType::AclItem => Ok(SqlScalarType::AclItem),
6382 CatalogType::Bool => Ok(SqlScalarType::Bool),
6383 CatalogType::Bytes => Ok(SqlScalarType::Bytes),
6384 CatalogType::Date => Ok(SqlScalarType::Date),
6385 CatalogType::Float32 => Ok(SqlScalarType::Float32),
6386 CatalogType::Float64 => Ok(SqlScalarType::Float64),
6387 CatalogType::Int16 => Ok(SqlScalarType::Int16),
6388 CatalogType::Int32 => Ok(SqlScalarType::Int32),
6389 CatalogType::Int64 => Ok(SqlScalarType::Int64),
6390 CatalogType::UInt16 => Ok(SqlScalarType::UInt16),
6391 CatalogType::UInt32 => Ok(SqlScalarType::UInt32),
6392 CatalogType::UInt64 => Ok(SqlScalarType::UInt64),
6393 CatalogType::MzTimestamp => Ok(SqlScalarType::MzTimestamp),
6394 CatalogType::Interval => Ok(SqlScalarType::Interval),
6395 CatalogType::Jsonb => Ok(SqlScalarType::Jsonb),
6396 CatalogType::Oid => Ok(SqlScalarType::Oid),
6397 CatalogType::PgLegacyChar => Ok(SqlScalarType::PgLegacyChar),
6398 CatalogType::PgLegacyName => Ok(SqlScalarType::PgLegacyName),
6399 CatalogType::Pseudo => {
6400 sql_bail!(
6401 "cannot reference pseudo type {}",
6402 catalog.resolve_full_name(entry.name()).to_string()
6403 )
6404 }
6405 CatalogType::RegClass => Ok(SqlScalarType::RegClass),
6406 CatalogType::RegProc => Ok(SqlScalarType::RegProc),
6407 CatalogType::RegType => Ok(SqlScalarType::RegType),
6408 CatalogType::String => Ok(SqlScalarType::String),
6409 CatalogType::Time => Ok(SqlScalarType::Time),
6410 CatalogType::Uuid => Ok(SqlScalarType::Uuid),
6411 CatalogType::Int2Vector => Ok(SqlScalarType::Int2Vector),
6412 CatalogType::MzAclItem => Ok(SqlScalarType::MzAclItem),
6413 CatalogType::Numeric => unreachable!("handled above"),
6414 CatalogType::Char => unreachable!("handled above"),
6415 CatalogType::VarChar => unreachable!("handled above"),
6416 CatalogType::Timestamp => unreachable!("handled above"),
6417 CatalogType::TimestampTz => unreachable!("handled above"),
6418 }
6419 }
6420 }
6421}
6422
6423struct AggregateTableFuncVisitor<'a> {
6426 scx: &'a StatementContext<'a>,
6427 aggs: Vec<Function<Aug>>,
6428 within_aggregate: bool,
6429 tables: BTreeMap<Function<Aug>, String>,
6430 table_disallowed_context: Vec<&'static str>,
6431 in_select_item: bool,
6432 id_gen: IdGen,
6433 err: Option<PlanError>,
6434}
6435
6436impl<'a> AggregateTableFuncVisitor<'a> {
6437 fn new(scx: &'a StatementContext<'a>) -> AggregateTableFuncVisitor<'a> {
6438 AggregateTableFuncVisitor {
6439 scx,
6440 aggs: Vec::new(),
6441 within_aggregate: false,
6442 tables: BTreeMap::new(),
6443 table_disallowed_context: Vec::new(),
6444 in_select_item: false,
6445 id_gen: Default::default(),
6446 err: None,
6447 }
6448 }
6449
6450 fn into_result(
6451 self,
6452 ) -> Result<(Vec<Function<Aug>>, BTreeMap<Function<Aug>, String>), PlanError> {
6453 match self.err {
6454 Some(err) => Err(err),
6455 None => {
6456 let mut seen = BTreeSet::new();
6459 let aggs = self
6460 .aggs
6461 .into_iter()
6462 .filter(move |agg| seen.insert(agg.clone()))
6463 .collect();
6464 Ok((aggs, self.tables))
6465 }
6466 }
6467 }
6468}
6469
6470impl<'a> VisitMut<'_, Aug> for AggregateTableFuncVisitor<'a> {
6471 fn visit_function_mut(&mut self, func: &mut Function<Aug>) {
6472 let item = match self.scx.get_item_by_resolved_name(&func.name) {
6473 Ok(i) => i,
6474 Err(_) => return,
6476 };
6477
6478 match item.func() {
6479 Ok(Func::Aggregate { .. }) if func.over.is_none() => {
6482 if self.within_aggregate {
6483 self.err = Some(sql_err!("nested aggregate functions are not allowed",));
6484 return;
6485 }
6486 self.aggs.push(func.clone());
6487 let Function {
6488 name: _,
6489 args,
6490 filter,
6491 over: _,
6492 distinct: _,
6493 } = func;
6494 if let Some(filter) = filter {
6495 self.visit_expr_mut(filter);
6496 }
6497 let old_within_aggregate = self.within_aggregate;
6498 self.within_aggregate = true;
6499 self.table_disallowed_context
6500 .push("aggregate function calls");
6501
6502 self.visit_function_args_mut(args);
6503
6504 self.within_aggregate = old_within_aggregate;
6505 self.table_disallowed_context.pop();
6506 }
6507 Ok(Func::Table { .. }) => {
6508 self.table_disallowed_context.push("other table functions");
6509 visit_mut::visit_function_mut(self, func);
6510 self.table_disallowed_context.pop();
6511 }
6512 _ => visit_mut::visit_function_mut(self, func),
6513 }
6514 }
6515
6516 fn visit_query_mut(&mut self, _query: &mut Query<Aug>) {
6517 }
6519
6520 fn visit_expr_mut(&mut self, expr: &mut Expr<Aug>) {
6521 let (disallowed_context, func) = match expr {
6522 Expr::Case { .. } => (Some("CASE"), None),
6523 Expr::HomogenizingFunction {
6524 function: HomogenizingFunction::Coalesce,
6525 ..
6526 } => (Some("COALESCE"), None),
6527 Expr::Function(func) if self.in_select_item => {
6528 let mut table_func = None;
6531 if let Ok(item) = self.scx.get_item_by_resolved_name(&func.name) {
6532 if let Ok(Func::Table { .. }) = item.func() {
6533 if let Some(context) = self.table_disallowed_context.last() {
6534 self.err = Some(sql_err!(
6535 "table functions are not allowed in {} (function {})",
6536 context,
6537 func.name
6538 ));
6539 return;
6540 }
6541 table_func = Some(func.clone());
6542 }
6543 }
6544 (None, table_func)
6547 }
6548 _ => (None, None),
6549 };
6550 if let Some(func) = func {
6551 visit_mut::visit_expr_mut(self, expr);
6553 if let Function {
6555 name: _,
6556 args: _,
6557 filter: None,
6558 over: None,
6559 distinct: false,
6560 } = &func
6561 {
6562 let unique_id = self.id_gen.allocate_id();
6564 let id = self
6565 .tables
6566 .entry(func)
6567 .or_insert_with(|| format!("table_func_{unique_id}"));
6568 *expr = Expr::Identifier(vec![Ident::new_unchecked(id.clone())]);
6571 }
6572 }
6573 if let Some(context) = disallowed_context {
6574 self.table_disallowed_context.push(context);
6575 }
6576
6577 visit_mut::visit_expr_mut(self, expr);
6578
6579 if disallowed_context.is_some() {
6580 self.table_disallowed_context.pop();
6581 }
6582 }
6583
6584 fn visit_select_item_mut(&mut self, si: &mut SelectItem<Aug>) {
6585 let old = self.in_select_item;
6586 self.in_select_item = true;
6587 visit_mut::visit_select_item_mut(self, si);
6588 self.in_select_item = old;
6589 }
6590}
6591
6592#[derive(Default)]
6593struct WindowFuncCollector {
6594 window_funcs: Vec<Expr<Aug>>,
6595}
6596
6597impl WindowFuncCollector {
6598 fn into_result(self) -> Vec<Expr<Aug>> {
6599 let mut seen = BTreeSet::new();
6601 let window_funcs_dedupped = self
6602 .window_funcs
6603 .into_iter()
6604 .filter(move |expr| seen.insert(expr.clone()))
6605 .rev()
6608 .collect();
6609 window_funcs_dedupped
6610 }
6611}
6612
6613impl Visit<'_, Aug> for WindowFuncCollector {
6614 fn visit_expr(&mut self, expr: &Expr<Aug>) {
6615 match expr {
6616 Expr::Function(func) => {
6617 if func.over.is_some() {
6618 self.window_funcs.push(expr.clone());
6619 }
6620 }
6621 _ => (),
6622 }
6623 visit::visit_expr(self, expr);
6624 }
6625
6626 fn visit_query(&mut self, _query: &Query<Aug>) {
6627 }
6629}
6630
6631#[derive(Debug, Eq, PartialEq, Copy, Clone)]
6633pub enum QueryLifetime {
6634 OneShot,
6636 Index,
6638 MaterializedView,
6640 Subscribe,
6642 View,
6644 Source,
6646}
6647
6648impl QueryLifetime {
6649 pub fn is_one_shot(&self) -> bool {
6653 let result = match self {
6654 QueryLifetime::OneShot => true,
6655 QueryLifetime::Index => false,
6656 QueryLifetime::MaterializedView => false,
6657 QueryLifetime::Subscribe => false,
6658 QueryLifetime::View => false,
6659 QueryLifetime::Source => false,
6660 };
6661 assert_eq!(!result, self.is_maintained());
6662 result
6663 }
6664
6665 pub fn is_maintained(&self) -> bool {
6668 match self {
6669 QueryLifetime::OneShot => false,
6670 QueryLifetime::Index => true,
6671 QueryLifetime::MaterializedView => true,
6672 QueryLifetime::Subscribe => true,
6673 QueryLifetime::View => true,
6674 QueryLifetime::Source => true,
6675 }
6676 }
6677
6678 pub fn allow_show(&self) -> bool {
6680 match self {
6681 QueryLifetime::OneShot => true,
6682 QueryLifetime::Index => false,
6683 QueryLifetime::MaterializedView => false,
6684 QueryLifetime::Subscribe => true, QueryLifetime::View => false,
6686 QueryLifetime::Source => false,
6687 }
6688 }
6689}
6690
6691#[derive(Debug, Clone)]
6693pub struct CteDesc {
6694 pub name: String,
6695 pub desc: RelationDesc,
6696}
6697
6698#[derive(Debug, Clone)]
6700pub struct QueryContext<'a> {
6701 pub scx: &'a StatementContext<'a>,
6703 pub lifetime: QueryLifetime,
6705 pub outer_scopes: Vec<Scope>,
6707 pub outer_relation_types: Vec<SqlRelationType>,
6709 pub ctes: BTreeMap<LocalId, CteDesc>,
6711 pub name_manager: Rc<RefCell<NameManager>>,
6713 pub recursion_guard: RecursionGuard,
6714}
6715
6716impl CheckedRecursion for QueryContext<'_> {
6717 fn recursion_guard(&self) -> &RecursionGuard {
6718 &self.recursion_guard
6719 }
6720}
6721
6722impl<'a> QueryContext<'a> {
6723 pub fn root(scx: &'a StatementContext, lifetime: QueryLifetime) -> QueryContext<'a> {
6724 QueryContext {
6725 scx,
6726 lifetime,
6727 outer_scopes: vec![],
6728 outer_relation_types: vec![],
6729 ctes: BTreeMap::new(),
6730 name_manager: Rc::new(RefCell::new(NameManager::new())),
6731 recursion_guard: RecursionGuard::with_limit(1024), }
6733 }
6734
6735 fn relation_type(&self, expr: &HirRelationExpr) -> SqlRelationType {
6736 expr.typ(&self.outer_relation_types, &self.scx.param_types.borrow())
6737 }
6738
6739 fn derived_context(&self, scope: Scope, relation_type: SqlRelationType) -> QueryContext<'a> {
6742 let ctes = self.ctes.clone();
6743 let outer_scopes = iter::once(scope).chain(self.outer_scopes.clone()).collect();
6744 let outer_relation_types = iter::once(relation_type)
6745 .chain(self.outer_relation_types.clone())
6746 .collect();
6747 let name_manager = Rc::clone(&self.name_manager);
6749
6750 QueryContext {
6751 scx: self.scx,
6752 lifetime: self.lifetime,
6753 outer_scopes,
6754 outer_relation_types,
6755 ctes,
6756 name_manager,
6757 recursion_guard: self.recursion_guard.clone(),
6758 }
6759 }
6760
6761 fn empty_derived_context(&self) -> QueryContext<'a> {
6763 let scope = Scope::empty();
6764 let ty = SqlRelationType::empty();
6765 self.derived_context(scope, ty)
6766 }
6767
6768 pub fn resolve_table_name(
6771 &self,
6772 object: ResolvedItemName,
6773 ) -> Result<(HirRelationExpr, Scope), PlanError> {
6774 match object {
6775 ResolvedItemName::Item {
6776 id,
6777 full_name,
6778 version,
6779 ..
6780 } => {
6781 let item = self.scx.get_item(&id).at_version(version);
6782 let desc = match item.relation_desc() {
6783 Some(desc) => desc.clone(),
6784 None => {
6785 return Err(PlanError::InvalidDependency {
6786 name: full_name.to_string(),
6787 item_type: item.item_type().to_string(),
6788 });
6789 }
6790 };
6791 let expr = HirRelationExpr::Get {
6792 id: Id::Global(item.global_id()),
6793 typ: desc.typ().clone(),
6794 };
6795
6796 let name = full_name.into();
6797 let scope = Scope::from_source(Some(name), desc.iter_names().cloned());
6798
6799 Ok((expr, scope))
6800 }
6801 ResolvedItemName::Cte { id, name } => {
6802 let name = name.into();
6803 let cte = self.ctes.get(&id).unwrap();
6804 let expr = HirRelationExpr::Get {
6805 id: Id::Local(id),
6806 typ: cte.desc.typ().clone(),
6807 };
6808
6809 let scope = Scope::from_source(Some(name), cte.desc.iter_names());
6810
6811 Ok((expr, scope))
6812 }
6813 ResolvedItemName::Error => bail_internal!("should have been caught in name resolution"),
6814 }
6815 }
6816
6817 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6820 self.scx.humanize_sql_scalar_type(typ, postgres_compat)
6821 }
6822}
6823
6824#[derive(Debug, Clone)]
6826pub struct ExprContext<'a> {
6827 pub qcx: &'a QueryContext<'a>,
6828 pub name: &'a str,
6830 pub scope: &'a Scope,
6833 pub relation_type: &'a SqlRelationType,
6836 pub allow_aggregates: bool,
6838 pub allow_subqueries: bool,
6840 pub allow_parameters: bool,
6842 pub allow_windows: bool,
6844}
6845
6846impl CheckedRecursion for ExprContext<'_> {
6847 fn recursion_guard(&self) -> &RecursionGuard {
6848 &self.qcx.recursion_guard
6849 }
6850}
6851
6852impl<'a> ExprContext<'a> {
6853 pub fn catalog(&self) -> &dyn SessionCatalog {
6854 self.qcx.scx.catalog
6855 }
6856
6857 pub fn with_name(&self, name: &'a str) -> ExprContext<'a> {
6858 let mut ecx = self.clone();
6859 ecx.name = name;
6860 ecx
6861 }
6862
6863 pub fn column_type<E>(&self, expr: &E) -> E::Type
6864 where
6865 E: AbstractExpr,
6866 {
6867 expr.typ(
6868 &self.qcx.outer_relation_types,
6869 self.relation_type,
6870 &self.qcx.scx.param_types.borrow(),
6871 )
6872 }
6873
6874 pub fn scalar_type<E>(&self, expr: &E) -> <E::Type as AbstractColumnType>::AbstractScalarType
6875 where
6876 E: AbstractExpr,
6877 {
6878 self.column_type(expr).scalar_type()
6879 }
6880
6881 fn derived_query_context(&self) -> QueryContext<'_> {
6882 let mut scope = self.scope.clone();
6883 scope.lateral_barrier = true;
6884 self.qcx.derived_context(scope, self.relation_type.clone())
6885 }
6886
6887 pub fn require_feature_flag(&self, flag: &'static FeatureFlag) -> Result<(), PlanError> {
6888 self.qcx.scx.require_feature_flag(flag)
6889 }
6890
6891 pub fn param_types(&self) -> &RefCell<BTreeMap<usize, SqlScalarType>> {
6892 &self.qcx.scx.param_types
6893 }
6894
6895 pub fn humanize_sql_scalar_type(&self, typ: &SqlScalarType, postgres_compat: bool) -> String {
6898 self.qcx.scx.humanize_sql_scalar_type(typ, postgres_compat)
6899 }
6900
6901 pub fn intern(&self, item: &ScopeItem) -> Arc<str> {
6902 self.qcx.name_manager.borrow_mut().intern_scope_item(item)
6903 }
6904}
6905
6906#[derive(Debug, Clone)]
6912pub struct NameManager(BTreeSet<Arc<str>>);
6913
6914impl NameManager {
6915 pub fn new() -> Self {
6917 Self(BTreeSet::new())
6918 }
6919
6920 fn intern<S: AsRef<str>>(&mut self, s: S) -> Arc<str> {
6923 let s = s.as_ref();
6924 if let Some(interned) = self.0.get(s) {
6925 Arc::clone(interned)
6926 } else {
6927 let interned: Arc<str> = Arc::from(s);
6928 self.0.insert(Arc::clone(&interned));
6929 interned
6930 }
6931 }
6932
6933 pub fn intern_scope_item(&mut self, item: &ScopeItem) -> Arc<str> {
6936 self.intern(item.column_name.as_str())
6952 }
6953}
6954
6955#[cfg(test)]
6956mod test {
6957 use super::*;
6958
6959 #[mz_ore::test]
6964 pub fn test_name_manager_string_interning() {
6965 let mut nm = NameManager::new();
6966
6967 let orig_hi = "hi";
6968 let hi = nm.intern(orig_hi);
6969 let hello = nm.intern("hello");
6970
6971 assert_ne!(hi.as_ptr(), hello.as_ptr());
6972
6973 let hi2 = nm.intern("hi");
6975 assert_eq!(hi.as_ptr(), hi2.as_ptr());
6976
6977 let s = format!(
6979 "{}{}",
6980 hi.chars().nth(0).unwrap(),
6981 hi2.chars().nth(1).unwrap()
6982 );
6983 assert_ne!(orig_hi.as_ptr(), s.as_ptr());
6985
6986 let hi3 = nm.intern(s);
6987 assert_eq!(hi.as_ptr(), hi3.as_ptr());
6988 }
6989}