You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/20 14:26:03 UTC
(datafusion) branch gh-readonly-queue/main/pr-25526-6700d02160b1d2aee228455a715813e4a586cfc3 created (now 918ab2c998)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25526-6700d02160b1d2aee228455a715813e4a586cfc3
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 918ab2c998 fix: preserve sorts for non-monotonic temporal casts (#25526)
This branch includes the following new commits:
new 918ab2c998 fix: preserve sorts for non-monotonic temporal casts (#25526)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: preserve sorts for non-monotonic temporal casts (#25526)
Posted by gi...@apache.org.
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 <69...@users.noreply.github.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org