chore: Upgrade to latest DataFusion revision (apache#909)

* update dependency version

* update avg

* update avg_decimal

* update sum_decimal

* variance

* stddev

* covariance

* correlation

* save progress

* code compiles

* clippy

* remove duplicate of down_cast_any_ref function

* remove duplicate of down_cast_any_ref function

* machete

* bump DF version again and use StatsType from DataFusion

* implement groups accumulator for stddev and variance

* refactor

* fmt

* revert group accumulator

Author: andygrove

Cargo.toml

@@ -35,7 +35,6 @@ datafusion = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-expr = { workspace = true }
 datafusion-physical-expr = { workspace = true }
-datafusion-physical-plan = { workspace = true }
 chrono-tz = { workspace = true }
 num = { workspace = true }
 regex = { workspace = true }

src/cast.rs

@@ -50,14 +50,15 @@ use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr::PhysicalExpr;
 
 use chrono::{NaiveDate, NaiveDateTime, TimeZone, Timelike};
+use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
 use num::{
     cast::AsPrimitive, integer::div_floor, traits::CheckedNeg, CheckedSub, Integer, Num,
     ToPrimitive,
 };
 use regex::Regex;
 
 use crate::timezone;
-use crate::utils::{array_with_timezone, down_cast_any_ref};
+use crate::utils::array_with_timezone;
 
 use crate::{EvalMode, SparkError, SparkResult};
 

src/if_expr.rs

@@ -26,11 +26,10 @@ use arrow::{
     record_batch::RecordBatch,
 };
 use datafusion::logical_expr::ColumnarValue;
+use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
 use datafusion_common::Result;
 use datafusion_physical_expr::{expressions::CaseExpr, PhysicalExpr};
 
-use crate::utils::down_cast_any_ref;
-
 /// IfExpr is a wrapper around CaseExpr, because `IF(a, b, c)` is semantically equivalent to
 /// `CASE WHEN a THEN b ELSE c END`.
 #[derive(Debug, Hash)]

src/regexp.rs

@@ -15,13 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::utils::down_cast_any_ref;
 use crate::SparkError;
 use arrow::compute::take;
 use arrow_array::builder::BooleanBuilder;
 use arrow_array::types::Int32Type;
 use arrow_array::{Array, BooleanArray, DictionaryArray, RecordBatch, StringArray};
 use arrow_schema::{DataType, Schema};
+use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
 use datafusion_common::{internal_err, Result};
 use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr::PhysicalExpr;

src/structs.rs

@@ -19,6 +19,7 @@ use arrow::record_batch::RecordBatch;
 use arrow_array::{Array, StructArray};
 use arrow_schema::{DataType, Field, Schema};
 use datafusion::logical_expr::ColumnarValue;
+use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
 use datafusion_common::{DataFusionError, Result as DataFusionResult, ScalarValue};
 use datafusion_physical_expr::PhysicalExpr;
 use std::{
@@ -28,8 +29,6 @@ use std::{
     sync::Arc,
 };
 
-use crate::utils::down_cast_any_ref;
-
 #[derive(Debug, Hash)]
 pub struct CreateNamedStruct {
     values: Vec<Arc<dyn PhysicalExpr>>,

src/temporal.rs

@@ -28,10 +28,11 @@ use arrow::{
 };
 use arrow_schema::{DataType, Schema, TimeUnit::Microsecond};
 use datafusion::logical_expr::ColumnarValue;
+use datafusion::physical_expr_common::physical_expr::down_cast_any_ref;
 use datafusion_common::{DataFusionError, ScalarValue::Utf8};
 use datafusion_physical_expr::PhysicalExpr;
 
-use crate::utils::{array_with_timezone, down_cast_any_ref};
+use crate::utils::array_with_timezone;
 
 use crate::kernels::temporal::{
     date_trunc_array_fmt_dyn, date_trunc_dyn, timestamp_trunc_array_fmt_dyn, timestamp_trunc_dyn,

src/utils.rs

@@ -20,7 +20,6 @@ use arrow_array::{
     types::{Int32Type, TimestampMicrosecondType},
 };
 use arrow_schema::{ArrowError, DataType};
-use std::any::Any;
 use std::sync::Arc;
 
 use crate::timezone::Tz;
@@ -30,23 +29,6 @@ use arrow::{
 };
 use chrono::{DateTime, Offset, TimeZone};
 
-use datafusion_physical_plan::PhysicalExpr;
-
-/// A utility function from DataFusion. It is not exposed by DataFusion.
-pub fn down_cast_any_ref(any: &dyn Any) -> &dyn Any {
-    if any.is::<Arc<dyn PhysicalExpr>>() {
-        any.downcast_ref::<Arc<dyn PhysicalExpr>>()
-            .unwrap()
-            .as_any()
-    } else if any.is::<Box<dyn PhysicalExpr>>() {
-        any.downcast_ref::<Box<dyn PhysicalExpr>>()
-            .unwrap()
-            .as_any()
-    } else {
-        any
-    }
-}
-
 /// Preprocesses input arrays to add timezone information from Spark to Arrow array datatype or
 /// to apply timezone offset.
 //