fix: time ± interval returns a wrapped time instead of an interval (#23279)

## Which issue does this PR close?

- Closes #22265
- Closes #22255

## Rationale for this change

`time + interval` (and `interval + time` / `time - interval`) returned
an `Interval` instead of a `time`, so the value was never wrapped within
the 24-hour clock:

```
> SELECT time '23:30' + interval '2 hours';
25 hours 30 mins   -- an Interval, not a time
```

PostgreSQL (and DuckDB) return a `time` that wraps around midnight:

```
01:30:00
```

The root cause is in type coercion: `time <op> interval` was coerced by
widening the `time` operand into an `Interval`, so the addition happened
between two intervals and the result kept the `Interval` type.

## What changes are included in this PR?

Following the existing `Date - Date` special case (in the same two
files), `time ± interval` is now handled explicitly:

- **Coercion** (`expr-common/src/type_coercion/binary.rs`): `time +
interval`, `interval + time`, and `time - interval` now coerce to
`(time, interval)` — the interval normalized to `MonthDayNano`, and the
time operand kept at its own unit, which is also the result type.
`interval - time`, which is not meaningful, is left unchanged.
- **Evaluation** (`physical-expr/src/expressions/binary.rs`): a new
`apply_time_interval` adds/subtracts the interval's sub-day component
and wraps the result modulo 24 hours, for all four time units
(`Time32(Second|Millisecond)`, `Time64(Microsecond|Nanosecond)`) and
both operand orders. The result **keeps the input time's unit**,
mirroring `timestamp/date + interval` (which preserve their unit and
apply the interval at that resolution); interval precision finer than
the time's unit is truncated, so `time(s) + interval '1 nanosecond'` is
a no-op, exactly like `timestamp(s) + interval '1 nanosecond'`. Only the
interval's sub-day portion affects a time-of-day; whole months and days
are ignored, matching PostgreSQL (e.g. `time '10:00' + interval '1 day 2
hours'` = `12:00:00`).

`time - time` (→ `Interval`) is unchanged.

## Are these changes tested?

Yes.
`datafusion/sqllogictest/test_files/datetime/arith_time_interval.slt`
was previously a characterization test that documented the incorrect
`Interval` output; it now asserts the correct wrapped `time` results —
including wrapping past midnight in both directions (`22:00 + 3h →
01:00:00`, `02:00 - 3h → 23:00:00`), ignoring whole days, preserving
each input time unit (`arrow_typeof` per unit), and the finer-than-unit
interval truncation.

## Are there any user-facing changes?

Yes — `time ± interval` now returns a `time` value (wrapped within 24
hours) instead of an `Interval`, aligning DataFusion with PostgreSQL and
DuckDB. The result keeps the input time's unit rather than widening it
(see the 55.0.0 upgrade guide).

Author: vismaytiwari

datafusion/expr-common/src/type_coercion/binary.rs

@@ -267,6 +267,23 @@ impl<'a> BinaryTypeCoercer<'a> {
                 ret: Int64,
             });
         }
+        Plus | Minus if is_time_interval_arithmetic(lhs, rhs, self.op) => {
+            // `time ± interval` yields a `time` wrapped within the 24-hour clock,
+            // matching PostgreSQL and DuckDB (e.g. `time '23:30' + interval '2 hours'`
+            // is `01:30:00`). The interval is normalized to `MonthDayNano`; the time
+            // operand keeps its own unit and is also the result type -- mirroring
+            // `timestamp/date + interval`, which preserve their unit and apply the
+            // interval at that resolution. So, like `timestamp(s) + interval
+            // '1 nanosecond'`, `time(s) + interval '1 nanosecond'` is a no-op rather
+            // than widening the type.
+            let (lhs, rhs, ret) = match (lhs, rhs) {
+                (Interval(_), time) => {
+                    (Interval(MonthDayNano), time.clone(), time.clone())
+                }
+                (time, _) => (time.clone(), Interval(MonthDayNano), time.clone()),
+            };
+            return Ok(Signature { lhs, rhs, ret });
+        }
         Plus | Minus | Multiply | Divide | Modulo  =>  {
             if let Ok(ret) = self.get_result(lhs, rhs) {
 
@@ -362,6 +379,23 @@ fn is_date_minus_date(lhs: &DataType, rhs: &DataType) -> bool {
     )
 }
 
+/// Returns true for `time + interval`, `interval + time`, or `time - interval`.
+///
+/// These follow PostgreSQL/DuckDB semantics where the result is a `time` value
+/// wrapped within the 24-hour clock, rather than being widened to an interval.
+fn is_time_interval_arithmetic(lhs: &DataType, rhs: &DataType, op: &Operator) -> bool {
+    use DataType::{Interval, Time32, Time64};
+    match op {
+        Operator::Plus => matches!(
+            (lhs, rhs),
+            (Time32(_) | Time64(_), Interval(_)) | (Interval(_), Time32(_) | Time64(_))
+        ),
+        // `interval - time` is not meaningful, so only `time - interval` is accepted.
+        Operator::Minus => matches!((lhs, rhs), (Time32(_) | Time64(_), Interval(_))),
+        _ => false,
+    }
+}
+
 /// Coercion rules for mathematics operators between decimal and non-decimal types.
 fn math_decimal_coercion(
     lhs_type: &DataType,

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

@@ -271,6 +271,189 @@ where
     }
 }
 
