Function install_lifted_mfp

src/transform/src/join_implementation.rs:943–990

This function continues the surgery that `implement_arrangements` started. (In theory, this function could be merged into `implement_arrangements`, but it would be a bit painful, because we need to access the join's `implementation` after `implement_arrangements`, which would be somewhat convoluted after what `install_lifted_mfp` does.) The given MFP should be the merged MFP from all the lifted

(new_join: &mut MirRelationExpr, mfp: MapFilterProject)

/// - Canonicalizes scalar expressions in maps and filters with respect to the join equivalences.
///   See inline comment for more details.
fn install_lifted_mfp(new_join: &mut MirRelationExpr, mfp: MapFilterProject) {
    if !mfp.is_identity() {
        let (mut map, mut filter, project) = mfp.as_map_filter_project();
        if let MirRelationExpr::Join { equivalences, .. } = new_join {
            for equivalence in equivalences.iter_mut() {
                for expr in equivalence.iter_mut() {
                    // permute `equivalences` in light of the project being lifted
                    expr.permute(&project);
                    // if column references refer to mapped expressions that have been
                    // lifted, replace the column reference with the mapped expression.
                    expr.visit_mut_pre(&mut |e| {
                        // This has to be a loop! This is because it can happen that the new
                        // expression is again a column reference, in which case the visitor
                        // wouldn't be called on it again. (The visitation continues with the
                        // children of the new expression, but won't visit the new expression
                        // itself again.)
                        while let MirScalarExpr::Column(c, _) = e {
                            if *c >= mfp.input_arity {
                                *e = map[*c - mfp.input_arity].clone();
                            } else {
                                break;
                            }
                        }
                    });
                }
            }
            // Canonicalize scalar expressions in maps and filters with respect to the join
            // equivalences. This often makes some filters identical, which are then removed.
            // The identical filters come from either
            //  - lifting several predicates that originally were pushed down by localizing to more
            //    than one inputs;
            //  - individual IS NOT NULL filters on each of the inputs, which become identical
            //    when rewritten using the join equivalences.
            //  (This allows for almost the same optimizations as when `Demand`
            //  used to insert Projections that were marking some columns to be
            //  identical, when Demand used to run after `JoinImplementation`.)
            let canonicalizer_map = mz_expr::canonicalize::get_canonicalizer_map(equivalences);
            for expr in map.iter_mut().chain(filter.iter_mut()) {
                expr.visit_mut_post(&mut |e| {
                    if let Some(canonical_expr) = canonicalizer_map.get(e) {
                        *e = canonical_expr.clone();
                    }
                })
            }
        }
        *new_join = new_join.clone().map(map).filter(filter).project(project);
    }
}

/// Permute the keys in `order` to compensate for projections being lifted from inputs.
/// `lifted_projections` has an optional projection for each input.