Refactor OnPair to FSST-shape: dict-as-blob, u16 codes child, Rust decode

Replaces the previous opaque-blob layout with one that mirrors how FSST
splits its symbols-as-buffer / codes-as-child encoding, and shifts every
read path off the C++ FFI.

Layout
------
  Buffer 0  dict_bytes              — dictionary blob built by C++ training
  Slot 0    dict_offsets   u32[]    — len = dict_size + 1
  Slot 1    codes          u16[]    — one token id per element, low `bits`
                                       bits populated (FastLanes-bit-packable)
  Slot 2    codes_offsets  u32[]    — per-row token offsets, len = n + 1
  Slot 3    uncompressed_lengths    — i32[], len = n
  Slot 4    validity                — optional Bool child

  metadata = { bits: u32, uncompressed_lengths_ptype: i32 }

Decode path
-----------
At compress time we call OnPair's C++ trainer to produce the dictionary
and bit-packed token stream, then immediately unpack the stream into u16
codes in Rust (`vortex_onpair_sys::unpack_codes_to_u16`) and drop the
C++ column. After that, nothing on the read path touches C++:

  decode_row(r):
      for c in codes[codes_offsets[r] .. codes_offsets[r+1]]:
          out.extend_from_slice(
              dict_bytes[dict_offsets[c] .. dict_offsets[c+1]]
          )

`canonicalize`, `scalar_at`, and the compute kernels all share a
`DecodeView` over the materialised children.

Compute kernels (pure Rust, no C++ scan)
----------------------------------------
* compare (Eq / NotEq): streams dict slices per row, short-circuits on
  the first mismatch.
* like ('lit', 'pre%', '%sub%'): same streaming approach for prefix; a
  full row decode + memmem for contains.
* filter: canonical round-trip + recompress (unchanged).
* slice: zero-copy — narrows codes_offsets / uncompressed_lengths /
  validity and shares the dict blob + codes child.
* cast: identity rewrap, no payload touched.

Tests
-----
All 7 unit tests + the 100 000-row big_data smoke test pass. On the
smoke corpus (release): compress 147 ms, full canonicalize 7.5 ms,
equals / starts_with / contains pushdown counts match a brute-force
reference exactly.

Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

encodings/onpair-sys/cxx/onpair_shim.cpp

@@ -14,9 +14,11 @@
 #include <string_view>
 #include <vector>
 
+using onpair::DECOMPRESS_BUFFER_PADDING;
+using onpair::DictionaryView;
 using onpair::OnPairColumn;
 using onpair::OnPairColumnView;
-using onpair::DECOMPRESS_BUFFER_PADDING;
+using onpair::StoreView;
 using onpair::encoding::DynamicThreshold;
 using onpair::encoding::TrainingConfig;
 
@@ -351,4 +353,43 @@ size_t onpair_column_dict_bytes(const OnPairColumnHandle* handle) {
     }
 }
 
+OnPairStatus onpair_column_parts(
+    const OnPairColumnHandle* handle,
+    OnPairColumnParts*        out_parts) {
+    if (handle == nullptr || out_parts == nullptr) {
+        return ONPAIR_ERR_INVALID_ARG;
+    }
+    auto* h = const_cast<ColumnHandle*>(reinterpret_cast<const ColumnHandle*>(handle));
+    try {
+        const auto& view = h->get_view();
+        const DictionaryView& dv = view.dictionary();
+        const StoreView&      sv = view.store();
+
+        const size_t   dict_size  = dv.num_tokens();
+        const uint32_t* dict_off  = dv.raw_offsets();
+        const size_t   dict_bytes = dict_size == 0 ? 0 : dict_off[dict_size];
+
+        const size_t   num_rows   = sv.num_strings();
+        const uint32_t bw         = static_cast<uint32_t>(sv.bits());
+        const size_t   tokens     = sv.num_tokens();
+        // The packed stream is laid out by BitWriter as a vector<uint64_t>;
+        // round-up-to-u64 of (tokens * bits) bits.
+        const size_t   packed_u64 = (tokens * bw + 63) / 64;
+
+        out_parts->dict_bytes           = dv.raw_bytes();
+        out_parts->dict_bytes_len       = dict_bytes;
+        out_parts->dict_offsets         = dict_off;
+        out_parts->dict_offsets_len     = dict_size + 1;
+        out_parts->codes_packed         = sv.packed_data();
+        out_parts->codes_packed_u64_len = packed_u64;
+        out_parts->codes_boundaries     = sv.boundaries();
+        out_parts->codes_boundaries_len = num_rows + 1;
+        out_parts->bits                 = bw;
+        out_parts->num_rows             = num_rows;
+        return ONPAIR_OK;
+    } catch (...) {
+        return ONPAIR_ERR_INTERNAL;
+    }
+}
+
 } // extern "C"

