Wire OnPair as a btrblocks string scheme

* Extract a small `parts_to_children` helper in `vortex-onpair`'s
  `compress.rs` so the lift-out-of-C++ step reads top-to-bottom rather
  than via a block-and-drop dance.

* Add `OnPairScheme` to `vortex-btrblocks::schemes::string`. The scheme
  matches utf8 strings, declares its four primitive children
  (dict_offsets / codes / codes_offsets / uncompressed_lengths) so the
  cascading compressor can re-encode them downstream
  (FastLanes-bit-pack on `codes`, etc.), defers the compression-ratio
  estimate to the sample-based path (same as FSST / Zstd), and
  reassembles the result via `OnPair::try_new`.

* Feature-gate it via a new `onpair` Cargo feature, enabled by default,
  so out-of-the-box `BtrBlocksCompressorBuilder::default()` includes it
  in `ALL_SCHEMES` and consumers without a C++ toolchain can opt out
  with `default-features = false`.

* Update the FSST scheme-selection test to accept either FSST or OnPair
  as the winning encoding — both target the same workload (short
  strings with high lexical overlap) and the sample-based selector now
  picks the one with the better ratio on the test corpus.

Test results
  vortex-onpair       7 unit + 1 100k smoke   all green
  vortex-btrblocks   36 unit + 3 doctests     all green (incl. new
                                              `test_onpair_in_default_scheme_list`)

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

Author: claude

Cargo.lock

@@ -80,35 +80,7 @@ where
 
     let column = Column::compress(&flat, &offsets, config)
         .map_err(|e| vortex_err!("OnPair compress failed: {e}"))?;
-
-    let bits;
-    let dict_bytes;
-    let dict_offsets;
-    let codes;
-    let codes_offsets;
-    {
-        let parts = column
-            .parts()
-            .map_err(|e| vortex_err!("OnPair parts failed: {e}"))?;
-        bits = parts.bits;
-
-        // Last dict_offset = total token bytes; unpack into a single
-        // contiguous ByteBuffer for the Vortex `dict_bytes` blob.
-        dict_bytes = BufferHandle::new_host(ByteBuffer::from(parts.dict_bytes.to_vec()));
-        dict_offsets = Buffer::<u32>::copy_from(parts.dict_offsets).into_array();
-
-        let total_tokens = *parts
-            .codes_boundaries
-            .last()
-            .ok_or_else(|| vortex_err!("OnPair: missing boundaries"))?
-            as usize;
-        let codes_vec = unpack_codes_to_u16(parts.codes_packed, total_tokens, bits);
-        codes = Buffer::<u16>::copy_from(codes_vec).into_array();
-
-        // Token-index boundaries are exactly the offsets into our flat u16
-        // `codes` array, so we can use them as-is.
-        codes_offsets = Buffer::<u32>::copy_from(parts.codes_boundaries).into_array();
-    }
+    let (bits, dict_bytes, dict_offsets, codes, codes_offsets) = parts_to_children(&column)?;
     drop(column);
 
     let uncompressed_lengths = uncompressed_lengths.into_array();
@@ -129,6 +101,30 @@ where
     )
 }
 
