pub fn create_from(
        source_relation: usize,
        equivalences: &[Vec<MirScalarExpr>],
        join_order: &[(usize, Vec<MirScalarExpr>)],
        input_mapper: expr::JoinInputMapper,
        map_filter_project: &mut MapFilterProject,
    ) -> Self {
        let temporal_mfp = map_filter_project.extract_temporal();

        // Construct initial join build state.
        // This state will evolves as we build the join dataflow.
        let mut join_build_state = JoinBuildState::new(
            input_mapper.global_columns(source_relation),
            &equivalences,
            &map_filter_project,
        );

        // We would prefer to extract a closure here, but we do not know if
        // the input will be arranged or not.
        let initial_closure = None;

        // Sequence of steps to apply.
        let mut stage_plans = Vec::with_capacity(join_order.len());

        // Track the set of bound input relations, for equivalence resolution.
        let mut bound_inputs = vec![source_relation];

        // The arity of the stream of updates, to be modified for each lookup relation
        let mut stream_arity = input_mapper.input_arity(source_relation);
        // Iterate through the join order instructions, assembling keys and
        // closures to use.
        for (lookup_relation, lookup_key) in join_order.iter() {
            // rebase the intended key to use global column identifiers.
            let lookup_key_rebased = lookup_key
                .iter()
                .map(|k| input_mapper.map_expr_to_global(k.clone(), *lookup_relation))
                .collect::<Vec<_>>();

            // Expressions to use as a key for the stream of incoming updates
            // are determined by locating the elements of `lookup_key` among
            // the existing bound `columns`. If that cannot be done, the plan
            // is irrecoverably defective and we panic.
            // TODO: explicitly validate this before rendering.
            let stream_key = lookup_key_rebased
                .iter()
                .map(|expr| {
                    let mut bound_expr = input_mapper
                        .find_bound_expr(expr, &bound_inputs, &join_build_state.equivalences)
                        .expect("Expression in join plan is not bound at time of use");
                    // Rewrite column references to physical locations.
                    bound_expr.permute_map(&join_build_state.column_map);
                    bound_expr
                })
                .collect::<Vec<_>>();

            // Introduce new columns and expressions they enable. Form a new closure.
            let closure = join_build_state.add_columns(
                input_mapper.global_columns(*lookup_relation),
                &lookup_key_rebased,
            );
            let (stream_permutation, stream_thinning) =
                Permutation::construct_from_expr(&stream_key, stream_arity);
            let (lookup_permutation, lookup_thinning) = Permutation::construct_from_expr(
                &lookup_key,
                input_mapper.input_arity(*lookup_relation),
            );
            stream_arity = closure.before.projection.len();

            bound_inputs.push(*lookup_relation);

            // record the stage plan as next in the path.
            stage_plans.push(LinearStagePlan {
                lookup_relation: *lookup_relation,
                stream_key,
                stream_permutation,
                stream_thinning,
                lookup_key: lookup_key.clone(),
                lookup_permutation,
                lookup_thinning,
                closure,
            });
        }

        // determine a final closure, and complete the path plan.
        let final_closure = join_build_state.complete();
        let final_closure = if final_closure.is_identity() {
            None
        } else {
            Some(final_closure)
        };

        // Now that `map_filter_project` has been captured in the state builder,
        // assign the remaining temporal predicates to it, for the caller's use.
        *map_filter_project = temporal_mfp;

        // Form and return the complete join plan.
        LinearJoinPlan {
            source_relation,
            initial_closure,
            stage_plans,
            final_closure,
        }
    }
/// Create a new join plan from the required arguments.
Create a new join plan from the required arguments.
Create a new join plan from the required arguments .
pub fn create_from(
        source_relation: usize,
        equivalences: &[Vec<MirScalarExpr>],
        join_order: &[(usize, Vec<MirScalarExpr>)],
        input_mapper: expr::JoinInputMapper,
        map_filter_project: &mut MapFilterProject,
    ) -> Self {
        let temporal_mfp = map_filter_project.extract_temporal();
        // Construct initial join build state.
        // This state will evolves as we build the join dataflow.
        let mut join_build_state = JoinBuildState::new(
            input_mapper.global_columns(source_relation),
            &equivalences,
            &map_filter_project,
        );
        // We would prefer to extract a closure here, but we do not know if
        // the input will be arranged or not.
        let initial_closure = None;
        // Sequence of steps to apply.
        let mut stage_plans = Vec::with_capacity(join_order.len());
        // Track the set of bound input relations, for equivalence resolution.
        let mut bound_inputs = vec![source_relation];
        // The arity of the stream of updates, to be modified for each lookup relation
        let mut stream_arity = input_mapper.input_arity(source_relation);
        // Iterate through the join order instructions, assembling keys and
        // closures to use.
        for (lookup_relation, lookup_key) in join_order.iter() {
            // rebase the intended key to use global column identifiers.
            let lookup_key_rebased = lookup_key
                .iter()
                .map(|k| input_mapper.map_expr_to_global(k.clone(), *lookup_relation))
                .collect::<Vec<_>>();
            // Expressions to use as a key for the stream of incoming updates
            // are determined by locating the elements of `lookup_key` among
            // the existing bound `columns`. If that cannot be done, the plan
            // is irrecoverably defective and we panic.
            // TODO: explicitly validate this before rendering.
            let stream_key = lookup_key_rebased
                .iter()
                .map(|expr| {
                    let mut bound_expr = input_mapper
                        .find_bound_expr(expr, &bound_inputs, &join_build_state.equivalences)
                        .expect("Expression in join plan is not bound at time of use");
                    // Rewrite column references to physical locations.
                    bound_expr.permute_map(&join_build_state.column_map);
                    bound_expr
                })
                .collect::<Vec<_>>();
            // Introduce new columns and expressions they enable. Form a new closure.
            let closure = join_build_state.add_columns(
                input_mapper.global_columns(*lookup_relation),
                &lookup_key_rebased,
            );
            let (stream_permutation, stream_thinning) =
                Permutation::construct_from_expr(&stream_key, stream_arity);
            let (lookup_permutation, lookup_thinning) = Permutation::construct_from_expr(
                &lookup_key,
                input_mapper.input_arity(*lookup_relation),
            );
            stream_arity = closure.before.projection.len();
            bound_inputs.push(*lookup_relation);
            // record the stage plan as next in the path.
            stage_plans.push(LinearStagePlan {
                lookup_relation: *lookup_relation,
                stream_key,
                stream_permutation,
                stream_thinning,
                lookup_key: lookup_key.clone(),
                lookup_permutation,
                lookup_thinning,
                closure,
            });
        }
        // determine a final closure, and complete the path plan.
        let final_closure = join_build_state.complete();
        let final_closure = if final_closure.is_identity() {
            None
        } else {
            Some(final_closure)
        };
        // Now that `map_filter_project` has been captured in the state builder,
        // assign the remaining temporal predicates to it, for the caller's use.
        *map_filter_project = temporal_mfp;
        // Form and return the complete join plan.
        LinearJoinPlan {
            source_relation,
            initial_closure,
            stage_plans,
            final_closure,
        }
    }
