feat(vortex-onpair): update to the onpair 0.1 API (#8675)

## Summary

Updates the experimental `vortex-onpair` encoding to the public API
introduced in `onpair` 0.1.

- Migrates compression and decoding to the new configuration, raw
column-parts, and validated compact-dictionary APIs.
- Updates compressed-domain equality comparisons for the new dictionary
representation.
- Supports adaptive `u32`/`u64` code offsets.
- Validates dictionaries loaded from serialized data, including
rejecting dictionaries that exceed the `u16` token address space.
- Adds regression coverage for empty and all-null inputs, large
dictionaries, narrowed children, sliced arrays, and `u64` code offsets.

## Scope

This PR intentionally does not include the new `LIKE`, `list_contains`,
or `take` pushdowns. Those will be submitted in a follow-up PR.

A separate follow-up will enable OnPair on the default read path without
enabling it in the default compressor.

## Dependency note

The dependency is temporarily pinned to an upstream OnPair commit
containing the dictionary-validation fix. Before merging, that fix will
be published in a new crates.io release and the Git dependency will be
replaced with the released version.

## Testing

- `cargo +nightly fmt --all -- --check`
- `cargo test -p vortex-onpair`
- `cargo test -p vortex-btrblocks --features unstable_encodings`
- `cargo clippy -p vortex-onpair -p vortex-btrblocks --all-targets
--all-features -- -D warnings`

---------

Signed-off-by: Francesco Gargiulo <gargiulo.fr@gmail.com>
Signed-off-by: Francesco Gargiulo <francesco@spiraldb.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 = "0.1.1"
 opentelemetry = "0.32.0"
 opentelemetry-otlp = "0.32.0"
 opentelemetry_sdk = "0.32.0"

Cargo.toml

@@ -10,26 +10,25 @@ 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 configuration uses OnPair's default dictionary budget and
+a fixed seed. Vortex stores token codes as an integer child array; downstream
+integer compression may narrow or bit-pack that child independently.
 
 ## 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 2 — `codes_offsets`: `PrimitiveArray<u32>`, length `num_rows + 1`.
-- Slot 3 — `uncompressed_lengths`: integer `PrimitiveArray`, length
-  `num_rows`.
+  including the read padding required by the decoder.
+- Slot 0 — `dict_offsets`: integer child, len `dict_size + 1`.
+- Slot 1 — `codes`: integer child, length `total_tokens`.
+- Slot 2 — `codes_offsets`: integer child, length `num_rows + 1`.
+- Slot 3 — `uncompressed_lengths`: integer child, length `num_rows`.
 - Slot 4 — optional validity child.
 
 All four integer slot children flow through the standard cascading

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;
@@ -162,6 +164,8 @@ fn compress(n: usize, shape: Shape, ctx: &mut ExecutionCtx) -> OnPairArray {
     );
     onpair_compress(varbin.as_array(), DEFAULT_DICT12_CONFIG, ctx)
         .unwrap_or_else(|e| panic!("onpair_compress failed: {e}"))
+        .try_downcast::<OnPair>()
+        .unwrap_or_else(|array| panic!("expected OnPair array, got {}", array.encoding_id()))
 }
 
 /// Canonicalise a slot child to the decoder's native primitive width.
@@ -175,13 +179,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 +201,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