Merge branch 'datafusion-54' into df54

# Conflicts:
#	native/Cargo.lock
#	native/Cargo.toml
#	native/core/src/parquet/schema_adapter.rs
#	native/core/src/parquet/util/test_common/page_util.rs
#	native/spark-expr/src/agg_funcs/correlation.rs
#	native/spark-expr/src/agg_funcs/covariance.rs
#	native/spark-expr/src/agg_funcs/variance.rs

Author: comphead

native/core/src/debug/debug_batch_stream.rs

@@ -15,13 +15,13 @@
 // 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;
 
 use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Schema};
+use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::Result;
 use datafusion::execution::SendableRecordBatchStream;
 use datafusion::logical_expr::ColumnarValue;
@@ -82,15 +82,18 @@ 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()
     }
     fn children(&self) -> Vec<&Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
         vec![&self.inner]
     }
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&dyn PhysicalExpr) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn datafusion::physical_plan::ExecutionPlan>>,
@@ -155,9 +158,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/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/spark-expr/src/array_funcs/array_position.rs

@@ -28,7 +28,6 @@ use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
 use num::Float;
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark array_position() function that returns the 1-based position of an element in an array.
@@ -313,10 +312,6 @@ impl SparkArrayPositionFunc {
 }
 
 impl ScalarUDFImpl for SparkArrayPositionFunc {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "spark_array_position"
     }

native/spark-expr/src/datetime_funcs/seconds_to_timestamp.rs

@@ -24,7 +24,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result, Sca
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 const MICROS_PER_SECOND: i64 = 1_000_000;
@@ -61,10 +60,6 @@ impl Default for SparkSecondsToTimestamp {
 }
 
 impl ScalarUDFImpl for SparkSecondsToTimestamp {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "seconds_to_timestamp"
     }

native/spark-expr/src/jvm_udf/mod.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;
@@ -107,10 +106,6 @@ impl PartialEq for JvmScalarUdfExpr {
 impl Eq for JvmScalarUdfExpr {}
 
 impl PhysicalExpr for JvmScalarUdfExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

@@ -657,6 +657,42 @@ class CometJoinSuite extends CometTestBase {
     }
   }
 
+  // Reproducer for SPARK-43113: full outer SMJ with a join filter that references
+  // a nullable column should not match when the filter evaluates to NULL.
+  test("SPARK-43113: Full outer SMJ with NULL in join filter") {
+    withTempView("l", "r") {
+      // testData2: (a, b) — all non-null
+      Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2))
+        .toDF("a", "b")
+        .createOrReplaceTempView("l")
+
+      // testData3: (a, b) — b is nullable
+      Seq((1, None), (2, Some(2)))
+        .toDF("a", "b")
+        .createOrReplaceTempView("r")
+
+      val query =
+        """select /*+ MERGE(r) */ *
+          |from l
+          |full outer join r
+          |on l.a = r.a
+          |and l.b < (r.b + 1)
+          |and l.b < (r.a + 1)""".stripMargin
+
+      val expected = Seq(
+        (Some(1), Some(1), None, None),
+        (Some(1), Some(2), None, None),
+        (None, None, Some(1), None),
+        (Some(2), Some(1), Some(2), Some(2)),
+        (Some(2), Some(2), Some(2), Some(2)),
+        (Some(3), Some(1), None, None),
+        (Some(3), Some(2), None, None)).toDF("a", "b", "a", "b")
+
+      val df = sql(query)
+      checkAnswer(df, expected)
+    }
+  }
+
   test("Broadcast exchange respects AQE shuffle partition coalescing") {
     // When a shuffle feeds into a broadcast exchange, AQE may coalesce the shuffle
     // partitions. The broadcast collect should execute through the AQEShuffleReadExec