/// Recursively combine nodes in the LogicalPlan based on the conditions
/// listed above.
Recursively combine nodes in the LogicalPlan based on the conditions
listed above.
Recursively combine nodes in the LogicalPlan based on the conditions listed above .
fn combine_lp_nodes(
    input_l: Node,
    input_r: Node,
    lp_arena: &mut Arena<ALogicalPlan>,
    expr_arena: &mut Arena<AExpr>,
) -> Option<ALogicalPlan> {
    use ALogicalPlan::*;
    match (lp_arena.get(input_l), lp_arena.get(input_r)) {
        (
            Aggregate {
                input: child_input_l,
                keys: keys_l,
                aggs: aggs_l,
                apply: apply_l,
                maintain_order: maintain_l,
                ..
            },
            Aggregate {
                input: child_input_r,
                keys: keys_r,
                aggs: aggs_r,
                apply: apply_r,
                maintain_order: maintain_r,
                ..
            },
            // skip if we have custom functions
        ) if {
            apply_l.is_none()
                && apply_r.is_none()
                // check if aggregation keys can be combined.
                && equal_aexprs(keys_l, keys_r, expr_arena)
                && maintain_l == maintain_r
        }
        =>
            {
                let maintain_order = *maintain_l;
                let keys = keys_l.clone();
                let aggs = aggs_l
                    .iter()
                    .copied()
                    .chain(aggs_r.iter().copied())
                    .collect();
                combine_lp_nodes(*child_input_l, *child_input_r, lp_arena, expr_arena)
                    .map(|input| {
                        let node = lp_arena.add(input);
                        ALogicalPlanBuilder::new(node, expr_arena, lp_arena)
                            .groupby(keys, aggs, None, maintain_order)
                            .build()
                    })
            }
        (Projection {input: child_input_l, expr: expr_l, ..},
            Projection {input: child_input_r, expr: expr_r, ..})
        // check if all previous nodes/ transformations are equal
        if ALogicalPlan::eq(*child_input_l, *child_input_r, lp_arena, expr_arena)
        => {
            let exprs: Vec<_> = expr_l.iter().copied().chain(expr_r.iter().copied()).collect();
            let exprs = remove_duplicate_aexprs(&exprs, expr_arena);
            combine_lp_nodes(*child_input_l, *child_input_r, lp_arena, expr_arena)
                .map(|input| {
                    let node = lp_arena.add(input);
                    ALogicalPlanBuilder::new(node, expr_arena, lp_arena)
                        .project(exprs)
                        .build()
                })
        }
        #[cfg(feature = "csv-file")]
        (CsvScan {path: path_l,
            schema,
            options: options_l,
            predicate,
            aggregate,
        },
            CsvScan {path: path_r, options: options_r, ..})
        if canonicalize(path_l).unwrap() == canonicalize(path_r).unwrap()
        => {
            let mut options_l = options_l.clone();
            let path = path_l.clone();
            let with_columns = match (&options_l.with_columns, &options_r.with_columns) {
                (Some(l), Some(r)) => Some(l.iter().cloned().chain(r.iter().cloned()).collect()),
                (Some(l), None) => Some(l.clone()),
                (None, Some(r)) => Some(r.clone()),
                _ => None
            };
            options_l.with_columns = with_columns;
            Some(CsvScan {
                path,
                schema: schema.clone(),
                options: options_l,
                predicate: *predicate,
                aggregate: aggregate.clone(),
            })
        }
        _ => {
            if ALogicalPlan::eq(input_l, input_r, lp_arena, expr_arena) {
                Some(lp_arena.take(input_l))
            } else {
                None
            }
        },
    }
}
fn combine_lp_nodes ( input_l : Node , input_r : Node , lp_arena : & mut Arena < ALogicalPlan > , expr_arena : & mut Arena < AExpr > , ) -> Option < ALogicalPlan > { use ALogicalPlan :: * ; match ( lp_arena . get ( input_l ) , lp_arena . get ( input_r ) ) { ( Aggregate { input : child_input_l , keys : keys_l , aggs : aggs_l , apply : apply_l , maintain_order : maintain_l , .. } , Aggregate { input : child_input_r , keys : keys_r , aggs : aggs_r , apply : apply_r , maintain_order : maintain_r , .. } , // skip if we have custom functions ) if { apply_l . is_none ( ) && apply_r . is_none ( ) // check if aggregation keys can be combined. && equal_aexprs ( keys_l , keys_r , expr_arena ) && maintain_l == maintain_r } => { let maintain_order = * maintain_l ; let keys = keys_l . clone ( ) ; let aggs = aggs_l . iter ( ) . copied ( ) . chain ( aggs_r . iter ( ) . copied ( ) ) . collect ( ) ; combine_lp_nodes ( * child_input_l , * child_input_r , lp_arena , expr_arena ) . map ( | input | { let node = lp_arena . add ( input ) ; ALogicalPlanBuilder :: new ( node , expr_arena , lp_arena ) . groupby ( keys , aggs , None , maintain_order ) . build ( ) } ) } ( Projection { input : child_input_l , expr : expr_l , .. } , Projection { input : child_input_r , expr : expr_r , .. } ) // check if all previous nodes/ transformations are equal if ALogicalPlan :: eq ( * child_input_l , * child_input_r , lp_arena , expr_arena ) => { let exprs : Vec < _ > = expr_l . iter ( ) . copied ( ) . chain ( expr_r . iter ( ) . copied ( ) ) . collect ( ) ; let exprs = remove_duplicate_aexprs ( & exprs , expr_arena ) ; combine_lp_nodes ( * child_input_l , * child_input_r , lp_arena , expr_arena ) . map ( | input | { let node = lp_arena . add ( input ) ; ALogicalPlanBuilder :: new ( node , expr_arena , lp_arena ) . project ( exprs ) . build ( ) } ) } # [ cfg ( feature " " ) ] ( CsvScan { path : path_l , schema , options : options_l , predicate , aggregate , } , CsvScan { path : path_r , options : options_r , .. } ) if canonicalize ( path_l ) . unwrap ( ) == canonicalize ( path_r ) . unwrap ( ) => { let mut options_l = options_l . clone ( ) ; let path = path_l . clone ( ) ; let with_columns = match ( & options_l . with_columns , & options_r . with_columns ) { ( Some ( l ) , Some ( r ) ) => Some ( l . iter ( ) . cloned ( ) . chain ( r . iter ( ) . cloned ( ) ) . collect ( ) ) , ( Some ( l ) , None ) => Some ( l . clone ( ) ) , ( None , Some ( r ) ) => Some ( r . clone ( ) ) , _ => None } ; options_l . with_columns = with_columns ; Some ( CsvScan { path , schema : schema . clone ( ) , options : options_l , predicate : * predicate , aggregate : aggregate . clone ( ) , } ) } _ => { if ALogicalPlan :: eq ( input_l , input_r , lp_arena , expr_arena ) { Some ( lp_arena . take ( input_l ) ) } else { None } } , } }
Recursively combine nodes in the LogicalPlan based on the conditions
listed above.
Recursively combine nodes in the LogicalPlan based on the conditions listed above .
input_l
Node
input_r
Node
lp_arena
&mut Arena<ALogicalPlan>
expr_arena
&mut Arena<AExpr>