Improve error message for metadata conflict in schema

Author: mkleen

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/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;