+/// Returns true for `time + interval` or `interval + time`.
+fn is_time_plus_interval(lhs: &DataType, rhs: &DataType) -> bool {
+    matches!(
+        (lhs, rhs),
+        (
+            DataType::Time32(_) | DataType::Time64(_),
+            DataType::Interval(_)
+        ) | (
+            DataType::Interval(_),
+            DataType::Time32(_) | DataType::Time64(_)
+        )
+    )
+}
+
+/// Returns true for `time - interval`.
+fn is_time_minus_interval(lhs: &DataType, rhs: &DataType) -> bool {
+    matches!(
+        (lhs, rhs),
+        (
+            DataType::Time32(_) | DataType::Time64(_),
+            DataType::Interval(_)
+        )
+    )
+}
+
+/// Evaluates `time + interval`, `interval + time`, or `time - interval`, returning a
+/// `time` wrapped within the 24-hour clock to match PostgreSQL and DuckDB (e.g.
+/// `time '23:30' + interval '2 hours'` is `01:30:00`). arrow's arithmetic kernels do
+/// not implement time-of-day arithmetic, so it is handled here.
+///
+/// The result keeps the input time's unit; the interval (normalized to `MonthDayNano`
+/// by the coercion layer) is applied at nanosecond precision and floored to that unit,
+/// mirroring `timestamp(unit) + interval`. Only the sub-day portion of the interval
+/// affects a time-of-day -- whole months and days are ignored, matching PostgreSQL. The
+/// floor is applied after the sign, so `time(s) + interval '1 nanosecond'` is a no-op
+/// while `time(s) - interval '1 nanosecond'` rolls back a second, exactly as the
+/// timestamp case does.
+fn apply_time_interval(
+    lhs: &ColumnarValue,
+    rhs: &ColumnarValue,
+    subtract: bool,
+) -> Result<ColumnarValue> {
+    // The `time` operand determines the result type; the other is the interval.
+    let (time, interval) = if matches!(lhs.data_type(), DataType::Interval(_)) {
+        (rhs, lhs)
+    } else {
+        (lhs, rhs)
+    };
+
+    // Dispatch on the time unit; `ns_per_unit` converts the interval's nanoseconds to
+    // that unit, and the arithmetic is done (and wrapped) at that resolution.
+    match time.data_type() {
+        DataType::Time32(TimeUnit::Second) => wrap_time_interval::<Time32SecondType>(
+            time,
+            interval,
+            subtract,
+            1_000_000_000,
+        ),
+        DataType::Time32(TimeUnit::Millisecond) => {
+            wrap_time_interval::<Time32MillisecondType>(
+                time, interval, subtract, 1_000_000,
+            )
+        }
+        DataType::Time64(TimeUnit::Microsecond) => {
+            wrap_time_interval::<Time64MicrosecondType>(time, interval, subtract, 1_000)
+        }
+        DataType::Time64(TimeUnit::Nanosecond) => {
+            wrap_time_interval::<Time64NanosecondType>(time, interval, subtract, 1)
+        }
+        other => internal_err!("time operand expected, got: {other}"),
+    }
+}
+
+/// Adds or subtracts an interval to/from a `time` of arrow primitive type `T`, wrapping
+/// the result within the 24-hour clock and keeping the type `T`. `ns_per_unit` is the
+/// number of nanoseconds in one unit of `T` (e.g. `1_000` for microseconds).
+fn wrap_time_interval<T: ArrowPrimitiveType>(
+    time: &ColumnarValue,
+    interval: &ColumnarValue,
+    subtract: bool,
+    ns_per_unit: i64,
+) -> Result<ColumnarValue>
+where
+    T::Native: Copy + Into<i64> + TryFrom<i64>,
+{
+    /// Nanoseconds in a 24-hour day.
+    const DAY_NANOS: i64 = 86_400_000_000_000;
+    // Units in a 24-hour day, at `T`'s resolution.
+    let day_units = DAY_NANOS / ns_per_unit;
+
+    // Wraps `time ± interval` into `[0, day_units)`. The interval is reduced modulo a day
+    // (so the sum stays within `i64`), applied at nanosecond precision, then floored to
+    // `T`'s unit -- matching `timestamp(unit) ± interval`. Because the floor is applied
+    // after the sign, `time(s) - interval '1 nanosecond'` rolls back a full second, just
+    // as the timestamp case does, while `time(s) + interval '1 nanosecond'` is a no-op.
+    // `div_euclid`/`rem_euclid` floor toward negative infinity, so the wrapped value stays
+    // in `[0, day_units)`, which always fits `T::Native`.
+    let wrap = |time_unit: i64, iv: IntervalMonthDayNano| -> T::Native {
+        let iv_ns = iv.nanoseconds % DAY_NANOS;
+        let signed_ns = if subtract { -iv_ns } else { iv_ns };
+        let delta = signed_ns.div_euclid(ns_per_unit);
+        let wrapped = (time_unit + delta).rem_euclid(day_units);
+        T::Native::try_from(wrapped).unwrap_or_default()
+    };
+
+    /// Extracts an `Interval(MonthDayNano)` scalar.
+    fn interval_scalar(scalar: &ScalarValue) -> Result<Option<IntervalMonthDayNano>> {
+        match scalar {
+            ScalarValue::IntervalMonthDayNano(value) => Ok(*value),
+            other => internal_err!(
+                "Interval(MonthDayNano) scalar expected, got: {}",
+                other.data_type()
+            ),
+        }
+    }
+
+    /// Extracts a time scalar as its unit count since midnight.
+    fn time_scalar_units(scalar: &ScalarValue) -> Result<Option<i64>> {
+        match scalar {
+            ScalarValue::Time32Second(value) | ScalarValue::Time32Millisecond(value) => {
+                Ok(value.map(i64::from))
+            }
+            ScalarValue::Time64Microsecond(value)
+            | ScalarValue::Time64Nanosecond(value) => Ok(*value),
+            other => {
+                internal_err!("time scalar expected, got: {}", other.data_type())
+            }
+        }
+    }
+
+    /// Builds a time scalar of type `P` from a unit count.
+    fn time_scalar<P: ArrowPrimitiveType>(value: Option<i64>) -> ScalarValue {
+        match P::DATA_TYPE {
+            DataType::Time32(TimeUnit::Second) => {
+                ScalarValue::Time32Second(value.map(|v| v as i32))
+            }
+            DataType::Time32(TimeUnit::Millisecond) => {
+                ScalarValue::Time32Millisecond(value.map(|v| v as i32))
+            }
+            DataType::Time64(TimeUnit::Microsecond) => {
+                ScalarValue::Time64Microsecond(value)
+            }
+            _ => ScalarValue::Time64Nanosecond(value),
+        }
+    }
+
+    match (time, interval) {
+        (ColumnarValue::Array(time), ColumnarValue::Array(interval)) => {
+            let time = time.as_primitive::<T>();
+            let interval = interval.as_primitive::<IntervalMonthDayNanoType>();
+            let result: PrimitiveArray<T> =
+                arrow::compute::binary(time, interval, |t, iv| wrap(t.into(), iv))?;
+            Ok(ColumnarValue::Array(Arc::new(result)))
+        }
+        (ColumnarValue::Array(time), ColumnarValue::Scalar(interval)) => {
+            let time = time.as_primitive::<T>();
+            match interval_scalar(interval)? {
+                Some(iv) => {
+                    let result: PrimitiveArray<T> = time.unary(|t| wrap(t.into(), iv));
+                    Ok(ColumnarValue::Array(Arc::new(result)))
+                }
+                None => Ok(ColumnarValue::Scalar(time_scalar::<T>(None))),
+            }
+        }
+        (ColumnarValue::Scalar(time), ColumnarValue::Array(interval)) => {
+            let interval = interval.as_primitive::<IntervalMonthDayNanoType>();
+            match time_scalar_units(time)? {
+                Some(t) => {
+                    let result: PrimitiveArray<T> = interval.unary(|iv| wrap(t, iv));
+                    Ok(ColumnarValue::Array(Arc::new(result)))
+                }
+                None => Ok(ColumnarValue::Scalar(time_scalar::<T>(None))),
+            }
+        }
+        (ColumnarValue::Scalar(time), ColumnarValue::Scalar(interval)) => {
+            let result = time_scalar_units(time)?
+                .zip(interval_scalar(interval)?)
+                .map(|(t, iv)| wrap(t, iv).into());
+            Ok(ColumnarValue::Scalar(time_scalar::<T>(result)))
+        }
+    }
+}
+
 impl PhysicalExpr for BinaryExpr {
     fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
         BinaryTypeCoercer::new(
@@ -353,6 +536,18 @@ impl PhysicalExpr for BinaryExpr {
         let input_schema = schema.as_ref();
 
         match self.op {
+            // `time ± interval` returns a wrapped `time` (PostgreSQL/DuckDB
+            // semantics); arrow's arithmetic kernels don't implement it.
+            Operator::Plus
+                if is_time_plus_interval(&left_data_type, &right_data_type) =>
+            {
+                return apply_time_interval(&lhs, &rhs, false);
+            }
+            Operator::Minus
+                if is_time_minus_interval(&left_data_type, &right_data_type) =>
+            {
+                return apply_time_interval(&lhs, &rhs, true);
+            }
             Operator::Plus if self.fail_on_overflow => return apply(&lhs, &rhs, add),
             Operator::Plus => return apply(&lhs, &rhs, add_wrapping),
             // Special case: Date - Date returns Int64 (days difference)