fix: keep fractional seconds in EXTRACT(EPOCH) for Duration

datafusion/functions/benches/datetime_expressions/date_part.rs

@@ -39,6 +39,7 @@ const BATCH_SIZE: usize = 1000;
 const TS_BOUND: i64 = 2_006_463_600;
 const SEC_DAY: i64 = 86_400;
 const DAYS_SINCE_EPOCH: i64 = TS_BOUND / SEC_DAY;
+const NULL_DENSITY: f64 = 0.1;
 
 fn generate_timestamp_ns_array(rng: &mut StdRng) -> TimestampNanosecondArray {
     TimestampNanosecondArray::from(
@@ -162,6 +163,31 @@ fn generate_duration_nanosecond_array(rng: &mut StdRng) -> DurationNanosecondArr
     )
 }
 
+// Durations under `i32::MAX` nanoseconds (about 2.1 seconds)
+fn generate_duration_nanosecond_small_array(rng: &mut StdRng) -> DurationNanosecondArray {
+    DurationNanosecondArray::from(
+        (0..BATCH_SIZE)
+            .map(|_| rng.random_range(0..i32::MAX as i64))
+            .collect::<Vec<_>>(),
+    )
+}
+
+fn generate_duration_nanosecond_array_with_nulls(
+    rng: &mut StdRng,
+) -> DurationNanosecondArray {
+    DurationNanosecondArray::from(
+        (0..BATCH_SIZE)
+            .map(|_| {
+                if rng.random_bool(NULL_DENSITY) {
+                    None
+                } else {
+                    Some(rng.random_range(0..TS_BOUND * 1_000_000_000))
+                }
+            })
+            .collect::<Vec<_>>(),
+    )
+}
+
 fn bench_date_part(
     c: &mut Criterion,
     udf: &Arc<ScalarUDF>,
@@ -213,6 +239,10 @@ fn criterion_benchmark(c: &mut Criterion) {
     let duration_ns = Arc::new(generate_duration_nanosecond_array(&mut rng)) as ArrayRef;
     let date32 = Arc::new(generate_date32_array(&mut rng)) as ArrayRef;
     let date64 = Arc::new(generate_date64_array(&mut rng)) as ArrayRef;
+    let duration_ns_small =
+        Arc::new(generate_duration_nanosecond_small_array(&mut rng)) as ArrayRef;
+    let duration_ns_nulls =
+        Arc::new(generate_duration_nanosecond_array_with_nulls(&mut rng)) as ArrayRef;
 
     let udf = date_part();
 
@@ -322,6 +352,8 @@ fn criterion_benchmark(c: &mut Criterion) {
         ("interval_dt", &interval_dt),
         ("interval_mdn", &interval_mdn),
         ("duration_ns", &duration_ns),
+        ("duration_ns_small", &duration_ns_small),
+        ("duration_ns_nulls", &duration_ns_nulls),
     ] {
         bench_date_part(
             c,

datafusion/functions/src/datetime/date_part.rs

@@ -39,7 +39,9 @@ use crate::datetime::common::parse_tz;
 use datafusion_common::{
     Result, ScalarValue,
     cast::{
-        as_date32_array, as_date64_array, as_int32_array, as_interval_dt_array,
+        as_date32_array, as_date64_array, as_duration_microsecond_array,
+        as_duration_millisecond_array, as_duration_nanosecond_array,
+        as_duration_second_array, as_int32_array, as_interval_dt_array,
         as_interval_mdn_array, as_interval_ym_array, as_primitive_array,
         as_time32_millisecond_array, as_time32_second_array, as_time64_microsecond_array,
         as_time64_nanosecond_array, as_timestamp_microsecond_array,
@@ -504,48 +506,6 @@ fn seconds_as_i32(array: &dyn Array, unit: TimeUnit) -> Result<ArrayRef> {
     }
 }
 
-/// Invoke [`date_part`] on an `array` (e.g. Timestamp) and convert the
-/// result to a total number of seconds, milliseconds, microseconds or
-/// nanoseconds
-///
-/// Given epoch return f64, this is a duplicated function to optimize for f64 type
-fn seconds(array: &dyn Array, unit: TimeUnit) -> Result<ArrayRef> {
-    let sf = match unit {
-        Second => 1_f64,
-        Millisecond => 1_000_f64,
-        Microsecond => 1_000_000_f64,
-        Nanosecond => 1_000_000_000_f64,
-    };
-    let secs = date_part(array, DatePart::Second)?;
-    // This assumes array is primitive and not a dictionary
-    let secs = as_int32_array(secs.as_ref())?;
-    let subsecs = date_part(array, DatePart::Nanosecond)?;
-    let subsecs = as_int32_array(subsecs.as_ref())?;
-
-    // Special case where there are no nulls.
-    if subsecs.null_count() == 0 {
-        let r: Float64Array = binary(secs, subsecs, |secs, subsecs| {
-            (secs as f64 + ((subsecs % 1_000_000_000) as f64 / 1_000_000_000_f64)) * sf
-        })?;
-        Ok(Arc::new(r))
-    } else {
-        // Nulls in secs are preserved, nulls in subsecs are treated as zero to account for the case
-        // where the number of nanoseconds overflows.
-        let r: Float64Array = secs
-            .iter()
-            .zip(subsecs)
-            .map(|(secs, subsecs)| {
-                secs.map(|secs| {
-                    let subsecs = subsecs.unwrap_or(0);
-                    (secs as f64 + ((subsecs % 1_000_000_000) as f64 / 1_000_000_000_f64))
-                        * sf
-                })
-            })
-            .collect();
-        Ok(Arc::new(r))
-    }
-}
-
 fn epoch(array: &dyn Array) -> Result<ArrayRef> {
     const SECONDS_IN_A_DAY: f64 = 86400_f64;
     // Note: Month-to-second conversion uses 30 days as an approximation.
@@ -589,7 +549,16 @@ fn epoch(array: &dyn Array) -> Result<ArrayRef> {
                     + x.nanoseconds as f64 / 1_000_000_000_f64
             })
         }
-        Duration(_) => return seconds(array, Second),
+        Duration(Second) => as_duration_second_array(array)?.unary(|x| x as f64),
+        Duration(Millisecond) => {
+            as_duration_millisecond_array(array)?.unary(|x| x as f64 / 1_000_f64)
+        }
+        Duration(Microsecond) => {
+            as_duration_microsecond_array(array)?.unary(|x| x as f64 / 1_000_000_f64)
+        }
+        Duration(Nanosecond) => {
+            as_duration_nanosecond_array(array)?.unary(|x| x as f64 / 1_000_000_000_f64)
+        }
         d => return exec_err!("Cannot convert {d:?} to epoch"),
     };
     Ok(Arc::new(f))

datafusion/sqllogictest/test_files/datetime/date_part.slt

@@ -1178,6 +1178,78 @@ SELECT extract(epoch from arrow_cast(2, 'Duration(Second)'))
 ----
 2
 
+# epoch keeps the fraction of a duration whose total sub-second units exceed i32::MAX
+query RR
+SELECT extract(epoch from arrow_cast(2147483647, 'Duration(Nanosecond)')),
+       extract(epoch from arrow_cast(2147483648, 'Duration(Nanosecond)'))
+----
+2.147483647 2.147483648
+
+# Reproducer from https://github.com/apache/datafusion/issues/26673: every row keeps its fraction.
+# Before the fix, the rows at or above 2^31 ns (2.147483648 s) lost their fraction.
+query TTR
+SELECT description,
+       arrow_typeof(finished_at - started_at),
+       extract(epoch from finished_at - started_at)
+FROM (VALUES
+  ('1.25 seconds',      timestamp '2025-01-01 00:00:00', timestamp '2025-01-01 00:00:01.250000'),
+  ('2.10 seconds',      timestamp '2025-01-01 00:00:00', timestamp '2025-01-01 00:00:02.100000'),
+  ('2.50 seconds',      timestamp '2025-01-01 00:00:00', timestamp '2025-01-01 00:00:02.500000'),
+  ('4.763 seconds',     timestamp '2025-01-01 00:00:00', timestamp '2025-01-01 00:00:04.763000'),
+  ('12.923350 seconds', timestamp '2025-01-01 00:00:00', timestamp '2025-01-01 00:00:12.923350')
+) AS t(description, started_at, finished_at)
+----
+1.25 seconds Duration(ns) 1.25
+2.10 seconds Duration(ns) 2.1
+2.50 seconds Duration(ns) 2.5
+4.763 seconds Duration(ns) 4.763
+12.923350 seconds Duration(ns) 12.92335
+
+query R
+SELECT extract(epoch from arrow_cast(2147483648, 'Duration(Microsecond)'))
+----
+2147.483648
+
+query R
+SELECT extract(epoch from arrow_cast(2147483648, 'Duration(Millisecond)'))
+----
+2147483.648
+
+query R
+SELECT extract(epoch from arrow_cast(2147483648, 'Duration(Second)'))
+----
+2147483648
+
+query R
+SELECT extract(epoch from arrow_cast(-2500000000, 'Duration(Nanosecond)'))
+----
+-2.5
+
+query R
+SELECT extract(epoch from arrow_cast(NULL, 'Duration(Nanosecond)'))
+----
+NULL
+
+# NULLs stay NULL and don't change the other rows of the batch
+query R
+SELECT extract(epoch from d) FROM (VALUES
+  (arrow_cast(2500000000, 'Duration(Nanosecond)')),
+  (NULL),
+  (arrow_cast(1500000000, 'Duration(Nanosecond)'))
+) AS t(d)
+----
+2.5
+NULL
+1.5
+
+query RRRR
+SELECT extract(epoch from arrow_cast(9223372036854775807, 'Duration(Nanosecond)')),
+       extract(epoch from arrow_cast(-9223372036854775807, 'Duration(Nanosecond)')),
+       extract(epoch from arrow_cast(9223372036854775807, 'Duration(Second)')),
+       extract(epoch from arrow_cast(-9223372036854775807, 'Duration(Second)'))
+----
+9223372036.854776 -9223372036.854776 9223372036854776000 -9223372036854776000
+
 query I
 SELECT extract(millisecond from arrow_cast(2, 'Duration(Second)'))
 ----