diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs index d7ff07d00f586..a6db7892e75f5 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 { @@ -696,11 +684,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..44e21964876ef 100644 --- a/datafusion/sqllogictest/test_files/metadata.slt +++ b/datafusion/sqllogictest/test_files/metadata.slt @@ -124,6 +124,30 @@ 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 + +# 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")