diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 2195d91b1caee..60e68ca79d924 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -283,18 +283,51 @@ impl CastExpr { } } -pub(crate) fn is_order_preserving_cast_family( - source_type: &DataType, - target_type: &DataType, -) -> bool { - (source_type.is_numeric() || *source_type == Boolean) && target_type.is_numeric() - || source_type.is_temporal() && target_type.is_temporal() - || source_type.eq(target_type) +/// UTC and fixed offsets have no timezone transitions. +fn is_fixed_offset(tz: &str) -> bool { + tz == "UTC" || tz.starts_with(['+', '-']) +} + +/// Whether successful casts preserve order when conversion failures return errors. +/// Unlike `check_bigger_cast`, this allows precision loss and is not sufficient +/// for propagating distinct counts or the ordering of subsequent sort keys. +fn is_order_preserving_cast(source_type: &DataType, target_type: &DataType) -> bool { + use arrow::datatypes::TimeUnit::*; + if source_type == target_type + || (source_type.is_numeric() || *source_type == Boolean) + && target_type.is_numeric() + { + return true; + } + // Temporal casts are not generally monotonic: extracting time-of-day wraps + // at midnight, and timezone transitions can reverse the local date. + match (source_type, target_type) { + (Date32 | Date64, Date32 | Date64) + | (Date32 | Date64, Timestamp(_, None)) + | (Timestamp(_, None), Date32) => true, + (Timestamp(_, Some(tz)), Date32) => is_fixed_offset(tz), + // Arrow converts timestamps to Date64 by scaling the epoch value, + // whereas Date32 extracts the date in the timestamp's timezone. + (Timestamp(_, _), Date64) => true, + // Only admit scaling that cannot wrap on overflow. Time64 widening + // and narrowing to Time32 do not currently check overflow in Arrow. + (Time32(Second | Millisecond), Time32(Second | Millisecond)) + | (Time32(Second | Millisecond), Time64(Microsecond | Nanosecond)) + | (Time64(Nanosecond), Time64(Microsecond)) + | (Duration(_), Duration(_)) => true, + (Timestamp(_, from_tz), Timestamp(_, to_tz)) => { + // Adding a timezone to naive timestamps interprets local times; + // other timezone changes only change metadata on the epoch value. + from_tz.is_some() || to_tz.as_deref().is_none_or(is_fixed_offset) + } + _ => false, + } } pub(crate) fn cast_expr_properties( child: &ExprProperties, target_type: &DataType, + null_on_failure: bool, ) -> Result { let unbounded = Interval::make_unbounded(target_type)?; let source_type = child.range.data_type(); @@ -302,7 +335,11 @@ pub(crate) fn cast_expr_properties( // strictly order-preserving; a narrowing cast may collapse distinct values, // breaking the ordering of subsequent sort keys. let bigger_cast = CastExpr::check_bigger_cast(target_type, &source_type); - if is_order_preserving_cast_family(&source_type, target_type) || bigger_cast { + // New NULLs from failed conversions may violate NULLS FIRST/LAST, even + // if the successfully converted values remain ordered. + if bigger_cast + || (!null_on_failure && is_order_preserving_cast(&source_type, target_type)) + { Ok(child .clone() .with_range(unbounded) @@ -396,10 +433,9 @@ impl PhysicalExpr for CastExpr { ])) } - /// A [`CastExpr`] preserves the ordering of its child if the cast is done - /// under the same datatype family. + /// Propagate ordering only for supported order-preserving conversions. fn get_properties(&self, children: &[ExprProperties]) -> Result { - cast_expr_properties(&children[0], self.cast_type()) + cast_expr_properties(&children[0], self.cast_type(), self.cast_options.safe) } fn fmt_sql(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -1693,6 +1729,171 @@ mod tests { } } + #[test] + fn test_temporal_cast_ordering() { + use TimeUnit::*; + use arrow::compute::SortOptions; + use datafusion_expr_common::sort_properties::SortProperties; + + let timezone = Some("America/Goose_Bay".into()); + // Expected ordering is independent of strictness: discarding precision + // may merge values, but must not reverse them. + let cases = [ + (Date32, Date64, true), + (Date64, Date32, true), + (Date32, Timestamp(Nanosecond, None), true), + (Date64, Timestamp(Second, None), true), + (Timestamp(Second, None), Date32, true), + (Timestamp(Second, Some("UTC".into())), Date32, true), + (Timestamp(Second, Some("+08:00".into())), Date32, true), + (Timestamp(Second, timezone.clone()), Date32, false), + (Timestamp(Second, timezone.clone()), Date64, true), + (Timestamp(Second, None), Timestamp(Nanosecond, None), true), + (Timestamp(Nanosecond, None), Timestamp(Second, None), true), + ( + Timestamp(Second, None), + Timestamp(Second, timezone.clone()), + false, + ), + ( + Timestamp(Second, None), + Timestamp(Second, Some("+08:00".into())), + true, + ), + ( + Timestamp(Second, timezone.clone()), + Timestamp(Second, None), + true, + ), + ( + Timestamp(Second, timezone.clone()), + Timestamp(Millisecond, Some("UTC".into())), + true, + ), + (Timestamp(Second, None), Time32(Second), false), + (Timestamp(Nanosecond, timezone), Time64(Nanosecond), false), + (Time32(Second), Time32(Millisecond), true), + (Time32(Millisecond), Time32(Second), true), + (Time32(Second), Time64(Nanosecond), true), + (Time64(Nanosecond), Time64(Microsecond), true), + (Time64(Microsecond), Time64(Nanosecond), false), + (Time64(Nanosecond), Time32(Second), false), + (Duration(Second), Duration(Nanosecond), true), + (Duration(Nanosecond), Duration(Second), true), + (Null, Timestamp(Second, None), false), + ]; + for (source, target, preserves_order) in cases { + let schema = Schema::new(vec![Field::new("a", source.clone(), true)]); + let expr = CastExpr::new( + col("a", &schema).expect("column exists"), + target.clone(), + None, + ); + for descending in [false, true] { + for nulls_first in [false, true] { + let ordered = SortProperties::Ordered(SortOptions { + descending, + nulls_first, + }); + let child = ExprProperties::new_unknown() + .with_range( + Interval::make_unbounded(&source) + .expect("supported interval type"), + ) + .with_order(ordered) + .with_strictly_order_preserving(true); + let properties = + expr.get_properties(&[child]).expect("cast properties"); + assert_eq!( + properties.sort_properties, + if preserves_order { + ordered + } else { + SortProperties::Unordered + }, + "{source} -> {target}, descending={descending}, nulls_first={nulls_first}" + ); + assert_eq!(properties.range.data_type(), target); + assert!(!properties.strictly_order_preserving); + } + } + } + } + + #[test] + fn test_cast_ordering_with_null_on_failure() { + use arrow::array::TimestampSecondArray; + use arrow::compute::SortOptions; + use datafusion_expr_common::sort_properties::SortProperties; + + // Overflow introduces a trailing NULL even though the input satisfies + // ASC NULLS FIRST. Successful values alone are still in order. + let source = Timestamp(TimeUnit::Second, None); + let target = Timestamp(TimeUnit::Nanosecond, None); + let schema = Arc::new(Schema::new(vec![Field::new("a", source.clone(), true)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(TimestampSecondArray::from(vec![ + None, + Some(0), + Some(i64::MAX), + ]))], + ) + .expect("valid input batch"); + let child = ExprProperties::new_unknown() + .with_range( + Interval::make_unbounded(&source).expect("supported interval type"), + ) + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_strictly_order_preserving(true); + let expr = CastExpr::new( + col("a", &schema).expect("column exists"), + target.clone(), + Some(DEFAULT_SAFE_CAST_OPTIONS), + ); + let actual = expr + .evaluate(&batch) + .expect("safe cast succeeds") + .into_array(batch.num_rows()) + .expect("array result"); + let expected = TimestampNanosecondArray::from(vec![None, Some(0), None]); + assert_eq!(actual.as_ref(), &expected); + let properties = expr + .get_properties(std::slice::from_ref(&child)) + .expect("safe cast properties"); + assert_eq!(properties.sort_properties, SortProperties::Unordered); + assert_eq!(properties.range.data_type(), target); + assert!(!properties.strictly_order_preserving); + + let expr = CastExpr::new(col("a", &schema).expect("column exists"), target, None); + assert!(expr.evaluate(&batch).is_err()); + assert_eq!( + expr.get_properties(&[child]) + .expect("cast properties") + .sort_properties, + SortProperties::Ordered(SortOptions::default()) + ); + + // A lossless cast preserves ordering even in NULL-on-failure mode. + let schema = Schema::new(vec![Field::new("a", Int32, true)]); + let child = ExprProperties::new_unknown() + .with_range( + Interval::make_unbounded(&Int32).expect("supported interval type"), + ) + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_strictly_order_preserving(true); + let expr = CastExpr::new( + col("a", &schema).expect("column exists"), + Int64, + Some(DEFAULT_SAFE_CAST_OPTIONS), + ); + let properties = expr + .get_properties(std::slice::from_ref(&child)) + .expect("safe lossless cast properties"); + assert_eq!(properties.sort_properties, child.sort_properties); + assert!(properties.strictly_order_preserving); + } + #[test] fn test_check_bigger_cast_precision_loss() { use DataType::*; diff --git a/datafusion/sqllogictest/test_files/cast.slt b/datafusion/sqllogictest/test_files/cast.slt index 7067b416de551..1fd3851b81c00 100644 --- a/datafusion/sqllogictest/test_files/cast.slt +++ b/datafusion/sqllogictest/test_files/cast.slt @@ -90,6 +90,95 @@ select * from t0 where v0<1e100; statement ok drop table t0; +# A timestamp ordering cannot satisfy a time-of-day ordering across midnight. +# LIMIT keeps the inner sort as a required part of the input plan. +statement ok +CREATE VIEW cast_midnight AS +SELECT column1 AS ts +FROM (VALUES + (TIMESTAMP '1970-01-01 23:59:59'), + (TIMESTAMP '1970-01-02 00:00:00'), + (TIMESTAMP '1970-01-02 00:00:01'), + (CAST(NULL AS TIMESTAMP)) +); + +query PD +SELECT ts, CAST(ts AS TIME) AS time_of_day +FROM (SELECT ts FROM cast_midnight ORDER BY ts ASC NULLS FIRST LIMIT 4) +ORDER BY time_of_day ASC NULLS FIRST; +---- +NULL NULL +1970-01-02T00:00:00 00:00:00 +1970-01-02T00:00:01 00:00:01 +1970-01-01T23:59:59 23:59:59 + +query PD +SELECT ts, CAST(ts AS TIME) AS time_of_day +FROM (SELECT ts FROM cast_midnight ORDER BY ts ASC NULLS LAST LIMIT 4) +ORDER BY time_of_day ASC NULLS LAST; +---- +1970-01-02T00:00:00 00:00:00 +1970-01-02T00:00:01 00:00:01 +1970-01-01T23:59:59 23:59:59 +NULL NULL + +query PD +SELECT ts, CAST(ts AS TIME) AS time_of_day +FROM (SELECT ts FROM cast_midnight ORDER BY ts DESC NULLS FIRST LIMIT 4) +ORDER BY time_of_day DESC NULLS FIRST; +---- +NULL NULL +1970-01-01T23:59:59 23:59:59 +1970-01-02T00:00:01 00:00:01 +1970-01-02T00:00:00 00:00:00 + +query PD +SELECT ts, CAST(ts AS TIME) AS time_of_day +FROM (SELECT ts FROM cast_midnight ORDER BY ts DESC NULLS LAST LIMIT 4) +ORDER BY time_of_day DESC NULLS LAST; +---- +1970-01-01T23:59:59 23:59:59 +1970-01-02T00:00:01 00:00:01 +1970-01-02T00:00:00 00:00:00 +NULL NULL + +# A named timezone rollback can also reverse the local date. +query ID +SELECT arrow_cast(ts, 'Int64') AS epoch, CAST(ts AS DATE) AS d +FROM ( + SELECT arrow_cast(column1, 'Timestamp(Second, Some("America/Goose_Bay"))') AS ts + FROM (VALUES (562129259::bigint), (562129260::bigint)) + ORDER BY ts LIMIT 2 +) +ORDER BY d; +---- +562129260 1987-10-24 +562129259 1987-10-25 + +# Reducing timestamp precision remains order-preserving, even when values merge. +statement ok +SET datafusion.explain.physical_plan_only = true; + +query TT +EXPLAIN SELECT arrow_cast(ts, 'Timestamp(Second, None)') AS truncated +FROM ( + SELECT column1 AS ts + FROM (VALUES + (TIMESTAMP '1970-01-01 00:00:00.001'), + (TIMESTAMP '1970-01-01 00:00:00.002'), + (TIMESTAMP '1970-01-01 00:00:01.000') + ) + ORDER BY ts LIMIT 3 +) +ORDER BY truncated; +---- +physical_plan +01)ProjectionExec: expr=[CAST(column1@0 AS Timestamp(s)) as truncated] +02)--SortExec: TopK(fetch=3), expr=[column1@0 ASC NULLS LAST], preserve_partitioning=[false] +03)----DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +SET datafusion.explain.physical_plan_only = false; # Regression for https://github.com/apache/datafusion/issues/25523. statement ok