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}