chore: bump DataFusion to rev c6f0d3c

Author: huaxingao

native/Cargo.lock

@@ -39,13 +39,13 @@ arrow-buffer = { version = "52.2.0" }
 arrow-data = { version = "52.2.0" }
 arrow-schema = { version = "52.2.0" }
 parquet = { version = "52.2.0", default-features = false, features = ["experimental"] }
-datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e" }
-datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", features = ["unicode_expressions", "crypto_expressions"] }
-datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", features = ["crypto_expressions"] }
-datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", default-features = false }
-datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", default-features = false }
-datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", default-features = false }
-datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "35c2e7e", default-features = false }
+datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c" }
+datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", features = ["unicode_expressions", "crypto_expressions"] }
+datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", features = ["crypto_expressions"] }
+datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", default-features = false }
+datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", default-features = false }
+datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", default-features = false }
+datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "c6f0d3c", default-features = false }
 datafusion-comet-spark-expr = { path = "spark-expr", version = "0.2.0" }
 datafusion-comet-proto = { path = "proto", version = "0.2.0" }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

native/Cargo.toml

@@ -20,6 +20,8 @@
 use arrow_schema::{DataType, Field, Schema, TimeUnit, DECIMAL128_MAX_PRECISION};
 use datafusion::functions_aggregate::bit_and_or_xor::{bit_and_udaf, bit_or_udaf, bit_xor_udaf};
 use datafusion::functions_aggregate::count::count_udaf;
+use datafusion::functions_aggregate::min_max::max_udaf;
+use datafusion::functions_aggregate::min_max::min_udaf;
 use datafusion::functions_aggregate::sum::sum_udaf;
 use datafusion::physical_plan::windows::BoundedWindowAggExec;
 use datafusion::physical_plan::InputOrderMode;
@@ -33,7 +35,7 @@ use datafusion::{
         execution_props::ExecutionProps,
         expressions::{
             in_list, BinaryExpr, CaseExpr, CastExpr, Column, IsNotNullExpr, IsNullExpr,
-            Literal as DataFusionLiteral, Max, Min, NotExpr,
+            Literal as DataFusionLiteral, NotExpr,
         },
         AggregateExpr, PhysicalExpr, PhysicalSortExpr, ScalarFunctionExpr,
     },
@@ -1254,14 +1256,36 @@ impl PhysicalPlanner {
                 .map_err(|e| ExecutionError::DataFusionError(e.to_string()))
             }
             AggExprStruct::Min(expr) => {
-                let child = self.create_expr(expr.child.as_ref().unwrap(), schema)?;
+                let child = self.create_expr(expr.child.as_ref().unwrap(), schema.clone())?;
                 let datatype = to_arrow_datatype(expr.datatype.as_ref().unwrap());
-                Ok(Arc::new(Min::new(child, "min", datatype)))
+                let child = Arc::new(CastExpr::new(child, datatype.clone(), None));
+                create_aggregate_expr(
+                    &min_udaf(),
+                    &[child],
+                    &[],
+                    &[],
+                    &[],
+                    schema.as_ref(),
+                    "min",
+                    false,
+                    false,
+                ).map_err(|e| ExecutionError::DataFusionError(e.to_string()))
             }
             AggExprStruct::Max(expr) => {
-                let child = self.create_expr(expr.child.as_ref().unwrap(), schema)?;
+                let child = self.create_expr(expr.child.as_ref().unwrap(), schema.clone())?;
                 let datatype = to_arrow_datatype(expr.datatype.as_ref().unwrap());
-                Ok(Arc::new(Max::new(child, "max", datatype)))
+                let child = Arc::new(CastExpr::new(child, datatype.clone(), None));
+                create_aggregate_expr(
+                    &max_udaf(),
+                    &[child],
+                    &[],
+                    &[],
+                    &[],
+                    schema.as_ref(),
+                    "max",
+                    false,
+                    false,
+                ).map_err(|e| ExecutionError::DataFusionError(e.to_string()))
             }
             AggExprStruct::Sum(expr) => {
                 let child = self.create_expr(expr.child.as_ref().unwrap(), schema.clone())?;