Move First Value UDAF and builtin first / last function to aggregate-functions

datafusion/core/Cargo.toml

@@ -93,12 +93,12 @@ bytes = { workspace = true }
 bzip2 = { version = "0.4.3", optional = true }
 chrono = { workspace = true }
 dashmap = { workspace = true }
-datafusion-functions-aggregate = { workspace = true }
 datafusion-common = { workspace = true, features = ["object_store"] }
 datafusion-common-runtime = { workspace = true }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-functions = { workspace = true }
+datafusion-functions-aggregate = { workspace = true }
 datafusion-functions-array = { workspace = true, optional = true }
 datafusion-optimizer = { workspace = true }
 datafusion-physical-expr = { workspace = true }

datafusion/core/src/execution/context/mod.rs

@@ -24,7 +24,6 @@ use std::string::String;
 use std::sync::{Arc, Weak};
 
 use super::options::ReadOptions;
-use crate::{functions, functions_aggregate, functions_array};
 use crate::{
     catalog::information_schema::{InformationSchemaProvider, INFORMATION_SCHEMA},
     catalog::listing_schema::ListingSchemaProvider,
@@ -58,6 +57,7 @@ use crate::{
     physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner},
     variable::{VarProvider, VarType},
 };
+use crate::{functions, functions_aggregate, functions_array};
 
 use arrow::datatypes::{DataType, SchemaRef};
 use arrow::record_batch::RecordBatch;

datafusion/functions-aggregate/src/first_last.rs

