fix: align CrossJoinExec schema metadata merge with the logical plan

`CrossJoinExec::new` merged schema-level metadata right-biased
(`left.metadata().extend(right)`), while the logical plan and the shared
physical `build_join_schema` are left-biased for inner joins. When both
inputs carry the same metadata key with different values, the cross join's
physical schema diverged from the logical one and tripped the physical
planner's `schema_satisfied_by` check (raised when the cross join feeds an
aggregate), failing with an internal error.

Route `CrossJoinExec::new` through the shared `build_join_schema` helper
(the one #16221 fixed for hash/nested-loop joins but never applied to cross
join) so metadata is merged consistently. Field output is unchanged for
inner joins.

Adds a unit test and a metadata.slt regression over two tables whose
schema metadata conflicts under the same key.

Closes #23434

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: fangchenli

datafusion/physical-plan/src/joins/cross_join.rs

@@ -23,7 +23,7 @@ use std::{sync::Arc, task::Poll};
 use super::utils::{
     BatchSplitter, BatchTransformer, BuildProbeJoinMetrics, NoopBatchTransformer,
     OnceAsync, OnceFut, StatefulStreamResult, adjust_right_output_partitioning,
-    reorder_output_after_swap,
+    build_join_schema, reorder_output_after_swap,
 };
 use crate::execution_plan::{EmissionType, boundedness_from_children};
 use crate::metrics::{ExecutionPlanMetricsSet, MetricsSet};
@@ -41,7 +41,7 @@ use crate::{
 
 use arrow::array::{RecordBatch, RecordBatchOptions};
 use arrow::compute::concat_batches;
-use arrow::datatypes::{Fields, Schema, SchemaRef};
+use arrow::datatypes::{Schema, SchemaRef};
 use datafusion_common::stats::Precision;
 use datafusion_common::{
     JoinType, Result, ScalarValue, assert_eq_or_internal_err, internal_err,
@@ -102,23 +102,11 @@ pub struct CrossJoinExec {
 impl CrossJoinExec {
     /// Create a new [CrossJoinExec].
     pub fn new(left: Arc<dyn ExecutionPlan>, right: Arc<dyn ExecutionPlan>) -> Self {
-        // left then right
-        let (all_columns, metadata) = {
-            let left_schema = left.schema();
-            let right_schema = right.schema();
-            let left_fields = left_schema.fields().iter();
-            let right_fields = right_schema.fields().iter();
-
-            let mut metadata = left_schema.metadata().clone();
-            metadata.extend(right_schema.metadata().clone());
-
-            (
-                left_fields.chain(right_fields).cloned().collect::<Fields>(),
-                metadata,
-            )
-        };
-
-        let schema = Arc::new(Schema::new(all_columns).with_metadata(metadata));
+        // Use the shared helper (inner join) so metadata merges the same way as
+        // the logical plan; merging it here independently let schemas diverge.
+        let (schema, _) =
+            build_join_schema(&left.schema(), &right.schema(), &JoinType::Inner);
+        let schema = Arc::new(schema);
         let cache = Self::compute_properties(&left, &right, Arc::clone(&schema)).unwrap();
 
         CrossJoinExec {

datafusion/sqllogictest/src/test_context.rs

@@ -521,6 +521,25 @@ pub async fn register_metadata_tables(ctx: &SessionContext) {
     .unwrap();
 
     ctx.register_batch("table_with_metadata", batch).unwrap();
+
+    // Same `metadata_key` as `table_with_metadata` but a different value, so a
+    // cross join of the two exercises the schema-metadata merge (#23434).
+    let alt_id =
+        Field::new("id", DataType::Int32, true).with_metadata(HashMap::from([(
+            String::from("metadata_key"),
+            String::from("the alt id field"),
+        )]));
+    let alt_schema = Schema::new(vec![alt_id]).with_metadata(HashMap::from([(
+        String::from("metadata_key"),
+        String::from("the other schema"),
+    )]));
+    let alt_batch = RecordBatch::try_new(
+        Arc::new(alt_schema),
+        vec![Arc::new(Int32Array::from(vec![Some(10), Some(20)])) as _],
+    )
+    .unwrap();
+    ctx.register_batch("table_with_metadata_alt", alt_batch)
+        .unwrap();
 }
 
 /// Create a UDF function named "example". See the `sample_udf.rs` example

datafusion/sqllogictest/test_files/metadata.slt

@@ -124,6 +124,16 @@ FROM
 ----
 6
 
+# Regression test: cross join over two tables with conflicting schema metadata,
+# feeding an aggregate, used to fail the physical planner's schema check.
+# count(DISTINCT ...) keeps a real cross join (plain count folds from stats).
+# See https://github.com/apache/datafusion/issues/23434
+query I
+SELECT count(DISTINCT "l"."id")
+FROM "table_with_metadata" AS "l", "table_with_metadata_alt" AS "r";
+----
+2
+
 # Regression test: missing field metadata, from the NULL field on the left side of the union
 query ITT
 (SELECT id, NULL::string as name, l_name FROM "table_with_metadata")