Skip to content

Commit 7385cfd

Browse files
committed
Change DateTimeParts to i32
Signed-off-by: Mikhail Kot <mikhail@spiraldb.com>
1 parent 5c8c397 commit 7385cfd

8 files changed

Lines changed: 226 additions & 38 deletions

File tree

‎Cargo.lock‎

Lines changed: 2 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎encodings/datetime-parts/Cargo.toml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,12 @@ vortex-mask = { workspace = true }
2626
vortex-session = { workspace = true }
2727

2828
[dev-dependencies]
29+
divan = { workspace = true }
30+
rand = { workspace = true }
2931
rstest = { workspace = true }
3032
vortex-array = { workspace = true, features = ["_test-harness"] }
3133
vortex-error = { workspace = true }
34+
35+
[[bench]]
36+
name = "split_temporal"
37+
harness = false
Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
#![expect(clippy::unwrap_used)]
5+
6+
use std::sync::LazyLock;
7+
8+
use divan::Bencher;
9+
use rand::RngExt;
10+
use rand::SeedableRng as _;
11+
use rand::rngs::StdRng;
12+
use vortex_array::IntoArray;
13+
use vortex_array::VortexSessionExecute;
14+
use vortex_array::arrays::PrimitiveArray;
15+
use vortex_array::arrays::TemporalArray;
16+
use vortex_array::extension::datetime::TimeUnit;
17+
use vortex_array::validity::Validity;
18+
use vortex_buffer::Buffer;
19+
use vortex_datetime_parts::split_temporal;
20+
use vortex_session::VortexSession;
21+
22+
fn main() {
23+
divan::main();
24+
}
25+
26+
static SESSION: LazyLock<VortexSession> = LazyLock::new(vortex_array::array_session);
27+
28+
const BENCH_ARGS: &[(usize, TimeUnit)] = &[
29+
(65_536, TimeUnit::Seconds),
30+
(65_536, TimeUnit::Milliseconds),
31+
(65_536, TimeUnit::Microseconds),
32+
(65_536, TimeUnit::Nanoseconds),
33+
];
34+
35+
#[divan::bench(args = BENCH_ARGS)]
36+
fn split(bencher: Bencher, args: (usize, TimeUnit)) {
37+
let (n, unit) = args;
38+
let divisor: i64 = match unit {
39+
TimeUnit::Seconds => 1,
40+
TimeUnit::Milliseconds => 1_000,
41+
TimeUnit::Microseconds => 1_000_000,
42+
TimeUnit::Nanoseconds => 1_000_000_000,
43+
TimeUnit::Days => unreachable!(),
44+
};
45+
let mut rng = StdRng::seed_from_u64(0);
46+
let timestamps = Buffer::from_iter((0..n).map(|_| {
47+
rng.random_range(1_500_000_000i64..1_800_000_000) * divisor + rng.random_range(0..divisor)
48+
}));
49+
let array = TemporalArray::new_timestamp(
50+
PrimitiveArray::new(timestamps, Validity::NonNullable).into_array(),
51+
unit,
52+
Some("UTC".into()),
53+
);
54+
55+
bencher
56+
.with_inputs(|| (array.clone(), SESSION.create_execution_ctx()))
57+
.bench_values(|(array, mut ctx)| split_temporal(array, &mut ctx).unwrap())
58+
}

‎encodings/datetime-parts/src/compress.rs‎

Lines changed: 91 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,26 @@
11
// SPDX-License-Identifier: Apache-2.0
22
// SPDX-FileCopyrightText: Copyright the Vortex contributors
33

4+
use std::mem::MaybeUninit;
5+
46
use vortex_array::ArrayRef;
57
use vortex_array::ExecutionCtx;
68
use vortex_array::IntoArray;
9+
use vortex_array::arrays::ConstantArray;
710
use vortex_array::arrays::PrimitiveArray;
811
use vortex_array::arrays::TemporalArray;
912
use vortex_array::builtins::ArrayBuiltins;
1013
use vortex_array::dtype::DType;
1114
use vortex_array::dtype::PType;
15+
use vortex_array::extension::datetime::TimeUnit;
1216
use vortex_buffer::BufferMut;
1317
use vortex_error::VortexResult;
18+
use vortex_error::vortex_bail;
1419

