This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new b4cde12a85 fix: Improve error message for metadata conflict in schema
(#23952)
b4cde12a85 is described below
commit b4cde12a85b1201cb6addd8d328d30f4c5691507
Author: Michael Kleen <[email protected]>
AuthorDate: Sat Aug 1 10:15:59 2026 +0200
fix: Improve error message for metadata conflict in schema (#23952)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/23461
## Rationale for this change
Based on the discussion with @2010YOUY01 in
https://github.com/apache/datafusion/issues/23461 this pr improves the
error message when there is a metadata conflict when schemas get merged.
It also reverts https://github.com/apache/datafusion/pull/23605, so
`CrossJoinExec` has again the original behavior.
```sql
select * from larger_table cross join smaller_table;
DataFusion error: PhysicalOptimizer rule 'join_selection' failed. Schema
mismatch caused by
Internal error: Schema metadata mismatch: Expected original metadata:
{"metadata_key": "right"}, got metadata: {"metadata_key": "left"}
```
## What changes are included in this PR?
- Revert https://github.com/apache/datafusion/pull/23605 the behavior
change for `CrossJoinExec`
- Improve error messages for schema validation in `physical_planner.rs`
- Data creation in `test_context.rs`
- Reproduction in `metadata.slt`
## Are these changes tested?
Yes
## Are there any user-facing changes?
Yes, error message improved when the schema metadata has conflicting
keys using a cross-join.
---
datafusion/core/src/physical_planner.rs | 48 ++++++++++++++-------
datafusion/physical-plan/src/joins/cross_join.rs | 54 ++----------------------
datafusion/sqllogictest/src/test_context.rs | 24 +++++++++++
datafusion/sqllogictest/test_files/metadata.slt | 6 +++
4 files changed, 65 insertions(+), 67 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index 222511b901..46e2957af0 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -3128,15 +3128,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
@@ -3155,28 +3154,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<'_> {
@@ -5038,7 +5054,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/physical-plan/src/joins/cross_join.rs
b/datafusion/physical-plan/src/joins/cross_join.rs
index 1a631aac98..16155aaafd 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<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()
diff --git a/datafusion/sqllogictest/src/test_context.rs
b/datafusion/sqllogictest/src/test_context.rs
index 99c3179ef1..92f18d8f1d 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 3e2a503e6b..0fc74fa6cf 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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