encodings/onpair-sys/cxx/onpair_shim.h

@@ -124,6 +124,29 @@ OnPairStatus onpair_column_dict_copy(
 // Bytes occupied by the dictionary (sum of entry lengths).
 size_t onpair_column_dict_bytes(const OnPairColumnHandle* handle);
 
+// --- Decomposition into raw arrays (Vortex layout) ------------------------
+//
+// Borrows pointers to the column's underlying Dictionary + Store vectors.
+// The pointers remain valid until `handle` is freed; the caller is expected
+// to copy them out into Vortex buffers/children and then drop the column.
+
+typedef struct OnPairColumnParts {
+    const uint8_t*  dict_bytes;
+    size_t          dict_bytes_len;       // = dict_offsets[dict_size] (true, unpadded)
+    const uint32_t* dict_offsets;
+    size_t          dict_offsets_len;     // = dict_size + 1
+    const uint64_t* codes_packed;         // LSB-first bit-packed token stream
+    size_t          codes_packed_u64_len; // u64 word count
+    const uint32_t* codes_boundaries;     // per-row token index
+    size_t          codes_boundaries_len; // = num_rows + 1
+    uint32_t        bits;                 // 9..=16
+    size_t          num_rows;
+} OnPairColumnParts;
+
+OnPairStatus onpair_column_parts(
+    const OnPairColumnHandle* handle,
+    OnPairColumnParts*        out_parts);
+
 #ifdef __cplusplus
 } // extern "C"
 #endif

encodings/onpair-sys/src/lib.rs

@@ -115,6 +115,26 @@ pub mod ffi {
             bytes_capacity: usize,
             out_offsets: *mut u64,
         ) -> u32;
+
+        pub fn onpair_column_parts(
+            handle: *const OnPairColumnHandle,
+            out_parts: *mut OnPairColumnParts,
+        ) -> u32;
+    }
+
+    #[repr(C)]
+    #[derive(Debug, Copy, Clone)]
+    pub struct OnPairColumnParts {
+        pub dict_bytes: *const u8,
+        pub dict_bytes_len: usize,
+        pub dict_offsets: *const u32,
+        pub dict_offsets_len: usize,
+        pub codes_packed: *const u64,
+        pub codes_packed_u64_len: usize,
+        pub codes_boundaries: *const u32,
+        pub codes_boundaries_len: usize,
+        pub bits: u32,
+        pub num_rows: usize,
     }
 }
 
@@ -322,8 +342,109 @@ impl Column {
     }
 }
 
+impl Column {
+    /// Borrow the column's raw decomposition: dictionary, bit-packed token
+    /// stream, and per-row boundaries. The returned pointers reference memory
+    /// owned by `self` and remain valid for as long as the column does.
+    pub fn parts(&self) -> Result<Parts<'_>, Error> {
+        let mut raw = OnPairColumnParts {
+            dict_bytes: std::ptr::null(),
+            dict_bytes_len: 0,
+            dict_offsets: std::ptr::null(),
+            dict_offsets_len: 0,
+            codes_packed: std::ptr::null(),
+            codes_packed_u64_len: 0,
+            codes_boundaries: std::ptr::null(),
+            codes_boundaries_len: 0,
+            bits: 0,
+            num_rows: 0,
+        };
+        let status = unsafe { onpair_column_parts(self.handle.as_ptr(), &raw mut raw) };
+        Error::check(status)?;
+        // SAFETY: the C side returns pointers into vectors owned by `self`
+        // (the underlying `OnPairColumn`); they remain valid for `&self`.
+        Ok(unsafe { Parts::from_raw(raw) })
+    }
+}
+
 impl Drop for Column {
     fn drop(&mut self) {
         unsafe { onpair_column_free(self.handle.as_ptr()) }
     }
 }