@@ -27,10 +27,7 @@ use datafusion_common::{
 use datafusion_expr::function::AccumulatorArgs;
 use datafusion_expr::type_coercion::aggregates::NUMERICS;
 use datafusion_expr::utils::format_state_name;
-use datafusion_expr::{
-    Accumulator, AccumulatorFactoryFunction, AggregateUDFImpl, Expr, Signature,
-    Volatility,
-};
+use datafusion_expr::{Accumulator, AggregateUDFImpl, Expr, Signature, Volatility};
 use datafusion_physical_expr_common::aggregate::utils::{
     down_cast_any_ref, get_sort_options, ordering_fields,
 };
@@ -48,14 +45,12 @@ make_udaf_function!(
     first_value,
     value,
     "Returns the first value in a group of values.",
-    first_value_udaf,
-    create_first_value_accumulator
+    first_value_udaf
 );
 
 pub struct FirstValue {
     signature: Signature,
     aliases: Vec<String>,
-    accumulator: AccumulatorFactoryFunction,
 }
 
 impl Debug for FirstValue {
@@ -68,12 +63,17 @@ impl Debug for FirstValue {
     }
 }
 
+impl Default for FirstValue {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl FirstValue {
-    pub fn new(accumulator: AccumulatorFactoryFunction) -> Self {
+    pub fn new() -> Self {
         Self {
             aliases: vec![String::from("FIRST_VALUE")],
             signature: Signature::uniform(1, NUMERICS.to_vec(), Volatility::Immutable),
-            accumulator,
         }
     }
 }
@@ -96,7 +96,45 @@ impl AggregateUDFImpl for FirstValue {
     }
 
     fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
-        (self.accumulator)(acc_args)
+        let mut all_sort_orders = vec![];
+
+        // Construct PhysicalSortExpr objects from Expr objects:
+        let mut sort_exprs = vec![];
+        for expr in acc_args.sort_exprs {
+            if let Expr::Sort(sort) = expr {
+                if let Expr::Column(col) = sort.expr.as_ref() {
+                    let name = &col.name;
+                    let e = expressions::column::col(name, acc_args.schema)?;
+                    sort_exprs.push(PhysicalSortExpr {
+                        expr: e,
+                        options: SortOptions {
+                            descending: !sort.asc,
+                            nulls_first: sort.nulls_first,
+                        },
+                    });
+                }
+            }
+        }
+        if !sort_exprs.is_empty() {
+            all_sort_orders.extend(sort_exprs);
+        }
+
+        let ordering_req = all_sort_orders;
+
+        let ordering_dtypes = ordering_req
+            .iter()
+            .map(|e| e.expr.data_type(acc_args.schema))
+            .collect::<Result<Vec<_>>>()?;
+
+        let requirement_satisfied = ordering_req.is_empty();
+
+        FirstValueAccumulator::try_new(
+            acc_args.data_type,
+            &ordering_dtypes,
+            ordering_req,
+            acc_args.ignore_nulls,
+        )
+        .map(|acc| Box::new(acc.with_requirement_satisfied(requirement_satisfied)) as _)
     }
 
     fn state_fields(
@@ -120,50 +158,6 @@ impl AggregateUDFImpl for FirstValue {
     }
 }
 
-pub(crate) fn create_first_value_accumulator(
-    acc_args: AccumulatorArgs,
-) -> Result<Box<dyn Accumulator>> {
-    let mut all_sort_orders = vec![];
-
-    // Construct PhysicalSortExpr objects from Expr objects:
-    let mut sort_exprs = vec![];
-    for expr in acc_args.sort_exprs {
-        if let Expr::Sort(sort) = expr {
-            if let Expr::Column(col) = sort.expr.as_ref() {
-                let name = &col.name;
-                let e = expressions::column::col(name, acc_args.schema)?;
-                sort_exprs.push(PhysicalSortExpr {
-                    expr: e,
-                    options: SortOptions {
-                        descending: !sort.asc,
-                        nulls_first: sort.nulls_first,
-                    },
-                });
-            }
-        }
-    }
-    if !sort_exprs.is_empty() {
-        all_sort_orders.extend(sort_exprs);
-    }
-
-    let ordering_req = all_sort_orders;
-
-    let ordering_dtypes = ordering_req
-        .iter()
-        .map(|e| e.expr.data_type(acc_args.schema))
-        .collect::<Result<Vec<_>>>()?;
-
-    let requirement_satisfied = ordering_req.is_empty();
-
-    FirstValueAccumulator::try_new(
-        acc_args.data_type,
-        &ordering_dtypes,
-        ordering_req,
-        acc_args.ignore_nulls,
-    )
-    .map(|acc| Box::new(acc.with_requirement_satisfied(requirement_satisfied)) as _)
-}
-
 #[derive(Debug)]
 pub struct FirstValueAccumulator {
     first: ScalarValue,

datafusion/functions-aggregate/src/macros.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 macro_rules! make_udaf_function {
-    ($UDAF:ty, $EXPR_FN:ident, $($arg:ident)*, $DOC:expr, $AGGREGATE_UDF_FN:ident, $ACCUMULATOR:ident) => {
+    ($UDAF:ty, $EXPR_FN:ident, $($arg:ident)*, $DOC:expr, $AGGREGATE_UDF_FN:ident) => {
         paste::paste! {
             // "fluent expr_fn" style function
             #[doc = $DOC]
@@ -44,10 +44,7 @@ macro_rules! make_udaf_function {
             pub fn $AGGREGATE_UDF_FN() -> std::sync::Arc<datafusion_expr::AggregateUDF> {
                 [< STATIC_ $UDAF >]
                     .get_or_init(|| {
-
-                        let accumulator = std::sync::Arc::new($ACCUMULATOR);
-                        std::sync::Arc::new(datafusion_expr::AggregateUDF::from(<$UDAF>::new(accumulator)))
-
+                        std::sync::Arc::new(datafusion_expr::AggregateUDF::from(<$UDAF>::default()))
                     })
                     .clone()
             }

datafusion/physical-expr/Cargo.toml

@@ -56,10 +56,10 @@ base64 = { version = "0.22", optional = true }
 blake2 = { version = "^0.10.2", optional = true }
 blake3 = { version = "1.0", optional = true }
 chrono = { workspace = true }
-datafusion-functions-aggregate = { workspace = true }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-execution = { workspace = true }
 datafusion-expr = { workspace = true }
+datafusion-functions-aggregate = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
 half = { workspace = true }
 hashbrown = { version = "0.14", features = ["raw"] }