From b20c6c6bd0e09a4c3fa6317b89162685f134a930 Mon Sep 17 00:00:00 2001 From: Fangchen Li Date: Thu, 9 Jul 2026 19:45:39 -0700 Subject: [PATCH 1/2] 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) --- .../physical-plan/src/joins/cross_join.rs | 58 ++++++++++++------- datafusion/sqllogictest/src/test_context.rs | 19 ++++++ .../sqllogictest/test_files/metadata.slt | 10 ++++ 3 files changed, 67 insertions(+), 20 deletions(-) diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs index 79295ba2fb556..8d8ff9b45e4c1 100644 --- a/datafusion/physical-plan/src/joins/cross_join.rs +++ b/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, right: Arc) -> 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::(), - 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 { @@ -692,11 +680,41 @@ impl CrossJoinStream { mod tests { use super::*; use crate::common; - use crate::test::{assert_join_metrics, build_table_scan_i32}; + use crate::test::{TestMemoryExec, assert_join_metrics, build_table_scan_i32}; + use arrow::datatypes::{DataType, Field}; use datafusion_common::{assert_contains, test_util::batches_to_sort_string}; use datafusion_execution::runtime_env::RuntimeEnvBuilder; use insta::assert_snapshot; + use std::collections::HashMap; + + // On a conflicting schema-metadata key, the cross join keeps the LEFT value, + // matching `build_join_schema` and the logical plan (mirrors the per-join-type + // metadata tests #16221 added in joins/utils.rs and the logical builder.rs). + #[test] + fn cross_join_schema_metadata_is_left_biased() { + let input = |field: &str, meta_value: &str| { + let schema = Arc::new( + Schema::new(vec![Field::new(field, DataType::Int32, false)]) + .with_metadata(HashMap::from([( + String::from("metadata_key"), + String::from(meta_value), + )])), + ); + TestMemoryExec::try_new_exec(&[vec![]], schema, None).unwrap() + }; + + let join = + CrossJoinExec::new(input("a", "left value"), input("b", "right value")); + + assert_eq!( + join.schema() + .metadata() + .get("metadata_key") + .map(String::as_str), + Some("left value"), + ); + } async fn join_collect( left: Arc, diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs index e0aaa91ef6369..24093245330bc 100644 --- a/datafusion/sqllogictest/src/test_context.rs +++ b/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 diff --git a/datafusion/sqllogictest/test_files/metadata.slt b/datafusion/sqllogictest/test_files/metadata.slt index 3e2a503e6b3fc..ac69e13524e54 100644 --- a/datafusion/sqllogictest/test_files/metadata.slt +++ b/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") From 6f4e74bd8817eb7e9b41678c544bc41fcd33b284 Mon Sep 17 00:00:00 2001 From: Fangchen Li Date: Fri, 10 Jul 2026 12:29:41 -0700 Subject: [PATCH 2/2] test: pin cross join metadata fix with join reordering disabled Add a metadata.slt variant that runs the cross join regression with `join_reordering = false`, so the fix is pinned on the initial (non-swapped) cross join plan and stays insulated from the separate pre-existing swap-path divergence. --- datafusion/sqllogictest/test_files/metadata.slt | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/datafusion/sqllogictest/test_files/metadata.slt b/datafusion/sqllogictest/test_files/metadata.slt index ac69e13524e54..44e21964876ef 100644 --- a/datafusion/sqllogictest/test_files/metadata.slt +++ b/datafusion/sqllogictest/test_files/metadata.slt @@ -134,6 +134,20 @@ FROM "table_with_metadata" AS "l", "table_with_metadata_alt" AS "r"; ---- 2 +# Same case with join reordering disabled, so the fix is pinned on the initial +# (non-swapped) cross join plan. The swap path has a separate pre-existing bug. +statement ok +set datafusion.optimizer.join_reordering = false; + +query I +SELECT count(DISTINCT "l"."id") +FROM "table_with_metadata" AS "l", "table_with_metadata_alt" AS "r"; +---- +2 + +statement ok +set datafusion.optimizer.join_reordering = true; + # 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")