+/// Borrow the raw C++ parts and lift them into owned Vortex children.
+/// Returns `(bits, dict_bytes, dict_offsets, codes, codes_offsets)`.
+fn parts_to_children(
+    column: &Column,
+) -> VortexResult<(u32, BufferHandle, ArrayRef, ArrayRef, ArrayRef)> {
+    let parts = column
+        .parts()
+        .map_err(|e| vortex_err!("OnPair parts failed: {e}"))?;
+    let bits = parts.bits;
+    let dict_bytes = BufferHandle::new_host(ByteBuffer::from(parts.dict_bytes.to_vec()));
+    let dict_offsets = Buffer::<u32>::copy_from(parts.dict_offsets).into_array();
+    let total_tokens = usize::try_from(
+        *parts
+            .codes_boundaries
+            .last()
+            .ok_or_else(|| vortex_err!("OnPair: missing codes_boundaries"))?,
+    )
+    .map_err(|_| vortex_err!("OnPair: total_tokens does not fit in usize"))?;
+    let codes_vec = unpack_codes_to_u16(parts.codes_packed, total_tokens, bits);
+    let codes = Buffer::<u16>::copy_from(codes_vec).into_array();
+    let codes_offsets = Buffer::<u32>::copy_from(parts.codes_boundaries).into_array();
+    Ok((bits, dict_bytes, dict_offsets, codes, codes_offsets))
+}
+
 /// Compress a byte-string accessor (typically a `VarBinArray` or
 /// `VarBinViewArray`).
 pub fn onpair_compress<A: ArrayAccessor<[u8]>>(

encodings/onpair/src/compress.rs

@@ -30,6 +30,7 @@ vortex-error = { workspace = true }
 vortex-fastlanes = { workspace = true }
 vortex-fsst = { workspace = true }
 vortex-mask = { workspace = true }
+vortex-onpair = { workspace = true, optional = true }
 vortex-pco = { workspace = true, optional = true }
 vortex-runend = { workspace = true }
 vortex-sequence = { workspace = true }
@@ -47,8 +48,10 @@ vortex-array = { workspace = true, features = ["_test-harness"] }
 vortex-session = { workspace = true }
 
 [features]
+default = ["onpair"]
 # This feature enabled unstable encodings for which we don't guarantee stability.
 unstable_encodings = ["dep:vortex-tensor", "vortex-zstd?/unstable_encodings"]
+onpair = ["dep:vortex-onpair"]
 pco = ["dep:pco", "dep:vortex-pco"]
 zstd = ["dep:vortex-zstd"]
 

vortex-btrblocks/Cargo.toml

@@ -54,6 +54,8 @@ pub const ALL_SCHEMES: &[&dyn Scheme] = &[
     ////////////////////////////////////////////////////////////////////////////////////////////////
     &string::StringDictScheme,
     &string::FSSTScheme,
+    #[cfg(feature = "onpair")]
+    &string::OnPairScheme,
     &string::StringConstantScheme,
     &string::NullDominatedSparseScheme,
     // Decimal schemes.

vortex-btrblocks/src/builder.rs

@@ -21,6 +21,14 @@ use vortex_fsst::FSST;
 use vortex_fsst::FSSTArrayExt;
 use vortex_fsst::fsst_compress;
 use vortex_fsst::fsst_train_compressor;
+#[cfg(feature = "onpair")]
+use vortex_onpair::DEFAULT_DICT12_CONFIG;
+#[cfg(feature = "onpair")]
+use vortex_onpair::OnPair;
+#[cfg(feature = "onpair")]
+use vortex_onpair::OnPairArrayExt;
+#[cfg(feature = "onpair")]
+use vortex_onpair::onpair_compress;
 use vortex_sparse::Sparse;
 use vortex_sparse::SparseExt as _;
 
@@ -36,6 +44,18 @@ use crate::SchemeExt;
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct FSSTScheme;
 
+/// OnPair short-string compression (dict-12, FSST-shape children).
+///
+/// Targets the same workload as FSST — large columns of short-to-medium
+/// strings with high lexical overlap — but uses a learned dictionary of
+/// frequent adjacent substrings and 12-bit codes. The codes / offsets /
+/// uncompressed-lengths children all flow through the cascading compressor
+/// the same way FSST's do, so any downstream bit-packing / FoR / etc. still
+/// applies.
+#[cfg(feature = "onpair")]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub struct OnPairScheme;
+
 /// Sparse encoding for null-dominated arrays.
 ///
 /// This is the same as the integer `SparseScheme`, but we only use this for null-dominated arrays.
@@ -138,6 +158,108 @@ impl Scheme for FSSTScheme {
     }
 }
 
+#[cfg(feature = "onpair")]
+impl Scheme for OnPairScheme {
+    fn scheme_name(&self) -> &'static str {
+        "vortex.string.onpair"
+    }
+
+    fn matches(&self, canonical: &Canonical) -> bool {
+        is_utf8_string(canonical)
+    }
+
+    /// Children, in slot order:
+    /// 0 = dict_offsets, 1 = codes, 2 = codes_offsets, 3 = uncompressed_lengths.
+    /// Validity is handled separately by the outer array.
+    fn num_children(&self) -> usize {
+        4
+    }
+
+    fn expected_compression_ratio(
+        &self,
+        _data: &ArrayAndStats,
+        _compress_ctx: CompressorContext,
+        _exec_ctx: &mut ExecutionCtx,
+    ) -> CompressionEstimate {
+        CompressionEstimate::Deferred(DeferredEstimate::Sample)
+    }
+
+    fn compress(
+        &self,
+        compressor: &CascadingCompressor,
+        data: &ArrayAndStats,
+        compress_ctx: CompressorContext,
+        exec_ctx: &mut ExecutionCtx,
+    ) -> VortexResult<ArrayRef> {
+        let utf8 = data.array_as_utf8().into_owned();
+        let onpair_array = onpair_compress(&utf8, utf8.len(), utf8.dtype(), DEFAULT_DICT12_CONFIG)?;
+
+        let dict_offsets = compress_primitive_child(
+            compressor,
+            onpair_array.dict_offsets(),
+            &compress_ctx,
+            self.id(),
+            0,
+            exec_ctx,
+        )?;
+        let codes = compress_primitive_child(
+            compressor,
+            onpair_array.codes(),
+            &compress_ctx,
+            self.id(),
+            1,
+            exec_ctx,
+        )?;
+        let codes_offsets = compress_primitive_child(
+            compressor,
+            onpair_array.codes_offsets(),
+            &compress_ctx,
+            self.id(),
+            2,
+            exec_ctx,
+        )?;
+        let uncompressed_lengths = compress_primitive_child(
+            compressor,
+            onpair_array.uncompressed_lengths(),
+            &compress_ctx,
+            self.id(),
+            3,
+            exec_ctx,
+        )?;
+
+        Ok(OnPair::try_new(
+            onpair_array.dtype().clone(),
+            onpair_array.dict_bytes_handle().clone(),
+            dict_offsets,
+            codes,
+            codes_offsets,
+            uncompressed_lengths,
+            onpair_array.array_validity(),
+            onpair_array.bits(),
+        )?
+        .into_array())
+    }
+}
+
+/// Helper: narrow a primitive child to its tightest int type, then hand it
+/// off to the cascading compressor.
+#[cfg(feature = "onpair")]
+fn compress_primitive_child(
+    compressor: &CascadingCompressor,
+    child: &ArrayRef,
+    compress_ctx: &CompressorContext,
+    scheme_id: vortex_compressor::scheme::SchemeId,
+    child_idx: usize,
+    exec_ctx: &mut ExecutionCtx,
+) -> VortexResult<ArrayRef> {
+    let narrowed = child
+        .clone()
+        .execute::<PrimitiveArray>(exec_ctx)?
+        .narrow()?
+        .into_array();
+    compressor.compress_child(&narrowed, compress_ctx, scheme_id, child_idx, exec_ctx)
+}
+
 impl Scheme for NullDominatedSparseScheme {
     fn scheme_name(&self) -> &'static str {
         "vortex.string.sparse"
@@ -411,8 +533,24 @@ mod scheme_selection_tests {
         Ok(())
     }
 
+    #[cfg(feature = "onpair")]
+    #[test]
+    fn test_onpair_in_default_scheme_list() {
+        use crate::SchemeExt;
+        use crate::schemes::string::OnPairScheme;
+
+        let ids: Vec<_> = crate::ALL_SCHEMES.iter().map(|s| s.id()).collect();
+        assert!(
+            ids.contains(&OnPairScheme.id()),
+            "OnPairScheme not registered in ALL_SCHEMES"
+        );
+    }
+
     #[test]
-    fn test_fsst_compressed() -> VortexResult<()> {
+    fn test_dictionary_string_scheme_compressed() -> VortexResult<()> {
+        // Dictionary-style string corpus: high lexical overlap, short rows.
+        // FSST and OnPair both target this shape; the cascading compressor
+        // picks whichever samples better, so accept either.
         let mut strings = Vec::with_capacity(1000);
         for i in 0..1000 {
             strings.push(Some(format!(
@@ -423,7 +561,16 @@ mod scheme_selection_tests {
         let array_ref = array.into_array();
         let compressed = BtrBlocksCompressor::default()
             .compress(&array_ref, &mut SESSION.create_execution_ctx())?;
-        assert!(compressed.is::<FSST>());
+        let is_fsst = compressed.is::<FSST>();
+        #[cfg(feature = "onpair")]
+        let is_onpair = compressed.is::<vortex_onpair::OnPair>();
+        #[cfg(not(feature = "onpair"))]
+        let is_onpair = false;
+        assert!(
+            is_fsst || is_onpair,
+            "expected FSST or OnPair, got {}",
+            compressed.encoding_id()
+        );
         Ok(())
     }
 }