Allow Timestamp outside of jiff's range

vortex-array/src/aggregate_fn/fns/min_max/extension.rs

@@ -10,6 +10,7 @@ use crate::ExecutionCtx;
 use crate::aggregate_fn::NumericalAggregateOpts;
 use crate::arrays::ExtensionArray;
 use crate::arrays::extension::ExtensionArrayExt;
+use crate::dtype::DType;
 use crate::dtype::Nullability;
 use crate::scalar::Scalar;
 
@@ -19,15 +20,23 @@ pub(super) fn accumulate_extension(
     ctx: &mut ExecutionCtx,
 ) -> VortexResult<()> {
     let non_nullable_ext_dtype = array.ext_dtype().with_nullability(Nullability::NonNullable);
-    let local = min_max(
+    let Some(MinMaxResult { min, max }) = min_max(
         array.storage_array(),
         ctx,
         NumericalAggregateOpts::default(),
     )?
-    .map(|MinMaxResult { min, max }| MinMaxResult {
-        min: Scalar::extension_ref(non_nullable_ext_dtype.clone(), min),
-        max: Scalar::extension_ref(non_nullable_ext_dtype, max),
-    });
+    else {
+        return Ok(());
+    };
+
+    let ext_dtype = DType::Extension(non_nullable_ext_dtype);
+    let local = match (
+        Scalar::try_new(ext_dtype.clone(), min.into_value()),
+        Scalar::try_new(ext_dtype, max.into_value()),
+    ) {
+        (Ok(min), Ok(max)) => Some(MinMaxResult { min, max }),
+        _ => None,
+    };
     partial.merge(local);
     Ok(())
 }

vortex-array/src/aggregate_fn/fns/min_max/mod.rs

@@ -465,6 +465,7 @@ mod tests {
     use crate::arrays::ChunkedArray;
     use crate::arrays::ConstantArray;
     use crate::arrays::DecimalArray;
+    use crate::arrays::ExtensionArray;
     use crate::arrays::FixedSizeListArray;
     use crate::arrays::ListArray;
     use crate::arrays::NullArray;
@@ -476,7 +477,10 @@ mod tests {
     use crate::dtype::PType;
     use crate::expr::stats::Precision;
     use crate::expr::stats::Stat;
+    use crate::extension::datetime::TimeUnit;
+    use crate::extension::datetime::Timestamp;
     use crate::scalar::DecimalValue;
+    use crate::scalar::PValue;
     use crate::scalar::Scalar;
     use crate::scalar::ScalarValue;
     use crate::validity::Validity;
@@ -1092,4 +1096,30 @@ mod tests {
         );
         Ok(())
     }
+
+    #[test]
+    fn test_timestamp_min_max_beyond_jiff_range() -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+        let ext_dtype = Timestamp::new(TimeUnit::Microseconds, Nullability::NonNullable).erased();
+        // 9999-12-30T22:00:00Z is jiff's maximum, 9999-12-31T00:00:00Z is past it.
+        let in_range = 253_402_207_200_000_000i64;
+        let out_of_range = 253_402_214_400_000_000i64;
+        let storage = buffer![out_of_range, in_range].into_array();
+        let array = ExtensionArray::try_new(ext_dtype.clone(), storage)?.into_array();
+
+        let result = min_max(&array, &mut ctx, NumericalAggregateOpts::default())?
+            .vortex_expect("should have result");
+        let dtype = DType::Extension(ext_dtype);
+        let expected_min = Scalar::try_new(
+            dtype.clone(),
+            Some(ScalarValue::Primitive(PValue::I64(in_range))),
+        )?;
+        let expected_max = Scalar::try_new(
+            dtype,
+            Some(ScalarValue::Primitive(PValue::I64(out_of_range))),
+        )?;
+        assert_eq!(result.min, expected_min);
+        assert_eq!(result.max, expected_max);
+        Ok(())
+    }
 }

vortex-array/src/arrays/extension/vtable/operations.rs

@@ -8,6 +8,7 @@ use crate::array::ArrayView;
 use crate::array::OperationsVTable;
 use crate::arrays::Extension;
 use crate::arrays::extension::ExtensionArrayExt;
