repr, expr: roll leap-second timestamps into the next minute

chrono represents a parsed :60 second as 59 with nanos >= 1e9, a value
that sorts before the following second while the epoch family (extract
epoch, casts to mz_timestamp, timestamp subtraction, date_bin, precision
rounding) counts it at or past that second. That breaks the monotonicity
contracts persist filter pushdown narrows ranges with: an interior leap
value escapes the endpoint-derived range, so a part holding one could be
wrongly discarded. Roll :60 over at the parse boundary, matching Postgres,
so the leap representation never enters a timestamp. TIME keeps it:
rolling a time over would wrap to 00:00:00 and reverse its ordering, and
the whole-second leap time is provably harmless (fractional leap seconds
are rejected at parse for all types).

add_date_time is the other way a leap representation reached a timestamp.
It passed a leap-second TIME's raw nanos straight through, so DATE + TIME
'23:59:60' still constructed one and disagreed with the equivalent
timestamp literal. It now rolls over the same way.

Both rollovers can overflow chrono's range: TIMESTAMP '262142-12-31
23:59:60' parses, because that is chrono's max date and the leap value
passes the HIGH_DATE check, and adding the rollover second overflows.
Overflow reports out of range in the timestamp, timestamptz, and
DATE + TIME paths instead of panicking.

Closes: PER-63

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: def-

src/expr/src/scalar/func.rs

@@ -344,9 +344,20 @@ fn add_date_time(
     date: Date,
     time: chrono::NaiveTime,
 ) -> Result<CheckedTimestamp<NaiveDateTime>, EvalError> {
+    // A leap-second TIME (nanos >= 1e9) rolls over into the next minute,
+    // matching what parsing the equivalent timestamp literal produces. The
+    // leap representation must not enter a timestamp: it sorts before the
+    // next second while epoch-style conversions count it at or past it,
+    // breaking the monotonicity contracts filter pushdown relies on.
+    let (extra_sec, nanos) = match time.nanosecond().checked_sub(1_000_000_000) {
+        Some(nanos) => (1, nanos),
+        None => (0, time.nanosecond()),
+    };
     let dt = NaiveDate::from(date)
-        .and_hms_nano_opt(time.hour(), time.minute(), time.second(), time.nanosecond())
-        .unwrap();
+        .and_hms_nano_opt(time.hour(), time.minute(), time.second(), nanos)
+        .unwrap()
+        .checked_add_signed(chrono::Duration::try_seconds(extra_sec).unwrap())
+        .ok_or(EvalError::TimestampOutOfRange)?;
     Ok(CheckedTimestamp::from_timestamplike(dt)?)
 }
 

src/repr/src/strconv.rs

@@ -525,12 +525,39 @@ fn parse_timestamp_inner(
     order: DateOrder,
 ) -> Result<CheckedTimestamp<NaiveDateTime>, ParseError> {
     match parse_timestamp_string(s, order) {
-        Ok((date, time, _)) => CheckedTimestamp::from_timestamplike(date.and_time(time))
-            .map_err(|_| ParseError::out_of_range("timestamp", s)),
+        Ok((date, time, _)) => roll_over_leap_second(date.and_time(time))
+            .ok_or_else(|| ParseError::out_of_range("timestamp", s))
+            .and_then(|dt| {
+                CheckedTimestamp::from_timestamplike(dt)
+                    .map_err(|_| ParseError::out_of_range("timestamp", s))
+            }),
         Err(e) => Err(ParseError::invalid_input_syntax("timestamp", s).with_details(e)),
     }
 }
 
