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.

Author: xudong963

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

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.

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(),