+
+/// Borrowed view over a column's raw arrays. See [`Column::parts`].
+#[derive(Copy, Clone)]
+pub struct Parts<'a> {
+    /// Concatenated dictionary entry bytes (unpadded).
+    pub dict_bytes: &'a [u8],
+    /// Length `dict_size + 1`; entry `i` spans `dict_bytes[dict_offsets[i]..dict_offsets[i + 1]]`.
+    pub dict_offsets: &'a [u32],
+    /// LSB-first bit-packed token stream, packed `bits` bits per token.
+    pub codes_packed: &'a [u64],
+    /// Length `num_rows + 1`; row `r` spans tokens `codes_boundaries[r]..codes_boundaries[r + 1]`.
+    pub codes_boundaries: &'a [u32],
+    /// Bits per token (9..=16).
+    pub bits: u32,
+    pub num_rows: usize,
+}
+
+impl<'a> Parts<'a> {
+    /// # Safety
+    /// Caller must guarantee the pointers in `raw` are valid for `'a`.
+    unsafe fn from_raw(raw: OnPairColumnParts) -> Self {
+        unsafe {
+            Self {
+                dict_bytes: slice_or_empty(raw.dict_bytes, raw.dict_bytes_len),
+                dict_offsets: slice_or_empty(raw.dict_offsets, raw.dict_offsets_len),
+                codes_packed: slice_or_empty(raw.codes_packed, raw.codes_packed_u64_len),
+                codes_boundaries: slice_or_empty(raw.codes_boundaries, raw.codes_boundaries_len),
+                bits: raw.bits,
+                num_rows: raw.num_rows,
+            }
+        }
+    }
+}
+
+#[inline]
+unsafe fn slice_or_empty<'a, T>(ptr: *const T, len: usize) -> &'a [T] {
+    if ptr.is_null() || len == 0 {
+        &[]
+    } else {
+        unsafe { std::slice::from_raw_parts(ptr, len) }
+    }
+}
+
+/// Read `bits` (1..=16) bits from `packed` starting at LSB-first bit position
+/// `bit_pos`. Matches OnPair's `BitWriter` layout.
+#[inline]
+pub fn read_bits_lsb(packed: &[u64], bit_pos: usize, bits: u32) -> u16 {
+    debug_assert!((1..=16).contains(&bits));
+    let word_idx = bit_pos / 64;
+    // SAFETY of cast: `bit_pos % 64` is always in `0..64`, which fits in u32.
+    #[allow(clippy::cast_possible_truncation)]
+    let bit_off = (bit_pos % 64) as u32;
+    let mask: u64 = (1u64 << bits) - 1;
+    let low = packed[word_idx] >> bit_off;
+    let combined = if bit_off + bits <= 64 {
+        low & mask
+    } else {
+        let high = packed[word_idx + 1] << (64 - bit_off);
+        (low | high) & mask
+    };
+    // SAFETY of cast: `combined` has been masked to at most `bits` (<=16) bits.
+    #[allow(clippy::cast_possible_truncation)]
+    let value = combined as u16;
+    value
+}
+
+/// Decompress an LSB-first bit-packed token stream into a flat `Vec<u16>`,
+/// one element per token. Each `u16` only uses its low `bits` bits.
+pub fn unpack_codes_to_u16(packed: &[u64], total_tokens: usize, bits: u32) -> Vec<u16> {
+    assert!((9..=16).contains(&bits), "bits must be in [9, 16]");
+    let mut out = Vec::with_capacity(total_tokens);
+    for t in 0..total_tokens {
+        out.push(read_bits_lsb(packed, t * bits as usize, bits));
+    }
+    out
+}

encodings/onpair/goldenfiles/onpair.metadata

@@ -1 +1 @@
-�
+