Skip to content

Commit 950de57

Browse files
committed
refactor: isolate bitpacked v1 serialization
Signed-off-by: "Matt Katz" <mhkatz97@gmail.com>
1 parent c6678f3 commit 950de57

5 files changed

Lines changed: 225 additions & 91 deletions

File tree

‎encodings/fastlanes/src/bitpacking/mod.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,13 @@ mod plugin;
1717
mod vtable;
1818

1919
pub(crate) use plugin::BitPackedPatchedPlugin;
20+
pub use plugin::BitPackedPlugin;
2021
pub use vtable::BitPacked;
2122
pub use vtable::BitPackedArray;
2223

2324
pub(crate) fn initialize(session: &vortex_session::VortexSession) {
2425
vtable::initialize(session);
2526
}
27+
28+
#[cfg(test)]
29+
mod serde_tests;

‎encodings/fastlanes/src/bitpacking/plugin.rs‎

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

4-
//! A custom [`ArrayPlugin`] that lets you load in and deserialize a `BitPacked` array with interior
5-
//! patches as a `PatchedArray` that wraps a patchless `BitPacked` array.
6-
//!
7-
//! This enables zero-cost backward compatibility with previously written datasets.
4+
//! Serialization plugins for the frozen bit-packed format and its external-patches adapter.
85
6+
use prost::Message;
97
use vortex_array::Array;
108
use vortex_array::ArrayDeserialization;
119
use vortex_array::ArrayId;
@@ -18,32 +16,41 @@ use vortex_array::VortexSessionExecute;
1816
use vortex_array::arrays::Patched;
1917
use vortex_error::VortexResult;
2018
use vortex_error::vortex_ensure;
21-
use vortex_error::vortex_err;
2219
use vortex_session::VortexSession;
2320

2421
use crate::BitPacked;
2522
use crate::BitPackedArrayExt;
23+
use crate::bitpacking::vtable::BitPackedMetadata;
2624

27-
/// Custom deserialization plugin that converts a BitPacked array with interior
28-
/// Patches into a PatchedArray holding a BitPacked array.
25+
/// Serialization boundary for the frozen `fastlanes.bitpacked` wire format.
2926
#[derive(Debug, Clone)]
30-
pub(crate) struct BitPackedPatchedPlugin;
27+
pub struct BitPackedPlugin;
3128

