mz_transform/
analysis.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//! Traits and types for reusable expression analysis
11
12pub mod equivalences;
13pub mod monotonic;
14
15use mz_expr::MirRelationExpr;
16
17pub use arity::Arity;
18pub use cardinality::Cardinality;
19pub use column_names::{ColumnName, ColumnNames};
20pub use common::{Derived, DerivedBuilder, DerivedView};
21pub use explain::annotate_plan;
22pub use non_negative::NonNegative;
23pub use subtree::SubtreeSize;
24pub use types::SqlRelationType;
25pub use unique_keys::UniqueKeys;
26
27/// An analysis that can be applied bottom-up to a `MirRelationExpr`.
28pub trait Analysis: 'static {
29    /// The type of value this analysis associates with an expression.
30    type Value: std::fmt::Debug;
31    /// Announce any dependencies this analysis has on other analyses.
32    ///
33    /// The method should invoke `builder.require::<Foo>()` for each other
34    /// analysis `Foo` this analysis depends upon.
35    fn announce_dependencies(_builder: &mut DerivedBuilder) {}
36    /// The analysis value derived for an expression, given other analysis results.
37    ///
38    /// The other analysis results include the results of this analysis for all children,
39    /// in `results`, and the results of other analyses this analysis has expressed a
40    /// dependence upon, in `depends`, for children and the expression itself.
41    /// The analysis results for `Self` can only be found in `results`, and are not
42    /// available in `depends`.
43    ///
44    /// Implementors of this method must defensively check references into `results`, as
45    /// it may be invoked on `LetRec` bindings that have not yet been populated. It is up
46    /// to the analysis what to do in that case, but conservative behavior is recommended.
47    ///
48    /// The `index` indicates the post-order index for the expression, for use in finding
49    /// the corresponding information in `results` and `depends`.
50    ///
51    /// The returned result will be associated with this expression for this analysis,
52    /// and the analyses will continue.
53    fn derive(
54        &self,
55        expr: &MirRelationExpr,
56        index: usize,
57        results: &[Self::Value],
58        depends: &Derived,
59    ) -> Self::Value;
60
61    /// When available, provide a lattice interface to allow optimistic recursion.
62    ///
63    /// Providing a non-`None` output indicates that the analysis intends re-iteration.
64    fn lattice() -> Option<Box<dyn Lattice<Self::Value>>> {
65        None
66    }
67}
68
69/// Lattice methods for repeated analysis
70pub trait Lattice<T> {
71    /// An element greater than all other elements.
72    fn top(&self) -> T;
73    /// Set `a` to the greatest lower bound of `a` and `b`, and indicate if `a` changed as a result.
74    fn meet_assign(&self, a: &mut T, b: T) -> bool;
75}
76
77/// Types common across multiple analyses
78pub mod common {
79
80    use std::any::{Any, TypeId};
81    use std::collections::BTreeMap;
82
83    use itertools::Itertools;
84    use mz_expr::LocalId;
85    use mz_expr::MirRelationExpr;
86    use mz_ore::assert_none;
87    use mz_repr::optimize::OptimizerFeatures;
88
89    use super::Analysis;
90    use super::subtree::SubtreeSize;
91
92    /// Container for analysis state and binding context.
93    #[derive(Default)]
94    #[allow(missing_debug_implementations)]
95    pub struct Derived {
96        /// A record of active analyses and their results, indexed by their type id.
97        analyses: BTreeMap<TypeId, Box<dyn AnalysisBundle>>,
98        /// Analyses ordered where each depends only on strictly prior analyses.
99        order: Vec<TypeId>,
100        /// Map from local identifier to result offset for analysis values.
101        bindings: BTreeMap<LocalId, usize>,
102    }
103
104    impl Derived {
105        /// Return the analysis results derived so far.
106        ///
107        /// # Panics
108        ///
109        /// This method panics if `A` was not installed as a required analysis.
110        pub fn results<A: Analysis>(&self) -> &[A::Value] {
111            let type_id = TypeId::of::<Bundle<A>>();
112            if let Some(bundle) = self.analyses.get(&type_id) {
113                if let Some(bundle) = bundle.as_any().downcast_ref::<Bundle<A>>() {
114                    return &bundle.results[..];
115                }
116            }
117            panic!("Analysis {:?} missing", std::any::type_name::<A>());
118        }
119        /// Bindings from local identifiers to result offsets for analysis values.
120        pub fn bindings(&self) -> &BTreeMap<LocalId, usize> {
121            &self.bindings
122        }
123        /// Result offsets for the state of a various number of children of the current expression.
124        ///
125        /// The integers are the zero-offset locations in the `SubtreeSize` analysis. The order of
126        /// the children is descending, from last child to first, because of how the information is
127        /// laid out, and the non-reversibility of the look-ups.
128        ///
129        /// It is an error to call this method with more children than expression has.
130        pub fn children_of_rev<'a>(
131            &'a self,
132            start: usize,
133            count: usize,
134        ) -> impl Iterator<Item = usize> + 'a {
135            let sizes = self.results::<SubtreeSize>();
136            let offset = 1;
137            (0..count).scan(offset, move |offset, _| {
138                let result = start - *offset;
139                *offset += sizes[result];
140                Some(result)
141            })
142        }
143
144        /// Recast the derived data as a view that can be subdivided into views over child state.
145        pub fn as_view<'a>(&'a self) -> DerivedView<'a> {
146            DerivedView {
147                derived: self,
148                lower: 0,
149                upper: self.results::<SubtreeSize>().len(),
150            }
151        }
152    }
153
154    /// The subset of a `Derived` corresponding to an expression and its children.
155    ///
156    /// Specifically, bounds an interval `[lower, upper)` that ends with the state
157    /// of an expression, at `upper-1`, and is preceded by the state of descendents.
158    ///
159    /// This is best thought of as a node in a tree rather
160    #[allow(missing_debug_implementations)]
161    #[derive(Copy, Clone)]
162    pub struct DerivedView<'a> {
163        derived: &'a Derived,
164        lower: usize,
165        upper: usize,
166    }
167
168    impl<'a> DerivedView<'a> {
169        /// The value associated with the expression.
170        pub fn value<A: Analysis>(&self) -> Option<&'a A::Value> {
171            self.results::<A>().last()
172        }
173
174        /// The post-order traversal index for the expression.
175        ///
176        /// This can be used to index into the full set of results, as provided by an
177        /// instance of `Derived`.
178        pub fn index(&self) -> usize {
179            self.upper - 1
180        }
181
182        /// The value bound to an identifier, if it has been derived.
183        ///
184        /// There are several reasons the value could not be derived, and this method
185        /// does not distinguish between them.
186        pub fn bound<A: Analysis>(&self, id: LocalId) -> Option<&'a A::Value> {
187            self.derived
188                .bindings
189                .get(&id)
190                .and_then(|index| self.derived.results::<A>().get(*index))
191        }
192
193        /// The results for expression and its children.
194        ///
195        /// The results for the expression itself will be the last element.
196        ///
197        /// # Panics
198        ///
199        /// This method panics if `A` was not installed as a required analysis.
200        pub fn results<A: Analysis>(&self) -> &'a [A::Value] {
201            &self.derived.results::<A>()[self.lower..self.upper]
202        }
203
204        /// Bindings from local identifiers to result offsets for analysis values.
205        ///
206        /// This method returns all bindings, which may include bindings not in scope for
207        /// the expression and its children; they should be ignored.
208        pub fn bindings(&self) -> &'a BTreeMap<LocalId, usize> {
209            self.derived.bindings()
210        }
211
212        /// Subviews over `self` corresponding to the children of the expression, in reverse order.
213        ///
214        /// These views should disjointly cover the same interval as `self`, except for the last element
215        /// which corresponds to the expression itself.
216        ///
217        /// The number of produced items should exactly match the number of children, which need not
218        /// be provided as an argument. This relies on the well-formedness of the view, which should
219        /// exhaust itself just as it enumerates its last (the first) child view.
220        pub fn children_rev(&self) -> impl Iterator<Item = DerivedView<'a>> + 'a {
221            // This logic is copy/paste from `Derived::children_of_rev` but it was annoying to layer
222            // it over the output of that function, and perhaps clearer to rewrite in any case.
223
224            // Discard the last element (the size of the expression's subtree).
225            // Repeatedly read out the last element, then peel off that many elements.
226            // Each extracted slice corresponds to a child of the current expression.
227            // We should end cleanly with an empty slice, otherwise there is an issue.
228            let sizes = self.results::<SubtreeSize>();
229            let sizes = &sizes[..sizes.len() - 1];
230
231            let offset = self.lower;
232            let derived = self.derived;
233            (0..).scan(sizes, move |sizes, _| {
234                if let Some(size) = sizes.last() {
235                    *sizes = &sizes[..sizes.len() - size];
236                    Some(Self {
237                        derived,
238                        lower: offset + sizes.len(),
239                        upper: offset + sizes.len() + size,
240                    })
241                } else {
242                    None
243                }
244            })
245        }
246
247        /// A convenience method for the view over the expressions last child.
248        ///
249        /// This method is appropriate to call on expressions with multiple children,
250        /// and in particular for `Let` and `LetRecv` variants where the body is the
251        /// last child.
252        ///
253        /// It is an error to call this on a view for an expression with no children.
254        pub fn last_child(&self) -> DerivedView<'a> {
255            self.children_rev().next().unwrap()
256        }
257    }
258
259    /// A builder wrapper to accumulate announced dependencies and construct default state.
260    #[allow(missing_debug_implementations)]
261    pub struct DerivedBuilder<'a> {
262        result: Derived,
263        features: &'a OptimizerFeatures,
264    }
265
266    impl<'a> DerivedBuilder<'a> {
267        /// Create a new [`DerivedBuilder`] parameterized by [`OptimizerFeatures`].
268        pub fn new(features: &'a OptimizerFeatures) -> Self {
269            // The default builder should include `SubtreeSize` to facilitate navigation.
270            let mut builder = DerivedBuilder {
271                result: Derived::default(),
272                features,
273            };
274            builder.require(SubtreeSize);
275            builder
276        }
277    }
278
279    impl<'a> DerivedBuilder<'a> {
280        /// Announces a dependence on an analysis `A`.
281        ///
282        /// This ensures that `A` will be performed, and before any analysis that
283        /// invokes this method.
284        pub fn require<A: Analysis>(&mut self, analysis: A) {
285            // The method recursively descends through required analyses, first
286            // installing each in `result.analyses` and second in `result.order`.
287            // The first is an obligation, and serves as an indication that we have
288            // found a cycle in dependencies.
289            let type_id = TypeId::of::<Bundle<A>>();
290            if !self.result.order.contains(&type_id) {
291                // If we have not sequenced `type_id` but have a bundle, it means
292                // we are in the process of fulfilling its requirements: a cycle.
293                if self.result.analyses.contains_key(&type_id) {
294                    panic!("Cyclic dependency detected: {}", std::any::type_name::<A>());
295                }
296                // Insert the analysis bundle first, so that we can detect cycles.
297                self.result.analyses.insert(
298                    type_id,
299                    Box::new(Bundle::<A> {
300                        analysis,
301                        results: Vec::new(),
302                        fuel: 100,
303                        allow_optimistic: self.features.enable_letrec_fixpoint_analysis,
304                    }),
305                );
306                A::announce_dependencies(self);
307                // All dependencies are successfully sequenced; sequence `type_id`.
308                self.result.order.push(type_id);
309            }
310        }
311        /// Complete the building: perform analyses and return the resulting `Derivation`.
312        pub fn visit(mut self, expr: &MirRelationExpr) -> Derived {
313            // A stack of expressions to process (`Ok`) and let bindings to fill (`Err`).
314            let mut todo = vec![Ok(expr)];
315            // Expressions in reverse post-order: each expression, followed by its children in reverse order.
316            // We will reverse this to get the post order, but must form it in reverse.
317            let mut rev_post_order = Vec::new();
318            while let Some(command) = todo.pop() {
319                match command {
320                    // An expression to visit.
321                    Ok(expr) => {
322                        match expr {
323                            MirRelationExpr::Let { id, value, body } => {
324                                todo.push(Ok(value));
325                                todo.push(Err(*id));
326                                todo.push(Ok(body));
327                            }
328                            MirRelationExpr::LetRec {
329                                ids, values, body, ..
330                            } => {
331                                for (id, value) in ids.iter().zip_eq(values) {
332                                    todo.push(Ok(value));
333                                    todo.push(Err(*id));
334                                }
335                                todo.push(Ok(body));
336                            }
337                            _ => {
338                                todo.extend(expr.children().map(Ok));
339                            }
340                        }
341                        rev_post_order.push(expr);
342                    }
343                    // A local id to install
344                    Err(local_id) => {
345                        // Capture the *remaining* work, which we'll need to flip around.
346                        let prior = self.result.bindings.insert(local_id, rev_post_order.len());
347                        assert_none!(prior, "Shadowing not allowed");
348                    }
349                }
350            }
351            // Flip the offsets now that we know a length.
352            for value in self.result.bindings.values_mut() {
353                *value = rev_post_order.len() - *value - 1;
354            }
355            // Visit the pre-order in reverse order: post-order.
356            rev_post_order.reverse();
357
358            // Apply each analysis to `expr` in order.
359            for id in self.result.order.iter() {
360                if let Some(mut bundle) = self.result.analyses.remove(id) {
361                    bundle.analyse(&rev_post_order[..], &self.result);
362                    self.result.analyses.insert(*id, bundle);
363                }
364            }
365
366            self.result
367        }
368    }
369
370    /// An abstraction for an analysis and associated state.
371    trait AnalysisBundle: Any {
372        /// Populates internal state for all of `exprs`.
373        ///
374        /// Result indicates whether new information was produced for `exprs.last()`.
375        fn analyse(&mut self, exprs: &[&MirRelationExpr], depends: &Derived) -> bool;
376        /// Upcasts `self` to a `&dyn Any`.
377        ///
378        /// NOTE: This is required until <https://github.com/rust-lang/rfcs/issues/2765> is fixed
379        fn as_any(&self) -> &dyn std::any::Any;
380    }
381
382    /// A wrapper for analysis state.
383    struct Bundle<A: Analysis> {
384        /// The algorithm instance used to derive the results.
385        analysis: A,
386        /// A vector of results.
387        results: Vec<A::Value>,
388        /// Counts down with each `LetRec` re-iteration, to avoid unbounded effort.
389        /// Should it reach zero, the analysis should discard its results and restart as if pessimistic.
390        fuel: usize,
391        /// Allow optimistic analysis for `A` (otherwise we always do pesimistic
392        /// analysis, even if a [`crate::analysis::Lattice`] is available for `A`).
393        allow_optimistic: bool,
394    }
395
396    impl<A: Analysis> AnalysisBundle for Bundle<A> {
397        fn analyse(&mut self, exprs: &[&MirRelationExpr], depends: &Derived) -> bool {
398            self.results.clear();
399            // Attempt optimistic analysis, and if that fails go pessimistic.
400            let update = A::lattice()
401                .filter(|_| self.allow_optimistic)
402                .and_then(|lattice| {
403                    for _ in exprs.iter() {
404                        self.results.push(lattice.top());
405                    }
406                    self.analyse_optimistic(exprs, 0, exprs.len(), depends, &*lattice)
407                        .ok()
408                })
409                .unwrap_or_else(|| {
410                    self.results.clear();
411                    self.analyse_pessimistic(exprs, depends)
412                });
413            assert_eq!(self.results.len(), exprs.len());
414            update
415        }
416        fn as_any(&self) -> &dyn std::any::Any {
417            self
418        }
419    }
420
421    impl<A: Analysis> Bundle<A> {
422        /// Analysis that starts optimistically but is only correct at a fixed point.
423        ///
424        /// Will fail out to `analyse_pessimistic` if the lattice is missing, or `self.fuel` is exhausted.
425        /// When successful, the result indicates whether new information was produced for `exprs.last()`.
426        fn analyse_optimistic(
427            &mut self,
428            exprs: &[&MirRelationExpr],
429            lower: usize,
430            upper: usize,
431            depends: &Derived,
432            lattice: &dyn crate::analysis::Lattice<A::Value>,
433        ) -> Result<bool, ()> {
434            // Repeatedly re-evaluate the whole tree bottom up, until no changes of fuel spent.
435            let mut changed = true;
436            while changed {
437                changed = false;
438
439                // Bail out if we have done too many `LetRec` passes in this analysis.
440                self.fuel -= 1;
441                if self.fuel == 0 {
442                    return Err(());
443                }
444
445                // Track if repetitions may be required, to avoid iteration if they are not.
446                let mut is_recursive = false;
447                // Update each derived value and track if any have changed.
448                for index in lower..upper {
449                    let value = self.derive(exprs[index], index, depends);
450                    changed = lattice.meet_assign(&mut self.results[index], value) || changed;
451                    if let MirRelationExpr::LetRec { .. } = &exprs[index] {
452                        is_recursive = true;
453                    }
454                }
455
456                // Un-set the potential loop if there was no recursion.
457                if !is_recursive {
458                    changed = false;
459                }
460            }
461            Ok(true)
462        }
463
464        /// Analysis that starts conservatively and can be stopped at any point.
465        ///
466        /// Result indicates whether new information was produced for `exprs.last()`.
467        fn analyse_pessimistic(&mut self, exprs: &[&MirRelationExpr], depends: &Derived) -> bool {
468            // TODO: consider making iterative, from some `bottom()` up using `join_assign()`.
469            self.results.clear();
470            for (index, expr) in exprs.iter().enumerate() {
471                self.results.push(self.derive(expr, index, depends));
472            }
473            true
474        }
475
476        #[inline]
477        fn derive(&self, expr: &MirRelationExpr, index: usize, depends: &Derived) -> A::Value {
478            self.analysis
479                .derive(expr, index, &self.results[..], depends)
480        }
481    }
482}
483
484/// Expression subtree sizes
485///
486/// This analysis counts the number of expressions in each subtree, and is most useful
487/// for navigating the results of other analyses that are offset by subtree sizes.
488pub mod subtree {
489
490    use super::{Analysis, Derived};
491    use mz_expr::MirRelationExpr;
492
493    /// Analysis that determines the size in child expressions of relation expressions.
494    #[derive(Debug)]
495    pub struct SubtreeSize;
496
497    impl Analysis for SubtreeSize {
498        type Value = usize;
499
500        fn derive(
501            &self,
502            expr: &MirRelationExpr,
503            index: usize,
504            results: &[Self::Value],
505            _depends: &Derived,
506        ) -> Self::Value {
507            match expr {
508                MirRelationExpr::Constant { .. } | MirRelationExpr::Get { .. } => 1,
509                _ => {
510                    let mut offset = 1;
511                    for _ in expr.children() {
512                        offset += results[index - offset];
513                    }
514                    offset
515                }
516            }
517        }
518    }
519}
520
521/// Expression arities
522mod arity {
523
524    use super::{Analysis, Derived};
525    use mz_expr::MirRelationExpr;
526
527    /// Analysis that determines the number of columns of relation expressions.
528    #[derive(Debug)]
529    pub struct Arity;
530
531    impl Analysis for Arity {
532        type Value = usize;
533
534        fn derive(
535            &self,
536            expr: &MirRelationExpr,
537            index: usize,
538            results: &[Self::Value],
539            depends: &Derived,
540        ) -> Self::Value {
541            let mut offsets = depends
542                .children_of_rev(index, expr.children().count())
543                .map(|child| results[child])
544                .collect::<Vec<_>>();
545            offsets.reverse();
546            expr.arity_with_input_arities(offsets.into_iter())
547        }
548    }
549}
550
551/// Expression types
552mod types {
553
554    use super::{Analysis, Derived, Lattice};
555    use itertools::Itertools;
556    use mz_expr::MirRelationExpr;
557    use mz_repr::SqlColumnType;
558
559    /// Analysis that determines the type of relation expressions.
560    ///
561    /// The value is `Some` when it discovers column types, and `None` in the case that
562    /// it has discovered no constraining information on the column types. The `None`
563    /// variant should only occur in the course of iteration, and should not be revealed
564    /// as an output of the analysis. One can `unwrap()` the result, and if it errors then
565    /// either the expression is malformed or the analysis has a bug.
566    ///
567    /// The analysis will panic if an expression is not well typed (i.e. if `try_col_with_input_cols`
568    /// returns an error).
569    #[derive(Debug)]
570    pub struct SqlRelationType;
571
572    impl Analysis for SqlRelationType {
573        type Value = Option<Vec<SqlColumnType>>;
574
575        fn derive(
576            &self,
577            expr: &MirRelationExpr,
578            index: usize,
579            results: &[Self::Value],
580            depends: &Derived,
581        ) -> Self::Value {
582            let offsets = depends
583                .children_of_rev(index, expr.children().count())
584                .map(|child| &results[child])
585                .collect::<Vec<_>>();
586
587            // For most expressions we'll apply `try_col_with_input_cols`, but for `Get` expressions
588            // we'll want to combine what we know (iteratively) with the stated `Get::typ`.
589            match expr {
590                MirRelationExpr::Get {
591                    id: mz_expr::Id::Local(i),
592                    typ,
593                    ..
594                } => {
595                    let mut result = typ.column_types.clone();
596                    if let Some(o) = depends.bindings().get(i) {
597                        if let Some(t) = results.get(*o) {
598                            if let Some(rec_typ) = t {
599                                // Reconcile nullability statements.
600                                // Unclear if we should trust `typ`.
601                                assert_eq!(result.len(), rec_typ.len());
602                                result.clone_from(rec_typ);
603                                for (res, col) in result.iter_mut().zip_eq(typ.column_types.iter())
604                                {
605                                    if !col.nullable {
606                                        res.nullable = false;
607                                    }
608                                }
609                            } else {
610                                // Our `None` information indicates that we are optimistically
611                                // assuming the best, including that all columns are non-null.
612                                // This should only happen in the first visit to a `Get` expr.
613                                // Use `typ`, but flatten nullability.
614                                for col in result.iter_mut() {
615                                    col.nullable = false;
616                                }
617                            }
618                        }
619                    }
620                    Some(result)
621                }
622                _ => {
623                    // Every expression with inputs should have non-`None` inputs at this point.
624                    let input_cols = offsets.into_iter().rev().map(|o| {
625                        o.as_ref()
626                            .expect("SqlRelationType analysis discovered type-less expression")
627                    });
628                    Some(expr.try_col_with_input_cols(input_cols).unwrap())
629                }
630            }
631        }
632
633        fn lattice() -> Option<Box<dyn Lattice<Self::Value>>> {
634            Some(Box::new(RTLattice))
635        }
636    }
637
638    struct RTLattice;
639
640    impl Lattice<Option<Vec<SqlColumnType>>> for RTLattice {
641        fn top(&self) -> Option<Vec<SqlColumnType>> {
642            None
643        }
644        fn meet_assign(
645            &self,
646            a: &mut Option<Vec<SqlColumnType>>,
647            b: Option<Vec<SqlColumnType>>,
648        ) -> bool {
649            match (a, b) {
650                (_, None) => false,
651                (Some(a), Some(b)) => {
652                    let mut changed = false;
653                    assert_eq!(a.len(), b.len());
654                    for (at, bt) in a.iter_mut().zip_eq(b.iter()) {
655                        assert_eq!(at.scalar_type, bt.scalar_type);
656                        if !at.nullable && bt.nullable {
657                            at.nullable = true;
658                            changed = true;
659                        }
660                    }
661                    changed
662                }
663                (a, b) => {
664                    *a = b;
665                    true
666                }
667            }
668        }
669    }
670}
671
672/// Expression unique keys
673mod unique_keys {
674
675    use super::arity::Arity;
676    use super::{Analysis, Derived, DerivedBuilder, Lattice};
677    use mz_expr::MirRelationExpr;
678
679    /// Analysis that determines the unique keys of relation expressions.
680    ///
681    /// The analysis value is a `Vec<Vec<usize>>`, which should be interpreted as a list
682    /// of sets of column identifiers, each set of which has the property that there is at
683    /// most one instance of each assignment of values to those columns.
684    ///
685    /// The sets are minimal, in that any superset of another set is removed from the list.
686    /// Any superset of unique key columns are also unique key columns.
687    #[derive(Debug)]
688    pub struct UniqueKeys;
689
690    impl Analysis for UniqueKeys {
691        type Value = Vec<Vec<usize>>;
692
693        fn announce_dependencies(builder: &mut DerivedBuilder) {
694            builder.require(Arity);
695        }
696
697        fn derive(
698            &self,
699            expr: &MirRelationExpr,
700            index: usize,
701            results: &[Self::Value],
702            depends: &Derived,
703        ) -> Self::Value {
704            let mut offsets = depends
705                .children_of_rev(index, expr.children().count())
706                .collect::<Vec<_>>();
707            offsets.reverse();
708
709            match expr {
710                MirRelationExpr::Get {
711                    id: mz_expr::Id::Local(i),
712                    typ,
713                    ..
714                } => {
715                    // We have information from `typ` and from the analysis.
716                    // We should "join" them, unioning and reducing the keys.
717                    let mut keys = typ.keys.clone();
718                    if let Some(o) = depends.bindings().get(i) {
719                        if let Some(ks) = results.get(*o) {
720                            for k in ks.iter() {
721                                antichain_insert(&mut keys, k.clone());
722                            }
723                            keys.extend(ks.iter().cloned());
724                            keys.sort();
725                            keys.dedup();
726                        }
727                    }
728                    keys
729                }
730                _ => {
731                    let arity = depends.results::<Arity>();
732                    expr.keys_with_input_keys(
733                        offsets.iter().map(|o| arity[*o]),
734                        offsets.iter().map(|o| &results[*o]),
735                    )
736                }
737            }
738        }
739
740        fn lattice() -> Option<Box<dyn Lattice<Self::Value>>> {
741            Some(Box::new(UKLattice))
742        }
743    }
744
745    fn antichain_insert(into: &mut Vec<Vec<usize>>, item: Vec<usize>) {
746        // Insert only if there is not a dominating element of `into`.
747        if into.iter().all(|key| !key.iter().all(|k| item.contains(k))) {
748            into.retain(|key| !key.iter().all(|k| item.contains(k)));
749            into.push(item);
750        }
751    }
752
753    /// Lattice for sets of columns that define a unique key.
754    ///
755    /// An element `Vec<Vec<usize>>` describes all sets of columns `Vec<usize>` that are a
756    /// superset of some set of columns in the lattice element.
757    struct UKLattice;
758
759    impl Lattice<Vec<Vec<usize>>> for UKLattice {
760        fn top(&self) -> Vec<Vec<usize>> {
761            vec![vec![]]
762        }
763        fn meet_assign(&self, a: &mut Vec<Vec<usize>>, b: Vec<Vec<usize>>) -> bool {
764            a.sort();
765            a.dedup();
766            let mut c = Vec::new();
767            for cols_a in a.iter_mut() {
768                cols_a.sort();
769                cols_a.dedup();
770                for cols_b in b.iter() {
771                    let mut cols_c = cols_a.iter().chain(cols_b).cloned().collect::<Vec<_>>();
772                    cols_c.sort();
773                    cols_c.dedup();
774                    antichain_insert(&mut c, cols_c);
775                }
776            }
777            c.sort();
778            c.dedup();
779            std::mem::swap(a, &mut c);
780            a != &mut c
781        }
782    }
783}
784
785/// Determines if accumulated frequences can be negative.
786///
787/// This analysis assumes that globally identified collection have the property, and it is
788/// incorrect to apply it to expressions that reference external collections that may have
789/// negative accumulations.
790mod non_negative {
791
792    use super::{Analysis, Derived, Lattice};
793    use crate::analysis::common_lattice::BoolLattice;
794    use mz_expr::{Id, MirRelationExpr};
795
796    /// Analysis that determines if all accumulations at all times are non-negative.
797    ///
798    /// The analysis assumes that `Id::Global` references only refer to non-negative collections.
799    #[derive(Debug)]
800    pub struct NonNegative;
801
802    impl Analysis for NonNegative {
803        type Value = bool;
804
805        fn derive(
806            &self,
807            expr: &MirRelationExpr,
808            index: usize,
809            results: &[Self::Value],
810            depends: &Derived,
811        ) -> Self::Value {
812            match expr {
813                MirRelationExpr::Constant { rows, .. } => rows
814                    .as_ref()
815                    .map(|r| r.iter().all(|(_, diff)| *diff >= mz_repr::Diff::ZERO))
816                    .unwrap_or(true),
817                MirRelationExpr::Get { id, .. } => match id {
818                    Id::Local(id) => {
819                        let index = *depends
820                            .bindings()
821                            .get(id)
822                            .expect("Dependency info not found");
823                        *results.get(index).unwrap_or(&false)
824                    }
825                    Id::Global(_) => true,
826                },
827                // Negate must be false unless input is "non-positive".
828                MirRelationExpr::Negate { .. } => false,
829                // Threshold ensures non-negativity.
830                MirRelationExpr::Threshold { .. } => true,
831                // Reduce errors on negative input.
832                MirRelationExpr::Reduce { .. } => true,
833                MirRelationExpr::Join { .. } => {
834                    // If all inputs are non-negative, the join is non-negative.
835                    depends
836                        .children_of_rev(index, expr.children().count())
837                        .all(|off| results[off])
838                }
839                MirRelationExpr::Union { base, inputs } => {
840                    // If all inputs are non-negative, the union is non-negative.
841                    let all_non_negative = depends
842                        .children_of_rev(index, expr.children().count())
843                        .all(|off| results[off]);
844
845                    if all_non_negative {
846                        return true;
847                    }
848
849                    // We look for the pattern `Union { base, Negate(Subset(base)) }`.
850                    // TODO: take some care to ensure that union fusion does not introduce a regression.
851                    if inputs.len() == 1 {
852                        if let MirRelationExpr::Negate { input } = &inputs[0] {
853                            // If `base` is non-negative, and `is_superset_of(base, input)`, return true.
854                            // TODO: this is not correct until we have `is_superset_of` validate non-negativity
855                            // as it goes, but it matches the current implementation.
856                            let mut children = depends.children_of_rev(index, 2);
857                            let _negate = children.next().unwrap();
858                            let base_id = children.next().unwrap();
859                            debug_assert_eq!(children.next(), None);
860                            if results[base_id] && is_superset_of(&*base, &*input) {
861                                return true;
862                            }
863                        }
864                    }
865
866                    false
867                }
868                _ => results[index - 1],
869            }
870        }
871
872        fn lattice() -> Option<Box<dyn Lattice<Self::Value>>> {
873            Some(Box::new(BoolLattice))
874        }
875    }
876
877    /// Returns true only if `rhs.negate().union(lhs)` contains only non-negative multiplicities
878    /// once consolidated.
879    ///
880    /// Informally, this happens when `rhs` is a multiset subset of `lhs`, meaning the multiplicity
881    /// of any record in `rhs` is at most the multiplicity of the same record in `lhs`.
882    ///
883    /// This method recursively descends each of `lhs` and `rhs` and performs a great many equality
884    /// tests, which has the potential to be quadratic. We should consider restricting its attention
885    /// to `Get` identifiers, under the premise that equal AST nodes would necessarily be identified
886    /// by common subexpression elimination. This requires care around recursively bound identifiers.
887    ///
888    /// These rules are .. somewhat arbitrary, and likely reflect observed opportunities. For example,
889    /// while we do relate `distinct(filter(A)) <= distinct(A)`, we do not relate `distinct(A) <= A`.
890    /// Further thoughts about the class of optimizations, and whether there should be more or fewer,
891    /// can be found here: <https://github.com/MaterializeInc/database-issues/issues/4044>.
892    fn is_superset_of(mut lhs: &MirRelationExpr, mut rhs: &MirRelationExpr) -> bool {
893        // This implementation is iterative.
894        // Before converting this implementation to recursive (e.g. to improve its accuracy)
895        // make sure to use the `CheckedRecursion` struct to avoid blowing the stack.
896        while lhs != rhs {
897            match rhs {
898                MirRelationExpr::Filter { input, .. } => rhs = &**input,
899                MirRelationExpr::TopK { input, .. } => rhs = &**input,
900                // Descend in both sides if the current roots are
901                // projections with the same `outputs` vector.
902                MirRelationExpr::Project {
903                    input: rhs_input,
904                    outputs: rhs_outputs,
905                } => match lhs {
906                    MirRelationExpr::Project {
907                        input: lhs_input,
908                        outputs: lhs_outputs,
909                    } if lhs_outputs == rhs_outputs => {
910                        rhs = &**rhs_input;
911                        lhs = &**lhs_input;
912                    }
913                    _ => return false,
914                },
915                // Descend in both sides if the current roots are reduces with empty aggregates
916                // on the same set of keys (that is, a distinct operation on those keys).
917                MirRelationExpr::Reduce {
918                    input: rhs_input,
919                    group_key: rhs_group_key,
920                    aggregates: rhs_aggregates,
921                    monotonic: _,
922                    expected_group_size: _,
923                } if rhs_aggregates.is_empty() => match lhs {
924                    MirRelationExpr::Reduce {
925                        input: lhs_input,
926                        group_key: lhs_group_key,
927                        aggregates: lhs_aggregates,
928                        monotonic: _,
929                        expected_group_size: _,
930                    } if lhs_aggregates.is_empty() && lhs_group_key == rhs_group_key => {
931                        rhs = &**rhs_input;
932                        lhs = &**lhs_input;
933                    }
934                    _ => return false,
935                },
936                _ => {
937                    // TODO: Imagine more complex reasoning here!
938                    return false;
939                }
940            }
941        }
942        true
943    }
944}
945
946mod column_names {
947    use std::ops::Range;
948    use std::sync::Arc;
949
950    use super::Analysis;
951    use mz_expr::{AggregateFunc, Id, MirRelationExpr, MirScalarExpr, TableFunc};
952    use mz_repr::explain::ExprHumanizer;
953    use mz_repr::{GlobalId, UNKNOWN_COLUMN_NAME};
954    use mz_sql::ORDINALITY_COL_NAME;
955
956    /// An abstract type denoting an inferred column name.
957    #[derive(Debug, Clone)]
958    pub enum ColumnName {
959        /// A column with name inferred to be equal to a GlobalId schema column.
960        Global(GlobalId, usize),
961        /// An anonymous expression named after the top-level function name.
962        Aggregate(AggregateFunc, Box<ColumnName>),
963        /// A column with a name that has been saved from the original SQL query.
964        Annotated(Arc<str>),
965        /// An column with an unknown name.
966        Unknown,
967    }
968
969    impl ColumnName {
970        /// Return `true` iff the variant has an inferred name.
971        pub fn is_known(&self) -> bool {
972            match self {
973                Self::Global(..) | Self::Aggregate(..) => true,
974                // We treat annotated columns as unknown because we would rather
975                // override them with inferred names, if we can.
976                Self::Annotated(..) | Self::Unknown => false,
977            }
978        }
979
980        /// Humanize the column to a [`String`], returns an empty [`String`] for
981        /// unknown columns (or columns named `UNKNOWN_COLUMN_NAME`).
982        pub fn humanize(&self, humanizer: &dyn ExprHumanizer) -> String {
983            match self {
984                Self::Global(id, c) => humanizer.humanize_column(*id, *c).unwrap_or_default(),
985                Self::Aggregate(func, expr) => {
986                    let func = func.name();
987                    let expr = expr.humanize(humanizer);
988                    if expr.is_empty() {
989                        String::from(func)
990                    } else {
991                        format!("{func}_{expr}")
992                    }
993                }
994                Self::Annotated(name) => name.to_string(),
995                Self::Unknown => String::new(),
996            }
997        }
998
999        /// Clone this column name if it is known, otherwise try to use the provided
1000        /// name if it is available.
1001        pub fn cloned_or_annotated(&self, name: &Option<Arc<str>>) -> Self {
1002            match self {
1003                Self::Global(..) | Self::Aggregate(..) | Self::Annotated(..) => self.clone(),
1004                Self::Unknown => name
1005                    .as_ref()
1006                    .filter(|name| name.as_ref() != UNKNOWN_COLUMN_NAME)
1007                    .map_or_else(|| Self::Unknown, |name| Self::Annotated(Arc::clone(name))),
1008            }
1009        }
1010    }
1011
1012    /// Compute the column types of each subtree of a [MirRelationExpr] from the
1013    /// bottom-up.
1014    #[derive(Debug)]
1015    pub struct ColumnNames;
1016
1017    impl ColumnNames {
1018        /// fallback schema consisting of ordinal column names: #0, #1, ...
1019        fn anonymous(range: Range<usize>) -> impl Iterator<Item = ColumnName> {
1020            range.map(|_| ColumnName::Unknown)
1021        }
1022
1023        /// fallback schema consisting of ordinal column names: #0, #1, ...
1024        fn extend_with_scalars(column_names: &mut Vec<ColumnName>, scalars: &Vec<MirScalarExpr>) {
1025            for scalar in scalars {
1026                column_names.push(match scalar {
1027                    MirScalarExpr::Column(c, name) => column_names[*c].cloned_or_annotated(&name.0),
1028                    _ => ColumnName::Unknown,
1029                });
1030            }
1031        }
1032    }
1033
1034    impl Analysis for ColumnNames {
1035        type Value = Vec<ColumnName>;
1036
1037        fn derive(
1038            &self,
1039            expr: &MirRelationExpr,
1040            index: usize,
1041            results: &[Self::Value],
1042            depends: &crate::analysis::Derived,
1043        ) -> Self::Value {
1044            use MirRelationExpr::*;
1045
1046            match expr {
1047                Constant { rows: _, typ } => {
1048                    // Fallback to an anonymous schema for constants.
1049                    ColumnNames::anonymous(0..typ.arity()).collect()
1050                }
1051                Get {
1052                    id: Id::Global(id),
1053                    typ,
1054                    access_strategy: _,
1055                } => {
1056                    // Emit ColumnName::Global instances for each column in the
1057                    // `Get` type. Those can be resolved to real names later when an
1058                    // ExpressionHumanizer is available.
1059                    (0..typ.columns().len())
1060                        .map(|c| ColumnName::Global(*id, c))
1061                        .collect()
1062                }
1063                Get {
1064                    id: Id::Local(id),
1065                    typ,
1066                    access_strategy: _,
1067                } => {
1068                    let index_child = *depends.bindings().get(id).expect("id in scope");
1069                    if index_child < results.len() {
1070                        results[index_child].clone()
1071                    } else {
1072                        // Possible because we infer LetRec bindings in order. This
1073                        // can be improved by introducing a fixpoint loop in the
1074                        // Env<A>::schedule_tasks LetRec handling block.
1075                        ColumnNames::anonymous(0..typ.arity()).collect()
1076                    }
1077                }
1078                Let {
1079                    id: _,
1080                    value: _,
1081                    body: _,
1082                } => {
1083                    // Return the column names of the `body`.
1084                    results[index - 1].clone()
1085                }
1086                LetRec {
1087                    ids: _,
1088                    values: _,
1089                    limits: _,
1090                    body: _,
1091                } => {
1092                    // Return the column names of the `body`.
1093                    results[index - 1].clone()
1094                }
1095                Project { input: _, outputs } => {
1096                    // Permute the column names of the input.
1097                    let input_column_names = &results[index - 1];
1098                    let mut column_names = vec![];
1099                    for col in outputs {
1100                        column_names.push(input_column_names[*col].clone());
1101                    }
1102                    column_names
1103                }
1104                Map { input: _, scalars } => {
1105                    // Extend the column names of the input with anonymous columns.
1106                    let mut column_names = results[index - 1].clone();
1107                    Self::extend_with_scalars(&mut column_names, scalars);
1108                    column_names
1109                }
1110                FlatMap {
1111                    input: _,
1112                    func,
1113                    exprs: _,
1114                } => {
1115                    // Extend the column names of the input with anonymous columns.
1116                    let mut column_names = results[index - 1].clone();
1117                    let func_output_start = column_names.len();
1118                    let func_output_end = column_names.len() + func.output_arity();
1119                    column_names.extend(Self::anonymous(func_output_start..func_output_end));
1120                    if let TableFunc::WithOrdinality { .. } = func {
1121                        // We know the name of the last column
1122                        // TODO(ggevay): generalize this to meaningful col names for all table functions
1123                        **column_names.last_mut().as_mut().expect(
1124                            "there is at least one output column, from the WITH ORDINALITY",
1125                        ) = ColumnName::Annotated(ORDINALITY_COL_NAME.into());
1126                    }
1127                    column_names
1128                }
1129                Filter {
1130                    input: _,
1131                    predicates: _,
1132                } => {
1133                    // Return the column names of the `input`.
1134                    results[index - 1].clone()
1135                }
1136                Join {
1137                    inputs: _,
1138                    equivalences: _,
1139                    implementation: _,
1140                } => {
1141                    let mut input_results = depends
1142                        .children_of_rev(index, expr.children().count())
1143                        .map(|child| &results[child])
1144                        .collect::<Vec<_>>();
1145                    input_results.reverse();
1146
1147                    let mut column_names = vec![];
1148                    for input_column_names in input_results {
1149                        column_names.extend(input_column_names.iter().cloned());
1150                    }
1151                    column_names
1152                }
1153                Reduce {
1154                    input: _,
1155                    group_key,
1156                    aggregates,
1157                    monotonic: _,
1158                    expected_group_size: _,
1159                } => {
1160                    // We clone and extend the input vector and then remove the part
1161                    // associated with the input at the end.
1162                    let mut column_names = results[index - 1].clone();
1163                    let input_arity = column_names.len();
1164
1165                    // Infer the group key part.
1166                    Self::extend_with_scalars(&mut column_names, group_key);
1167                    // Infer the aggregates part.
1168                    for aggregate in aggregates.iter() {
1169                        // The inferred name will consist of (1) the aggregate
1170                        // function name and (2) the aggregate expression (iff
1171                        // it is a simple column reference).
1172                        let func = aggregate.func.clone();
1173                        let expr = match aggregate.expr.as_column() {
1174                            Some(c) => column_names.get(c).unwrap_or(&ColumnName::Unknown).clone(),
1175                            None => ColumnName::Unknown,
1176                        };
1177                        column_names.push(ColumnName::Aggregate(func, Box::new(expr)));
1178                    }
1179                    // Remove the prefix associated with the input
1180                    column_names.drain(0..input_arity);
1181
1182                    column_names
1183                }
1184                TopK {
1185                    input: _,
1186                    group_key: _,
1187                    order_key: _,
1188                    limit: _,
1189                    offset: _,
1190                    monotonic: _,
1191                    expected_group_size: _,
1192                } => {
1193                    // Return the column names of the `input`.
1194                    results[index - 1].clone()
1195                }
1196                Negate { input: _ } => {
1197                    // Return the column names of the `input`.
1198                    results[index - 1].clone()
1199                }
1200                Threshold { input: _ } => {
1201                    // Return the column names of the `input`.
1202                    results[index - 1].clone()
1203                }
1204                Union { base: _, inputs: _ } => {
1205                    // Use the first non-empty column across all inputs.
1206                    let mut column_names = vec![];
1207
1208                    let mut inputs_results = depends
1209                        .children_of_rev(index, expr.children().count())
1210                        .map(|child| &results[child])
1211                        .collect::<Vec<_>>();
1212
1213                    let base_results = inputs_results.pop().unwrap();
1214                    inputs_results.reverse();
1215
1216                    for (i, mut column_name) in base_results.iter().cloned().enumerate() {
1217                        for input_results in inputs_results.iter() {
1218                            if !column_name.is_known() && input_results[i].is_known() {
1219                                column_name = input_results[i].clone();
1220                                break;
1221                            }
1222                        }
1223                        column_names.push(column_name);
1224                    }
1225
1226                    column_names
1227                }
1228                ArrangeBy { input: _, keys: _ } => {
1229                    // Return the column names of the `input`.
1230                    results[index - 1].clone()
1231                }
1232            }
1233        }
1234    }
1235}
1236
1237mod explain {
1238    //! Derived Analysis framework and definitions.
1239
1240    use std::collections::BTreeMap;
1241
1242    use mz_expr::MirRelationExpr;
1243    use mz_expr::explain::{ExplainContext, HumanizedExplain, HumanizerMode};
1244    use mz_ore::stack::RecursionLimitError;
1245    use mz_repr::explain::{Analyses, AnnotatedPlan};
1246
1247    use crate::analysis::equivalences::{Equivalences, HumanizedEquivalenceClasses};
1248
1249    // Analyses should have shortened paths when exported.
1250    use super::DerivedBuilder;
1251
1252    impl<'c> From<&ExplainContext<'c>> for DerivedBuilder<'c> {
1253        fn from(context: &ExplainContext<'c>) -> DerivedBuilder<'c> {
1254            let mut builder = DerivedBuilder::new(context.features);
1255            if context.config.subtree_size {
1256                builder.require(super::SubtreeSize);
1257            }
1258            if context.config.non_negative {
1259                builder.require(super::NonNegative);
1260            }
1261            if context.config.types {
1262                builder.require(super::SqlRelationType);
1263            }
1264            if context.config.arity {
1265                builder.require(super::Arity);
1266            }
1267            if context.config.keys {
1268                builder.require(super::UniqueKeys);
1269            }
1270            if context.config.cardinality {
1271                builder.require(super::Cardinality::with_stats(
1272                    context.cardinality_stats.clone(),
1273                ));
1274            }
1275            if context.config.column_names || context.config.humanized_exprs {
1276                builder.require(super::ColumnNames);
1277            }
1278            if context.config.equivalences {
1279                builder.require(Equivalences);
1280            }
1281            builder
1282        }
1283    }
1284
1285    /// Produce an [`AnnotatedPlan`] wrapping the given [`MirRelationExpr`] along
1286    /// with [`Analyses`] derived from the given context configuration.
1287    pub fn annotate_plan<'a>(
1288        plan: &'a MirRelationExpr,
1289        context: &'a ExplainContext,
1290    ) -> Result<AnnotatedPlan<'a, MirRelationExpr>, RecursionLimitError> {
1291        let mut annotations = BTreeMap::<&MirRelationExpr, Analyses>::default();
1292        let config = context.config;
1293
1294        // We want to annotate the plan with analyses in the following cases:
1295        // 1. An Analysis was explicitly requested in the ExplainConfig.
1296        // 2. Humanized expressions were requested in the ExplainConfig (in which
1297        //    case we need to derive the ColumnNames Analysis).
1298        if config.requires_analyses() || config.humanized_exprs {
1299            // get the annotation keys
1300            let subtree_refs = plan.post_order_vec();
1301            // get the annotation values
1302            let builder = DerivedBuilder::from(context);
1303            let derived = builder.visit(plan);
1304
1305            if config.subtree_size {
1306                for (expr, subtree_size) in std::iter::zip(
1307                    subtree_refs.iter(),
1308                    derived.results::<super::SubtreeSize>().into_iter(),
1309                ) {
1310                    let analyses = annotations.entry(expr).or_default();
1311                    analyses.subtree_size = Some(*subtree_size);
1312                }
1313            }
1314            if config.non_negative {
1315                for (expr, non_negative) in std::iter::zip(
1316                    subtree_refs.iter(),
1317                    derived.results::<super::NonNegative>().into_iter(),
1318                ) {
1319                    let analyses = annotations.entry(expr).or_default();
1320                    analyses.non_negative = Some(*non_negative);
1321                }
1322            }
1323
1324            if config.arity {
1325                for (expr, arity) in std::iter::zip(
1326                    subtree_refs.iter(),
1327                    derived.results::<super::Arity>().into_iter(),
1328                ) {
1329                    let analyses = annotations.entry(expr).or_default();
1330                    analyses.arity = Some(*arity);
1331                }
1332            }
1333
1334            if config.types {
1335                for (expr, types) in std::iter::zip(
1336                    subtree_refs.iter(),
1337                    derived.results::<super::SqlRelationType>().into_iter(),
1338                ) {
1339                    let analyses = annotations.entry(expr).or_default();
1340                    analyses.types = Some(types.clone());
1341                }
1342            }
1343
1344            if config.keys {
1345                for (expr, keys) in std::iter::zip(
1346                    subtree_refs.iter(),
1347                    derived.results::<super::UniqueKeys>().into_iter(),
1348                ) {
1349                    let analyses = annotations.entry(expr).or_default();
1350                    analyses.keys = Some(keys.clone());
1351                }
1352            }
1353
1354            if config.cardinality {
1355                for (expr, card) in std::iter::zip(
1356                    subtree_refs.iter(),
1357                    derived.results::<super::Cardinality>().into_iter(),
1358                ) {
1359                    let analyses = annotations.entry(expr).or_default();
1360                    analyses.cardinality = Some(card.to_string());
1361                }
1362            }
1363
1364            if config.column_names || config.humanized_exprs {
1365                for (expr, column_names) in std::iter::zip(
1366                    subtree_refs.iter(),
1367                    derived.results::<super::ColumnNames>().into_iter(),
1368                ) {
1369                    let analyses = annotations.entry(expr).or_default();
1370                    let value = column_names
1371                        .iter()
1372                        .map(|column_name| column_name.humanize(context.humanizer))
1373                        .collect();
1374                    analyses.column_names = Some(value);
1375                }
1376            }
1377
1378            if config.equivalences {
1379                for (expr, equivs) in std::iter::zip(
1380                    subtree_refs.iter(),
1381                    derived.results::<Equivalences>().into_iter(),
1382                ) {
1383                    let analyses = annotations.entry(expr).or_default();
1384                    analyses.equivalences = Some(match equivs.as_ref() {
1385                        Some(equivs) => HumanizedEquivalenceClasses {
1386                            equivalence_classes: equivs,
1387                            cols: analyses.column_names.as_ref(),
1388                            mode: HumanizedExplain::new(config.redacted),
1389                        }
1390                        .to_string(),
1391                        None => "<empty collection>".to_string(),
1392                    });
1393                }
1394            }
1395        }
1396
1397        Ok(AnnotatedPlan { plan, annotations })
1398    }
1399}
1400
1401/// Definition and helper structs for the [`Cardinality`] Analysis.
1402mod cardinality {
1403    use std::collections::{BTreeMap, BTreeSet};
1404
1405    use mz_expr::{
1406        BinaryFunc, Id, JoinImplementation, MirRelationExpr, MirScalarExpr, TableFunc, UnaryFunc,
1407        VariadicFunc,
1408    };
1409    use mz_ore::cast::{CastFrom, CastLossy, TryCastFrom};
1410    use mz_repr::GlobalId;
1411
1412    use ordered_float::OrderedFloat;
1413
1414    use super::{Analysis, Arity, SubtreeSize, UniqueKeys};
1415
1416    /// Compute the estimated cardinality of each subtree of a [MirRelationExpr] from the bottom up.
1417    #[allow(missing_debug_implementations)]
1418    pub struct Cardinality {
1419        /// Cardinalities for globally named entities
1420        pub stats: BTreeMap<GlobalId, usize>,
1421    }
1422
1423    impl Cardinality {
1424        /// A cardinality estimator with provided statistics for the given global identifiers
1425        pub fn with_stats(stats: BTreeMap<GlobalId, usize>) -> Self {
1426            Cardinality { stats }
1427        }
1428    }
1429
1430    impl Default for Cardinality {
1431        fn default() -> Self {
1432            Cardinality {
1433                stats: BTreeMap::new(),
1434            }
1435        }
1436    }
1437
1438    /// Cardinality estimates
1439    #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
1440    pub enum CardinalityEstimate {
1441        Unknown,
1442        Estimate(OrderedFloat<f64>),
1443    }
1444
1445    impl CardinalityEstimate {
1446        pub fn max(lhs: CardinalityEstimate, rhs: CardinalityEstimate) -> CardinalityEstimate {
1447            use CardinalityEstimate::*;
1448            match (lhs, rhs) {
1449                (Estimate(lhs), Estimate(rhs)) => Estimate(std::cmp::max(lhs, rhs)),
1450                _ => Unknown,
1451            }
1452        }
1453
1454        pub fn rounded(&self) -> Option<usize> {
1455            match self {
1456                CardinalityEstimate::Estimate(OrderedFloat(f)) => {
1457                    let rounded = f.ceil();
1458                    let flattened = usize::cast_from(
1459                        u64::try_cast_from(rounded)
1460                            .expect("positive and representable cardinality estimate"),
1461                    );
1462
1463                    Some(flattened)
1464                }
1465                CardinalityEstimate::Unknown => None,
1466            }
1467        }
1468    }
1469
1470    impl std::ops::Add for CardinalityEstimate {
1471        type Output = CardinalityEstimate;
1472
1473        fn add(self, rhs: Self) -> Self::Output {
1474            use CardinalityEstimate::*;
1475            match (self, rhs) {
1476                (Estimate(lhs), Estimate(rhs)) => Estimate(lhs + rhs),
1477                _ => Unknown,
1478            }
1479        }
1480    }
1481
1482    impl std::ops::Sub for CardinalityEstimate {
1483        type Output = CardinalityEstimate;
1484
1485        fn sub(self, rhs: Self) -> Self::Output {
1486            use CardinalityEstimate::*;
1487            match (self, rhs) {
1488                (Estimate(lhs), Estimate(rhs)) => Estimate(lhs - rhs),
1489                _ => Unknown,
1490            }
1491        }
1492    }
1493
1494    impl std::ops::Sub<CardinalityEstimate> for f64 {
1495        type Output = CardinalityEstimate;
1496
1497        fn sub(self, rhs: CardinalityEstimate) -> Self::Output {
1498            use CardinalityEstimate::*;
1499            if let Estimate(OrderedFloat(rhs)) = rhs {
1500                Estimate(OrderedFloat(self - rhs))
1501            } else {
1502                Unknown
1503            }
1504        }
1505    }
1506
1507    impl std::ops::Mul for CardinalityEstimate {
1508        type Output = CardinalityEstimate;
1509
1510        fn mul(self, rhs: Self) -> Self::Output {
1511            use CardinalityEstimate::*;
1512            match (self, rhs) {
1513                (Estimate(lhs), Estimate(rhs)) => Estimate(lhs * rhs),
1514                _ => Unknown,
1515            }
1516        }
1517    }
1518
1519    impl std::ops::Mul<f64> for CardinalityEstimate {
1520        type Output = CardinalityEstimate;
1521
1522        fn mul(self, rhs: f64) -> Self::Output {
1523            if let CardinalityEstimate::Estimate(OrderedFloat(lhs)) = self {
1524                CardinalityEstimate::Estimate(OrderedFloat(lhs * rhs))
1525            } else {
1526                CardinalityEstimate::Unknown
1527            }
1528        }
1529    }
1530
1531    impl std::ops::Div for CardinalityEstimate {
1532        type Output = CardinalityEstimate;
1533
1534        fn div(self, rhs: Self) -> Self::Output {
1535            use CardinalityEstimate::*;
1536            match (self, rhs) {
1537                (Estimate(lhs), Estimate(rhs)) => Estimate(lhs / rhs),
1538                _ => Unknown,
1539            }
1540        }
1541    }
1542
1543    impl std::ops::Div<f64> for CardinalityEstimate {
1544        type Output = CardinalityEstimate;
1545
1546        fn div(self, rhs: f64) -> Self::Output {
1547            use CardinalityEstimate::*;
1548            if let Estimate(lhs) = self {
1549                Estimate(lhs / OrderedFloat(rhs))
1550            } else {
1551                Unknown
1552            }
1553        }
1554    }
1555
1556    impl std::iter::Sum for CardinalityEstimate {
1557        fn sum<I: Iterator<Item = Self>>(iter: I) -> Self {
1558            iter.fold(CardinalityEstimate::from(0.0), |acc, elt| acc + elt)
1559        }
1560    }
1561
1562    impl std::iter::Product for CardinalityEstimate {
1563        fn product<I: Iterator<Item = Self>>(iter: I) -> Self {
1564            iter.fold(CardinalityEstimate::from(1.0), |acc, elt| acc * elt)
1565        }
1566    }
1567
1568    impl From<usize> for CardinalityEstimate {
1569        fn from(value: usize) -> Self {
1570            Self::Estimate(OrderedFloat(f64::cast_lossy(value)))
1571        }
1572    }
1573
1574    impl From<f64> for CardinalityEstimate {
1575        fn from(value: f64) -> Self {
1576            Self::Estimate(OrderedFloat(value))
1577        }
1578    }
1579
1580    /// The default selectivity for predicates we know nothing about.
1581    ///
1582    /// But see also expr/src/scalar.rs for `FilterCharacteristics::worst_case_scaling_factor()` for a more nuanced take.
1583    pub const WORST_CASE_SELECTIVITY: OrderedFloat<f64> = OrderedFloat(0.1);
1584
1585    // This section defines how we estimate cardinality for each syntactic construct.
1586    //
1587    // We split it up into functions to make it all a bit more tractable to work with.
1588    impl Cardinality {
1589        fn flat_map(&self, tf: &TableFunc, input: CardinalityEstimate) -> CardinalityEstimate {
1590            match tf {
1591                TableFunc::Wrap { types, width } => {
1592                    input * (f64::cast_lossy(types.len()) / f64::cast_lossy(*width))
1593                }
1594                _ => {
1595                    // TODO(mgree) what explosion factor should we make up?
1596                    input * CardinalityEstimate::from(4.0)
1597                }
1598            }
1599        }
1600
1601        fn predicate(
1602            &self,
1603            predicate_expr: &MirScalarExpr,
1604            unique_columns: &BTreeSet<usize>,
1605        ) -> OrderedFloat<f64> {
1606            let index_selectivity = |expr: &MirScalarExpr| -> Option<OrderedFloat<f64>> {
1607                match expr {
1608                    MirScalarExpr::Column(col, _) => {
1609                        if unique_columns.contains(col) {
1610                            // TODO(mgree): when we have index cardinality statistics, they should go here when `expr` is a `MirScalarExpr::Column` that's in `unique_columns`
1611                            None
1612                        } else {
1613                            None
1614                        }
1615                    }
1616                    _ => None,
1617                }
1618            };
1619
1620            match predicate_expr {
1621                MirScalarExpr::Column(_, _)
1622                | MirScalarExpr::Literal(_, _)
1623                | MirScalarExpr::CallUnmaterializable(_) => OrderedFloat(1.0),
1624                MirScalarExpr::CallUnary { func, expr } => match func {
1625                    UnaryFunc::Not(_) => OrderedFloat(1.0) - self.predicate(expr, unique_columns),
1626                    UnaryFunc::IsTrue(_) | UnaryFunc::IsFalse(_) => OrderedFloat(0.5),
1627                    UnaryFunc::IsNull(_) => {
1628                        if let Some(icard) = index_selectivity(expr) {
1629                            icard
1630                        } else {
1631                            WORST_CASE_SELECTIVITY
1632                        }
1633                    }
1634                    _ => WORST_CASE_SELECTIVITY,
1635                },
1636                MirScalarExpr::CallBinary { func, expr1, expr2 } => {
1637                    match func {
1638                        BinaryFunc::Eq(_) => {
1639                            match (index_selectivity(expr1), index_selectivity(expr2)) {
1640                                (Some(isel1), Some(isel2)) => std::cmp::max(isel1, isel2),
1641                                (Some(isel), None) | (None, Some(isel)) => isel,
1642                                (None, None) => WORST_CASE_SELECTIVITY,
1643                            }
1644                        }
1645                        // 1.0 - the Eq case
1646                        BinaryFunc::NotEq(_) => {
1647                            match (index_selectivity(expr1), index_selectivity(expr2)) {
1648                                (Some(isel1), Some(isel2)) => {
1649                                    OrderedFloat(1.0) - std::cmp::max(isel1, isel2)
1650                                }
1651                                (Some(isel), None) | (None, Some(isel)) => OrderedFloat(1.0) - isel,
1652                                (None, None) => OrderedFloat(1.0) - WORST_CASE_SELECTIVITY,
1653                            }
1654                        }
1655                        BinaryFunc::Lt(_)
1656                        | BinaryFunc::Lte(_)
1657                        | BinaryFunc::Gt(_)
1658                        | BinaryFunc::Gte(_) => {
1659                            // TODO(mgree) if we have high/low key values and one of the columns is an index, we can do better
1660                            OrderedFloat(0.33)
1661                        }
1662                        _ => OrderedFloat(1.0), // TOOD(mgree): are there other interesting cases?
1663                    }
1664                }
1665                MirScalarExpr::CallVariadic { func, exprs } => match func {
1666                    VariadicFunc::And => exprs
1667                        .iter()
1668                        .map(|expr| self.predicate(expr, unique_columns))
1669                        .product(),
1670                    VariadicFunc::Or => {
1671                        // TODO(mgree): BETWEEN will get compiled down to an AND of appropriate bounds---we could try to detect it and be clever
1672
1673                        // F(expr1 OR expr2) = F(expr1) + F(expr2) - F(expr1) * F(expr2), but generalized
1674                        let mut exprs = exprs.into_iter();
1675
1676                        let mut expr1;
1677
1678                        if let Some(first) = exprs.next() {
1679                            expr1 = self.predicate(first, unique_columns);
1680                        } else {
1681                            return OrderedFloat(1.0);
1682                        }
1683
1684                        for expr2 in exprs {
1685                            let expr2 = self.predicate(expr2, unique_columns);
1686                            expr1 = expr1 + expr2 - expr1 * expr2;
1687                        }
1688                        expr1
1689                    }
1690                    _ => OrderedFloat(1.0),
1691                },
1692                MirScalarExpr::If { cond: _, then, els } => std::cmp::max(
1693                    self.predicate(then, unique_columns),
1694                    self.predicate(els, unique_columns),
1695                ),
1696            }
1697        }
1698
1699        fn filter(
1700            &self,
1701            predicates: &Vec<MirScalarExpr>,
1702            keys: &Vec<Vec<usize>>,
1703            input: CardinalityEstimate,
1704        ) -> CardinalityEstimate {
1705            // TODO(mgree): should we try to do something for indices built on multiple columns?
1706            let mut unique_columns = BTreeSet::new();
1707            for key in keys {
1708                if key.len() == 1 {
1709                    unique_columns.insert(key[0]);
1710                }
1711            }
1712
1713            let mut estimate = input;
1714            for expr in predicates {
1715                let selectivity = self.predicate(expr, &unique_columns);
1716                debug_assert!(
1717                    OrderedFloat(0.0) <= selectivity && selectivity <= OrderedFloat(1.0),
1718                    "predicate selectivity {selectivity} should be in the range [0,1]"
1719                );
1720                estimate = estimate * selectivity.0;
1721            }
1722
1723            estimate
1724        }
1725
1726        fn join(
1727            &self,
1728            equivalences: &Vec<Vec<MirScalarExpr>>,
1729            _implementation: &JoinImplementation,
1730            unique_columns: BTreeMap<usize, usize>,
1731            mut inputs: Vec<CardinalityEstimate>,
1732        ) -> CardinalityEstimate {
1733            if inputs.is_empty() {
1734                return CardinalityEstimate::from(0.0);
1735            }
1736
1737            for equiv in equivalences {
1738                // those sources which have a unique key
1739                let mut unique_sources = BTreeSet::new();
1740                let mut all_unique = true;
1741
1742                for expr in equiv {
1743                    if let MirScalarExpr::Column(col, _) = expr {
1744                        if let Some(idx) = unique_columns.get(col) {
1745                            unique_sources.insert(*idx);
1746                        } else {
1747                            all_unique = false;
1748                        }
1749                    } else {
1750                        all_unique = false;
1751                    }
1752                }
1753
1754                // no unique columns in this equivalence
1755                if unique_sources.is_empty() {
1756                    continue;
1757                }
1758
1759                // ALL unique columns in this equivalence
1760                if all_unique {
1761                    // these inputs have unique keys for _all_ of the equivalence, so they're a bound on how many rows we'll get from those sources
1762                    // we'll find the leftmost such input and use it to hold the minimum; the other sources we set to 1.0 (so they have no effect)
1763                    let mut sources = unique_sources.iter();
1764
1765                    let lhs_idx = *sources.next().unwrap();
1766                    let mut lhs =
1767                        std::mem::replace(&mut inputs[lhs_idx], CardinalityEstimate::from(1.0));
1768                    for &rhs_idx in sources {
1769                        let rhs =
1770                            std::mem::replace(&mut inputs[rhs_idx], CardinalityEstimate::from(1.0));
1771                        lhs = CardinalityEstimate::min(lhs, rhs);
1772                    }
1773
1774                    inputs[lhs_idx] = lhs;
1775
1776                    // best option! go look at the next equivalence
1777                    continue;
1778                }
1779
1780                // some unique columns in this equivalence
1781                for idx in unique_sources {
1782                    // when joining R and S on R.x = S.x, if R.x is unique and S.x is not, we're bounded above by the cardinality of S
1783                    inputs[idx] = CardinalityEstimate::from(1.0);
1784                }
1785            }
1786
1787            let mut product = CardinalityEstimate::from(1.0);
1788            for input in inputs {
1789                product = product * input;
1790            }
1791            product
1792        }
1793
1794        fn reduce(
1795            &self,
1796            group_key: &Vec<MirScalarExpr>,
1797            expected_group_size: &Option<u64>,
1798            input: CardinalityEstimate,
1799        ) -> CardinalityEstimate {
1800            // TODO(mgree): if no `group_key` is present, we can do way better
1801
1802            if let Some(group_size) = expected_group_size {
1803                input / f64::cast_lossy(*group_size)
1804            } else if group_key.is_empty() {
1805                CardinalityEstimate::from(1.0)
1806            } else {
1807                // in the worst case, every row is its own group
1808                input
1809            }
1810        }
1811
1812        fn topk(
1813            &self,
1814            group_key: &Vec<usize>,
1815            limit: &Option<MirScalarExpr>,
1816            expected_group_size: &Option<u64>,
1817            input: CardinalityEstimate,
1818        ) -> CardinalityEstimate {
1819            // TODO: support simple arithmetic expressions
1820            let k = limit
1821                .as_ref()
1822                .and_then(|l| l.as_literal_int64())
1823                .map_or(1, |l| std::cmp::max(0, l));
1824
1825            if let Some(group_size) = expected_group_size {
1826                input * (f64::cast_lossy(k) / f64::cast_lossy(*group_size))
1827            } else if group_key.is_empty() {
1828                CardinalityEstimate::from(f64::cast_lossy(k))
1829            } else {
1830                // in the worst case, every row is its own group
1831                input.clone()
1832            }
1833        }
1834
1835        fn threshold(&self, input: CardinalityEstimate) -> CardinalityEstimate {
1836            // worst case scaling factor is 1
1837            input.clone()
1838        }
1839    }
1840
1841    impl Analysis for Cardinality {
1842        type Value = CardinalityEstimate;
1843
1844        fn announce_dependencies(builder: &mut crate::analysis::DerivedBuilder) {
1845            builder.require(crate::analysis::Arity);
1846            builder.require(crate::analysis::UniqueKeys);
1847        }
1848
1849        fn derive(
1850            &self,
1851            expr: &MirRelationExpr,
1852            index: usize,
1853            results: &[Self::Value],
1854            depends: &crate::analysis::Derived,
1855        ) -> Self::Value {
1856            use MirRelationExpr::*;
1857
1858            let sizes = depends.as_view().results::<SubtreeSize>();
1859            let arity = depends.as_view().results::<Arity>();
1860            let keys = depends.as_view().results::<UniqueKeys>();
1861
1862            match expr {
1863                Constant { rows, .. } => {
1864                    CardinalityEstimate::from(rows.as_ref().map_or_else(|_| 0, |v| v.len()))
1865                }
1866                Get { id, .. } => match id {
1867                    Id::Local(id) => depends
1868                        .bindings()
1869                        .get(id)
1870                        .and_then(|id| results.get(*id))
1871                        .copied()
1872                        .unwrap_or(CardinalityEstimate::Unknown),
1873                    Id::Global(id) => self
1874                        .stats
1875                        .get(id)
1876                        .copied()
1877                        .map(CardinalityEstimate::from)
1878                        .unwrap_or(CardinalityEstimate::Unknown),
1879                },
1880                Let { .. } | Project { .. } | Map { .. } | ArrangeBy { .. } | Negate { .. } => {
1881                    results[index - 1].clone()
1882                }
1883                LetRec { .. } =>
1884                // TODO(mgree): implement a recurrence-based approach (or at least identify common idioms, e.g. transitive closure)
1885                {
1886                    CardinalityEstimate::Unknown
1887                }
1888                Union { base: _, inputs: _ } => depends
1889                    .children_of_rev(index, expr.children().count())
1890                    .map(|off| results[off].clone())
1891                    .sum(),
1892                FlatMap { func, .. } => {
1893                    let input = results[index - 1];
1894                    self.flat_map(func, input)
1895                }
1896                Filter { predicates, .. } => {
1897                    let input = results[index - 1];
1898                    let keys = depends.results::<UniqueKeys>();
1899                    let keys = &keys[index - 1];
1900                    self.filter(predicates, keys, input)
1901                }
1902                Join {
1903                    equivalences,
1904                    implementation,
1905                    inputs,
1906                    ..
1907                } => {
1908                    let mut input_results = Vec::with_capacity(inputs.len());
1909
1910                    // maps a column to the index in `inputs` that it belongs to
1911                    let mut unique_columns = BTreeMap::new();
1912                    let mut key_offset = 0;
1913
1914                    let mut offset = 1;
1915                    for idx in 0..inputs.len() {
1916                        let input = results[index - offset];
1917                        input_results.push(input);
1918
1919                        let arity = arity[index - offset];
1920                        let keys = &keys[index - offset];
1921                        for key in keys {
1922                            if key.len() == 1 {
1923                                unique_columns.insert(key_offset + key[0], idx);
1924                            }
1925                        }
1926                        key_offset += arity;
1927
1928                        offset += &sizes[index - offset];
1929                    }
1930
1931                    self.join(equivalences, implementation, unique_columns, input_results)
1932                }
1933                Reduce {
1934                    group_key,
1935                    expected_group_size,
1936                    ..
1937                } => {
1938                    let input = results[index - 1];
1939                    self.reduce(group_key, expected_group_size, input)
1940                }
1941                TopK {
1942                    group_key,
1943                    limit,
1944                    expected_group_size,
1945                    ..
1946                } => {
1947                    let input = results[index - 1];
1948                    self.topk(group_key, limit, expected_group_size, input)
1949                }
1950                Threshold { .. } => {
1951                    let input = results[index - 1];
1952                    self.threshold(input)
1953                }
1954            }
1955        }
1956    }
1957
1958    impl std::fmt::Display for CardinalityEstimate {
1959        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1960            match self {
1961                CardinalityEstimate::Estimate(OrderedFloat(estimate)) => write!(f, "{estimate}"),
1962                CardinalityEstimate::Unknown => write!(f, "<UNKNOWN>"),
1963            }
1964        }
1965    }
1966}
1967
1968mod common_lattice {
1969    use crate::analysis::Lattice;
1970
1971    pub struct BoolLattice;
1972
1973    impl Lattice<bool> for BoolLattice {
1974        // `true` > `false`.
1975        fn top(&self) -> bool {
1976            true
1977        }
1978        // `false` is the greatest lower bound. `into` changes if it's true and `item` is false.
1979        fn meet_assign(&self, into: &mut bool, item: bool) -> bool {
1980            let changed = *into && !item;
1981            *into = *into && item;
1982            changed
1983        }
1984    }
1985}