+use crate::dtype::DType;
 use crate::scalar::Scalar;
 
 impl OperationsVTable<Extension> for Extension {
@@ -16,9 +17,10 @@ impl OperationsVTable<Extension> for Extension {
         index: usize,
         ctx: &mut ExecutionCtx,
     ) -> VortexResult<Scalar> {
-        Ok(Scalar::extension_ref(
-            array.ext_dtype().clone(),
-            array.storage_array().execute_scalar(index, ctx)?,
-        ))
+        let storage_scalar = array.storage_array().execute_scalar(index, ctx)?;
+        Scalar::try_new(
+            DType::Extension(array.ext_dtype().clone()),
+            storage_scalar.into_value(),
+        )
     }
 }

vortex-array/src/extension/datetime/timestamp.rs

@@ -91,20 +91,44 @@ pub enum TimestampValue<'a> {
 
 impl fmt::Display for TimestampValue<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let (span, tz) = match self {
-            TimestampValue::Seconds(v, tz) => (Span::new().seconds(*v), *tz),
-            TimestampValue::Milliseconds(v, tz) => (Span::new().milliseconds(*v), *tz),
-            TimestampValue::Microseconds(v, tz) => (Span::new().microseconds(*v), *tz),
-            TimestampValue::Nanoseconds(v, tz) => (Span::new().nanoseconds(*v), *tz),
+        let (span, raw, unit, tz) = match self {
+            TimestampValue::Seconds(v, tz) => {
+                (Span::new().try_seconds(*v), *v, TimeUnit::Seconds, *tz)
+            }
+            TimestampValue::Milliseconds(v, tz) => (
+                Span::new().try_milliseconds(*v),
+                *v,
+                TimeUnit::Milliseconds,
+                *tz,
+            ),
+            TimestampValue::Microseconds(v, tz) => (
+                Span::new().try_microseconds(*v),
+                *v,
+                TimeUnit::Microseconds,
+                *tz,
+            ),
+            TimestampValue::Nanoseconds(v, tz) => (
+                Span::new().try_nanoseconds(*v),
+                *v,
+                TimeUnit::Nanoseconds,
+                *tz,
+            ),
+        };
+
+        let Some(ts) = span
+            .ok()
+            .and_then(|span| jiff::Timestamp::UNIX_EPOCH.checked_add(span).ok())
+        else {
+            // storage values may exceed jiff's Timestamp range
+            return write!(f, "{raw}{unit} since Unix epoch");
         };
-        let ts = jiff::Timestamp::UNIX_EPOCH + span;
 
         match tz {
             None => write!(f, "{ts}"),
-            Some(tz) => {
-                let adjusted_ts = ts.in_tz(tz.as_ref()).vortex_expect("unknown timezone");
-                write!(f, "{adjusted_ts}",)
-            }
+            Some(tz) => match ts.in_tz(tz.as_ref()) {
+                Ok(adjusted_ts) => write!(f, "{adjusted_ts}"),
+                Err(_) => write!(f, "{ts}"),
+            },
         }
     }
 }
@@ -226,37 +250,18 @@ impl ExtVTable for Timestamp {
         let ts_value = storage_value.as_primitive().cast::<i64>()?;
         let tz = metadata.tz.as_ref();
 
-        let (span, value) = match metadata.unit {
-            TimeUnit::Nanoseconds => (
-                Span::new().nanoseconds(ts_value),
-                TimestampValue::Nanoseconds(ts_value, tz),
-            ),
-            TimeUnit::Microseconds => (
-                Span::new().microseconds(ts_value),
-                TimestampValue::Microseconds(ts_value, tz),
-            ),
-            TimeUnit::Milliseconds => (
-                Span::new().milliseconds(ts_value),
-                TimestampValue::Milliseconds(ts_value, tz),
-            ),
-            TimeUnit::Seconds => (
-                Span::new().seconds(ts_value),
-                TimestampValue::Seconds(ts_value, tz),
-            ),
-            TimeUnit::Days => vortex_bail!("Timestamp does not support Days time unit"),
-        };
-
-        // Validate the storage value is within the valid range for Timestamp.
-        let ts = jiff::Timestamp::UNIX_EPOCH
-            .checked_add(span)
-            .map_err(|e| vortex_err!("Invalid timestamp scalar: {}", e))?;
-
         if let Some(tz) = tz {
-            ts.in_tz(tz.as_ref())
+            jiff::tz::TimeZone::get(tz.as_ref())
                 .map_err(|e| vortex_err!("Invalid timezone for timestamp scalar: {}", e))?;
         }
 
-        Ok(value)
+        Ok(match metadata.unit {
+            TimeUnit::Nanoseconds => TimestampValue::Nanoseconds(ts_value, tz),
+            TimeUnit::Microseconds => TimestampValue::Microseconds(ts_value, tz),
+            TimeUnit::Milliseconds => TimestampValue::Milliseconds(ts_value, tz),
+            TimeUnit::Seconds => TimestampValue::Seconds(ts_value, tz),
+            TimeUnit::Days => vortex_bail!("Timestamp does not support Days time unit"),
+        })
     }
 }
 
