Port OnPair to onpair 0.1.0; add compressed-domain LIKE, list_contains, and take kernels

Cargo.toml

@@ -199,7 +199,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", rev = "a31dfd507f5e4f07f104a9fd11346dd2d6a20f9b", version = "0.1.0" }
 opentelemetry = "0.32.0"
 opentelemetry-otlp = "0.32.0"
 opentelemetry_sdk = "0.32.0"

encodings/experimental/onpair/README.md

@@ -10,23 +10,27 @@ 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`, `take`, byte length,
+constant equality / inequality, simple constant `LIKE` patterns, and constant
+string-list membership (`IN (...)`). Unsupported operators or pattern shapes
+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/benches/decode.rs

@@ -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/goldenfiles/onpair.metadata

@@ -1 +1 @@
-€  €è(0
8
+€  €è(0
8