This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25526-6700d02160b1d2aee228455a715813e4a586cfc3 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 918ab2c99844e4b6f79e17f275343b7c59c139f7 Author: Malingshu <[email protected]> AuthorDate: Sun Sep 20 14:25:39 2026 +0000 fix: preserve sorts for non-monotonic temporal casts (#25526) ## Which issue does this PR close? - Closes #25506. ## Rationale for this change `ORDER BY CAST(ts AS TIME)` can return incorrectly ordered results when an already ordered timestamp input crosses midnight. A named timezone rollback can similarly make a timestamp-to-date conversion non-monotonic. In both cases, DataFusion may incorrectly reuse the input ordering and remove the sort required by the cast result. ## What changes are included in this PR? Replace the broad temporal-to-temporal ordering rule with an explicit list of order-preserving temporal casts. Keep the existing lossless cast check separate because it provides the stronger guarantees used for strict ordering and statistics propagation. Cast expressions that may turn conversion failures into NULL no longer propagate ordering unless the conversion is known to be lossless, since a new NULL can violate `NULLS FIRST` or `NULLS LAST`. ## What is the testing strategy for this PR? The unit tests cover safe and unsafe temporal cast combinations, both sort directions, both NULL placements, unknown input types, and NULL-on-failure casts. The cases added to `cast.slt` cover timestamp-to-time conversion across midnight, timestamp-to-date conversion across a named timezone rollback, and verify that reducing timestamp precision still avoids an unnecessary sort. ## Are there any user-facing changes? Queries that order by a non-monotonic temporal cast now retain the required sort and return correctly ordered results. There are no public API changes. --- datafusion/physical-expr/src/expressions/cast.rs | 223 +++++++++++++++++++++-- datafusion/sqllogictest/test_files/cast.slt | 89 +++++++++ 2 files changed, 301 insertions(+), 11 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 2195d91b1c..60e68ca79d 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<ExprProperties> { 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<ExprProperties> { - 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 7067b416de..1fd3851b81 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 --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
