Fix union equivalence schema rewrite with stale constants

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