pub fn create_from ( source_relation : usize , equivalences : & [ Vec < MirScalarExpr > ] , join_order : & [ ( usize , Vec < MirScalarExpr > ) ] , input_mapper : expr :: JoinInputMapper , map_filter_project : & mut MapFilterProject , ) -> Self { let temporal_mfp = map_filter_project . extract_temporal ( ) ; // Construct initial join build state. // This state will evolves as we build the join dataflow. let mut join_build_state = JoinBuildState :: new ( input_mapper . global_columns ( source_relation ) , & equivalences , & map_filter_project , ) ; // We would prefer to extract a closure here, but we do not know if // the input will be arranged or not. let initial_closure = None ; // Sequence of steps to apply. let mut stage_plans = Vec :: with_capacity ( join_order . len ( ) ) ; // Track the set of bound input relations, for equivalence resolution. let mut bound_inputs = vec ! [ source_relation ] ; // The arity of the stream of updates, to be modified for each lookup relation let mut stream_arity = input_mapper . input_arity ( source_relation ) ; // Iterate through the join order instructions, assembling keys and // closures to use. for ( lookup_relation , lookup_key ) in join_order . iter ( ) { // rebase the intended key to use global column identifiers. let lookup_key_rebased = lookup_key . iter ( ) . map ( | k | input_mapper . map_expr_to_global ( k . clone ( ) , * lookup_relation ) ) . collect :: < Vec < _ > > ( ) ; // Expressions to use as a key for the stream of incoming updates // are determined by locating the elements of `lookup_key` among // the existing bound `columns`. If that cannot be done, the plan // is irrecoverably defective and we panic. // TODO: explicitly validate this before rendering. let stream_key = lookup_key_rebased . iter ( ) . map ( | expr | { let mut bound_expr = input_mapper . find_bound_expr ( expr , & bound_inputs , & join_build_state . equivalences ) . expect ( " " ) ; // Rewrite column references to physical locations. bound_expr . permute_map ( & join_build_state . column_map ) ; bound_expr } ) . collect :: < Vec < _ > > ( ) ; // Introduce new columns and expressions they enable. Form a new closure. let closure = join_build_state . add_columns ( input_mapper . global_columns ( * lookup_relation ) , & lookup_key_rebased , ) ; let ( stream_permutation , stream_thinning ) = Permutation :: construct_from_expr ( & stream_key , stream_arity ) ; let ( lookup_permutation , lookup_thinning ) = Permutation :: construct_from_expr ( & lookup_key , input_mapper . input_arity ( * lookup_relation ) , ) ; stream_arity = closure . before . projection . len ( ) ; bound_inputs . push ( * lookup_relation ) ; // record the stage plan as next in the path. stage_plans . push ( LinearStagePlan { lookup_relation : * lookup_relation , stream_key , stream_permutation , stream_thinning , lookup_key : lookup_key . clone ( ) , lookup_permutation , lookup_thinning , closure , } ) ; } // determine a final closure, and complete the path plan. let final_closure = join_build_state . complete ( ) ; let final_closure = if final_closure . is_identity ( ) { None } else { Some ( final_closure ) } ; // Now that `map_filter_project` has been captured in the state builder, // assign the remaining temporal predicates to it, for the caller's use. * map_filter_project = temporal_mfp ; // Form and return the complete join plan. LinearJoinPlan { source_relation , initial_closure , stage_plans , final_closure , } }
Create a new join plan from the required arguments.
Create a new join plan from the required arguments .
source_relation
usize
equivalences
&[Vec<MirScalarExpr>]
join_order
&[(usize, Vec<MirScalarExpr>)]
input_mapper
expr::JoinInputMapper
map_filter_project
&mut MapFilterProject
source_relation
usize
equivalences
&[Vec<MirScalarExpr>]
join_order
&[(usize, Vec<MirScalarExpr>)]
input_mapper
expr::JoinInputMapper
map_filter_project
&mut MapFilterProject