1520
use crate::timestamp;
21+
/// All parts are stored as i32. seconds and subseconds always fit in i32.
22+
/// For days we have a upper day limitation due to "jiff" validation so days
23+
/// also fit in i32.
1624
pub struct TemporalParts {
1725
pub days: ArrayRef,
1826
pub seconds: ArrayRef,
@@ -39,25 +47,98 @@ pub fn split_temporal(array: TemporalArray, ctx: &mut ExecutionCtx) -> VortexRes
3947
))?
4048
.execute::<PrimitiveArray>(ctx)?;
4149

50+
let time_unit = array.temporal_metadata().time_unit();
51+
if matches!(time_unit, TimeUnit::Days) {
52+
vortex_bail!("Cannot handle day-level data");
53+
}
54+
4255
let length = timestamps.len();
43-
let mut days = BufferMut::with_capacity(length);
44-
let mut seconds = BufferMut::with_capacity(length);
45-
let mut subseconds = BufferMut::with_capacity(length);
46-
47-
for &ts in timestamps.as_slice::<i64>() {
48-
let ts_parts = timestamp::split(ts, array.temporal_metadata().time_unit())?;
49-
days.push(ts_parts.days);
50-
seconds.push(ts_parts.seconds);
51-
subseconds.push(ts_parts.subseconds);
56+
57+
// If we don't [..length], compiler can't infer all 3 or 4 slices are the
58+
// same length, and zip() in split_slice_* checks iterator boundaries.
59+
let timestamps = &timestamps.as_slice::<i64>()[..length];
60+
61+
let mut days: BufferMut<i32> = BufferMut::with_capacity(timestamps.len());
62+
let mut seconds: BufferMut<i32> = BufferMut::with_capacity(timestamps.len());
63+
let days_ptr = &mut days.spare_capacity_mut()[..length];
64+
let seconds_ptr = &mut seconds.spare_capacity_mut()[..length];
65+
66+
if matches!(time_unit, TimeUnit::Seconds) {
67+
split_slice_seconds(days_ptr, seconds_ptr, timestamps);
68+
69+
// SAFETY: all items in [0; length) are filled in split_slice_seconds
70+
unsafe {
71+
days.set_len(length);
72+
seconds.set_len(length);
73+
}
74+
75+
return Ok(TemporalParts {
76+
days: PrimitiveArray::new(days.freeze(), temporal_values.validity()?).into_array(),
77+
seconds: seconds.into_array(),
78+
subseconds: ConstantArray::new(0, length).into_array(),
79+
});
80+
}
81+
82+
let mut subseconds: BufferMut<i32> = BufferMut::with_capacity(timestamps.len());
83+
let subseconds_ptr = &mut subseconds.spare_capacity_mut()[..length];
84+
85+
match time_unit {
86+
TimeUnit::Nanoseconds => {
87+
split_slice::<1_000_000_000>(days_ptr, seconds_ptr, subseconds_ptr, timestamps)
88+
}
89+
TimeUnit::Microseconds => {
90+
split_slice::<1_000_000>(days_ptr, seconds_ptr, subseconds_ptr, timestamps)
91+
}
92+
TimeUnit::Milliseconds => {
93+
split_slice::<1_000>(days_ptr, seconds_ptr, subseconds_ptr, timestamps)
94+
}
95+
_ => unreachable!("Handled before"),
96+
};
97+
98+
// SAFETY: all items in [0; length) are filled in split_slice
99+
unsafe {
100+
days.set_len(length);
101+
seconds.set_len(length);
102+
subseconds.set_len(length);
52103
}
53104

54105
Ok(TemporalParts {
55-
days: PrimitiveArray::new(days, temporal_values.validity()?).into_array(),
106+
days: PrimitiveArray::new(days.freeze(), temporal_values.validity()?).into_array(),
56107
seconds: seconds.into_array(),
57108
subseconds: subseconds.into_array(),
58109
})
59110
}
60111

