Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
25 changes: 12 additions & 13 deletions encodings/experimental/onpair/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
55 changes: 31 additions & 24 deletions encodings/experimental/onpair/benches/decode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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.
Expand All @@ -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)
}

Expand All @@ -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);
});
Expand Down
2 changes: 1 addition & 1 deletion encodings/experimental/onpair/goldenfiles/onpair.metadata
Original file line number Diff line number Diff line change
@@ -1 +1 @@
 € €è(08
€ €è(08
Loading
Loading