Alamb/merge resolve

datafusion-examples/examples/dataframe_subquery.rs

@@ -19,7 +19,7 @@ use arrow_schema::DataType;
 use std::sync::Arc;
 
 use datafusion::error::Result;
-use datafusion::functions_aggregate::average::avg;
+use datafusion::functions_aggregate::expr_fn::{avg, max};
 use datafusion::prelude::*;
 use datafusion::test_util::arrow_test_data;
 use datafusion_common::ScalarValue;

datafusion/core/src/dataframe/mod.rs

@@ -48,13 +48,13 @@ use datafusion_common::config::{CsvOptions, FormatOptions, JsonOptions};
 use datafusion_common::{
     plan_err, Column, DFSchema, DataFusionError, ParamValues, SchemaError, UnnestOptions,
 };
-use datafusion_expr::{case, is_null, lit};
-use datafusion_expr::{
-    max, min, utils::COUNT_STAR_EXPANSION, TableProviderFilterPushDown, UNNAMED_TABLE,
+use datafusion_expr::{case, is_null, lit, TableProviderFilterPushDown, UNNAMED_TABLE};
+use datafusion_functions_aggregate::expr_fn::{
+    avg, count, max, median, min, stddev, sum,
 };
-use datafusion_functions_aggregate::expr_fn::{avg, count, median, stddev, sum};
 
 use async_trait::async_trait;
+use datafusion_expr::utils::COUNT_STAR_EXPANSION;
 
 /// Contains options that control how data is
 /// written out from a DataFrame

datafusion/core/src/physical_optimizer/aggregate_statistics.rs

@@ -173,6 +173,23 @@ fn take_optimizable_column_and_table_count(
     None
 }
 
+fn unwrap_min(agg_expr: &dyn AggregateExpr) -> Option<&AggregateFunctionExpr> {
+    if let Some(casted_expr) = agg_expr.as_any().downcast_ref::<AggregateFunctionExpr>() {
+        if casted_expr.fun().name() == "MIN" {
+            return Some(casted_expr);
+        }
+    }
+    None
+}
+
+fn unwrap_max(agg_expr: &dyn AggregateExpr) -> Option<&AggregateFunctionExpr> {
+    if let Some(casted_expr) = agg_expr.as_any().downcast_ref::<AggregateFunctionExpr>() {
+        if casted_expr.fun().name() == "MAX" {
+            return Some(casted_expr);
+        }
+    }
+    None
+}
 /// If this agg_expr is a min that is exactly defined in the statistics, return it.
 fn take_optimizable_min(
     agg_expr: &dyn AggregateExpr,
@@ -182,9 +199,7 @@ fn take_optimizable_min(
         match *num_rows {
             0 => {
                 // MIN/MAX with 0 rows is always null
-                if let Some(casted_expr) =
-                    agg_expr.as_any().downcast_ref::<expressions::Min>()
-                {
+                if let Some(casted_expr) = unwrap_min(agg_expr) {
                     if let Ok(min_data_type) =
                         ScalarValue::try_from(casted_expr.field().unwrap().data_type())
                     {
@@ -194,9 +209,7 @@ fn take_optimizable_min(
             }
             value if value > 0 => {
                 let col_stats = &stats.column_statistics;
-                if let Some(casted_expr) =
-                    agg_expr.as_any().downcast_ref::<expressions::Min>()
-                {
+                if let Some(casted_expr) = unwrap_min(agg_expr) {
                     if casted_expr.expressions().len() == 1 {
                         // TODO optimize with exprs other than Column
                         if let Some(col_expr) = casted_expr.expressions()[0]
@@ -232,9 +245,7 @@ fn take_optimizable_max(
         match *num_rows {
             0 => {
                 // MIN/MAX with 0 rows is always null
-                if let Some(casted_expr) =
-                    agg_expr.as_any().downcast_ref::<expressions::Max>()
-                {
+                if let Some(casted_expr) = unwrap_max(agg_expr) {
                     if let Ok(max_data_type) =
                         ScalarValue::try_from(casted_expr.field().unwrap().data_type())
                     {
@@ -244,9 +255,7 @@ fn take_optimizable_max(
             }
             value if value > 0 => {
                 let col_stats = &stats.column_statistics;
-                if let Some(casted_expr) =
-                    agg_expr.as_any().downcast_ref::<expressions::Max>()
-                {
+                if let Some(casted_expr) = unwrap_max(agg_expr) {
                     if casted_expr.expressions().len() == 1 {
                         // TODO optimize with exprs other than Column
                         if let Some(col_expr) = casted_expr.expressions()[0]

datafusion/core/tests/dataframe/mod.rs

@@ -52,11 +52,11 @@ use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_expr::expr::{GroupingSet, Sort};
 use datafusion_expr::var_provider::{VarProvider, VarType};
 use datafusion_expr::{
-    array_agg, cast, col, exists, expr, in_subquery, lit, max, out_ref_col, placeholder,
+    array_agg, cast, col, exists, expr, in_subquery, lit, out_ref_col, placeholder,
     scalar_subquery, when, wildcard, Expr, ExprSchemable, WindowFrame, WindowFrameBound,
     WindowFrameUnits, WindowFunctionDefinition,
 };
-use datafusion_functions_aggregate::expr_fn::{avg, count, sum};
+use datafusion_functions_aggregate::expr_fn::{avg, count, max, sum};
 
 #[tokio::test]
 async fn test_count_wildcard_on_sort() -> Result<()> {

datafusion/core/tests/fuzz_cases/window_fuzz.rs

@@ -35,10 +35,11 @@ use datafusion_common_runtime::SpawnedTask;
 use datafusion_expr::type_coercion::aggregates::coerce_types;
 use datafusion_expr::type_coercion::functions::data_types_with_aggregate_udf;
 use datafusion_expr::{
-    AggregateFunction, BuiltInWindowFunction, WindowFrame, WindowFrameBound,
-    WindowFrameUnits, WindowFunctionDefinition,
+    BuiltInWindowFunction, WindowFrame, WindowFrameBound, WindowFrameUnits,
+    WindowFunctionDefinition,
 };
 use datafusion_functions_aggregate::count::count_udaf;
+use datafusion_functions_aggregate::min_max::{max_udaf, min_udaf};
 use datafusion_functions_aggregate::sum::sum_udaf;
 use datafusion_physical_expr::expressions::{cast, col, lit};
 use datafusion_physical_expr::{PhysicalExpr, PhysicalSortExpr};
@@ -360,14 +361,14 @@ fn get_random_function(
     window_fn_map.insert(
         "min",
         (
-            WindowFunctionDefinition::AggregateFunction(AggregateFunction::Min),
+            WindowFunctionDefinition::AggregateUDF(min_udaf()),
             vec![arg.clone()],
         ),
     );
     window_fn_map.insert(
         "max",
         (
-            WindowFunctionDefinition::AggregateFunction(AggregateFunction::Max),
+            WindowFunctionDefinition::AggregateUDF(max_udaf()),
             vec![arg.clone()],
         ),
     );

datafusion/expr/src/aggregate_function.rs

@@ -33,10 +33,6 @@ use strum_macros::EnumIter;
 // https://datafusion.apache.org/contributor-guide/index.html#how-to-add-a-new-aggregate-function
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash, EnumIter)]
 pub enum AggregateFunction {
-    /// Minimum
-    Min,
-    /// Maximum
-    Max,
     /// Aggregation into an array
     ArrayAgg,
     /// N'th value in a group according to some ordering
@@ -49,8 +45,6 @@ impl AggregateFunction {
     pub fn name(&self) -> &str {
         use AggregateFunction::*;
         match self {
-            Min => "MIN",
-            Max => "MAX",
             ArrayAgg => "ARRAY_AGG",
             NthValue => "NTH_VALUE",
             Grouping => "GROUPING",
@@ -69,8 +63,6 @@ impl FromStr for AggregateFunction {
     fn from_str(name: &str) -> Result<AggregateFunction> {
         Ok(match name {
             // general
-            "max" => AggregateFunction::Max,
-            "min" => AggregateFunction::Min,
             "array_agg" => AggregateFunction::ArrayAgg,
             "nth_value" => AggregateFunction::NthValue,
             // other
@@ -109,11 +101,6 @@ impl AggregateFunction {
             })?;
 
         match self {
-            AggregateFunction::Max | AggregateFunction::Min => {
-                // For min and max agg function, the returned type is same as input type.
-                // The coerced_data_types is same with input_types.
-                Ok(coerced_data_types[0].clone())
-            }
             AggregateFunction::ArrayAgg => Ok(DataType::List(Arc::new(Field::new(
                 "item",
                 coerced_data_types[0].clone(),
@@ -128,7 +115,6 @@ impl AggregateFunction {
     /// nullability
     pub fn nullable(&self) -> Result<bool> {
         match self {
-            AggregateFunction::Max | AggregateFunction::Min => Ok(true),
             AggregateFunction::ArrayAgg => Ok(false),
             AggregateFunction::Grouping => Ok(true),
             AggregateFunction::NthValue => Ok(true),
@@ -144,18 +130,6 @@ impl AggregateFunction {
             AggregateFunction::Grouping | AggregateFunction::ArrayAgg => {
                 Signature::any(1, Volatility::Immutable)
             }
-            AggregateFunction::Min | AggregateFunction::Max => {
-                let valid = STRINGS
-                    .iter()
-                    .chain(NUMERICS.iter())
-                    .chain(TIMESTAMPS.iter())
-                    .chain(DATES.iter())
-                    .chain(TIMES.iter())
-                    .chain(BINARYS.iter())
-                    .cloned()
-                    .collect::<Vec<_>>();
-                Signature::uniform(1, valid, Volatility::Immutable)
-            }
             AggregateFunction::NthValue => Signature::any(2, Volatility::Immutable),
         }
     }

datafusion/expr/src/expr.rs

@@ -2456,6 +2456,7 @@ mod test {
             "first_value",
             "last_value",
             "nth_value",
+            "avg",
             "min",
             "max",
         ];
@@ -2474,18 +2475,6 @@ mod test {
 
     #[test]
     fn test_find_df_window_function() {
-        assert_eq!(
-            find_df_window_func("max"),
-            Some(WindowFunctionDefinition::AggregateFunction(
-                aggregate_function::AggregateFunction::Max
-            ))
-        );
-        assert_eq!(
-            find_df_window_func("min"),
-            Some(WindowFunctionDefinition::AggregateFunction(
-                aggregate_function::AggregateFunction::Min
-            ))
-        );
         assert_eq!(
             find_df_window_func("cume_dist"),
             Some(WindowFunctionDefinition::BuiltInWindowFunction(

datafusion/expr/src/expr_fn.rs

@@ -147,30 +147,6 @@ pub fn not(expr: Expr) -> Expr {
     expr.not()
 }
 
-/// Create an expression to represent the min() aggregate function
-pub fn min(expr: Expr) -> Expr {
-    Expr::AggregateFunction(AggregateFunction::new(
-        aggregate_function::AggregateFunction::Min,
-        vec![expr],
-        false,
-        None,
-        None,
-        None,
-    ))
-}
-
-/// Create an expression to represent the max() aggregate function
-pub fn max(expr: Expr) -> Expr {
-    Expr::AggregateFunction(AggregateFunction::new(
-        aggregate_function::AggregateFunction::Max,
-        vec![expr],
-        false,
-        None,
-        None,
-        None,
-    ))
-}
-
 /// Create an expression to represent the array_agg() aggregate function
 pub fn array_agg(expr: Expr) -> Expr {
     Expr::AggregateFunction(AggregateFunction::new(

datafusion/expr/src/expr_rewriter/order_by.rs

@@ -156,11 +156,12 @@ mod test {
     use arrow::datatypes::{DataType, Field, Schema};
 
     use crate::{
-        cast, col, lit, logical_plan::builder::LogicalTableSource, min,
-        test::function_stub::avg, try_cast, LogicalPlanBuilder,
+        cast, col, lit, logical_plan::builder::LogicalTableSource, try_cast,
+        LogicalPlanBuilder,
     };
 
     use super::*;
+    use crate::test::function_stub::{avg, min};
 
     #[test]
     fn rewrite_sort_cols_by_agg() {