This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23375-b790763529ed709ad1b264f434e06a1b7878c947 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 215ebb8471dcc29234f84b79ba859d64f4733e06 Author: xudong.w <[email protected]> AuthorDate: Thu Jul 9 14:54:00 2026 +0800 Fix union equivalence schema rewrite with stale constants (#23375) ## Which issue does this PR close? - Closes #23374. ## Rationale for this change `UnionExec::try_new` can panic while computing equivalence properties if stale constant metadata is carried across a projection and then rewritten to the union output schema. In the observed shape, a filter such as `ticker = 'ESU6'` can leave a uniform string constant in equivalence properties. After a parent projection drops `ticker`, union property schema rewriting can see the remaining column slot as a timestamp column and attempt to cast `'ESU6'` to `Timestamp`, which fails during planning. Equivalence constants are optimizer metadata, so an unrepresentable constant after schema rewrite should be discarded rather than failing query planning. ## What changes are included in this PR? - Drops a uniform constant during `EquivalenceProperties::with_new_schema` if its value cannot be cast to the rewritten expression type. - Removes trivial equivalence classes after dropping such constants. - Propagates `UnionExec::compute_properties` errors from `UnionExec::try_new` instead of unwrapping. - Adds a regression test for union equivalence schema rewrite with an unrepresentable stale constant value. ## Are these changes tested? Yes: ## Are there any user-facing changes? No API change. This prevents a planner panic for affected `UNION ALL` + filter + projection query shapes. --- .../src/equivalence/properties/mod.rs | 13 ++++++++- .../src/equivalence/properties/union.rs | 32 +++++++++++++++++++++- datafusion/physical-plan/src/union.rs | 4 +-- 3 files changed, 44 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index bb74cd1d9c..17c3898fd9 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -1314,7 +1314,18 @@ impl EquivalenceProperties { if let (Some(data_type), Some(AcrossPartitions::Uniform(Some(value)))) = (data_type, &mut eq_class.constant) { - *value = value.cast_to(&data_type)?; + match value.cast_to(&data_type) { + Ok(cast_value) => *value = cast_value, + Err(_) => { + // This is optimizer metadata. If a stale constant + // value cannot be represented after schema rewrite, + // drop the constant instead of failing planning. + eq_class.constant = None; + } + } + } + if eq_class.is_trivial() { + continue; } eq_classes.push(eq_class); } diff --git a/datafusion/physical-expr/src/equivalence/properties/union.rs b/datafusion/physical-expr/src/equivalence/properties/union.rs index d77129472a..ea4094e751 100644 --- a/datafusion/physical-expr/src/equivalence/properties/union.rs +++ b/datafusion/physical-expr/src/equivalence/properties/union.rs @@ -311,7 +311,7 @@ mod tests { use crate::equivalence::tests::{create_test_schema, parse_sort_expr}; use crate::expressions::col; - use arrow::datatypes::{DataType, Field, Schema}; + use arrow::datatypes::{DataType, Field, Schema, TimeUnit}; use datafusion_common::ScalarValue; use itertools::Itertools; @@ -899,6 +899,36 @@ mod tests { Ok(()) } + #[test] + fn test_union_drops_unrepresentable_constant_value_after_schema_rewrite() -> Result<()> + { + let input_schema = Arc::new(Schema::new(vec![Field::new( + "ticker", + DataType::Timestamp(TimeUnit::Nanosecond, None), + true, + )])); + let output_schema = Arc::new(Schema::new(vec![Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Nanosecond, None), + true, + )])); + + let ticker = col("ticker", &input_schema)?; + let stale_value = ScalarValue::Utf8(Some("ESU6".to_owned())); + let const_expr = ConstExpr::new( + Arc::clone(&ticker), + AcrossPartitions::Uniform(Some(stale_value)), + ); + + let mut input = EquivalenceProperties::new(input_schema); + input.add_constants(vec![const_expr])?; + + let union_props = calculate_union(vec![input], output_schema)?; + assert!(union_props.constants().is_empty()); + + Ok(()) + } + /// Return a new schema with the same types, but new field names /// /// The new field names are the old field names with `text` appended. diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index d6f664c005..b330c30583 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -130,9 +130,7 @@ impl UnionExec { // The schema of the inputs and the union schema is consistent when: // - They have the same number of fields, and // - Their fields have same types at the same indices. - // Here, we know that schemas are consistent and the call below can - // not return an error. - let cache = Self::compute_properties(&inputs, schema).unwrap(); + let cache = Self::compute_properties(&inputs, schema)?; Ok(Arc::new(UnionExec { inputs, metrics: ExecutionPlanMetricsSet::new(), --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
