(datafusion) branch main updated: Fix union equivalence schema rewrite with stale constants (#23375)

[gi...@apache.org]

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 215ebb8471 Fix union equivalence schema rewrite with stale constants (#23375)
215ebb8471 is described below

commit 215ebb8471dcc29234f84b79ba859d64f4733e06
Author: xudong.w <wx...@gmail.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org