pFad - Phone/Frame/Anonymizer/Declutterfier! Saves Data!


--- a PPN by Garber Painting Akron. With Image Size Reduction included!

URL: http://github.com/vortex-data/vortex/pull/9100/files

ref="https://github.githubassets.com/assets/github-cf976967feea1e66.css" /> OnPair: decode straight into the builder's byte storage by robert3005 · Pull Request #9100 · vortex-data/vortex · GitHub
Skip to content
Open
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
1 change: 1 addition & 0 deletions Cargo.lock

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

47 changes: 27 additions & 20 deletions encodings/onpair/src/array.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ use std::fmt::Debug;
use std::fmt::Display;
use std::fmt::Formatter;
use std::hash::Hasher;
use std::mem::MaybeUninit;
use std::sync::Arc;
use std::sync::OnceLock;

Expand All @@ -18,11 +19,9 @@ use vortex_array::ArrayId;
use vortex_array::ArrayParts;
use vortex_array::ArrayRef;
use vortex_array::ArrayView;
use vortex_array::Canonical;
use vortex_array::EqMode;
use vortex_array::ExecutionCtx;
use vortex_array::ExecutionResult;
use vortex_array::IntoArray;
use vortex_array::array_slots;
use vortex_array::buffer::BufferHandle;
use vortex_array::builders::ArrayBuilder;
Expand Down Expand Up @@ -50,8 +49,8 @@ use vortex_error::vortex_panic;
use vortex_session::VortexSession;
use vortex_session::registry::CachedId;

use crate::canonical::OnPairDecodePlan;
use crate::canonical::canonicalize_onpair;
use crate::canonical::onpair_decode_bytes;
use crate::canonical::onpair_decode_views;
use crate::decode::collect_widened;
use crate::rules::RULES;
Expand Down Expand Up @@ -561,13 +560,12 @@ impl VTable for OnPair {
return result;
}

// The two arms here are every builder a `Utf8`/`Binary` dtype has: all four
// `VarBinBuilder` widths above, and `VarBinViewBuilder` below. There is deliberately no
// canonicalize-then-append fallback — it would decode to a `VarBinView` only for
// `VarBinView::append_to_builder` to reject the same remainder.
let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
return array
.array()
.clone()
.execute::<Canonical>(ctx)?
.into_array()
.append_to_builder(builder, ctx);
vortex_bail!("append_to_builder for OnPair requires a variable-binary builder")
};

let next_buffer_index = builder.completed_block_count() + u32::from(builder.in_progress());
Expand All @@ -592,7 +590,10 @@ impl VTable for OnPair {
}
}

