Method action

src/transform/src/join_implementation.rs:130–526

Determines the join implementation for join operators.

(
        &self,
        relation: &mut MirRelationExpr,
        mfp_above: MapFilterProject,
        indexes: &IndexMap,
        stats: &dyn StatisticsOracle,
        features: &OptimizerFeatures,
  

    /// Determines the join implementation for join operators.
    pub fn action(
        &self,
        relation: &mut MirRelationExpr,
        mfp_above: MapFilterProject,
        indexes: &IndexMap,
        stats: &dyn StatisticsOracle,
        features: &OptimizerFeatures,
    ) -> Result<(), TransformError> {
        if let MirRelationExpr::Join {
            inputs,
            equivalences,
            // (Note that `JoinImplementation` runs in a fixpoint loop.)
            // If the current implementation is
            //  - Unimplemented, then we need to come up with an implementation.
            //  - Differential, then we consider switching to a Delta join, because we might have
            //    inserted some ArrangeBys that create new arrangements when we came up with the
            //    Differential plan, in which case a Delta join might have become viable.
            //  - Delta, then we are good already.
            //  - IndexedFilter, then we just leave that alone, because those are out of scope
            //    for JoinImplementation (they are created by `LiteralConstraints`).
            // We don't want to change from a Differential plan to an other Differential plan, or
            // from a Delta plan to an other Delta plan, because the second run cannot distinguish
            // between an ArrangeBy that marks an already existing arrangement and an ArrangeBy
            // that was inserted by a previous run of JoinImplementation. (We should eventually
            // refactor this to make ArrangeBy unambiguous somehow. Maybe move JoinImplementation
            // to the lowering.)
            implementation: implementation @ (Unimplemented | Differential(..)),
        } = relation
        {
            // If we eagerly plan delta joins, we don't need the second run to "pick up" delta joins
            // that could be planned with the arrangements from a differential. If such a delta
            // join were viable, we'd have already planned it the first time.
            if features.enable_eager_delta_joins && !matches!(implementation, Unimplemented) {
                return Ok(());
            }

            let input_types = inputs.iter().map(|i| i.typ()).collect::<Vec<_>>();

            // Canonicalize the equivalence classes
            if matches!(implementation, Unimplemented) {
                // Let's do this only if it's the first run of JoinImplementation, in which case we
                // are guaranteed to produce a new plan, which will be compatible with the modified
                // equivalences from the below call. Otherwise, if we already have a Differential or
                // a Delta join, then we might discard the new plan and go with the old plan, which
                // was created previously for the old equivalences, and might be invalid for the
                // modified equivalences from the below call. Note that this issue can arise only if
                // `canonicalize_equivalences` is not idempotent, which unfortunately seems to be
                // the case.
                mz_expr::canonicalize::canonicalize_equivalences(
                    equivalences,
                    input_types.iter().map(|t| &t.column_types),
                );
            }

            // Common information of broad utility.
            let input_mapper = JoinInputMapper::new_from_input_types(&input_types);

            // The first fundamental question is whether we should employ a delta query or not.