From 40b5410a81991dd9f102933f65a21187bd830ccd Mon Sep 17 00:00:00 2001 From: Michael Kleen Date: Wed, 29 Jul 2026 12:27:09 +0200 Subject: [PATCH 1/2] Revert "fix: Preserve metadata when a cross-join is swapped (#23605)" This reverts commit d75742edbfc4fb10fea20145b36ab776287276de. --- .../physical-plan/src/joins/cross_join.rs | 54 ++----------------- 1 file changed, 3 insertions(+), 51 deletions(-) diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs index 1a631aac980ab..16155aaafdd9c 100644 --- a/datafusion/physical-plan/src/joins/cross_join.rs +++ b/datafusion/physical-plan/src/joins/cross_join.rs @@ -186,32 +186,8 @@ impl CrossJoinExec { /// operators on the join's children. Check [`super::HashJoinExec::swap_inputs`] /// for more details. pub fn swap_inputs(&self) -> Result> { - // Rebuild schema with columns from right to left, preserve existing metadata - let new_columns = self - .right - .schema() - .fields - .iter() - .chain(self.left.schema().fields.iter()) - .cloned() - .collect::(); - - let new_schema = Arc::new( - Schema::new(new_columns).with_metadata(self.schema.metadata.clone()), - ); - - let new_cache = - Self::compute_properties(&self.right, &self.left, Arc::clone(&new_schema))?; - - let new_join = CrossJoinExec { - left: Arc::clone(&self.right), - right: Arc::clone(&self.left), - schema: new_schema, - left_fut: Default::default(), - metrics: ExecutionPlanMetricsSet::default(), - cache: Arc::new(new_cache), - }; - + let new_join = + CrossJoinExec::new(Arc::clone(&self.right), Arc::clone(&self.left)); reorder_output_after_swap( Arc::new(new_join), &self.left.schema(), @@ -775,9 +751,7 @@ impl CrossJoinStream { mod tests { use super::*; use crate::common; - use crate::test::{TestMemoryExec, assert_join_metrics, build_table_scan_i32}; - use arrow_schema::{DataType, Field}; - use std::collections::HashMap; + use crate::test::{assert_join_metrics, build_table_scan_i32}; use datafusion_common::{assert_contains, test_util::batches_to_sort_string}; use datafusion_execution::runtime_env::RuntimeEnvBuilder; @@ -1070,28 +1044,6 @@ mod tests { Ok(()) } - #[test] - fn test_swapped_cross_join_schema_on_conflicting_metadata() { - 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() - }; - // Conflicting metadata on left and right input, right side wins "metadata_key" -> "right value" - let join = - CrossJoinExec::new(input("a", "left value"), input("b", "right value")); - - let swapped_join = join.swap_inputs().unwrap(); - - // The metadata of the cross-join and the swapped cross-join (with projection on top) must be the same - assert_eq!(join.schema().metadata(), swapped_join.schema().metadata()); - } - /// Returns the column names on the schema fn columns(schema: &Schema) -> Vec { schema.fields().iter().map(|f| f.name().clone()).collect() From 745aed7ca01146b16fc635f86470325e448f81db Mon Sep 17 00:00:00 2001 From: Michael Kleen Date: Wed, 29 Jul 2026 12:30:46 +0200 Subject: [PATCH 2/2] Improve error message for metadata conflict in schema --- datafusion/core/src/physical_planner.rs | 48 ++++++++++++------- datafusion/sqllogictest/src/test_context.rs | 24 ++++++++++ .../sqllogictest/test_files/metadata.slt | 6 +++ 3 files changed, 62 insertions(+), 16 deletions(-) diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 4e914556b4cc0..dcdda28113291 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -3258,15 +3258,14 @@ impl<'a> OptimizationInvariantChecker<'a> { previous_schema: &Arc, ) -> Result<()> { // if the rule is not permitted to change the schema, confirm that it did not change. - if self.rule.schema_check() - && !is_allowed_schema_change(previous_schema.as_ref(), plan.schema().as_ref()) - { - internal_err!( - "PhysicalOptimizer rule '{}' failed. Schema mismatch. Expected original schema: {}, got new schema: {}", - self.rule.name(), - previous_schema, - plan.schema() - )? + if self.rule.schema_check() { + is_allowed_schema_change(previous_schema.as_ref(), plan.schema().as_ref()) + .map_err(|e| { + e.context(format!( + "PhysicalOptimizer rule '{}' failed. Schema mismatch.", + self.rule.name(), + )) + })? } // check invariants per each ExecutionPlan node @@ -3285,28 +3284,45 @@ impl<'a> OptimizationInvariantChecker<'a> { /// This change is allowed because for any field the non-nullable domain `F` is a strict subset /// of the nullable domain `F ∪ { NULL }`. A physical schema that guarantees a stricter subset /// of values will not violate any assumptions made based on the less strict schema. -fn is_allowed_schema_change(old: &Schema, new: &Schema) -> bool { +fn is_allowed_schema_change(old: &Schema, new: &Schema) -> Result<()> { if new.metadata != old.metadata { - return false; + return internal_err!( + "Schema metadata mismatch: Expected original metadata: {:?}, got metadata: {:?}", + old.metadata, + new.metadata + ); } if new.fields.len() != old.fields.len() { - return false; + return internal_err!( + "Schema field mismatch: Expected original field count: {}, got field count: {}", + old.fields.len(), + new.fields.len() + ); } let new_fields = new.fields.iter().map(|f| f.as_ref()); let old_fields = old.fields.iter().map(|f| f.as_ref()); old_fields .zip(new_fields) - .all(|(old, new)| is_allowed_field_change(old, new)) + .try_for_each(|(old, new)| is_allowed_field_change(old, new)) } -fn is_allowed_field_change(old_field: &Field, new_field: &Field) -> bool { - new_field.name() == old_field.name() +fn is_allowed_field_change(old_field: &Field, new_field: &Field) -> Result<()> { + if new_field.name() == old_field.name() && new_field.data_type() == old_field.data_type() && new_field.metadata() == old_field.metadata() && (new_field.is_nullable() == old_field.is_nullable() || !new_field.is_nullable()) + { + Ok(()) + } else { + internal_err!( + "Schema field unallowed change: old field: {:?}, new field: {:?}", + old_field, + new_field + ) + } } impl<'n> TreeNodeVisitor<'n> for OptimizationInvariantChecker<'_> { @@ -4973,7 +4989,7 @@ digraph { let expected_err = OptimizationInvariantChecker::new(&rule) .check(&ok_plan, &different_schema) .unwrap_err(); - assert!(expected_err.to_string().contains("PhysicalOptimizer rule 'OptimizerRuleWithSchemaCheck' failed. Schema mismatch. Expected original schema")); + assert!(expected_err.to_string().contains("PhysicalOptimizer rule 'OptimizerRuleWithSchemaCheck' failed. Schema mismatch.")); // The recursive `check_invariants` walk only runs under `debug_assertions` // (see `OptimizationInvariantChecker::check`). In release builds the walk is diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs index 99c3179ef1056..92f18d8f1d738 100644 --- a/datafusion/sqllogictest/src/test_context.rs +++ b/datafusion/sqllogictest/src/test_context.rs @@ -182,6 +182,7 @@ impl TestContext { "metadata.slt" | "arrow_field.slt" => { info!("Registering metadata table tables"); register_metadata_tables(test_ctx.session_ctx()); + register_conflicting_metadata_tables(test_ctx.session_ctx()) } "union_function.slt" => { info!("Registering table with union column"); @@ -765,3 +766,26 @@ fn register_async_abs_udf(ctx: &SessionContext) { let udf = AsyncScalarUDF::new(Arc::new(async_abs)); ctx.register_udf(udf.into_scalar_udf()); } + +fn register_conflicting_metadata_tables(ctx: &SessionContext) { + let schema_left = + Schema::new(vec![Field::new("a", DataType::Int32, false)]).with_metadata( + HashMap::from([(String::from("metadata_key"), String::from("left"))]), + ); + let data_left = + Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10])) as ArrayRef; + + let batch_left = + RecordBatch::try_new(Arc::new(schema_left), vec![Arc::new(data_left)]).unwrap(); + ctx.register_batch("larger_table", batch_left).unwrap(); + + let schema_right = + Schema::new(vec![Field::new("b", DataType::Int32, false)]).with_metadata( + HashMap::from([(String::from("metadata_key"), String::from("right"))]), + ); + let data_right = Arc::new(Int32Array::from(vec![1])) as ArrayRef; + + let batch_right = + RecordBatch::try_new(Arc::new(schema_right), vec![Arc::new(data_right)]).unwrap(); + ctx.register_batch("smaller_table", batch_right).unwrap(); +} diff --git a/datafusion/sqllogictest/test_files/metadata.slt b/datafusion/sqllogictest/test_files/metadata.slt index 3e2a503e6b3fc..180eb249179ec 100644 --- a/datafusion/sqllogictest/test_files/metadata.slt +++ b/datafusion/sqllogictest/test_files/metadata.slt @@ -520,3 +520,9 @@ NULL the id field statement ok drop table table_with_metadata; + +# Test that metadata on conflicting values raises an error. +# The larger_table has 10 values, smaller_tables 1 value and the fields of each table +# have conflicting metadata, same key different values See test:context.rs register_conflicting_metadata_tables +statement error DataFusion error: PhysicalOptimizer rule 'join_selection' failed\. Schema mismatch\.\ncaused by\nInternal error: Schema metadata mismatch: Expected original metadata: \{"metadata_key": "right"\}, got metadata: \{"metadata_key": "left"\}\. +select * from larger_table cross join smaller_table;