mz_sql/plan/lowering.rs
1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Lowering is the process of transforming a `HirRelationExpr`
11//! into a `MirRelationExpr`.
12//!
13//! The most crucial part of lowering is decorrelation; i.e.: rewriting a
14//! `HirScalarExpr` that may contain subqueries (e.g. `SELECT` or `EXISTS`)
15//! with instances of `MirScalarExpr` that contain none of these.
16//!
17//! Informally, a subquery should be viewed as a query that is executed in
18//! the context of some outer relation, for each row of that relation. The
19//! subqueries often contain references to the columns of the outer
20//! relation.
21//!
22//! The transformation we perform maintains an `outer` relation and then
23//! traverses the relation expression that may contain references to those
24//! outer columns. As subqueries are discovered, the current relation
25//! expression is recast as the outer expression until such a point as the
26//! scalar expression's evaluation can be determined and appended to each
27//! row of the previously outer relation.
28//!
29//! It is important that the outer columns (the initial columns) act as keys
30//! for all nested computation. When counts or other aggregations are
31//! performed, they should include not only the indicated keys but also all
32//! of the outer columns.
33//!
34//! The decorrelation transformation is initialized with an empty outer
35//! relation, but it seems entirely appropriate to decorrelate queries that
36//! contain "holes" from prepared statements, as if the query was a subquery
37//! against a relation containing the assignments of values to those holes.
38
39use std::collections::{BTreeMap, BTreeSet};
40use std::iter::repeat;
41
42use itertools::Itertools;
43use mz_expr::visit::Visit;
44use mz_expr::{AccessStrategy, AggregateFunc, MirRelationExpr, MirScalarExpr};
45use mz_ore::collections::CollectionExt;
46use mz_ore::stack::maybe_grow;
47use mz_repr::*;
48
49use crate::optimizer_metrics::OptimizerMetrics;
50use crate::plan::hir::{
51 AggregateExpr, ColumnOrder, ColumnRef, HirRelationExpr, HirScalarExpr, JoinKind, WindowExprType,
52};
53use crate::plan::{PlanError, transform_hir};
54use crate::session::vars::SystemVars;
55
56mod variadic_left;
57
58/// Maps a leveled column reference to a specific column.
59///
60/// Leveled column references are nested, so that larger levels are
61/// found early in a record and level zero is found at the end.
62///
63/// The column map only stores references for levels greater than zero,
64/// and column references at level zero simply start at the first column
65/// after all prior references.
66#[derive(Debug, Clone)]
67struct ColumnMap {
68 inner: BTreeMap<ColumnRef, usize>,
69}
70
71impl ColumnMap {
72 fn empty() -> ColumnMap {
73 Self::new(BTreeMap::new())
74 }
75
76 fn new(inner: BTreeMap<ColumnRef, usize>) -> ColumnMap {
77 ColumnMap { inner }
78 }
79
80 fn get(&self, col_ref: &ColumnRef) -> usize {
81 if col_ref.level == 0 {
82 self.inner.len() + col_ref.column
83 } else {
84 self.inner[col_ref]
85 }
86 }
87
88 fn len(&self) -> usize {
89 self.inner.len()
90 }
91
92 /// Updates references in the `ColumnMap` for use in a nested scope. The
93 /// provided `arity` must specify the arity of the current scope.
94 fn enter_scope(&self, arity: usize) -> ColumnMap {
95 // From the perspective of the nested scope, all existing column
96 // references will be one level greater.
97 let existing = self
98 .inner
99 .clone()
100 .into_iter()
101 .update(|(col, _i)| col.level += 1);
102
103 // All columns in the current scope become explicit entries in the
104 // immediate parent scope.
105 let new = (0..arity).map(|i| {
106 (
107 ColumnRef {
108 level: 1,
109 column: i,
110 },
111 self.len() + i,
112 )
113 });
114
115 ColumnMap::new(existing.chain(new).collect())
116 }
117}
118
119/// Map with the CTEs currently in scope.
120type CteMap = BTreeMap<mz_expr::LocalId, CteDesc>;
121
122/// Information about needed when finding a reference to a CTE in scope.
123#[derive(Clone)]
124struct CteDesc {
125 /// The new ID assigned to the lowered version of the CTE, which may not match
126 /// the ID of the input CTE.
127 new_id: mz_expr::LocalId,
128 /// The relation type of the CTE including the columns from the outer
129 /// context at the beginning.
130 relation_type: RelationType,
131 /// The outer relation the CTE was applied to.
132 outer_relation: MirRelationExpr,
133}
134
135#[derive(Debug)]
136pub struct Config {
137 /// Enable outer join lowering implemented in database-issues#6747.
138 pub enable_new_outer_join_lowering: bool,
139 /// Enable outer join lowering implemented in database-issues#7561.
140 pub enable_variadic_left_join_lowering: bool,
141 pub enable_guard_subquery_tablefunc: bool,
142}
143
144impl From<&SystemVars> for Config {
145 fn from(vars: &SystemVars) -> Self {
146 Self {
147 enable_new_outer_join_lowering: vars.enable_new_outer_join_lowering(),
148 enable_variadic_left_join_lowering: vars.enable_variadic_left_join_lowering(),
149 enable_guard_subquery_tablefunc: vars.enable_guard_subquery_tablefunc(),
150 }
151 }
152}
153
154/// Context passed to the lowering. This is wired to most parts of the lowering.
155pub(crate) struct Context<'a> {
156 /// Feature flags affecting the behavior of lowering.
157 pub config: &'a Config,
158 /// Optional, because some callers don't have an `OptimizerMetrics` handy. When it's None, we
159 /// simply don't write metrics.
160 pub metrics: Option<&'a OptimizerMetrics>,
161}
162
163impl HirRelationExpr {
164 /// Rewrite `self` into a `MirRelationExpr`.
165 /// This requires rewriting all correlated subqueries (nested `HirRelationExpr`s) into flat queries
166 #[mz_ore::instrument(target = "optimizer", level = "trace", name = "hir_to_mir")]
167 pub fn lower<C: Into<Config>>(
168 self,
169 config: C,
170 metrics: Option<&OptimizerMetrics>,
171 ) -> Result<MirRelationExpr, PlanError> {
172 let context = Context {
173 config: &config.into(),
174 metrics,
175 };
176 let result = match self {
177 // We directly rewrite a Constant into the corresponding `MirRelationExpr::Constant`
178 // to ensure that the downstream optimizer can easily bypass most
179 // irrelevant optimizations (e.g. reduce folding) for this expression
180 // without having to re-learn the fact that it is just a constant,
181 // as it would if the constant were wrapped in a Let-Get pair.
182 HirRelationExpr::Constant { rows, typ } => {
183 let rows: Vec<_> = rows.into_iter().map(|row| (row, Diff::ONE)).collect();
184 MirRelationExpr::Constant {
185 rows: Ok(rows),
186 typ,
187 }
188 }
189 mut other => {
190 let mut id_gen = mz_ore::id_gen::IdGen::default();
191 transform_hir::split_subquery_predicates(&mut other)?;
192 transform_hir::try_simplify_quantified_comparisons(&mut other)?;
193 transform_hir::fuse_window_functions(&mut other, &context)?;
194 MirRelationExpr::constant(vec![vec![]], RelationType::new(vec![])).let_in(
195 &mut id_gen,
196 |id_gen, get_outer| {
197 other.applied_to(
198 id_gen,
199 get_outer,
200 &ColumnMap::empty(),
201 &mut CteMap::new(),
202 &context,
203 )
204 },
205 )?
206 }
207 };
208
209 mz_repr::explain::trace_plan(&result);
210
211 Ok(result)
212 }
213
214 /// Return a `MirRelationExpr` which evaluates `self` once for each row of `get_outer`.
215 ///
216 /// For uncorrelated `self`, this should be the cross-product between `get_outer` and `self`.
217 /// When `self` references columns of `get_outer`, much more work needs to occur.
218 ///
219 /// The `col_map` argument contains mappings to some of the columns of `get_outer`, though
220 /// perhaps not all of them. It should be used as the basis of resolving column references,
221 /// but care must be taken when adding new columns that `get_outer.arity()` is where they
222 /// will start, rather than any function of `col_map`.
223 ///
224 /// The `get_outer` expression should be a `Get` with no duplicate rows, describing the distinct
225 /// assignment of values to outer rows.
226 fn applied_to(
227 self,
228 id_gen: &mut mz_ore::id_gen::IdGen,
229 get_outer: MirRelationExpr,
230 col_map: &ColumnMap,
231 cte_map: &mut CteMap,
232 context: &Context,
233 ) -> Result<MirRelationExpr, PlanError> {
234 maybe_grow(|| {
235 use MirRelationExpr as SR;
236
237 use HirRelationExpr::*;
238
239 if let MirRelationExpr::Get { .. } = &get_outer {
240 } else {
241 panic!(
242 "get_outer: expected a MirRelationExpr::Get, found {:?}",
243 get_outer
244 );
245 }
246 assert_eq!(col_map.len(), get_outer.arity());
247 Ok(match self {
248 Constant { rows, typ } => {
249 // Constant expressions are not correlated with `get_outer`, and should be cross-products.
250 get_outer.product(SR::Constant {
251 rows: Ok(rows.into_iter().map(|row| (row, Diff::ONE)).collect()),
252 typ,
253 })
254 }
255 Get { id, typ } => match id {
256 mz_expr::Id::Local(local_id) => {
257 let cte_desc = cte_map.get(&local_id).unwrap();
258 let get_cte = SR::Get {
259 id: mz_expr::Id::Local(cte_desc.new_id.clone()),
260 typ: cte_desc.relation_type.clone(),
261 access_strategy: AccessStrategy::UnknownOrLocal,
262 };
263 if get_outer == cte_desc.outer_relation {
264 // If the CTE was applied to the same exact relation, we can safely
265 // return a `Get` relation.
266 get_cte
267 } else {
268 // Otherwise, the new outer relation may contain more columns from some
269 // intermediate scope placed between the definition of the CTE and this
270 // reference of the CTE and/or more operations applied on top of the
271 // outer relation.
272 //
273 // An example of the latter is the following query:
274 //
275 // SELECT *
276 // FROM x,
277 // LATERAL(WITH a(m) as (SELECT max(y.a) FROM y WHERE y.a < x.a)
278 // SELECT (SELECT m FROM a) FROM y) b;
279 //
280 // When the CTE is lowered, the outer relation is `Get x`. But then,
281 // the reference of the CTE is applied to `Distinct(Join(Get x, Get y), x.*)`
282 // which has the same cardinality as `Get x`.
283 //
284 // In any case, `get_outer` is guaranteed to contain the columns of the
285 // outer relation the CTE was applied to at its prefix. Since, we must
286 // return a relation containing `get_outer`'s column at the beginning,
287 // we must build a join between `get_outer` and `get_cte` on their common
288 // columns.
289 let oa = get_outer.arity();
290 let cte_outer_columns = cte_desc.relation_type.arity() - typ.arity();
291 let equivalences = (0..cte_outer_columns)
292 .map(|pos| {
293 vec![
294 MirScalarExpr::column(pos),
295 MirScalarExpr::column(pos + oa),
296 ]
297 })
298 .collect();
299
300 // Project out the second copy of the common between `get_outer` and
301 // `cte_desc.outer_relation`.
302 let projection = (0..oa)
303 .chain(oa + cte_outer_columns..oa + cte_outer_columns + typ.arity())
304 .collect_vec();
305 SR::join_scalars(vec![get_outer, get_cte], equivalences)
306 .project(projection)
307 }
308 }
309 _ => {
310 // Get statements are only to external sources, and are not correlated with `get_outer`.
311 get_outer.product(SR::Get {
312 id,
313 typ,
314 access_strategy: AccessStrategy::UnknownOrLocal,
315 })
316 }
317 },
318 Let {
319 name: _,
320 id,
321 value,
322 body,
323 } => {
324 let value =
325 value.applied_to(id_gen, get_outer.clone(), col_map, cte_map, context)?;
326 value.let_in(id_gen, |id_gen, get_value| {
327 let (new_id, typ) = if let MirRelationExpr::Get {
328 id: mz_expr::Id::Local(id),
329 typ,
330 ..
331 } = get_value
332 {
333 (id, typ)
334 } else {
335 panic!(
336 "get_value: expected a MirRelationExpr::Get with local Id, found {:?}",
337 get_value
338 );
339 };
340 // Add the information about the CTE to the map and remove it when
341 // it goes out of scope.
342 let old_value = cte_map.insert(
343 id.clone(),
344 CteDesc {
345 new_id,
346 relation_type: typ,
347 outer_relation: get_outer.clone(),
348 },
349 );
350 let body = body.applied_to(id_gen, get_outer, col_map, cte_map, context);
351 if let Some(old_value) = old_value {
352 cte_map.insert(id, old_value);
353 } else {
354 cte_map.remove(&id);
355 }
356 body
357 })?
358 }
359 LetRec {
360 limit,
361 bindings,
362 body,
363 } => {
364 let num_bindings = bindings.len();
365
366 // We use the outer type with the HIR types to form MIR CTE types.
367 let outer_column_types = get_outer.typ().column_types;
368
369 // Rename and introduce all bindings.
370 let mut shadowed_bindings = Vec::with_capacity(num_bindings);
371 let mut mir_ids = Vec::with_capacity(num_bindings);
372 for (_name, id, _value, typ) in bindings.iter() {
373 let mir_id = mz_expr::LocalId::new(id_gen.allocate_id());
374 mir_ids.push(mir_id);
375 let shadowed = cte_map.insert(
376 id.clone(),
377 CteDesc {
378 new_id: mir_id,
379 relation_type: RelationType::new(
380 outer_column_types
381 .iter()
382 .cloned()
383 .chain(typ.column_types.iter().cloned())
384 .collect::<Vec<_>>(),
385 ),
386 outer_relation: get_outer.clone(),
387 },
388 );
389 shadowed_bindings.push((*id, shadowed));
390 }
391
392 let mut mir_values = Vec::with_capacity(num_bindings);
393 for (_name, _id, value, _typ) in bindings.into_iter() {
394 mir_values.push(value.applied_to(
395 id_gen,
396 get_outer.clone(),
397 col_map,
398 cte_map,
399 context,
400 )?);
401 }
402
403 let mir_body = body.applied_to(id_gen, get_outer, col_map, cte_map, context)?;
404
405 // Remove our bindings and reinstate any shadowed bindings.
406 for (id, shadowed) in shadowed_bindings {
407 if let Some(shadowed) = shadowed {
408 cte_map.insert(id, shadowed);
409 } else {
410 cte_map.remove(&id);
411 }
412 }
413
414 MirRelationExpr::LetRec {
415 ids: mir_ids,
416 values: mir_values,
417 // Copy the limit to each binding.
418 limits: repeat(limit).take(num_bindings).collect(),
419 body: Box::new(mir_body),
420 }
421 }
422 Project { input, outputs } => {
423 // Projections should be applied to the decorrelated `inner`, and to its columns,
424 // which means rebasing `outputs` to start `get_outer.arity()` columns later.
425 let input =
426 input.applied_to(id_gen, get_outer.clone(), col_map, cte_map, context)?;
427 let outputs = (0..get_outer.arity())
428 .chain(outputs.into_iter().map(|i| get_outer.arity() + i))
429 .collect::<Vec<_>>();
430 input.project(outputs)
431 }
432 Map { input, mut scalars } => {
433 // Scalar expressions may contain correlated subqueries. We must be cautious!
434
435 // We lower scalars in chunks, and must keep track of the
436 // arity of the HIR fragments lowered so far.
437 let mut lowered_arity = input.arity();
438
439 let mut input =
440 input.applied_to(id_gen, get_outer, col_map, cte_map, context)?;
441
442 // Lower subqueries in maximally sized batches, such as no subquery in the current
443 // batch depends on columns from the same batch.
444 // Note that subqueries in this projection may reference columns added by this
445 // Map operator, so we need to ensure these columns exist before lowering the
446 // subquery.
447 while !scalars.is_empty() {
448 let end_idx = scalars
449 .iter_mut()
450 .position(|s| {
451 let mut requires_nonexistent_column = false;
452 #[allow(deprecated)]
453 s.visit_columns(0, &mut |depth, col| {
454 if col.level == depth {
455 requires_nonexistent_column |= col.column >= lowered_arity
456 }
457 });
458 requires_nonexistent_column
459 })
460 .unwrap_or(scalars.len());
461 assert!(
462 end_idx > 0,
463 "a Map expression references itself or a later column; lowered_arity: {}, expressions: {:?}",
464 lowered_arity,
465 scalars
466 );
467
468 lowered_arity = lowered_arity + end_idx;
469 let scalars = scalars.drain(0..end_idx).collect_vec();
470
471 let old_arity = input.arity();
472 let (with_subqueries, subquery_map) = HirScalarExpr::lower_subqueries(
473 &scalars, id_gen, col_map, cte_map, input, context,
474 )?;
475 input = with_subqueries;
476
477 // We will proceed sequentially through the scalar expressions, for each transforming
478 // the decorrelated `input` into a relation with potentially more columns capable of
479 // addressing the needs of the scalar expression.
480 // Having done so, we add the scalar value of interest and trim off any other newly
481 // added columns.
482 //
483 // The sequential traversal is present as expressions are allowed to depend on the
484 // values of prior expressions.
485 let mut scalar_columns = Vec::new();
486 for scalar in scalars {
487 let scalar = scalar.applied_to(
488 id_gen,
489 col_map,
490 cte_map,
491 &mut input,
492 &Some(&subquery_map),
493 context,
494 )?;
495 input = input.map_one(scalar);
496 scalar_columns.push(input.arity() - 1);
497 }
498
499 // Discard any new columns added by the lowering of the scalar expressions
500 input = input.project((0..old_arity).chain(scalar_columns).collect());
501 }
502
503 input
504 }
505 CallTable { func, exprs } => {
506 // FlatMap expressions may contain correlated subqueries. Unlike Map they are not
507 // allowed to refer to the results of previous expressions, and we have a simpler
508 // implementation that appends all relevant columns first, then applies the flatmap
509 // operator to the result, then strips off any columns introduce by subqueries.
510
511 let mut input = get_outer;
512 let old_arity = input.arity();
513
514 let exprs = exprs
515 .into_iter()
516 .map(|e| e.applied_to(id_gen, col_map, cte_map, &mut input, &None, context))
517 .collect::<Result<Vec<_>, _>>()?;
518
519 let new_arity = input.arity();
520 let output_arity = func.output_arity();
521 input = input.flat_map(func, exprs);
522 if old_arity != new_arity {
523 // this means we added some columns to handle subqueries, and now we need to get rid of them
524 input = input.project(
525 (0..old_arity)
526 .chain(new_arity..new_arity + output_arity)
527 .collect(),
528 );
529 }
530 input
531 }
532 Filter { input, predicates } => {
533 // Filter expressions may contain correlated subqueries.
534 // We extend `get_outer` with sufficient values to determine the value of the predicate,
535 // then filter the results, then strip off any columns that were added for this purpose.
536 let mut input =
537 input.applied_to(id_gen, get_outer, col_map, cte_map, context)?;
538 for predicate in predicates {
539 let old_arity = input.arity();
540 let predicate = predicate
541 .applied_to(id_gen, col_map, cte_map, &mut input, &None, context)?;
542 let new_arity = input.arity();
543 input = input.filter(vec![predicate]);
544 if old_arity != new_arity {
545 // this means we added some columns to handle subqueries, and now we need to get rid of them
546 input = input.project((0..old_arity).collect());
547 }
548 }
549 input
550 }
551 Join {
552 left,
553 right,
554 on,
555 kind,
556 } if right.is_correlated() => {
557 // A correlated join is a join in which the right expression has
558 // access to the columns in the left expression. It turns out
559 // this is *exactly* our branch operator, plus some additional
560 // null handling in the case of left joins. (Right and full
561 // lateral joins are not permitted.)
562 //
563 // As with normal joins, the `on` predicate may be correlated,
564 // and we treat it as a filter that follows the branch.
565
566 assert!(kind.can_be_correlated());
567
568 let left = left.applied_to(id_gen, get_outer, col_map, cte_map, context)?;
569 left.let_in(id_gen, |id_gen, get_left| {
570 let apply_requires_distinct_outer = false;
571 let mut join = branch(
572 id_gen,
573 get_left.clone(),
574 col_map,
575 cte_map,
576 *right,
577 apply_requires_distinct_outer,
578 context,
579 |id_gen, right, get_left, col_map, cte_map, context| {
580 right.applied_to(id_gen, get_left, col_map, cte_map, context)
581 },
582 )?;
583
584 // Plan the `on` predicate.
585 let old_arity = join.arity();
586 let on =
587 on.applied_to(id_gen, col_map, cte_map, &mut join, &None, context)?;
588 join = join.filter(vec![on]);
589 let new_arity = join.arity();
590 if old_arity != new_arity {
591 // This means we added some columns to handle
592 // subqueries, and now we need to get rid of them.
593 join = join.project((0..old_arity).collect());
594 }
595
596 // If a left join, reintroduce any rows from the left that
597 // are missing, with nulls filled in for the right columns.
598 if let JoinKind::LeftOuter { .. } = kind {
599 let default = join
600 .typ()
601 .column_types
602 .into_iter()
603 .skip(get_left.arity())
604 .map(|typ| (Datum::Null, typ.scalar_type))
605 .collect();
606 get_left.lookup(id_gen, join, default)
607 } else {
608 Ok::<_, PlanError>(join)
609 }
610 })?
611 }
612 Join {
613 left,
614 right,
615 on,
616 kind,
617 } => {
618 if context.config.enable_variadic_left_join_lowering {
619 // Attempt to extract a stack of left joins.
620 if let JoinKind::LeftOuter = kind {
621 let mut rights = vec![(&*right, &on)];
622 let mut left_test = &left;
623 while let Join {
624 left,
625 right,
626 on,
627 kind: JoinKind::LeftOuter,
628 } = &**left_test
629 {
630 rights.push((&**right, on));
631 left_test = left;
632 }
633 if rights.len() > 1 {
634 // Defensively clone `cte_map` as it may be mutated.
635 let cte_map_clone = cte_map.clone();
636 if let Ok(Some(magic)) = variadic_left::attempt_left_join_magic(
637 left_test,
638 rights,
639 id_gen,
640 get_outer.clone(),
641 col_map,
642 cte_map,
643 context,
644 ) {
645 return Ok(magic);
646 } else {
647 cte_map.clone_from(&cte_map_clone);
648 }
649 }
650 }
651 }
652
653 // Both join expressions should be decorrelated, and then joined by their
654 // leading columns to form only those pairs corresponding to the same row
655 // of `get_outer`.
656 //
657 // The `on` predicate may contain correlated subqueries, and we treat it
658 // as though it was a filter, with the caveat that we also translate outer
659 // joins in this step. The post-filtration results need to be considered
660 // against the records present in the left and right (decorrelated) inputs,
661 // depending on the type of join.
662 let oa = get_outer.arity();
663 let left =
664 left.applied_to(id_gen, get_outer.clone(), col_map, cte_map, context)?;
665 let lt = left.typ().column_types.into_iter().skip(oa).collect_vec();
666 let la = lt.len();
667 left.let_in(id_gen, |id_gen, get_left| {
668 let right_col_map = col_map.enter_scope(0);
669 let right = right.applied_to(
670 id_gen,
671 get_outer.clone(),
672 &right_col_map,
673 cte_map,
674 context,
675 )?;
676 let rt = right.typ().column_types.into_iter().skip(oa).collect_vec();
677 let ra = rt.len();
678 right.let_in(id_gen, |id_gen, get_right| {
679 let mut product = SR::join(
680 vec![get_left.clone(), get_right.clone()],
681 (0..oa).map(|i| vec![(0, i), (1, i)]).collect(),
682 )
683 // Project away the repeated copy of get_outer's columns.
684 .project(
685 (0..(oa + la))
686 .chain((oa + la + oa)..(oa + la + oa + ra))
687 .collect(),
688 );
689
690 // Decorrelate and lower the `on` clause.
691 let on = on.applied_to(
692 id_gen,
693 col_map,
694 cte_map,
695 &mut product,
696 &None,
697 context,
698 )?;
699 // Collect the types of all subqueries appearing in
700 // the `on` clause. The subquery results were
701 // appended to `product` in the `on.applied_to(...)`
702 // call above.
703 let on_subquery_types = product
704 .typ()
705 .column_types
706 .drain(oa + la + ra..)
707 .collect_vec();
708 // Remember if `on` had any subqueries.
709 let on_has_subqueries = !on_subquery_types.is_empty();
710
711 // Attempt an efficient equijoin implementation, in which outer joins are
712 // more efficiently rendered than in general. This can return `None` if
713 // such a plan is not possible, for example if `on` does not describe an
714 // equijoin between columns of `left` and `right`.
715 if kind != JoinKind::Inner {
716 if let Some(joined) = attempt_outer_equijoin(
717 get_left.clone(),
718 get_right.clone(),
719 on.clone(),
720 on_subquery_types,
721 kind.clone(),
722 oa,
723 id_gen,
724 context,
725 )? {
726 if let Some(metrics) = context.metrics {
727 metrics.inc_outer_join_lowering("equi");
728 }
729 return Ok(joined);
730 }
731 }
732
733 // Otherwise, perform a more general join.
734 if let Some(metrics) = context.metrics {
735 metrics.inc_outer_join_lowering("general");
736 }
737 let mut join = product.filter(vec![on]);
738 if on_has_subqueries {
739 // This means that `on.applied_to(...)` appended
740 // some columns to handle subqueries, and now we
741 // need to get rid of them.
742 join = join.project((0..oa + la + ra).collect());
743 }
744 join.let_in(id_gen, |id_gen, get_join| {
745 let mut result = get_join.clone();
746 if let JoinKind::LeftOuter { .. } | JoinKind::FullOuter { .. } =
747 kind
748 {
749 let left_outer = get_left.clone().anti_lookup::<PlanError>(
750 id_gen,
751 get_join.clone(),
752 rt.into_iter()
753 .map(|typ| (Datum::Null, typ.scalar_type))
754 .collect(),
755 )?;
756 result = result.union(left_outer);
757 }
758 if let JoinKind::RightOuter | JoinKind::FullOuter = kind {
759 let right_outer = get_right
760 .clone()
761 .anti_lookup::<PlanError>(
762 id_gen,
763 get_join
764 // need to swap left and right to make the anti_lookup work
765 .project(
766 (0..oa)
767 .chain((oa + la)..(oa + la + ra))
768 .chain((oa)..(oa + la))
769 .collect(),
770 ),
771 lt.into_iter()
772 .map(|typ| (Datum::Null, typ.scalar_type))
773 .collect(),
774 )?
775 // swap left and right back again
776 .project(
777 (0..oa)
778 .chain((oa + ra)..(oa + ra + la))
779 .chain((oa)..(oa + ra))
780 .collect(),
781 );
782 result = result.union(right_outer);
783 }
784 Ok::<MirRelationExpr, PlanError>(result)
785 })
786 })
787 })?
788 }
789 Union { base, inputs } => {
790 // Union is uncomplicated.
791 SR::Union {
792 base: Box::new(base.applied_to(
793 id_gen,
794 get_outer.clone(),
795 col_map,
796 cte_map,
797 context,
798 )?),
799 inputs: inputs
800 .into_iter()
801 .map(|input| {
802 input.applied_to(
803 id_gen,
804 get_outer.clone(),
805 col_map,
806 cte_map,
807 context,
808 )
809 })
810 .collect::<Result<Vec<_>, _>>()?,
811 }
812 }
813 Reduce {
814 input,
815 group_key,
816 aggregates,
817 expected_group_size,
818 } => {
819 // Reduce may contain expressions with correlated subqueries.
820 // In addition, here an empty reduction key signifies that we need to supply default values
821 // in the case that there are no results (as in a SQL aggregation without an explicit GROUP BY).
822 let mut input =
823 input.applied_to(id_gen, get_outer.clone(), col_map, cte_map, context)?;
824 let applied_group_key = (0..get_outer.arity())
825 .chain(group_key.iter().map(|i| get_outer.arity() + i))
826 .collect();
827 let applied_aggregates = aggregates
828 .into_iter()
829 .map(|aggregate| {
830 aggregate.applied_to(id_gen, col_map, cte_map, &mut input, context)
831 })
832 .collect::<Result<Vec<_>, _>>()?;
833 let input_type = input.typ();
834 let default = applied_aggregates
835 .iter()
836 .map(|agg| {
837 (
838 agg.func.default(),
839 agg.typ(&input_type.column_types).scalar_type,
840 )
841 })
842 .collect();
843 // NOTE we don't need to remove any extra columns from aggregate.applied_to above because the reduce will do that anyway
844 let mut reduced =
845 input.reduce(applied_group_key, applied_aggregates, expected_group_size);
846
847 // Introduce default values in the case the group key is empty.
848 if group_key.is_empty() {
849 reduced = get_outer.lookup::<PlanError>(id_gen, reduced, default)?;
850 }
851 reduced
852 }
853 Distinct { input } => {
854 // Distinct is uncomplicated.
855 input
856 .applied_to(id_gen, get_outer, col_map, cte_map, context)?
857 .distinct()
858 }
859 TopK {
860 input,
861 group_key,
862 order_key,
863 limit,
864 offset,
865 expected_group_size,
866 } => {
867 // TopK is uncomplicated, except that we must group by the columns of `get_outer` as well.
868 let mut input =
869 input.applied_to(id_gen, get_outer.clone(), col_map, cte_map, context)?;
870 let mut applied_group_key: Vec<_> = (0..get_outer.arity())
871 .chain(group_key.iter().map(|i| get_outer.arity() + i))
872 .collect();
873 let applied_order_key = order_key
874 .iter()
875 .map(|column_order| ColumnOrder {
876 column: column_order.column + get_outer.arity(),
877 desc: column_order.desc,
878 nulls_last: column_order.nulls_last,
879 })
880 .collect();
881
882 let old_arity = input.arity();
883
884 // Lower `limit`, which may introduce new columns if is a correlated subquery.
885 let mut limit_mir = None;
886 if let Some(limit) = limit {
887 limit_mir = Some(
888 limit
889 .applied_to(id_gen, col_map, cte_map, &mut input, &None, context)?,
890 );
891 }
892
893 let new_arity = input.arity();
894 // Extend the key to contain any new columns.
895 applied_group_key.extend(old_arity..new_arity);
896
897 let offset = offset
898 .try_into_literal_int64()
899 .expect("Should be a Literal by this time")
900 .try_into()
901 .expect("Should have checked non-negativity of OFFSET clause already");
902 let mut result = input.top_k(
903 applied_group_key,
904 applied_order_key,
905 limit_mir,
906 offset,
907 expected_group_size,
908 );
909
910 // If new columns were added for `limit` we must remove them.
911 if old_arity != new_arity {
912 result = result.project((0..old_arity).collect());
913 }
914
915 result
916 }
917 Negate { input } => {
918 // Negate is uncomplicated.
919 input
920 .applied_to(id_gen, get_outer, col_map, cte_map, context)?
921 .negate()
922 }
923 Threshold { input } => {
924 // Threshold is uncomplicated.
925 input
926 .applied_to(id_gen, get_outer, col_map, cte_map, context)?
927 .threshold()
928 }
929 })
930 })
931 }
932}
933
934impl HirScalarExpr {
935 /// Rewrite `self` into a `mz_expr::ScalarExpr` which can be applied to the modified `inner`.
936 ///
937 /// This method is responsible for decorrelating subqueries in `self` by introducing further columns
938 /// to `inner`, and rewriting `self` to refer to its physical columns (specified by `usize` positions).
939 /// The most complicated logic is for the scalar expressions that involve subqueries, each of which are
940 /// documented in more detail closer to their logic.
941 ///
942 /// This process presumes that `inner` is the result of decorrelation, meaning its first several columns
943 /// may be inherited from outer relations. The `col_map` column map should provide specific offsets where
944 /// each of these references can be found.
945 fn applied_to(
946 self,
947 id_gen: &mut mz_ore::id_gen::IdGen,
948 col_map: &ColumnMap,
949 cte_map: &mut CteMap,
950 inner: &mut MirRelationExpr,
951 subquery_map: &Option<&BTreeMap<HirScalarExpr, usize>>,
952 context: &Context,
953 ) -> Result<MirScalarExpr, PlanError> {
954 maybe_grow(|| {
955 use MirScalarExpr as SS;
956
957 use HirScalarExpr::*;
958
959 if let Some(subquery_map) = subquery_map {
960 if let Some(col) = subquery_map.get(&self) {
961 return Ok(SS::column(*col));
962 }
963 }
964
965 Ok::<MirScalarExpr, PlanError>(match self {
966 Column(col_ref, name) => SS::Column(col_map.get(&col_ref), name),
967 Literal(row, typ, _name) => SS::Literal(Ok(row), typ),
968 Parameter(_, _name) => {
969 panic!("cannot decorrelate expression with unbound parameters")
970 }
971 CallUnmaterializable(func, _name) => SS::CallUnmaterializable(func),
972 CallUnary {
973 func,
974 expr,
975 name: _,
976 } => SS::CallUnary {
977 func,
978 expr: Box::new(expr.applied_to(
979 id_gen,
980 col_map,
981 cte_map,
982 inner,
983 subquery_map,
984 context,
985 )?),
986 },
987 CallBinary {
988 func,
989 expr1,
990 expr2,
991 name: _,
992 } => SS::CallBinary {
993 func,
994 expr1: Box::new(expr1.applied_to(
995 id_gen,
996 col_map,
997 cte_map,
998 inner,
999 subquery_map,
1000 context,
1001 )?),
1002 expr2: Box::new(expr2.applied_to(
1003 id_gen,
1004 col_map,
1005 cte_map,
1006 inner,
1007 subquery_map,
1008 context,
1009 )?),
1010 },
1011 CallVariadic {
1012 func,
1013 exprs,
1014 name: _,
1015 } => SS::CallVariadic {
1016 func,
1017 exprs: exprs
1018 .into_iter()
1019 .map(|expr| {
1020 expr.applied_to(id_gen, col_map, cte_map, inner, subquery_map, context)
1021 })
1022 .collect::<Result<Vec<_>, _>>()?,
1023 },
1024 If {
1025 cond,
1026 then,
1027 els,
1028 name,
1029 } => {
1030 // The `If` case is complicated by the fact that we do not want to
1031 // apply the `then` or `else` logic to tuples that respectively do
1032 // not or do pass the `cond` test. Our strategy is to independently
1033 // decorrelate the `then` and `else` logic, and apply each to tuples
1034 // that respectively pass and do not pass the `cond` logic (which is
1035 // executed, and so decorrelated, for all tuples).
1036 //
1037 // Informally, we turn the `if` statement into:
1038 //
1039 // let then_case = inner.filter(cond).map(then);
1040 // let else_case = inner.filter(!cond).map(else);
1041 // return then_case.concat(else_case);
1042 //
1043 // We only require this if either expression would result in any
1044 // computation beyond the expr itself, which we will interpret as
1045 // "introduces additional columns". In the absence of correlation,
1046 // we should just retain a `ScalarExpr::If` expression; the inverse
1047 // transformation as above is complicated to recover after the fact,
1048 // and we would benefit from not introducing the complexity.
1049
1050 let inner_arity = inner.arity();
1051 let cond_expr =
1052 cond.applied_to(id_gen, col_map, cte_map, inner, subquery_map, context)?;
1053
1054 // Defensive copies, in case we mangle these in decorrelation.
1055 let inner_clone = inner.clone();
1056 let then_clone = then.clone();
1057 let else_clone = els.clone();
1058
1059 let cond_arity = inner.arity();
1060 let then_expr =
1061 then.applied_to(id_gen, col_map, cte_map, inner, subquery_map, context)?;
1062 let else_expr =
1063 els.applied_to(id_gen, col_map, cte_map, inner, subquery_map, context)?;
1064
1065 if cond_arity == inner.arity() {
1066 // If no additional columns were added, we simply return the
1067 // `If` variant with the updated expressions.
1068 SS::If {
1069 cond: Box::new(cond_expr),
1070 then: Box::new(then_expr),
1071 els: Box::new(else_expr),
1072 }
1073 } else {
1074 // If columns were added, we need a more careful approach, as
1075 // described above. First, we need to de-correlate each of
1076 // the two expressions independently, and apply their cases
1077 // as `MirRelationExpr::Map` operations.
1078
1079 *inner = inner_clone.let_in(id_gen, |id_gen, get_inner| {
1080 // Restrict to records satisfying `cond_expr` and apply `then` as a map.
1081 let mut then_inner = get_inner.clone().filter(vec![cond_expr.clone()]);
1082 let then_expr = then_clone.applied_to(
1083 id_gen,
1084 col_map,
1085 cte_map,
1086 &mut then_inner,
1087 subquery_map,
1088 context,
1089 )?;
1090 let then_arity = then_inner.arity();
1091 then_inner = then_inner
1092 .map_one(then_expr)
1093 .project((0..inner_arity).chain(Some(then_arity)).collect());
1094
1095 // Restrict to records not satisfying `cond_expr` and apply `els` as a map.
1096 let mut else_inner = get_inner.filter(vec![SS::CallVariadic {
1097 func: mz_expr::VariadicFunc::Or,
1098 exprs: vec![
1099 cond_expr
1100 .clone()
1101 .call_binary(SS::literal_false(), mz_expr::BinaryFunc::Eq),
1102 cond_expr.clone().call_is_null(),
1103 ],
1104 }]);
1105 let else_expr = else_clone.applied_to(
1106 id_gen,
1107 col_map,
1108 cte_map,
1109 &mut else_inner,
1110 subquery_map,
1111 context,
1112 )?;
1113 let else_arity = else_inner.arity();
1114 else_inner = else_inner
1115 .map_one(else_expr)
1116 .project((0..inner_arity).chain(Some(else_arity)).collect());
1117
1118 // concatenate the two results.
1119 Ok::<MirRelationExpr, PlanError>(then_inner.union(else_inner))
1120 })?;
1121
1122 SS::Column(inner_arity, name)
1123 }
1124 }
1125
1126 // Subqueries!
1127 // These are surprisingly subtle. Things to be careful of:
1128
1129 // Anything in the subquery that cares about row counts (Reduce/Distinct/Negate/Threshold) must not:
1130 // * change the row counts of the outer query
1131 // * accidentally compute its own value using the row counts of the outer query
1132 // Use `branch` to calculate the subquery once for each __distinct__ key in the outer
1133 // query and then join the answers back on to the original rows of the outer query.
1134
1135 // When the subquery would return 0 rows for some row in the outer query, `subquery.applied_to(get_inner)` will not have any corresponding row.
1136 // Use `lookup` if you need to add default values for cases when the subquery returns 0 rows.
1137 Exists(expr, name) => {
1138 let apply_requires_distinct_outer = true;
1139 *inner = apply_existential_subquery(
1140 id_gen,
1141 inner.take_dangerous(),
1142 col_map,
1143 cte_map,
1144 *expr,
1145 apply_requires_distinct_outer,
1146 context,
1147 )?;
1148 SS::Column(inner.arity() - 1, name)
1149 }
1150
1151 Select(expr, name) => {
1152 let apply_requires_distinct_outer = true;
1153 *inner = apply_scalar_subquery(
1154 id_gen,
1155 inner.take_dangerous(),
1156 col_map,
1157 cte_map,
1158 *expr,
1159 apply_requires_distinct_outer,
1160 context,
1161 )?;
1162 SS::Column(inner.arity() - 1, name)
1163 }
1164 Windowing(expr, _name) => {
1165 let partition_by = expr.partition_by;
1166 let order_by = expr.order_by;
1167
1168 // argument lowering for scalar window functions
1169 // (We need to specify the & _ in the arguments because of this problem:
1170 // https://users.rust-lang.org/t/the-implementation-of-fnonce-is-not-general-enough/72141/3 )
1171 let scalar_lower_args =
1172 |_id_gen: &mut _,
1173 _col_map: &_,
1174 _cte_map: &mut _,
1175 _get_inner: &mut _,
1176 _subquery_map: &Option<&_>,
1177 order_by_mir: Vec<MirScalarExpr>,
1178 original_row_record,
1179 original_row_record_type: ScalarType| {
1180 let agg_input = MirScalarExpr::CallVariadic {
1181 func: mz_expr::VariadicFunc::ListCreate {
1182 elem_type: original_row_record_type.clone(),
1183 },
1184 exprs: vec![original_row_record],
1185 };
1186 let mut agg_input = vec![agg_input];
1187 agg_input.extend(order_by_mir.clone());
1188 let agg_input = MirScalarExpr::CallVariadic {
1189 func: mz_expr::VariadicFunc::RecordCreate {
1190 field_names: (0..agg_input.len())
1191 .map(|_| ColumnName::from("?column?"))
1192 .collect_vec(),
1193 },
1194 exprs: agg_input,
1195 };
1196 let list_type = ScalarType::List {
1197 element_type: Box::new(original_row_record_type),
1198 custom_id: None,
1199 };
1200 let agg_input_type = ScalarType::Record {
1201 fields: std::iter::once(&list_type)
1202 .map(|t| {
1203 (ColumnName::from("?column?"), t.clone().nullable(false))
1204 })
1205 .collect(),
1206 custom_id: None,
1207 }
1208 .nullable(false);
1209
1210 Ok((agg_input, agg_input_type))
1211 };
1212
1213 // argument lowering for value window functions and aggregate window functions
1214 let value_or_aggr_lower_args = |hir_encoded_args: Box<HirScalarExpr>| {
1215 |id_gen: &mut _,
1216 col_map: &_,
1217 cte_map: &mut _,
1218 get_inner: &mut _,
1219 subquery_map: &Option<&_>,
1220 order_by_mir: Vec<MirScalarExpr>,
1221 original_row_record,
1222 original_row_record_type| {
1223 // Creates [((OriginalRow, EncodedArgs), OrderByExprs...)]
1224
1225 // Compute the encoded args for all rows
1226 let mir_encoded_args = hir_encoded_args.applied_to(
1227 id_gen,
1228 col_map,
1229 cte_map,
1230 get_inner,
1231 subquery_map,
1232 context,
1233 )?;
1234 let mir_encoded_args_type = mir_encoded_args
1235 .typ(&get_inner.typ().column_types)
1236 .scalar_type;
1237
1238 // Build a new record that has two fields:
1239 // 1. the original row in a record
1240 // 2. the encoded args (which can be either a single value, or a record
1241 // if the window function has multiple arguments, such as `lag`)
1242 let fn_input_record_fields: Box<[_]> =
1243 [original_row_record_type, mir_encoded_args_type]
1244 .iter()
1245 .map(|t| {
1246 (ColumnName::from("?column?"), t.clone().nullable(false))
1247 })
1248 .collect();
1249 let fn_input_record = MirScalarExpr::CallVariadic {
1250 func: mz_expr::VariadicFunc::RecordCreate {
1251 field_names: fn_input_record_fields
1252 .iter()
1253 .map(|(n, _)| n.clone())
1254 .collect_vec(),
1255 },
1256 exprs: vec![original_row_record, mir_encoded_args],
1257 };
1258 let fn_input_record_type = ScalarType::Record {
1259 fields: fn_input_record_fields,
1260 custom_id: None,
1261 }
1262 .nullable(false);
1263
1264 // Build a new record with the record above + the ORDER BY exprs
1265 // This follows the standard encoding of ORDER BY exprs used by aggregate functions
1266 let mut agg_input = vec![fn_input_record];
1267 agg_input.extend(order_by_mir.clone());
1268 let agg_input = MirScalarExpr::CallVariadic {
1269 func: mz_expr::VariadicFunc::RecordCreate {
1270 field_names: (0..agg_input.len())
1271 .map(|_| ColumnName::from("?column?"))
1272 .collect_vec(),
1273 },
1274 exprs: agg_input,
1275 };
1276
1277 let agg_input_type = ScalarType::Record {
1278 fields: [(
1279 ColumnName::from("?column?"),
1280 fn_input_record_type.nullable(false),
1281 )]
1282 .into(),
1283 custom_id: None,
1284 }
1285 .nullable(false);
1286
1287 Ok((agg_input, agg_input_type))
1288 }
1289 };
1290
1291 match expr.func {
1292 WindowExprType::Scalar(scalar_window_expr) => {
1293 let mir_aggr_func = scalar_window_expr.into_expr();
1294 Self::window_func_applied_to(
1295 id_gen,
1296 col_map,
1297 cte_map,
1298 inner,
1299 subquery_map,
1300 partition_by,
1301 order_by,
1302 mir_aggr_func,
1303 scalar_lower_args,
1304 context,
1305 )?
1306 }
1307 WindowExprType::Value(value_window_expr) => {
1308 let (hir_encoded_args, mir_aggr_func) = value_window_expr.into_expr();
1309
1310 Self::window_func_applied_to(
1311 id_gen,
1312 col_map,
1313 cte_map,
1314 inner,
1315 subquery_map,
1316 partition_by,
1317 order_by,
1318 mir_aggr_func,
1319 value_or_aggr_lower_args(hir_encoded_args),
1320 context,
1321 )?
1322 }
1323 WindowExprType::Aggregate(aggr_window_expr) => {
1324 let (hir_encoded_args, mir_aggr_func) = aggr_window_expr.into_expr();
1325
1326 Self::window_func_applied_to(
1327 id_gen,
1328 col_map,
1329 cte_map,
1330 inner,
1331 subquery_map,
1332 partition_by,
1333 order_by,
1334 mir_aggr_func,
1335 value_or_aggr_lower_args(hir_encoded_args),
1336 context,
1337 )?
1338 }
1339 }
1340 }
1341 })
1342 })
1343 }
1344
1345 fn window_func_applied_to<F>(
1346 id_gen: &mut mz_ore::id_gen::IdGen,
1347 col_map: &ColumnMap,
1348 cte_map: &mut CteMap,
1349 inner: &mut MirRelationExpr,
1350 subquery_map: &Option<&BTreeMap<HirScalarExpr, usize>>,
1351 partition_by: Vec<HirScalarExpr>,
1352 order_by: Vec<HirScalarExpr>,
1353 mir_aggr_func: AggregateFunc,
1354 lower_args: F,
1355 context: &Context,
1356 ) -> Result<MirScalarExpr, PlanError>
1357 where
1358 F: FnOnce(
1359 &mut mz_ore::id_gen::IdGen,
1360 &ColumnMap,
1361 &mut CteMap,
1362 &mut MirRelationExpr,
1363 &Option<&BTreeMap<HirScalarExpr, usize>>,
1364 Vec<MirScalarExpr>,
1365 MirScalarExpr,
1366 ScalarType,
1367 ) -> Result<(MirScalarExpr, ColumnType), PlanError>,
1368 {
1369 // Example MIRs for a window function (specifically, a window aggregation):
1370 //
1371 // CREATE TABLE t7(x INT, y INT);
1372 //
1373 // explain decorrelated plan for select sum(x*y) over (partition by x+y order by x-y, x/y) from t7;
1374 //
1375 // Decorrelated Plan
1376 // Project (#3)
1377 // Map (#2)
1378 // Project (#3..=#5)
1379 // Map (record_get[0](record_get[1](#2)), record_get[1](record_get[1](#2)), record_get[0](#2))
1380 // FlatMap unnest_list(#1)
1381 // Reduce group_by=[#2] aggregates=[window_agg[sum order_by=[#0 asc nulls_last, #1 asc nulls_last]](row(row(row(#0, #1), (#0 * #1)), (#0 - #1), (#0 / #1)))]
1382 // Map ((#0 + #1))
1383 // CrossJoin
1384 // Constant
1385 // - ()
1386 // Get materialize.public.t7
1387 //
1388 // The same query after optimizations:
1389 //
1390 // explain select sum(x*y) over (partition by x+y order by x-y, x/y) from t7;
1391 //
1392 // Optimized Plan
1393 // Explained Query:
1394 // Project (#2)
1395 // Map (record_get[0](#1))
1396 // FlatMap unnest_list(#0)
1397 // Project (#1)
1398 // Reduce group_by=[(#0 + #1)] aggregates=[window_agg[sum order_by=[#0 asc nulls_last, #1 asc nulls_last]](row(row(row(#0, #1), (#0 * #1)), (#0 - #1), (#0 / #1)))]
1399 // ReadStorage materialize.public.t7
1400 //
1401 // The `row(row(row(...), ...), ...)` stuff means the following:
1402 // `row(row(row(<original row>), <arguments to window function>), <order by values>...)`
1403 // - The <arguments to window function> can be either a single value or itself a
1404 // `row` if there are multiple arguments.
1405 // - The <order by values> are _not_ wrapped in a `row`, even if there are more than one
1406 // ORDER BY columns.
1407 // - The <original row> currently always captures the entire original row. This should
1408 // improve when we make `ProjectionPushdown` smarter, see
1409 // https://github.com/MaterializeInc/database-issues/issues/5090
1410 //
1411 // TODO:
1412 // We should probably introduce some dedicated Datum constructor functions instead of `row`
1413 // to make MIR plans and MIR construction/manipulation code more readable. Additionally, we
1414 // might even introduce dedicated Datum enum variants, so that the rendering code also
1415 // becomes more readable (and possibly slightly more performant).
1416
1417 *inner = inner
1418 .take_dangerous()
1419 .let_in(id_gen, |id_gen, mut get_inner| {
1420 let order_by_mir = order_by
1421 .into_iter()
1422 .map(|o| {
1423 o.applied_to(
1424 id_gen,
1425 col_map,
1426 cte_map,
1427 &mut get_inner,
1428 subquery_map,
1429 context,
1430 )
1431 })
1432 .collect::<Result<Vec<_>, _>>()?;
1433
1434 // Record input arity here so that any group_keys that need to mutate get_inner
1435 // don't add those columns to the aggregate input.
1436 let input_type = get_inner.typ();
1437 let input_arity = input_type.arity();
1438 // The reduction that computes the window function must be keyed on the columns
1439 // from the outer context, plus the expressions in the partition key. The current
1440 // subquery will be 'executed' for every distinct row from the outer context so
1441 // by putting the outer columns in the grouping key we isolate each re-execution.
1442 let mut group_key = col_map
1443 .inner
1444 .iter()
1445 .map(|(_, outer_col)| *outer_col)
1446 .sorted()
1447 .collect_vec();
1448 for p in partition_by {
1449 let key = p.applied_to(
1450 id_gen,
1451 col_map,
1452 cte_map,
1453 &mut get_inner,
1454 subquery_map,
1455 context,
1456 )?;
1457 if let MirScalarExpr::Column(c, _name) = key {
1458 group_key.push(c);
1459 } else {
1460 get_inner = get_inner.map_one(key);
1461 group_key.push(get_inner.arity() - 1);
1462 }
1463 }
1464
1465 get_inner.let_in(id_gen, |id_gen, mut get_inner| {
1466 // Original columns of the relation
1467 let fields: Box<_> = input_type
1468 .column_types
1469 .iter()
1470 .take(input_arity)
1471 .map(|t| (ColumnName::from("?column?"), t.clone()))
1472 .collect();
1473
1474 // Original row made into a record
1475 let original_row_record = MirScalarExpr::CallVariadic {
1476 func: mz_expr::VariadicFunc::RecordCreate {
1477 field_names: fields.iter().map(|(name, _)| name.clone()).collect_vec(),
1478 },
1479 exprs: (0..input_arity).map(MirScalarExpr::column).collect_vec(),
1480 };
1481 let original_row_record_type = ScalarType::Record {
1482 fields,
1483 custom_id: None,
1484 };
1485
1486 let (agg_input, agg_input_type) = lower_args(
1487 id_gen,
1488 col_map,
1489 cte_map,
1490 &mut get_inner,
1491 subquery_map,
1492 order_by_mir,
1493 original_row_record,
1494 original_row_record_type,
1495 )?;
1496
1497 let aggregate = mz_expr::AggregateExpr {
1498 func: mir_aggr_func,
1499 expr: agg_input,
1500 distinct: false,
1501 };
1502
1503 // Actually call reduce with the window function
1504 // The output of the aggregation function should be a list of tuples that has
1505 // the result in the first position, and the original row in the second position
1506 let mut reduce = get_inner
1507 .reduce(group_key.clone(), vec![aggregate.clone()], None)
1508 .flat_map(
1509 mz_expr::TableFunc::UnnestList {
1510 el_typ: aggregate
1511 .func
1512 .output_type(agg_input_type)
1513 .scalar_type
1514 .unwrap_list_element_type()
1515 .clone(),
1516 },
1517 vec![MirScalarExpr::column(group_key.len())],
1518 );
1519 let record_col = reduce.arity() - 1;
1520
1521 // Unpack the record output by the window function
1522 for c in 0..input_arity {
1523 reduce = reduce.take_dangerous().map_one(MirScalarExpr::CallUnary {
1524 func: mz_expr::UnaryFunc::RecordGet(mz_expr::func::RecordGet(c)),
1525 expr: Box::new(MirScalarExpr::CallUnary {
1526 func: mz_expr::UnaryFunc::RecordGet(mz_expr::func::RecordGet(1)),
1527 expr: Box::new(MirScalarExpr::column(record_col)),
1528 }),
1529 });
1530 }
1531
1532 // Append the column with the result of the window function.
1533 reduce = reduce.take_dangerous().map_one(MirScalarExpr::CallUnary {
1534 func: mz_expr::UnaryFunc::RecordGet(mz_expr::func::RecordGet(0)),
1535 expr: Box::new(MirScalarExpr::column(record_col)),
1536 });
1537
1538 let agg_col = record_col + 1 + input_arity;
1539 Ok::<_, PlanError>(reduce.project((record_col + 1..agg_col + 1).collect_vec()))
1540 })
1541 })?;
1542 Ok(MirScalarExpr::column(inner.arity() - 1))
1543 }
1544
1545 /// Applies the subqueries in the given list of scalar expressions to every distinct
1546 /// value of the given relation and returns a join of the given relation with all
1547 /// the subqueries found, and the mapping of scalar expressions with columns projected
1548 /// by the returned join that will hold their results.
1549 fn lower_subqueries(
1550 exprs: &[Self],
1551 id_gen: &mut mz_ore::id_gen::IdGen,
1552 col_map: &ColumnMap,
1553 cte_map: &mut CteMap,
1554 inner: MirRelationExpr,
1555 context: &Context,
1556 ) -> Result<(MirRelationExpr, BTreeMap<HirScalarExpr, usize>), PlanError> {
1557 let mut subquery_map = BTreeMap::new();
1558 let output = inner.let_in(id_gen, |id_gen, get_inner| {
1559 let mut subqueries = Vec::new();
1560 let distinct_inner = get_inner.clone().distinct();
1561 for expr in exprs.iter() {
1562 expr.visit_pre_post(
1563 &mut |e| match e {
1564 // For simplicity, subqueries within a conditional statement will be
1565 // lowered when lowering the conditional expression.
1566 HirScalarExpr::If { .. } => Some(vec![]),
1567 _ => None,
1568 },
1569 &mut |e| match e {
1570 HirScalarExpr::Select(expr, _name) => {
1571 let apply_requires_distinct_outer = false;
1572 let subquery = apply_scalar_subquery(
1573 id_gen,
1574 distinct_inner.clone(),
1575 col_map,
1576 cte_map,
1577 (**expr).clone(),
1578 apply_requires_distinct_outer,
1579 context,
1580 )
1581 .unwrap();
1582
1583 subqueries.push((e.clone(), subquery));
1584 }
1585 HirScalarExpr::Exists(expr, _name) => {
1586 let apply_requires_distinct_outer = false;
1587 let subquery = apply_existential_subquery(
1588 id_gen,
1589 distinct_inner.clone(),
1590 col_map,
1591 cte_map,
1592 (**expr).clone(),
1593 apply_requires_distinct_outer,
1594 context,
1595 )
1596 .unwrap();
1597 subqueries.push((e.clone(), subquery));
1598 }
1599 _ => {}
1600 },
1601 )?;
1602 }
1603
1604 if subqueries.is_empty() {
1605 Ok::<MirRelationExpr, PlanError>(get_inner)
1606 } else {
1607 let inner_arity = get_inner.arity();
1608 let mut total_arity = inner_arity;
1609 let mut join_inputs = vec![get_inner];
1610 let mut join_input_arities = vec![inner_arity];
1611 for (expr, subquery) in subqueries.into_iter() {
1612 // Avoid lowering duplicated subqueries
1613 if !subquery_map.contains_key(&expr) {
1614 let subquery_arity = subquery.arity();
1615 assert_eq!(subquery_arity, inner_arity + 1);
1616 join_inputs.push(subquery);
1617 join_input_arities.push(subquery_arity);
1618 total_arity += subquery_arity;
1619
1620 // Column with the value of the subquery
1621 subquery_map.insert(expr, total_arity - 1);
1622 }
1623 }
1624 // Each subquery projects all the columns of the outer context (distinct_inner)
1625 // plus 1 column, containing the result of the subquery. Those columns must be
1626 // joined with the outer/main relation (get_inner).
1627 let input_mapper =
1628 mz_expr::JoinInputMapper::new_from_input_arities(join_input_arities);
1629 let equivalences = (0..inner_arity)
1630 .map(|col| {
1631 join_inputs
1632 .iter()
1633 .enumerate()
1634 .map(|(input, _)| {
1635 MirScalarExpr::column(input_mapper.map_column_to_global(col, input))
1636 })
1637 .collect_vec()
1638 })
1639 .collect_vec();
1640 Ok(MirRelationExpr::join_scalars(join_inputs, equivalences))
1641 }
1642 })?;
1643 Ok((output, subquery_map))
1644 }
1645
1646 /// Rewrites `self` into a `mz_expr::ScalarExpr`.
1647 ///
1648 /// Returns an _internal_ error if the expression contains
1649 /// - a subquery
1650 /// - a column reference to an outer level
1651 /// - a parameter
1652 /// - a window function call
1653 ///
1654 /// Should succeed if [`HirScalarExpr::is_constant`] would return true on `self`.
1655 pub fn lower_uncorrelated(self) -> Result<MirScalarExpr, PlanError> {
1656 use MirScalarExpr as SS;
1657
1658 use HirScalarExpr::*;
1659
1660 Ok(match self {
1661 Column(ColumnRef { level: 0, column }, name) => SS::Column(column, name),
1662 Literal(datum, typ, _name) => SS::Literal(Ok(datum), typ),
1663 CallUnmaterializable(func, _name) => SS::CallUnmaterializable(func),
1664 CallUnary {
1665 func,
1666 expr,
1667 name: _,
1668 } => SS::CallUnary {
1669 func,
1670 expr: Box::new(expr.lower_uncorrelated()?),
1671 },
1672 CallBinary {
1673 func,
1674 expr1,
1675 expr2,
1676 name: _,
1677 } => SS::CallBinary {
1678 func,
1679 expr1: Box::new(expr1.lower_uncorrelated()?),
1680 expr2: Box::new(expr2.lower_uncorrelated()?),
1681 },
1682 CallVariadic {
1683 func,
1684 exprs,
1685 name: _,
1686 } => SS::CallVariadic {
1687 func,
1688 exprs: exprs
1689 .into_iter()
1690 .map(|expr| expr.lower_uncorrelated())
1691 .collect::<Result<_, _>>()?,
1692 },
1693 If {
1694 cond,
1695 then,
1696 els,
1697 name: _,
1698 } => SS::If {
1699 cond: Box::new(cond.lower_uncorrelated()?),
1700 then: Box::new(then.lower_uncorrelated()?),
1701 els: Box::new(els.lower_uncorrelated()?),
1702 },
1703 Select { .. } | Exists { .. } | Parameter(..) | Column(..) | Windowing(..) => {
1704 sql_bail!(
1705 "Internal error: unexpected HirScalarExpr in lower_uncorrelated: {:?}",
1706 self
1707 );
1708 }
1709 })
1710 }
1711}
1712
1713/// Prepare to apply `inner` to `outer`. Note that `inner` is a correlated (SQL)
1714/// expression, while `outer` is a non-correlated (dataflow) expression. `inner`
1715/// will, in effect, be executed once for every distinct row in `outer`, and the
1716/// results will be joined with `outer`. Note that columns in `outer` that are
1717/// not depended upon by `inner` are thrown away before the distinct, so that we
1718/// don't perform needless computation of `inner`.
1719///
1720/// `branch` will inspect the contents of `inner` to determine whether `inner`
1721/// is not multiplicity sensitive (roughly, contains only maps, filters,
1722/// projections, and calls to table functions). If it is not multiplicity
1723/// sensitive, `branch` will *not* distinctify outer. If this is problematic,
1724/// e.g. because the `apply` callback itself introduces multiplicity-sensitive
1725/// operations that were not present in `inner`, then set
1726/// `apply_requires_distinct_outer` to ensure that `branch` chooses the plan
1727/// that distinctifies `outer`.
1728///
1729/// The caller must supply the `apply` function that applies the rewritten
1730/// `inner` to `outer`.
1731fn branch<F>(
1732 id_gen: &mut mz_ore::id_gen::IdGen,
1733 outer: MirRelationExpr,
1734 col_map: &ColumnMap,
1735 cte_map: &mut CteMap,
1736 inner: HirRelationExpr,
1737 apply_requires_distinct_outer: bool,
1738 context: &Context,
1739 apply: F,
1740) -> Result<MirRelationExpr, PlanError>
1741where
1742 F: FnOnce(
1743 &mut mz_ore::id_gen::IdGen,
1744 HirRelationExpr,
1745 MirRelationExpr,
1746 &ColumnMap,
1747 &mut CteMap,
1748 &Context,
1749 ) -> Result<MirRelationExpr, PlanError>,
1750{
1751 // TODO: It would be nice to have a version of this code w/o optimizations,
1752 // at the least for purposes of understanding. It was difficult for one reader
1753 // to understand the required properties of `outer` and `col_map`.
1754
1755 // If the inner expression is sufficiently simple, it is safe to apply it
1756 // *directly* to outer, rather than applying it to the distinctified key
1757 // (see below).
1758 //
1759 // As an example, consider the following two queries:
1760 //
1761 // CREATE TABLE t (a int, b int);
1762 // SELECT a, series FROM t, generate_series(1, t.b) series;
1763 //
1764 // The "simple" path for the `SELECT` yields
1765 //
1766 // %0 =
1767 // | Get t
1768 // | FlatMap generate_series(1, #1)
1769 //
1770 // while the non-simple path yields:
1771 //
1772 // %0 =
1773 // | Get t
1774 //
1775 // %1 =
1776 // | Get t
1777 // | Distinct group=(#1)
1778 // | FlatMap generate_series(1, #0)
1779 //
1780 // %2 =
1781 // | LeftJoin %1 %2 (= #1 #2)
1782 //
1783 // There is a tradeoff here: the simple plan is stateless, but the non-
1784 // simple plan may do (much) less computation if there are only a few
1785 // distinct values of `t.b`.
1786 //
1787 // We apply a very simple heuristic here and take the simple path if `inner`
1788 // contains only maps, filters, projections, and calls to table functions.
1789 // The intuition is that straightforward usage of table functions should
1790 // take the simple path, while everything else should not. (In theory we
1791 // think this transformation is valid as long as `inner` does not contain a
1792 // Reduce, Distinct, or TopK node, but it is not always an optimization in
1793 // the general case.)
1794 //
1795 // TODO(benesch): this should all be handled by a proper optimizer, but
1796 // detecting the moment of decorrelation in the optimizer right now is too
1797 // hard.
1798 let mut is_simple = true;
1799 #[allow(deprecated)]
1800 inner.visit(0, &mut |expr, _| match expr {
1801 HirRelationExpr::Constant { .. }
1802 | HirRelationExpr::Project { .. }
1803 | HirRelationExpr::Map { .. }
1804 | HirRelationExpr::Filter { .. }
1805 | HirRelationExpr::CallTable { .. } => (),
1806 _ => is_simple = false,
1807 });
1808 if is_simple && !apply_requires_distinct_outer {
1809 let new_col_map = col_map.enter_scope(outer.arity() - col_map.len());
1810 return outer.let_in(id_gen, |id_gen, get_outer| {
1811 apply(id_gen, inner, get_outer, &new_col_map, cte_map, context)
1812 });
1813 }
1814
1815 // The key consists of the columns from the outer expression upon which the
1816 // inner relation depends. We discover these dependencies by walking the
1817 // inner relation expression and looking for column references whose level
1818 // escapes inner.
1819 //
1820 // At the end of this process, `key` contains the decorrelated position of
1821 // each outer column, according to the passed-in `col_map`, and
1822 // `new_col_map` maps each outer column to its new ordinal position in key.
1823 let mut outer_cols = BTreeSet::new();
1824 #[allow(deprecated)]
1825 inner.visit_columns(0, &mut |depth, col| {
1826 // Test if the column reference escapes the subquery.
1827 if col.level > depth {
1828 outer_cols.insert(ColumnRef {
1829 level: col.level - depth,
1830 column: col.column,
1831 });
1832 }
1833 });
1834 // Collect all the outer columns referenced by any CTE referenced by
1835 // the inner relation.
1836 #[allow(deprecated)]
1837 inner.visit(0, &mut |e, _| match e {
1838 HirRelationExpr::Get {
1839 id: mz_expr::Id::Local(id),
1840 ..
1841 } => {
1842 if let Some(cte_desc) = cte_map.get(id) {
1843 let cte_outer_arity = cte_desc.outer_relation.arity();
1844 outer_cols.extend(
1845 col_map
1846 .inner
1847 .iter()
1848 .filter(|(_, position)| **position < cte_outer_arity)
1849 .map(|(c, _)| {
1850 // `col_map` maps column references to column positions in
1851 // `outer`'s projection.
1852 // `outer_cols` is meant to contain the external column
1853 // references in `inner`.
1854 // Since `inner` defines a new scope, any column reference
1855 // in `col_map` is one level deeper when seen from within
1856 // `inner`, hence the +1.
1857 ColumnRef {
1858 level: c.level + 1,
1859 column: c.column,
1860 }
1861 }),
1862 );
1863 }
1864 }
1865 HirRelationExpr::Let { id, .. } => {
1866 // Note: if ID uniqueness is not guaranteed, we can't use `visit` since
1867 // we would need to remove the old CTE with the same ID temporarily while
1868 // traversing the definition of the new CTE under the same ID.
1869 assert!(!cte_map.contains_key(id));
1870 }
1871 _ => {}
1872 });
1873 let mut new_col_map = BTreeMap::new();
1874 let mut key = vec![];
1875 for col in outer_cols {
1876 new_col_map.insert(col, key.len());
1877 key.push(col_map.get(&ColumnRef {
1878 // Note: `outer_cols` contains the external column references within `inner`.
1879 // We must compensate for `inner`'s scope when translating column references
1880 // as seen within `inner` to column references as seen from `outer`'s context,
1881 // hence the -1.
1882 level: col.level - 1,
1883 column: col.column,
1884 }));
1885 }
1886 let new_col_map = ColumnMap::new(new_col_map);
1887 outer.let_in(id_gen, |id_gen, get_outer| {
1888 let keyed_outer = if key.is_empty() {
1889 // Don't depend on outer at all if the branch is not correlated,
1890 // which yields vastly better query plans. Note that this is a bit
1891 // weird in that the branch will be computed even if outer has no
1892 // rows, whereas if it had been correlated it would not (and *could*
1893 // not) have been computed if outer had no rows, but the callers of
1894 // this function don't mind these somewhat-weird semantics.
1895 MirRelationExpr::constant(vec![vec![]], RelationType::new(vec![]))
1896 } else {
1897 get_outer.clone().distinct_by(key.clone())
1898 };
1899 keyed_outer.let_in(id_gen, |id_gen, get_keyed_outer| {
1900 let oa = get_outer.arity();
1901 let branch = apply(
1902 id_gen,
1903 inner,
1904 get_keyed_outer,
1905 &new_col_map,
1906 cte_map,
1907 context,
1908 )?;
1909 let ba = branch.arity();
1910 let joined = MirRelationExpr::join(
1911 vec![get_outer.clone(), branch],
1912 key.iter()
1913 .enumerate()
1914 .map(|(i, &k)| vec![(0, k), (1, i)])
1915 .collect(),
1916 )
1917 // throw away the right-hand copy of the key we just joined on
1918 .project((0..oa).chain((oa + key.len())..(oa + ba)).collect());
1919 Ok(joined)
1920 })
1921 })
1922}
1923
1924fn apply_scalar_subquery(
1925 id_gen: &mut mz_ore::id_gen::IdGen,
1926 outer: MirRelationExpr,
1927 col_map: &ColumnMap,
1928 cte_map: &mut CteMap,
1929 scalar_subquery: HirRelationExpr,
1930 apply_requires_distinct_outer: bool,
1931 context: &Context,
1932) -> Result<MirRelationExpr, PlanError> {
1933 branch(
1934 id_gen,
1935 outer,
1936 col_map,
1937 cte_map,
1938 scalar_subquery,
1939 apply_requires_distinct_outer,
1940 context,
1941 |id_gen, expr, get_inner, col_map, cte_map, context| {
1942 // compute for every row in get_inner
1943 let select = expr.applied_to(id_gen, get_inner.clone(), col_map, cte_map, context)?;
1944 let col_type = select.typ().column_types.into_last();
1945
1946 let inner_arity = get_inner.arity();
1947 // We must determine a count for each `get_inner` prefix,
1948 // and report an error if that count exceeds one.
1949 let guarded = select.let_in(id_gen, |_id_gen, get_select| {
1950 // Count for each `get_inner` prefix.
1951 let counts = get_select.clone().reduce(
1952 (0..inner_arity).collect::<Vec<_>>(),
1953 vec![mz_expr::AggregateExpr {
1954 func: mz_expr::AggregateFunc::Count,
1955 expr: MirScalarExpr::literal_true(),
1956 distinct: false,
1957 }],
1958 None,
1959 );
1960
1961 let use_guard = context.config.enable_guard_subquery_tablefunc;
1962
1963 // Errors should result from counts > 1.
1964 let errors = if use_guard {
1965 counts
1966 .flat_map(
1967 mz_expr::TableFunc::GuardSubquerySize {
1968 column_type: col_type.clone().scalar_type,
1969 },
1970 vec![MirScalarExpr::column(inner_arity)],
1971 )
1972 .project(
1973 (0..inner_arity)
1974 .chain(Some(inner_arity + 1))
1975 .collect::<Vec<_>>(),
1976 )
1977 } else {
1978 counts
1979 .filter(vec![MirScalarExpr::column(inner_arity).call_binary(
1980 MirScalarExpr::literal_ok(Datum::Int64(1), ScalarType::Int64),
1981 mz_expr::BinaryFunc::Gt,
1982 )])
1983 .project((0..inner_arity).collect::<Vec<_>>())
1984 .map_one(MirScalarExpr::literal(
1985 Err(mz_expr::EvalError::MultipleRowsFromSubquery),
1986 col_type.clone().scalar_type,
1987 ))
1988 };
1989 // Return `get_select` and any errors added in.
1990 Ok::<_, PlanError>(get_select.union(errors))
1991 })?;
1992 // append Null to anything that didn't return any rows
1993 let default = vec![(Datum::Null, col_type.scalar_type)];
1994 get_inner.lookup(id_gen, guarded, default)
1995 },
1996 )
1997}
1998
1999fn apply_existential_subquery(
2000 id_gen: &mut mz_ore::id_gen::IdGen,
2001 outer: MirRelationExpr,
2002 col_map: &ColumnMap,
2003 cte_map: &mut CteMap,
2004 subquery_expr: HirRelationExpr,
2005 apply_requires_distinct_outer: bool,
2006 context: &Context,
2007) -> Result<MirRelationExpr, PlanError> {
2008 branch(
2009 id_gen,
2010 outer,
2011 col_map,
2012 cte_map,
2013 subquery_expr,
2014 apply_requires_distinct_outer,
2015 context,
2016 |id_gen, expr, get_inner, col_map, cte_map, context| {
2017 let exists = expr
2018 // compute for every row in get_inner
2019 .applied_to(id_gen, get_inner.clone(), col_map, cte_map, context)?
2020 // throw away actual values and just remember whether or not there were __any__ rows
2021 .distinct_by((0..get_inner.arity()).collect())
2022 // Append true to anything that returned any rows.
2023 .map(vec![MirScalarExpr::literal_true()]);
2024
2025 // append False to anything that didn't return any rows
2026 get_inner.lookup(id_gen, exists, vec![(Datum::False, ScalarType::Bool)])
2027 },
2028 )
2029}
2030
2031impl AggregateExpr {
2032 fn applied_to(
2033 self,
2034 id_gen: &mut mz_ore::id_gen::IdGen,
2035 col_map: &ColumnMap,
2036 cte_map: &mut CteMap,
2037 inner: &mut MirRelationExpr,
2038 context: &Context,
2039 ) -> Result<mz_expr::AggregateExpr, PlanError> {
2040 let AggregateExpr {
2041 func,
2042 expr,
2043 distinct,
2044 } = self;
2045
2046 Ok(mz_expr::AggregateExpr {
2047 func: func.into_expr(),
2048 expr: expr.applied_to(id_gen, col_map, cte_map, inner, &None, context)?,
2049 distinct,
2050 })
2051 }
2052}
2053
2054/// Attempts an efficient outer join, if `on` has equijoin structure.
2055///
2056/// Both `left` and `right` are decorrelated inputs.
2057///
2058/// The first `oa` columns correspond to an outer context: we should do the
2059/// outer join independently for each prefix. In the case that `on` contains
2060/// just some equality tests between columns of `left` and `right` and some
2061/// local predicates, we can employ a relatively simple plan.
2062///
2063/// The last `on_subquery_types.len()` columns correspond to results from
2064/// subqueries defined in the `on` clause - we treat those as theta-join
2065/// conditions that prohibit the use of the simple plan attempted here.
2066fn attempt_outer_equijoin(
2067 left: MirRelationExpr,
2068 right: MirRelationExpr,
2069 on: MirScalarExpr,
2070 on_subquery_types: Vec<ColumnType>,
2071 kind: JoinKind,
2072 oa: usize,
2073 id_gen: &mut mz_ore::id_gen::IdGen,
2074 context: &Context,
2075) -> Result<Option<MirRelationExpr>, PlanError> {
2076 // TODO(database-issues#6827): In theory, we can be smarter and also handle `on`
2077 // predicates that reference subqueries as long as these subqueries don't
2078 // reference `left` and `right` at the same time.
2079 //
2080 // TODO(database-issues#6828): This code can be improved as follows:
2081 //
2082 // 1. Move the `canonicalize_predicates(...)` call to `applied_to`.
2083 // 2. Use the canonicalized `on` predicate in the non-equijoin based
2084 // lowering strategy.
2085 // 3. Move the `OnPredicates::new(...)` call to `applied_to`.
2086 // 4. Pass the classified `OnPredicates` as a parameter.
2087 // 5. Guard calls of this function with `on_predicates.is_equijoin()`.
2088 //
2089 // Steps (1 + 2) require further investigation because we might change the
2090 // error semantics in case the `on` predicate contains a literal error..
2091
2092 let l_type = left.typ();
2093 let r_type = right.typ();
2094 let la = l_type.column_types.len() - oa;
2095 let ra = r_type.column_types.len() - oa;
2096 let sa = on_subquery_types.len();
2097
2098 // The output type contains [outer, left, right, sa] attributes.
2099 let mut output_type = Vec::with_capacity(oa + la + ra + sa);
2100 output_type.extend(l_type.column_types);
2101 output_type.extend(r_type.column_types.into_iter().skip(oa));
2102 output_type.extend(on_subquery_types);
2103
2104 // Generally healthy to do, but specifically `USING` conditions sometimes
2105 // put an `AND true` at the end of the `ON` condition.
2106 //
2107 // TODO(aalexandrov): maybe we should already be doing this in `applied_to`.
2108 // However, in that case it's not clear that we won't see regressions if
2109 // `on` simplifies to a literal error.
2110 let mut on = vec![on];
2111 mz_expr::canonicalize::canonicalize_predicates(&mut on, &output_type);
2112
2113 // Form the left and right types without the outer attributes.
2114 output_type.drain(0..oa);
2115 let lt = output_type.drain(0..la).collect_vec();
2116 let rt = output_type.drain(0..ra).collect_vec();
2117 assert!(output_type.len() == sa);
2118
2119 let on_predicates = OnPredicates::new(oa, la, ra, sa, on.clone(), context);
2120 if !on_predicates.is_equijoin(context) {
2121 return Ok(None);
2122 }
2123
2124 // If we've gotten this far, we can do the clever thing.
2125 // We'll want to use left and right multiple times
2126 let result = left.let_in(id_gen, |id_gen, get_left| {
2127 right.let_in(id_gen, |id_gen, get_right| {
2128 // TODO: we know that we can re-use the arrangements of left and right
2129 // needed for the inner join with each of the conditional outer joins.
2130 // It is not clear whether we should hint that, or just let the planner
2131 // and optimizer run and see what happens.
2132
2133 // We'll want the inner join (minus repeated columns)
2134 let join = MirRelationExpr::join(
2135 vec![get_left.clone(), get_right.clone()],
2136 (0..oa).map(|i| vec![(0, i), (1, i)]).collect(),
2137 )
2138 // remove those columns from `right` repeating the first `oa` columns.
2139 .project(
2140 (0..(oa + la))
2141 .chain((oa + la + oa)..(oa + la + oa + ra))
2142 .collect(),
2143 )
2144 // apply the filter constraints here, to ensure nulls are not matched.
2145 .filter(on);
2146
2147 // We'll want to re-use the results of the join multiple times.
2148 join.let_in(id_gen, |id_gen, get_join| {
2149 let mut result = get_join.clone();
2150
2151 // A collection of keys present in both left and right collections.
2152 let join_keys = on_predicates.join_keys();
2153 let both_keys_arity = join_keys.len();
2154 let both_keys = get_join.restrict(join_keys).distinct();
2155
2156 // The plan is now to determine the left and right rows matched in the
2157 // inner join, subtract them from left and right respectively, pad what
2158 // remains with nulls, and fold them in to `result`.
2159
2160 both_keys.let_in(id_gen, |_id_gen, get_both| {
2161 if let JoinKind::LeftOuter { .. } | JoinKind::FullOuter = kind {
2162 // Rows in `left` matched in the inner equijoin. This is
2163 // a semi-join between `left` and `both_keys`.
2164 let left_present = MirRelationExpr::join_scalars(
2165 vec![
2166 get_left
2167 .clone()
2168 // Push local predicates.
2169 .filter(on_predicates.lhs()),
2170 get_both.clone(),
2171 ],
2172 itertools::zip_eq(
2173 on_predicates.eq_lhs(),
2174 (0..both_keys_arity).map(|k| MirScalarExpr::column(oa + la + k)),
2175 )
2176 .map(|(l_key, b_key)| [l_key, b_key].to_vec())
2177 .collect(),
2178 )
2179 .project((0..(oa + la)).collect());
2180
2181 // Determine the types of nulls to use as filler.
2182 let right_fill = rt
2183 .into_iter()
2184 .map(|typ| MirScalarExpr::literal_null(typ.scalar_type))
2185 .collect();
2186
2187 // Add to `result` absent elements, filled with typed nulls.
2188 result = left_present
2189 .negate()
2190 .union(get_left.clone())
2191 .map(right_fill)
2192 .union(result);
2193 }
2194
2195 if let JoinKind::RightOuter | JoinKind::FullOuter = kind {
2196 // Rows in `right` matched in the inner equijoin. This
2197 // is a semi-join between `right` and `both_keys`.
2198 let right_present = MirRelationExpr::join_scalars(
2199 vec![
2200 get_right
2201 .clone()
2202 // Push local predicates.
2203 .filter(on_predicates.rhs()),
2204 get_both,
2205 ],
2206 itertools::zip_eq(
2207 on_predicates.eq_rhs(),
2208 (0..both_keys_arity).map(|k| MirScalarExpr::column(oa + ra + k)),
2209 )
2210 .map(|(r_key, b_key)| [r_key, b_key].to_vec())
2211 .collect(),
2212 )
2213 .project((0..(oa + ra)).collect());
2214
2215 // Determine the types of nulls to use as filler.
2216 let left_fill = lt
2217 .into_iter()
2218 .map(|typ| MirScalarExpr::literal_null(typ.scalar_type))
2219 .collect();
2220
2221 // Add to `result` absent elements, prepended with typed nulls.
2222 result = right_present
2223 .negate()
2224 .union(get_right.clone())
2225 .map(left_fill)
2226 // Permute left fill before right values.
2227 .project(
2228 itertools::chain!(
2229 0..oa, // Preserve `outer`.
2230 oa + ra..oa + la + ra, // Increment the next `la` cols by `ra`.
2231 oa..oa + ra // Decrement the next `ra` cols by `la`.
2232 )
2233 .collect(),
2234 )
2235 .union(result)
2236 }
2237
2238 Ok::<_, PlanError>(result)
2239 })
2240 })
2241 })
2242 })?;
2243 Ok(Some(result))
2244}
2245
2246/// A struct that represents the predicates in the `on` clause in a form
2247/// suitable for efficient planning outer joins with equijoin predicates.
2248struct OnPredicates {
2249 /// A store for classified `ON` predicates.
2250 ///
2251 /// Predicates that reference a single side are adjusted to assume an
2252 /// `outer × <side>` schema.
2253 predicates: Vec<OnPredicate>,
2254 /// Number of outer context columns.
2255 oa: usize,
2256}
2257
2258impl OnPredicates {
2259 const I_OUT: usize = 0; // outer context input position
2260 const I_LHS: usize = 1; // lhs input position
2261 const I_RHS: usize = 2; // rhs input position
2262 const I_SUB: usize = 3; // on subqueries input position
2263
2264 /// Classify the predicates in the `on` clause of an outer join.
2265 ///
2266 /// The other parameters are arities of the input parts:
2267 ///
2268 /// - `oa` is the arity of the `outer` context.
2269 /// - `la` is the arity of the `left` input.
2270 /// - `ra` is the arity of the `right` input.
2271 /// - `sa` is the arity of the `on` subqueries.
2272 ///
2273 /// The constructor assumes that:
2274 ///
2275 /// 1. The `on` parameter will be applied on a result that has the following
2276 /// schema `outer × left × right × on_subqueries`.
2277 /// 2. The `on` parameter is already adjusted to assume that schema.
2278 /// 3. The `on` parameter is obtained by canonicalizing the original `on:
2279 /// MirScalarExpr` with `canonicalize_predicates`.
2280 fn new(
2281 oa: usize,
2282 la: usize,
2283 ra: usize,
2284 sa: usize,
2285 on: Vec<MirScalarExpr>,
2286 _context: &Context,
2287 ) -> Self {
2288 use mz_expr::BinaryFunc::Eq;
2289
2290 // Re-bind those locally for more compact pattern matching.
2291 const I_LHS: usize = OnPredicates::I_LHS;
2292 const I_RHS: usize = OnPredicates::I_RHS;
2293
2294 // Self parameters.
2295 let mut predicates = Vec::with_capacity(on.len());
2296
2297 // Helpers for populating `predicates`.
2298 let inner_join_mapper = mz_expr::JoinInputMapper::new_from_input_arities([oa, la, ra, sa]);
2299 let rhs_permutation = itertools::chain!(0..oa + la, oa..oa + ra).collect::<Vec<_>>();
2300 let lookup_inputs = |expr: &MirScalarExpr| -> Vec<usize> {
2301 inner_join_mapper
2302 .lookup_inputs(expr)
2303 .filter(|&i| i != Self::I_OUT)
2304 .collect()
2305 };
2306 let has_subquery_refs = |expr: &MirScalarExpr| -> bool {
2307 inner_join_mapper
2308 .lookup_inputs(expr)
2309 .any(|i| i == Self::I_SUB)
2310 };
2311
2312 // Iterate over `on` elements and populate `predicates`.
2313 for mut predicate in on {
2314 if predicate.might_error() {
2315 tracing::debug!(case = "thetajoin (error)", "OnPredicates::new");
2316 // Treat predicates that can produce a literal error as Theta.
2317 predicates.push(OnPredicate::Theta(predicate));
2318 } else if has_subquery_refs(&predicate) {
2319 tracing::debug!(case = "thetajoin (subquery)", "OnPredicates::new");
2320 // Treat predicates referencing an `on` subquery as Theta.
2321 predicates.push(OnPredicate::Theta(predicate));
2322 } else if let MirScalarExpr::CallBinary {
2323 func: Eq,
2324 expr1,
2325 expr2,
2326 } = &mut predicate
2327 {
2328 // Obtain the non-outer inputs referenced by each side.
2329 let inputs1 = lookup_inputs(expr1);
2330 let inputs2 = lookup_inputs(expr2);
2331
2332 match (&inputs1[..], &inputs2[..]) {
2333 // Neither side references an input. This could be a
2334 // constant expression or an expression that depends only on
2335 // the outer context.
2336 ([], []) => {
2337 predicates.push(OnPredicate::Const(predicate));
2338 }
2339 // Both sides reference different inputs.
2340 ([I_LHS], [I_RHS]) => {
2341 let lhs = expr1.take();
2342 let mut rhs = expr2.take();
2343 rhs.permute(&rhs_permutation);
2344 predicates.push(OnPredicate::Eq(lhs.clone(), rhs.clone()));
2345 predicates.push(OnPredicate::LhsConsequence(lhs.call_is_null().not()));
2346 predicates.push(OnPredicate::RhsConsequence(rhs.call_is_null().not()));
2347 }
2348 // Both sides reference different inputs (swapped).
2349 ([I_RHS], [I_LHS]) => {
2350 let lhs = expr2.take();
2351 let mut rhs = expr1.take();
2352 rhs.permute(&rhs_permutation);
2353 predicates.push(OnPredicate::Eq(lhs.clone(), rhs.clone()));
2354 predicates.push(OnPredicate::LhsConsequence(lhs.call_is_null().not()));
2355 predicates.push(OnPredicate::RhsConsequence(rhs.call_is_null().not()));
2356 }
2357 // Both sides reference the left input or no input.
2358 ([I_LHS], [I_LHS]) | ([I_LHS], []) | ([], [I_LHS]) => {
2359 predicates.push(OnPredicate::Lhs(predicate));
2360 }
2361 // Both sides reference the right input or no input.
2362 ([I_RHS], [I_RHS]) | ([I_RHS], []) | ([], [I_RHS]) => {
2363 predicate.permute(&rhs_permutation);
2364 predicates.push(OnPredicate::Rhs(predicate));
2365 }
2366 // At least one side references more than one input.
2367 _ => {
2368 tracing::debug!(case = "thetajoin (eq)", "OnPredicates::new");
2369 predicates.push(OnPredicate::Theta(predicate));
2370 }
2371 }
2372 } else {
2373 // Obtain the non-outer inputs referenced by this predicate.
2374 let inputs = lookup_inputs(&predicate);
2375
2376 match &inputs[..] {
2377 // The predicate references no inputs. This could be a
2378 // constant expression or an expression that depends only on
2379 // the outer context.
2380 [] => {
2381 predicates.push(OnPredicate::Const(predicate));
2382 }
2383 // The predicate references only the left input.
2384 [I_LHS] => {
2385 predicates.push(OnPredicate::Lhs(predicate));
2386 }
2387 // The predicate references only the right input.
2388 [I_RHS] => {
2389 predicate.permute(&rhs_permutation);
2390 predicates.push(OnPredicate::Rhs(predicate));
2391 }
2392 // The predicate references both inputs.
2393 _ => {
2394 tracing::debug!(case = "thetajoin (non-eq)", "OnPredicates::new");
2395 predicates.push(OnPredicate::Theta(predicate));
2396 }
2397 }
2398 }
2399 }
2400
2401 Self { predicates, oa }
2402 }
2403
2404 /// Check if the predicates can be lowered with an equijoin-based strategy.
2405 fn is_equijoin(&self, context: &Context) -> bool {
2406 // Count each `OnPredicate` variant in `self.predicates`.
2407 let (const_cnt, lhs_cnt, rhs_cnt, eq_cnt, eq_cols, theta_cnt) =
2408 self.predicates.iter().fold(
2409 (0, 0, 0, 0, 0, 0),
2410 |(const_cnt, lhs_cnt, rhs_cnt, eq_cnt, eq_cols, theta_cnt), p| {
2411 (
2412 const_cnt + usize::from(matches!(p, OnPredicate::Const(..))),
2413 lhs_cnt + usize::from(matches!(p, OnPredicate::Lhs(..))),
2414 rhs_cnt + usize::from(matches!(p, OnPredicate::Rhs(..))),
2415 eq_cnt + usize::from(matches!(p, OnPredicate::Eq(..))),
2416 eq_cols + usize::from(matches!(p, OnPredicate::Eq(lhs, rhs) if lhs.is_column() && rhs.is_column())),
2417 theta_cnt + usize::from(matches!(p, OnPredicate::Theta(..))),
2418 )
2419 },
2420 );
2421
2422 let is_equijion = if context.config.enable_new_outer_join_lowering {
2423 // New classifier.
2424 eq_cnt > 0 && theta_cnt == 0
2425 } else {
2426 // Old classifier.
2427 eq_cnt > 0 && eq_cnt == eq_cols && theta_cnt + const_cnt + lhs_cnt + rhs_cnt == 0
2428 };
2429
2430 // Log an entry only if this is an equijoin according to the new classifier.
2431 if eq_cnt > 0 && theta_cnt == 0 {
2432 tracing::debug!(
2433 const_cnt,
2434 lhs_cnt,
2435 rhs_cnt,
2436 eq_cnt,
2437 eq_cols,
2438 theta_cnt,
2439 "OnPredicates::is_equijoin"
2440 );
2441 }
2442
2443 is_equijion
2444 }
2445
2446 /// Return an [`MirRelationExpr`] list that represents the keys for the
2447 /// equijoin. The list will contain the outer columns as a prefix.
2448 fn join_keys(&self) -> JoinKeys {
2449 // We could return either the `lhs` or the `rhs` of the keys used to
2450 // form the inner join as they are equated by the join condition.
2451 let join_keys = self.eq_lhs().collect::<Vec<_>>();
2452
2453 if join_keys.iter().all(|k| k.is_column()) {
2454 tracing::debug!(case = "outputs", "OnPredicates::join_keys");
2455 JoinKeys::Outputs(join_keys.iter().flat_map(|k| k.as_column()).collect())
2456 } else {
2457 tracing::debug!(case = "scalars", "OnPredicates::join_keys");
2458 JoinKeys::Scalars(join_keys)
2459 }
2460 }
2461
2462 /// Return an iterator over the left-hand sides of all [`OnPredicate::Eq`]
2463 /// conditions in the predicates list.
2464 ///
2465 /// The iterator will start with column references to the outer columns as a
2466 /// prefix.
2467 fn eq_lhs(&self) -> impl Iterator<Item = MirScalarExpr> + '_ {
2468 itertools::chain(
2469 (0..self.oa).map(MirScalarExpr::column),
2470 self.predicates.iter().filter_map(|e| match e {
2471 OnPredicate::Eq(lhs, _) => Some(lhs.clone()),
2472 _ => None,
2473 }),
2474 )
2475 }
2476
2477 /// Return an iterator over the right-hand sides of all [`OnPredicate::Eq`]
2478 /// conditions in the predicates list.
2479 ///
2480 /// The iterator will start with column references to the outer columns as a
2481 /// prefix.
2482 fn eq_rhs(&self) -> impl Iterator<Item = MirScalarExpr> + '_ {
2483 itertools::chain(
2484 (0..self.oa).map(MirScalarExpr::column),
2485 self.predicates.iter().filter_map(|e| match e {
2486 OnPredicate::Eq(_, rhs) => Some(rhs.clone()),
2487 _ => None,
2488 }),
2489 )
2490 }
2491
2492 /// Return an iterator over the [`OnPredicate::Lhs`], [`OnPredicate::LhsConsequence`] and
2493 /// [`OnPredicate::Const`] conditions in the predicates list.
2494 fn lhs(&self) -> impl Iterator<Item = MirScalarExpr> + '_ {
2495 self.predicates.iter().filter_map(|p| match p {
2496 // We treat Const predicates local to both inputs.
2497 OnPredicate::Const(p) => Some(p.clone()),
2498 OnPredicate::Lhs(p) => Some(p.clone()),
2499 OnPredicate::LhsConsequence(p) => Some(p.clone()),
2500 _ => None,
2501 })
2502 }
2503
2504 /// Return an iterator over the [`OnPredicate::Rhs`], [`OnPredicate::RhsConsequence`] and
2505 /// [`OnPredicate::Const`] conditions in the predicates list.
2506 fn rhs(&self) -> impl Iterator<Item = MirScalarExpr> + '_ {
2507 self.predicates.iter().filter_map(|p| match p {
2508 // We treat Const predicates local to both inputs.
2509 OnPredicate::Const(p) => Some(p.clone()),
2510 OnPredicate::Rhs(p) => Some(p.clone()),
2511 OnPredicate::RhsConsequence(p) => Some(p.clone()),
2512 _ => None,
2513 })
2514 }
2515}
2516
2517enum OnPredicate {
2518 // A predicate that is either constant or references only outer columns.
2519 Const(MirScalarExpr),
2520 // A local predicate on the left-hand side of the join, i.e., it references only the left input
2521 // and possibly outer columns.
2522 //
2523 // This is one of the original predicates from the ON clause.
2524 //
2525 // One _must_ apply this predicate.
2526 Lhs(MirScalarExpr),
2527 // A local predicate on the left-hand side of the join, i.e., it references only the left input
2528 // and possibly outer columns.
2529 //
2530 // This is not one of the original predicates from the ON clause, but is just a consequence
2531 // of an original predicate in the ON clause, where the original predicate references both
2532 // inputs, but the consequence references only the left input.
2533 //
2534 // For example, the original predicate `input1.x = input2.a` has the consequence
2535 // `input1.x IS NOT NULL`. Applying such a consequence before the input is fed into the join
2536 // prevents null skew, and also makes more CSE opportunities available when the left input's key
2537 // doesn't have a NOT NULL constraint, saving us an arrangement.
2538 //
2539 // Applying the predicate is optional, because the original predicate will be applied anyway.
2540 LhsConsequence(MirScalarExpr),
2541 // A local predicate on the right-hand side of the join.
2542 //
2543 // This is one of the original predicates from the ON clause.
2544 //
2545 // One _must_ apply this predicate.
2546 Rhs(MirScalarExpr),
2547 // A consequence of an original ON predicate, see above.
2548 RhsConsequence(MirScalarExpr),
2549 // An equality predicate between the two sides.
2550 Eq(MirScalarExpr, MirScalarExpr),
2551 // a non-equality predicate between the two sides.
2552 #[allow(dead_code)]
2553 Theta(MirScalarExpr),
2554}
2555
2556/// A set of join keys referencing an input.
2557///
2558/// This is used in the [`MirRelationExpr::Join`] lowering code in order to
2559/// avoid changes (and thereby possible regressions) in plans that have equijoin
2560/// predicates consisting only of column refs.
2561///
2562/// If we were running `CanonicalizeMfp` as part of `NormalizeOps` we might be
2563/// able to get rid of this code, but as it stands `Map` simplification seems
2564/// more cumbersome than `Project` simplification, so do this just to be sure.
2565enum JoinKeys {
2566 // A predicate that is either constant or references only outer columns.
2567 Outputs(Vec<usize>),
2568 // A local predicate on the left-hand side of the join.
2569 Scalars(Vec<MirScalarExpr>),
2570}
2571
2572impl JoinKeys {
2573 fn len(&self) -> usize {
2574 match self {
2575 JoinKeys::Outputs(outputs) => outputs.len(),
2576 JoinKeys::Scalars(scalars) => scalars.len(),
2577 }
2578 }
2579}
2580
2581/// Extension methods for [`MirRelationExpr`] required in the HIR ⇒ MIR lowering
2582/// code.
2583trait LoweringExt {
2584 /// See [`MirRelationExpr::restrict`].
2585 fn restrict(self, join_keys: JoinKeys) -> Self;
2586}
2587
2588impl LoweringExt for MirRelationExpr {
2589 /// Restrict the set of columns of an input to the sequence of [`JoinKeys`].
2590 fn restrict(self, join_keys: JoinKeys) -> Self {
2591 let num_keys = join_keys.len();
2592 match join_keys {
2593 JoinKeys::Outputs(outputs) => self.project(outputs),
2594 JoinKeys::Scalars(scalars) => {
2595 let input_arity = self.arity();
2596 let outputs = (input_arity..input_arity + num_keys).collect();
2597 self.map(scalars).project(outputs)
2598 }
2599 }
2600 }
2601}