+/// Postgres normalizes a parsed `:60` second by rolling it into the next
+/// minute. chrono instead keeps a leap-second representation (nanos >= 1e9)
+/// that sorts before the following second while epoch-style conversions
+/// count it at or past that second, which breaks the monotonicity contracts
+/// persist filter pushdown derives ranges with. Roll it over at the parse
+/// boundary so the leap representation never enters a parsed timestamp.
+/// `TIME` keeps the leap representation: rolling it over would wrap to
+/// 00:00:00 and reverse its ordering, and the whole-second leap time is
+/// harmless.
+///
+/// Returns `None` when the rollover overflows chrono's range (a leap second
+/// on the maximum date), which callers report as out of range.
+fn roll_over_leap_second(dt: NaiveDateTime) -> Option<NaiveDateTime> {
+    use chrono::Timelike;
+    match dt.nanosecond().checked_sub(1_000_000_000) {
+        Some(nanos) => dt
+            .with_nanosecond(nanos)
+            .expect("in range")
+            .checked_add_signed(Duration::try_seconds(1).unwrap()),
+        None => Some(dt),
+    }
+}
+
 /// Writes a [`NaiveDateTime`] timestamp to `buf`.
 pub fn format_timestamp<F>(buf: &mut F, ts: &NaiveDateTime) -> Nestable
 where
@@ -569,7 +596,8 @@ fn parse_timestamptz_inner(
     parse_timestamp_string(s, order)
         .and_then(|(date, time, timezone)| {
             use Timezone::*;
-            let mut dt = date.and_time(time);
+            let mut dt = roll_over_leap_second(date.and_time(time))
+                .ok_or_else(|| "timestamp out of range".to_owned())?;
             let offset = match timezone {
                 FixedOffset(offset) => offset,
                 Tz(tz) => match tz.offset_from_local_datetime(&dt).latest() {
@@ -2236,6 +2264,17 @@ mod tests {
 
     use super::*;
 
+    /// Rolling a leap second over must not panic at the timestamp maximum:
+    /// chrono's max date parses, and adding the rollover second overflows.
+    /// The leap value on the max date errors as out of range instead.
+    #[mz_ore::test]
+    fn leap_second_rollover_at_max_date_errors() {
+        assert!(parse_timestamp("262142-12-31 23:59:60").is_err());
+        assert!(parse_timestamptz("262142-12-31 23:59:60+00").is_err());
+        // One second below the maximum still rolls over successfully.
+        assert_ok!(parse_timestamp("262142-12-31 23:59:59"));
+    }
+
     proptest! {
         #[mz_ore::test]
         #[cfg_attr(miri, ignore)] // too slow

test/sqllogictest/explain/pushdown.slt

@@ -185,6 +185,35 @@ ALTER SYSTEM RESET persist_stats_audit_percent
 ----
 COMPLETE 0
 
+# Leap-second timestamps roll over at parse time (Postgres compat). The
+# chrono leap representation sorted below the next second while the epoch
+# family (extract epoch, casts to mz_timestamp, subtraction, date_bin)
+# counted it at or past that second, breaking the monotonicity contracts
+# pushdown narrows ranges with.
+
+query TT
+SELECT TIMESTAMP '2015-06-30 23:59:60' = TIMESTAMP '2015-07-01 00:00:00', TIMESTAMPTZ '2015-06-30 23:59:60+00' = TIMESTAMPTZ '2015-07-01 00:00:00+00'
+----
+true  true
+
+statement ok
+CREATE TABLE leap (ts timestamp)
+
+statement ok
+INSERT INTO leap VALUES ('2015-06-30 23:59:59.4'), ('2015-06-30 23:59:60'), ('2015-07-01 00:00:00.2')
+
+query I
+SELECT count(*) FROM leap WHERE ts::mz_timestamp = 1435708800000::mz_timestamp
+----
+1
+
+# DATE + TIME rolls a leap second over the same way the timestamp literal
+# does, so the two spellings agree.
+query T
+SELECT DATE '2015-06-30' + TIME '23:59:60' = TIMESTAMP '2015-07-01 00:00:00'
+----
+true
+
 # EXPLAIN FILTER PUSHDOWN FOR MATERIALIZED VIEW is also supported
 
 statement ok