deps: Upgrade to DataFusion 54.0.0

native/Cargo.toml

@@ -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/core/Cargo.toml

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

native/core/src/debug/debug_batch_stream.rs

@@ -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/execution/expressions/arithmetic.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/list_positions.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/strings.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/subquery.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/memory_pools/fair_pool.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/logging_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()",

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

@@ -16,7 +16,7 @@
 // under the License.
 
 use std::{
-    fmt::{Debug, Formatter, Result as FmtResult},
+    fmt::{Debug, Display, Formatter, Result as FmtResult},
     sync::{
         atomic::{AtomicUsize, Ordering::Relaxed},
         Arc,
@@ -90,10 +90,24 @@ impl Drop for CometUnifiedMemoryPool {
     }
 }
 
+impl Display for CometUnifiedMemoryPool {
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        write!(
+            f,
+            "CometUnifiedMemoryPool(used={})",
+            self.used.load(Relaxed)
+        )
+    }
+}
+
 unsafe impl Send for CometUnifiedMemoryPool {}
 unsafe impl Sync for CometUnifiedMemoryPool {}
 
 impl MemoryPool for CometUnifiedMemoryPool {
+    fn name(&self) -> &str {
+        "CometUnifiedMemoryPool"
+    }
+
     fn grow(&self, reservation: &MemoryReservation, additional: usize) {
         self.try_grow(reservation, additional).unwrap();
     }

native/core/src/execution/merge_as_partial.rs

@@ -26,7 +26,6 @@
 //! outputs state) but redirects `update_batch` calls to `merge_batch`, giving merge
 //! semantics with state output.
 
-use std::any::Any;
 use std::fmt::Debug;
 use std::hash::{Hash, Hasher};
 
@@ -100,10 +99,6 @@ impl MergeAsPartialUDF {
 }
 
 impl AggregateUDFImpl for MergeAsPartialUDF {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }

native/core/src/execution/metrics/utils.rs

@@ -71,14 +71,19 @@ pub(crate) fn to_native_metric_node(
         children: Vec::with_capacity(children.len()),
     };
 
-    if let Some(metrics) = node_metrics {
-        for m in metrics.iter() {
-            let value = m.value();
-            native_metric_node
-                .metrics
-                .insert(value.name().to_string(), value.as_usize() as i64);
-        }
-    }
+    // Aggregate metrics by name using DataFusion's aggregate_by_name(), which
+    // correctly handles duplicate metric names (e.g. BaselineMetrics registered
+    // by both FileStream and ParquetMorselizer on the same ExecutionPlanMetricsSet).
+    // The additional_native_plans branch below already does this.
+    node_metrics
+        .unwrap_or_default()
+        .aggregate_by_name()
+        .iter()
+        .map(|m| m.value())
+        .map(|m| (m.name(), m.as_usize() as i64))
+        .for_each(|(name, value)| {
+            native_metric_node.metrics.insert(name.to_string(), value);
+        });
 
     for child_plan in children {
         let child_node = to_native_metric_node(child_plan)?;

native/core/src/execution/operators/expand.rs

@@ -29,7 +29,6 @@ use datafusion::{
 };
 use futures::{Stream, StreamExt};
 use std::{
-    any::Any,
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
@@ -91,10 +90,6 @@ impl DisplayAs for ExpandExec {
 }
 
 impl ExecutionPlan for ExpandExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }

native/core/src/execution/operators/iceberg_scan.rs

@@ -17,7 +17,6 @@
 
 //! Native Iceberg table scan operator using iceberg-rust
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::fmt;
 use std::pin::Pin;
@@ -121,10 +120,6 @@ impl ExecutionPlan for IcebergScanExec {
         "IcebergScanExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.output_schema)
     }

native/core/src/execution/operators/parquet_writer.rs

@@ -18,7 +18,6 @@
 //! Parquet writer operator for writing RecordBatches to Parquet files
 
 use std::{
-    any::Any,
     collections::HashMap,
     fmt,
     fmt::{Debug, Formatter},
@@ -404,10 +403,6 @@ impl DisplayAs for ParquetWriterExec {
 
 #[async_trait]
 impl ExecutionPlan for ParquetWriterExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "ParquetWriterExec"
     }

native/core/src/execution/operators/scan.rs

@@ -43,7 +43,6 @@ use itertools::Itertools;
 use jni::objects::{Global, JObject, JValue};
 use std::rc::Rc;
 use std::{
-    any::Any,
     pin::Pin,
     sync::{Arc, Mutex},
     task::{Context, Poll},
@@ -286,10 +285,6 @@ fn schema_from_data_types(data_types: &[DataType]) -> SchemaRef {
 }
 
 impl ExecutionPlan for ScanExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }

native/core/src/execution/operators/shuffle_scan.rs

@@ -37,7 +37,6 @@ use datafusion::{
 use futures::Stream;
 use jni::objects::{Global, JByteBuffer, JObject};
 use std::{
-    any::Any,
     pin::Pin,
     sync::{Arc, Mutex},
     task::{Context, Poll},
@@ -221,10 +220,6 @@ fn schema_from_data_types(data_types: &[DataType]) -> SchemaRef {
 }
 
 impl ExecutionPlan for ShuffleScanExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }