diff --git a/Cargo.lock b/Cargo.lock index c764cd8eab7..d9f80fc40b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6045,9 +6045,9 @@ checksum = "cfe21416a02c693fb9f980befcb230ecc70b0b3d1cc4abf88b9675c4c1457f0c" [[package]] name = "onpair" -version = "0.1.1" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "daf79077d8decdf5714242f1813f844f23c80a0cf009aba52881effb4c36ecbe" +checksum = "e546e7cd983c9998a5a52e9c7eecaf38e518f2e6fc9c9ec649e8b53748a96bb0" dependencies = [ "hashbrown 0.16.1", "rand 0.9.5", @@ -6587,7 +6587,7 @@ version = "0.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "044b1fa4f259f4df9ad5078e587b208f5d288a25407575fcddb9face30c7c692" dependencies = [ - "rand 0.9.5", + "rand 0.8.7", "socket2", "thiserror 2.0.19", ] diff --git a/Cargo.toml b/Cargo.toml index 997b3281d25..d096bce0c35 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -196,7 +196,7 @@ object_store = { version = "0.13.2", default-features = false } object_store_opendal = "0.57.0" once_cell = "1.21" oneshot = { version = "0.2.0", features = ["async"] } -onpair = "0.1.1" +onpair = "0.2.0" opendal = { version = "0.57.0", default-features = false } opentelemetry = "0.32.0" opentelemetry-otlp = "0.32.0" diff --git a/benchmarks/string-bench/src/codec.rs b/benchmarks/string-bench/src/codec.rs index dc8364b73d1..046cb230bb2 100644 --- a/benchmarks/string-bench/src/codec.rs +++ b/benchmarks/string-bench/src/codec.rs @@ -23,7 +23,7 @@ use vortex_bench::measurements::CustomUnitMeasurement; use vortex_fsst::fsst_compress; use vortex_fsst::fsst_train_compressor; use vortex_onpair::Config; -use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::DEFAULT_CONFIG; use vortex_onpair::MaxDictBits; use vortex_onpair::onpair_compress; @@ -59,7 +59,7 @@ impl DirectCandidate { })?; Ok(Self::OnPair(Config { max_dict_bits, - ..DEFAULT_DICT12_CONFIG + ..DEFAULT_CONFIG })) } diff --git a/benchmarks/string-bench/src/serialized.rs b/benchmarks/string-bench/src/serialized.rs index 7f38ef33f85..29d2115ed49 100644 --- a/benchmarks/string-bench/src/serialized.rs +++ b/benchmarks/string-bench/src/serialized.rs @@ -45,7 +45,7 @@ use vortex_btrblocks::schemes::string::FSSTScheme; use vortex_btrblocks::schemes::string::NullDominatedSparseScheme; use vortex_btrblocks::schemes::string::OnPairScheme; use vortex_btrblocks::schemes::string::StringDictScheme; -use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::DEFAULT_CONFIG; use crate::StringColumn; use crate::StringEncoder; @@ -87,7 +87,7 @@ fn serialized_encoder_label(encoder: StringEncoder) -> String { match encoder { // The config `OnPairScheme` compresses with. When btrblocks gains a // configurable budget, pass the benchmark's own config here. - StringEncoder::OnPair => onpair_label(&DEFAULT_DICT12_CONFIG), + StringEncoder::OnPair => onpair_label(&DEFAULT_CONFIG), StringEncoder::Fsst => encoder.label().to_string(), } } diff --git a/encodings/onpair/benches/decode.rs b/encodings/onpair/benches/decode.rs index 8a0f54bc2d8..57debb32090 100644 --- a/encodings/onpair/benches/decode.rs +++ b/encodings/onpair/benches/decode.rs @@ -45,7 +45,7 @@ use vortex_array::dtype::NativePType; use vortex_array::dtype::Nullability; use vortex_buffer::Buffer; use vortex_mask::Mask; -use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::DEFAULT_CONFIG; use vortex_onpair::OnPair; use vortex_onpair::OnPairArray; use vortex_onpair::OnPairArraySlotsExt; @@ -162,7 +162,7 @@ fn compress(n: usize, shape: Shape, ctx: &mut ExecutionCtx) -> OnPairArray { strings.iter().map(|s| Some(s.as_bytes())), DType::Utf8(Nullability::NonNullable), ); - onpair_compress(varbin.as_array(), DEFAULT_DICT12_CONFIG, ctx) + onpair_compress(varbin.as_array(), DEFAULT_CONFIG, ctx) .unwrap_or_else(|e| panic!("onpair_compress failed: {e}")) .try_downcast::() .unwrap_or_else(|array| panic!("expected OnPair array, got {}", array.encoding_id())) @@ -181,8 +181,11 @@ fn materialise(arr: &OnPairArray, ctx: &mut ExecutionCtx) -> (DecodeInputs, usiz let view = arr.as_view(); let dict_offsets = widen::(view.dict_offsets(), ctx); let dict_bytes = view.dict_bytes_handle().clone(); - CompactDictionaryView::validate(dict_bytes.as_host().as_slice(), dict_offsets.as_slice()) - .expect("valid OnPair dictionary"); + CompactDictionaryView::validate_safety( + dict_bytes.as_host().as_slice(), + dict_offsets.as_slice(), + ) + .expect("valid OnPair dictionary"); let inputs = DecodeInputs { dict_bytes, dict_offsets, diff --git a/encodings/onpair/src/array.rs b/encodings/onpair/src/array.rs index b67e09dfbd0..168155cc471 100644 --- a/encodings/onpair/src/array.rs +++ b/encodings/onpair/src/array.rs @@ -9,7 +9,10 @@ use std::sync::Arc; use std::sync::OnceLock; use num_traits::AsPrimitive; +use onpair::CompactDictionary; use onpair::CompactDictionaryView; +use onpair::Dictionary; +use onpair::DictionaryStorage; use prost::Message as _; use vortex_array::Array; use vortex_array::ArrayEq; @@ -103,8 +106,10 @@ impl OnPairMetadata { #[array_slots(OnPair)] pub struct OnPairSlots { - /// Primitive integer dictionary offsets, length `dict_size + 1`. The - /// cascading compressor may re-encode this child independently. + /// Dictionary-offset child, with length `dict_size + 1`. The cascading + /// compressor may re-encode this child independently; the materialised + /// offsets used by the runtime dictionary cache are derived from it on + /// first use. #[slot(0)] pub dict_offsets: ArrayRef, /// Primitive integer token codes. Downstream integer compression may @@ -124,12 +129,40 @@ pub struct OnPairSlots { pub validity: Option, } -/// Inner data for an OnPair-encoded array. +/// Immutable storage for a materialised OnPair dictionary. +/// +/// The two buffers remain owned by Vortex. Implementing [`DictionaryStorage`] +/// this way lets `onpair::CompactDictionary` retain them without copying +/// either the dictionary bytes or the widened offsets. +#[derive(Clone, Debug)] +struct OnPairDictionaryStorage { + bytes: ByteBuffer, + offsets: Buffer, +} + +impl DictionaryStorage for OnPairDictionaryStorage { + #[inline] + fn bytes(&self) -> &[u8] { + self.bytes.as_slice() + } + + #[inline] + fn offsets(&self) -> &[u32] { + self.offsets.as_slice() + } +} + +/// Non-child data for an OnPair-encoded array. /// -/// Holds only the dictionary blob (buffer 0). Every other piece — -/// `dict_offsets`, the per-token `codes`, the per-row `codes_offsets`, the -/// per-row `uncompressed_lengths`, and the optional validity child — is a -/// Vortex slot child so it can be re-encoded by the cascading compressor. +/// The serialized dictionary bytes stay in buffer 0, while the dictionary +/// offsets remain a recursive child so the existing on-disk format is +/// unchanged. Once the offsets are materialised, `dictionary` retains an +/// upstream storage-backed [`CompactDictionary`] around the same immutable +/// buffers. Constructing an array from serialized parts does not materialise or +/// validate the dictionary: Vortex only reconstructs the recursive offset +/// child at that point. Converting that child to contiguous `u32` offsets and +/// validating the dictionary are deferred until the first decode/search +/// operation that needs dictionary token access. #[derive(Clone)] pub struct OnPairData { /// The dictionary blob (buffer 0). @@ -137,63 +170,44 @@ pub struct OnPairData { /// INVARIANT: this buffer is an OnPair compact dictionary byte buffer, /// including its trailing read padding. dict_bytes: BufferHandle, - len: usize, - /// The `dict_offsets` child widened to `u32`, memoized on first use so the - /// child is decompressed and the dictionary content is validated at most - /// once per dictionary — never per operation. (`dict_bytes` needs no such - /// cache: it is a raw buffer, so access is already zero-cost.) + /// The storage-backed dictionary, memoized after successful initialization. + /// Initialization decompresses the child and safety-validates the dictionary; + /// once cached, later operations do neither again. The dictionary owns + /// clones of the immutable Vortex buffer handles, so this does not copy the + /// bytes. A failed validation is not cached, and concurrent first users may + /// duplicate initialization work before one value wins the `OnceLock`. /// /// INVARIANT: once populated, the offsets passed - /// [`CompactDictionaryView::validate`] against `dict_bytes` (or came from - /// the trainer via [`init_dict_offsets`](Self::init_dict_offsets)), and - /// they are the widened values of the array's `dict_offsets` child. The - /// `Arc` cell is shared only between arrays with identical `dict_bytes` - /// and logically identical `dict_offsets` (slice / filter / cast keep - /// both). - dict_offsets: Arc>>, + /// [`CompactDictionary::validate_safety`] against `dict_bytes`. + /// The `Arc` cell is shared only between arrays with identical dictionary + /// bytes and logically identical offsets (slice / filter / cast keep both). + dictionary: Arc>>, } impl OnPairData { - /// Build [`OnPairData`] from the dictionary blob and the number of rows. - pub fn new(dict_bytes: BufferHandle, len: usize) -> Self { + /// Build [`OnPairData`] from the dictionary blob. + pub fn new(dict_bytes: BufferHandle) -> Self { Self { dict_bytes, - len, - dict_offsets: Arc::new(OnceLock::new()), + dictionary: Arc::new(OnceLock::new()), } } - /// Seed the widened-offsets cell with dictionary offsets that are already - /// known to be conformant, so the first operation skips validation. + /// Build [`OnPairData`] with the dictionary already materialised, so the + /// first decode skips both widening and validation. /// - /// This is the crate-internal trust mint (the moral equivalent of - /// [`onpair::CompactDictionary::new_unchecked`]): compression seeds it - /// with the trainer's offsets, which are conformant by construction. - /// - /// # Safety - /// `(self.dict_bytes, offsets)` must satisfy every [`onpair`] compact - /// dictionary invariant (i.e. [`CompactDictionaryView::validate`] would - /// succeed on them), and `offsets` must be the widened values of the - /// array's `dict_offsets` child. [`dict_view`] relies on this to build - /// unchecked dictionary views. - pub(crate) unsafe fn init_dict_offsets(&self, offsets: Buffer) { - debug_assert!( - CompactDictionaryView::validate(self.dict_bytes().as_slice(), offsets.as_slice()) - .is_ok(), - "init_dict_offsets called with a non-conformant dictionary" - ); - // A benign race can only ever install another conformant value. - drop(self.dict_offsets.set(offsets)); - } - - /// Number of rows in the array. - pub fn len(&self) -> usize { - self.len - } - - /// Whether the array has zero rows. - pub fn is_empty(&self) -> bool { - self.len == 0 + /// `offsets` must be the widened values of the `dict_offsets` child the + /// caller attaches to the array; validation proves only that they are + /// structurally safe against `dict_bytes`. + pub(crate) fn try_new_with_dictionary( + dict_bytes: BufferHandle, + offsets: Buffer, + ) -> VortexResult { + let dictionary = build_dictionary(dict_bytes.as_host().clone(), offsets)?; + Ok(Self { + dict_bytes, + dictionary: Arc::new(OnceLock::from(dictionary)), + }) } /// The dictionary blob as a host byte buffer. @@ -207,50 +221,48 @@ impl OnPairData { } } -/// A conformant [`CompactDictionaryView`] over `array`'s dictionary. +/// Safety-validate `(bytes, offsets)` and seal them into a storage-backed +/// dictionary. +fn build_dictionary( + bytes: ByteBuffer, + offsets: Buffer, +) -> VortexResult> { + CompactDictionary::validate_safety(OnPairDictionaryStorage { bytes, offsets }) + .map_err(|e| vortex_err!(InvalidArgument: "Unsafe OnPair dictionary: {e}")) +} + +/// A safety-validated [`CompactDictionaryView`] over `array`'s dictionary. /// -/// The first call per dictionary widens the `dict_offsets` child to `u32` and -/// validates the dictionary content; both results are memoized in -/// [`OnPairData`], so subsequent calls — including on arrays derived by -/// slice / filter / cast, which share the cell — pay neither cost again. +/// The first successful initialization widens the `dict_offsets` child and +/// safety-validates the dictionary structure; the resulting storage-backed +/// dictionary is memoized in [`OnPairData`]. Once cached, subsequent calls — +/// including on arrays derived by slice / filter / cast, which share the cell — +/// pay neither cost again. pub(crate) fn dict_view<'a>( array: ArrayView<'a, OnPair>, ctx: &mut ExecutionCtx, ) -> VortexResult> { let data = array.data(); - let offsets = match data.dict_offsets.get() { - Some(offsets) => offsets, + let dictionary = match data.dictionary.get() { + Some(dictionary) => dictionary, None => { let widened = collect_widened::(array.dict_offsets(), ctx)?; - CompactDictionaryView::validate(data.dict_bytes().as_slice(), widened.as_slice()) - .map_err(|e| vortex_err!(InvalidArgument: "Invalid OnPair dictionary: {e}"))?; - data.dict_offsets.get_or_init(|| widened) + let dictionary = build_dictionary(data.dict_bytes().clone(), widened)?; + data.dictionary.get_or_init(|| dictionary) } }; - // SAFETY: the cell only ever holds offsets that satisfy the compact - // dictionary invariants against this `dict_bytes` (validated above, or - // guaranteed by `init_dict_offsets`'s contract), and both buffers are - // immutable. - Ok(unsafe { - CompactDictionaryView::new_unchecked(data.dict_bytes().as_slice(), offsets.as_slice()) - }) + Ok(dictionary.as_view()) } impl Display for OnPairData { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!( - f, - "len: {}, dict_bytes_len: {}", - self.len, - self.dict_bytes.len() - ) + write!(f, "dict_bytes_len: {}", self.dict_bytes.len()) } } impl Debug for OnPairData { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.debug_struct("OnPairData") - .field("len", &self.len) .field("dict_bytes_len", &self.dict_bytes.len()) .finish() } @@ -284,6 +296,37 @@ impl OnPair { codes_offsets: ArrayRef, uncompressed_lengths: ArrayRef, validity: Validity, + ) -> VortexResult { + Self::try_new_with_data( + dtype, + OnPairData::new(dict_bytes), + dict_offsets, + codes, + codes_offsets, + uncompressed_lengths, + validity, + ) + } + + /// Build an [`OnPairArray`] from already-materialised parts while reusing + /// an existing [`OnPairData`]. + /// + /// Reusing the data preserves the dictionary byte-buffer handle and the + /// shared lazy dictionary cache. This is useful when recursive compression + /// replaces the slot children without changing the logical dictionary. + /// + /// If `data` contains a memoized dictionary, `dict_offsets` must be + /// logically equivalent to the offsets used to create that dictionary. + /// The constructor intentionally does not validate the dictionary itself; + /// dictionary validation remains lazy and is performed on first use. + pub fn try_new_with_data( + dtype: DType, + data: OnPairData, + dict_offsets: ArrayRef, + codes: ArrayRef, + codes_offsets: ArrayRef, + uncompressed_lengths: ArrayRef, + validity: Validity, ) -> VortexResult { validate_parts( &dtype, @@ -292,32 +335,30 @@ impl OnPair { &codes_offsets, &uncompressed_lengths, )?; - let len = uncompressed_lengths.len(); - let data = OnPairData::new(dict_bytes, len); - let slots = OnPairSlots { - dict_offsets, - codes, - codes_offsets, - uncompressed_lengths, - validity: validity_to_child(&validity, len), - } - .into_slots(); Ok(unsafe { - Array::from_parts_unchecked(ArrayParts::new(OnPair, dtype, len, data).with_slots(slots)) + Self::new_unchecked( + dtype, + data, + dict_offsets, + codes, + codes_offsets, + uncompressed_lengths, + validity, + ) }) } /// Build an [`OnPairArray`] without validation, carrying `data` — and with - /// it the memoized widened `dict_offsets` — from an existing array. + /// it the memoized dictionary — from an existing array. /// /// # Safety /// The parts must satisfy the same invariants [`try_new`](Self::try_new) - /// checks. If `data`'s widened-offsets cell is populated (or shared with a - /// live array), `dict_offsets` must hold the same logical offsets the cell + /// checks. If `data`'s dictionary cell is populated (or shared with a live + /// array), `dict_offsets` must hold the same logical offsets the dictionary /// was built from. pub(crate) unsafe fn new_unchecked( dtype: DType, - mut data: OnPairData, + data: OnPairData, dict_offsets: ArrayRef, codes: ArrayRef, codes_offsets: ArrayRef, @@ -325,7 +366,6 @@ impl OnPair { validity: Validity, ) -> OnPairArray { let len = uncompressed_lengths.len(); - data.len = len; let slots = OnPairSlots { dict_offsets, codes, @@ -386,7 +426,7 @@ impl VTable for OnPair { fn validate( &self, - data: &Self::TypedArrayData, + _data: &Self::TypedArrayData, dtype: &DType, len: usize, slots: &[Option], @@ -402,9 +442,6 @@ impl VTable for OnPair { if s.uncompressed_lengths.len() != len { vortex_bail!(InvalidArgument: "uncompressed_lengths must have same len as outer array"); } - if data.len != len { - vortex_bail!(InvalidArgument: "OnPairData len {} != outer len {}", data.len, len); - } Ok(()) } @@ -438,10 +475,10 @@ impl VTable for OnPair { ); let mut data = array.data().clone(); data.dict_bytes = buffers[0].clone(); - // The replacement blob may differ from the one the memoized offsets - // were validated against, so drop the (shared) cell rather than + // The replacement blob may differ from the one the memoized dictionary + // was validated against, so drop the (shared) cell rather than // carry a claim we can no longer prove. - data.dict_offsets = Arc::new(OnceLock::new()); + data.dictionary = Arc::new(OnceLock::new()); Ok( ArrayParts::new(self.clone(), array.dtype().clone(), array.len(), data) .with_slots(array.slots().iter().cloned().collect()), @@ -528,7 +565,7 @@ impl VTable for OnPair { other => vortex_bail!(InvalidArgument: "Expected 4 or 5 children, got {other}"), }; - let data = OnPairData::new(buffers[0].clone(), len); + let data = OnPairData::new(buffers[0].clone()); let slots = OnPairSlots { dict_offsets, codes, diff --git a/encodings/onpair/src/canonical.rs b/encodings/onpair/src/canonical.rs index 5179def3bf1..a799ea52519 100644 --- a/encodings/onpair/src/canonical.rs +++ b/encodings/onpair/src/canonical.rs @@ -24,7 +24,7 @@ use vortex_buffer::ByteBuffer; use vortex_buffer::ByteBufferMut; use vortex_error::VortexResult; use vortex_error::vortex_ensure; -use vortex_error::vortex_panic; +use vortex_error::vortex_err; use crate::OnPair; use crate::OnPairArraySlotsExt; @@ -91,18 +91,14 @@ pub(crate) fn onpair_decode_bytes( let codes = collect_widened::(&array.codes().slice(code_start..code_end)?, ctx)?; let dict = dict_view(array, ctx)?; let mut out_bytes = ByteBufferMut::with_capacity(total_size); - let written = - match onpair::try_decode_into(codes.as_slice(), dict, out_bytes.spare_capacity_mut()) { - Ok(written) => written, - Err(_) => { - vortex_panic!("OnPair codes decode to more bytes than uncompressed_lengths records") - } - }; - if written != total_size { - vortex_panic!( - "OnPair codes decoded to {written} bytes but uncompressed_lengths records {total_size}" - ); - } + let written = onpair::try_decode_into(codes.as_slice(), dict, out_bytes.spare_capacity_mut()) + .map_err(|_| { + vortex_err!("OnPair codes decode to more bytes than uncompressed_lengths records") + })?; + vortex_ensure!( + written == total_size, + "OnPair codes decoded to {written} bytes but uncompressed_lengths records {total_size}" + ); // SAFETY: `try_decode_into` initialised exactly `written` bytes. unsafe { out_bytes.set_len(written) }; Ok((out_bytes, lengths)) diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index a2e63f2c8a2..e2cf69353ec 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -23,12 +23,7 @@ use vortex_error::vortex_err; use vortex_mask::AllOr; use crate::OnPair; - -/// Default OnPair training configuration: 12-bit codes ("dict-12"). -pub const DEFAULT_DICT12_CONFIG: Config = Config { - seed: Some(42), - ..onpair::DEFAULT_CONFIG -}; +use crate::OnPairData; /// Compress any [`ArrayRef`] whose canonical form is a string array. /// @@ -102,18 +97,19 @@ pub fn onpair_compress( let uncompressed_lengths = uncompressed_lengths.into_array(); - let encoded = OnPair::try_new( - array.dtype().clone(), + let data = OnPairData::try_new_with_dictionary( dict_bytes_to_buffer(dict_bytes), - dict_offsets.clone().into_array(), + dict_offsets.clone(), + )?; + let encoded = OnPair::try_new_with_data( + array.dtype().clone(), + data, + dict_offsets.into_array(), codes, codes_offsets, uncompressed_lengths, validity, )?; - // SAFETY: the trainer's dictionary is conformant by construction, and - // `dict_offsets` is exactly the u32 child attached above. - unsafe { encoded.init_dict_offsets(dict_offsets) }; Ok(encoded.into_array()) } diff --git a/encodings/onpair/src/compute/cast.rs b/encodings/onpair/src/compute/cast.rs index c31893f82fd..a746d3dcc07 100644 --- a/encodings/onpair/src/compute/cast.rs +++ b/encodings/onpair/src/compute/cast.rs @@ -82,7 +82,7 @@ mod tests { use vortex_error::VortexResult; use vortex_session::VortexSession; - use crate::compress::DEFAULT_DICT12_CONFIG; + use crate::DEFAULT_CONFIG; use crate::compress::onpair_compress; static SESSION: LazyLock = LazyLock::new(|| { @@ -107,7 +107,7 @@ mod tests { fn test_cast_onpair_conformance(#[case] array: VarBinArray) -> VortexResult<()> { let array = array.into_array(); let mut ctx = SESSION.create_execution_ctx(); - let onpair = onpair_compress(&array, DEFAULT_DICT12_CONFIG, &mut ctx)?; + let onpair = onpair_compress(&array, DEFAULT_CONFIG, &mut ctx)?; test_cast_conformance(&onpair.into_array(), &mut ctx); Ok(()) } diff --git a/encodings/onpair/src/compute/compare.rs b/encodings/onpair/src/compute/compare.rs index bb10fcfd6aa..3ed449731d9 100644 --- a/encodings/onpair/src/compute/compare.rs +++ b/encodings/onpair/src/compute/compare.rs @@ -109,8 +109,8 @@ mod tests { use vortex_error::vortex_err; use vortex_session::VortexSession; + use crate::DEFAULT_CONFIG; use crate::OnPair; - use crate::compress::DEFAULT_DICT12_CONFIG; use crate::compress::onpair_compress; static SESSION: LazyLock = LazyLock::new(|| { @@ -133,7 +133,7 @@ mod tests { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(input.as_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(input.as_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); let result = arr .binary(ConstantArray::new("", input.len()).into_array(), op)? @@ -150,7 +150,7 @@ mod tests { DType::Utf8(Nullability::Nullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(input.as_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(input.as_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); let eq_empty = arr .clone() @@ -183,7 +183,7 @@ mod tests { DType::Utf8(Nullability::Nullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(input.as_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(input.as_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); let rhs = ConstantArray::new("hello", arr.len()).into_array(); let eq = arr @@ -218,7 +218,7 @@ mod tests { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(input.as_array(), DEFAULT_DICT12_CONFIG, &mut ctx)? + let arr = onpair_compress(input.as_array(), DEFAULT_CONFIG, &mut ctx)? .try_downcast::() .map_err(|array| vortex_err!("expected OnPair array, got {}", array.encoding_id()))?; let rhs = ConstantArray::new("hello", arr.len()).into_array(); @@ -258,7 +258,7 @@ mod tests { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(input.as_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(input.as_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); let sliced = arr.slice(1..4)?; assert!(sliced.is::(), "slice dropped OnPair encoding"); let sliced = sliced diff --git a/encodings/onpair/src/lib.rs b/encodings/onpair/src/lib.rs index 4c2ae4529cc..8b91154fe39 100644 --- a/encodings/onpair/src/lib.rs +++ b/encodings/onpair/src/lib.rs @@ -23,6 +23,7 @@ mod tests; pub use array::*; pub use compress::*; pub use onpair::Config; +pub use onpair::DEFAULT_CONFIG; pub use onpair::Error as OnPairError; pub use onpair::MaxDictBits; pub use onpair::Threshold; diff --git a/encodings/onpair/src/tests.rs b/encodings/onpair/src/tests.rs index 61b5591f54d..9f20c4ce08b 100644 --- a/encodings/onpair/src/tests.rs +++ b/encodings/onpair/src/tests.rs @@ -28,11 +28,11 @@ use vortex_array::validity::Validity; use vortex_buffer::BufferMut; use vortex_session::VortexSession; +use crate::DEFAULT_CONFIG; use crate::OnPair; use crate::OnPairArrayExt; use crate::OnPairArraySlotsExt; use crate::OnPairMetadata; -use crate::compress::DEFAULT_DICT12_CONFIG; use crate::compress::onpair_compress; static SESSION: LazyLock = LazyLock::new(vortex_array::array_session); @@ -41,7 +41,7 @@ fn compress_onpair( array: &vortex_array::ArrayRef, ctx: &mut vortex_array::ExecutionCtx, ) -> vortex_error::VortexResult { - onpair_compress(array, DEFAULT_DICT12_CONFIG, ctx)? + onpair_compress(array, DEFAULT_CONFIG, ctx)? .try_downcast::() .map_err(|array| { vortex_error::vortex_err!("expected OnPair array, got {}", array.encoding_id()) @@ -95,13 +95,13 @@ fn test_onpair_rejects_100k_token_dictionary() -> vortex_error::VortexResult<()> } dict_bytes.resize(dict_bytes.len() + onpair::MAX_TOKEN_SIZE, 0); - assert!(CompactDictionaryView::validate(&dict_bytes, &dict_offsets).is_err()); + assert!(CompactDictionaryView::validate_safety(&dict_bytes, &dict_offsets).is_err()); Ok(()) } -/// Dictionary content is validated lazily — construction stays lightweight, -/// and a corrupt dictionary is rejected by the first operation that decodes -/// or searches through it, including on derived (sliced) arrays. +/// Dictionary safety is validated lazily — construction stays lightweight, and +/// a structurally corrupt dictionary is rejected by the first operation that +/// decodes or searches through it, including on derived (sliced) arrays. #[cfg_attr(miri, ignore)] #[test] fn test_corrupt_dictionary_rejected_on_first_use() -> vortex_error::VortexResult<()> { @@ -159,7 +159,7 @@ fn test_onpair_roundtrip() -> vortex_error::VortexResult<()> { let input = sample_input(); let mut ctx = SESSION.create_execution_ctx(); - let compressed = onpair_compress(&input.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let compressed = onpair_compress(&input.into_array(), DEFAULT_CONFIG, &mut ctx)?; assert!(compressed.clone().into_array().is::()); let decoded = compressed @@ -247,7 +247,7 @@ fn test_onpair_nullable_canonicalize() -> vortex_error::VortexResult<()> { DType::Utf8(Nullability::Nullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&input.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&input.into_array(), DEFAULT_CONFIG, &mut ctx)?; let canonical = arr.into_array().execute::(&mut ctx)?; let mask = canonical .validity()? @@ -266,7 +266,7 @@ fn test_onpair_nullable_canonicalize() -> vortex_error::VortexResult<()> { fn test_onpair_scalar_at() -> vortex_error::VortexResult<()> { let input = sample_input(); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&input.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&input.into_array(), DEFAULT_CONFIG, &mut ctx)?; let s = arr.into_array().execute_scalar(2, &mut ctx)?; let v = s.as_utf8().value().unwrap(); assert_eq!(v.as_bytes(), b"https://www.test.org/page"); @@ -291,7 +291,7 @@ fn test_onpair_scalar_at_window() -> vortex_error::VortexResult<()> { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&varbin.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(&varbin.into_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); for &i in &[0usize, 1, 999, 1000, n - 1] { let got = arr.execute_scalar(i, &mut ctx)?; @@ -340,7 +340,7 @@ fn test_onpair_unroll_tail_boundaries(#[case] n: usize) -> vortex_error::VortexR DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&input.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&input.into_array(), DEFAULT_CONFIG, &mut ctx)?; let canonical = arr.into_array().execute::(&mut ctx)?; let mask = canonical .validity()? @@ -364,7 +364,7 @@ fn test_onpair_empty() -> vortex_error::VortexResult<()> { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&input.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&input.into_array(), DEFAULT_CONFIG, &mut ctx)?; assert_eq!(arr.len(), 0); let canonical = arr.into_array().execute::(&mut ctx)?; assert_eq!(canonical.len(), 0); @@ -381,7 +381,7 @@ fn test_onpair_all_null() -> vortex_error::VortexResult<()> { ) .into_array(); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&input, DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&input, DEFAULT_CONFIG, &mut ctx)?; assert!(arr.is::()); assert_arrays_eq!(arr, input, &mut ctx); @@ -595,7 +595,7 @@ fn test_onpair_slice_canonicalize() -> vortex_error::VortexResult<()> { DType::Utf8(Nullability::NonNullable), ); let mut ctx = SESSION.create_execution_ctx(); - let arr = onpair_compress(&varbin.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?.into_array(); + let arr = onpair_compress(&varbin.into_array(), DEFAULT_CONFIG, &mut ctx)?.into_array(); // interior (start>0, end0, // end=n), and a near-full window. diff --git a/encodings/onpair/tests/big_data.rs b/encodings/onpair/tests/big_data.rs index 15f50bb3322..7feed9b1718 100644 --- a/encodings/onpair/tests/big_data.rs +++ b/encodings/onpair/tests/big_data.rs @@ -25,7 +25,7 @@ use vortex_array::builtins::ArrayBuiltins; use vortex_array::dtype::DType; use vortex_array::dtype::Nullability; use vortex_array::scalar_fn::fns::operators::Operator; -use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::DEFAULT_CONFIG; use vortex_onpair::onpair_compress; use vortex_session::VortexSession; @@ -74,7 +74,7 @@ fn smoke_100k_rows() -> vortex_error::VortexResult<()> { let mut ctx = SESSION.create_execution_ctx(); let t0 = Instant::now(); - let arr = onpair_compress(&varbin.into_array(), DEFAULT_DICT12_CONFIG, &mut ctx)?; + let arr = onpair_compress(&varbin.into_array(), DEFAULT_CONFIG, &mut ctx)?; let compress_elapsed = t0.elapsed(); eprintln!( "compressed {} rows ({} raw bytes) in {:?}", diff --git a/vortex-btrblocks/src/schemes/string/onpair.rs b/vortex-btrblocks/src/schemes/string/onpair.rs index a771593efc2..75ec97d2191 100644 --- a/vortex-btrblocks/src/schemes/string/onpair.rs +++ b/vortex-btrblocks/src/schemes/string/onpair.rs @@ -15,7 +15,7 @@ use vortex_compressor::scheme::CompressionEstimate; use vortex_compressor::scheme::DeferredEstimate; use vortex_compressor::scheme::SchemeId; use vortex_error::VortexResult; -use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::DEFAULT_CONFIG; use vortex_onpair::OnPair; use vortex_onpair::OnPairArrayExt; use vortex_onpair::OnPairArraySlotsExt; @@ -79,7 +79,7 @@ impl Scheme for OnPairScheme { exec_ctx: &mut ExecutionCtx, ) -> VortexResult { let utf8 = data.array_as_varbinview().into_owned(); - let encoded = onpair_compress(utf8.as_array(), DEFAULT_DICT12_CONFIG, exec_ctx)?; + let encoded = onpair_compress(utf8.as_array(), DEFAULT_CONFIG, exec_ctx)?; let Some(onpair_array) = encoded.as_opt::() else { return Ok(encoded); }; @@ -117,9 +117,9 @@ impl Scheme for OnPairScheme { exec_ctx, )?; - Ok(OnPair::try_new( + Ok(OnPair::try_new_with_data( onpair_array.dtype().clone(), - onpair_array.dict_bytes_handle().clone(), + onpair_array.data().clone(), dict_offsets, codes, codes_offsets, diff --git a/vortex-btrblocks/tests/onpair_roundtrip.rs b/vortex-btrblocks/tests/onpair_roundtrip.rs index 272fa930bd0..7668a836d86 100644 --- a/vortex-btrblocks/tests/onpair_roundtrip.rs +++ b/vortex-btrblocks/tests/onpair_roundtrip.rs @@ -188,7 +188,7 @@ fn empty_and_short_string_roundtrip() { } /// Regression for the Euro2016 compress-bench panic -/// (`onpair::decompress`: "dictionary offsets must be nondecreasing"). +/// (`onpair::decompress`: "dictionary offsets must be strictly increasing"). /// /// A large, high-cardinality corpus fills the OnPair dictionary toward its /// 4096-entry cap, so the cascading compressor narrows `dict_offsets` to `u16`