/// Wrap a plan in an aggregate
Wrap a plan in an aggregate
Wrap a plan in an aggregate
fn aggregate(
        &self,
        input: &LogicalPlan,
        select_exprs: &[Expr],
        having_expr_opt: &Option<Expr>,
        group_by_exprs: Vec<Expr>,
        aggr_exprs: Vec<Expr>,
    ) -> Result<(LogicalPlan, Vec<Expr>, Option<Expr>)> {
        let aggr_projection_exprs = group_by_exprs
            .iter()
            .chain(aggr_exprs.iter())
            .cloned()
            .collect::<Vec<Expr>>();
        let plan = LogicalPlanBuilder::from(&input)
            .aggregate(group_by_exprs, aggr_exprs)?
            .build()?;
        // After aggregation, these are all of the columns that will be
        // available to next phases of planning.
        let column_exprs_post_aggr = aggr_projection_exprs
            .iter()
            .map(|expr| expr_as_column_expr(expr, input))
            .collect::<Result<Vec<Expr>>>()?;
        // Rewrite the SELECT expression to use the columns produced by the
        // aggregation.
        let select_exprs_post_aggr = select_exprs
            .iter()
            .map(|expr| rebase_expr(expr, &aggr_projection_exprs, input))
            .collect::<Result<Vec<Expr>>>()?;
        if !can_columns_satisfy_exprs(&column_exprs_post_aggr, &select_exprs_post_aggr)? {
            return Err(DataFusionError::Plan(
                "Projection references non-aggregate values".to_owned(),
            ));
        }
        // Rewrite the HAVING expression to use the columns produced by the
        // aggregation.
        let having_expr_post_aggr_opt = if let Some(having_expr) = having_expr_opt {
            let having_expr_post_aggr =
                rebase_expr(having_expr, &aggr_projection_exprs, input)?;
            if !can_columns_satisfy_exprs(
                &column_exprs_post_aggr,
                &[having_expr_post_aggr.clone()],
            )? {
                return Err(DataFusionError::Plan(
                    "Having references non-aggregate values".to_owned(),
                ));
            }
            Some(having_expr_post_aggr)
        } else {
            None
        };
        Ok((plan, select_exprs_post_aggr, having_expr_post_aggr_opt))
    }
fn aggregate ( & self , input : & LogicalPlan , select_exprs : & [ Expr ] , having_expr_opt : & Option < Expr > , group_by_exprs : Vec < Expr > , aggr_exprs : Vec < Expr > , ) -> Result < ( LogicalPlan , Vec < Expr > , Option < Expr > ) > { let aggr_projection_exprs = group_by_exprs . iter ( ) . chain ( aggr_exprs . iter ( ) ) . cloned ( ) . collect :: < Vec < Expr > > ( ) ; let plan = LogicalPlanBuilder :: from ( & input ) . aggregate ( group_by_exprs , aggr_exprs ) ? . build ( ) ? ; // After aggregation, these are all of the columns that will be // available to next phases of planning. let column_exprs_post_aggr = aggr_projection_exprs . iter ( ) . map ( | expr | expr_as_column_expr ( expr , input ) ) . collect :: < Result < Vec < Expr > > > ( ) ? ; // Rewrite the SELECT expression to use the columns produced by the // aggregation. let select_exprs_post_aggr = select_exprs . iter ( ) . map ( | expr | rebase_expr ( expr , & aggr_projection_exprs , input ) ) . collect :: < Result < Vec < Expr > > > ( ) ? ; if ! can_columns_satisfy_exprs ( & column_exprs_post_aggr , & select_exprs_post_aggr ) ? { return Err ( DataFusionError :: Plan ( " " . to_owned ( ) , ) ) ; } // Rewrite the HAVING expression to use the columns produced by the // aggregation. let having_expr_post_aggr_opt = if let Some ( having_expr ) = having_expr_opt { let having_expr_post_aggr = rebase_expr ( having_expr , & aggr_projection_exprs , input ) ? ; if ! can_columns_satisfy_exprs ( & column_exprs_post_aggr , & [ having_expr_post_aggr . clone ( ) ] , ) ? { return Err ( DataFusionError :: Plan ( " " . to_owned ( ) , ) ) ; } Some ( having_expr_post_aggr ) } else { None } ; Ok ( ( plan , select_exprs_post_aggr , having_expr_post_aggr_opt ) ) }
Wrap a plan in an aggregate
Wrap a plan in an aggregate
&self
input
&LogicalPlan
select_exprs
&[Expr]
having_expr_opt
&Option<Expr>
group_by_exprs
Vec<Expr>
aggr_exprs
Vec<Expr>