feat(vortex-onpair): port to onpair 0.1.0 API

Port the OnPair encoding to the 0.1.0 public API, including raw column parts, validated compact dictionaries, and MaxDictBits configuration.

Pin the latest upstream validation fix, add compressed-domain equality pushdown, and strengthen file-borne dictionary and code validation.

Signed-off-by: Francesco Gargiulo <gargiulo.fr@gmail.com>

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: gargiulofrancesco

Cargo.lock

@@ -192,7 +192,7 @@ num_enum = { version = "0.7.3", default-features = false }
 object_store = { version = "0.13.2", default-features = false }
 once_cell = "1.21"
 oneshot = { version = "0.2.0", features = ["async"] }
-onpair = { version = "0.0.4" }
+onpair = { git = "https://github.com/spiraldb/onpair.git", rev = "28df115c799674c663adedc8f81620ac82c7a890" }
 opentelemetry = "0.32.0"
 opentelemetry-otlp = "0.32.0"
 opentelemetry_sdk = "0.32.0"

Cargo.toml

@@ -10,23 +10,26 @@ cascading-compressor support on every integer child.
 
 ## Compute
 
-Like the FSST encoding, this crate provides `cast` and `filter`
-pushdown. Other operators fall back to ordinary decompression.
+Like the FSST encoding, this crate pushes down common operations over the
+encoded representation. It supports `cast`, `filter`, byte length, and
+constant equality / inequality. Unsupported operators fall back to ordinary
+decompression.
 
 ## Default Configuration
 
-The default training preset is **dict-12**: 12 bits per token,
-dictionary capped at 4 096 entries. Token codes are stored as a
-`PrimitiveArray<u16>`; downstream `FastLanes::BitPacking` losslessly
-narrows the child to exactly `bits`-bit codes on disk.
+The default training preset is **dict-12**: the OnPair trainer may build a
+dictionary with up to 4 096 tokens. After compression, the runtime code width is
+derived from the actual dictionary size. Vortex stores token codes as an integer
+child array; downstream integer compression may narrow or bit-pack that child
+independently of the OnPair metadata.
 
 ## Layout
 
 - Buffer 0 — `dict_bytes`: dictionary blob built by the OnPair trainer,
   padded with `MAX_TOKEN_SIZE` trailing zero bytes so the over-copy
   decoder can read 16 bytes past the last token.
 - Slot 0 — `dict_offsets`: `PrimitiveArray<u32>`, len `dict_size + 1`.
-- Slot 1 — `codes`: `PrimitiveArray<u16>`, length `total_tokens`.
+- Slot 1 — `codes`: integer `PrimitiveArray`, length `total_tokens`.
 - Slot 2 — `codes_offsets`: `PrimitiveArray<u32>`, length `num_rows + 1`.
 - Slot 3 — `uncompressed_lengths`: integer `PrimitiveArray`, length
   `num_rows`.

encodings/experimental/onpair/README.md

@@ -3,7 +3,7 @@
 //
 //! Decode-path microbenchmarks for the OnPair Vortex array.
 //!
-//! * `decompress_into` — the upstream `onpair::decompress_into` decoder hot
+//! * `decode_into` — the upstream `onpair::decode_into` decoder hot
 //!   loop, fed by pre-materialised [`DecodeInputs`]. Measures the inner loop
 //!   only (no child `execute`, no allocation).
 //! * `canonicalize_to_varbinview` — the full Vortex
@@ -28,7 +28,7 @@ use std::mem::MaybeUninit;
 use std::sync::LazyLock;
 
 use divan::Bencher;
-use onpair::Parts;
+use onpair::CompactDictionaryView;
 use vortex_array::ArrayRef;
 use vortex_array::ExecutionCtx;
 use vortex_array::IntoArray;
@@ -38,43 +38,45 @@ use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::VarBinArray;
 use vortex_array::arrays::VarBinViewArray;
 use vortex_array::arrays::filter::FilterKernel;
+use vortex_array::buffer::BufferHandle;
 use vortex_array::builtins::ArrayBuiltins;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::NativePType;
 use vortex_array::dtype::Nullability;
 use vortex_buffer::Buffer;
