diff --git a/.github/workflows/bench-pr.yml b/.github/workflows/bench-pr.yml index 12137726122..c49c3b212bf 100644 --- a/.github/workflows/bench-pr.yml +++ b/.github/workflows/bench-pr.yml @@ -61,7 +61,7 @@ jobs: env: RUSTFLAGS: "-C target-cpu=native" run: | - cargo build --package ${{ matrix.benchmark.id }} --profile release_debug ${{ matrix.benchmark.build_args }} --features unstable_encodings + cargo build --package ${{ matrix.benchmark.id }} --profile release_debug ${{ matrix.benchmark.build_args }} - name: Setup Polar Signals if: github.event.pull_request.head.repo.fork == false diff --git a/.github/workflows/sql-benchmarks.yml b/.github/workflows/sql-benchmarks.yml index 243db83aff8..b71959c12e8 100644 --- a/.github/workflows/sql-benchmarks.yml +++ b/.github/workflows/sql-benchmarks.yml @@ -576,10 +576,12 @@ jobs: RUSTFLAGS: "-C target-cpu=native" run: | packages=(--bin data-gen --bin datafusion-bench --bin duckdb-bench) + features=() if [ "${{ inputs.mode }}" != "pr" ]; then packages+=(--bin lance-bench) + features+=(--features unstable_encodings) fi - cargo build "${packages[@]}" --profile release_debug --features unstable_encodings + cargo build "${packages[@]}" "${features[@]}" --profile release_debug - name: Generate data shell: bash diff --git a/Cargo.lock b/Cargo.lock index 871b73a08c2..093ee5cf58b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9823,6 +9823,7 @@ dependencies = [ "vortex-error", "vortex-fastlanes", "vortex-fsst", + "vortex-mask", "vortex-onpair", "vortex-pco", "vortex-runend", diff --git a/vortex-btrblocks/Cargo.toml b/vortex-btrblocks/Cargo.toml index 255d2fbd075..a4c8ff83ff3 100644 --- a/vortex-btrblocks/Cargo.toml +++ b/vortex-btrblocks/Cargo.toml @@ -44,6 +44,7 @@ insta = { workspace = true } rstest = { workspace = true } test-with = { workspace = true } vortex-array = { workspace = true, features = ["_test-harness"] } +vortex-mask = { workspace = true } vortex-session = { workspace = true } [features] @@ -64,3 +65,8 @@ test = false name = "compress_listview" harness = false test = false + +[[bench]] +name = "stable_tradeoff" +harness = false +test = false diff --git a/vortex-btrblocks/benches/stable_tradeoff.rs b/vortex-btrblocks/benches/stable_tradeoff.rs new file mode 100644 index 00000000000..af82d62d7b4 --- /dev/null +++ b/vortex-btrblocks/benches/stable_tradeoff.rs @@ -0,0 +1,414 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Targeted size-versus-execution benchmark for stable encodings. + +#![expect(clippy::unwrap_used)] + +use std::sync::Arc; +use std::sync::LazyLock; + +use divan::Bencher; +use divan::counter::ItemsCount; +use vortex_array::ArrayRef; +use vortex_array::Canonical; +use vortex_array::IntoArray; +use vortex_array::VortexSessionExecute; +use vortex_array::arrays::BoolArray; +use vortex_array::arrays::ConstantArray; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::arrays::VarBinViewArray; +use vortex_array::arrays::scalar_fn::ScalarFnFactoryExt; +use vortex_array::builtins::ArrayBuiltins; +use vortex_array::scalar_fn::fns::like::Like; +use vortex_array::scalar_fn::fns::like::LikeOptions; +use vortex_array::scalar_fn::fns::operators::Operator; +use vortex_btrblocks::ArrayAndStats; +use vortex_btrblocks::BtrBlocksCompressorBuilder; +use vortex_btrblocks::Cost; +use vortex_btrblocks::CostModel; +use vortex_btrblocks::ScanCost; +use vortex_btrblocks::SchemeExt; +use vortex_btrblocks::SchemeId; +use vortex_btrblocks::SizeCost; +use vortex_btrblocks::schemes::integer::BitPackingScheme; +use vortex_btrblocks::schemes::integer::RunEndScheme; +use vortex_btrblocks::schemes::integer::SparseScheme; +use vortex_btrblocks::schemes::string::FSSTScheme; +use vortex_btrblocks::schemes::string::StringDictScheme; +use vortex_buffer::Buffer; +use vortex_compressor::cost::Candidate; +use vortex_mask::Mask; +use vortex_session::VortexSession; + +const LEN: usize = 1 << 18; +const UNIQUE: usize = LEN / 2; +const INTEGER_LEN: usize = 1 << 20; + +static SESSION: LazyLock = LazyLock::new(|| { + let session = vortex_array::array_session(); + vortex_fastlanes::initialize(&session); + vortex_fsst::initialize(&session); + session +}); + +static INPUT: LazyLock = LazyLock::new(|| { + VarBinViewArray::from_iter_str((0..LEN).map(|index| { + format!( + "tenant-{:06}-event-checkout-completed-region-us-east", + index % UNIQUE + ) + })) + .into_array() +}); + +static SIZE_WINNER: LazyLock = LazyLock::new(|| { + BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(SizeCost)) + .build() + .compress(&INPUT, &mut SESSION.create_execution_ctx()) + .unwrap() +}); + +static FSST: LazyLock = LazyLock::new(|| compress_with_root(&INPUT, FSSTScheme.id())); +static DICT: LazyLock = + LazyLock::new(|| compress_with_root(&INPUT, StringDictScheme.id())); + +static INTEGER_INPUT: LazyLock = LazyLock::new(|| { + let values: Buffer = (0..INTEGER_LEN) + .map(|index| { + if index.is_multiple_of(20) { + u32::try_from(index).unwrap() + } else { + 0 + } + }) + .collect(); + values.into_array() +}); + +static INTEGER_SIZE_WINNER: LazyLock = LazyLock::new(|| { + BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(SizeCost)) + .build() + .compress(&INTEGER_INPUT, &mut SESSION.create_execution_ctx()) + .unwrap() +}); + +static SPARSE: LazyLock = + LazyLock::new(|| compress_with_root(&INTEGER_INPUT, SparseScheme.id())); +static BITPACKED: LazyLock = + LazyLock::new(|| compress_with_root(&INTEGER_INPUT, BitPackingScheme.id())); +static INTEGER_FILTER: LazyLock = + LazyLock::new(|| Mask::from_iter((0..INTEGER_LEN).map(|index| index.is_multiple_of(10)))); + +static RUN_INPUT: LazyLock = LazyLock::new(|| { + let values: Buffer = (0..INTEGER_LEN) + .map(|index| u32::try_from(index / 16).unwrap()) + .collect(); + values.into_array() +}); + +static RUN_SIZE_WINNER: LazyLock = LazyLock::new(|| { + BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(SizeCost)) + .build() + .compress(&RUN_INPUT, &mut SESSION.create_execution_ctx()) + .unwrap() +}); + +static RUNEND: LazyLock = + LazyLock::new(|| compress_with_root(&RUN_INPUT, RunEndScheme.id())); +static RUN_BITPACKED: LazyLock = + LazyLock::new(|| compress_with_root(&RUN_INPUT, BitPackingScheme.id())); +static RUN_SCAN_COST: LazyLock = LazyLock::new(|| { + BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(ScanCost::default())) + .build() + .compress(&RUN_INPUT, &mut SESSION.create_execution_ctx()) + .unwrap() +}); + +/// Forces one root family while preserving normal size-based descendant selection. +#[derive(Debug)] +struct ForceRootCost { + root: SchemeId, +} + +impl CostModel for ForceRootCost { + fn cost(&self, candidate: &Candidate<'_>) -> Option { + if candidate.cascade().is_empty() && candidate.scheme_id() != self.root { + return None; + } + SizeCost.cost(candidate) + } + + fn canonical_cost(&self, data: &ArrayAndStats, n_values: u64) -> Cost { + SizeCost.canonical_cost(data, n_values) + } +} + +fn compress_with_root(input: &ArrayRef, root: SchemeId) -> ArrayRef { + BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(ForceRootCost { root })) + .build() + .compress(input, &mut SESSION.create_execution_ctx()) + .unwrap() +} + +fn bench_decode(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(LEN)) + .with_inputs(|| { + ( + LazyLock::force(array).clone(), + SESSION.create_execution_ctx(), + ) + }) + .bench_refs(|(array, ctx)| { + divan::black_box(array.clone().execute::(ctx).unwrap()) + }); +} + +fn bench_eq(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(LEN)) + .with_inputs(|| { + let lhs = LazyLock::force(array).clone(); + let rhs = + ConstantArray::new("tenant-131071-event-checkout-completed-region-us-east", LEN) + .into_array(); + (lhs, rhs, SESSION.create_execution_ctx()) + }) + .bench_refs(|(lhs, rhs, ctx)| { + divan::black_box( + lhs.clone() + .binary(rhs.clone(), Operator::Eq) + .unwrap() + .execute::(ctx) + .unwrap(), + ) + }); +} + +fn bench_like(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(LEN)) + .with_inputs(|| { + let lhs = LazyLock::force(array).clone(); + let pattern = ConstantArray::new("%checkout-completed%", LEN).into_array(); + (lhs, pattern, SESSION.create_execution_ctx()) + }) + .bench_refs(|(lhs, pattern, ctx)| { + divan::black_box( + Like.try_new_array(LEN, LikeOptions::default(), [lhs.clone(), pattern.clone()]) + .unwrap() + .into_array() + .execute::(ctx) + .unwrap(), + ) + }); +} + +fn bench_integer_decode(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(INTEGER_LEN)) + .with_inputs(|| { + ( + LazyLock::force(array).clone(), + SESSION.create_execution_ctx(), + ) + }) + .bench_refs(|(array, ctx)| { + divan::black_box(array.clone().execute::(ctx).unwrap()) + }); +} + +fn bench_integer_eq(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(INTEGER_LEN)) + .with_inputs(|| { + let lhs = LazyLock::force(array).clone(); + let rhs = ConstantArray::new(0u32, INTEGER_LEN).into_array(); + (lhs, rhs, SESSION.create_execution_ctx()) + }) + .bench_refs(|(lhs, rhs, ctx)| { + divan::black_box( + lhs.clone() + .binary(rhs.clone(), Operator::Eq) + .unwrap() + .execute::(ctx) + .unwrap(), + ) + }); +} + +fn bench_integer_filter(bencher: Bencher, array: &'static LazyLock) { + bencher + .counter(ItemsCount::new(INTEGER_LEN)) + .with_inputs(|| { + ( + LazyLock::force(array).clone(), + LazyLock::force(&INTEGER_FILTER).clone(), + SESSION.create_execution_ctx(), + ) + }) + .bench_refs(|(array, mask, ctx)| { + divan::black_box( + array + .clone() + .filter(mask.clone()) + .unwrap() + .execute::(ctx) + .unwrap(), + ) + }); +} + +#[divan::bench] +fn fsst_decode(bencher: Bencher) { + bench_decode(bencher, &FSST); +} + +#[divan::bench] +fn dict_decode(bencher: Bencher) { + bench_decode(bencher, &DICT); +} + +#[divan::bench] +fn fsst_eq(bencher: Bencher) { + bench_eq(bencher, &FSST); +} + +#[divan::bench] +fn dict_eq(bencher: Bencher) { + bench_eq(bencher, &DICT); +} + +#[divan::bench] +fn fsst_like(bencher: Bencher) { + bench_like(bencher, &FSST); +} + +#[divan::bench] +fn dict_like(bencher: Bencher) { + bench_like(bencher, &DICT); +} + +#[divan::bench] +fn sparse_decode(bencher: Bencher) { + bench_integer_decode(bencher, &SPARSE); +} + +#[divan::bench] +fn bitpacked_decode(bencher: Bencher) { + bench_integer_decode(bencher, &BITPACKED); +} + +#[divan::bench] +fn sparse_eq_fill(bencher: Bencher) { + bench_integer_eq(bencher, &SPARSE); +} + +#[divan::bench] +fn bitpacked_eq_fill(bencher: Bencher) { + bench_integer_eq(bencher, &BITPACKED); +} + +#[divan::bench] +fn sparse_filter_10pct(bencher: Bencher) { + bench_integer_filter(bencher, &SPARSE); +} + +#[divan::bench] +fn bitpacked_filter_10pct(bencher: Bencher) { + bench_integer_filter(bencher, &BITPACKED); +} + +#[divan::bench] +fn runend_decode(bencher: Bencher) { + bench_integer_decode(bencher, &RUNEND); +} + +#[divan::bench] +fn run_bitpacked_decode(bencher: Bencher) { + bench_integer_decode(bencher, &RUN_BITPACKED); +} + +#[divan::bench] +fn runend_eq(bencher: Bencher) { + bench_integer_eq(bencher, &RUNEND); +} + +#[divan::bench] +fn run_bitpacked_eq(bencher: Bencher) { + bench_integer_eq(bencher, &RUN_BITPACKED); +} + +#[divan::bench] +fn runend_filter_10pct(bencher: Bencher) { + bench_integer_filter(bencher, &RUNEND); +} + +#[divan::bench] +fn run_bitpacked_filter_10pct(bencher: Bencher) { + bench_integer_filter(bencher, &RUN_BITPACKED); +} + +#[divan::bench] +fn run_scan_cost_decode(bencher: Bencher) { + bench_integer_decode(bencher, &RUN_SCAN_COST); +} + +#[divan::bench] +fn run_scan_cost_eq(bencher: Bencher) { + bench_integer_eq(bencher, &RUN_SCAN_COST); +} + +#[divan::bench] +fn run_scan_cost_filter_10pct(bencher: Bencher) { + bench_integer_filter(bencher, &RUN_SCAN_COST); +} + +fn main() { + let size_winner = LazyLock::force(&SIZE_WINNER); + let fsst = LazyLock::force(&FSST); + let dict = LazyLock::force(&DICT); + println!( + "size winner: encoding={} bytes={}; fsst={} dict={} dict/fsst={:.3}", + size_winner.encoding_id(), + size_winner.nbytes(), + fsst.nbytes(), + dict.nbytes(), + dict.nbytes() as f64 / fsst.nbytes() as f64, + ); + let integer_size_winner = LazyLock::force(&INTEGER_SIZE_WINNER); + let sparse = LazyLock::force(&SPARSE); + let bitpacked = LazyLock::force(&BITPACKED); + println!( + "integer size winner: encoding={} bytes={}; sparse={} bitpacked={} bitpacked/sparse={:.3}", + integer_size_winner.encoding_id(), + integer_size_winner.nbytes(), + sparse.nbytes(), + bitpacked.nbytes(), + bitpacked.nbytes() as f64 / sparse.nbytes() as f64, + ); + let run_size_winner = LazyLock::force(&RUN_SIZE_WINNER); + let runend = LazyLock::force(&RUNEND); + let run_bitpacked = LazyLock::force(&RUN_BITPACKED); + let run_scan_cost = LazyLock::force(&RUN_SCAN_COST); + println!( + "run size winner: encoding={} bytes={}; runend={} bitpacked={} bitpacked/runend={:.3}", + run_size_winner.encoding_id(), + run_size_winner.nbytes(), + runend.nbytes(), + run_bitpacked.nbytes(), + run_bitpacked.nbytes() as f64 / runend.nbytes() as f64, + ); + println!( + "run scan-cost alternative: encoding={} bytes={}", + run_scan_cost.encoding_id(), + run_scan_cost.nbytes(), + ); + divan::main(); +} diff --git a/vortex-btrblocks/src/builder.rs b/vortex-btrblocks/src/builder.rs index 26632e70860..32111adc8bf 100644 --- a/vortex-btrblocks/src/builder.rs +++ b/vortex-btrblocks/src/builder.rs @@ -3,10 +3,16 @@ //! Builder for configuring `BtrBlocksCompressor` instances. +use std::sync::Arc; + +use vortex_compressor::cost::CostModel; +use vortex_compressor::cost::SizeCost; use vortex_utils::aliases::hash_set::HashSet; use crate::BtrBlocksCompressor; use crate::CascadingCompressor; +#[cfg(not(feature = "unstable_encodings"))] +use crate::ScanCost; use crate::Scheme; use crate::SchemeExt; use crate::SchemeId; @@ -68,8 +74,10 @@ pub const ALL_SCHEMES: &[&dyn Scheme] = &[ /// Builder for creating configured [`BtrBlocksCompressor`] instances. /// -/// By default, all schemes in [`ALL_SCHEMES`] are enabled in a deterministic order. Feature-gated -/// schemes (Pco, Zstd) are not in `ALL_SCHEMES` and must be added explicitly via +/// By default, all schemes in [`ALL_SCHEMES`] are enabled in a deterministic order. Stable builds +/// rank candidates with [`ScanCost`]; builds that opt into `unstable_encodings` preserve +/// [`SizeCost`] selection. Feature-gated schemes (Pco, Zstd) are not in `ALL_SCHEMES` and must be +/// added explicitly via /// [`with_new_scheme`](BtrBlocksCompressorBuilder::with_new_scheme) or `with_compact` when the /// `zstd` feature is enabled. /// @@ -90,12 +98,14 @@ pub const ALL_SCHEMES: &[&dyn Scheme] = &[ #[derive(Debug, Clone)] pub struct BtrBlocksCompressorBuilder { schemes: Vec<&'static dyn Scheme>, + cost_model: Arc, } impl Default for BtrBlocksCompressorBuilder { fn default() -> Self { Self { schemes: ALL_SCHEMES.to_vec(), + cost_model: default_cost_model(), } } } @@ -107,6 +117,7 @@ impl BtrBlocksCompressorBuilder { pub fn empty() -> Self { Self { schemes: Vec::new(), + cost_model: Arc::new(SizeCost), } } @@ -202,12 +213,33 @@ impl BtrBlocksCompressorBuilder { self } + /// Uses a custom model to rank compression candidates. + /// + /// Stable builds default to [`ScanCost`]. Builds with `unstable_encodings` default to + /// [`SizeCost`] because the scan calibration has not been validated for those schemes. + /// [`ReadTimeCost`](vortex_compressor::cost::ReadTimeCost) is available for deterministic + /// sequential-scan objectives that combine estimated I/O and decode time. + pub fn with_cost_model(mut self, cost_model: Arc) -> Self { + self.cost_model = cost_model; + self + } + /// Builds the configured [`BtrBlocksCompressor`]. pub fn build(self) -> BtrBlocksCompressor { - BtrBlocksCompressor(CascadingCompressor::new(self.schemes)) + BtrBlocksCompressor(CascadingCompressor::new(self.schemes).with_cost_model(self.cost_model)) } } +#[cfg(not(feature = "unstable_encodings"))] +fn default_cost_model() -> Arc { + Arc::new(ScanCost::default()) +} + +#[cfg(feature = "unstable_encodings")] +fn default_cost_model() -> Arc { + Arc::new(SizeCost) +} + #[cfg(test)] mod tests { use super::*; diff --git a/vortex-btrblocks/src/lib.rs b/vortex-btrblocks/src/lib.rs index 37915b427b6..54bcb9add0a 100644 --- a/vortex-btrblocks/src/lib.rs +++ b/vortex-btrblocks/src/lib.rs @@ -30,8 +30,8 @@ //! //! Each `Scheme` implementation declares whether it [`matches`](Scheme::matches) a given //! canonical form and, if so, estimates the compression ratio (often by compressing a ~1% -//! sample). There is no dynamic registry — the set of schemes is fixed at build time via -//! [`ALL_SCHEMES`]. +//! sample). The configured cost model ranks those estimates. There is no dynamic registry — the +//! set of schemes is fixed at build time via [`ALL_SCHEMES`]. //! //! Schemes can produce arrays that are themselves further compressed (e.g. FoR then BitPacking), //! up to [`MAX_CASCADE`] (3) layers deep. Descendant exclusion rules for of [`SchemeId`] prevents @@ -68,6 +68,7 @@ mod builder; mod canonical_compressor; +mod scan_cost; /// Compression scheme implementations. pub mod schemes; @@ -76,8 +77,13 @@ pub mod schemes; pub use builder::ALL_SCHEMES; pub use builder::BtrBlocksCompressorBuilder; pub use canonical_compressor::BtrBlocksCompressor; +pub use scan_cost::ScanCost; pub use schemes::patches::compress_patches; pub use vortex_compressor::CascadingCompressor; +pub use vortex_compressor::cost::Cost; +pub use vortex_compressor::cost::CostModel; +pub use vortex_compressor::cost::ReadTimeCost; +pub use vortex_compressor::cost::SizeCost; pub use vortex_compressor::scheme::CompressorContext; pub use vortex_compressor::scheme::MAX_CASCADE; pub use vortex_compressor::scheme::Scheme; diff --git a/vortex-btrblocks/src/scan_cost.rs b/vortex-btrblocks/src/scan_cost.rs new file mode 100644 index 00000000000..f5e6d0b67f5 --- /dev/null +++ b/vortex-btrblocks/src/scan_cost.rs @@ -0,0 +1,42 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! A calibrated cost model for analytical scans. + +use vortex_compressor::cost::Candidate; +use vortex_compressor::cost::Cost; +use vortex_compressor::cost::CostModel; +use vortex_compressor::cost::ReadTimeCost; +use vortex_compressor::stats::ArrayAndStats; + +use crate::SchemeExt; +use crate::schemes::integer::RunEndScheme; + +/// Estimates cached analytical scan time instead of minimizing encoded size alone. +/// +/// The model combines estimated bytes with deterministic per-scheme decode charges. Its initial +/// calibration assumes 20 bytes/ns effective read bandwidth and charges RunEnd 0.12 ns/value. +/// This avoids selecting RunEnd for modest size wins when a faster stable encoding is available. +/// Schemes without a calibration currently receive no decode charge. +/// +/// Use [`crate::SizeCost`] with +/// [`crate::BtrBlocksCompressorBuilder::with_cost_model`] when minimum encoded size is the desired +/// objective. Use [`ReadTimeCost`] directly to supply workload- or hardware-specific coefficients. +#[derive(Debug, Clone)] +pub struct ScanCost(ReadTimeCost); + +impl Default for ScanCost { + fn default() -> Self { + Self(ReadTimeCost::new(20.0, 0.0).with_scheme_cost(RunEndScheme.id(), 0.12)) + } +} + +impl CostModel for ScanCost { + fn cost(&self, candidate: &Candidate<'_>) -> Option { + self.0.cost(candidate) + } + + fn canonical_cost(&self, data: &ArrayAndStats, n_values: u64) -> Cost { + self.0.canonical_cost(data, n_values) + } +} diff --git a/vortex-btrblocks/src/schemes/integer/scheme_selection_tests.rs b/vortex-btrblocks/src/schemes/integer/scheme_selection_tests.rs index e4227a472ec..ab7fac65b6d 100644 --- a/vortex-btrblocks/src/schemes/integer/scheme_selection_tests.rs +++ b/vortex-btrblocks/src/schemes/integer/scheme_selection_tests.rs @@ -4,6 +4,7 @@ //! Tests to verify that each integer compression scheme produces the expected encoding. use std::iter; +use std::sync::Arc; use std::sync::LazyLock; use rand::Rng; @@ -28,6 +29,8 @@ use vortex_session::VortexSession; use vortex_sparse::Sparse; use crate::BtrBlocksCompressor; +use crate::BtrBlocksCompressorBuilder; +use crate::SizeCost; static SESSION: LazyLock = LazyLock::new(vortex_array::array_session); #[test] @@ -113,19 +116,39 @@ fn test_dict_compressed() -> VortexResult<()> { Ok(()) } -#[test] -fn test_runend_compressed() -> VortexResult<()> { +fn runend_input() -> PrimitiveArray { let mut values: Vec = Vec::new(); for i in 0..100 { values.extend(iter::repeat_n((i32::MAX - 50).wrapping_add(i), 10)); } - let array = PrimitiveArray::new(Buffer::copy_from(&values), Validity::NonNullable); - let btr = BtrBlocksCompressor::default(); - let compressed = btr.compress(&array.into_array(), &mut SESSION.create_execution_ctx())?; + + PrimitiveArray::new(Buffer::copy_from(&values), Validity::NonNullable) +} + +#[test] +fn test_size_cost_selects_runend() -> VortexResult<()> { + let btr = BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(SizeCost)) + .build(); + let compressed = btr.compress( + &runend_input().into_array(), + &mut SESSION.create_execution_ctx(), + )?; assert!(compressed.is::()); Ok(()) } +#[test] +#[cfg(not(feature = "unstable_encodings"))] +fn test_scan_cost_avoids_runend_for_a_modest_size_win() -> VortexResult<()> { + let compressed = BtrBlocksCompressor::default().compress( + &runend_input().into_array(), + &mut SESSION.create_execution_ctx(), + )?; + assert!(!compressed.is::()); + Ok(()) +} + #[test] fn test_sequence_compressed() -> VortexResult<()> { let values: Vec = (0..1000).map(|i| i * 7).collect(); @@ -137,7 +160,7 @@ fn test_sequence_compressed() -> VortexResult<()> { } #[test] -fn test_rle_compressed() -> VortexResult<()> { +fn test_size_cost_selects_runend_for_rle_input() -> VortexResult<()> { let mut values: Vec = Vec::new(); for i in 0..1024 { // Scramble the per-run value so the data is run-length-dominant but not monotone: this @@ -147,7 +170,9 @@ fn test_rle_compressed() -> VortexResult<()> { values.extend(iter::repeat_n(v, 10)); } let array = PrimitiveArray::new(Buffer::copy_from(&values), Validity::NonNullable); - let btr = BtrBlocksCompressor::default(); + let btr = BtrBlocksCompressorBuilder::default() + .with_cost_model(Arc::new(SizeCost)) + .build(); let compressed = btr.compress(&array.into_array(), &mut SESSION.create_execution_ctx())?; eprintln!("{}", compressed.display_tree()); assert!(compressed.is::()); diff --git a/vortex-btrblocks/tests/snapshots/golden__default__int_runs.snap b/vortex-btrblocks/tests/snapshots/golden__default__int_runs.snap index 271de8faf93..a6633ba9a64 100644 --- a/vortex-btrblocks/tests/snapshots/golden__default__int_runs.snap +++ b/vortex-btrblocks/tests/snapshots/golden__default__int_runs.snap @@ -3,13 +3,23 @@ source: vortex-btrblocks/tests/golden.rs expression: rendered --- input: i32, len=16384, nbytes=65536 -root: vortex.runend(i32, len=16384) nbytes=3968 - metadata: offset: 0 - ends: fastlanes.for(u16, len=1020) nbytes=1792 - metadata: reference: 13u16 - encoded: fastlanes.bitpacked(u16, len=1020) nbytes=1792 - metadata: bit_width: 14, offset: 0 - values: fastlanes.for(i32, len=1020) nbytes=2176 +root: vortex.dict(i32, len=16384) nbytes=17435 + metadata: all_values_referenced: true + codes: fastlanes.rle(u16, len=16384) nbytes=15259 + metadata: offset: 0 + values: vortex.primitive(u16, len=1035) nbytes=2070 + metadata: ptype: u16 + indices: fastlanes.bitpacked(u8, len=16384) nbytes=13157 + metadata: bit_width: 6, offset: 0 + patch_indices: vortex.primitive(u16, len=279) nbytes=558 + metadata: ptype: u16 + patch_values: vortex.primitive(u8, len=279) nbytes=279 + metadata: ptype: u8 + patch_chunk_offsets: vortex.primitive(u16, len=16) nbytes=32 + metadata: ptype: u16 + values_idx_offsets: vortex.primitive(u16, len=16) nbytes=32 + metadata: ptype: u16 + values: fastlanes.for(i32, len=1014) nbytes=2176 metadata: reference: -49931i32 - encoded: fastlanes.bitpacked(i32, len=1020) nbytes=2176 + encoded: fastlanes.bitpacked(i32, len=1014) nbytes=2176 metadata: bit_width: 17, offset: 0 diff --git a/vortex-btrblocks/tests/snapshots/golden__default__list_of_int_runs.snap b/vortex-btrblocks/tests/snapshots/golden__default__list_of_int_runs.snap index c9554add05a..f821d9a0165 100644 --- a/vortex-btrblocks/tests/snapshots/golden__default__list_of_int_runs.snap +++ b/vortex-btrblocks/tests/snapshots/golden__default__list_of_int_runs.snap @@ -3,17 +3,27 @@ source: vortex-btrblocks/tests/golden.rs expression: rendered --- input: list(i32), len=4066, nbytes=81804 -root: vortex.list(list(i32), len=4066) nbytes=11146 +root: vortex.list(list(i32), len=4066) nbytes=24613 metadata: - elements: vortex.runend(i32, len=16384) nbytes=3968 - metadata: offset: 0 - ends: fastlanes.for(u16, len=1020) nbytes=1792 - metadata: reference: 13u16 - encoded: fastlanes.bitpacked(u16, len=1020) nbytes=1792 - metadata: bit_width: 14, offset: 0 - values: fastlanes.for(i32, len=1020) nbytes=2176 + elements: vortex.dict(i32, len=16384) nbytes=17435 + metadata: all_values_referenced: true + codes: fastlanes.rle(u16, len=16384) nbytes=15259 + metadata: offset: 0 + values: vortex.primitive(u16, len=1035) nbytes=2070 + metadata: ptype: u16 + indices: fastlanes.bitpacked(u8, len=16384) nbytes=13157 + metadata: bit_width: 6, offset: 0 + patch_indices: vortex.primitive(u16, len=279) nbytes=558 + metadata: ptype: u16 + patch_values: vortex.primitive(u8, len=279) nbytes=279 + metadata: ptype: u8 + patch_chunk_offsets: vortex.primitive(u16, len=16) nbytes=32 + metadata: ptype: u16 + values_idx_offsets: vortex.primitive(u16, len=16) nbytes=32 + metadata: ptype: u16 + values: fastlanes.for(i32, len=1014) nbytes=2176 metadata: reference: -49931i32 - encoded: fastlanes.bitpacked(i32, len=1020) nbytes=2176 + encoded: fastlanes.bitpacked(i32, len=1014) nbytes=2176 metadata: bit_width: 17, offset: 0 offsets: fastlanes.bitpacked(u16, len=4067) nbytes=7178 metadata: bit_width: 14, offset: 0 diff --git a/vortex-compressor/src/cost/mod.rs b/vortex-compressor/src/cost/mod.rs index fda8c2c0c86..025885a6e54 100644 --- a/vortex-compressor/src/cost/mod.rs +++ b/vortex-compressor/src/cost/mod.rs @@ -47,5 +47,8 @@ mod model; pub use model::Cost; pub use model::CostModel; +mod read_time; +pub use read_time::ReadTimeCost; + mod size; pub use size::SizeCost; diff --git a/vortex-compressor/src/cost/read_time.rs b/vortex-compressor/src/cost/read_time.rs new file mode 100644 index 00000000000..39dd68ef24a --- /dev/null +++ b/vortex-compressor/src/cost/read_time.rs @@ -0,0 +1,205 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! A deterministic read-time estimate for sequential scans. + +use vortex_utils::aliases::hash_map::HashMap; + +use crate::cost::Candidate; +use crate::cost::Cost; +use crate::cost::CostModel; +use crate::scheme::SchemeId; +use crate::stats::ArrayAndStats; + +/// Estimates sequential read time as I/O time plus scheme decode time. +/// +/// The model prices a candidate as: +/// +/// `estimated_bytes / effective_bandwidth + values * decode_ns_per_value` +/// +/// `effective_bandwidth` is expressed in bytes per nanosecond (numerically equal to decimal +/// GB/s), while scheme charges are expressed in nanoseconds per value. The coefficients are +/// configuration, never measurements taken on the write path, so selection remains deterministic. +/// +/// This intentionally models only the scheme currently being selected. It does not yet price +/// operation pushdown, random access, or the complete produced encoding tree; those require the +/// capability and calibration artifacts described by the cost-model tracking plan. +#[derive(Debug, Clone)] +pub struct ReadTimeCost { + /// Estimated bytes read per nanosecond. + effective_bandwidth: f64, + /// Decode charge used when a scheme has no explicit calibration. + fallback_decode_ns_per_value: f64, + /// Per-scheme decode charges in nanoseconds per value. + decode_ns_per_value: HashMap, +} + +impl ReadTimeCost { + /// Creates a read-time model with a conservative charge for unregistered schemes. + /// + /// `effective_bandwidth` is bytes per nanosecond and `fallback_decode_ns_per_value` is + /// nanoseconds per decoded value. + /// + /// # Panics + /// + /// Panics if either argument is non-finite, if bandwidth is not positive, or if the fallback + /// decode charge is negative. + pub fn new(effective_bandwidth: f64, fallback_decode_ns_per_value: f64) -> Self { + assert!( + effective_bandwidth.is_finite() && effective_bandwidth > 0.0, + "effective bandwidth must be finite and positive" + ); + assert!( + (u64::MAX as f64 / effective_bandwidth).is_finite(), + "effective bandwidth is too small to produce finite costs" + ); + assert!( + fallback_decode_ns_per_value.is_finite() && fallback_decode_ns_per_value >= 0.0, + "fallback decode cost must be finite and non-negative" + ); + Self { + effective_bandwidth, + fallback_decode_ns_per_value, + decode_ns_per_value: HashMap::default(), + } + } + + /// Sets the decode charge for a scheme in nanoseconds per value. + /// + /// # Panics + /// + /// Panics if the charge is non-finite or negative. + pub fn with_scheme_cost(mut self, scheme: SchemeId, decode_ns_per_value: f64) -> Self { + assert!( + decode_ns_per_value.is_finite() && decode_ns_per_value >= 0.0, + "scheme decode cost must be finite and non-negative" + ); + self.decode_ns_per_value.insert(scheme, decode_ns_per_value); + self + } + + /// Returns the configured effective bandwidth in bytes per nanosecond. + pub fn effective_bandwidth(&self) -> f64 { + self.effective_bandwidth + } +} + +impl CostModel for ReadTimeCost { + fn cost(&self, candidate: &Candidate<'_>) -> Option { + let ratio = candidate.estimate().estimated_compression_ratio()?; + // A larger representation cannot beat canonical because decode charges are non-negative. + if !ratio.is_finite() || ratio.is_subnormal() || ratio <= 1.0 { + return None; + } + + let estimated_nbytes = candidate.input_nbytes() as f64 / ratio; + let decode_ns_per_value = self + .decode_ns_per_value + .get(&candidate.scheme_id()) + .copied() + .unwrap_or(self.fallback_decode_ns_per_value); + let cost = estimated_nbytes / self.effective_bandwidth + + candidate.n_values() as f64 * decode_ns_per_value; + cost.is_finite().then(|| Cost::new(cost)) + } + + fn canonical_cost(&self, data: &ArrayAndStats, _n_values: u64) -> Cost { + Cost::new(data.array().nbytes() as f64 / self.effective_bandwidth) + } +} + +#[cfg(test)] +mod tests { + use vortex_array::ArrayRef; + use vortex_array::Canonical; + use vortex_array::ExecutionCtx; + use vortex_array::IntoArray; + use vortex_array::arrays::PrimitiveArray; + use vortex_array::validity::Validity; + use vortex_buffer::buffer; + use vortex_error::VortexResult; + + use super::*; + use crate::CascadingCompressor; + use crate::scheme::CandidateEstimate; + use crate::scheme::CompressorContext; + use crate::scheme::Scheme; + use crate::scheme::SchemeEvaluation; + use crate::scheme::SchemeExt; + use crate::stats::GenerateStatsOptions; + + #[derive(Debug)] + struct TestScheme; + + impl Scheme for TestScheme { + fn scheme_name(&self) -> &'static str { + "test.read_time_cost" + } + + fn matches(&self, _canonical: &Canonical) -> bool { + true + } + + fn evaluate( + &self, + _data: &ArrayAndStats, + _compress_ctx: CompressorContext, + _exec_ctx: &mut ExecutionCtx, + ) -> SchemeEvaluation { + SchemeEvaluation::Skip + } + + fn compress( + &self, + _compressor: &CascadingCompressor, + _data: &ArrayAndStats, + _compress_ctx: CompressorContext, + _exec_ctx: &mut ExecutionCtx, + ) -> VortexResult { + unreachable!("test helper is never selected") + } + } + + fn test_data() -> ArrayAndStats { + ArrayAndStats::new( + PrimitiveArray::new(buffer![1i32, 2, 3, 4], Validity::NonNullable).into_array(), + GenerateStatsOptions::default(), + ) + } + + #[test] + fn combines_io_and_decode_time() { + let data = test_data(); + let model = ReadTimeCost::new(4.0, 1.0).with_scheme_cost(TestScheme.id(), 0.5); + let candidate = Candidate::new( + &TestScheme, + CandidateEstimate::from_compression_ratio(2.0), + &data, + None, + &[], + ); + + // 16 input bytes / ratio 2 / 4 bytes/ns + 4 values * 0.5 ns/value. + assert_eq!(model.cost(&candidate).map(Cost::value), Some(4.0)); + assert_eq!(model.canonical_cost(&data, 4).value(), 4.0); + } + + #[test] + fn decode_charge_can_reject_a_smaller_candidate() { + let data = test_data(); + let model = ReadTimeCost::new(4.0, 1.0); + let candidate = Candidate::new( + &TestScheme, + CandidateEstimate::from_compression_ratio(2.0), + &data, + None, + &[], + ); + + assert!( + model + .cost(&candidate) + .is_some_and(|cost| { cost >= model.canonical_cost(&data, candidate.n_values()) }) + ); + } +} diff --git a/vortex/src/lib.rs b/vortex/src/lib.rs index 2703020906c..d34be18e914 100644 --- a/vortex/src/lib.rs +++ b/vortex/src/lib.rs @@ -141,8 +141,13 @@ pub mod buffer { pub mod compressor { pub use vortex_btrblocks::BtrBlocksCompressor; pub use vortex_btrblocks::BtrBlocksCompressorBuilder; + pub use vortex_btrblocks::Cost; + pub use vortex_btrblocks::CostModel; + pub use vortex_btrblocks::ReadTimeCost; + pub use vortex_btrblocks::ScanCost; pub use vortex_btrblocks::Scheme; pub use vortex_btrblocks::SchemeId; + pub use vortex_btrblocks::SizeCost; } /// Logical Vortex data types.