vortex_btrblocks/schemes/string/
onpair.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4//! OnPair short-string compression (dict-12).
5
6use vortex_array::ArrayRef;
7use vortex_array::Canonical;
8use vortex_array::ExecutionCtx;
9use vortex_array::IntoArray;
10use vortex_array::arrays::PrimitiveArray;
11use vortex_array::arrays::primitive::PrimitiveArrayExt;
12use vortex_compressor::estimate::CompressionEstimate;
13use vortex_compressor::estimate::DeferredEstimate;
14use vortex_compressor::scheme::SchemeId;
15use vortex_error::VortexResult;
16use vortex_onpair::DEFAULT_DICT12_CONFIG;
17use vortex_onpair::OnPair;
18use vortex_onpair::OnPairArrayExt;
19use vortex_onpair::OnPairArraySlotsExt;
20use vortex_onpair::onpair_compress;
21
22use crate::ArrayAndStats;
23use crate::CascadingCompressor;
24use crate::CompressorContext;
25use crate::Scheme;
26use crate::SchemeExt;
27use crate::schemes::integer::try_compress_delta;
28
29/// OnPair short-string compression (dict-12).
30///
31/// A default string-fragmentation scheme (alongside [`super::FSSTScheme`]) —
32/// targets large columns of short-to-medium strings with high lexical
33/// overlap, like URLs or log lines. Uses a learned dictionary of frequent
34/// adjacent substrings (built by the OnPair trainer at compress time) and
35/// 12-bit token codes stored as a u16 child, with offsets /
36/// uncompressed-lengths flowing through the cascading compressor like any
37/// other primitive children.
38#[derive(Debug, Copy, Clone, PartialEq, Eq)]
39pub struct OnPairScheme;
40
41impl Scheme for OnPairScheme {
42    fn scheme_name(&self) -> &'static str {
43        "vortex.string.onpair"
44    }
45
46    fn matches(&self, canonical: &Canonical) -> bool {
47        canonical.dtype().is_utf8()
48    }
49
50    /// 4 primitive slot children flow through the cascading compressor:
51    /// `dict_offsets` (u32 → typically `FoR`/`BitPacked`), `codes` (u16 →
52    /// `FastLanes::BitPacked` to exactly `bits` = 12 by default),
53    /// `codes_offsets` (u32 → `FoR`), `uncompressed_lengths` (i32 → narrow
54    /// + `FoR`). Validity stays untouched.
55    fn num_children(&self) -> usize {
56        4
57    }
58
59    fn expected_compression_ratio(
60        &self,
61        _data: &ArrayAndStats,
62        _compress_ctx: CompressorContext,
63        _exec_ctx: &mut ExecutionCtx,
64    ) -> CompressionEstimate {
65        CompressionEstimate::Deferred(DeferredEstimate::Sample)
66    }
67
68    fn compress(
69        &self,
70        compressor: &CascadingCompressor,
71        data: &ArrayAndStats,
72        compress_ctx: CompressorContext,
73        exec_ctx: &mut ExecutionCtx,
74    ) -> VortexResult<ArrayRef> {
75        let utf8 = data.array_as_varbinview().into_owned();
76        let onpair_array = onpair_compress(utf8.as_array(), DEFAULT_DICT12_CONFIG, exec_ctx)?;
77
78        let dict_offsets = compress_offsets_child(
79            compressor,
80            onpair_array.dict_offsets(),
81            &compress_ctx,
82            self.id(),
83            0,
84            exec_ctx,
85        )?;
86        let codes = compress_primitive_child(
87            compressor,
88            onpair_array.codes(),
89            &compress_ctx,
90            self.id(),
91            1,
92            exec_ctx,
93        )?;
94        let codes_offsets = compress_offsets_child(
95            compressor,
96            onpair_array.codes_offsets(),
97            &compress_ctx,
98            self.id(),
99            2,
100            exec_ctx,
101        )?;
102        let uncompressed_lengths = compress_primitive_child(
103            compressor,
104            onpair_array.uncompressed_lengths(),
105            &compress_ctx,
106            self.id(),
107            3,
108            exec_ctx,
109        )?;
110
111        Ok(OnPair::try_new(
112            onpair_array.dtype().clone(),
113            onpair_array.dict_bytes_handle().clone(),
114            dict_offsets,
115            codes,
116            codes_offsets,
117            uncompressed_lengths,
118            onpair_array.array_validity(),
119            onpair_array.bits(),
120        )?
121        .into_array())
122    }
123}
124
125/// Narrow a primitive child to its tightest int type, then forward it to
126/// the cascading compressor.
127fn compress_primitive_child(
128    compressor: &CascadingCompressor,
129    child: &ArrayRef,
130    compress_ctx: &CompressorContext,
131    scheme_id: SchemeId,
132    child_idx: usize,
133    exec_ctx: &mut ExecutionCtx,
134) -> VortexResult<ArrayRef> {
135    let narrowed = child
136        .clone()
137        .execute::<PrimitiveArray>(exec_ctx)?
138        .narrow(exec_ctx)?
139        .into_array();
140    compressor.compress_child(&narrowed, compress_ctx, scheme_id, child_idx, exec_ctx)
141}
142
143/// Minimum child length before delta is even attempted. Delta carries fixed
144/// overhead (a separate `bases` array plus FastLanes' 1024-element lane
145/// packing), so on short children it can only lose.
146const OFFSETS_DELTA_MIN_LEN: usize = 2048;
147
148/// Compress a monotonic offsets child. For children of at least
149/// [`OFFSETS_DELTA_MIN_LEN`] it tries both the normal cascading path and a
150/// delta path and keeps whichever produces fewer bytes; shorter children
151/// skip delta entirely. `dict_offsets` and `codes_offsets` are cumulative
152/// (monotonic), so delta (per-entry deltas) usually packs much tighter than
153/// FoR+bitpacking over the full range.
154fn compress_offsets_child(
155    compressor: &CascadingCompressor,
156    child: &ArrayRef,
157    compress_ctx: &CompressorContext,
158    scheme_id: SchemeId,
159    child_idx: usize,
160    exec_ctx: &mut ExecutionCtx,
161) -> VortexResult<ArrayRef> {
162    let narrowed = child
163        .clone()
164        .execute::<PrimitiveArray>(exec_ctx)?
165        .narrow(exec_ctx)?
166        .into_array();
167    let plain =
168        compressor.compress_child(&narrowed, compress_ctx, scheme_id, child_idx, exec_ctx)?;
169    if narrowed.len() < OFFSETS_DELTA_MIN_LEN {
170        return Ok(plain);
171    }
172    let delta = try_compress_delta(
173        compressor,
174        &narrowed,
175        compress_ctx,
176        scheme_id,
177        child_idx,
178        exec_ctx,
179    )?;
180    if delta.nbytes() < plain.nbytes() {
181        Ok(delta)
182    } else {
183        Ok(plain)
184    }
185}
vortex_btrblocks/schemes/string/onpair.rs

vortex_btrblocks/schemes/string/
onpair.rs