-use vortex_buffer::ByteBuffer;
 use vortex_mask::Mask;
 use vortex_onpair::DEFAULT_DICT12_CONFIG;
 use vortex_onpair::OnPair;
 use vortex_onpair::OnPairArray;
 use vortex_onpair::OnPairArraySlotsExt;
 
 /// Host-resident decode inputs, materialised once so the decode-loop benchmark
-/// measures only `onpair::decompress_into` (not child `execute`/allocation).
+/// measures only `onpair::decode_into` (not child `execute`/allocation).
 struct DecodeInputs {
-    dict_bytes: ByteBuffer,
+    dict_bytes: BufferHandle,
     dict_offsets: Buffer<u32>,
     codes: Buffer<u16>,
-    bits: u32,
 }
 
 impl DecodeInputs {
-    fn as_parts(&self) -> Parts<'_> {
-        Parts {
-            dict_bytes: self.dict_bytes.as_slice(),
-            dict_offsets: self.dict_offsets.as_slice(),
-            bits: self.bits,
-            codes: self.codes.as_slice(),
+    fn dict(&self) -> CompactDictionaryView<'_> {
+        // SAFETY: `materialise` validates this borrowed dictionary once after
+        // widening offsets. The benchmark then keeps both buffers immutable.
+        unsafe {
+            CompactDictionaryView::new_unchecked(
+                self.dict_bytes.as_host().as_slice(),
+                self.dict_offsets.as_slice(),
+            )
         }
     }
 
-    fn decompressed_len(&self) -> usize {
-        onpair::decompressed_len(self.as_parts())
+    fn decoded_len(&self) -> usize {
+        onpair::decoded_len(self.codes.as_slice(), self.dict())
     }
 
-    fn decompress_into(&self, out: &mut [MaybeUninit<u8>]) -> usize {
-        onpair::decompress_into(self.as_parts(), out)
+    fn decode_into(&self, out: &mut [MaybeUninit<u8>]) -> usize {
+        // SAFETY: callers allocate `decoded_len + DECODE_PADDING` bytes.
+        unsafe { onpair::decode_into(self.codes.as_slice(), self.dict(), out) }
     }
 }
 use vortex_onpair::onpair_compress;
@@ -175,13 +177,16 @@ fn widen<T: NativePType>(arr: &ArrayRef, ctx: &mut ExecutionCtx) -> Buffer<T> {
 
 fn materialise(arr: &OnPairArray, ctx: &mut ExecutionCtx) -> (DecodeInputs, usize) {
     let view = arr.as_view();
+    let dict_offsets = widen::<u32>(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");
     let inputs = DecodeInputs {
-        dict_bytes: view.dict_bytes().clone(),
-        dict_offsets: widen::<u32>(view.dict_offsets(), ctx),
+        dict_bytes,
+        dict_offsets,
         codes: widen::<u16>(view.codes(), ctx),
-        bits: view.bits(),
     };
-    let total = inputs.decompressed_len();
+    let total = inputs.decoded_len();
     (inputs, total)
 }
 
@@ -194,16 +199,16 @@ const CASES: &[(Shape, usize)] = &[
 ];
 
 /// Raw decode loop time, excluding child `execute` and the output allocation.
-/// Hits `onpair::decompress_into` directly.
+/// Hits `onpair::decode_into` directly.
 #[divan::bench(args = CASES)]
-fn decompress_into_bench(bencher: Bencher, case: (Shape, usize)) {
+fn decode_into_bench(bencher: Bencher, case: (Shape, usize)) {
     let mut ctx = SESSION.create_execution_ctx();
     let (shape, n) = case;
     let arr = compress(n, shape, &mut ctx);
     let (inputs, total) = materialise(&arr, &mut ctx);
     bencher.bench_local(|| {
-        let mut out: Vec<u8> = Vec::with_capacity(total);
-        let written = inputs.decompress_into(out.spare_capacity_mut());
+        let mut out: Vec<u8> = Vec::with_capacity(total + onpair::DECODE_PADDING);
+        let written = inputs.decode_into(out.spare_capacity_mut());
         unsafe { out.set_len(written) };
         divan::black_box(out);
     });

encodings/experimental/onpair/benches/decode.rs

@@ -1 +1 @@
-�  ��(0
8
+�  ��(0
8