Method aggregate

datafusion/core/src/dataframe/mod.rs:646–679

Return a new `DataFrame` that aggregates the rows of the current `DataFrame`, first optionally grouping by the given expressions. # Example ``` # use datafusion::prelude::*; # use datafusion::error::Result; # use datafusion::functions_aggregate::expr_fn::min; # use datafusion_common::assert_batches_sorted_eq; # #[tokio::main] # async fn main() -> Result<()> { let ctx = SessionContext::new(); let

(
        self,
        group_expr: Vec<Expr>,
        aggr_expr: Vec<Expr>,
    )

    /// # }
    /// ```
    pub fn aggregate(
        self,
        group_expr: Vec<Expr>,
        aggr_expr: Vec<Expr>,
    ) -> Result<DataFrame> {
        let is_grouping_set = matches!(group_expr.as_slice(), [Expr::GroupingSet(_)]);
        let aggr_expr_len = aggr_expr.len();
        let options =
            LogicalPlanBuilderOptions::new().with_add_implicit_group_by_exprs(true);
        let plan = LogicalPlanBuilder::from(self.plan)
            .with_options(options)
            .aggregate(group_expr, aggr_expr)?
            .build()?;
        let plan = if is_grouping_set {
            let grouping_id_pos = plan.schema().fields().len() - 1 - aggr_expr_len;
            // For grouping sets we do a project to not expose the internal grouping id
            let exprs = plan
                .schema()
                .columns()
                .into_iter()
                .enumerate()
                .filter(|(idx, _)| *idx != grouping_id_pos)
                .map(|(_, column)| Expr::Column(column))
                .collect::<Vec<_>>();
            LogicalPlanBuilder::from(plan).project(exprs)?.build()?
        } else {
            plan
        };
        Ok(DataFrame {
            session_state: self.session_state,
            plan,
            projection_requires_validation: !is_grouping_set,
        })
    }

    /// Return a new DataFrame that adds the result of evaluating one or more
    /// window functions ([`Expr::WindowFunction`]) to the existing columns