@@ -282,6 +287,17 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn validate_timestamp_scalar_beyond_jiff_range() -> VortexResult<()> {
+        let dtype = DType::Extension(Timestamp::new(TimeUnit::Microseconds, Nullable).erased());
+        Scalar::try_new(
+            dtype,
+            Some(ScalarValue::Primitive(PValue::I64(253_402_214_400_000_000))),
+        )?;
+
+        Ok(())
+    }
+
     #[cfg_attr(miri, ignore)]
     #[test]
     fn reject_timestamp_with_invalid_timezone() {
@@ -321,6 +337,20 @@ mod tests {
         );
     }
 
+    #[cfg_attr(miri, ignore)]
+    #[test]
+    fn display_timestamp_scalar_beyond_jiff_range() {
+        let dtype = DType::Extension(Timestamp::new(TimeUnit::Microseconds, Nullable).erased());
+        let scalar = Scalar::new(
+            dtype,
+            Some(ScalarValue::Primitive(PValue::I64(253_402_214_400_000_000))),
+        );
+        assert_eq!(
+            format!("{}", scalar.as_extension()),
+            "253402214400000000µs since Unix epoch"
+        );
+    }
+
     #[test]
     fn least_supertype_timestamp_units() {
         use crate::dtype::Nullability::NonNullable;

vortex-file/tests/issue_8860_timestamp_stats.rs

@@ -0,0 +1,58 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+#![expect(clippy::tests_outside_test_module)]
+
+use std::sync::LazyLock;
+
+use vortex_array::IntoArray;
+use vortex_array::arrays::ExtensionArray;
+use vortex_array::arrays::PrimitiveArray;
+use vortex_array::arrays::StructArray;
+use vortex_array::dtype::Nullability::NonNullable;
+use vortex_array::extension::datetime::TimeUnit;
+use vortex_array::extension::datetime::Timestamp;
+use vortex_error::VortexResult;
+use vortex_file::WriteOptionsSessionExt;
+use vortex_io::session::RuntimeSession;
+use vortex_layout::session::LayoutSession;
+use vortex_session::VortexSession;
+
+mod common;
+
+use common::enable_all_registered_array_encodings;
+
+static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
+    let session = vortex_array::array_session()
+        .with::<LayoutSession>()
+        .with::<RuntimeSession>();
+    vortex_file::register_default_encodings(&session);
+    enable_all_registered_array_encodings(&session);
+    session
+});
+
+async fn write_micros(value: i64) -> VortexResult<()> {
+    let ext_dtype = Timestamp::new(TimeUnit::Microseconds, NonNullable).erased();
+    let storage = PrimitiveArray::from_iter([value]).into_array();
+    let ts = ExtensionArray::try_new(ext_dtype, storage)?.into_array();
+    let data = StructArray::from_fields(&[("ts", ts)])?.into_array();
+
+    let mut bytes = Vec::new();
+    SESSION
+        .write_options()
+        .write(&mut bytes, data.to_array_stream())
+        .await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn write_timestamp_last_day_of_9999_does_not_panic() -> VortexResult<()> {
+    // 9999-12-31T00:00:00Z is past jiff's max of 9999-12-30T22:00:00Z
+    write_micros(253_402_214_400_000_000).await
+}
+
+#[tokio::test]
+async fn write_timestamp_in_jiff_range() -> VortexResult<()> {
+    // 9999-12-30T22:00:00Z jiff's maximum
+    write_micros(253_402_207_200_000_000).await
+}