diff --git a/vortex-array/src/aggregate_fn/fns/min_max/extension.rs b/vortex-array/src/aggregate_fn/fns/min_max/extension.rs index e981c7c6af1..1267dbc1683 100644 --- a/vortex-array/src/aggregate_fn/fns/min_max/extension.rs +++ b/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(()) } diff --git a/vortex-array/src/aggregate_fn/fns/min_max/mod.rs b/vortex-array/src/aggregate_fn/fns/min_max/mod.rs index a5091cf4a9f..71f8bcffe11 100644 --- a/vortex-array/src/aggregate_fn/fns/min_max/mod.rs +++ b/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(()) + } } diff --git a/vortex-array/src/arrays/extension/vtable/operations.rs b/vortex-array/src/arrays/extension/vtable/operations.rs index 66de94b596a..a5d461de0af 100644 --- a/vortex-array/src/arrays/extension/vtable/operations.rs +++ b/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 for Extension { @@ -16,9 +17,10 @@ impl OperationsVTable for Extension { index: usize, ctx: &mut ExecutionCtx, ) -> VortexResult { - 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(), + ) } } diff --git a/vortex-array/src/extension/datetime/timestamp.rs b/vortex-array/src/extension/datetime/timestamp.rs index d75820ee181..e4556757eaa 100644 --- a/vortex-array/src/extension/datetime/timestamp.rs +++ b/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::()?; 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; diff --git a/vortex-file/tests/issue_8860_timestamp_stats.rs b/vortex-file/tests/issue_8860_timestamp_stats.rs new file mode 100644 index 00000000000..5f37b06415e --- /dev/null +++ b/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 = LazyLock::new(|| { + let session = vortex_array::array_session() + .with::() + .with::(); + 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 +}