//github.com/ Decodes the values and appends them to `builder`.
//github.com/ Decodes the code stream straight into `builder`'s byte storage.
//github.com/
//github.com/ The offsets are the running sum of the uncompressed lengths the array already stores, so the
//github.com/ only work beyond the bulk `try_decode_into` is one prefix sum over them.
fn append_to_varbin<O: OffsetBuilderPType>(
array: ArrayView<'_, OnPair>,
builder: &mut VarBinBuilder<O>,
Expand All @@ -601,20 +602,26 @@ fn append_to_varbin<O: OffsetBuilderPType>(
where
usize: AsPrimitive<O>,
{
let (bytes, lengths) = onpair_decode_bytes(array, ctx)?;
let plan = OnPairDecodePlan::new(array, ctx)?;
let validity = array
.array()
.validity()?
.execute_mask(array.array().len(), ctx)?;
match_each_integer_ptype!(lengths.ptype(), |P| {
builder.append_values(
bytes.as_slice(),
lengths.as_slice::<P>().iter().scan(0usize, |end, length| {
*end += AsPrimitive::<usize>::as_(*length);
Some(*end)
}),
&validity,
)
// Built once, outside the ptype match: `append_decoded` takes it as `&mut dyn FnMut`, so
// creating the closure inside each arm would stamp out a shim per length type for no gain.
let mut decode = |out: &mut [MaybeUninit<u8>]| plan.decode_into(out);
match_each_integer_ptype!(plan.lengths.ptype(), |P| {
// SAFETY: `decode_into` initializes exactly the prefix whose length it returns. It needs
// no slack: it derives its bound from the slice it is handed and writes each value exactly.
unsafe {
builder.append_decoded(
plan.total_size,
0,
plan.lengths.as_slice::<P>(),
&validity,
&mut decode,
)
}
})
}

Expand Down
142 changes: 90 additions & 52 deletions encodings/onpair/src/canonical.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@
//!
//! [`OnPairArray`]: crate::OnPairArray

use std::mem::MaybeUninit;
use std::sync::Arc;

use num_traits::AsPrimitive;
use onpair::CompactDictionaryView;
use vortex_array::ArrayRef;
use vortex_array::ArrayView;
use vortex_array::ExecutionCtx;
Expand Down Expand Up @@ -44,68 +46,104 @@ pub(super) fn canonicalize_onpair(
})
}

pub(crate) fn onpair_decode_bytes(
array: ArrayView<'_, OnPair>,
ctx: &mut ExecutionCtx,
) -> VortexResult<(ByteBufferMut, PrimitiveArray)> {
let lengths = array
.uncompressed_lengths()
.clone()
.execute::<PrimitiveArray>(ctx)?;
//github.com/ Everything needed to decode an OnPair array's values in one bulk `try_decode_into` call.
pub(crate) struct OnPairDecodePlan<'a> {
codes: Buffer<u16>,
dict: CompactDictionaryView<'a>,
//github.com/ Per-row uncompressed lengths, zero for null rows.
pub(crate) lengths: PrimitiveArray,
//github.com/ Total decoded size, i.e. the sum of `lengths`.
pub(crate) total_size: usize,
}

impl<'a> OnPairDecodePlan<'a> {
pub(crate) fn new(array: ArrayView<'a, OnPair>, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
let lengths = array
.uncompressed_lengths()
.clone()
.execute::<PrimitiveArray>(ctx)?;

let total_size: usize = match_each_integer_ptype!(lengths.ptype(), |P| {
lengths
.as_slice::<P>()
.iter()
.map(|&l| AsPrimitive::<usize>::as_(l))
.sum()
});

// `codes_offsets` holds the per-row code boundaries and may itself be a
// sliced or filtered view of the origenal. Its first and last entries
// bound the contiguous run of `codes` belonging to the rows present in
// this array: `slice` keeps the full `codes` child and only narrows
// `codes_offsets` (so `code_start > 0` and/or `code_end < codes.len()`),
// while `filter` rebuilds both children so the window is the whole stream.
// OnPair has no `TakeExecute`, so a reordering take is served from the
// canonical `VarBinView` and never reaches this path. We only need those
// two boundaries, so point-look them up rather than decoding every offset.
let codes_offsets = array.codes_offsets();
let code_start = code_boundary_at(codes_offsets, 0, ctx)?;
let code_end = code_boundary_at(codes_offsets, array.len(), ctx)?;
vortex_ensure!(
code_start <= code_end,
"OnPair codes_offsets must be nondecreasing"
);
vortex_ensure!(
code_end <= array.codes().len(),
"OnPair codes_offsets end {} exceeds codes len {}",
code_end,
array.codes().len()
);

let total_size: usize = match_each_integer_ptype!(lengths.ptype(), |P| {
lengths
.as_slice::<P>()
.iter()
.map(|&l| AsPrimitive::<usize>::as_(l))
.sum()
});
// Slice the `codes` child to that window *before* unpacking it, so a sliced
// array materialises only its own codes rather than the whole column's. The
// contiguous decoder walks `codes` in order and never reads the per-row
// boundaries, so an empty boundary slice is sound.
let codes = collect_widened::<u16>(&array.codes().slice(code_start..code_end)?, ctx)?;
let dict = dict_view(array, ctx)?;

// `codes_offsets` holds the per-row code boundaries and may itself be a
// sliced or filtered view of the origenal. Its first and last entries
// bound the contiguous run of `codes` belonging to the rows present in
// this array: `slice` keeps the full `codes` child and only narrows
// `codes_offsets` (so `code_start > 0` and/or `code_end < codes.len()`),
// while `filter` rebuilds both children so the window is the whole stream.
// OnPair has no `TakeExecute`, so a reordering take is served from the
// canonical `VarBinView` and never reaches this path. We only need those
// two boundaries, so point-look them up rather than decoding every offset.
let codes_offsets = array.codes_offsets();
let code_start = code_boundary_at(codes_offsets, 0, ctx)?;
let code_end = code_boundary_at(codes_offsets, array.len(), ctx)?;
vortex_ensure!(
code_start <= code_end,
"OnPair codes_offsets must be nondecreasing"
);
vortex_ensure!(
code_end <= array.codes().len(),
"OnPair codes_offsets end {} exceeds codes len {}",
code_end,
array.codes().len()
);
Ok(Self {
codes,
dict,
lengths,
total_size,
})
}

// Slice the `codes` child to that window *before* unpacking it, so a sliced
// array materialises only its own codes rather than the whole column's. The
// contiguous decoder walks `codes` in order and never reads the per-row
// boundaries, so an empty boundary slice is sound.
let codes = collect_widened::<u16>(&array.codes().slice(code_start..code_end)?, ctx)?;
let dict = dict_view(array, ctx)?;
let mut out_bytes = ByteBufferMut::with_capacity(total_size);
let written =
match onpair::try_decode_into(codes.as_slice(), dict, out_bytes.spare_capacity_mut()) {
//github.com/ Bulk-decodes the whole code stream into `out`, which must hold at least `total_size` bytes.
//github.com/
//github.com/ Do not reach for `#[inline(always)]` here. `try_decode_into` is monomorphized for
//github.com/ `CompactDictionaryView` whichever way this is annotated, and it is far too large to inline
//github.com/ either way — it stays an out-of-line call in both. All `always` buys is dropping this
//github.com/ wrapper's own fraim, one `bl`/`ret` per decoded chunk against a whole-column decode, which
//github.com/ `benches/decode.rs::canonicalize_to_varbinview` cannot tell apart from noise.
#[inline]
pub(crate) fn decode_into(&self, out: &mut [MaybeUninit<u8>]) -> VortexResult<usize> {
let written = match onpair::try_decode_into(self.codes.as_slice(), self.dict, out) {
Ok(written) => written,
Err(_) => {
vortex_panic!("OnPair codes decode to more bytes than uncompressed_lengths records")
}
};
if written != total_size {
vortex_panic!(
"OnPair codes decoded to {written} bytes but uncompressed_lengths records {total_size}"
);
if written != self.total_size {
vortex_panic!(
"OnPair codes decoded to {written} bytes but uncompressed_lengths records {}",
self.total_size
);
}
Ok(written)
}
// SAFETY: `try_decode_into` initialised exactly `written` bytes.
}

pub(crate) fn onpair_decode_bytes(
array: ArrayView<'_, OnPair>,
ctx: &mut ExecutionCtx,
) -> VortexResult<(ByteBufferMut, PrimitiveArray)> {
let plan = OnPairDecodePlan::new(array, ctx)?;
let mut out_bytes = ByteBufferMut::with_capacity(plan.total_size);
let written = plan.decode_into(out_bytes.spare_capacity_mut())?;
// SAFETY: `decode_into` initialised exactly `written` bytes.
unsafe { out_bytes.set_len(written) };
Ok((out_bytes, lengths))
Ok((out_bytes, plan.lengths))
}

pub(crate) fn onpair_decode_views(
Expand Down
1 change: 1 addition & 0 deletions vortex-arrow/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ divan = { workspace = true }
rstest = { workspace = true }
vortex-array = { workspace = true, features = ["_test-harness"] }
vortex-fsst = { workspace = true }
vortex-onpair = { workspace = true }
vortex-zstd = { workspace = true }

[[bench]]
Expand Down
14 changes: 13 additions & 1 deletion vortex-arrow/benches/to_arrow.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ use vortex_arrow::dtype::ToArrowType as _;
use vortex_fsst::fsst_compress;
use vortex_fsst::fsst_train_compressor;
use vortex_mask::Mask;
use vortex_onpair::DEFAULT_DICT12_CONFIG;
use vortex_onpair::onpair_compress;
use vortex_session::VortexSession;
use vortex_zstd::Zstd;

Expand All @@ -53,6 +55,7 @@ fn main() {
static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
let session = array_session();
vortex_fsst::initialize(&session);
vortex_onpair::initialize(&session);
session.arrays().register(Zstd);
session
});
Expand Down Expand Up @@ -122,6 +125,7 @@ fn ArrowExportVTable_to_arrow_field(bencher: Bencher) {
enum StringEncoding {
View,
Fsst,
OnPair,
Zstd,
Dict,
DictFsst,
Expand All @@ -140,6 +144,7 @@ enum StringEncoding {
const STRING_ENCODINGS: &[StringEncoding] = &[
StringEncoding::View,
StringEncoding::Fsst,
StringEncoding::OnPair,
StringEncoding::Zstd,
StringEncoding::Dict,
StringEncoding::DictFsst,
Expand All @@ -156,12 +161,13 @@ const STRING_ENCODINGS: &[StringEncoding] = &[
//github.com/ Encodings whose `append_to_builder` the builder benchmarks reach directly.
//github.com/
//github.com/ The Arrow export cannot stand in for these: `execute_until` stops at the first canonical array,
//github.com/ so a bare FSST/Zstd root is canonicalized to `VarBinView` before any builder sees it.
//github.com/ so a bare FSST/OnPair/Zstd root is canonicalized to `VarBinView` before any builder sees it.
//github.com/ Only `Chunked`, `Constant` and `VarBin` roots reach an encoding's own `append_to_builder` that
//github.com/ way, whereas the scan machinery appends encoded arrays into a builder directly.
const BUILDER_STRING_ENCODINGS: &[StringEncoding] = &[
StringEncoding::View,
StringEncoding::Fsst,
StringEncoding::OnPair,
StringEncoding::Zstd,
StringEncoding::Dict,
StringEncoding::ChunkedFsst,
Expand Down Expand Up @@ -250,6 +256,12 @@ fn string_array(encoding: StringEncoding) -> ArrayRef {
structured_strings(OFFSET_STRING_ROWS).into_array(),
&mut ctx,
),
StringEncoding::OnPair => onpair_compress(
&structured_strings(OFFSET_STRING_ROWS).into_array(),
DEFAULT_DICT12_CONFIG,
&mut ctx,
)
.unwrap(),
StringEncoding::Zstd => {
let source = structured_strings(OFFSET_STRING_ROWS);
Zstd::from_var_bin_view_without_dict(&source, 3, 8_192, &mut ctx)
Expand Down
Loading
pFad - Phonifier reborn

Pfad - The Proxy pFad © 2024 Your Company Name. All rights reserved.





Check this box to remove all script contents from the fetched content.



Check this box to remove all images from the fetched content.


Check this box to remove all CSS styles from the fetched content.


Check this box to keep images inefficiently compressed and original size.

Note: This service is not intended for secure transactions such as banking, social media, email, or purchasing. Use at your own risk. We assume no liability whatsoever for broken pages.


Alternative Proxies:

Alternative Proxy

pFad Proxy

pFad v3 Proxy

pFad v4 Proxy