32-
impl ArrayPlugin for BitPackedPatchedPlugin {
29+
impl ArrayPlugin for BitPackedPlugin {
3330
fn id(&self) -> ArrayId {
34-
// We reuse the existing `BitPacked` ID so that we can take over its
35-
// deserialization pathway.
36-
// TODO(joe): dedup method name
3731
ArrayVTable::id(&BitPacked)
3832
}
3933

4034
fn serialize(
4135
&self,
4236
array: &ArrayRef,
43-
session: &VortexSession,
37+
_session: &VortexSession,
4438
) -> VortexResult<Option<ArraySerialization>> {
45-
// delegate to BitPacked VTable for serialization
46-
ArrayPlugin::serialize(&BitPacked, array, session)
39+
let view = array.as_::<BitPacked>();
40+
let metadata = BitPackedMetadata {
41+
bit_width: view.bit_width() as u32,
42+
offset: view.offset() as u32,
43+
patches: view
44+
.patches()
45+
.map(|p| p.to_metadata(view.len(), view.dtype()))
46+
.transpose()?,
47+
}
48+
.encode_to_vec();
49+
Ok(Some(ArraySerialization::from_array(
50+
self.id(),
51+
array,
52+
metadata,
53+
)))
4754
}
4855

4956
fn deserialize(
@@ -54,18 +61,50 @@ impl ArrayPlugin for BitPackedPatchedPlugin {
5461
vortex_ensure!(
5562
parts.serialized_id == self.id(),
5663
"BitPacked plugin does not recognize serialized ID {}",
57-
parts.serialized_id,
64+
parts.serialized_id
5865
);
59-
let bitpacked = Array::<BitPacked>::try_from_parts(ArrayVTable::deserialize(
66+
Ok(Array::<BitPacked>::try_from_parts(ArrayVTable::deserialize(
6067
&BitPacked,
6168
parts.dtype,
6269
parts.len,
6370
parts.metadata,
6471
parts.buffers,
6572
parts.children,
6673
session,
67-
)?)
68-
.map_err(|_| vortex_err!("BitPacked plugin should only deserialize fastlanes.bitpacked"))?;
74+
)?)?
75+
.into_array())
76+
}
77+
}
78+
79+
/// Custom deserialization plugin that converts a BitPacked array with interior
80+
/// Patches into a PatchedArray holding a BitPacked array.
81+
#[derive(Debug, Clone)]
82+
pub(crate) struct BitPackedPatchedPlugin;
83+
84+
impl ArrayPlugin for BitPackedPatchedPlugin {
85+
fn id(&self) -> ArrayId {
86+
// We reuse the existing `BitPacked` ID so that we can take over its
87+
// deserialization pathway.
88+
// TODO(joe): dedup method name
89+
ArrayVTable::id(&BitPacked)
90+
}
91+
92+
fn serialize(
93+
&self,
94+
array: &ArrayRef,
95+
session: &VortexSession,
96+
) -> VortexResult<Option<ArraySerialization>> {
97+
// Both plugins share the same wire contract.
98+
BitPackedPlugin.serialize(array, session)
99+
}
100+
101+
fn deserialize(
102+
&self,
103+
parts: ArrayDeserialization<'_>,
104+
session: &VortexSession,
105+
) -> VortexResult<ArrayRef> {
106+
let bitpacked = BitPackedPlugin.deserialize(parts, session)?;
107+
let bitpacked = bitpacked.as_::<BitPacked>().into_owned();
69108

70109
// Create a new BitPackedArray without the interior patches installed.
71110
let Some(patches) = bitpacked.patches() else {
Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// SPDX-FileCopyrightText: Copyright the Vortex contributors
3+
4+
use std::sync::LazyLock;
5+
6+
use prost::Message;
7+
use vortex_array::ArrayContext;
8+
use vortex_array::ArrayId;
9+
use vortex_array::ArrayRef;
10+
use vortex_array::ArrayVTable;
11+
use vortex_array::VortexSessionExecute;
12+
use vortex_array::arrays::PrimitiveArray;
13+
use vortex_array::assert_arrays_eq;
14+
use vortex_array::serde::SerializeOptions;
15+
use vortex_array::serde::SerializedArray;
16+
use vortex_array::session::ArraySessionExt;
17+
use vortex_buffer::ByteBufferMut;
18+
use vortex_error::VortexResult;
19+
use vortex_error::vortex_err;
20+
use vortex_session::VortexSession;
21+
use vortex_session::registry::ReadContext;
22+
23+
use crate::BitPacked;
24+
use crate::BitPackedArray;
25+
use crate::bitpacking::bitpack_compress::bitpack_to_best_bit_width;
26+
use crate::bitpacking::vtable::BitPackedMetadata;
27+
28+
static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
29+
let session = vortex_array::array_session();
30+
crate::initialize(&session);
31+
session
32+
});
33+
34+
fn serde_roundtrip(array: &BitPackedArray) -> VortexResult<(ArrayId, Vec<u8>, ArrayRef)> {
35+
let array_ref = array.as_array();
36+
let serialization = SESSION
37+
.array_serialize(array_ref)?
38+
.ok_or_else(|| vortex_err!("BitPacked must serialize"))?;
39+
let array_ctx = ArrayContext::empty();
40+
let buffers = array_ref.serialize(&array_ctx, &SESSION, &SerializeOptions::default())?;
41+
let mut bytes = ByteBufferMut::empty();
42+
for buffer in buffers {
43+
bytes.extend_from_slice(&buffer);
44+
}
45+
let read = SerializedArray::try_from(bytes.freeze())?.decode(
46+
array_ref.dtype(),
47+
array_ref.len(),
48+
&ReadContext::new(array_ctx.to_ids()),
49+
&SESSION,
50+
)?;
51+
Ok((serialization.serialized_id, serialization.metadata, read))
52+
}
53+
54+
#[test]
55+
fn uniform_widths_serialize_as_original_format() -> VortexResult<()> {
56+
let mut ctx = SESSION.create_execution_ctx();
57+
let values: Vec<u32> = (0..3000).map(|i| i % 128).collect();
58+
let packed =
59+
bitpack_to_best_bit_width(&PrimitiveArray::from_iter(values.iter().copied()), &mut ctx)?;
60+
assert!(packed.as_array().children().is_empty());
61+
assert!(
62+
SESSION
63+
.array_serialize(packed.as_array())?
64+
.ok_or_else(|| vortex_err!("must serialize"))?
65+
.children
66+
.is_empty()
67+
);
68+
let (id, metadata, read) = serde_roundtrip(&packed)?;
69+
assert_eq!(id, ArrayVTable::id(&BitPacked));
70+
let original = BitPackedMetadata {
71+
bit_width: 7,
72+
offset: 0,
73+
patches: None,
74+
}
75+
.encode_to_vec();
76+
assert_eq!(metadata, original);
77+
assert_arrays_eq!(
78+
read,
79+
PrimitiveArray::from_iter(values.iter().copied()),
80+
&mut ctx
81+
);
82+
Ok(())
83+
}

‎encodings/fastlanes/src/bitpacking/vtable/mod.rs‎

Lines changed: 79 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -161,20 +161,10 @@ impl VTable for BitPacked {
161161
}
162162

163163
fn serialize(
164-
array: ArrayView<'_, Self>,
164+
_array: ArrayView<'_, Self>,
165165
_session: &VortexSession,
166166
) -> VortexResult<Option<Vec<u8>>> {
167-
Ok(Some(
168-
BitPackedMetadata {
169-
bit_width: array.bit_width() as u32,
170-
offset: array.offset() as u32,
171-
patches: array
172-
.patches()
173-
.map(|p| p.to_metadata(array.len(), array.dtype()))
174-
.transpose()?,
175-
}
176-
.encode_to_vec(),
177-
))
167+
vortex_bail!("BitPacked serialization requires BitPackedPlugin")
178168
}
179169

180170
fn deserialize(
@@ -187,71 +177,24 @@ impl VTable for BitPacked {
187177
_session: &VortexSession,
188178
) -> VortexResult<ArrayParts<Self>> {
189179
let metadata = BitPackedMetadata::decode(metadata)?;
190-
if buffers.len() != 1 {
191-
vortex_bail!("Expected 1 buffer, got {}", buffers.len());
192-
}
193-
let packed = buffers[0].clone();
194-
195-
let load_validity = |child_idx: usize| {
196-
if children.len() == child_idx {
197-
Ok(Validity::from(dtype.nullability()))
198-
} else if children.len() == child_idx + 1 {
199-
let validity = children.get(child_idx, &Validity::DTYPE, len)?;
200-
Ok(Validity::Array(validity))
201-
} else {
202-
vortex_bail!(
203-
"Expected {} or {} children, got {}",
204-
child_idx,
205-
child_idx + 1,
206-
children.len()
207-
);
208-
}
209-
};
210-
211-
let validity_idx = match &metadata.patches {
212-
None => 0,
213-
Some(patches_meta) if patches_meta.chunk_offsets_dtype()?.is_some() => 3,
214-
Some(_) => 2,
215-
};
216-
217-
let validity = load_validity(validity_idx)?;
218-
219-
let patches = metadata
220-
.patches
221-
.map(|p| {
222-
let indices = children.get(0, &p.indices_dtype()?, p.len()?)?;
223-
let values = children.get(1, dtype, p.len()?)?;
224-
let chunk_offsets = p
225-
.chunk_offsets_dtype()?
226-
.map(|dtype| children.get(2, &dtype, p.chunk_offsets_len() as usize))
227-
.transpose()?;
228-
229-
Patches::new(len, p.offset()?, indices, values, chunk_offsets)
230-
})
231-
.transpose()?;
180+
let packed = single_buffer(buffers)?;
181+
let offset = offset_from_metadata(metadata.offset)?;
182+
let bit_width = u8::try_from(metadata.bit_width).map_err(|_| {
183+
vortex_err!(
184+
"BitPackedMetadata bit_width {} does not fit in u8",
185+
metadata.bit_width
186+
)
187+
})?;
188+
let (patches, validity, _) =
189+
deserialize_children(children, metadata.patches, dtype, len, 0)?;
232190

233191
let slots = {
234-
let mut s = ArraySlots::with_capacity(4);
192+
let mut s = ArraySlots::with_capacity(BitPackedSlots::COUNT);
235193
PatchesData::push_slots(&mut s, patches.as_ref());
236194
s.push(validity_to_child(&validity, len));
237195
s
238196
};
239-
let data = BitPackedData::try_new(
240-
packed,
241-
patches,
242-
u8::try_from(metadata.bit_width).map_err(|_| {
243-
vortex_err!(
244-
"BitPackedMetadata bit_width {} does not fit in u8",
245-
metadata.bit_width
246-
)
247-
})?,
248-
u16::try_from(metadata.offset).map_err(|_| {
249-
vortex_err!(
250-
"BitPackedMetadata offset {} does not fit in u16",
251-
metadata.offset
252-
)
253-
})?,
254-
)?;
197+
let data = BitPackedData::try_new(packed, patches, bit_width, offset)?;
255198
Ok(ArrayParts::new(self.clone(), dtype.clone(), len, data).with_slots(slots))
256199
}
257200

@@ -299,6 +242,71 @@ impl VTable for BitPacked {
299242
}
300243
}
301244

245+
/// The single packed buffer of a serialized bit-packed array.
246+
pub(crate) fn single_buffer(buffers: &[BufferHandle]) -> VortexResult<BufferHandle> {
247+
vortex_ensure!(
248+
buffers.len() == 1,
249+
"Expected 1 buffer, got {}",
250+
buffers.len()
251+
);
252+
Ok(buffers[0].clone())
253+
}
254+
255+
/// The offset into the first chunk, which the metadata stores as a `u32`.
256+
pub(crate) fn offset_from_metadata(offset: u32) -> VortexResult<u16> {
257+
u16::try_from(offset)
258+
.map_err(|_| vortex_err!("BitPackedMetadata offset {offset} does not fit in u16"))
259+
}
260+
261+
/// Read the patches and validity children that both wire formats share.
262+
///
263+
/// Children run: the patches, then a validity bitmap if there is one, then `trailing` children
264+
/// the caller reads itself. Returns the index of the first trailing child.
265+
pub(crate) fn deserialize_children(
266+
children: &dyn ArrayChildren,
267+
patches: Option<PatchesMetadata>,
268+
dtype: &DType,
269+
len: usize,
270+
trailing: usize,
271+
) -> VortexResult<(Option<Patches>, Validity, usize)> {
272+
let num_patch_children = match &patches {
273+
None => 0,
274+
Some(patches_meta) if patches_meta.chunk_offsets_dtype()?.is_some() => 3,
275+
Some(_) => 2,
276+
};
277+
let num_fixed = num_patch_children + trailing;
278+
let has_validity = match children.len().checked_sub(num_fixed) {
279+
Some(0) => false,
280+
Some(1) => true,
281+
_ => vortex_bail!(
282+
"Expected {num_fixed} or {} children, got {}",
283+
num_fixed + 1,
284+
children.len()
285+
),
286+
};
287+
let validity = if has_validity {
288+
Validity::Array(children.get(num_patch_children, &Validity::DTYPE, len)?)
289+
} else {
290+
Validity::from(dtype.nullability())
291+
};
292+
let patches = patches
293+
.map(|p| {
294+
let indices = children.get(0, &p.indices_dtype()?, p.len()?)?;
295+
let values = children.get(1, dtype, p.len()?)?;
296+
let chunk_offsets = p
297+
.chunk_offsets_dtype()?
298+
.map(|dtype| children.get(2, &dtype, p.chunk_offsets_len() as usize))
299+
.transpose()?;
300+
Patches::new(len, p.offset()?, indices, values, chunk_offsets)
301+
})
302+
.transpose()?;
303+
Ok((
304+
patches,
305+
validity,
306+
num_patch_children + usize::from(has_validity),
307+
))
308+
}
309+
302310
#[derive(Clone, Debug)]
303311
pub struct BitPacked;
304312

0 commit comments

Comments
 (0)