fix: Improve error message for metadata conflict in schema

datafusion/core/src/physical_planner.rs

@@ -3258,15 +3258,14 @@ impl<'a> OptimizationInvariantChecker<'a> {
         previous_schema: &Arc<Schema>,
     ) -> 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

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<Arc<dyn ExecutionPlan>> {
-        // 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::<Fields>();
-
-        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<T: BatchTransformer> CrossJoinStream<T> {
 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<String> {
         schema.fields().iter().map(|f| f.name().clone()).collect()

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();
+}

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;