diff --git a/Cargo.lock b/Cargo.lock index e79211b5436..11a58bb49f6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9783,6 +9783,7 @@ dependencies = [ "reqwest 0.13.4", "serde", "serde_json", + "sha2 0.11.0", "spatialbench", "spatialbench-arrow", "sysinfo", diff --git a/vortex-array/benches/take_fsl.rs b/vortex-array/benches/take_fsl.rs index 5dc491c28c5..cfd434b16c4 100644 --- a/vortex-array/benches/take_fsl.rs +++ b/vortex-array/benches/take_fsl.rs @@ -6,6 +6,7 @@ //! Parameterized over: //! - Number of indices to take //! - Fixed size list length (elements per list) +//! - Element byte width #![expect(clippy::cast_possible_truncation)] #![expect(clippy::unwrap_used)] @@ -13,16 +14,28 @@ use std::sync::LazyLock; use divan::Bencher; +use divan::counter::BytesCount; +use num_traits::FromPrimitive; use rand::RngExt; use rand::SeedableRng; use rand::rngs::StdRng; +use vortex_array::ExecutionCtx; use vortex_array::IntoArray; use vortex_array::RecursiveCanonical; use vortex_array::VortexSessionExecute; use vortex_array::array_session; +use vortex_array::arrays::ConstantArray; use vortex_array::arrays::FixedSizeListArray; +use vortex_array::arrays::PiecewiseSequenceArray; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::arrays::fixed_size_list::FixedSizeListArrayExt; +use vortex_array::dtype::IntegerPType; +use vortex_array::dtype::NativePType; +use vortex_array::dtype::half::f16; +use vortex_array::match_smallest_offset_type; use vortex_array::validity::Validity; use vortex_buffer::Buffer; +use vortex_buffer::BufferMut; use vortex_session::VortexSession; fn main() { @@ -39,18 +52,45 @@ const NUM_LISTS: usize = 500; const NUM_INDICES: &[usize] = &[100, 1_000]; /// Fixed size list lengths (elements per list). -const LIST_SIZES: &[usize] = &[16, 64, 256, 1024, 4096]; +const LIST_SIZES: &[usize] = &[16, 64, 128, 256, 512, 1024, 2048, 4096]; + +/// F16 list lengths for isolating the per-index, piecewise, and manual range-copy strategies. +const F16_STRATEGY_LIST_SIZES: &[usize] = &[1, 2, 4, 8, 16, 64, 128, 256, 512, 1024, 2048, 4096]; + +/// F16 strategy benchmarks keep a smaller take width so the forced slow strategies stay cheap. +const F16_STRATEGY_NUM_INDICES: &[usize] = &[10]; /// Creates a FixedSizeListArray with the given list size and number of lists. -fn create_fsl(list_size: usize, num_lists: usize) -> FixedSizeListArray { +fn create_fsl(list_size: usize, num_lists: usize) -> FixedSizeListArray +where + T: NativePType + FromPrimitive, +{ + create_fsl_with_validity::(list_size, num_lists, Validity::NonNullable) +} + +fn create_fsl_with_validity( + list_size: usize, + num_lists: usize, + validity: Validity, +) -> FixedSizeListArray +where + T: NativePType + FromPrimitive, +{ + let total_elements = list_size * num_lists; + let elements: Buffer = (0..total_elements) + .map(|idx| T::from_u16((idx % 251) as u16).unwrap()) + .collect(); + FixedSizeListArray::new(elements.into_array(), list_size as u32, validity, num_lists) +} + +fn create_i64_fsl_with_validity( + list_size: usize, + num_lists: usize, + validity: Validity, +) -> FixedSizeListArray { let total_elements = list_size * num_lists; let elements: Buffer = (0..total_elements as i64).collect(); - FixedSizeListArray::new( - elements.into_array(), - list_size as u32, - Validity::NonNullable, - num_lists, - ) + FixedSizeListArray::new(elements.into_array(), list_size as u32, validity, num_lists) } /// Creates random indices for taking from the array. @@ -63,11 +103,50 @@ fn create_random_indices(num_indices: usize, max_index: usize) -> Buffer { #[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] fn take_fsl_random(bencher: Bencher, num_indices: usize) { - let fsl = create_fsl(LIST_SIZE, NUM_LISTS); + let fsl = create_i64_fsl_with_validity(LIST_SIZE, NUM_LISTS, Validity::NonNullable); + bench_take_fsl_random::(bencher, num_indices, fsl); +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_f16_random(bencher: Bencher, num_indices: usize) { + take_fsl_random_typed::(bencher, num_indices); +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_u8_random(bencher: Bencher, num_indices: usize) { + take_fsl_random_typed::(bencher, num_indices); +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_u32_random(bencher: Bencher, num_indices: usize) { + take_fsl_random_typed::(bencher, num_indices); +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_u64_random(bencher: Bencher, num_indices: usize) { + take_fsl_random_typed::(bencher, num_indices); +} + +fn take_fsl_random_typed(bencher: Bencher, num_indices: usize) +where + T: NativePType + FromPrimitive, +{ + let fsl = create_fsl::(LIST_SIZE, NUM_LISTS); + bench_take_fsl_random::(bencher, num_indices, fsl); +} + +fn bench_take_fsl_random( + bencher: Bencher, + num_indices: usize, + fsl: FixedSizeListArray, +) where + T: NativePType, +{ let indices = create_random_indices(num_indices, NUM_LISTS); let indices_array = indices.into_array(); bencher + .counter(BytesCount::of_many::(num_indices * LIST_SIZE)) .with_inputs(|| (&fsl, &indices_array, SESSION.create_execution_ctx())) .bench_refs(|(array, indices, execution_ctx)| { array @@ -79,28 +158,185 @@ fn take_fsl_random(bencher: Bencher, num_indices: usize) }); } -#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] -fn take_fsl_nullable_random(bencher: Bencher, num_indices: usize) { - let total_elements = LIST_SIZE * NUM_LISTS; - let elements: Buffer = (0..total_elements as i64).collect(); - - // Create validity with ~10% nulls - let mut rng = StdRng::seed_from_u64(123); - let validity = Validity::from_iter((0..NUM_LISTS).map(|_| rng.random_ratio(9, 10))); +#[divan::bench(args = F16_STRATEGY_NUM_INDICES, consts = F16_STRATEGY_LIST_SIZES)] +fn take_fsl_f16_force_per_index(bencher: Bencher, num_indices: usize) { + let fsl = create_fsl::(LIST_SIZE, NUM_LISTS); + let indices = create_random_indices(num_indices, NUM_LISTS); - let fsl = FixedSizeListArray::new(elements.into_array(), LIST_SIZE as u32, validity, NUM_LISTS); + bencher + .counter(BytesCount::of_many::(num_indices * LIST_SIZE)) + .with_inputs(|| (&fsl, &indices, SESSION.create_execution_ctx())) + .bench_refs(|(array, indices, execution_ctx)| { + match_smallest_offset_type!(array.elements().len(), |E| { + take_fsl_f16_per_index_strategy::(array, indices) + }) + .into_array() + .execute::(execution_ctx) + .unwrap() + }); +} +#[divan::bench(args = F16_STRATEGY_NUM_INDICES, consts = F16_STRATEGY_LIST_SIZES)] +fn take_fsl_f16_force_piecewise_sequence( + bencher: Bencher, + num_indices: usize, +) { + let fsl = create_fsl::(LIST_SIZE, NUM_LISTS); let indices = create_random_indices(num_indices, NUM_LISTS); - let indices_array = indices.into_array(); bencher - .with_inputs(|| (&fsl, &indices_array, SESSION.create_execution_ctx())) + .counter(BytesCount::of_many::(num_indices * LIST_SIZE)) + .with_inputs(|| (&fsl, &indices, SESSION.create_execution_ctx())) .bench_refs(|(array, indices, execution_ctx)| { - array - .clone() - .take(indices.clone()) + take_fsl_f16_piecewise_sequence_strategy::(array, indices) + .into_array() + .execute::(execution_ctx) .unwrap() + }); +} + +#[divan::bench(args = F16_STRATEGY_NUM_INDICES, consts = F16_STRATEGY_LIST_SIZES)] +fn take_fsl_f16_force_manual_range_copy( + bencher: Bencher, + num_indices: usize, +) { + let fsl = create_fsl::(LIST_SIZE, NUM_LISTS); + let indices = create_random_indices(num_indices, NUM_LISTS); + + bencher + .counter(BytesCount::of_many::(num_indices * LIST_SIZE)) + .with_inputs(|| (&fsl, &indices, SESSION.create_execution_ctx())) + .bench_refs(|(array, indices, execution_ctx)| { + take_fsl_f16_manual_range_copy_strategy::(array, indices, execution_ctx) + .into_array() .execute::(execution_ctx) .unwrap() }); } + +fn take_fsl_f16_per_index_strategy( + array: &FixedSizeListArray, + indices: &Buffer, +) -> FixedSizeListArray { + let mut element_indices = BufferMut::::with_capacity(indices.len() * LIST_SIZE); + for &idx in indices.as_ref() { + let start = idx as usize * LIST_SIZE; + let end = start + LIST_SIZE; + for element_idx in start..end { + // SAFETY: capacity is exactly `indices.len() * LIST_SIZE`, and this loop appends + // exactly `LIST_SIZE` element indices for each input index. + unsafe { element_indices.push_unchecked(E::from_usize(element_idx).unwrap()) }; + } + } + + let element_indices = + PrimitiveArray::new(element_indices.freeze(), Validity::NonNullable).into_array(); + let elements = array.elements().take(element_indices).unwrap(); + + // SAFETY: `elements` was built by taking exactly `LIST_SIZE` elements per input index, so its + // length is `indices.len() * LIST_SIZE`; the output is non-nullable by construction. + unsafe { + FixedSizeListArray::new_unchecked( + elements, + LIST_SIZE as u32, + Validity::NonNullable, + indices.len(), + ) + } +} + +fn take_fsl_f16_piecewise_sequence_strategy( + array: &FixedSizeListArray, + indices: &Buffer, +) -> FixedSizeListArray { + let starts = indices + .as_ref() + .iter() + .map(|&idx| idx * LIST_SIZE as u64) + .collect::>(); + let run_count = starts.len(); + let starts = PrimitiveArray::from_iter(starts).into_array(); + let lengths = ConstantArray::new(LIST_SIZE as u64, run_count).into_array(); + let multipliers = ConstantArray::new(1u64, run_count).into_array(); + + // SAFETY: benchmark indices are generated in-bounds; lengths and multiplier 1 are + // non-nullable unsigned constants; output length is exactly `indices.len() * LIST_SIZE`. + let element_indices = unsafe { + PiecewiseSequenceArray::new_unchecked( + starts, + lengths, + multipliers, + indices.len() * LIST_SIZE, + ) + } + .into_array(); + let elements = array.elements().take(element_indices).unwrap(); + + // SAFETY: each generated run has width `LIST_SIZE`, and there is one run per input index, + // so `elements.len() == indices.len() * LIST_SIZE`. + unsafe { + FixedSizeListArray::new_unchecked( + elements, + LIST_SIZE as u32, + Validity::NonNullable, + indices.len(), + ) + } +} + +fn take_fsl_f16_manual_range_copy_strategy( + array: &FixedSizeListArray, + indices: &Buffer, + execution_ctx: &mut ExecutionCtx, +) -> FixedSizeListArray { + let elements = array + .elements() + .clone() + .execute::(execution_ctx) + .unwrap(); + let source = elements.as_slice::(); + let mut values = BufferMut::::with_capacity(indices.len() * LIST_SIZE); + + for &idx in indices.as_ref() { + let start = idx as usize * LIST_SIZE; + values.extend_from_slice(&source[start..start + LIST_SIZE]); + } + + // SAFETY: the buffer was filled with exactly `LIST_SIZE` copied values per input index, so it + // has the element length required by the constructed FSL. + unsafe { + FixedSizeListArray::new_unchecked( + PrimitiveArray::new(values.freeze(), Validity::NonNullable).into_array(), + LIST_SIZE as u32, + Validity::NonNullable, + indices.len(), + ) + } +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_nullable_random(bencher: Bencher, num_indices: usize) { + // Create validity with ~10% nulls + let mut rng = StdRng::seed_from_u64(123); + let validity = Validity::from_iter((0..NUM_LISTS).map(|_| rng.random_ratio(9, 10))); + + let fsl = create_i64_fsl_with_validity(LIST_SIZE, NUM_LISTS, validity); + bench_take_fsl_random::(bencher, num_indices, fsl); +} + +#[divan::bench(args = NUM_INDICES, consts = LIST_SIZES)] +fn take_fsl_f16_nullable_random(bencher: Bencher, num_indices: usize) { + take_fsl_nullable_random_typed::(bencher, num_indices); +} + +fn take_fsl_nullable_random_typed(bencher: Bencher, num_indices: usize) +where + T: NativePType + FromPrimitive, +{ + // Create validity with ~10% nulls + let mut rng = StdRng::seed_from_u64(123); + let validity = Validity::from_iter((0..NUM_LISTS).map(|_| rng.random_ratio(9, 10))); + + let fsl = create_fsl_with_validity::(LIST_SIZE, NUM_LISTS, validity); + bench_take_fsl_random::(bencher, num_indices, fsl); +} diff --git a/vortex-array/src/arrays/bool/compute/take.rs b/vortex-array/src/arrays/bool/compute/take.rs index 8469c7e25c2..1297a0a18e7 100644 --- a/vortex-array/src/arrays/bool/compute/take.rs +++ b/vortex-array/src/arrays/bool/compute/take.rs @@ -4,9 +4,12 @@ use itertools::Itertools as _; use num_traits::AsPrimitive; use vortex_buffer::BitBuffer; +use vortex_buffer::BitBufferMut; use vortex_buffer::BitBufferView; use vortex_buffer::get_bit; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use vortex_mask::Mask; use crate::ArrayRef; @@ -15,12 +18,17 @@ use crate::array::ArrayView; use crate::arrays::Bool; use crate::arrays::BoolArray; use crate::arrays::ConstantArray; +use crate::arrays::PiecewiseSequence; use crate::arrays::PrimitiveArray; use crate::arrays::bool::BoolArrayExt; use crate::arrays::dict::TakeExecute; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::builtins::ArrayBuiltins; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_integer_ptype; +use crate::match_each_unsigned_integer_ptype; use crate::scalar::Scalar; impl TakeExecute for Bool { @@ -29,6 +37,12 @@ impl TakeExecute for Bool { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_contiguous_ranges(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + let indices_nulls_zeroed = match indices.validity()?.execute_mask(indices.len(), ctx)? { Mask::AllTrue(_) => indices.clone(), Mask::AllFalse(_) => { @@ -55,6 +69,47 @@ impl TakeExecute for Bool { } } +fn take_contiguous_ranges( + array: ArrayView<'_, Bool>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let source = array.to_bit_buffer(); + let output_len = indices_ref.len(); + let buffer = match &lengths { + ConstantOrArray::Constant(length) => { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + take_bit_slices_constant_length( + &source, + starts.as_slice::(), + *length, + output_len, + )? + }) + } + ConstantOrArray::Array(lengths) => { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + take_bit_slices( + &source, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + )? + }) + }) + } + }; + + Ok(Some( + BoolArray::new(buffer, array.validity()?.take(indices_ref)?).into_array(), + )) +} + fn take_valid_indices>(bools: BitBufferView<'_>, indices: &[I]) -> BitBuffer { // For boolean arrays that roughly fit into a single page (at least, on Linux), it's worth // the overhead to convert to a Vec. @@ -82,6 +137,58 @@ fn take_bool_impl>(bools: BitBufferView<'_>, indices: &[I] }) } +fn take_bit_slices_constant_length( + source: &BitBuffer, + starts: &[S], + length: usize, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + + let mut values = BitBufferMut::with_capacity(output_len); + for &start in starts { + let start = start.as_(); + values.append_buffer(&source.slice(start..).slice(..length)); + } + + Ok(values.freeze()) +} + +fn take_bit_slices( + source: &BitBuffer, + starts: &[S], + lengths: &[L], + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, +{ + let mut values = BitBufferMut::with_capacity(output_len); + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + values.append_buffer(&source.slice(start..).slice(..length)); + } + + vortex_ensure!( + values.len() == output_len, + "PiecewiseSequenceArray expanded length {} does not match declared length {output_len}", + values.len() + ); + Ok(values.freeze()) +} + #[cfg(test)] mod test { use rstest::rstest; diff --git a/vortex-array/src/arrays/decimal/compute/take.rs b/vortex-array/src/arrays/decimal/compute/take.rs index ce786f8e4ab..1b0a6aff7d6 100644 --- a/vortex-array/src/arrays/decimal/compute/take.rs +++ b/vortex-array/src/arrays/decimal/compute/take.rs @@ -1,21 +1,32 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +use itertools::Itertools as _; use vortex_buffer::Buffer; +use vortex_buffer::BufferMut; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; use crate::arrays::Decimal; use crate::arrays::DecimalArray; +use crate::arrays::PiecewiseSequence; use crate::arrays::PrimitiveArray; use crate::arrays::dict::TakeExecute; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::copy_slice_to_spare; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::dtype::IntegerPType; use crate::dtype::NativeDecimalType; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_decimal_value_type; use crate::match_each_integer_ptype; +use crate::match_each_unsigned_integer_ptype; +use crate::validity::Validity; impl TakeExecute for Decimal { fn take( @@ -23,6 +34,12 @@ impl TakeExecute for Decimal { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_contiguous_ranges(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + let indices = indices.clone().execute::(ctx)?; let validity = array.validity()?.take(&indices.clone().into_array())?; @@ -42,10 +59,190 @@ impl TakeExecute for Decimal { } } +fn take_contiguous_ranges( + array: ArrayView<'_, Decimal>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let validity = array.validity()?.take(indices_ref)?; + let output_len = indices_ref.len(); + let taken = match lengths { + ConstantOrArray::Constant(length) => { + take_slices_constant_length(array, &starts, length, validity, output_len)? + } + ConstantOrArray::Array(lengths) => { + take_slices(array, &starts, &lengths, validity, output_len)? + } + }; + Ok(Some(taken)) +} + +fn take_slices_constant_length( + array: ArrayView<'_, Decimal>, + starts: &PrimitiveArray, + length: usize, + validity: Validity, + output_len: usize, +) -> VortexResult { + match_each_decimal_value_type!(array.values_type(), |D| { + take_slices_constant_length_typed::(array, starts, length, validity, output_len) + }) +} + +fn take_slices_constant_length_typed( + array: ArrayView<'_, Decimal>, + starts: &PrimitiveArray, + length: usize, + validity: Validity, + output_len: usize, +) -> VortexResult +where + D: NativeDecimalType, +{ + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + let values = take_slices_constant_length_to_buffer::( + starts.as_slice::(), + length, + array.buffer::().as_slice(), + output_len, + )?; + + // SAFETY: contiguous gather preserves the decimal dtype and value representation. + Ok( + unsafe { DecimalArray::new_unchecked(values, array.decimal_dtype(), validity) } + .into_array(), + ) + }) +} + +fn take_slices( + array: ArrayView<'_, Decimal>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult { + match_each_decimal_value_type!(array.values_type(), |D| { + take_slices_typed::(array, starts, lengths, validity, output_len) + }) +} + +fn take_slices_typed( + array: ArrayView<'_, Decimal>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult +where + D: NativeDecimalType, +{ + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + take_slices_start_typed::(array, starts, lengths, validity, output_len) + }) +} + +fn take_slices_start_typed( + array: ArrayView<'_, Decimal>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult +where + D: NativeDecimalType, + S: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + let values = take_slices_to_buffer::( + starts.as_slice::(), + lengths.as_slice::(), + array.buffer::().as_slice(), + output_len, + )?; + + // SAFETY: contiguous gather preserves the decimal dtype and value representation. + Ok( + unsafe { DecimalArray::new_unchecked(values, array.decimal_dtype(), validity) } + .into_array(), + ) + }) +} + fn take_to_buffer(indices: &[I], values: &[T]) -> Buffer { indices.iter().map(|idx| values[idx.as_()]).collect() } +fn take_slices_constant_length_to_buffer( + starts: &[S], + length: usize, + values: &[T], + output_len: usize, +) -> VortexResult> +where + S: UnsignedPType, + T: NativeDecimalType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + + let mut result = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for &start in starts { + let start = start.as_(); + cursor = copy_slice_to_spare(&mut result, cursor, &values[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {cursor} does not match declared length {output_len}" + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { result.set_len(output_len) }; + Ok(result.freeze()) +} + +fn take_slices_to_buffer( + starts: &[S], + lengths: &[L], + values: &[T], + output_len: usize, +) -> VortexResult> +where + S: UnsignedPType, + L: UnsignedPType, + T: NativeDecimalType, +{ + let mut result = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + cursor = copy_slice_to_spare(&mut result, cursor, &values[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {} does not match declared length {output_len}", + cursor + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { result.set_len(output_len) }; + Ok(result.freeze()) +} + #[cfg(test)] mod tests { use rstest::rstest; diff --git a/vortex-array/src/arrays/extension/compute/rules.rs b/vortex-array/src/arrays/extension/compute/rules.rs index 2d02d1ae7a7..0699f2f68f4 100644 --- a/vortex-array/src/arrays/extension/compute/rules.rs +++ b/vortex-array/src/arrays/extension/compute/rules.rs @@ -11,6 +11,7 @@ use crate::arrays::ConstantArray; use crate::arrays::Extension; use crate::arrays::ExtensionArray; use crate::arrays::Filter; +use crate::arrays::dict::TakeReduceAdaptor; use crate::arrays::extension::ExtensionArrayExt; use crate::arrays::filter::FilterReduceAdaptor; use crate::arrays::slice::SliceReduceAdaptor; @@ -50,6 +51,7 @@ pub(crate) const PARENT_RULES: ParentRuleSet = ParentRuleSet::new(&[ ParentRuleSet::lift(&FilterReduceAdaptor(Extension)), ParentRuleSet::lift(&MaskReduceAdaptor(Extension)), ParentRuleSet::lift(&SliceReduceAdaptor(Extension)), + ParentRuleSet::lift(&TakeReduceAdaptor(Extension)), ]); /// Push filter operations into the storage array of an extension array. diff --git a/vortex-array/src/arrays/extension/compute/take.rs b/vortex-array/src/arrays/extension/compute/take.rs index 8fb0da11222..d9ac1b4f600 100644 --- a/vortex-array/src/arrays/extension/compute/take.rs +++ b/vortex-array/src/arrays/extension/compute/take.rs @@ -10,8 +10,24 @@ use crate::array::ArrayView; use crate::arrays::Extension; use crate::arrays::ExtensionArray; use crate::arrays::dict::TakeExecute; +use crate::arrays::dict::TakeReduce; use crate::arrays::extension::ExtensionArrayExt; +impl TakeReduce for Extension { + fn take(array: ArrayView<'_, Extension>, indices: &ArrayRef) -> VortexResult> { + let taken_storage = array.storage_array().take(indices.clone())?; + Ok(Some( + ExtensionArray::new( + array + .ext_dtype() + .with_nullability(taken_storage.dtype().nullability()), + taken_storage, + ) + .into_array(), + )) + } +} + impl TakeExecute for Extension { fn take( array: ArrayView<'_, Extension>, diff --git a/vortex-array/src/arrays/fixed_size_list/compute/take.rs b/vortex-array/src/arrays/fixed_size_list/compute/take.rs index 193730f62c4..5f2e0a50835 100644 --- a/vortex-array/src/arrays/fixed_size_list/compute/take.rs +++ b/vortex-array/src/arrays/fixed_size_list/compute/take.rs @@ -1,221 +1,251 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors -use vortex_buffer::BitBufferMut; +use itertools::Itertools; +use vortex_buffer::Buffer; use vortex_buffer::BufferMut; -use vortex_error::VortexExpect; use vortex_error::VortexResult; -use vortex_error::vortex_panic; -use vortex_mask::Mask; +use vortex_error::vortex_bail; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; -use crate::arrays::BoolArray; +use crate::arrays::ConstantArray; use crate::arrays::FixedSizeList; use crate::arrays::FixedSizeListArray; +use crate::arrays::PiecewiseSequenceArray; use crate::arrays::Primitive; use crate::arrays::PrimitiveArray; -use crate::arrays::bool::BoolArrayExt; use crate::arrays::dict::TakeExecute; use crate::arrays::fixed_size_list::FixedSizeListArrayExt; use crate::arrays::primitive::PrimitiveArrayExt; +use crate::builders::builder_with_capacity; +use crate::dtype::DType; use crate::dtype::IntegerPType; use crate::executor::ExecutionCtx; -use crate::match_each_unsigned_integer_ptype; -use crate::match_smallest_offset_type; +use crate::match_each_integer_ptype; +use crate::optimizer::ArrayOptimizer; use crate::validity::Validity; /// Take implementation for [`FixedSizeListArray`]. /// -/// Unlike `ListView`, `FixedSizeListArray` must rebuild the elements array because it requires -/// that elements start at offset 0 and be perfectly packed without gaps. We expand list indices -/// into element indices and push them down to the child elements array. +/// `FixedSizeListArray` must rebuild its elements array because selected lists need to become +/// packed from offset 0. The FSL layer translates selected list rows into ordered element runs and +/// delegates the execution strategy to the elements child via `PiecewiseSequenceArray` indices. impl TakeExecute for FixedSizeList { fn take( array: ArrayView<'_, FixedSizeList>, indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { - let max_element_idx = array.elements().len(); - // Indices are non-negative; dispatch over the 4 unsigned widths (the executed array is - // reinterpreted to unsigned in `take_with_indices`). `E` is already unsigned. - match_each_unsigned_integer_ptype!(indices.dtype().as_ptype().to_unsigned(), |I| { - match_smallest_offset_type!(max_element_idx, |E| { - take_with_indices::(array, indices, ctx) - }) - }) - .map(Some) + if array.is_empty() { + return take_empty_fsl(array, indices, ctx).map(Some); + } + + take_non_empty_fsl(array, indices, ctx).map(Some) } } -/// Dispatches to the appropriate take implementation based on list size and nullability. -fn take_with_indices( +fn take_empty_fsl( array: ArrayView<'_, FixedSizeList>, indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult { - let list_size = array.list_size() as usize; + debug_assert!(array.is_empty()); - let indices_array = indices.clone().execute::(ctx)?; - // Reinterpret to unsigned so `as_slice::` (with unsigned `I`) matches; values are unchanged. - let indices_array = indices_array.reinterpret_cast(indices_array.ptype().to_unsigned()); - - // Make sure to handle degenerate case where lists have size 0 (these can take fast paths). - if list_size == 0 { - debug_assert!( - array.elements().is_empty(), - "degenerate list must have empty elements" + let new_len = indices.len(); + if new_len != 0 { + let indices_validity = indices.validity()?.execute_mask(new_len, ctx)?; + vortex_ensure!( + indices_validity.all_false(), + "cannot take valid indices from an empty FixedSizeList" ); + } - // Since there are no elements to take, we just need to take on the validity map. - let new_validity = array.validity()?.take(indices)?; - let new_len = indices_array.len(); - - Ok( - // SAFETY: list_size is 0, elements array is empty, and validity has the correct length. - unsafe { - FixedSizeListArray::new_unchecked( - array.elements().clone(), // Remember that this is an empty array. - array.list_size(), - new_validity, - new_len, - ) - } - .into_array(), + let list_size = array.list_size() as usize; + let elements_len = new_len.checked_mul(list_size).ok_or_else(|| { + vortex_err!( + "FixedSizeList take output length overflow: {new_len} lists of size {list_size}" ) + })?; + let new_elements = default_elements(array, elements_len); + let new_validity = if new_len == 0 { + array.validity()?.take(indices)? } else { - // The result's nullability is the union of the input nullabilities. - if array.dtype().is_nullable() || indices_array.dtype().is_nullable() { - let indices_array = indices_array.as_view(); - take_nullable_fsl::(array, indices_array, ctx) - } else { - let indices_array = indices_array.as_view(); - take_non_nullable_fsl::(array, indices_array) - } + Validity::AllInvalid + }; + + // SAFETY: empty output needs no child values; otherwise the index validity mask proves every + // output row is null. Placeholder child elements have the exact length required by FSL. + unsafe { + FixedSizeListArray::new_unchecked(new_elements, array.list_size(), new_validity, new_len) } + .into_array() + .optimize_ctx(ctx.session()) } -/// Takes from an array when both the array and indices are non-nullable. -fn take_non_nullable_fsl( +fn take_non_empty_fsl( array: ArrayView<'_, FixedSizeList>, - indices_array: ArrayView<'_, Primitive>, + indices: &ArrayRef, + ctx: &mut ExecutionCtx, ) -> VortexResult { - let list_size = array.list_size() as usize; - let indices: &[I] = indices_array.as_slice::(); - let new_len = indices.len(); - - // Build the element indices directly without validity tracking. - let mut elements_indices = BufferMut::::with_capacity(new_len * list_size); + debug_assert!(!array.is_empty()); - // Build the element indices for each list. - for data_idx in indices { - let data_idx = data_idx - .to_usize() - .unwrap_or_else(|| vortex_panic!("Failed to convert index to usize: {}", data_idx)); - - let list_start = data_idx * list_size; - let list_end = (data_idx + 1) * list_size; + let DType::Primitive(ptype, _) = indices.dtype() else { + vortex_bail!("Invalid indices dtype: {}", indices.dtype()) + }; + if !ptype.is_int() { + vortex_bail!("Invalid indices dtype: {}", indices.dtype()); + } - // Expand the list into individual element indices. - for i in list_start..list_end { - // SAFETY: We've allocated enough space for enough indices for all `new_len` lists (that each consist of `list_size = list_end - list_start` elements), so we know we have enough capacity. - unsafe { - elements_indices.push_unchecked(E::from_usize(i).vortex_expect("i < list_end")) - }; - } + if array.list_size() == 0 { + return take_non_empty_degenerate_fsl(array, indices, ctx); } - let elements_indices = elements_indices.freeze(); - debug_assert_eq!(elements_indices.len(), new_len * list_size); + let indices_array = indices.clone().execute::(ctx)?; + match_each_integer_ptype!(indices_array.ptype(), |I| { + take_non_empty_non_degenerate_fsl::(array, indices, indices_array.as_view(), ctx) + }) +} - let elements_indices_array = PrimitiveArray::new(elements_indices, Validity::NonNullable); - let new_elements = array.elements().take(elements_indices_array.into_array())?; - debug_assert_eq!(new_elements.len(), new_len * list_size); +fn take_non_empty_degenerate_fsl( + array: ArrayView<'_, FixedSizeList>, + indices: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult { + debug_assert!(!array.is_empty()); + debug_assert_eq!(array.list_size(), 0); + vortex_ensure!( + array.elements().is_empty(), + "degenerate list must have empty elements" + ); - // Both inputs are non-nullable, so the result is non-nullable. - Ok(unsafe { + let indices_array = indices.clone().execute::(ctx)?; + match_each_integer_ptype!(indices_array.ptype(), |I| { + bounds_check_valid_indices::(&indices_array.as_view(), array.as_ref().len(), ctx) + })?; + let new_validity = array.validity()?.take(indices)?; + let new_len = indices_array.len(); + + // SAFETY: degenerate FSL inputs have no elements, valid index payloads were checked against + // the source length, and `Validity::take` produces validity for `new_len`. + unsafe { FixedSizeListArray::new_unchecked( - new_elements, + array.elements().clone(), array.list_size(), - Validity::NonNullable, + new_validity, new_len, ) } - .into_array()) + .into_array() + .optimize_ctx(ctx.session()) } -/// Takes from an array when either the array or indices are nullable. -fn take_nullable_fsl( +fn take_non_empty_non_degenerate_fsl( array: ArrayView<'_, FixedSizeList>, + indices: &ArrayRef, indices_array: ArrayView<'_, Primitive>, ctx: &mut ExecutionCtx, ) -> VortexResult { + debug_assert!(!array.is_empty()); + debug_assert_ne!(array.list_size(), 0); + + let (new_elements, new_len) = + take_non_empty_non_degenerate_elements::(array, indices_array, ctx)?; + let new_validity = array.validity()?.take(indices)?; + + // SAFETY: `new_elements` has `new_len * list_size` elements, and `Validity::take` produces + // validity for `new_len`. + unsafe { + FixedSizeListArray::new_unchecked(new_elements, array.list_size(), new_validity, new_len) + } + .into_array() + .optimize_ctx(ctx.session()) +} + +fn take_non_empty_non_degenerate_elements( + array: ArrayView<'_, FixedSizeList>, + indices_array: ArrayView<'_, Primitive>, + ctx: &mut ExecutionCtx, +) -> VortexResult<(ArrayRef, usize)> { + debug_assert!(!array.is_empty()); + debug_assert_ne!(array.list_size(), 0); + let list_size = array.list_size() as usize; + let array_len = array.as_ref().len(); let indices: &[I] = indices_array.as_slice::(); let new_len = indices.len(); - - let array_validity = array - .fixed_size_list_validity() - .execute_mask(array.as_ref().len(), ctx) - .vortex_expect("Failed to compute validity mask"); - let indices_len = indices_array.as_ref().len(); - let indices_validity = match indices_array - .validity() - .vortex_expect("Failed to compute validity mask") - { - Validity::NonNullable | Validity::AllValid => Mask::new_true(indices_len), - Validity::AllInvalid => Mask::new_false(indices_len), - Validity::Array(a) => a.execute::(ctx)?.execute_mask(ctx), - }; - - // We must use placeholder zeros for null lists to maintain the array length without - // propagating nullability to the element array's take operation. - let mut elements_indices = BufferMut::::with_capacity(new_len * list_size); - let mut new_validity_builder = BitBufferMut::with_capacity(new_len); - - // Build the element indices while tracking which lists are null. - for (data_idx, is_index_valid) in indices.iter().zip(indices_validity.iter()) { - let data_idx = data_idx - .to_usize() - .unwrap_or_else(|| vortex_panic!("Failed to convert index to usize: {}", data_idx)); - - // The list is null if the index is null or the indexed element is null. - if !is_index_valid || !array_validity.value(data_idx) { - // Append placeholder zeros for null lists. These will be masked by the validity array. - // We cannot use append_nulls here as explained above. - unsafe { elements_indices.push_n_unchecked(E::zero(), list_size) }; - new_validity_builder.append(false); - } else { - // Append the actual element indices for this list. - let list_start = data_idx * list_size; - let list_end = (data_idx + 1) * list_size; - - // Expand the list into individual element indices. - for i in list_start..list_end { - // SAFETY: We've allocated enough space for enough indices for all `new_len` lists (that each consist of `list_size = list_end - list_start` elements), so we know we have enough capacity. - unsafe { - elements_indices.push_unchecked(E::from_usize(i).vortex_expect("i < list_end")) - }; + let elements_len = new_len.checked_mul(list_size).ok_or_else(|| { + vortex_err!( + "FixedSizeList take output length overflow: {new_len} lists of size {list_size}" + ) + })?; + + let indices_validity = indices_array.validity()?.execute_mask(new_len, ctx)?; + let starts = indices + .iter() + .zip_eq(indices_validity.iter()) + .map(|(&data_idx, is_index_valid)| { + if !is_index_valid { + return Ok(0); } - new_validity_builder.append(true); - } - } + let data_idx: usize = data_idx.as_(); + if data_idx >= array_len { + vortex_bail!(OutOfBounds: data_idx, 0, array_len); + } + Ok((data_idx * list_size) as u64) + }) + .process_results(|iter| iter.collect::>())?; - let elements_indices = elements_indices.freeze(); - debug_assert_eq!(elements_indices.len(), new_len * list_size); + let new_elements = + take_element_runs(array.elements(), starts.freeze(), list_size, elements_len)?; - let elements_indices_array = PrimitiveArray::new(elements_indices, Validity::NonNullable); - let new_elements = array.elements().take(elements_indices_array.into_array())?; - debug_assert_eq!(new_elements.len(), new_len * list_size); + Ok((new_elements, new_len)) +} - // At least one input was nullable, so the result is nullable. - let new_validity = Validity::from(new_validity_builder.freeze()); - debug_assert!(new_validity.maybe_len().is_none_or(|vl| vl == new_len)); +fn bounds_check_valid_indices( + indices_array: &ArrayView<'_, Primitive>, + array_len: usize, + ctx: &mut ExecutionCtx, +) -> VortexResult<()> { + let indices: &[I] = indices_array.as_slice::(); + let indices_validity = indices_array.validity()?.execute_mask(indices.len(), ctx)?; - Ok(unsafe { - FixedSizeListArray::new_unchecked(new_elements, array.list_size(), new_validity, new_len) + for (&data_idx, is_index_valid) in indices.iter().zip_eq(indices_validity.iter()) { + if is_index_valid { + let data_idx = data_idx.as_(); + if data_idx >= array_len { + vortex_bail!(OutOfBounds: data_idx, 0, array_len); + } + } } - .into_array()) + Ok(()) +} + +fn default_elements(array: ArrayView<'_, FixedSizeList>, len: usize) -> ArrayRef { + let mut builder = builder_with_capacity(array.elements().dtype(), len); + builder.append_defaults(len); + builder.finish() +} + +fn take_element_runs( + elements: &ArrayRef, + starts: Buffer, + length: usize, + output_len: usize, +) -> VortexResult { + let run_count = starts.len(); + let starts = PrimitiveArray::new(starts, Validity::NonNullable).into_array(); + let lengths = ConstantArray::new(length as u64, run_count).into_array(); + let multipliers = ConstantArray::new(1u64, run_count).into_array(); + + // SAFETY: callers produced one start per output row after validating list indices against the + // source FSL length. `length` and multiplier 1 are represented as non-nullable unsigned + // constant arrays, and `output_len` was computed as `run_count * length`. + let indices = + unsafe { PiecewiseSequenceArray::new_unchecked(starts, lengths, multipliers, output_len) }; + elements.take(indices.into_array()) } diff --git a/vortex-array/src/arrays/fixed_size_list/tests/take.rs b/vortex-array/src/arrays/fixed_size_list/tests/take.rs index af8f94f2d5f..93262a74b37 100644 --- a/vortex-array/src/arrays/fixed_size_list/tests/take.rs +++ b/vortex-array/src/arrays/fixed_size_list/tests/take.rs @@ -147,6 +147,28 @@ fn test_take_fsl_with_null_indices_preserves_elements() { assert_arrays_eq!(expected, result, &mut ctx); } +#[test] +fn test_take_fsl_null_index_ignores_out_of_bounds_payload() { + let mut ctx = array_session().create_execution_ctx(); + let elements = buffer![1i32, 2, 3, 4, 5, 6].into_array(); + let fsl = FixedSizeListArray::new(elements.into_array(), 2, Validity::NonNullable, 3); + + let indices = PrimitiveArray::new( + buffer![1u32, 99, 2], + Validity::from_iter([true, false, true]), + ) + .into_array(); + let result = fsl.take(indices).unwrap(); + + let expected = FixedSizeListArray::new( + buffer![3i32, 4, 0, 0, 5, 6].into_array(), + 2, + Validity::from_iter([true, false, true]), + 3, + ); + assert_arrays_eq!(expected, result, &mut ctx); +} + // Element index overflow: with u8 indices and list_size=16, data_idx=16 produces element index // 16*16=256 which overflows u8. The take kernel must widen the element index type. #[rstest] diff --git a/vortex-array/src/arrays/list/compute/take.rs b/vortex-array/src/arrays/list/compute/take.rs index 14e89d4c238..56b6ce22511 100644 --- a/vortex-array/src/arrays/list/compute/take.rs +++ b/vortex-array/src/arrays/list/compute/take.rs @@ -1,26 +1,34 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors -use vortex_error::VortexExpect; +use itertools::Itertools as _; +use vortex_buffer::BufferMut; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; +use vortex_mask::Mask; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; +use crate::arrays::ConstantArray; use crate::arrays::List; use crate::arrays::ListArray; +use crate::arrays::PiecewiseSequence; +use crate::arrays::PiecewiseSequenceArray; use crate::arrays::Primitive; use crate::arrays::PrimitiveArray; use crate::arrays::dict::TakeExecute; use crate::arrays::list::ListArrayExt; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::arrays::primitive::PrimitiveArrayExt; -use crate::builders::ArrayBuilder; -use crate::builders::PrimitiveBuilder; use crate::dtype::IntegerPType; -use crate::dtype::Nullability; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_unsigned_integer_ptype; use crate::match_smallest_offset_type; +use crate::validity::Validity; // TODO(connor)[ListView]: Re-revert to the version where we simply convert to a `ListView` and call // the `ListView::take` compute function once `ListView` is more stable. @@ -37,21 +45,30 @@ impl TakeExecute for List { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_piecewise_sequence(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + + let new_validity = array.validity()?.take(indices)?; let indices = indices.clone().execute::(ctx)?; let indices = indices.reinterpret_cast(indices.ptype().to_unsigned()); let offsets = array.offsets().clone().execute::(ctx)?; let offsets = offsets.reinterpret_cast(offsets.ptype().to_unsigned()); + let validity_mask = new_validity.execute_mask(indices.len(), ctx)?; // This is an over-approximation of the total number of elements in the resulting array. let total_approx = array.elements().len().saturating_mul(indices.len()); match_each_unsigned_integer_ptype!(offsets.ptype(), |O| { match_each_unsigned_integer_ptype!(indices.ptype(), |I| { match_smallest_offset_type!(total_approx, |OutputOffsetType| { - _take::( + take_with_piecewise_elements::( array, offsets.as_view(), indices.as_view(), - ctx, + new_validity, + &validity_mask, ) .map(Some) }) @@ -60,148 +77,535 @@ impl TakeExecute for List { } } -fn _take( +fn take_with_piecewise_elements< + I: IntegerPType, + O: IntegerPType, + OutputOffsetType: IntegerPType, +>( array: ArrayView<'_, List>, offsets_array: ArrayView<'_, Primitive>, indices_array: ArrayView<'_, Primitive>, - ctx: &mut ExecutionCtx, + new_validity: Validity, + validity_mask: &Mask, ) -> VortexResult { - let data_validity = array - .list_validity() - .execute_mask(array.as_ref().len(), ctx)?; - let indices_validity = indices_array - .validity() - .vortex_expect("Failed to compute validity mask") - .execute_mask(indices_array.as_ref().len(), ctx)?; - - if !indices_validity.all_true() || !data_validity.all_true() { - return _take_nullable::(array, offsets_array, indices_array, ctx); - } - let offsets: &[O] = offsets_array.as_slice(); let indices: &[I] = indices_array.as_slice(); - let mut new_offsets = PrimitiveBuilder::::with_capacity( - Nullability::NonNullable, - indices.len(), - ); - let mut elements_to_take = - PrimitiveBuilder::with_capacity(Nullability::NonNullable, 2 * indices.len()); - - let mut current_offset = OutputOffsetType::zero(); - new_offsets.append_zero(); + let offsets_capacity = indices + .len() + .checked_add(1) + .ok_or_else(|| vortex_err!("List take offsets length overflow"))?; + let mut new_offsets = BufferMut::::with_capacity(offsets_capacity); + let mut element_starts = BufferMut::::with_capacity(indices.len()); + let mut element_lengths = BufferMut::::with_capacity(indices.len()); + + let mut current_offset = 0usize; + new_offsets.push(OutputOffsetType::zero()); + + for (&data_idx, is_valid) in indices.iter().zip_eq(validity_mask.iter()) { + if !is_valid { + new_offsets.push(new_offset_value::(current_offset)?); + element_starts.push(0); + element_lengths.push(0); + continue; + } - for &data_idx in indices { let data_idx: usize = data_idx.as_(); let start = offsets[data_idx]; let stop = offsets[data_idx + 1]; - - // Annoyingly, we can't turn (start..end) into a range, so we're doing that manually. - // - // We could convert start and end to usize, but that would impose a potentially - // harder constraint - now we don't care if they fit into usize as long as their - // difference does. - let additional: usize = (stop - start).as_(); - - // TODO(0ax1): optimize this - elements_to_take.reserve_exact(additional); - for i in 0..additional { - elements_to_take.append_value(start + O::from_usize(i).vortex_expect("i < additional")); - } - current_offset += - OutputOffsetType::from_usize((stop - start).as_()).vortex_expect("offset conversion"); - new_offsets.append_value(current_offset); + let start: usize = start.as_(); + let stop: usize = stop.as_(); + let length = stop + .checked_sub(start) + .ok_or_else(|| vortex_err!("List offsets are not monotonic at offset {stop}"))?; + + current_offset = current_offset + .checked_add(length) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + new_offsets.push(new_offset_value::(current_offset)?); + element_starts.push(start as u64); + element_lengths.push(length as u64); } - let elements_to_take = elements_to_take.finish(); - let new_offsets = new_offsets.finish(); - - let new_elements = array.elements().take(elements_to_take)?; + let new_offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable).into_array(); + let multipliers = ConstantArray::new(1u64, element_starts.len()).into_array(); + + // SAFETY: valid source rows contribute ranges derived from list offsets; null index/source + // rows contribute zero-length placeholder ranges. `current_offset` is the sum of all generated + // element lengths, and multiplier 1 preserves contiguous ranges. + let element_indices = unsafe { + PiecewiseSequenceArray::new_unchecked( + element_starts.into_array(), + element_lengths.into_array(), + multipliers, + current_offset, + ) + }; + let new_elements = array.elements().take(element_indices.into_array())?; - Ok(ListArray::try_new( - new_elements, - new_offsets, - array.validity()?.take(indices_array.array())?, - )? - .into_array()) + // SAFETY: offsets are rebuilt from the gathered element ranges and have one entry per output + // row plus the leading zero; validity is produced by the usual take-validity path. + Ok(unsafe { ListArray::new_unchecked(new_elements, new_offsets, new_validity) }.into_array()) } -// Kept out-of-line: as a single-callsite generic helper it would otherwise be inlined into every -// monomorphization of `_take`, duplicating the entire nullable path across all specializations. -#[inline(never)] -fn _take_nullable( +fn take_piecewise_sequence( array: ArrayView<'_, List>, - offsets_array: ArrayView<'_, Primitive>, - indices_array: ArrayView<'_, Primitive>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, ctx: &mut ExecutionCtx, -) -> VortexResult { - let offsets: &[O] = offsets_array.as_slice(); - let indices: &[I] = indices_array.as_slice(); +) -> VortexResult> { let data_validity = array .list_validity() .execute_mask(array.as_ref().len(), ctx)?; - let indices_validity = indices_array - .validity() - .vortex_expect("Failed to compute validity mask") - .execute_mask(indices_array.as_ref().len(), ctx)?; - - let mut new_offsets = PrimitiveBuilder::::with_capacity( - Nullability::NonNullable, - indices.len(), + if !data_validity.all_true() { + return Ok(None); + } + + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let offsets = array.offsets().clone().execute::(ctx)?; + let offsets = offsets.reinterpret_cast(offsets.ptype().to_unsigned()); + let output_len = indices_ref.len(); + + let taken = match &lengths { + ConstantOrArray::Constant(length) => take_piecewise_sequence_constant_dispatch( + array, + &starts, + *length, + &offsets, + indices_ref, + output_len, + )?, + ConstantOrArray::Array(lengths) => take_piecewise_sequence_lengths_dispatch( + array, + &starts, + lengths, + &offsets, + indices_ref, + output_len, + )?, + }; + Ok(Some(taken)) +} + +fn take_piecewise_sequence_constant_dispatch( + array: ArrayView<'_, List>, + starts: &PrimitiveArray, + length: usize, + offsets: &PrimitiveArray, + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + take_piecewise_sequence_constant_start_dispatch::( + array, + starts, + length, + offsets, + indices_ref, + output_len, + ) + }) +} + +fn take_piecewise_sequence_constant_start_dispatch( + array: ArrayView<'_, List>, + starts: &PrimitiveArray, + length: usize, + offsets: &PrimitiveArray, + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(offsets.ptype(), |O| { + take_piecewise_sequence_constant_length::( + array, + starts.as_slice::(), + length, + offsets.as_slice::(), + indices_ref, + output_len, + ) + }) +} + +fn take_piecewise_sequence_lengths_dispatch( + array: ArrayView<'_, List>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + take_piecewise_sequence_lengths_start_dispatch::( + array, + starts, + lengths, + offsets, + indices_ref, + output_len, + ) + }) +} + +fn take_piecewise_sequence_lengths_start_dispatch( + array: ArrayView<'_, List>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + take_piecewise_sequence_lengths_start_length_dispatch::( + array, + starts, + lengths, + offsets, + indices_ref, + output_len, + ) + }) +} + +fn take_piecewise_sequence_lengths_start_length_dispatch( + array: ArrayView<'_, List>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(offsets.ptype(), |O| { + take_piecewise_sequence_typed::( + array, + starts.as_slice::(), + lengths.as_slice::(), + offsets.as_slice::(), + indices_ref, + output_len, + ) + }) +} + +fn take_piecewise_sequence_constant_length( + array: ArrayView<'_, List>, + starts: &[S], + length: usize, + offsets: &[Offset], + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, + Offset: UnsignedPType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" ); + let total_elements = + piecewise_list_elements_len_constant(array.elements().len(), offsets, starts, length)?; + let validity = array.validity()?.take(indices_ref)?; + + match_smallest_offset_type!(total_elements, |OutputOffset| { + let gathered = gather_piecewise_list_constant_length::( + array.elements(), + offsets, + starts, + length, + output_len, + total_elements, + )?; + + // SAFETY: output offsets are rebuilt from valid monotonic source offsets; output elements + // are exactly the gathered child ranges referenced by those offsets; validity has one bit + // per output row. + Ok( + unsafe { ListArray::new_unchecked(gathered.elements, gathered.offsets, validity) } + .into_array(), + ) + }) +} - // This will be the indices we push down to the child array to call `take` with. - // - // There are 2 things to note here: - // - We do not know how many elements we need to take from our child since lists are variable - // size: thus we arbitrarily choose a capacity of `2 * # of indices`. - // - The type of the primitive builder needs to fit the largest offset of the (parent) - // `ListArray`, so we make this `PrimitiveBuilder` generic over `O` (instead of `I`). - let mut elements_to_take = - PrimitiveBuilder::::with_capacity(Nullability::NonNullable, 2 * indices.len()); - - let mut current_offset = OutputOffsetType::zero(); - new_offsets.append_zero(); - - for (data_idx, index_valid) in indices.iter().zip(indices_validity.iter()) { - if !index_valid { - new_offsets.append_value(current_offset); +fn take_piecewise_sequence_typed( + array: ArrayView<'_, List>, + starts: &[S], + lengths: &[L], + offsets: &[Offset], + indices_ref: &ArrayRef, + output_len: usize, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, + Offset: UnsignedPType, +{ + let mut computed_len = 0usize; + for &length in lengths { + let length: usize = length.as_(); + computed_len = computed_len + .checked_add(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + } + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + let total_elements = + piecewise_list_elements_len(array.elements().len(), offsets, starts, lengths)?; + + match_smallest_offset_type!(total_elements, |OutputOffset| { + let gathered = gather_piecewise_list::( + array.elements(), + offsets, + starts, + lengths, + output_len, + total_elements, + )?; + let validity = array.validity()?.take(indices_ref)?; + + // SAFETY: output offsets are rebuilt from valid monotonic source offsets; output elements + // are exactly the gathered child ranges referenced by those offsets; validity has one bit + // per output row. + Ok( + unsafe { ListArray::new_unchecked(gathered.elements, gathered.offsets, validity) } + .into_array(), + ) + }) +} + +struct GatheredList { + elements: ArrayRef, + offsets: ArrayRef, +} + +fn piecewise_list_elements_len_constant( + elements_len: usize, + offsets: &[Offset], + starts: &[S], + length: usize, +) -> VortexResult +where + S: UnsignedPType, + Offset: UnsignedPType, +{ + let mut total = 0usize; + for &start in starts { + let start: usize = start.as_(); + if length == 0 { continue; } - let data_idx: usize = data_idx.as_(); + let offset_range = &offsets[start..][..=length]; + let element_start: usize = offset_range[0].as_(); + let element_end: usize = offset_range[length].as_(); + vortex_ensure!( + element_start <= element_end && element_end <= elements_len, + "List offsets range {element_start}..{element_end} exceeds elements length {elements_len}", + ); + total = total + .checked_add(element_end - element_start) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + } + Ok(total) +} - if !data_validity.value(data_idx) { - new_offsets.append_value(current_offset); +fn piecewise_list_elements_len( + elements_len: usize, + offsets: &[Offset], + starts: &[S], + lengths: &[L], +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, + Offset: UnsignedPType, +{ + let mut total = 0usize; + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start: usize = start.as_(); + let length: usize = length.as_(); + if length == 0 { continue; } - let start = offsets[data_idx]; - let stop = offsets[data_idx + 1]; + let offset_range = &offsets[start..][..=length]; + let element_start: usize = offset_range[0].as_(); + let element_end: usize = offset_range[length].as_(); + vortex_ensure!( + element_start <= element_end && element_end <= elements_len, + "List offsets range {element_start}..{element_end} exceeds elements length {elements_len}", + ); + total = total + .checked_add(element_end - element_start) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + } + Ok(total) +} + +fn gather_piecewise_list_constant_length( + elements: &ArrayRef, + offsets: &[Offset], + starts: &[S], + length: usize, + output_len: usize, + total_elements: usize, +) -> VortexResult +where + S: UnsignedPType, + Offset: UnsignedPType, + OutputOffset: IntegerPType, +{ + let offsets_capacity = output_len + .checked_add(1) + .ok_or_else(|| vortex_err!("List take offsets length overflow"))?; + let mut new_offsets = BufferMut::::with_capacity(offsets_capacity); + let mut element_starts = BufferMut::::with_capacity(starts.len()); + let mut element_lengths = BufferMut::::with_capacity(starts.len()); + let mut output_elements = 0usize; + + new_offsets.push(OutputOffset::zero()); + for &start in starts { + let start: usize = start.as_(); + if length == 0 { + continue; + } + + let offset_range = &offsets[start..][..=length]; + let element_start: usize = offset_range[0].as_(); + let element_end: usize = offset_range[length].as_(); + for &offset in &offset_range[1..] { + let offset: usize = offset.as_(); + let relative = offset + .checked_sub(element_start) + .ok_or_else(|| vortex_err!("List offsets are not monotonic at offset {offset}"))?; + let output_offset = output_elements + .checked_add(relative) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + new_offsets.push(new_offset_value::(output_offset)?); + } - // See the note in `_take` on the reasoning. - let additional: usize = (stop - start).as_(); + let element_length = element_end - element_start; + element_starts.push(element_start as u64); + element_lengths.push(element_length as u64); + output_elements = output_elements + .checked_add(element_length) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + } + debug_assert_eq!(output_elements, total_elements); + + let offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable).into_array(); + let multipliers = ConstantArray::new(1u64, element_starts.len()).into_array(); + // SAFETY: element ranges are derived from validated source list offsets, and total_elements is + // the sum of the gathered element range lengths. Multiplier 1 preserves contiguous ranges. + let element_indices = unsafe { + PiecewiseSequenceArray::new_unchecked( + element_starts.into_array(), + element_lengths.into_array(), + multipliers, + total_elements, + ) + }; + let elements = elements.take(element_indices.into_array())?; + + Ok(GatheredList { elements, offsets }) +} + +fn gather_piecewise_list( + elements: &ArrayRef, + offsets: &[Offset], + starts: &[S], + lengths: &[L], + output_len: usize, + total_elements: usize, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, + Offset: UnsignedPType, + OutputOffset: IntegerPType, +{ + let offsets_capacity = output_len + .checked_add(1) + .ok_or_else(|| vortex_err!("List take offsets length overflow"))?; + let mut new_offsets = BufferMut::::with_capacity(offsets_capacity); + let mut element_starts = BufferMut::::with_capacity(starts.len()); + let mut element_lengths = BufferMut::::with_capacity(lengths.len()); + let mut output_elements = 0usize; + + new_offsets.push(OutputOffset::zero()); + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start: usize = start.as_(); + let length: usize = length.as_(); + if length == 0 { + continue; + } - elements_to_take.reserve_exact(additional); - for i in 0..additional { - elements_to_take.append_value(start + O::from_usize(i).vortex_expect("i < additional")); + let offset_range = &offsets[start..][..=length]; + let element_start: usize = offset_range[0].as_(); + let element_end: usize = offset_range[length].as_(); + for &offset in &offset_range[1..] { + let offset: usize = offset.as_(); + let relative = offset + .checked_sub(element_start) + .ok_or_else(|| vortex_err!("List offsets are not monotonic at offset {offset}"))?; + let output_offset = output_elements + .checked_add(relative) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; + new_offsets.push(new_offset_value::(output_offset)?); } - current_offset += - OutputOffsetType::from_usize((stop - start).as_()).vortex_expect("offset conversion"); - new_offsets.append_value(current_offset); + + let element_length = element_end - element_start; + element_starts.push(element_start as u64); + element_lengths.push(element_length as u64); + output_elements = output_elements + .checked_add(element_length) + .ok_or_else(|| vortex_err!("List take output elements length overflow"))?; } + debug_assert_eq!(output_elements, total_elements); + + let offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable).into_array(); + let multipliers = ConstantArray::new(1u64, element_starts.len()).into_array(); + // SAFETY: element ranges are derived from validated source list offsets, and total_elements is + // the sum of the gathered element range lengths. Multiplier 1 preserves contiguous ranges. + let element_indices = unsafe { + PiecewiseSequenceArray::new_unchecked( + element_starts.into_array(), + element_lengths.into_array(), + multipliers, + total_elements, + ) + }; + let elements = elements.take(element_indices.into_array())?; - let elements_to_take = elements_to_take.finish(); - let new_offsets = new_offsets.finish(); - let new_elements = array.elements().take(elements_to_take)?; + Ok(GatheredList { elements, offsets }) +} - Ok(ListArray::try_new( - new_elements, - new_offsets, - array.validity()?.take(indices_array.array())?, - )? - .into_array()) +fn new_offset_value(value: usize) -> VortexResult { + T::from_usize(value).ok_or_else(|| { + vortex_err!( + "List take offset value {value} does not fit in {}", + T::PTYPE + ) + }) } #[cfg(test)] @@ -215,9 +619,12 @@ mod test { use crate::VortexSessionExecute; use crate::array_session; use crate::arrays::BoolArray; + use crate::arrays::ConstantArray; use crate::arrays::ListArray; use crate::arrays::ListViewArray; + use crate::arrays::PiecewiseSequenceArray; use crate::arrays::PrimitiveArray; + use crate::assert_arrays_eq; use crate::compute::conformance::take::test_take_conformance; use crate::dtype::DType; use crate::dtype::Nullability; @@ -306,6 +713,55 @@ mod test { ); } + #[test] + fn null_index_ignores_out_of_bounds_payload() { + let mut ctx = array_session().create_execution_ctx(); + let list = ListArray::try_new( + buffer![1i32, 2, 3, 4].into_array(), + buffer![0u32, 2, 4].into_array(), + Validity::NonNullable, + ) + .unwrap() + .into_array(); + + let idx = PrimitiveArray::new( + buffer![1u32, 99, 0], + Validity::from_iter([true, false, true]), + ) + .into_array(); + let result = list.take(idx).unwrap(); + + let expected = ListArray::new( + buffer![3i32, 4, 1, 2].into_array(), + buffer![0u32, 2, 2, 4].into_array(), + Validity::from_iter([true, false, true]), + ); + assert_arrays_eq!(expected, result, &mut ctx); + } + + #[test] + fn null_source_row_ignores_invalid_offset_payload() { + let mut ctx = array_session().create_execution_ctx(); + let list = unsafe { + ListArray::new_unchecked( + buffer![1i32, 2].into_array(), + buffer![0u32, 2, 999].into_array(), + Validity::from_iter([true, false]), + ) + } + .into_array(); + + let idx = buffer![0u32, 1].into_array(); + let result = list.take(idx).unwrap(); + + let expected = ListArray::new( + buffer![1i32, 2].into_array(), + buffer![0u32, 2, 2].into_array(), + Validity::from_iter([true, false]), + ); + assert_arrays_eq!(expected, result, &mut ctx); + } + #[test] fn change_validity() { let list = ListArray::try_new( @@ -403,6 +859,54 @@ mod test { ); } + #[test] + fn piecewise_sequence_take() { + let mut ctx = array_session().create_execution_ctx(); + let list = ListArray::try_new( + buffer![0i32, 1, 2, 3, 4, 5, 6].into_array(), + buffer![0u32, 2, 5, 5, 7].into_array(), + Validity::NonNullable, + ) + .unwrap() + .into_array(); + let idx = PiecewiseSequenceArray::try_new( + buffer![1u64, 0].into_array(), + buffer![2u64, 1].into_array(), + ConstantArray::new(1u64, 2).into_array(), + 3, + ) + .unwrap() + .into_array(); + + let result = list + .take(idx) + .unwrap() + .execute::(&mut ctx) + .unwrap(); + + let element_dtype: Arc = Arc::new(I32.into()); + assert_eq!( + result.execute_scalar(0, &mut ctx).unwrap(), + Scalar::list( + Arc::clone(&element_dtype), + vec![2i32.into(), 3.into(), 4.into()], + Nullability::NonNullable + ) + ); + assert_eq!( + result.execute_scalar(1, &mut ctx).unwrap(), + Scalar::list(Arc::clone(&element_dtype), vec![], Nullability::NonNullable) + ); + assert_eq!( + result.execute_scalar(2, &mut ctx).unwrap(), + Scalar::list( + element_dtype, + vec![0i32.into(), 1.into()], + Nullability::NonNullable + ) + ); + } + #[test] fn test_take_empty_array() { let list = ListArray::try_new( diff --git a/vortex-array/src/arrays/listview/compute/take.rs b/vortex-array/src/arrays/listview/compute/take.rs index 2b6c016d2c3..01ffb80dab7 100644 --- a/vortex-array/src/arrays/listview/compute/take.rs +++ b/vortex-array/src/arrays/listview/compute/take.rs @@ -50,14 +50,20 @@ fn apply_take(array: ArrayView<'_, ListView>, indices: &ArrayRef) -> VortexResul // duplicates. let nullable_new_offsets = offsets.take(indices.clone())?; let nullable_new_sizes = sizes.take(indices.clone())?; + let validity_array = new_validity.to_array(indices.len()); - // `take` returns nullable arrays; cast back to non-nullable (filling with zeros to represent - // the null lists — the validity mask tracks nullness separately). + // Null output rows may carry arbitrary physical offset/size payloads from either the indices + // or the source rows. Mask with the final validity before filling so metadata placeholders are + // safe and non-nullable. let new_offsets = match_each_integer_ptype!(nullable_new_offsets.dtype().as_ptype(), |O| { - nullable_new_offsets.fill_null(Scalar::primitive(O::zero(), Nullability::NonNullable))? + nullable_new_offsets + .mask(validity_array.clone())? + .fill_null(Scalar::primitive(O::zero(), Nullability::NonNullable))? }); let new_sizes = match_each_integer_ptype!(nullable_new_sizes.dtype().as_ptype(), |S| { - nullable_new_sizes.fill_null(Scalar::primitive(S::zero(), Nullability::NonNullable))? + nullable_new_sizes + .mask(validity_array)? + .fill_null(Scalar::primitive(S::zero(), Nullability::NonNullable))? }); // SAFETY: Take operation maintains all `ListViewArray` invariants: diff --git a/vortex-array/src/arrays/listview/rebuild.rs b/vortex-array/src/arrays/listview/rebuild.rs index 342c9eccf19..6e36958522e 100644 --- a/vortex-array/src/arrays/listview/rebuild.rs +++ b/vortex-array/src/arrays/listview/rebuild.rs @@ -5,16 +5,17 @@ use num_traits::FromPrimitive; use vortex_buffer::BufferMut; use vortex_error::VortexExpect; use vortex_error::VortexResult; +use vortex_error::vortex_err; +use vortex_mask::Mask; -use crate::Canonical; use crate::ExecutionCtx; use crate::IntoArray; use crate::arrays::ConstantArray; use crate::arrays::ListViewArray; +use crate::arrays::PiecewiseSequenceArray; use crate::arrays::PrimitiveArray; use crate::arrays::listview::ListViewArrayExt; use crate::arrays::primitive::PrimitiveArrayExt; -use crate::builders::builder_with_capacity; use crate::builtins::ArrayBuiltins; use crate::dtype::IntegerPType; use crate::dtype::Nullability; @@ -140,19 +141,18 @@ impl ListViewArray { // for sizes as well. match_each_unsigned_integer_ptype!(sizes_ptype.to_unsigned(), |S| { match offsets_ptype.to_unsigned() { - PType::U8 => self.naive_rebuild::(ctx), - PType::U16 => self.naive_rebuild::(ctx), - PType::U32 => self.naive_rebuild::(ctx), - PType::U64 => self.naive_rebuild::(ctx), + PType::U8 => self.rebuild_with_take_or_piecewise::(ctx), + PType::U16 => self.rebuild_with_take_or_piecewise::(ctx), + PType::U32 => self.rebuild_with_take_or_piecewise::(ctx), + PType::U64 => self.rebuild_with_take_or_piecewise::(ctx), _ => unreachable!("invalid offsets PType"), } }) } /// Picks between [`rebuild_with_take`](Self::rebuild_with_take) and - /// [`rebuild_list_by_list`](Self::rebuild_list_by_list) based on element dtype and average - /// list size. - fn naive_rebuild( + /// [`rebuild_with_piecewise`](Self::rebuild_with_piecewise) based on average list size. + fn rebuild_with_take_or_piecewise( &self, ctx: &mut ExecutionCtx, ) -> VortexResult { @@ -167,12 +167,12 @@ impl ListViewArray { if Self::should_use_take(total, self.len()) { self.rebuild_with_take::(ctx) } else { - self.rebuild_list_by_list::(ctx) + self.rebuild_with_piecewise::(ctx) } } /// Returns `true` when we are confident that `rebuild_with_take` will - /// outperform `rebuild_list_by_list`. + /// outperform `rebuild_with_piecewise`. /// /// Take is dramatically faster for small lists (often 10-100×) because it /// avoids per-list builder overhead. LBL is the safer default for larger @@ -243,86 +243,69 @@ impl ListViewArray { .reinterpret_cast(size_ptype) .into_array(); - // SAFETY: same invariants as `rebuild_list_by_list` — offsets are sequential and - // non-overlapping, all (offset, size) pairs reference valid elements, and the validity - // array is preserved from the original. + // SAFETY: offsets are sequential and non-overlapping, all (offset, size) pairs reference + // valid elements, and the validity array is preserved from the original. Ok(unsafe { ListViewArray::new_unchecked(elements, offsets, sizes, self.validity()?) .with_zero_copy_to_list(true) }) } - /// Rebuilds elements list-by-list: canonicalize elements upfront, then for each list `slice` - /// the relevant range and `append_to_builder` into a typed builder. - fn rebuild_list_by_list( + /// Rebuilds elements using one contiguous-run take over the element child. + fn rebuild_with_piecewise( &self, ctx: &mut ExecutionCtx, ) -> VortexResult { - let element_dtype = self - .dtype() - .as_list_element_opt() - .vortex_expect("somehow had a canonical list that was not a list"); - let new_offset_ptype = rebuilt_offset_ptype(self.offsets().dtype().as_ptype()); let size_ptype = self.sizes().dtype().as_ptype(); let offsets_canonical = self.offsets().clone().execute::(ctx)?; let offsets_canonical = offsets_canonical.reinterpret_cast(offsets_canonical.ptype().to_unsigned()); - let offsets_slice = offsets_canonical.as_slice::(); let sizes_canonical = self.sizes().clone().execute::(ctx)?; let sizes_canonical = sizes_canonical.reinterpret_cast(sizes_canonical.ptype().to_unsigned()); - let sizes_slice = sizes_canonical.as_slice::(); - - let len = offsets_slice.len(); - - let mut new_offsets = BufferMut::::with_capacity(len); - // TODO(connor)[ListView]: Do we really need to do this? - // The only reason we need to rebuild the sizes here is that the validity may indicate that - // a list is null even though it has a non-zero size. This rebuild will set the size of all - // null lists to 0. - let mut new_sizes = BufferMut::::with_capacity(len); - - // Canonicalize the elements up front as we will be slicing the elements quite a lot. - let elements_canonical = self - .elements() - .clone() - .execute::(ctx)? - .into_array(); - - // Note that we do not know what the exact capacity should be of the new elements since - // there could be overlaps in the existing `ListViewArray`. - let mut new_elements_builder = - builder_with_capacity(element_dtype.as_ref(), self.elements().len()); - - // Resolve validity to a mask once instead of probing it per row (see `rebuild_with_take`). - let validity = self.validity()?.execute_mask(len, ctx)?; - - let mut n_elements = NewOffset::zero(); - for index in 0..len { - if !validity.value(index) { - // For NULL lists, place them after the previous item's data to maintain the - // no-overlap invariant for zero-copy to `ListArray` arrays. - new_offsets.push(n_elements); - new_sizes.push(S::zero()); - continue; - } - let offset = offsets_slice[index]; - let size = sizes_slice[index]; - - let start = offset.as_(); - let stop = start + size.as_(); - - new_offsets.push(n_elements); - new_sizes.push(size); - elements_canonical - .slice(start..stop)? - .append_to_builder(new_elements_builder.as_mut(), ctx)?; + let len = offsets_canonical.len(); + let validity = self.validity()?; + let validity_mask = validity.execute_mask(len, ctx)?; - n_elements += num_traits::cast(size).vortex_expect("Cast failed"); - } + let ranges = match_each_unsigned_integer_ptype!(offsets_canonical.ptype(), |O| { + rebuild_ranges::( + offsets_canonical.as_slice::(), + sizes_canonical.as_slice::(), + &validity_mask, + ) + })?; + + let RebuildRanges { + new_offsets, + new_sizes, + starts, + lengths, + elements_len, + } = ranges; + let constant_length = lengths + .first() + .copied() + .filter(|first| lengths.iter().all(|length| *length == *first)); + let lengths = match constant_length { + Some(length) => ConstantArray::new(length, starts.len()).into_array(), + None => lengths.into_array(), + }; + + // SAFETY: range starts and lengths are derived from valid ListView metadata; elements_len + // is the sum of all generated range lengths. Multiplier 1 preserves contiguous ranges. + let multipliers = ConstantArray::new(1u64, starts.len()).into_array(); + let element_indices = unsafe { + PiecewiseSequenceArray::new_unchecked( + starts.into_array(), + lengths, + multipliers, + elements_len, + ) + }; + let elements = self.elements().take(element_indices.into_array())?; // Built unsigned; reinterpret back to the signed-preserving result types. let offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable) @@ -331,24 +314,11 @@ impl ListViewArray { let sizes = PrimitiveArray::new(new_sizes.freeze(), Validity::NonNullable) .reinterpret_cast(size_ptype) .into_array(); - let elements = new_elements_builder.finish(); - debug_assert_eq!( - n_elements.as_(), - elements.len(), - "The accumulated elements somehow had the wrong length" - ); - - // SAFETY: - // - All offsets are sequential and non-overlapping (`n_elements` tracks running total). - // - Each `offset[i] + size[i]` equals `offset[i+1]` for all valid indices (including null - // lists). - // - All elements referenced by (offset, size) pairs exist within the new `elements` array. - // - The validity array is preserved from the original array unchanged - // - The array satisfies the zero-copy-to-list property by having sorted offsets, no gaps, - // and no overlaps. + // SAFETY: offsets are sequential and non-overlapping, all (offset, size) pairs reference + // valid elements, and validity is preserved from the original array. Ok(unsafe { - ListViewArray::new_unchecked(elements, offsets, sizes, self.validity()?) + ListViewArray::new_unchecked(elements, offsets, sizes, validity) .with_zero_copy_to_list(true) }) } @@ -414,6 +384,69 @@ impl ListViewArray { } } +struct RebuildRanges { + new_offsets: BufferMut, + new_sizes: BufferMut, + starts: BufferMut, + lengths: BufferMut, + elements_len: usize, +} + +fn rebuild_ranges( + offsets: &[O], + sizes: &[S], + validity_mask: &Mask, +) -> VortexResult> +where + NewOffset: IntegerPType, + O: IntegerPType, + S: IntegerPType, +{ + let len = offsets.len(); + let mut new_offsets = BufferMut::::with_capacity(len); + let mut new_sizes = BufferMut::::with_capacity(len); + let mut starts = BufferMut::::with_capacity(len); + let mut lengths = BufferMut::::with_capacity(len); + let mut n_elements = NewOffset::zero(); + let mut elements_len = 0usize; + + for (index, is_valid) in validity_mask.iter().enumerate() { + if !is_valid { + new_offsets.push(n_elements); + new_sizes.push(S::zero()); + starts.push(0); + lengths.push(0); + continue; + } + + let size = sizes[index]; + let start: usize = offsets[index].as_(); + let length: usize = size.as_(); + start.checked_add(length).ok_or_else(|| { + vortex_err!( + "ListView rebuild element range overflow for start {start} and length {length}" + ) + })?; + elements_len = elements_len + .checked_add(length) + .ok_or_else(|| vortex_err!("ListView rebuild elements length overflow"))?; + + new_offsets.push(n_elements); + new_sizes.push(size); + starts.push(start as u64); + lengths.push(length as u64); + n_elements += num_traits::cast(size).vortex_expect("Cast failed"); + } + + Ok(RebuildRanges { + new_offsets, + new_sizes, + starts, + lengths, + elements_len, + }) +} + #[cfg(test)] mod tests { use vortex_buffer::BitBuffer; @@ -511,6 +544,44 @@ mod tests { Ok(()) } + #[test] + fn test_rebuild_flatten_null_row_ignores_invalid_range_payload() -> VortexResult<()> { + let elements = PrimitiveArray::from_iter(vec![1i32, 2, 3]).into_array(); + let offsets = PrimitiveArray::from_iter(vec![0u32, 999, 2]).into_array(); + let sizes = PrimitiveArray::from_iter(vec![2u32, 999, 1]).into_array(); + let validity = Validity::from_iter([true, false, true]); + + // SAFETY: this intentionally models a null row whose physical offset and size payloads are + // invalid. Rebuild must ignore those payloads and only emit safe placeholder ranges for the + // null row. + let listview = unsafe { ListViewArray::new_unchecked(elements, offsets, sizes, validity) }; + + let mut ctx = SESSION.create_execution_ctx(); + let flattened = listview.rebuild(ListViewRebuildMode::MakeZeroCopyToList, &mut ctx)?; + + assert_eq!(flattened.offset_at(0), 0); + assert_eq!(flattened.size_at(0), 2); + assert_eq!(flattened.offset_at(1), 2); + assert_eq!(flattened.size_at(1), 0); + assert_eq!(flattened.offset_at(2), 2); + assert_eq!(flattened.size_at(2), 1); + assert!(flattened.validity()?.execute_is_valid(0, &mut ctx)?); + assert!(!flattened.validity()?.execute_is_valid(1, &mut ctx)?); + assert!(flattened.validity()?.execute_is_valid(2, &mut ctx)?); + + assert_arrays_eq!( + flattened.list_elements_at(0)?, + PrimitiveArray::from_iter([1i32, 2]), + &mut ctx + ); + assert_arrays_eq!( + flattened.list_elements_at(2)?, + PrimitiveArray::from_iter([3i32]), + &mut ctx + ); + Ok(()) + } + #[test] fn test_rebuild_trim_elements_basic() -> VortexResult<()> { // Test trimming both leading and trailing unused elements while preserving gaps in the @@ -561,6 +632,35 @@ mod tests { Ok(()) } + #[test] + fn test_rebuild_flatten_large_lists_with_piecewise_indices() -> VortexResult<()> { + let elements = PrimitiveArray::from_iter(0i32..300).into_array(); + let offsets = PrimitiveArray::from_iter(vec![10u32, 0]).into_array(); + let sizes = PrimitiveArray::from_iter(vec![150u32, 130]).into_array(); + let listview = ListViewArray::new(elements, offsets, sizes, Validity::NonNullable); + + let mut ctx = SESSION.create_execution_ctx(); + let flattened = listview.rebuild(ListViewRebuildMode::MakeZeroCopyToList, &mut ctx)?; + + assert_eq!(flattened.elements().len(), 280); + assert_eq!(flattened.offset_at(0), 0); + assert_eq!(flattened.size_at(0), 150); + assert_eq!(flattened.offset_at(1), 150); + assert_eq!(flattened.size_at(1), 130); + + assert_arrays_eq!( + flattened.list_elements_at(0)?, + PrimitiveArray::from_iter(10i32..160), + &mut ctx + ); + assert_arrays_eq!( + flattened.list_elements_at(1)?, + PrimitiveArray::from_iter(0i32..130), + &mut ctx + ); + Ok(()) + } + #[test] fn test_rebuild_with_trailing_nulls_regression() -> VortexResult<()> { // Regression test for issue #5412 diff --git a/vortex-array/src/arrays/listview/tests/take.rs b/vortex-array/src/arrays/listview/tests/take.rs index 2af397e262f..8fb7a4af1bb 100644 --- a/vortex-array/src/arrays/listview/tests/take.rs +++ b/vortex-array/src/arrays/listview/tests/take.rs @@ -101,6 +101,29 @@ fn test_take_with_gaps() { ); } +#[test] +fn test_take_null_source_row_zeros_offset_size_payloads() { + let elements = buffer![1i32, 2].into_array(); + let offsets = buffer![0u32, 999].into_array(); + let sizes = buffer![2u32, 999].into_array(); + let validity = Validity::from_iter([true, false]); + let listview = + unsafe { ListViewArray::new_unchecked(elements, offsets, sizes, validity) }.into_array(); + + let result = listview.take(buffer![1u32].into_array()).unwrap(); + let result_list = result + .execute::(&mut SESSION.create_execution_ctx()) + .unwrap(); + + assert_eq!(result_list.offset_at(0), 0); + assert_eq!(result_list.size_at(0), 0); + assert!( + result_list + .is_invalid(0, &mut SESSION.create_execution_ctx()) + .unwrap() + ); +} + #[test] fn test_take_constant_arrays() { // ListView-specific: Test with ConstantArray for offsets/sizes. diff --git a/vortex-array/src/arrays/mod.rs b/vortex-array/src/arrays/mod.rs index 15b6b5aa6be..001e20dee65 100644 --- a/vortex-array/src/arrays/mod.rs +++ b/vortex-array/src/arrays/mod.rs @@ -88,6 +88,10 @@ pub mod patched; pub use patched::Patched; pub use patched::PatchedArray; +pub mod piecewise_sequence; +pub use piecewise_sequence::PiecewiseSequence; +pub use piecewise_sequence::PiecewiseSequenceArray; + pub mod primitive; pub use primitive::Primitive; pub use primitive::PrimitiveArray; diff --git a/vortex-array/src/arrays/piecewise_sequence/array.rs b/vortex-array/src/arrays/piecewise_sequence/array.rs new file mode 100644 index 00000000000..1559cf7d3ad --- /dev/null +++ b/vortex-array/src/arrays/piecewise_sequence/array.rs @@ -0,0 +1,70 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use smallvec::smallvec; +use vortex_error::VortexResult; + +use crate::ArrayRef; +use crate::array::Array; +use crate::array::ArrayParts; +use crate::array::EmptyArrayData; +use crate::array::TypedArrayRef; +use crate::array_slots; +use crate::arrays::PiecewiseSequence; +use crate::dtype::PType; + +#[array_slots(PiecewiseSequence)] +pub struct PiecewiseSequenceSlots { + /// The inclusive start index of each sequential piece. + pub starts: ArrayRef, + /// The length of each sequential piece. + pub lengths: ArrayRef, + /// The distance between consecutive indices in each piece. + pub multipliers: ArrayRef, +} + +/// Extension methods for [`Array`] values using the [`PiecewiseSequence`] encoding. +pub trait PiecewiseSequenceArrayExt: + TypedArrayRef + PiecewiseSequenceArraySlotsExt +{ +} +impl> PiecewiseSequenceArrayExt for T {} + +impl Array { + /// Constructs a new `PiecewiseSequenceArray` from start, length, and multiplier arrays. + /// + /// This validates only structural invariants: all children must be non-nullable unsigned + /// integer arrays with matching lengths, and the outer array length is the declared expanded + /// index length. Individual ranges are checked when the index array is executed or consumed by + /// a take implementation. + pub fn try_new( + starts: ArrayRef, + lengths: ArrayRef, + multipliers: ArrayRef, + len: usize, + ) -> VortexResult { + Array::try_from_parts( + ArrayParts::new(PiecewiseSequence, PType::U64.into(), len, EmptyArrayData) + .with_slots(smallvec![Some(starts), Some(lengths), Some(multipliers)]), + ) + } + + /// Constructs a new `PiecewiseSequenceArray` without validation. + /// + /// # Safety + /// + /// The caller must guarantee the same structural invariants as [`Self::try_new`]. + pub unsafe fn new_unchecked( + starts: ArrayRef, + lengths: ArrayRef, + multipliers: ArrayRef, + len: usize, + ) -> Self { + unsafe { + Array::from_parts_unchecked( + ArrayParts::new(PiecewiseSequence, PType::U64.into(), len, EmptyArrayData) + .with_slots(smallvec![Some(starts), Some(lengths), Some(multipliers)]), + ) + } + } +} diff --git a/vortex-array/src/arrays/piecewise_sequence/mod.rs b/vortex-array/src/arrays/piecewise_sequence/mod.rs new file mode 100644 index 00000000000..eba974604bc --- /dev/null +++ b/vortex-array/src/arrays/piecewise_sequence/mod.rs @@ -0,0 +1,216 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Index encoding for concatenated sequential ranges. +//! +//! A `PiecewiseSequenceArray` represents the expanded index sequence +//! `starts[i] + j * multipliers[i]` for `j` in `0..lengths[i]` for each piece `i`. It is +//! intended for take operations that can gather regular runs without materializing one index per +//! element. + +use std::ptr; + +use itertools::Itertools; +use num_traits::AsPrimitive; +use vortex_buffer::BufferMut; +use vortex_error::VortexResult; +use vortex_error::vortex_bail; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; + +use crate::ArrayRef; +use crate::array::ArrayView; +use crate::arrays::PrimitiveArray; +use crate::arrays::piecewise_sequence::array::PiecewiseSequenceArraySlotsExt; +use crate::dtype::UnsignedPType; +use crate::executor::ExecutionCtx; +use crate::scalar::PValue; + +pub mod array; +mod vtable; + +#[cfg(test)] +mod tests; + +pub use array::PiecewiseSequenceArrayExt; +pub use vtable::*; + +pub(crate) fn check_index_arrays( + starts: &ArrayRef, + lengths: &ArrayRef, + multipliers: &ArrayRef, +) -> VortexResult<()> { + check_index_array("starts", starts)?; + check_index_array("lengths", lengths)?; + check_index_array("multipliers", multipliers)?; + vortex_ensure!( + starts.len() == lengths.len(), + "PiecewiseSequenceArray starts length {} does not match lengths length {}", + starts.len(), + lengths.len() + ); + vortex_ensure!( + starts.len() == multipliers.len(), + "PiecewiseSequenceArray starts length {} does not match multipliers length {}", + starts.len(), + multipliers.len() + ); + Ok(()) +} + +pub(crate) fn execute_index_arrays( + array: ArrayView<'_, PiecewiseSequence>, + ctx: &mut ExecutionCtx, +) -> VortexResult<(PrimitiveArray, PrimitiveArray, PrimitiveArray)> { + let starts = array.starts().clone().execute::(ctx)?; + let lengths = array.lengths().clone().execute::(ctx)?; + let multipliers = array.multipliers().clone().execute::(ctx)?; + check_index_arrays(starts.as_ref(), lengths.as_ref(), multipliers.as_ref())?; + Ok((starts, lengths, multipliers)) +} + +pub(crate) enum ConstantOrArray { + Constant(usize), + Array(PrimitiveArray), +} + +pub(crate) fn maybe_contiguous_slices( + array: ArrayView<'_, PiecewiseSequence>, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + if !is_constant_multiplier_one(array.multipliers()) { + return Ok(None); + } + + check_index_arrays(array.starts(), array.lengths(), array.multipliers())?; + let starts = array.starts().clone().execute::(ctx)?; + if let Some(length) = constant_unsigned_usize(array.lengths())? { + check_index_arrays(starts.as_ref(), array.lengths(), array.multipliers())?; + return Ok(Some((starts, ConstantOrArray::Constant(length)))); + } + + let lengths = array.lengths().clone().execute::(ctx)?; + check_index_arrays(starts.as_ref(), lengths.as_ref(), array.multipliers())?; + Ok(Some((starts, ConstantOrArray::Array(lengths)))) +} + +pub(crate) fn is_constant_multiplier_one(multipliers: &ArrayRef) -> bool { + let Some(scalar) = multipliers.as_constant() else { + return false; + }; + matches!( + scalar.as_primitive_opt().and_then(|scalar| scalar.pvalue()), + Some(PValue::U8(1) | PValue::U16(1) | PValue::U32(1) | PValue::U64(1)) + ) +} + +pub(crate) fn copy_slice_to_spare( + buffer: &mut BufferMut, + cursor: usize, + source: &[T], + output_len: usize, +) -> VortexResult { + let end = cursor + .checked_add(source.len()) + .ok_or_else(|| vortex_err!("slice copy output length overflows usize"))?; + vortex_ensure!( + end <= output_len, + "slice copy length {end} exceeds declared output length {output_len}" + ); + vortex_ensure!( + buffer.is_empty(), + "slice copy buffer already has {} initialized values", + buffer.len() + ); + vortex_ensure!( + output_len <= buffer.capacity(), + "slice copy output length {output_len} exceeds buffer capacity {}", + buffer.capacity() + ); + + // SAFETY: the checks above prove `cursor..end` is inside the spare capacity of an + // uninitialized buffer allocated for at least `output_len` values. + let dst = unsafe { buffer.spare_capacity_mut().get_unchecked_mut(cursor..end) }; + // SAFETY: `dst` has exactly `source.len()` writable slots and does not overlap with `source`. + unsafe { + ptr::copy_nonoverlapping(source.as_ptr(), dst.as_mut_ptr().cast(), source.len()); + } + Ok(end) +} + +fn constant_unsigned_usize(array: &ArrayRef) -> VortexResult> { + let Some(scalar) = array.as_constant() else { + return Ok(None); + }; + + let Some(pvalue) = scalar.as_primitive_opt().and_then(|scalar| scalar.pvalue()) else { + vortex_bail!("PiecewiseSequenceArray constant length must be an unsigned integer"); + }; + + Ok(Some(match pvalue { + PValue::U8(value) => value as usize, + PValue::U16(value) => value as usize, + PValue::U32(value) => value as usize, + PValue::U64(value) => value.as_(), + _ => vortex_bail!("PiecewiseSequenceArray constant length must be an unsigned integer"), + })) +} + +fn check_index_array(name: &str, array: &ArrayRef) -> VortexResult<()> { + vortex_ensure!( + array.dtype().is_unsigned_int(), + "PiecewiseSequenceArray {name} must have unsigned integer dtype, got {}", + array.dtype() + ); + vortex_ensure!( + !array.dtype().is_nullable(), + "PiecewiseSequenceArray {name} must be non-nullable, got {}", + array.dtype() + ); + Ok(()) +} + +pub(crate) fn materialize_ranges( + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + multipliers: &PrimitiveArray, + output_len: usize, +) -> VortexResult> +where + S: UnsignedPType, + L: UnsignedPType, + M: UnsignedPType, +{ + let starts = starts.as_slice::(); + let lengths = lengths.as_slice::(); + let multipliers = multipliers.as_slice::(); + let mut values = BufferMut::with_capacity(output_len); + let mut computed_len = 0usize; + + for ((&start, &length), &multiplier) in starts.iter().zip_eq(lengths).zip_eq(multipliers) { + let start: usize = start.as_(); + let length: usize = length.as_(); + let multiplier: usize = multiplier.as_(); + if length != 0 { + let last_offset = length - 1; + let last_delta = last_offset + .checked_mul(multiplier) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray range overflows usize"))?; + start + .checked_add(last_delta) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray range overflows usize"))?; + } + computed_len = computed_len + .checked_add(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + + values.extend((0..length).map(|offset| (start + offset * multiplier) as u64)); + } + + if computed_len != output_len { + vortex_bail!( + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + } + Ok(values) +} diff --git a/vortex-array/src/arrays/piecewise_sequence/tests.rs b/vortex-array/src/arrays/piecewise_sequence/tests.rs new file mode 100644 index 00000000000..7a2a9401ec5 --- /dev/null +++ b/vortex-array/src/arrays/piecewise_sequence/tests.rs @@ -0,0 +1,390 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use num_traits::AsPrimitive; +use vortex_buffer::ByteBufferMut; +use vortex_buffer::buffer; +use vortex_error::VortexResult; +use vortex_session::registry::ReadContext; + +use crate::ArrayContext; +use crate::ArrayRef; +use crate::IntoArray; +use crate::VortexSessionExecute; +use crate::array_session; +use crate::arrays::BoolArray; +use crate::arrays::ConstantArray; +use crate::arrays::DecimalArray; +use crate::arrays::FixedSizeListArray; +use crate::arrays::ListArray; +use crate::arrays::PiecewiseSequence; +use crate::arrays::PiecewiseSequenceArray; +use crate::arrays::PrimitiveArray; +use crate::arrays::VarBinArray; +use crate::arrays::VarBinViewArray; +use crate::assert_arrays_eq; +use crate::dtype::DType; +use crate::dtype::DecimalDType; +use crate::dtype::Nullability; +use crate::serde::SerializeOptions; +use crate::serde::SerializedArray; +use crate::validity::Validity; + +fn piecewise_indices( + starts: impl IntoIterator, + lengths: &[u64], +) -> VortexResult { + let len = lengths + .iter() + .map(|&length| -> usize { length.as_() }) + .sum(); + let starts = PrimitiveArray::from_iter(starts).into_array(); + let lengths = PrimitiveArray::from_iter(lengths.iter().copied()).into_array(); + let multipliers = ConstantArray::new(1u64, lengths.len()).into_array(); + Ok(PiecewiseSequenceArray::try_new(starts, lengths, multipliers, len)?.into_array()) +} + +fn piecewise_indices_constant_length( + starts: impl IntoIterator, + length: u64, +) -> VortexResult { + let starts = starts.into_iter().collect::>(); + let length_usize: usize = length.as_(); + let len = starts.len() * length_usize; + let piece_count = starts.len(); + let starts = PrimitiveArray::from_iter(starts).into_array(); + let lengths = ConstantArray::new(length, piece_count).into_array(); + let multipliers = ConstantArray::new(1u64, piece_count).into_array(); + Ok(PiecewiseSequenceArray::try_new(starts, lengths, multipliers, len)?.into_array()) +} + +#[test] +fn materializes_piecewise_indices() -> VortexResult<()> { + let starts = buffer![3u64, 15, 21].into_array(); + let lengths = buffer![3u64, 3, 3].into_array(); + let multipliers = ConstantArray::new(1u64, 3).into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 9)?.into_array(); + + let expected = PrimitiveArray::from_iter([3u64, 4, 5, 15, 16, 17, 21, 22, 23]).into_array(); + assert_arrays_eq!(array, expected, &mut array_session().create_execution_ctx()); + Ok(()) +} + +#[test] +fn materializes_repeated_and_empty_ranges() -> VortexResult<()> { + let starts = buffer![5u64, 2, 5].into_array(); + let lengths = buffer![2u64, 0, 2].into_array(); + let multipliers = ConstantArray::new(1u64, 3).into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 4)?.into_array(); + + let expected = PrimitiveArray::from_iter([5u64, 6, 5, 6]).into_array(); + assert_arrays_eq!(array, expected, &mut array_session().create_execution_ctx()); + Ok(()) +} + +#[test] +fn materializes_multiplied_ranges() -> VortexResult<()> { + let starts = buffer![3u64, 15].into_array(); + let lengths = buffer![3u64, 2].into_array(); + let multipliers = buffer![2u64, 4].into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 5)?.into_array(); + + let expected = PrimitiveArray::from_iter([3u64, 5, 7, 15, 19]).into_array(); + assert_arrays_eq!(array, expected, &mut array_session().create_execution_ctx()); + Ok(()) +} + +#[test] +fn supports_constant_lengths() -> VortexResult<()> { + let starts = buffer![0u64, 10, 20].into_array(); + let lengths = ConstantArray::new(2u64, 3).into_array(); + let multipliers = ConstantArray::new(1u64, 3).into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 6)?.into_array(); + + let expected = PrimitiveArray::from_iter([0u64, 1, 10, 11, 20, 21]).into_array(); + assert_arrays_eq!(array, expected, &mut array_session().create_execution_ctx()); + Ok(()) +} + +#[test] +fn scalar_at_maps_into_piece() -> VortexResult<()> { + let starts = buffer![3u64, 15, 21].into_array(); + let lengths = buffer![3u64, 3, 3].into_array(); + let multipliers = buffer![1u64, 1, 1].into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 9)?.into_array(); + let mut ctx = array_session().create_execution_ctx(); + + assert_eq!(array.execute_scalar(0, &mut ctx)?, 3u64.into()); + assert_eq!(array.execute_scalar(4, &mut ctx)?, 16u64.into()); + assert_eq!(array.execute_scalar(8, &mut ctx)?, 23u64.into()); + Ok(()) +} + +#[test] +fn constructor_defers_range_value_validation() -> VortexResult<()> { + let starts = buffer![u64::MAX].into_array(); + let lengths = buffer![2u64].into_array(); + let multipliers = buffer![1u64].into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 2)?.into_array(); + + let err = array + .execute::(&mut array_session().create_execution_ctx()) + .unwrap_err(); + assert!( + err.to_string() + .contains("PiecewiseSequenceArray range overflows usize"), + "{err}" + ); + Ok(()) +} + +#[test] +fn execution_checks_declared_length() -> VortexResult<()> { + let starts = buffer![0u64, 3].into_array(); + let lengths = buffer![2u64, 2].into_array(); + let multipliers = ConstantArray::new(1u64, 2).into_array(); + let array = PiecewiseSequenceArray::try_new(starts, lengths, multipliers, 3)?.into_array(); + + let err = array + .execute::(&mut array_session().create_execution_ctx()) + .unwrap_err(); + assert!( + err.to_string() + .contains("PiecewiseSequenceArray expanded length 4 does not match declared length 3"), + "{err}" + ); + Ok(()) +} + +#[test] +fn serde_roundtrip_preserves_piecewise_indices() -> VortexResult<()> { + let array = PiecewiseSequenceArray::try_new( + buffer![3u32, 15, 21].into_array(), + buffer![2u16, 0, 2].into_array(), + buffer![2u8, 1, 3].into_array(), + 4, + )? + .into_array(); + let dtype = array.dtype().clone(); + let len = array.len(); + + let array_ctx = ArrayContext::empty(); + let serialized = array.serialize(&array_ctx, &array_session(), &SerializeOptions::default())?; + + let mut concat = ByteBufferMut::empty(); + for buffer in serialized { + concat.extend_from_slice(buffer.as_ref()); + } + + let parts = SerializedArray::try_from(concat.freeze())?; + let decoded = parts.decode( + &dtype, + len, + &ReadContext::new(array_ctx.to_ids()), + &array_session(), + )?; + + assert!(decoded.is::()); + assert_arrays_eq!( + decoded, + PrimitiveArray::from_iter([3u64, 5, 21, 24]).into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn primitive_take_consumes_piecewise_indices() -> VortexResult<()> { + let values = PrimitiveArray::from_iter(0i32..20).into_array(); + let taken = values.take(piecewise_indices([3, 10], &[2, 3])?)?; + + assert_arrays_eq!( + taken, + PrimitiveArray::from_iter([3i32, 4, 10, 11, 12]).into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn primitive_take_handles_non_unit_multiplier() -> VortexResult<()> { + let values = PrimitiveArray::from_iter(0i32..20).into_array(); + let indices = PiecewiseSequenceArray::try_new( + buffer![3u64].into_array(), + buffer![3u64].into_array(), + buffer![2u64].into_array(), + 3, + )? + .into_array(); + let taken = values.take(indices)?; + + assert_arrays_eq!( + taken, + PrimitiveArray::from_iter([3i32, 5, 7]).into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn bool_take_consumes_piecewise_indices() -> VortexResult<()> { + let values = BoolArray::from_iter([true, false, true, true, false, false]).into_array(); + let taken = values.take(piecewise_indices([1, 4], &[2, 2])?)?; + + assert_arrays_eq!( + taken, + BoolArray::from_iter([false, true, false, false]).into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn decimal_take_consumes_piecewise_indices() -> VortexResult<()> { + let decimal_dtype = DecimalDType::new(19, 2); + let values = DecimalArray::from_iter([10i128, 20, 30, 40, 50, 60], decimal_dtype).into_array(); + let taken = values.take(piecewise_indices([1, 4], &[2, 1])?)?; + + assert_arrays_eq!( + taken, + DecimalArray::from_iter([20i128, 30, 50], decimal_dtype).into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn varbinview_take_consumes_piecewise_indices() -> VortexResult<()> { + let values = VarBinViewArray::from_iter( + ["a", "bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(); + let taken = values.take(piecewise_indices([1, 3], &[2, 1])?)?; + + assert_arrays_eq!( + taken, + VarBinViewArray::from_iter( + ["bb", "ccc", "dddd"].map(Some), + DType::Utf8(Nullability::NonNullable) + ) + .into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn varbin_take_consumes_piecewise_indices() -> VortexResult<()> { + let values = VarBinArray::from_iter( + ["a", "bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(); + let taken = values.take(piecewise_indices([1, 3], &[2, 1])?)?; + + assert_arrays_eq!( + taken, + VarBinArray::from_iter( + ["bb", "ccc", "dddd"].map(Some), + DType::Utf8(Nullability::NonNullable) + ) + .into_array(), + &mut array_session().create_execution_ctx() + ); + Ok(()) +} + +#[test] +fn contiguous_take_consumers_support_constant_piecewise_lengths() -> VortexResult<()> { + let mut ctx = array_session().create_execution_ctx(); + let indices = piecewise_indices_constant_length([1, 3], 2)?; + + let primitive = PrimitiveArray::from_iter(0i32..10).into_array(); + assert_arrays_eq!( + primitive.take(indices.clone())?, + PrimitiveArray::from_iter([1i32, 2, 3, 4]).into_array(), + &mut ctx + ); + + let bools = BoolArray::from_iter([false, true, false, true, false]).into_array(); + assert_arrays_eq!( + bools.take(indices.clone())?, + BoolArray::from_iter([true, false, true, false]).into_array(), + &mut ctx + ); + + let decimal_dtype = DecimalDType::new(19, 2); + let decimals = DecimalArray::from_iter([10i128, 20, 30, 40, 50], decimal_dtype).into_array(); + assert_arrays_eq!( + decimals.take(indices.clone())?, + DecimalArray::from_iter([20i128, 30, 40, 50], decimal_dtype).into_array(), + &mut ctx + ); + + let varbinview = VarBinViewArray::from_iter( + ["a", "bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(); + assert_arrays_eq!( + varbinview.take(indices.clone())?, + VarBinViewArray::from_iter( + ["bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(), + &mut ctx + ); + + let varbin = VarBinArray::from_iter( + ["a", "bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(); + assert_arrays_eq!( + varbin.take(indices)?, + VarBinArray::from_iter( + ["bb", "ccc", "dddd", "eeeee"].map(Some), + DType::Utf8(Nullability::NonNullable), + ) + .into_array(), + &mut ctx + ); + + Ok(()) +} + +#[test] +fn list_take_consumes_constant_piecewise_lengths() -> VortexResult<()> { + let mut ctx = array_session().create_execution_ctx(); + let list = ListArray::try_new( + buffer![0i32, 1, 2, 3, 4, 5, 6].into_array(), + buffer![0u32, 2, 5, 5, 7].into_array(), + Validity::NonNullable, + )? + .into_array(); + let list_indices = piecewise_indices_constant_length([1, 0], 2)?; + let expected = ListArray::try_new( + buffer![2i32, 3, 4, 0, 1, 2, 3, 4].into_array(), + buffer![0u32, 3, 3, 5, 8].into_array(), + Validity::NonNullable, + )? + .into_array(); + assert_arrays_eq!(list.take(list_indices)?, expected, &mut ctx); + + Ok(()) +} + +#[test] +fn fixed_size_list_take_builds_piecewise_element_indices() -> VortexResult<()> { + let elements = PrimitiveArray::from_iter(0i32..12).into_array(); + let array = FixedSizeListArray::new(elements, 3, Validity::NonNullable, 4).into_array(); + let taken = array.take(buffer![1u64, 3].into_array())?; + + let expected_elements = PrimitiveArray::from_iter([3i32, 4, 5, 9, 10, 11]).into_array(); + let expected = + FixedSizeListArray::new(expected_elements, 3, Validity::NonNullable, 2).into_array(); + assert_arrays_eq!(taken, expected, &mut array_session().create_execution_ctx()); + Ok(()) +} diff --git a/vortex-array/src/arrays/piecewise_sequence/vtable.rs b/vortex-array/src/arrays/piecewise_sequence/vtable.rs new file mode 100644 index 00000000000..ac2eba61116 --- /dev/null +++ b/vortex-array/src/arrays/piecewise_sequence/vtable.rs @@ -0,0 +1,275 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use itertools::Itertools; +use prost::Message; +use smallvec::smallvec; +use vortex_error::VortexResult; +use vortex_error::vortex_bail; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; +use vortex_error::vortex_panic; +use vortex_session::VortexSession; +use vortex_session::registry::CachedId; + +use crate::ArrayParts; +use crate::ArrayRef; +use crate::ExecutionCtx; +use crate::ExecutionResult; +use crate::IntoArray; +use crate::array::Array; +use crate::array::ArrayId; +use crate::array::ArrayView; +use crate::array::EmptyArrayData; +use crate::array::OperationsVTable; +use crate::array::VTable; +use crate::array::ValidityVTable; +use crate::array::with_empty_buffers; +use crate::arrays::PrimitiveArray; +use crate::arrays::piecewise_sequence::array::PiecewiseSequenceArraySlotsExt; +use crate::arrays::piecewise_sequence::array::PiecewiseSequenceSlots; +use crate::arrays::piecewise_sequence::check_index_arrays; +use crate::arrays::piecewise_sequence::execute_index_arrays; +use crate::arrays::piecewise_sequence::materialize_ranges; +use crate::arrays::primitive::PrimitiveArrayExt; +use crate::buffer::BufferHandle; +use crate::dtype::DType; +use crate::dtype::PType; +use crate::dtype::UnsignedPType; +use crate::match_each_unsigned_integer_ptype; +use crate::scalar::Scalar; +use crate::serde::ArrayChildren; +use crate::validity::Validity; + +/// A [`PiecewiseSequence`]-encoded Vortex index array. +pub type PiecewiseSequenceArray = Array; + +#[derive(Clone, prost::Message)] +struct PiecewiseSequenceMetadata { + #[prost(uint64, tag = "1")] + num_pieces: u64, + #[prost(enumeration = "PType", tag = "2")] + starts_ptype: i32, + #[prost(enumeration = "PType", tag = "3")] + lengths_ptype: i32, + #[prost(enumeration = "PType", tag = "4")] + multipliers_ptype: i32, +} + +#[derive(Clone, Debug)] +pub struct PiecewiseSequence; + +impl VTable for PiecewiseSequence { + type TypedArrayData = EmptyArrayData; + type OperationsVTable = Self; + type ValidityVTable = Self; + + fn id(&self) -> ArrayId { + static ID: CachedId = CachedId::new("vortex.piecewise-sequence"); + *ID + } + + fn validate( + &self, + _data: &Self::TypedArrayData, + dtype: &DType, + _len: usize, + slots: &[Option], + ) -> VortexResult<()> { + vortex_ensure!( + dtype == &DType::from(PType::U64), + "PiecewiseSequenceArray dtype must be u64, got {dtype}" + ); + vortex_ensure!( + slots.len() == PiecewiseSequenceSlots::NAMES.len(), + "PiecewiseSequenceArray requires {} slots, got {}", + PiecewiseSequenceSlots::NAMES.len(), + slots.len() + ); + let starts = slots[PiecewiseSequenceSlots::STARTS] + .as_ref() + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray starts slot must be present"))?; + let lengths = slots[PiecewiseSequenceSlots::LENGTHS] + .as_ref() + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray lengths slot must be present"))?; + let multipliers = slots[PiecewiseSequenceSlots::MULTIPLIERS] + .as_ref() + .ok_or_else(|| { + vortex_err!("PiecewiseSequenceArray multipliers slot must be present") + })?; + check_index_arrays(starts, lengths, multipliers) + } + + fn nbuffers(_array: ArrayView<'_, Self>) -> usize { + 0 + } + + fn buffer(_array: ArrayView<'_, Self>, _idx: usize) -> BufferHandle { + vortex_panic!("PiecewiseSequenceArray has no buffers") + } + + fn buffer_name(_array: ArrayView<'_, Self>, _idx: usize) -> Option { + None + } + + fn with_buffers( + &self, + array: ArrayView<'_, Self>, + buffers: &[BufferHandle], + ) -> VortexResult> { + with_empty_buffers(self, array, buffers) + } + + fn slot_name(_array: ArrayView<'_, Self>, idx: usize) -> String { + PiecewiseSequenceSlots::NAMES[idx].to_string() + } + + fn serialize( + array: ArrayView<'_, Self>, + _session: &VortexSession, + ) -> VortexResult>> { + Ok(Some( + PiecewiseSequenceMetadata { + num_pieces: u64::try_from(array.starts().len()).map_err(|_| { + vortex_err!( + "PiecewiseSequenceArray piece count {} overflowed u64", + array.starts().len() + ) + })?, + starts_ptype: PType::try_from(array.starts().dtype())? as i32, + lengths_ptype: PType::try_from(array.lengths().dtype())? as i32, + multipliers_ptype: PType::try_from(array.multipliers().dtype())? as i32, + } + .encode_to_vec(), + )) + } + + fn deserialize( + &self, + dtype: &DType, + len: usize, + metadata: &[u8], + buffers: &[BufferHandle], + children: &dyn ArrayChildren, + _session: &VortexSession, + ) -> VortexResult> { + let metadata = PiecewiseSequenceMetadata::decode(metadata)?; + vortex_ensure!( + dtype == &DType::from(PType::U64), + "PiecewiseSequenceArray dtype must be u64, got {dtype}" + ); + vortex_ensure!( + buffers.is_empty(), + "PiecewiseSequenceArray expects no buffers, got {}", + buffers.len() + ); + vortex_ensure!( + children.len() == PiecewiseSequenceSlots::NAMES.len(), + "PiecewiseSequenceArray expects {} children, got {}", + PiecewiseSequenceSlots::NAMES.len(), + children.len() + ); + + let num_pieces = usize::try_from(metadata.num_pieces).map_err(|_| { + vortex_err!( + "PiecewiseSequenceArray piece count {} does not fit in usize", + metadata.num_pieces + ) + })?; + let starts = children.get( + PiecewiseSequenceSlots::STARTS, + &metadata.starts_ptype().into(), + num_pieces, + )?; + let lengths = children.get( + PiecewiseSequenceSlots::LENGTHS, + &metadata.lengths_ptype().into(), + num_pieces, + )?; + let multipliers = children.get( + PiecewiseSequenceSlots::MULTIPLIERS, + &metadata.multipliers_ptype().into(), + num_pieces, + )?; + + Ok( + ArrayParts::new(self.clone(), dtype.clone(), len, EmptyArrayData) + .with_slots(smallvec![Some(starts), Some(lengths), Some(multipliers)]), + ) + } + + fn execute(array: Array, ctx: &mut ExecutionCtx) -> VortexResult { + let (starts, lengths, multipliers) = execute_index_arrays(array.as_view(), ctx)?; + + let values = match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + match_each_unsigned_integer_ptype!(multipliers.ptype(), |M| { + materialize_ranges::(&starts, &lengths, &multipliers, array.len())? + }) + }) + }); + Ok(ExecutionResult::done( + PrimitiveArray::new(values.freeze(), Validity::NonNullable).into_array(), + )) + } +} + +impl OperationsVTable for PiecewiseSequence { + fn scalar_at( + array: ArrayView<'_, PiecewiseSequence>, + index: usize, + ctx: &mut ExecutionCtx, + ) -> VortexResult { + let (starts, lengths, multipliers) = execute_index_arrays(array, ctx)?; + + let value = match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + match_each_unsigned_integer_ptype!(multipliers.ptype(), |M| { + scalar_at::(&starts, &lengths, &multipliers, index)? + }) + }) + }); + Ok(value.into()) + } +} + +impl ValidityVTable for PiecewiseSequence { + fn validity(_array: ArrayView<'_, PiecewiseSequence>) -> VortexResult { + Ok(Validity::NonNullable) + } +} + +fn scalar_at( + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + multipliers: &PrimitiveArray, + index: usize, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, + M: UnsignedPType, +{ + let mut remaining = index; + for ((&start, &length), &multiplier) in starts + .as_slice::() + .iter() + .zip_eq(lengths.as_slice::()) + .zip_eq(multipliers.as_slice::()) + { + let length: usize = length.as_(); + if remaining < length { + let start: usize = start.as_(); + let multiplier: usize = multiplier.as_(); + let offset = remaining + .checked_mul(multiplier) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray range overflows usize"))?; + let value = start + .checked_add(offset) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray range overflows usize"))?; + return Ok(value as u64); + } + remaining -= length; + } + vortex_bail!("PiecewiseSequenceArray index {index} out of bounds") +} diff --git a/vortex-array/src/arrays/primitive/compute/take/mod.rs b/vortex-array/src/arrays/primitive/compute/take/mod.rs index dd562281608..43a079f0a3c 100644 --- a/vortex-array/src/arrays/primitive/compute/take/mod.rs +++ b/vortex-array/src/arrays/primitive/compute/take/mod.rs @@ -6,25 +6,35 @@ mod avx2; use std::sync::LazyLock; +use itertools::Itertools as _; use vortex_buffer::Buffer; use vortex_buffer::BufferMut; use vortex_error::VortexResult; use vortex_error::vortex_bail; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use vortex_mask::Mask; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; use crate::arrays::ConstantArray; +use crate::arrays::PiecewiseSequence; use crate::arrays::Primitive; use crate::arrays::PrimitiveArray; use crate::arrays::dict::TakeExecute; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::copy_slice_to_spare; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::builtins::ArrayBuiltins; use crate::dtype::DType; use crate::dtype::IntegerPType; +use crate::dtype::NativePType; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_integer_ptype; use crate::match_each_native_ptype; +use crate::match_each_unsigned_integer_ptype; use crate::scalar::Scalar; use crate::validity::Validity; @@ -79,6 +89,12 @@ impl TakeExecute for Primitive { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_contiguous_ranges(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + let DType::Primitive(ptype, null) = indices.dtype() else { vortex_bail!("Invalid indices dtype: {}", indices.dtype()) }; @@ -123,6 +139,28 @@ impl TakeExecute for Primitive { } } +fn take_contiguous_ranges( + array: ArrayView<'_, Primitive>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let validity = array.validity()?.take(indices_ref)?; + let output_len = indices_ref.len(); + let taken = match lengths { + ConstantOrArray::Constant(length) => { + take_slices_constant_length(array, &starts, length, validity, output_len)? + } + ConstantOrArray::Array(lengths) => { + take_slices(array, &starts, &lengths, validity, output_len)? + } + }; + Ok(Some(taken)) +} + // Compiler may see this as unused based on enabled features #[inline(always)] fn take_primitive_scalar(buffer: &[T], indices: &[I]) -> Buffer { @@ -144,6 +182,154 @@ fn take_primitive_scalar(buffer: &[T], indices: &[I]) result.freeze() } +fn take_slices( + array: ArrayView<'_, Primitive>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult { + match_each_native_ptype!(array.ptype(), |T| { + take_slices_typed::(array, starts, lengths, validity, output_len) + }) +} + +fn take_slices_typed( + array: ArrayView<'_, Primitive>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult +where + T: NativePType, +{ + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + take_slices_start_typed::(array, starts, lengths, validity, output_len) + }) +} + +fn take_slices_start_typed( + array: ArrayView<'_, Primitive>, + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + validity: Validity, + output_len: usize, +) -> VortexResult +where + T: NativePType, + S: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + let values = take_slices_to_buffer::( + array.as_slice::(), + starts.as_slice::(), + lengths.as_slice::(), + output_len, + )?; + Ok(PrimitiveArray::new(values, validity).into_array()) + }) +} + +fn take_slices_to_buffer( + source: &[T], + starts: &[S], + lengths: &[L], + output_len: usize, +) -> VortexResult> +where + T: Copy, + S: UnsignedPType, + L: UnsignedPType, +{ + let mut values = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + cursor = copy_slice_to_spare(&mut values, cursor, &source[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {} does not match declared length {output_len}", + cursor + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { values.set_len(output_len) }; + Ok(values.freeze()) +} + +fn take_slices_constant_length( + array: ArrayView<'_, Primitive>, + starts: &PrimitiveArray, + length: usize, + validity: Validity, + output_len: usize, +) -> VortexResult { + match_each_native_ptype!(array.ptype(), |T| { + take_slices_constant_length_typed::(array, starts, length, validity, output_len) + }) +} + +fn take_slices_constant_length_typed( + array: ArrayView<'_, Primitive>, + starts: &PrimitiveArray, + length: usize, + validity: Validity, + output_len: usize, +) -> VortexResult +where + T: NativePType, +{ + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + let values = take_slices_constant_length_to_buffer::( + array.as_slice::(), + starts.as_slice::(), + length, + output_len, + )?; + Ok(PrimitiveArray::new(values, validity).into_array()) + }) +} + +fn take_slices_constant_length_to_buffer( + source: &[T], + starts: &[S], + length: usize, + output_len: usize, +) -> VortexResult> +where + T: Copy, + S: UnsignedPType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + + let mut values = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for &start in starts { + let start = start.as_(); + cursor = copy_slice_to_spare(&mut values, cursor, &source[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {cursor} does not match declared length {output_len}" + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { values.set_len(output_len) }; + Ok(values.freeze()) +} + #[cfg(any(target_arch = "x86_64", target_arch = "x86"))] #[cfg(test)] mod test { diff --git a/vortex-array/src/arrays/varbin/compute/take.rs b/vortex-array/src/arrays/varbin/compute/take.rs index 6a6454d0603..7a81f05a4d3 100644 --- a/vortex-array/src/arrays/varbin/compute/take.rs +++ b/vortex-array/src/arrays/varbin/compute/take.rs @@ -1,26 +1,34 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +use itertools::Itertools as _; use vortex_buffer::BitBufferMut; use vortex_buffer::BufferMut; use vortex_buffer::ByteBufferMut; use vortex_error::VortexExpect; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use vortex_error::vortex_panic; use vortex_mask::Mask; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; +use crate::arrays::PiecewiseSequence; use crate::arrays::PrimitiveArray; use crate::arrays::VarBin; use crate::arrays::VarBinArray; use crate::arrays::dict::TakeExecute; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::copy_slice_to_spare; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::arrays::primitive::PrimitiveArrayExt; use crate::arrays::varbin::VarBinArrayExt; use crate::dtype::DType; use crate::dtype::IntegerPType; use crate::dtype::PType; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_unsigned_integer_ptype; use crate::validity::Validity; @@ -43,6 +51,12 @@ impl TakeExecute for VarBin { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_contiguous_ranges(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + // TODO(joe): Be lazy with execute let offsets = array.offsets().clone().execute::(ctx)?; let data = array.bytes(); @@ -113,6 +127,54 @@ impl TakeExecute for VarBin { } } +fn take_contiguous_ranges( + array: ArrayView<'_, VarBin>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let offsets = array.offsets().clone().execute::(ctx)?; + let out_offset_ptype = taken_offset_ptype(offsets.ptype()); + let offsets = offsets.reinterpret_cast(offsets.ptype().to_unsigned()); + let bytes = array.bytes(); + let data = bytes.as_slice(); + let dtype = array.dtype().clone(); + let output_len = indices_ref.len(); + + let result = match &lengths { + ConstantOrArray::Constant(length) => gather_slices_constant_dispatch( + &starts, + *length, + &offsets, + data, + output_len, + out_offset_ptype, + )?, + ConstantOrArray::Array(lengths) => gather_slices_dispatch( + &starts, + lengths, + &offsets, + data, + output_len, + out_offset_ptype, + )?, + }; + + let validity = array.validity()?.take(indices_ref)?; + + // SAFETY: output offsets are built from valid input offsets, start at zero, are monotonically + // non-decreasing, and the copied data buffer has exactly the referenced byte length. + unsafe { + Ok(Some( + VarBinArray::new_unchecked(result.offsets, result.data.freeze(), dtype, validity) + .into_array(), + )) + } +} + fn take( dtype: DType, offsets: &[Offset], @@ -183,6 +245,363 @@ fn take( } } +struct GatheredPiecewiseVarBin { + offsets: ArrayRef, + data: ByteBufferMut, +} + +fn gather_slices_constant_dispatch( + starts: &PrimitiveArray, + length: usize, + offsets: &PrimitiveArray, + data: &[u8], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + gather_slices_constant_start_dispatch::( + starts, + length, + offsets, + data, + output_len, + out_offset_ptype, + ) + }) +} + +fn gather_slices_constant_start_dispatch( + starts: &PrimitiveArray, + length: usize, + offsets: &PrimitiveArray, + data: &[u8], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult +where + S: UnsignedPType, +{ + match offsets.ptype() { + PType::U8 => gather_slices_constant_length::( + offsets.as_slice::(), + data, + starts.as_slice::(), + length, + output_len, + out_offset_ptype, + ), + PType::U16 => gather_slices_constant_length::( + offsets.as_slice::(), + data, + starts.as_slice::(), + length, + output_len, + out_offset_ptype, + ), + PType::U32 => gather_slices_constant_length::( + offsets.as_slice::(), + data, + starts.as_slice::(), + length, + output_len, + out_offset_ptype, + ), + PType::U64 => gather_slices_constant_length::( + offsets.as_slice::(), + data, + starts.as_slice::(), + length, + output_len, + out_offset_ptype, + ), + _ => unreachable!("offsets were reinterpreted to an unsigned integer ptype"), + } +} + +fn gather_slices_dispatch( + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + data: &[u8], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + gather_slices_start_dispatch::( + starts, + lengths, + offsets, + data, + output_len, + out_offset_ptype, + ) + }) +} + +fn gather_slices_start_dispatch( + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + data: &[u8], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult +where + S: UnsignedPType, +{ + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + gather_slices_start_length_dispatch::( + starts, + lengths, + offsets, + data, + output_len, + out_offset_ptype, + ) + }) +} + +fn gather_slices_start_length_dispatch( + starts: &PrimitiveArray, + lengths: &PrimitiveArray, + offsets: &PrimitiveArray, + data: &[u8], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, +{ + match offsets.ptype() { + PType::U8 => gather_slices::( + offsets.as_slice::(), + data, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + out_offset_ptype, + ), + PType::U16 => gather_slices::( + offsets.as_slice::(), + data, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + out_offset_ptype, + ), + PType::U32 => gather_slices::( + offsets.as_slice::(), + data, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + out_offset_ptype, + ), + PType::U64 => gather_slices::( + offsets.as_slice::(), + data, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + out_offset_ptype, + ), + _ => unreachable!("offsets were reinterpreted to an unsigned integer ptype"), + } +} + +fn gather_slices_constant_length( + offsets: &[Offset], + data: &[u8], + starts: &[S], + length: usize, + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult +where + S: UnsignedPType, + Offset: IntegerPType, + NewOffset: IntegerPType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + + let mut new_offsets = BufferMut::::with_capacity(output_len + 1); + new_offsets.push(NewOffset::zero()); + let mut output_bytes = 0usize; + + for &start in starts { + let start = start.as_(); + if length == 0 { + continue; + } + + let offset_range = &offsets[start..][..=length]; + let byte_start = offset_range[0].as_(); + let byte_end = offset_range[length].as_(); + vortex_ensure!( + byte_start <= byte_end && byte_end <= data.len(), + "VarBin offsets range {byte_start}..{byte_end} exceeds data length {}", + data.len() + ); + + for &offset in &offset_range[1..] { + let offset = offset.as_(); + let relative = offset.checked_sub(byte_start).ok_or_else(|| { + vortex_err!("VarBin offsets are not monotonic at offset {offset}") + })?; + let output_offset = output_bytes.checked_add(relative).ok_or_else(|| { + vortex_err!("PiecewiseSequence VarBin output byte length overflow") + })?; + new_offsets.push(new_offset_value::(output_offset)?); + } + + output_bytes = output_bytes + .checked_add(byte_end - byte_start) + .ok_or_else(|| vortex_err!("PiecewiseSequence VarBin output byte length overflow"))?; + } + + let mut new_data = ByteBufferMut::with_capacity(output_bytes); + let mut cursor = 0usize; + for &start in starts { + let start = start.as_(); + if length == 0 { + continue; + } + + let offset_range = &offsets[start..][..=length]; + let byte_start = offset_range[0].as_(); + let byte_end = offset_range[length].as_(); + cursor = copy_slice_to_spare( + &mut new_data, + cursor, + &data[byte_start..][..byte_end - byte_start], + output_bytes, + )?; + } + vortex_ensure!( + cursor == output_bytes, + "VarBin byte copy length {cursor} does not match computed byte length {output_bytes}" + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_bytes`, and `cursor == + // output_bytes` proves all bytes were initialized. + unsafe { new_data.set_len(output_bytes) }; + + let offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable) + .reinterpret_cast(out_offset_ptype) + .into_array(); + Ok(GatheredPiecewiseVarBin { + offsets, + data: new_data, + }) +} + +fn gather_slices( + offsets: &[Offset], + data: &[u8], + starts: &[S], + lengths: &[L], + output_len: usize, + out_offset_ptype: PType, +) -> VortexResult +where + S: UnsignedPType, + L: UnsignedPType, + Offset: IntegerPType, + NewOffset: IntegerPType, +{ + let mut new_offsets = BufferMut::::with_capacity(output_len + 1); + new_offsets.push(NewOffset::zero()); + let mut output_bytes = 0usize; + + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + if length == 0 { + continue; + } + + let offset_range = &offsets[start..][..=length]; + let byte_start = offset_range[0].as_(); + let byte_end = offset_range[length].as_(); + vortex_ensure!( + byte_start <= byte_end && byte_end <= data.len(), + "VarBin offsets range {byte_start}..{byte_end} exceeds data length {}", + data.len() + ); + + for &offset in &offset_range[1..] { + let offset = offset.as_(); + let relative = offset.checked_sub(byte_start).ok_or_else(|| { + vortex_err!("VarBin offsets are not monotonic at offset {offset}") + })?; + let output_offset = output_bytes.checked_add(relative).ok_or_else(|| { + vortex_err!("PiecewiseSequence VarBin output byte length overflow") + })?; + new_offsets.push(new_offset_value::(output_offset)?); + } + + output_bytes = output_bytes + .checked_add(byte_end - byte_start) + .ok_or_else(|| vortex_err!("PiecewiseSequence VarBin output byte length overflow"))?; + } + vortex_ensure!( + new_offsets.len() == output_len + 1, + "PiecewiseSequenceArray expanded length {} does not match declared length {output_len}", + new_offsets.len() - 1 + ); + + let mut new_data = ByteBufferMut::with_capacity(output_bytes); + let mut cursor = 0usize; + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + if length == 0 { + continue; + } + + let offset_range = &offsets[start..][..=length]; + let byte_start = offset_range[0].as_(); + let byte_end = offset_range[length].as_(); + cursor = copy_slice_to_spare( + &mut new_data, + cursor, + &data[byte_start..][..byte_end - byte_start], + output_bytes, + )?; + } + vortex_ensure!( + cursor == output_bytes, + "VarBin byte copy length {cursor} does not match computed byte length {output_bytes}" + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_bytes`, and `cursor == + // output_bytes` proves all bytes were initialized. + unsafe { new_data.set_len(output_bytes) }; + + let offsets = PrimitiveArray::new(new_offsets.freeze(), Validity::NonNullable) + .reinterpret_cast(out_offset_ptype) + .into_array(); + Ok(GatheredPiecewiseVarBin { + offsets, + data: new_data, + }) +} + +fn new_offset_value(value: usize) -> VortexResult { + T::from(value).ok_or_else(|| { + vortex_err!( + "PiecewiseSequence VarBin offset value {value} does not fit in {}", + T::PTYPE + ) + }) +} + fn take_nullable( dtype: DType, offsets: &[Offset], diff --git a/vortex-array/src/arrays/varbinview/compute/take.rs b/vortex-array/src/arrays/varbinview/compute/take.rs index ce10abda3d0..d6256b33078 100644 --- a/vortex-array/src/arrays/varbinview/compute/take.rs +++ b/vortex-array/src/arrays/varbinview/compute/take.rs @@ -4,23 +4,33 @@ use std::iter; use std::sync::Arc; +use itertools::Itertools as _; use num_traits::AsPrimitive; use vortex_buffer::Buffer; +use vortex_buffer::BufferMut; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; +use vortex_error::vortex_err; use vortex_mask::AllOr; use vortex_mask::Mask; use crate::ArrayRef; use crate::IntoArray; use crate::array::ArrayView; +use crate::arrays::PiecewiseSequence; use crate::arrays::PrimitiveArray; use crate::arrays::VarBinView; use crate::arrays::VarBinViewArray; use crate::arrays::dict::TakeExecute; +use crate::arrays::piecewise_sequence::ConstantOrArray; +use crate::arrays::piecewise_sequence::copy_slice_to_spare; +use crate::arrays::piecewise_sequence::maybe_contiguous_slices; use crate::arrays::varbinview::BinaryView; use crate::buffer::BufferHandle; +use crate::dtype::UnsignedPType; use crate::executor::ExecutionCtx; use crate::match_each_integer_ptype; +use crate::match_each_unsigned_integer_ptype; impl TakeExecute for VarBinView { /// Take involves creating a new array that references the old array, just with the given set of views. @@ -29,6 +39,12 @@ impl TakeExecute for VarBinView { indices: &ArrayRef, ctx: &mut ExecutionCtx, ) -> VortexResult> { + if let Some(piecewise_indices) = indices.as_opt::() + && let Some(taken) = take_contiguous_ranges(array, piecewise_indices, indices, ctx)? + { + return Ok(Some(taken)); + } + let validity = array.validity()?.take(indices)?; let indices = indices.clone().execute::(ctx)?; @@ -57,6 +73,58 @@ impl TakeExecute for VarBinView { } } +fn take_contiguous_ranges( + array: ArrayView<'_, VarBinView>, + indices: ArrayView<'_, PiecewiseSequence>, + indices_ref: &ArrayRef, + ctx: &mut ExecutionCtx, +) -> VortexResult> { + let Some((starts, lengths)) = maybe_contiguous_slices(indices, ctx)? else { + return Ok(None); + }; + let source = array.views(); + let output_len = indices_ref.len(); + let views = match &lengths { + ConstantOrArray::Constant(length) => { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + gather_view_slices_constant_length( + source, + starts.as_slice::(), + *length, + output_len, + )? + }) + } + ConstantOrArray::Array(lengths) => { + match_each_unsigned_integer_ptype!(starts.ptype(), |S| { + match_each_unsigned_integer_ptype!(lengths.ptype(), |L| { + gather_view_slices( + source, + starts.as_slice::(), + lengths.as_slice::(), + output_len, + )? + }) + }) + } + }; + let validity = array.validity()?.take(indices_ref)?; + + // SAFETY: ranges were validated against the source views, and copied views still reference the + // same backing data buffers. + unsafe { + Ok(Some( + VarBinViewArray::new_handle_unchecked( + BufferHandle::new_host(views.into_byte_buffer()), + Arc::clone(array.data_buffers()), + array.dtype().clone(), + validity, + ) + .into_array(), + )) + } +} + fn take_views>( views_ref: &[BinaryView], indices: &[I], @@ -85,6 +153,70 @@ fn take_views>( } } +fn gather_view_slices_constant_length( + source: &[BinaryView], + starts: &[S], + length: usize, + output_len: usize, +) -> VortexResult> +where + S: UnsignedPType, +{ + let computed_len = starts + .len() + .checked_mul(length) + .ok_or_else(|| vortex_err!("PiecewiseSequenceArray output length overflows usize"))?; + vortex_ensure!( + computed_len == output_len, + "PiecewiseSequenceArray expanded length {computed_len} does not match declared length {output_len}" + ); + + let mut views = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for &start in starts { + let start = start.as_(); + cursor = copy_slice_to_spare(&mut views, cursor, &source[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {cursor} does not match declared length {output_len}" + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { views.set_len(output_len) }; + Ok(views.freeze()) +} + +fn gather_view_slices( + source: &[BinaryView], + starts: &[S], + lengths: &[L], + output_len: usize, +) -> VortexResult> +where + S: UnsignedPType, + L: UnsignedPType, +{ + let mut views = BufferMut::::with_capacity(output_len); + let mut cursor = 0usize; + for (&start, &length) in starts.iter().zip_eq(lengths) { + let start = start.as_(); + let length = length.as_(); + cursor = copy_slice_to_spare(&mut views, cursor, &source[start..][..length], output_len)?; + } + + vortex_ensure!( + cursor == output_len, + "PiecewiseSequenceArray expanded length {} does not match declared length {output_len}", + cursor + ); + // SAFETY: `copy_slice_to_spare` checked every write against `output_len`, and `cursor == + // output_len` proves all slots were initialized. + unsafe { views.set_len(output_len) }; + Ok(views.freeze()) +} + #[cfg(test)] mod tests { use rstest::rstest; diff --git a/vortex-array/src/session/mod.rs b/vortex-array/src/session/mod.rs index 2cb8414cbfd..98a769ceea9 100644 --- a/vortex-array/src/session/mod.rs +++ b/vortex-array/src/session/mod.rs @@ -25,6 +25,7 @@ use crate::arrays::List; use crate::arrays::ListView; use crate::arrays::Masked; use crate::arrays::Null; +use crate::arrays::PiecewiseSequence; use crate::arrays::Primitive; use crate::arrays::Struct; use crate::arrays::VarBin; @@ -81,6 +82,7 @@ impl Default for ArraySession { this.register(Dict); this.register(List); this.register(Masked); + this.register(PiecewiseSequence); this.register(VarBin); this diff --git a/vortex-bench/Cargo.toml b/vortex-bench/Cargo.toml index 02f043b8e68..d3a23c976f1 100644 --- a/vortex-bench/Cargo.toml +++ b/vortex-bench/Cargo.toml @@ -58,6 +58,7 @@ regex = { workspace = true } reqwest = { workspace = true, features = ["stream"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } +sha2 = { workspace = true } spatialbench = { workspace = true } spatialbench-arrow = { workspace = true } spatialbench-parquet = { workspace = true }