deps: Upgrade to DataFusion 54.0.0 (#4062)

Co-authored-by: Matt Butrovich <mbutrovich@gmail.com>
Co-authored-by: Matt Butrovich <mbutrovich@users.noreply.github.com>
Co-authored-by: comphead <comphead@ukr.net>
Co-authored-by: Oleks V <comphead@users.noreply.github.com>

Author: 3 people

native/Cargo.lock

@@ -38,10 +38,10 @@ arrow = { version = "58.3.0", features = ["prettyprint", "ffi", "chrono-tz"] }
 async-trait = { version = "0.1" }
 bytes = { version = "1.11.1" }
 parquet = { version = "58.3.0", default-features = false, features = ["experimental"] }
-datafusion = { version = "53.1.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
-datafusion-datasource = { version = "53.1.0" }
-datafusion-physical-expr-adapter = { version = "53.1.0" }
-datafusion-spark = { version = "53.1.0", features = ["core"] }
+datafusion = { version = "54.0.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
+datafusion-datasource = { version = "54.0.0" }
+datafusion-physical-expr-adapter = { version = "54.0.0" }
+datafusion-spark = { version = "54.0.0", features = ["core"] }
 datafusion-comet-spark-expr = { path = "spark-expr" }
 datafusion-comet-common = { path = "common" }
 datafusion-comet-jni-bridge = { path = "jni-bridge" }

native/Cargo.toml

@@ -92,7 +92,7 @@ jni = { version = "0.22.4", features = ["invocation"] }
 lazy_static = "1.4"
 assertables = "10"
 hex = "0.4.3"
-datafusion-functions-nested = { version = "53.1.0" }
+datafusion-functions-nested = { version = "54.0.0" }
 
 [features]
 backtrace = ["datafusion/backtrace"]

native/core/Cargo.toml

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -82,9 +81,6 @@ impl datafusion::physical_plan::ExecutionPlan for DebugExecutionDataStream {
     fn name(&self) -> &str {
         "DebugExecutionDataStream"
     }
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
     fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         self.inner.properties()
     }
@@ -155,9 +151,6 @@ impl Hash for DebugExecutionDataPhyExpr {
 }
 
 impl PhysicalExpr for DebugExecutionDataPhyExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
     fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
         self.inner.data_type(input_schema)
     }

native/core/src/debug/debug_batch_stream.rs

@@ -77,10 +77,6 @@ impl Hash for CheckedBinaryExpr {
 }
 
 impl PhysicalExpr for CheckedBinaryExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         self.child.fmt_sql(f)
     }

native/core/src/execution/expressions/arithmetic.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -68,10 +67,6 @@ impl Hash for ListPositionsExpr {
 }
 
 impl PhysicalExpr for ListPositionsExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }

native/core/src/execution/expressions/list_positions.rs

@@ -91,7 +91,7 @@ impl ExpressionBuilder for RlikeBuilder {
         let left = planner.create_expr(expr.left.as_ref().unwrap(), Arc::clone(&input_schema))?;
         let right = planner.create_expr(expr.right.as_ref().unwrap(), input_schema)?;
 
-        match right.as_any().downcast_ref::<Literal>().unwrap().value() {
+        match right.downcast_ref::<Literal>().unwrap().value() {
             ScalarValue::Utf8(Some(pattern)) => Ok(Arc::new(RLike::try_new(left, pattern)?)),
             _ => Err(ExecutionError::GeneralError(
                 "RLike only supports scalar patterns".to_string(),

native/core/src/execution/expressions/strings.rs

@@ -29,7 +29,6 @@ use jni::{
     sys::{jboolean, jbyte, jint, jlong, jshort},
 };
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -63,10 +62,6 @@ impl Display for Subquery {
 }
 
 impl PhysicalExpr for Subquery {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }

native/core/src/execution/expressions/subquery.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use std::{
-    fmt::{Debug, Formatter, Result as FmtResult},
+    fmt::{Debug, Display, Formatter, Result as FmtResult},
     sync::Arc,
 };
 
@@ -83,10 +83,25 @@ impl CometFairMemoryPool {
     }
 }
 
+impl Display for CometFairMemoryPool {
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        let state = self.state.lock();
+        write!(
+            f,
+            "CometFairMemoryPool(pool_size={}, used={}, num={})",
+            self.pool_size, state.used, state.num
+        )
+    }
+}
+
 unsafe impl Send for CometFairMemoryPool {}
 unsafe impl Sync for CometFairMemoryPool {}
 
 impl MemoryPool for CometFairMemoryPool {
+    fn name(&self) -> &str {
+        "CometFairMemoryPool"
+    }
+
     fn register(&self, _: &MemoryConsumer) {
         let mut state = self.state.lock();
         state.num = state

native/core/src/execution/memory_pools/fair_pool.rs

@@ -19,6 +19,7 @@ use datafusion::execution::memory_pool::{
     MemoryConsumer, MemoryLimit, MemoryPool, MemoryReservation,
 };
 use log::{info, warn};
+use std::fmt;
 use std::sync::Arc;
 
 #[derive(Debug)]
@@ -27,6 +28,12 @@ pub(crate) struct LoggingMemoryPool {
     pool: Arc<dyn MemoryPool>,
 }
 
+impl fmt::Display for LoggingMemoryPool {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "LoggingMemoryPool(task={})", self.task_attempt_id)
+    }
+}
+
 impl LoggingMemoryPool {
     pub fn new(task_attempt_id: u64, pool: Arc<dyn MemoryPool>) -> Self {
         Self {
@@ -37,6 +44,10 @@ impl LoggingMemoryPool {
 }
 
 impl MemoryPool for LoggingMemoryPool {
+    fn name(&self) -> &str {
+        "LoggingMemoryPool"
+    }
+
     fn register(&self, consumer: &MemoryConsumer) {
         info!(
             "[Task {}] MemoryPool[{}].register()",