diff --git a/datafusion/physical-expr/src/equivalence/properties/mod.rs b/datafusion/physical-expr/src/equivalence/properties/mod.rs index a98341b10765a..0f7169015a7ff 100644 --- a/datafusion/physical-expr/src/equivalence/properties/mod.rs +++ b/datafusion/physical-expr/src/equivalence/properties/mod.rs @@ -1305,7 +1305,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 d77129472a8ba..ea4094e75159a 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 43c99dd99c885..e868acdfa908e 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -148,9 +148,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(),