Skip to content

Replace DynVarBinBuilder with a generic VarBinBuilder #9096

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robert3005 wants to merge 4 commits into from
Open

Replace DynVarBinBuilder with a generic VarBinBuilder #9096

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
79d9d41
Replace DynVarBinBuilder with a generic VarBinBuilder
robert3005 Jul 30, 2026
105a560
DCO Remediation Commit for Robert Kruszewski <github@robertk.io>
robert3005 Jul 31, 2026
b81a551
fixes
robert3005 Jul 31, 2026
e072374
Address review: cover every offset width and push values into the bui…
robert3005 Jul 31, 2026
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
View file
Open in desktop

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

51 changes: 33 additions & 18 deletions encodings/fsst/src/array.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,15 @@ use vortex_array::arrays::VarBinArray;
use vortex_array::arrays::varbin::VarBinArraySlotsExt;
use vortex_array::buffer::BufferHandle;
use vortex_array::builders::ArrayBuilder;
use vortex_array::builders::DynVarBinBuilder;
use vortex_array::builders::VarBinBuilder;
use vortex_array::builders::VarBinViewBuilder;
use vortex_array::dtype::DType;
use vortex_array::dtype::Nullability;
use vortex_array::dtype::OffsetBuilderPType;
use vortex_array::dtype::PType;
use vortex_array::legacy_session;
use vortex_array::match_each_integer_ptype;
use vortex_array::match_each_varbin_builder;
use vortex_array::serde::ArrayChildren;
use vortex_array::validity::Validity;
use vortex_array::vtable::VTable;
Expand Down Expand Up @@ -307,23 +309,10 @@ impl VTable for FSST {
builder: &mut dyn ArrayBuilder,
ctx: &mut ExecutionCtx,
) -> VortexResult<()> {
if let Some(builder) = builder.as_any_mut().downcast_mut::<DynVarBinBuilder>() {
let (bytes, lengths) = fsst_decode_bytes(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,
)
})?;
return Ok(());
if let Some(result) =
match_each_varbin_builder!(builder, |builder| append_to_varbin(array, builder, ctx))
{
return result;
}

let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
Expand Down Expand Up @@ -361,6 +350,32 @@ impl VTable for FSST {
}
}

/// Decompresses the values and appends them to `builder`.
fn append_to_varbin<O: OffsetBuilderPType>(
array: ArrayView<'_, FSST>,
builder: &mut VarBinBuilder<O>,
ctx: &mut ExecutionCtx,
) -> VortexResult<()>
where
usize: AsPrimitive<O>,
{
let (bytes, lengths) = fsst_decode_bytes(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,
)
})
}

