diff --git a/Cargo.lock b/Cargo.lock index 6691c1efd5c..97afbbb2336 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11455,6 +11455,7 @@ dependencies = [ "tracing", "vortex-array", "vortex-buffer", + "vortex-decimal-byte-parts", "vortex-error", "vortex-fsst", "vortex-mask", diff --git a/encodings/decimal-byte-parts/Cargo.toml b/encodings/decimal-byte-parts/Cargo.toml index dcccabf001a..64fb492fa6d 100644 --- a/encodings/decimal-byte-parts/Cargo.toml +++ b/encodings/decimal-byte-parts/Cargo.toml @@ -37,6 +37,10 @@ vortex-bench-support = { workspace = true } name = "dbp_assemble" harness = false +[[bench]] +name = "dbp_compute" +harness = false + [[bench]] name = "dbp_split" harness = false diff --git a/encodings/decimal-byte-parts/benches/dbp_compute.rs b/encodings/decimal-byte-parts/benches/dbp_compute.rs new file mode 100644 index 00000000000..f206f2eb505 --- /dev/null +++ b/encodings/decimal-byte-parts/benches/dbp_compute.rs @@ -0,0 +1,317 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Decimal compute on DecimalByteParts arrays whose MSP holds the whole value. +//! +//! Every case has no lower parts. Most cases store the values in an MSP narrower than the +//! canonical width of the dtype, and `msp_i64_p18_canonical` is the control whose MSP has the +//! canonical width. Sum and min-max read the decoded decimal, and the filter filters the MSP and +//! then decodes the result, so their cost follows the width that decoding produces. Comparisons +//! against a constant run on the MSP. + +use std::fmt; +use std::sync::LazyLock; + +use divan::Bencher; +use mimalloc::MiMalloc; +use rand::RngExt; +use rand::SeedableRng; +use rand::distr::uniform::SampleRange; +use rand::distr::uniform::SampleUniform; +use rand::rngs::StdRng; +use vortex_array::ArrayRef; +use vortex_array::Canonical; +use vortex_array::ExecutionCtx; +use vortex_array::IntoArray; +use vortex_array::VortexSessionExecute; +use vortex_array::aggregate_fn::NumericalAggregateOpts; +use vortex_array::aggregate_fn::fns::min_max; +use vortex_array::aggregate_fn::fns::sum; +use vortex_array::arrays::BoolArray; +use vortex_array::arrays::ConstantArray; +use vortex_array::arrays::DecimalArray; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::builtins::ArrayBuiltins; +use vortex_array::dtype::DecimalDType; +use vortex_array::dtype::NativePType; +use vortex_array::dtype::Nullability; +use vortex_array::dtype::PType; +use vortex_array::scalar::DecimalValue; +use vortex_array::scalar::Scalar; +use vortex_array::scalar_fn::fns::operators::Operator; +use vortex_array::validity::Validity; +use vortex_buffer::BitBuffer; +use vortex_buffer::Buffer; +use vortex_buffer::ByteBuffer; +use vortex_decimal_byte_parts::DecimalByteParts; +use vortex_error::VortexExpect; +use vortex_mask::Mask; +use vortex_session::VortexSession; + +#[global_allocator] +static GLOBAL: MiMalloc = MiMalloc; + +fn main() { + LazyLock::force(&SESSION); + divan::main(); +} + +static SESSION: LazyLock = LazyLock::new(|| { + let session = VortexSession::empty(); + vortex_decimal_byte_parts::initialize(&session); + session +}); + +// Sized to keep CodSpeed simulation under 1ms per benchmark. +const LEN: usize = 8_192; + +/// The largest magnitude of a `Decimal(18, _)` value. +const MAX_P18: i64 = 999_999_999_999_999_999; + +#[derive(Clone, Copy)] +enum Case { + /// Values that fit an `i32` under a dtype whose canonical width is `i64`. The compressor + /// writes this shape when it narrows such a column. + MspI32P18, + /// [`Case::MspI32P18`] with about a tenth of the rows null and zeros in the null slots. + MspI32P18Nullable, + /// Every value fits an `i16` but needs precision 5, above the precision 4 that `i16` holds. + GapMspI16P10, + /// Values that fit an `i64` under a dtype whose canonical width is `i128`. + MspI64P38, + /// The control: the MSP already has the canonical width. + MspI64P18Canonical, +} + +const CASES: [Case; 5] = [ + Case::MspI32P18, + Case::MspI32P18Nullable, + Case::GapMspI16P10, + Case::MspI64P38, + Case::MspI64P18Canonical, +]; + +impl Case { + fn decimal_dtype(self) -> DecimalDType { + match self { + Self::MspI32P18 | Self::MspI32P18Nullable | Self::MspI64P18Canonical => { + DecimalDType::new(18, 2) + } + Self::GapMspI16P10 => DecimalDType::new(10, 2), + Self::MspI64P38 => DecimalDType::new(38, 2), + } + } + + fn input(self) -> Input { + let mut rng = StdRng::seed_from_u64(42); + match self { + Self::MspI32P18 => Input::non_nullable( + random_values(&mut rng, -999_999_999i32..=999_999_999), + self.decimal_dtype(), + ), + Self::MspI32P18Nullable => { + let validity: BitBuffer = (0..LEN).map(|_| rng.random_bool(0.9)).collect(); + let values: Buffer = validity + .iter() + .map(|valid| { + if valid { + rng.random_range(-999_999_999..=999_999_999) + } else { + 0 + } + }) + .collect(); + Input { + msp_values: values.into_byte_buffer(), + msp_ptype: PType::I32, + msp_validity: Some(validity), + decimal_dtype: self.decimal_dtype(), + } + } + Self::GapMspI16P10 => Input::non_nullable( + random_values(&mut rng, 10_000i16..=i16::MAX), + self.decimal_dtype(), + ), + Self::MspI64P38 | Self::MspI64P18Canonical => Input::non_nullable( + random_values(&mut rng, -MAX_P18..=MAX_P18), + self.decimal_dtype(), + ), + } + } + + /// A constant that the MSP can hold, near the middle of the values. + fn constant_in_msp_range(self) -> DecimalValue { + match self { + Self::MspI32P18 | Self::MspI32P18Nullable | Self::MspI64P18Canonical => { + DecimalValue::I64(0) + } + Self::GapMspI16P10 => DecimalValue::I64(21_383), + Self::MspI64P38 => DecimalValue::I128(0), + } + } + + /// A constant above every value that the MSP can hold, for the cases where the DBP compare + /// kernel answers without reading values. + /// + /// The kernel answers this way only when every row is valid, so the nullable case has no + /// such constant. Every `Decimal(18, _)` value fits an `i64`, so the canonical case has none + /// either. + fn constant_above_msp_range(self) -> Option { + match self { + Self::MspI32P18 => Some(DecimalValue::I64(i64::from(i32::MAX) + 1)), + Self::GapMspI16P10 => Some(DecimalValue::I64(i64::from(i16::MAX) + 1)), + Self::MspI64P38 => Some(DecimalValue::I128(i128::from(i64::MAX) + 1)), + Self::MspI32P18Nullable | Self::MspI64P18Canonical => None, + } + } + + fn constant(self, value: DecimalValue) -> ArrayRef { + let scalar = Scalar::decimal(value, self.decimal_dtype(), Nullability::NonNullable); + ConstantArray::new(scalar, LEN).into_array() + } +} + +impl fmt::Display for Case { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::MspI32P18 => "msp_i32_p18", + Self::MspI32P18Nullable => "msp_i32_p18_nullable", + Self::GapMspI16P10 => "gap_msp_i16_p10", + Self::MspI64P38 => "msp_i64_p38", + Self::MspI64P18Canonical => "msp_i64_p18_canonical", + }) + } +} + +fn out_of_range_cases() -> Vec { + CASES + .into_iter() + .filter(|case| case.constant_above_msp_range().is_some()) + .collect() +} + +/// The parts of one benchmark input. +/// +/// `sum` and `min_max` cache their result as a statistic on the array they read. Each iteration +/// therefore builds a new array from these parts, which shares the buffers but not the +/// statistics. +struct Input { + msp_values: ByteBuffer, + msp_ptype: PType, + msp_validity: Option, + decimal_dtype: DecimalDType, +} + +impl Input { + fn non_nullable(values: Buffer, decimal_dtype: DecimalDType) -> Self { + Self { + msp_values: values.into_byte_buffer(), + msp_ptype: T::PTYPE, + msp_validity: None, + decimal_dtype, + } + } + + fn array(&self) -> ArrayRef { + let validity = self + .msp_validity + .clone() + .map_or(Validity::NonNullable, Validity::from); + let msp = + PrimitiveArray::from_byte_buffer(self.msp_values.clone(), self.msp_ptype, validity); + DecimalByteParts::try_new(msp.into_array(), self.decimal_dtype) + .vortex_expect("the MSP of every benchmark input is a signed integer array") + .into_array() + } +} + +fn random_values(rng: &mut StdRng, range: R) -> Buffer +where + T: NativePType + SampleUniform, + R: SampleRange + Clone, +{ + (0..LEN).map(|_| rng.random_range(range.clone())).collect() +} + +/// Runs `operation` on a new array of `case` in each iteration. +fn bench_case( + bencher: Bencher, + case: Case, + operation: impl Fn(ArrayRef, &mut ExecutionCtx) -> O + Sync, +) { + let input = case.input(); + bencher + .with_inputs(|| (input.array(), SESSION.create_execution_ctx())) + .bench_values(|(array, mut ctx)| operation(array, &mut ctx)); +} + +#[divan::bench(args = CASES)] +fn decode(bencher: Bencher, case: Case) { + bench_case(bencher, case, |array, ctx| { + array + .execute::(ctx) + .vortex_expect("decode DecimalByteParts") + }); +} + +/// Compares a new array of `case` with `Gte` against a constant of `value` in each iteration. +fn bench_compare(bencher: Bencher, case: Case, value: DecimalValue) { + let constant = case.constant(value); + bench_case(bencher, case, |array, ctx| { + array + .binary(constant.clone(), Operator::Gte) + .vortex_expect("compare DecimalByteParts") + .execute::(ctx) + .vortex_expect("execute the comparison") + }); +} + +#[divan::bench(args = CASES)] +fn compare_constant(bencher: Bencher, case: Case) { + bench_compare(bencher, case, case.constant_in_msp_range()); +} + +#[divan::bench(args = out_of_range_cases())] +fn compare_constant_out_of_range(bencher: Bencher, case: Case) { + let value = case + .constant_above_msp_range() + .vortex_expect("out_of_range_cases keeps only cases with a constant above the MSP range"); + bench_compare(bencher, case, value); +} + +#[divan::bench(args = CASES)] +fn filter(bencher: Bencher, case: Case) { + let input = case.input(); + let mut rng = StdRng::seed_from_u64(7); + let selection: BitBuffer = (0..LEN).map(|_| rng.random_bool(0.5)).collect(); + bencher + .with_inputs(|| { + ( + input.array(), + Mask::from_buffer(selection.clone()), + SESSION.create_execution_ctx(), + ) + }) + .bench_values(|(array, mask, mut ctx)| { + array + .filter(mask) + .vortex_expect("filter DecimalByteParts") + .execute::(&mut ctx) + .vortex_expect("execute the filter") + }); +} + +#[divan::bench(args = CASES)] +fn sum(bencher: Bencher, case: Case) { + bench_case(bencher, case, |array, ctx| { + sum::sum(&array, ctx).vortex_expect("sum DecimalByteParts") + }); +} + +#[divan::bench(args = CASES)] +fn min_max(bencher: Bencher, case: Case) { + bench_case(bencher, case, |array, ctx| { + min_max::min_max(&array, ctx, NumericalAggregateOpts::default()) + .vortex_expect("min-max DecimalByteParts") + }); +} diff --git a/vortex-array/benches/cast_decimal.rs b/vortex-array/benches/cast_decimal.rs index 79c215c2f40..58122a586dd 100644 --- a/vortex-array/benches/cast_decimal.rs +++ b/vortex-array/benches/cast_decimal.rs @@ -7,7 +7,8 @@ //! masked kernel (`try_map_masked_into`) instead of the dense one (`try_map_into`). //! - **work**: an *in-place* cast that only widens precision at the same scale, so the values //! buffer is reused untouched (`O(1)`), vs a *copy* cast that must allocate a new buffer and -//! rescan every value (`O(n)`). +//! rescan every value (`O(n)`), vs a *widening* cast whose target precision has a wider +//! canonical width than the input. #![expect(clippy::unwrap_used)] @@ -22,8 +23,10 @@ use vortex_array::IntoArray; use vortex_array::VortexSessionExecute; use vortex_array::arrays::DecimalArray; use vortex_array::builtins::ArrayBuiltins; +use vortex_array::dtype::BigCast; use vortex_array::dtype::DType; use vortex_array::dtype::DecimalDType; +use vortex_array::dtype::NativeDecimalType; use vortex_array::dtype::Nullability; use vortex_array::optimizer::kernels::KernelSession; use vortex_array::validity::Validity; @@ -48,12 +51,17 @@ const SIZES: &[usize] = &[16_384]; static SESSION: LazyLock = LazyLock::new(|| VortexSession::empty().with::()); -/// Builds an `i64`-backed `Decimal(precision, scale)` array of `n` values that all fit in -/// precision 10, optionally with ~50% nulls. -fn decimal_array(n: usize, precision: u8, scale: i8, nullable: bool) -> ArrayRef { +/// Builds a `T`-backed `Decimal(precision, scale)` array of `n` values that are all below +/// `10^9`, optionally with ~50% nulls. +fn decimal_array( + n: usize, + precision: u8, + scale: i8, + nullable: bool, +) -> ArrayRef { let mut rng = StdRng::seed_from_u64(42); - let values: BufferMut = (0..n) - .map(|_| rng.random_range(0..1_000_000_000i64)) + let values: BufferMut = (0..n) + .map(|_| ::from(rng.random_range(0..1_000_000_000i64)).unwrap()) .collect(); let validity = if nullable { Validity::from_iter((0..n).map(|_| rng.random_bool(0.5))) @@ -82,7 +90,7 @@ fn bench_cast(bencher: Bencher, array: ArrayRef, target: DType) { fn in_place_non_nullable(bencher: Bencher, n: usize) { bench_cast( bencher, - decimal_array(n, 10, 2, false), + decimal_array::(n, 10, 2, false), DType::Decimal(DecimalDType::new(18, 2), Nullability::NonNullable), ); } @@ -91,7 +99,7 @@ fn in_place_non_nullable(bencher: Bencher, n: usize) { fn in_place_nullable(bencher: Bencher, n: usize) { bench_cast( bencher, - decimal_array(n, 10, 2, true), + decimal_array::(n, 10, 2, true), DType::Decimal(DecimalDType::new(18, 2), Nullability::Nullable), ); } @@ -103,7 +111,7 @@ fn in_place_nullable(bencher: Bencher, n: usize) { fn copy_non_nullable(bencher: Bencher, n: usize) { bench_cast( bencher, - decimal_array(n, 18, 2, false), + decimal_array::(n, 18, 2, false), DType::Decimal(DecimalDType::new(10, 2), Nullability::NonNullable), ); } @@ -112,7 +120,28 @@ fn copy_non_nullable(bencher: Bencher, n: usize) { fn copy_nullable(bencher: Bencher, n: usize) { bench_cast( bencher, - decimal_array(n, 18, 2, true), + decimal_array::(n, 18, 2, true), DType::Decimal(DecimalDType::new(10, 2), Nullability::Nullable), ); } + +// Widening: the target precision has a wider canonical width than the input, so every value is +// copied into a buffer of the wider type. + +#[divan::bench(args = SIZES)] +fn widen_non_nullable(bencher: Bencher, n: usize) { + bench_cast( + bencher, + decimal_array::(n, 9, 2, false), + DType::Decimal(DecimalDType::new(18, 2), Nullability::NonNullable), + ); +} + +#[divan::bench(args = SIZES)] +fn widen_nullable(bencher: Bencher, n: usize) { + bench_cast( + bencher, + decimal_array::(n, 9, 2, true), + DType::Decimal(DecimalDType::new(18, 2), Nullability::Nullable), + ); +} diff --git a/vortex-arrow/Cargo.toml b/vortex-arrow/Cargo.toml index 8605b37a3f0..7ee40444e84 100644 --- a/vortex-arrow/Cargo.toml +++ b/vortex-arrow/Cargo.toml @@ -41,6 +41,7 @@ divan = { workspace = true } mimalloc = { workspace = true } rstest = { workspace = true } vortex-array = { workspace = true, features = ["_test-harness"] } +vortex-decimal-byte-parts = { workspace = true } vortex-fsst = { workspace = true } vortex-onpair = { workspace = true } vortex-zstd = { workspace = true } diff --git a/vortex-arrow/benches/to_arrow.rs b/vortex-arrow/benches/to_arrow.rs index 0d06847ac16..ae773ce70d7 100644 --- a/vortex-arrow/benches/to_arrow.rs +++ b/vortex-arrow/benches/to_arrow.rs @@ -37,6 +37,7 @@ use vortex_array::memory::MemorySession; use vortex_array::optimizer::kernels::KernelSession; use vortex_array::session::ArraySession; use vortex_array::session::ArraySessionExt; +use vortex_array::validity::Validity; #[expect( deprecated, reason = "benchmark comparing deprecated method with new one" @@ -44,6 +45,8 @@ use vortex_array::session::ArraySessionExt; use vortex_arrow::ArrowArrayExecutor; use vortex_arrow::ArrowSession; use vortex_arrow::ArrowSessionExt; +use vortex_buffer::Buffer; +use vortex_decimal_byte_parts::DecimalByteParts; use vortex_fsst::fsst_compress; use vortex_fsst::fsst_train_compressor; use vortex_onpair::DEFAULT_CONFIG; @@ -66,6 +69,7 @@ static SESSION: LazyLock = LazyLock::new(|| { .with::() .with::() .with::(); + vortex_decimal_byte_parts::initialize(&session); vortex_fsst::initialize(&session); vortex_onpair::initialize(&session); session.arrays().register(Zstd); @@ -446,6 +450,140 @@ fn append_to_view_builder(bencher: Bencher, case: StringCase) { }); } +#[derive(Clone, Copy)] +enum DecimalExportCase { + CanonicalI64P18, + /// The values of `CanonicalI64P18` under a dtype whose canonical width is `i128`. + CanonicalI128P38, + MspI32P18, + /// `MspI32P18` with every third row null and zeros in the null slots. + MspI32P18Nullable, + /// Every value fits an `i16` but needs precision 5, above the precision 4 that `i16` holds. + GapMspI16P10, +} + +impl DecimalExportCase { + fn decimal_dtype(self) -> DecimalDType { + match self { + Self::CanonicalI64P18 | Self::MspI32P18 | Self::MspI32P18Nullable => { + DecimalDType::new(18, 2) + } + Self::CanonicalI128P38 => DecimalDType::new(38, 2), + Self::GapMspI16P10 => DecimalDType::new(10, 2), + } + } + + fn array(self) -> ArrayRef { + let decimal_dtype = self.decimal_dtype(); + match self { + Self::CanonicalI64P18 => DecimalArray::new( + spread_values::(-MAX_P18, MAX_P18), + decimal_dtype, + Validity::NonNullable, + ) + .into_array(), + Self::CanonicalI128P38 => DecimalArray::new( + spread_values::(-MAX_P18, MAX_P18), + decimal_dtype, + Validity::NonNullable, + ) + .into_array(), + Self::MspI32P18 => decimal_byte_parts( + PrimitiveArray::new( + spread_values::(-999_999_999, 999_999_999), + Validity::NonNullable, + ), + decimal_dtype, + ), + Self::MspI32P18Nullable => { + let validity = Validity::from_iter((0..DECIMAL_ROWS).map(|index| index % 3 != 0)); + let values: Buffer = spread_values::(-999_999_999, 999_999_999) + .iter() + .enumerate() + .map(|(index, &value)| if index % 3 == 0 { 0 } else { value }) + .collect(); + decimal_byte_parts(PrimitiveArray::new(values, validity), decimal_dtype) + } + Self::GapMspI16P10 => decimal_byte_parts( + PrimitiveArray::new( + spread_values::(10_000, i64::from(i16::MAX)), + Validity::NonNullable, + ), + decimal_dtype, + ), + } + } +} + +impl Display for DecimalExportCase { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::CanonicalI64P18 => "canonical_i64_p18", + Self::CanonicalI128P38 => "canonical_i128_p38", + Self::MspI32P18 => "msp_i32_p18", + Self::MspI32P18Nullable => "msp_i32_p18_nullable", + Self::GapMspI16P10 => "gap_msp_i16_p10", + }) + } +} + +const DECIMAL_EXPORT_CASES: &[DecimalExportCase] = &[ + DecimalExportCase::CanonicalI64P18, + DecimalExportCase::CanonicalI128P38, + DecimalExportCase::MspI32P18, + DecimalExportCase::MspI32P18Nullable, + DecimalExportCase::GapMspI16P10, +]; + +const DECIMAL_ROWS: usize = 100_000; + +/// The largest magnitude of a `Decimal(18, _)` value. +const MAX_P18: i64 = 999_999_999_999_999_999; + +/// `DECIMAL_ROWS` values spread across `min..=max` in an unsorted, fixed order. +/// +/// Row `i` holds `min + (i * STEP) % (max - min + 1)`. `STEP` is the prime `2^61 - 1`, so it is +/// coprime to every range size below it: consecutive rows land far apart, and the values repeat +/// only after every value of the range appears. +fn spread_values>(min: i64, max: i64) -> Buffer { + const STEP: i128 = (1 << 61) - 1; + let count = i128::from(max) - i128::from(min) + 1; + (0..DECIMAL_ROWS) + .map(|index| { + let offset = (i128::try_from(index).unwrap() * STEP) % count; + let value = i64::try_from(i128::from(min) + offset).unwrap(); + T::try_from(value).ok().unwrap() + }) + .collect() +} + +fn decimal_byte_parts(msp: PrimitiveArray, decimal_dtype: DecimalDType) -> ArrayRef { + DecimalByteParts::try_new(msp.into_array(), decimal_dtype) + .unwrap() + .into_array() +} + +/// Measures export of one decimal array to an Arrow `Decimal128` array. +#[divan::bench(args = DECIMAL_EXPORT_CASES)] +fn decimal_export(bencher: Bencher, case: DecimalExportCase) { + let array = case.array(); + let decimal_dtype = case.decimal_dtype(); + let field = Field::new( + "value", + DataType::Decimal128(decimal_dtype.precision(), decimal_dtype.scale()), + array.dtype().is_nullable(), + ); + bencher + .with_inputs(|| (array.clone(), SESSION.create_execution_ctx())) + .input_counter(|(array, _)| ItemsCount::new(array.len())) + .bench_values(|(array, mut ctx)| { + SESSION + .arrow() + .execute_arrow(array, Some(&field), &mut ctx) + .unwrap() + }); +} + #[divan::bench] fn to_arrow_array(bencher: Bencher) { bencher diff --git a/vortex-btrblocks/benches/compress.rs b/vortex-btrblocks/benches/compress.rs index 5bc2eefbaba..27f9d9c9185 100644 --- a/vortex-btrblocks/benches/compress.rs +++ b/vortex-btrblocks/benches/compress.rs @@ -13,19 +13,29 @@ mod benchmarks { use divan::counter::BytesCount; use divan::counter::ItemsCount; use rand::Rng; + use rand::RngExt; use rand::SeedableRng; use rand::prelude::StdRng; use vortex_array::ArrayRef; use vortex_array::IntoArray; use vortex_array::VortexSessionExecute; use vortex_array::aggregate_fn::session::AggregateFnSession; + use vortex_array::arrays::DecimalArray; use vortex_array::arrays::PrimitiveArray; + use vortex_array::dtype::DecimalDType; use vortex_array::memory::MemorySession; use vortex_array::optimizer::kernels::KernelSession; use vortex_array::session::ArraySession; + use vortex_array::validity::Validity; use vortex_btrblocks::BtrBlocksCompressorBuilder; use vortex_btrblocks::CompressionSession; + use vortex_buffer::Buffer; use vortex_buffer::buffer_mut; + use vortex_edition::EDITION_DECLARATIONS; + use vortex_edition::EDITION_FAMILIES; + use vortex_edition::EditionSession; + use vortex_edition::EditionSessionExt; + use vortex_edition::declarations::core::CORE_2026_08_3; use vortex_session::VortexSession; use vortex_utils::aliases::hash_set::HashSet; @@ -38,6 +48,30 @@ mod benchmarks { .with::() }); + /// A session that allows only what a default writer may produce, set up as the golden tests + /// set it up: the encodings the compressor emits, limited by the default `core` edition. + static WRITE_SESSION: LazyLock = LazyLock::new(|| { + let session = VortexSession::empty().with::(); + vortex_alp::initialize(&session); + vortex_datetime_parts::initialize(&session); + vortex_decimal_byte_parts::initialize(&session); + vortex_fastlanes::initialize(&session); + vortex_fsst::initialize(&session); + vortex_onpair::initialize(&session); + vortex_runend::initialize(&session); + vortex_sequence::initialize(&session); + vortex_sparse::initialize(&session); + vortex_zigzag::initialize(&session); + for family in EDITION_FAMILIES { + session.editions().declare_family(family).unwrap(); + } + for declaration in EDITION_DECLARATIONS { + session.register_edition(declaration).unwrap(); + } + session.enable_edition(CORE_2026_08_3).unwrap(); + session + }); + fn make_clickbench_window_name() -> ArrayRef { // A test that's meant to mirror the WindowName column from ClickBench. let mut values = buffer_mut![-1i32; 65_536]; @@ -76,6 +110,35 @@ mod benchmarks { .unwrap() }); } + + /// Compresses a canonical `Decimal(12, 2)` column whose values are drawn like those of the + /// golden `decimal_prices` column. + #[divan::bench] + fn btrblocks_decimal(bencher: Bencher) { + let mut rng = StdRng::seed_from_u64(115); + let values: Buffer = (0..65_536) + .map(|_| rng.random_range(0..10_000_000i64)) + .collect(); + // NB: the default `core` edition does not permit `vortex.decimal_byte_parts.v2`, so the + // decimal scheme refines to its v1 variant, as in a default write. An unrestricted + // compressor would refine to v2 instead. + let compressor = BtrBlocksCompressorBuilder::from_session(&WRITE_SESSION).build(); + bencher + .with_inputs(|| { + ( + DecimalArray::new( + values.clone(), + DecimalDType::new(12, 2), + Validity::NonNullable, + ) + .into_array(), + WRITE_SESSION.create_execution_ctx(), + ) + }) + .input_counter(|(array, _)| ItemsCount::new(array.len())) + .input_counter(|(array, _)| BytesCount::of_many::(array.len())) + .bench_refs(|(array, ctx)| compressor.compress(array, ctx).unwrap()); + } } #[global_allocator] diff --git a/vortex-duckdb/Cargo.toml b/vortex-duckdb/Cargo.toml index 38a9799a41b..5baeef1565d 100644 --- a/vortex-duckdb/Cargo.toml +++ b/vortex-duckdb/Cargo.toml @@ -66,6 +66,10 @@ wkb = { workspace = true } name = "bool_export" harness = false +[[bench]] +name = "decimal_export" +harness = false + [lints] workspace = true diff --git a/vortex-duckdb/benches/decimal_export.rs b/vortex-duckdb/benches/decimal_export.rs new file mode 100644 index 00000000000..aa9e02757cf --- /dev/null +++ b/vortex-duckdb/benches/decimal_export.rs @@ -0,0 +1,178 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Export of one decimal column into DuckDB data chunks. +//! +//! DuckDB stores a decimal in the width that its precision needs, with `i16` as the narrowest +//! width. A canonical array whose width matches DuckDB exports without a copy, and any other +//! array copies its values into the DuckDB vector. Each iteration builds the exporter, which +//! decodes the column, and then exports until the exporter is exhausted. + +use std::fmt; +use std::sync::LazyLock; + +use divan::Bencher; +use divan::counter::ItemsCount; +use mimalloc::MiMalloc; +use vortex::VortexSessionDefault; +use vortex::array::ArrayRef; +use vortex::array::IntoArray; +use vortex::array::VortexSessionExecute; +use vortex::array::arrays::DecimalArray; +use vortex::array::arrays::PrimitiveArray; +use vortex::array::arrays::StructArray; +use vortex::array::validity::Validity; +use vortex::buffer::Buffer; +use vortex::dtype::DecimalDType; +use vortex::dtype::NativeDecimalType; +use vortex::dtype::NativePType; +use vortex::encodings::decimal_byte_parts::DecimalByteParts; +use vortex::error::VortexExpect; +use vortex::session::VortexSession; +use vortex_duckdb::_benchmarking::ArrayExporter; +use vortex_duckdb::_benchmarking::ConversionCache; +use vortex_duckdb::duckdb::DataChunk; +use vortex_duckdb::duckdb::LogicalType; + +#[global_allocator] +static GLOBAL: MiMalloc = MiMalloc; + +fn main() { + LazyLock::force(&SESSION); + divan::main(); +} + +static SESSION: LazyLock = LazyLock::new(VortexSession::default); + +const ROWS: usize = 100_000; + +#[derive(Clone, Copy)] +enum Case { + /// Exports without a copy. + CanonicalI64P18, + /// Exports without a copy. + CanonicalI16P4, + /// Copies, because DuckDB stores precision 2 as `i16`. + CanonicalI8P2, + /// A DecimalByteParts array with an `i32` MSP under `Decimal(18, 2)`. + MspI32P18, + /// A DecimalByteParts array under `Decimal(10, 2)` whose values all fit an `i16` but need + /// precision 5, above the precision 4 that `i16` holds. + GapMspI16P10, +} + +const CASES: &[Case] = &[ + Case::CanonicalI64P18, + Case::CanonicalI16P4, + Case::CanonicalI8P2, + Case::MspI32P18, + Case::GapMspI16P10, +]; + +impl Case { + fn decimal_dtype(self) -> DecimalDType { + match self { + Self::CanonicalI64P18 | Self::MspI32P18 => DecimalDType::new(18, 2), + Self::CanonicalI16P4 => DecimalDType::new(4, 2), + Self::CanonicalI8P2 => DecimalDType::new(2, 1), + Self::GapMspI16P10 => DecimalDType::new(10, 2), + } + } + + fn column(self) -> ArrayRef { + let decimal_dtype = self.decimal_dtype(); + match self { + Self::CanonicalI64P18 => { + canonical(spread_values::(-MAX_P18, MAX_P18), decimal_dtype) + } + Self::CanonicalI16P4 => canonical(spread_values::(-9_999, 9_999), decimal_dtype), + Self::CanonicalI8P2 => canonical(spread_values::(-99, 99), decimal_dtype), + Self::MspI32P18 => decimal_byte_parts( + spread_values::(-999_999_999, 999_999_999), + decimal_dtype, + ), + Self::GapMspI16P10 => decimal_byte_parts( + spread_values::(10_000, i64::from(i16::MAX)), + decimal_dtype, + ), + } + } +} + +impl fmt::Display for Case { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::CanonicalI64P18 => "canonical_i64_p18", + Self::CanonicalI16P4 => "canonical_i16_p4", + Self::CanonicalI8P2 => "canonical_i8_p2", + Self::MspI32P18 => "msp_i32_p18", + Self::GapMspI16P10 => "gap_msp_i16_p10", + }) + } +} + +/// The largest magnitude of a `Decimal(18, _)` value. +const MAX_P18: i64 = 999_999_999_999_999_999; + +/// `ROWS` values spread across `min..=max` in an unsorted, fixed order. +/// +/// Row `i` holds `min + (i * STEP) % (max - min + 1)`. `STEP` is the prime `2^61 - 1`, so it is +/// coprime to every range size below it: consecutive rows land far apart, and the values repeat +/// only after every value of the range appears. +fn spread_values>(min: i64, max: i64) -> Buffer { + const STEP: i128 = (1 << 61) - 1; + let count = i128::from(max) - i128::from(min) + 1; + (0..ROWS) + .map(|index| { + let index = i128::try_from(index).vortex_expect("the row count fits an i128"); + let value = i64::try_from(i128::from(min) + (index * STEP) % count) + .vortex_expect("every value lies in min..=max"); + T::try_from(value) + .ok() + .vortex_expect("every value fits the type of its case") + }) + .collect() +} + +fn canonical(values: Buffer, decimal_dtype: DecimalDType) -> ArrayRef { + DecimalArray::new(values, decimal_dtype, Validity::NonNullable).into_array() +} + +fn decimal_byte_parts(msp: Buffer, decimal_dtype: DecimalDType) -> ArrayRef { + let msp = PrimitiveArray::new(msp, Validity::NonNullable).into_array(); + DecimalByteParts::try_new(msp, decimal_dtype) + .vortex_expect("the MSP of every benchmark input is a signed integer array") + .into_array() +} + +#[divan::bench(args = CASES)] +fn decimal_export(bencher: Bencher, case: Case) { + let decimal_dtype = case.decimal_dtype(); + let scale = + u8::try_from(decimal_dtype.scale()).vortex_expect("every case has a non-negative scale"); + let logical_type = LogicalType::decimal_type(decimal_dtype.precision(), scale) + .vortex_expect("DuckDB supports the precision of every case"); + let array = StructArray::from_fields(&[("value", case.column())]) + .vortex_expect("build the exported struct"); + let cache = ConversionCache::default(); + + bencher + .with_inputs(|| { + ( + DataChunk::new([logical_type.clone()]), + SESSION.create_execution_ctx(), + ) + }) + .input_counter(|_| ItemsCount::new(ROWS)) + .bench_values(|(mut chunk, ctx)| { + let mut exporter = + ArrayExporter::try_new(&array, &cache, ctx).vortex_expect("build the exporter"); + while exporter + .export(&mut chunk, None) + .vortex_expect("export a chunk") + {} + // Returning the exporter keeps the drop of a copied column's decoded buffer out of the + // timing, as it is for the columns that export without a copy. + (exporter, chunk) + }); +} diff --git a/vortex-duckdb/src/exporter/cache.rs b/vortex-duckdb/src/exporter/cache.rs index ec133dcc326..56f9afc1c78 100644 --- a/vortex-duckdb/src/exporter/cache.rs +++ b/vortex-duckdb/src/exporter/cache.rs @@ -17,9 +17,9 @@ use crate::duckdb::Vector; /// to avoid redundant conversions of the same array instances. /// We hold on to the `ArrayRef` to ensure that the key (ptr addr) doesn't get reused. pub struct ConversionCache { - pub dict_cache: DashMap, - pub values_cache: DashMap>)>, - pub canonical_cache: DashMap, + pub(crate) dict_cache: DashMap, + pub(crate) values_cache: DashMap>)>, + pub(crate) canonical_cache: DashMap, } impl Default for ConversionCache { diff --git a/vortex-duckdb/src/lib.rs b/vortex-duckdb/src/lib.rs index 1209213135f..bef68dfdf9d 100644 --- a/vortex-duckdb/src/lib.rs +++ b/vortex-duckdb/src/lib.rs @@ -43,6 +43,13 @@ mod copy; #[cfg(test)] mod e2e_test; +/// The exporter internals that the benchmarks in `benches/` build on. This is not a stable API. +#[doc(hidden)] +pub mod _benchmarking { + pub use crate::exporter::ArrayExporter; + pub use crate::exporter::ConversionCache; +} + // A global runtime for Vortex operations within DuckDB. static RUNTIME: LazyLock = LazyLock::new(CurrentThreadRuntime::new); /// Process-wide registry, so repeated scans against the same bucket share one client. diff --git a/vortex/benches/common_encoding_tree_throughput.rs b/vortex/benches/common_encoding_tree_throughput.rs index 0d523c133e0..7b18d2a78bd 100644 --- a/vortex/benches/common_encoding_tree_throughput.rs +++ b/vortex/benches/common_encoding_tree_throughput.rs @@ -31,6 +31,7 @@ use vortex::array::optimizer::kernels::KernelSession; use vortex::array::scalar::Scalar; use vortex::array::session::ArraySession; use vortex::dtype::DType; +use vortex::dtype::DecimalDType; use vortex::dtype::PType; use vortex::encodings::alp; use vortex::encodings::alp::ALP; @@ -40,6 +41,8 @@ use vortex::encodings::alp::alp_encode; use vortex::encodings::datetime_parts; use vortex::encodings::datetime_parts::DateTimeParts; use vortex::encodings::datetime_parts::split_temporal; +use vortex::encodings::decimal_byte_parts; +use vortex::encodings::decimal_byte_parts::DecimalByteParts; use vortex::encodings::fastlanes; use vortex::encodings::fastlanes::BitPacked; use vortex::encodings::fastlanes::FoR; @@ -77,6 +80,7 @@ static SESSION: LazyLock = LazyLock::new(|| { .with::(); alp::initialize(&session); datetime_parts::initialize(&session); + decimal_byte_parts::initialize(&session); fastlanes::initialize(&session); fsst::initialize(&session); runend::initialize(&session); @@ -431,6 +435,22 @@ mod setup { .unwrap() .into_array() } + + /// Create DecimalByteParts <- BitPacked encoding tree, the tree that the compressor builds for + /// the golden `decimal_prices` column. + pub fn dbp_bp_i32() -> ArrayRef { + let mut rng = StdRng::seed_from_u64(115); + let msp = + PrimitiveArray::from_iter((0..NUM_VALUES).map(|_| rng.random_range(0..10_000_000i32))) + .into_array(); + + let mut ctx = SESSION.create_execution_ctx(); + let msp_bp = BitPacked::encode(&msp, 24, &mut ctx).unwrap(); + + DecimalByteParts::try_new(msp_bp.into_array(), DecimalDType::new(12, 2)) + .unwrap() + .into_array() + } } // Complex encoding tree benchmarks @@ -480,6 +500,7 @@ macro_rules! setup_fn { setup_fn!(setup::dict_fsst_varbin_string), setup_fn!(setup::dict_fsst_varbin_bp_string), setup_fn!(setup::datetime_for_bp), + setup_fn!(setup::dbp_bp_i32), ] )] fn decompress(bencher: Bencher, setup_fn: SetupFn) {