112+
#[inline]
113+
fn split_slice<const DIVISOR: i64>(
114+
days: &mut [MaybeUninit<i32>],
115+
seconds: &mut [MaybeUninit<i32>],
116+
subseconds: &mut [MaybeUninit<i32>],
117+
timestamps: &[i64],
118+
) {
119+
for (((day, second), subseconds), ts) in
120+
days.iter_mut().zip(seconds).zip(subseconds).zip(timestamps)
121+
{
122+
let parts = timestamp::split_with_divisor::<DIVISOR>(*ts);
123+
day.write(parts.days);
124+
second.write(parts.seconds);
125+
subseconds.write(parts.subseconds);
126+
}
127+
}
128+
129+
#[inline]
130+
fn split_slice_seconds(
131+
days: &mut [MaybeUninit<i32>],
132+
seconds: &mut [MaybeUninit<i32>],
133+
timestamps: &[i64],
134+
) {
135+
for ((day, second), ts) in days.iter_mut().zip(seconds).zip(timestamps) {
136+
let parts = timestamp::split_with_divisor::<1>(*ts);
137+
day.write(parts.days);
138+
second.write(parts.seconds);
139+
}
140+
}
141+
61142
#[cfg(test)]
62143
mod tests {
63144
use rstest::rstest;

‎encodings/datetime-parts/src/compute/compare.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,7 @@ fn compare_gt(
176176

177177
fn compare_dtp(
178178
lhs: &ArrayRef,
179-
rhs: i64,
179+
rhs: i32,
180180
operator: CompareOperator,
181181
nullability: Nullability,
182182
) -> VortexResult<ArrayRef> {

‎encodings/datetime-parts/src/compute/rules.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ impl ArrayParentReduceRule<DateTimeParts> for DTPComparisonPushDownRule {
143143

144144
/// Try to extract the days value from a constant timestamp.
145145
/// Returns None if the constant is not a timestamp or has non-zero seconds/subseconds.
146-
fn try_extract_days_constant(array: &ArrayRef) -> Option<i64> {
146+
fn try_extract_days_constant(array: &ArrayRef) -> Option<i32> {
147147
let constant = array.as_constant()?;
148148

149149
// Extract the timestamp value

‎encodings/datetime-parts/src/ops.rs‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -37,24 +37,24 @@ impl OperationsVTable<DateTimeParts> for DateTimeParts {
3737
return Ok(Scalar::null(DType::Extension(ext)));
3838
}
3939

40-
let days: i64 = array
40+
let days: i32 = array
4141
.days()
4242
.execute_scalar(index, ctx)?
4343
.as_primitive()
44-
.as_::<i64>()
45-
.vortex_expect("days fits in i64");
46-
let seconds: i64 = array
44+
.as_::<i32>()
45+
.vortex_expect("days fits in i32");
46+
let seconds: i32 = array
4747
.seconds()
4848
.execute_scalar(index, ctx)?
4949
.as_primitive()
50-
.as_::<i64>()
51-
.vortex_expect("seconds fits in i64");
52-
let subseconds: i64 = array
50+
.as_::<i32>()
51+
.vortex_expect("seconds fits in i32");
52+
let subseconds: i32 = array
5353
.subseconds()
5454
.execute_scalar(index, ctx)?
5555
.as_primitive()
56-
.as_::<i64>()
57-
.vortex_expect("subseconds fits in i64");
56+
.as_::<i32>()
57+
.vortex_expect("subseconds fits in i32");
5858

5959
let ts = timestamp::combine(
6060
TimestampParts {

‎encodings/datetime-parts/src/timestamp.rs‎

Lines changed: 58 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
// SPDX-License-Identifier: Apache-2.0
22
// SPDX-FileCopyrightText: Copyright the Vortex contributors
33

4+
use num_traits::AsPrimitive;
45
use vortex_array::extension::datetime::TimeUnit;
56
use vortex_error::VortexResult;
67
use vortex_error::vortex_bail;
@@ -11,13 +12,13 @@ pub const SECONDS_PER_DAY: i64 = 86_400; // 24 * 60 * 60
1112
/// Parts of a Unix timestamp (time since 1970-01-01 00:00:00 UTC).
1213
///
1314
/// Broken down into:
14-
/// * Days since epoch
15+
/// * Days since epoch (max range is "jiff"'s range)
1516
/// * Seconds within day (0-86399)
1617
/// * Subseconds (range depends on `TimeUnit`)
1718
pub struct TimestampParts {
18-
pub days: i64,
19-
pub seconds: i64,
20-
pub subseconds: i64,
19+
pub days: i32,
20+
pub seconds: i32,
21+
pub subseconds: i32,
2122
}
2223

2324
/// Splits a Unix timestamp into its component parts.
@@ -31,22 +32,29 @@ pub struct TimestampParts {
3132
///
3233
/// Returns an error if `time_unit` is days, which cannot be split.
3334
pub fn split(timestamp: i64, time_unit: TimeUnit) -> VortexResult<TimestampParts> {
34-
let divisor = match time_unit {
35-
TimeUnit::Nanoseconds => 1_000_000_000,
36-
TimeUnit::Microseconds => 1_000_000,
37-
TimeUnit::Milliseconds => 1_000,
38-
TimeUnit::Seconds => 1,
35+
Ok(match time_unit {
36+
TimeUnit::Nanoseconds => split_with_divisor::<1_000_000_000>(timestamp),
37+
TimeUnit::Microseconds => split_with_divisor::<1_000_000>(timestamp),
38+
TimeUnit::Milliseconds => split_with_divisor::<1_000>(timestamp),
39+
TimeUnit::Seconds => split_with_divisor::<1>(timestamp),
3940
TimeUnit::Days => vortex_bail!("Cannot handle day-level data"),
40-
};
41-
42-
let ticks_per_day = SECONDS_PER_DAY * divisor;
43-
Ok(TimestampParts {
44-
days: timestamp / ticks_per_day,
45-
seconds: (timestamp % ticks_per_day) / divisor,
46-
subseconds: (timestamp % ticks_per_day) % divisor,
4741
})
4842
}
4943

44+
/// Split a Unix timestamp into parts using a constant DIVISOR.
45+
#[inline]
46+
pub(crate) fn split_with_divisor<const DIVISOR: i64>(timestamp: i64) -> TimestampParts {
47+
let days = timestamp / (SECONDS_PER_DAY * DIVISOR);
48+
let total_seconds = timestamp / DIVISOR;
49+
let seconds = total_seconds - days * SECONDS_PER_DAY;
50+
let subseconds = timestamp - total_seconds * DIVISOR;
51+
TimestampParts {
52+
days: days.as_(),
53+
seconds: seconds.as_(),
54+
subseconds: subseconds.as_(),
55+
}
56+
}
57+
5058
/// Combines timestamp parts back into a Unix timestamp.
5159
///
5260
/// # Arguments
@@ -66,13 +74,46 @@ pub fn combine(ts_parts: TimestampParts, time_unit: TimeUnit) -> i64 {
6674
TimeUnit::Days => vortex_panic!("Cannot handle day-level data"),
6775
};
6876

69-
ts_parts.days * SECONDS_PER_DAY * divisor + ts_parts.seconds * divisor + ts_parts.subseconds
77+
ts_parts.days as i64 * SECONDS_PER_DAY * divisor
78+
+ ts_parts.seconds as i64 * divisor
79+
+ ts_parts.subseconds as i64
7080
}
7181

7282
#[cfg(test)]
7383
mod tests {
84+
use rstest::rstest;
85+
7486
use super::*;
7587

88+
#[rstest]
89+
#[case(TimeUnit::Seconds, 1)]
90+
#[case(TimeUnit::Milliseconds, 1_000)]
91+
#[case(TimeUnit::Microseconds, 1_000_000)]
92+
#[case(TimeUnit::Nanoseconds, 1_000_000_000)]
93+
fn test_split_divmod(#[case] unit: TimeUnit, #[case] divisor: i64) -> VortexResult<()> {
94+
let ticks_per_day = SECONDS_PER_DAY * divisor;
95+
for ts in [
96+
0,
97+
1,
98+
-1,
99+
divisor - 1,
100+
3723 * divisor + divisor / 2,
101+
-(3723 * divisor + divisor / 2),
102+
ticks_per_day - 1,
103+
ticks_per_day,
104+
-ticks_per_day,
105+
i64::MAX,
106+
i64::MIN,
107+
] {
108+
let parts = split(ts, unit)?;
109+
assert_eq!(parts.days, (ts / ticks_per_day) as i32);
110+
assert_eq!(parts.seconds, ((ts % ticks_per_day) / divisor) as i32);
111+
assert_eq!(parts.subseconds, ((ts % ticks_per_day) % divisor) as i32);
112+
assert_eq!(combine(parts, unit), ts);
113+
}
114+
Ok(())
115+
}
116+
76117
#[test]
77118
fn test_split_seconds() {
78119
// 1970-01-02 01:02:03 UTC

0 commit comments

Comments
 (0)