#[array_slots(FSST)]
pub struct FSSTSlots {
/// Lengths of the original values before compression, can be compressed.
Expand Down
4 changes: 2 additions & 2 deletions encodings/fsst/src/canonical.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ mod tests {
use vortex_array::arrays::VarBinViewArray;
use vortex_array::arrays::varbin::VarBinArrayExt;
use vortex_array::builders::ArrayBuilder;
use vortex_array::builders::DynVarBinBuilder;
use vortex_array::builders::VarBinBuilder;
use vortex_array::builders::VarBinViewBuilder;
use vortex_array::dtype::DType;
use vortex_array::dtype::Nullability;
Expand Down Expand Up @@ -212,7 +212,7 @@ mod tests {

{
let mut builder =
DynVarBinBuilder::with_capacity(chunked_arr.dtype().clone(), false, data.len());
VarBinBuilder::<i32>::with_capacity(chunked_arr.dtype().clone(), data.len());
chunked_arr
.into_array()
.append_to_builder(&mut builder, &mut ctx)?;
Expand Down
31 changes: 20 additions & 11 deletions encodings/fsst/src/compress.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ use vortex_array::arrays::varbinview::BinaryView;
use vortex_array::buffer::BufferHandle;
use vortex_array::dtype::DType;
use vortex_array::dtype::IntegerPType;
use vortex_array::dtype::OffsetBuilderPType;
use vortex_array::match_each_integer_ptype;
use vortex_buffer::Buffer;
use vortex_buffer::BufferMut;
Expand Down Expand Up @@ -197,9 +198,13 @@ fn compress_views<O>(
ctx: &mut ExecutionCtx,
) -> VortexResult<FSSTArray>
where
O: IntegerPType + 'static,
O: OffsetBuilderPType + 'static,
{
let mut sink = FsstSink::<O>::with_capacity(strings.len(), compressor);
let mut sink = FsstSink::<O>::with_capacity(
DType::Binary(strings.dtype().nullability()),
strings.len(),
compressor,
);
let views = strings.views();
let buffers = strings.data_buffers();
match mask.bit_buffer() {
Expand Down Expand Up @@ -230,9 +235,13 @@ fn compress_varbin<O>(
ctx: &mut ExecutionCtx,
) -> VortexResult<FSSTArray>
where
O: IntegerPType + 'static,
O: OffsetBuilderPType + 'static,
{
let mut sink = FsstSink::<O>::with_capacity(strings.len(), compressor);
let mut sink = FsstSink::<O>::with_capacity(
DType::Binary(strings.dtype().nullability()),
strings.len(),
compressor,
);
let bytes = strings.bytes().as_slice();
match_each_integer_ptype!(offsets.ptype(), |I| {
let off = offsets.as_slice::<I>();
Expand Down Expand Up @@ -269,18 +278,18 @@ where
}

/// Per-row output state for an FSST compression pass.
struct FsstSink<'c, O: IntegerPType + 'static> {
struct FsstSink<'c, O: OffsetBuilderPType + 'static> {
buffer: Vec<u8>,
builder: VarBinBuilder<O>,
uncompressed_lengths: BufferMut<i32>,
compressor: &'c Compressor,
}

impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
fn with_capacity(len: usize, compressor: &'c Compressor) -> Self {
impl<'c, O: OffsetBuilderPType + 'static> FsstSink<'c, O> {
fn with_capacity(dtype: DType, len: usize, compressor: &'c Compressor) -> Self {
Self {
buffer: Vec::with_capacity(DEFAULT_BUFFER_LEN),
builder: VarBinBuilder::<O>::with_capacity(len),
builder: VarBinBuilder::<O>::with_capacity(dtype, len),
uncompressed_lengths: BufferMut::with_capacity(len),
compressor,
}
Expand All @@ -289,7 +298,7 @@ impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
#[inline]
fn emit(&mut self, row: Option<&[u8]>) {
let Some(s) = row else {
self.builder.append_null();
self.builder.push_null();
self.uncompressed_lengths.push(0);
return;
};
Expand All @@ -310,8 +319,8 @@ impl<'c, O: IntegerPType + 'static> FsstSink<'c, O> {
self.builder.append_value(&self.buffer);
}

fn finish(self, dtype: DType, ctx: &mut ExecutionCtx) -> VortexResult<FSSTArray> {
let codes = self.builder.finish(DType::Binary(dtype.nullability()));
fn finish(mut self, dtype: DType, ctx: &mut ExecutionCtx) -> VortexResult<FSSTArray> {
let codes = self.builder.finish_into_varbin();
FSST::try_new(
dtype,
Buffer::copy_from(self.compressor.symbol_table()),
Expand Down
28 changes: 15 additions & 13 deletions encodings/fsst/src/kernel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,8 @@ mod tests {
});

fn build_test_fsst_array() -> ArrayRef {
let mut builder = VarBinBuilder::<i32>::with_capacity(10);
let mut builder =
VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 10);
builder.append_value(b"hello world");
builder.append_value(b"foo bar baz");
builder.append_value(b"testing fsst compression");
Expand All @@ -72,7 +73,7 @@ mod tests {
builder.append_value(b"qrstuvwxyz");
builder.append_value(b"0123456789");
builder.append_value(b"final string");
let input = builder.finish(DType::Utf8(Nullability::NonNullable));
let input = builder.finish_into_varbin();

let mut ctx = SESSION.create_execution_ctx();
let arr = input.into_array();
Expand Down Expand Up @@ -131,7 +132,8 @@ mod tests {
// Test case with special characters and nulls
// Values: ["", "", "", "", "", "", "", "", "", "", "", ",", "A<<<<<<<", "", "", "", "", null, null, null, null, null, null]
// Mask: only the last element is selected (true at index 22)
let mut builder = VarBinBuilder::<i32>::with_capacity(23);
let mut builder =
VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::Nullable), 23);
// 11 empty strings
for _ in 0..11 {
builder.append_value(b"");
Expand All @@ -146,9 +148,9 @@ mod tests {
}
// 6 nulls
for _ in 0..6 {
builder.append_null();
builder.push_null();
}
let input = builder.finish(DType::Utf8(Nullability::Nullable));
let input = builder.finish_into_varbin();
let array = input.clone().into_array();

let mut ctx = SESSION.create_execution_ctx();
Expand All @@ -172,12 +174,13 @@ mod tests {

#[test]
fn filter_only_null() -> VortexResult<()> {
let mut builder = VarBinBuilder::<i32>::with_capacity(3);
builder.append_null();
let mut builder =
VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::Nullable), 3);
builder.push_null();
builder.append_value(b"A");
builder.append_null();
builder.push_null();

let input = builder.finish(DType::Utf8(Nullability::Nullable));
let input = builder.finish_into_varbin();
let array = input.clone().into_array();

let mut ctx = SESSION.create_execution_ctx();
Expand Down Expand Up @@ -213,15 +216,14 @@ mod tests {

#[test]
fn test_fsst_byte_length() -> VortexResult<()> {
let mut builder = VarBinBuilder::<i32>::with_capacity(3);
let mut builder =
VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 3);
builder.append_value(b"hello");
builder.append_value(b"world!!");
builder.append_value("Пуховички"); // 9 characters, 18 bytes
builder.append_value(b"");

let varbin = builder
.finish(DType::Utf8(Nullability::NonNullable))
.into_array();
let varbin = builder.finish_into_varbin().into_array();
let mut ctx = SESSION.create_execution_ctx();
let compressor = fsst_train_compressor(&varbin, &mut ctx)?;
let fsst = fsst_compress(&varbin, &compressor, &mut ctx)?.into_array();
Expand Down
13 changes: 5 additions & 8 deletions encodings/fsst/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,14 @@ static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {

/// this function is VERY slow on miri, so we only want to run it once
pub(crate) fn build_fsst_array(ctx: &mut ExecutionCtx) -> ArrayRef {
let mut input_array = VarBinBuilder::<i32>::with_capacity(3);
let mut input_array =
VarBinBuilder::<i32>::with_capacity(DType::Utf8(Nullability::NonNullable), 3);
input_array.append_value(b"The Greeks never said that the limit could not be overstepped");
input_array.append_value(
b"They said it existed and that whoever dared to exceed it was mercilessly struck down",
);
input_array.append_value(b"Nothing in present history can contradict them");
let input_array = input_array
.finish(DType::Utf8(Nullability::NonNullable))
.into_array();
let input_array = input_array.finish_into_varbin().into_array();

let compressor = fsst_train_compressor(&input_array, ctx).unwrap();
fsst_compress(&input_array, &compressor, ctx)
Expand Down Expand Up @@ -162,13 +161,11 @@ fn fsst_compress_offsets_overflow_i32() {

println!("building large VarBinArray");
let string = vec![b'a'; STRING_LEN];
let mut builder = VarBinBuilder::<i64>::with_capacity(N);
let mut builder = VarBinBuilder::<i64>::with_capacity(DType::Utf8(Nullability::NonNullable), N);
for _ in 0..N {
builder.append_value(&string);
}
let array = builder
.finish(DType::Utf8(Nullability::NonNullable))
.into_array();
let array = builder.finish_into_varbin().into_array();

let compressor = CompressorBuilder::default().build();
let len = array.len();
Expand Down
51 changes: 33 additions & 18 deletions encodings/onpair/src/array.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,14 @@ use vortex_array::IntoArray;
use vortex_array::array_slots;
use vortex_array::buffer::BufferHandle;
use vortex_array::builders::ArrayBuilder;
use vortex_array::builders::DynVarBinBuilder;
use vortex_array::builders::VarBinBuilder;
use vortex_array::builders::VarBinViewBuilder;
use vortex_array::dtype::DType;
use vortex_array::dtype::Nullability;
use vortex_array::dtype::OffsetBuilderPType;
use vortex_array::dtype::PType;
use vortex_array::match_each_integer_ptype;
use vortex_array::match_each_varbin_builder;
use vortex_array::serde::ArrayChildren;
use vortex_array::validity::Validity;
use vortex_array::vtable::VTable;
Expand Down Expand Up @@ -553,23 +555,10 @@ impl VTable for OnPair {
builder: &mut dyn ArrayBuilder,
ctx: &mut ExecutionCtx,
) -> VortexResult<()> {
if let Some(builder) = builder.as_any_mut().downcast_mut::<DynVarBinBuilder>() {
let (bytes, lengths) = onpair_decode_bytes(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,
)
})?;
return Ok(());
if let Some(result) =
match_each_varbin_builder!(builder, |builder| append_to_varbin(array, builder, ctx))
{
return result;
}

let Some(builder) = builder.as_any_mut().downcast_mut::<VarBinViewBuilder>() else {
Expand Down Expand Up @@ -603,6 +592,32 @@ impl VTable for OnPair {
}
}

/// Decodes the values and appends them to `builder`.
fn append_to_varbin<O: OffsetBuilderPType>(
array: ArrayView<'_, OnPair>,
builder: &mut VarBinBuilder<O>,
ctx: &mut ExecutionCtx,
) -> VortexResult<()>
where
usize: AsPrimitive<O>,
{
let (bytes, lengths) = onpair_decode_bytes(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,
)
})
}

impl ValidityVTable<OnPair> for OnPair {
fn validity(array: ArrayView<'_, OnPair>) -> VortexResult<Validity> {
Ok(child_to_validity(
Expand Down
4 changes: 2 additions & 2 deletions encodings/onpair/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use vortex_array::arrays::VarBinViewArray;
use vortex_array::arrays::filter::FilterKernel;
use vortex_array::assert_arrays_eq;
use vortex_array::buffer::BufferHandle;
use vortex_array::builders::DynVarBinBuilder;
use vortex_array::builders::VarBinBuilder;
use vortex_array::builtins::ArrayBuiltins;
use vortex_array::dtype::DType;
use vortex_array::dtype::Nullability;
Expand Down Expand Up @@ -66,7 +66,7 @@ fn test_direct_offset_builder() -> vortex_error::VortexResult<()> {
let mut ctx = SESSION.create_execution_ctx();
let input = sample_input();
let encoded = compress_onpair(input.as_ref(), &mut ctx)?;
let mut builder = DynVarBinBuilder::with_capacity(input.dtype().clone(), false, input.len());
let mut builder = VarBinBuilder::<i32>::with_capacity(input.dtype().clone(), input.len());
encoded
.into_array()
.append_to_builder(&mut builder, &mut ctx)?;
Expand Down
Loading
Loading