From c2e946adcce311c4e7c11f1b91340f78f57dd97e Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Tue, 8 Sep 2026 17:19:03 +0100 Subject: [PATCH 1/6] perf(onpair): update dependency and train directly from rows Signed-off-by: Francesco Gargiulo --- Cargo.lock | 5 ++- Cargo.toml | 2 +- encodings/onpair/src/compress.rs | 62 ++++++++++++++++++++------------ 3 files changed, 42 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 159b3b020d4..86746c83d4e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7026,9 +7026,8 @@ checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" [[package]] name = "onpair" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e546e7cd983c9998a5a52e9c7eecaf38e518f2e6fc9c9ec649e8b53748a96bb0" +version = "0.2.1" +source = "git+https://github.com/spiraldb/onpair?rev=9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b#9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b" dependencies = [ "hashbrown 0.16.1", "rand 0.9.5", diff --git a/Cargo.toml b/Cargo.toml index 3c0368e190e..515de40af91 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -203,7 +203,7 @@ num_enum = { version = "0.7.3", default-features = false } object_store = { version = "0.13.2", default-features = false } object_store_opendal = "0.58.0" once_cell = "1.21.4" -onpair = "0.2.0" +onpair = { version = "0.2.1", git = "https://github.com/spiraldb/onpair", rev = "9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b" } opendal = { version = "0.58.1", default-features = false } opentelemetry = "0.32.0" opentelemetry-otlp = "0.32.0" diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index e2cf69353ec..d73678dde60 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -4,6 +4,7 @@ //! Train + compress entry points for the OnPair encoding. use onpair::Config; +use onpair::Rows; use vortex_array::ArrayRef; use vortex_array::ExecutionCtx; use vortex_array::IntoArray; @@ -19,7 +20,6 @@ use vortex_buffer::ByteBuffer; use vortex_buffer::ByteBufferMut; use vortex_error::VortexExpect; use vortex_error::VortexResult; -use vortex_error::vortex_err; use vortex_mask::AllOr; use crate::OnPair; @@ -43,14 +43,8 @@ pub fn onpair_compress( } let views = array.views(); - let flat_bytes: usize = views.iter().map(|v| v.len() as usize).sum(); - - // TODO(francesco): we flatten because onpair training needs a contiguous `(bytes, offsets)` - // pair. Allowing onpair to train on a slice-of-slices would let us skip this copy. - let mut flat: Vec = Vec::with_capacity(flat_bytes); - let mut offsets: Vec = Vec::with_capacity(len + 1); let mut uncompressed_lengths: BufferMut = BufferMut::with_capacity(len); - offsets.push(0); + let mut total_bytes = 0usize; let buffers = array .data_buffers() .as_ref() @@ -61,32 +55,32 @@ pub fn onpair_compress( match mask.bit_buffer() { AllOr::All => { for view in views { - let bytes = view_bytes(view, &buffers); - flat.extend_from_slice(bytes); - offsets.push(u64::try_from(flat.len()).vortex_expect("offset must fit in u64")); uncompressed_lengths .push(i32::try_from(view.len()).vortex_expect("must fit in i32")); + total_bytes += view.len() as usize; } } AllOr::None => unreachable!("all-null input handled above"), AllOr::Some(validity) => { for (view, valid) in views.iter().zip(validity.iter()) { if valid { - let bytes = view_bytes(view, &buffers); - flat.extend_from_slice(bytes); - offsets.push(u64::try_from(flat.len()).vortex_expect("offset must fit in u64")); uncompressed_lengths .push(i32::try_from(view.len()).vortex_expect("must fit in i32")); + total_bytes += view.len() as usize; } else { - offsets.push(u64::try_from(flat.len()).vortex_expect("offset must fit in u64")); uncompressed_lengths.push(0); } } } } - let column = onpair::compress(&flat, &offsets, config) - .map_err(|e| vortex_err!("OnPair compress failed: {e}"))?; + let rows = ViewRows { + views, + buffers: &buffers, + lengths: uncompressed_lengths.as_slice(), + total_bytes, + }; + let column = onpair::compress_rows::<_, u64>(&rows, config); let (dict, codes, row_offsets) = column.into_raw(); let (dict_bytes, dict_offsets) = dict.into_raw(); let codes_offsets = codes_offsets_array(&row_offsets); @@ -113,12 +107,34 @@ pub fn onpair_compress( Ok(encoded.into_array()) } -fn view_bytes<'a>(view: &'a BinaryView, buffers: &'a [&ByteBuffer]) -> &'a [u8] { - if view.is_inlined() { - view.as_inlined().value() - } else { - let view_ref = view.as_view(); - &buffers[view_ref.buffer_index as usize][view_ref.as_range()] +/// Reads inline and external values in place, treating null rows as empty. +struct ViewRows<'a> { + views: &'a [BinaryView], + buffers: &'a [&'a ByteBuffer], + lengths: &'a [i32], + total_bytes: usize, +} + +impl Rows for ViewRows<'_> { + fn num_rows(&self) -> usize { + self.views.len() + } + + fn total_bytes(&self) -> usize { + self.total_bytes + } + + #[inline] + fn row(&self, i: usize) -> &[u8] { + let view = &self.views[i]; + if self.lengths[i] == 0 { + &[] + } else if view.is_inlined() { + view.as_inlined().value() + } else { + let view_ref = view.as_view(); + &self.buffers[view_ref.buffer_index as usize][view_ref.as_range()] + } } } From fd0ae9e2c3adedcc0e27b047c5ec7adecb04e306 Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Tue, 8 Sep 2026 17:54:25 +0100 Subject: [PATCH 2/6] build(onpair): use published 0.2.1 release Signed-off-by: Francesco Gargiulo --- Cargo.lock | 3 ++- Cargo.toml | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 86746c83d4e..f7c9dcaa4f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7027,7 +7027,8 @@ checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" [[package]] name = "onpair" version = "0.2.1" -source = "git+https://github.com/spiraldb/onpair?rev=9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b#9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34272d1ea63266148a9a5b2b96db80a45245d034862302735f2f93413a6f9172" dependencies = [ "hashbrown 0.16.1", "rand 0.9.5", diff --git a/Cargo.toml b/Cargo.toml index 515de40af91..7a7c94aee79 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -203,7 +203,7 @@ num_enum = { version = "0.7.3", default-features = false } object_store = { version = "0.13.2", default-features = false } object_store_opendal = "0.58.0" once_cell = "1.21.4" -onpair = { version = "0.2.1", git = "https://github.com/spiraldb/onpair", rev = "9a9b4b46c4df10fa8dbe8a9b9400727fde9f714b" } +onpair = "0.2.1" opendal = { version = "0.58.1", default-features = false } opentelemetry = "0.32.0" opentelemetry-otlp = "0.32.0" From ab288626cf6ed96937cf6cd2a526f757387beb32 Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Wed, 9 Sep 2026 11:28:49 +0100 Subject: [PATCH 3/6] fix(onpair): store uncompressed row lengths as u32 Signed-off-by: Francesco Gargiulo --- encodings/onpair/src/compress.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index d73678dde60..e1bf2bce52c 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -43,7 +43,7 @@ pub fn onpair_compress( } let views = array.views(); - let mut uncompressed_lengths: BufferMut = BufferMut::with_capacity(len); + let mut uncompressed_lengths: BufferMut = BufferMut::with_capacity(len); let mut total_bytes = 0usize; let buffers = array .data_buffers() @@ -55,8 +55,7 @@ pub fn onpair_compress( match mask.bit_buffer() { AllOr::All => { for view in views { - uncompressed_lengths - .push(i32::try_from(view.len()).vortex_expect("must fit in i32")); + uncompressed_lengths.push(view.len()); total_bytes += view.len() as usize; } } @@ -64,8 +63,7 @@ pub fn onpair_compress( AllOr::Some(validity) => { for (view, valid) in views.iter().zip(validity.iter()) { if valid { - uncompressed_lengths - .push(i32::try_from(view.len()).vortex_expect("must fit in i32")); + uncompressed_lengths.push(view.len()); total_bytes += view.len() as usize; } else { uncompressed_lengths.push(0); @@ -111,7 +109,7 @@ pub fn onpair_compress( struct ViewRows<'a> { views: &'a [BinaryView], buffers: &'a [&'a ByteBuffer], - lengths: &'a [i32], + lengths: &'a [u32], total_bytes: usize, } From e22376638d4b4c61d0498fc14edcd9978eef8772 Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Wed, 9 Sep 2026 11:44:31 +0100 Subject: [PATCH 4/6] refactor(onpair): reuse cached view lengths Signed-off-by: Francesco Gargiulo --- encodings/onpair/src/compress.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index e1bf2bce52c..9e1b75d4bc9 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -55,16 +55,18 @@ pub fn onpair_compress( match mask.bit_buffer() { AllOr::All => { for view in views { - uncompressed_lengths.push(view.len()); - total_bytes += view.len() as usize; + let length = view.len(); + uncompressed_lengths.push(length); + total_bytes += length as usize; } } AllOr::None => unreachable!("all-null input handled above"), AllOr::Some(validity) => { for (view, valid) in views.iter().zip(validity.iter()) { if valid { - uncompressed_lengths.push(view.len()); - total_bytes += view.len() as usize; + let length = view.len(); + uncompressed_lengths.push(length); + total_bytes += length as usize; } else { uncompressed_lengths.push(0); } From 0ba1757de6b74fef6da6e497231cf2d838ea1fec Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Wed, 9 Sep 2026 12:05:39 +0100 Subject: [PATCH 5/6] refactor(onpair): zip views with uncompressed lengths Signed-off-by: Francesco Gargiulo --- encodings/onpair/src/compress.rs | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index 9e1b75d4bc9..1b0f4c68d18 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -43,7 +43,7 @@ pub fn onpair_compress( } let views = array.views(); - let mut uncompressed_lengths: BufferMut = BufferMut::with_capacity(len); + let mut uncompressed_lengths: BufferMut = BufferMut::zeroed(len); let mut total_bytes = 0usize; let buffers = array .data_buffers() @@ -54,21 +54,21 @@ pub fn onpair_compress( match mask.bit_buffer() { AllOr::All => { - for view in views { - let length = view.len(); - uncompressed_lengths.push(length); - total_bytes += length as usize; + for (view, length) in views.iter().zip(uncompressed_lengths.iter_mut()) { + *length = view.len(); + total_bytes += *length as usize; } } AllOr::None => unreachable!("all-null input handled above"), AllOr::Some(validity) => { - for (view, valid) in views.iter().zip(validity.iter()) { + for ((view, length), valid) in views + .iter() + .zip(uncompressed_lengths.iter_mut()) + .zip(validity.iter()) + { if valid { - let length = view.len(); - uncompressed_lengths.push(length); - total_bytes += length as usize; - } else { - uncompressed_lengths.push(0); + *length = view.len(); + total_bytes += *length as usize; } } } From 50f03079b6f0d3c91080606f7b4b1c18c18750b5 Mon Sep 17 00:00:00 2001 From: Francesco Gargiulo Date: Thu, 10 Sep 2026 14:19:03 +0100 Subject: [PATCH 6/6] perf(onpair): iterate only valid rows when collecting lengths Signed-off-by: Francesco Gargiulo --- encodings/onpair/src/compress.rs | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/encodings/onpair/src/compress.rs b/encodings/onpair/src/compress.rs index 1b0f4c68d18..b46428445d8 100644 --- a/encodings/onpair/src/compress.rs +++ b/encodings/onpair/src/compress.rs @@ -44,7 +44,6 @@ pub fn onpair_compress( let views = array.views(); let mut uncompressed_lengths: BufferMut = BufferMut::zeroed(len); - let mut total_bytes = 0usize; let buffers = array .data_buffers() .as_ref() @@ -52,27 +51,29 @@ pub fn onpair_compress( .map(|b| b.as_host()) .collect::>(); - match mask.bit_buffer() { + // Keep sums local to each arm: a shared callback accumulator can prevent + // vectorization of the all-valid loop. + let total_bytes = match mask.bit_buffer() { AllOr::All => { + let mut total = 0; for (view, length) in views.iter().zip(uncompressed_lengths.iter_mut()) { *length = view.len(); - total_bytes += *length as usize; + total += *length as usize; } + total } AllOr::None => unreachable!("all-null input handled above"), AllOr::Some(validity) => { - for ((view, length), valid) in views - .iter() - .zip(uncompressed_lengths.iter_mut()) - .zip(validity.iter()) - { - if valid { - *length = view.len(); - total_bytes += *length as usize; - } - } + let lengths = uncompressed_lengths.as_mut_slice(); + let mut total = 0; + validity.for_each_set_index(|i| { + let length = views[i].len(); + lengths[i] = length; + total += length as usize; + }); + total } - } + }; let rows = ViewRows { views,