fix: preserve sorts for non-monotonic temporal casts

datafusion/physical-expr/src/expressions/cast.rs

@@ -283,26 +283,63 @@ 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();
     // A lossless cast recognized by check_bigger_cast is one-to-one, so it is
     // 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::*;

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