Update to last DataFusion commit

native/Cargo.toml

@@ -39,14 +39,14 @@ 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/viirya/arrow-datafusion.git", rev = "f98693e" }
-datafusion = { default-features = false, git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", features = ["unicode_expressions", "crypto_expressions"] }
-datafusion-functions = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", features = ["crypto_expressions"] }
-datafusion-functions-nested = { git = "https://github.com/viirya/arrow-datafusion.git",  rev = "f98693e", default-features = false }
-datafusion-expr = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", default-features = false }
-datafusion-execution = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", default-features = false }
-datafusion-physical-plan = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", default-features = false }
-datafusion-physical-expr = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e", default-features = false }
+datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "dff590b" }
+datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "dff590b", features = ["unicode_expressions", "crypto_expressions"] }
+datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "dff590b", features = ["crypto_expressions"] }
+datafusion-functions-nested = { git = "https://github.com/apache/datafusion.git",  rev = "dff590b", default-features = false }
+datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "dff590b", default-features = false }
+datafusion-execution = { git = "https://github.com/apache/datafusion.git", rev = "dff590b", default-features = false }
+datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "dff590b", default-features = false }
+datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "dff590b", 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/core/src/execution/datafusion/expressions/avg.rs

@@ -122,6 +122,10 @@ impl AggregateExpr for Avg {
             ),
         }
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Float64(None))
+    }
 }
 
 impl PartialEq<dyn Any> for Avg {

native/core/src/execution/datafusion/expressions/avg_decimal.rs

@@ -142,6 +142,22 @@ impl AggregateExpr for AvgDecimal {
             ),
         }
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        match &self.result_data_type {
+            Decimal128(target_precision, target_scale) => {
+                Ok(make_decimal128(
+                    None,
+                    *target_precision,
+                    *target_scale,
+                ))
+            }
+            _ => not_impl_err!(
+                "The result_data_type of AvgDecimal should be Decimal128 but got{}",
+                self.result_data_type
+            ),
+        }
+    }
 }
 
 impl PartialEq<dyn Any> for AvgDecimal {
@@ -211,6 +227,10 @@ impl AvgDecimalAccumulator {
     }
 }
 
+fn make_decimal128(value: Option<i128>, precision: u8, scale: i8) -> ScalarValue {
+    ScalarValue::Decimal128(value, precision, scale)
+}
+
 impl Accumulator for AvgDecimalAccumulator {
     fn state(&mut self) -> Result<Vec<ScalarValue>> {
         Ok(vec![
@@ -265,10 +285,6 @@ impl Accumulator for AvgDecimalAccumulator {
     }
 
     fn evaluate(&mut self) -> Result<ScalarValue> {
-        fn make_decimal128(value: Option<i128>, precision: u8, scale: i8) -> ScalarValue {
-            ScalarValue::Decimal128(value, precision, scale)
-        }
-
         let scaler = 10_i128.pow(self.target_scale.saturating_sub(self.sum_scale) as u32);
         let target_min = MIN_DECIMAL_FOR_EACH_PRECISION[self.target_precision as usize - 1];
         let target_max = MAX_DECIMAL_FOR_EACH_PRECISION[self.target_precision as usize - 1];

native/core/src/execution/datafusion/expressions/correlation.rs

@@ -121,6 +121,10 @@ impl AggregateExpr for Correlation {
     fn name(&self) -> &str {
         &self.name
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Float64(None))
+    }
 }
 
 impl PartialEq<dyn Any> for Correlation {

native/core/src/execution/datafusion/expressions/covariance.rs

@@ -118,6 +118,10 @@ impl AggregateExpr for Covariance {
     fn name(&self) -> &str {
         &self.name
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Float64(None))
+    }
 }
 
 impl PartialEq<dyn Any> for Covariance {

native/core/src/execution/datafusion/expressions/stddev.rs

@@ -108,6 +108,10 @@ impl AggregateExpr for Stddev {
     fn name(&self) -> &str {
         &self.name
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Float64(None))
+    }
 }
 
 impl PartialEq<dyn Any> for Stddev {

native/core/src/execution/datafusion/expressions/sum_decimal.rs

@@ -112,6 +112,13 @@ impl AggregateExpr for SumDecimal {
             self.scale,
         )))
     }
+
+    fn default_value(&self, _data_type: &DataType) -> DFResult<ScalarValue> {
+        ScalarValue::new_primitive::<Decimal128Type>(
+            None,
+            &DataType::Decimal128(self.precision, self.scale),
+        )
+    }
 }
 
 impl PartialEq<dyn Any> for SumDecimal {

native/core/src/execution/datafusion/expressions/variance.rs

@@ -106,6 +106,10 @@ impl AggregateExpr for Variance {
     fn name(&self) -> &str {
         &self.name
     }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Float64(None))
+    }
 }
 
 impl PartialEq<dyn Any> for Variance {