diff --git a/.github/workflows/bench-pr.yml b/.github/workflows/bench-pr.yml index a92704e6783..2a12c0a60b2 100644 --- a/.github/workflows/bench-pr.yml +++ b/.github/workflows/bench-pr.yml @@ -33,6 +33,10 @@ jobs: build_args: "--features lance" - id: compress-bench name: Compression + # No run_args: the default suite emits the three tracked metrics + # (size, write, read). The codec microbenchmark is a local diagnostic. + - id: string-bench + name: String Encoding steps: - uses: runs-on/action@v2 if: github.event.pull_request.head.repo.fork == false @@ -104,7 +108,8 @@ jobs: VORTEX_EXPERIMENTAL_PATCHED_ARRAY: "1" FLAT_LAYOUT_INLINE_ARRAY_NODE: "1" run: | - bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} -d gh-json -o results.json + bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} \ + ${{ matrix.benchmark.run_args }} -d gh-json -o results.json - name: Setup AWS CLI if: github.event.pull_request.head.repo.fork == false diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml index a2969c63c30..a046acc8ff4 100644 --- a/.github/workflows/bench.yml +++ b/.github/workflows/bench.yml @@ -44,11 +44,17 @@ jobs: - id: random-access-bench name: Random Access build_args: "--features lance" - formats: "parquet,lance,vortex" + v4_ingest: true - id: compress-bench name: Compression build_args: "--features lance" - formats: "parquet,lance,vortex" + run_args: "--formats parquet,lance,vortex --ingest-jsonl results.ingest.jsonl" + v4_ingest: true + # No run_args: the default suite emits the three tracked metrics + # (size, write, read). The codec microbenchmark is a local diagnostic. + - id: string-bench + name: String Encoding + v4_ingest: false steps: - uses: runs-on/action@v2 if: github.repository == 'vortex-data/vortex' @@ -116,7 +122,8 @@ jobs: VORTEX_EXPERIMENTAL_PATCHED_ARRAY: "1" FLAT_LAYOUT_INLINE_ARRAY_NODE: "1" run: | - bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} --formats ${{ matrix.benchmark.formats }} -d gh-json -o results.json --ingest-jsonl results.ingest.jsonl + bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} \ + ${{ matrix.benchmark.run_args }} -d gh-json -o results.json - name: Setup AWS CLI uses: aws-actions/configure-aws-credentials@e6de054238d6b7531b4efff3b6587d9aade6a06c # v6 @@ -130,28 +137,30 @@ jobs: bash scripts/cat-s3.sh vortex-ci-benchmark-results data.json.gz results.json # v4 (Postgres) ingest -- the REQUIRED benchmark-results pipeline feeding the live - # benchmarks website; a failure here fails the job. Gated on the ingest-role ARN var - # (the assume-role input that MUST exist for OIDC to succeed). post-ingest.py mints - # the RDS IAM token internally (boto3) from the assumed GitHubBenchmarkIngestRole; - # sslmode=verify-full validates the cert. + # benchmarks website for suites that emit v4 records; a failure here fails the job. + # All suites also upload S3 results above. string-bench does not emit + # results.ingest.jsonl, so these steps are disabled for it. + # Also gated on the ingest-role ARN var (the assume-role input that MUST exist for + # OIDC to succeed). post-ingest.py mints the RDS IAM token internally (boto3) from + # the assumed GitHubBenchmarkIngestRole; sslmode=verify-full validates the cert. # # `sync: false` -- the ingest runs `uv run --no-project --with`, which needs only # the uv binary, never the synced workspace. A full sync would rebuild # vortex-python via sccache->S3, which fails under the ingest-role creds # (rds-db:connect only) and is pure waste here. - name: Install uv for v4 ingest - if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' && matrix.benchmark.v4_ingest uses: spiraldb/actions/.github/actions/setup-uv@a746510eafaa926484c354541cfc49b2ec06cc63 # 0.18.6 with: sync: false - name: Configure AWS credentials for v4 ingest (OIDC) - if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' && matrix.benchmark.v4_ingest uses: aws-actions/configure-aws-credentials@e6de054238d6b7531b4efff3b6587d9aade6a06c # v6 with: role-to-assume: ${{ vars.GH_BENCH_INGEST_ROLE_ARN }} aws-region: ${{ vars.RDS_BENCH_REGION }} - name: Ingest results to v4 Postgres - if: vars.GH_BENCH_INGEST_ROLE_ARN != '' + if: vars.GH_BENCH_INGEST_ROLE_ARN != '' && matrix.benchmark.v4_ingest shell: bash env: RDS_BENCH_INSTANCE_ENDPOINT: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }} diff --git a/Cargo.lock b/Cargo.lock index 25fb98611d0..5be1371fac3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8384,6 +8384,28 @@ dependencies = [ "serde_json", ] +[[package]] +name = "string-bench" +version = "0.1.0" +dependencies = [ + "anyhow", + "bytes", + "clap", + "futures", + "indicatif", + "parquet 58.4.0", + "regex", + "tabled", + "tokio", + "tracing", + "vortex", + "vortex-arrow", + "vortex-bench", + "vortex-btrblocks", + "vortex-fsst", + "vortex-onpair", +] + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index add8f6e8892..5cce58ae60d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,6 +68,7 @@ members = [ "benchmarks/datafusion-bench", "benchmarks/duckdb-bench", "benchmarks/random-access-bench", + "benchmarks/string-bench", "vortex-geo", ] exclude = ["java/testfiles", "wasm-test"] diff --git a/benchmarks/string-bench/Cargo.toml b/benchmarks/string-bench/Cargo.toml new file mode 100644 index 00000000000..e2d3736b195 --- /dev/null +++ b/benchmarks/string-bench/Cargo.toml @@ -0,0 +1,46 @@ +[package] +name = "string-bench" +description = "Vortex string benchmarks for direct compression and serialized write/read round trips" +authors.workspace = true +categories.workspace = true +edition.workspace = true +homepage.workspace = true +include.workspace = true +keywords.workspace = true +license.workspace = true +readme.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true +publish = false + +[dependencies] +anyhow = { workspace = true } +bytes = { workspace = true } +clap = { workspace = true, features = ["derive"] } +futures = { workspace = true } +indicatif = { workspace = true } +parquet = { workspace = true } +regex = { workspace = true } +tabled = { workspace = true, features = ["std"] } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +vortex = { workspace = true } +vortex-arrow = { workspace = true } +vortex-bench = { workspace = true } +vortex-btrblocks = { workspace = true } +vortex-fsst = { workspace = true } +vortex-onpair = { workspace = true } + +[features] +unstable_encodings = [ + "vortex/unstable_encodings", + "vortex-btrblocks/unstable_encodings", +] + +[[bin]] +name = "string-bench" +required-features = ["unstable_encodings"] + +[lints] +workspace = true diff --git a/benchmarks/string-bench/README.md b/benchmarks/string-bench/README.md new file mode 100644 index 00000000000..41f52b65fd1 --- /dev/null +++ b/benchmarks/string-bench/README.md @@ -0,0 +1,112 @@ + + + +# string-bench + +Compares Vortex string encoders on configured real-world Utf8 columns. + +## What it measures + +Three metrics per (column, encoder), from the default `vortex` suite: + +| Metric | Meaning | Table unit | `gh-json` unit | +| --- | --- | --- | --- | +| `size` | Serialized file bytes over canonical uncompressed bytes | % | % | +| `write` | Repartition + zone stats + compress (string scheme and children) + layout + serialize | MB/s | ms | +| `read` | Open + scan, decoding each row split to canonical form | MB/s | ms | + +Values are the median of `--iterations` runs. The `gh-json` output reports +durations in milliseconds so that lower is better for every emitted metric; the +human table reports MB/s over canonical uncompressed bytes. + +### The canonical baseline + +Every metric normalizes against canonical uncompressed bytes: one 16-byte view +per row plus the bytes of the strings too long to inline, so a string of 12 +bytes or fewer costs only its view. The codec suite uses the same baseline. + +### How faithful the timings are + +Write and read run the real Vortex writer and scan on a current-thread runtime +with no worker pool, so both timings are single-threaded CPU costs. + +`read` fuses the canonical decode into each row split's scan task — the shape +production uses, where `into_record_batch_stream` fuses the Arrow conversion — +and drops each decoded chunk before the next split runs, so steady-state memory +is one chunk, not the whole column. Splits are awaited one at a time rather than +through the scan's own stream, which spawns `concurrency * +available_parallelism()` of them at once; on one thread that read-ahead buys no +parallelism, it only holds more chunks in memory and ties the result to the +host's core count. + +Excluded by design: + +- **Physical I/O** — the file lives in a `Vec` and `open_buffer` slices it, + so there is no read-driver request coalescing, no segment cache, and none of + the copying a real file or object-store read does. +- **Parallel scan throughput** — production spreads splits across workers, so + these per-thread costs do not translate directly into query time. +- **Filters and projections** — the whole column is read with no predicate, so + `read` is the full-decode cost, not decode-with-mask or compute pushdown. + +`size` covers encoded children, metadata, padding, and file markers. + +### Codec diagnostic + +`--suite codec` runs a separate microbenchmark: one whole-column encoder call, +with no Vortex layout, child compression, serialization, or I/O. + +It is not part of the tracked set. Its size metric is also not interchangeable +with the file suite's: it trains one dictionary or symbol table for the entire +column and leaves the encoded array's children uncompressed, whereas Vortex +writes in chunks and compresses those children. + +## Inputs and local data + +The current input catalog is: + +| Output name | Source | +| --- | --- | +| `clickbench/URL/shard-N` | ClickBench `hits_N.parquet`, `URL` column | +| `tpch/l_comment` | TPC-H SF1 `lineitem`, `l_comment` column | + +On first use, the selected ClickBench shard is downloaded through +`vortex-bench`'s shared idempotent downloader. TPC-H SF1 is generated +deterministically through the shared TPC-H dataset helper. Both are stored +under `vortex-bench/data/`, which is gitignored, and reused on later runs. +Input preparation is outside benchmark timing. + +## Running locally + +```bash +# Tracked metrics: size, write, read for every configured column and encoder. +cargo run -p string-bench --profile release_debug --features unstable_encodings + +# Focus on selected columns or encoders. +cargo run -p string-bench --profile release_debug --features unstable_encodings -- \ + --columns URL --encoders onpair + +# Add the direct codec microbenchmark. +cargo run -p string-bench --profile release_debug --features unstable_encodings -- \ + --suite both + +# Emit benchmark-comparator JSONL. +cargo run -p string-bench --profile release_debug --features unstable_encodings -- \ + --display-format gh-json --output-path results.json +``` + +Run `cargo run -p string-bench --features unstable_encodings -- --help` for all +filters and tuning options. + +Before timing, the benchmark checks that each requested encoding was produced +and, unless `--no-verify` is set, compares decoded output with the input. + +## CI + +The develop benchmark workflow runs the default suite after each merge to +`develop` and publishes the results to the shared benchmark history. + +Metric names are `//`, for example +`read/clickbench/URL/shard-0/onpair-12` and `size/tpch/l_comment/fsst`. The unit +is reported separately in the JSON output and the CI table, which groups rows by +unit — so metric-first names keep the rows you compare adjacent. diff --git a/benchmarks/string-bench/src/codec.rs b/benchmarks/string-bench/src/codec.rs new file mode 100644 index 00000000000..dc8364b73d1 --- /dev/null +++ b/benchmarks/string-bench/src/codec.rs @@ -0,0 +1,276 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Direct array-level codec microbenchmarks. +//! +//! This module deliberately measures the selected encoder's train + compress +//! path over one canonical whole-column array. It does not exercise the Vortex +//! file layout, per-chunk dictionaries, child compression, or file I/O. + +use std::hint::black_box; +use std::time::Duration; +use std::time::Instant; + +use anyhow::Result; +use anyhow::bail; +use vortex::array::ArrayRef; +use vortex::array::ExecutionCtx; +use vortex::array::IntoArray; +use vortex::array::VortexSessionExecute; +use vortex::array::arrays::VarBinViewArray; +use vortex_bench::Format; +use vortex_bench::measurements::CustomUnitMeasurement; +use vortex_fsst::fsst_compress; +use vortex_fsst::fsst_train_compressor; +use vortex_onpair::Config; +use vortex_onpair::DEFAULT_DICT12_CONFIG; +use vortex_onpair::MaxDictBits; +use vortex_onpair::onpair_compress; + +use crate::SESSION; +use crate::StringColumn; +use crate::StringEncoder; +use crate::duration_ms; +use crate::median; +use crate::onpair_label; +use crate::prepare_column; +use crate::throughput; +use crate::verify_canonicalized; + +/// A fully configured direct array-compression candidate: an encoder family +/// plus the fixed configuration that path needs. +pub enum DirectCandidate { + /// OnPair with a deterministic config, whose dictionary budget is the only + /// thing the benchmark varies. + OnPair(Config), + /// FSST (no configuration). + Fsst, +} + +impl DirectCandidate { + /// OnPair with a dictionary budget of up to `2^max_dict_bits` tokens, + /// reusing every other default. `max_dict_bits` must be in `9..=16`; OnPair + /// stores codes as `u16`, so all of those fit. + pub fn on_pair(max_dict_bits: u8) -> Result { + let max_dict_bits = MaxDictBits::new(max_dict_bits).map_err(|e| { + anyhow::anyhow!( + "invalid maximum dictionary bit width {max_dict_bits} (want 9..=16): {e}" + ) + })?; + Ok(Self::OnPair(Config { + max_dict_bits, + ..DEFAULT_DICT12_CONFIG + })) + } + + /// The encoder family, for the post-compression type check. + fn family(&self) -> StringEncoder { + match self { + Self::OnPair(_) => StringEncoder::OnPair, + Self::Fsst => StringEncoder::Fsst, + } + } + + /// Stable label used in benchmark output, e.g. `onpair-12` or `fsst`. + fn label(&self) -> String { + match self { + Self::OnPair(config) => onpair_label(config), + Self::Fsst => StringEncoder::Fsst.label().to_string(), + } + } + + /// Run the encoder's direct array-level train + compress path. + pub(crate) fn compress(&self, array: &ArrayRef, ctx: &mut ExecutionCtx) -> Result { + match self { + Self::OnPair(config) => Ok(onpair_compress(array, *config, ctx)?), + Self::Fsst => { + let compressor = fsst_train_compressor(array, ctx)?; + Ok(fsst_compress(array, &compressor, ctx)?.into_array()) + } + } + } +} + +/// Measured in-memory statistics for a single string column under one encoder +/// configuration (an OnPair dictionary size, or FSST). +pub struct ColumnResult { + /// Column identifier. + pub name: String, + /// Encoder configuration label, e.g. `onpair-12` or `fsst`. + pub encoder: String, + /// Number of rows. + pub rows: usize, + /// Canonical uncompressed array bytes used to normalize size and + /// throughput: one 16-byte view per row plus the bytes of the strings too + /// long to inline. + pub uncompressed_bytes: u64, + /// Buffer bytes referenced by the encoded array. + pub encoded_bytes: u64, + /// Direct array-level train + compress times, one per iteration. + pub compression_runs: Vec, +} + +impl ColumnResult { + /// Median direct array compression time across iterations. + fn compression_median(&self) -> Duration { + median(&self.compression_runs) + } + + /// Encoded array buffer bytes as a percentage of canonical uncompressed + /// array bytes. Lower is better. + /// + /// This direct representation has not had its children compressed by the + /// file writer. + pub fn encoded_size_pct(&self) -> f64 { + self.encoded_bytes as f64 / self.uncompressed_bytes as f64 * 100.0 + } + + /// Direct compression throughput in MB/s of canonical uncompressed array + /// bytes, from the median run. + pub fn compression_mbps(&self) -> f64 { + throughput(self.uncompressed_bytes, self.compression_median()) / 1e6 + } + + /// Emit lower-is-better timings and size percentages as Vortex custom-unit + /// metrics, named `codec///`. + /// + /// The `codec/` prefix keeps these distinct from the tracked file metrics, + /// which CI reports: this suite is a local diagnostic, so its measurements + /// only reach `gh-json` when it is explicitly selected. + /// + /// `Format` has no in-memory Vortex variant, so these measurements use + /// `Format::OnDiskVortex` as their reporting target. + pub fn measurements(&self) -> Vec { + let suffix = format!("{}/{}", self.name, self.encoder); + vec![ + CustomUnitMeasurement { + name: format!("codec/size/{suffix}"), + format: Format::OnDiskVortex, + unit: "%".into(), + value: self.encoded_size_pct(), + }, + CustomUnitMeasurement { + name: format!("codec/compress/{suffix}"), + format: Format::OnDiskVortex, + unit: "ms".into(), + value: duration_ms(self.compression_median()), + }, + ] + } +} + +/// Time `candidate`'s direct array-level train + compress path on `column` across +/// `iterations` runs, recording every run so the caller can report the median. +/// +/// Fails if the column contains nulls or does not compress to `candidate`'s +/// encoding, rather than silently changing the measured workload. +pub fn bench_column( + column: &StringColumn, + iterations: usize, + warmup: usize, + candidate: &DirectCandidate, + verify: bool, +) -> Result { + crate::validate_iterations(iterations)?; + let mut ctx = SESSION.create_execution_ctx(); + + let (canonical, uncompressed_bytes) = prepare_column(column, &mut ctx)?; + let input = canonical.clone().into_array(); + let rows = canonical.len(); + + // At least one warm-up produces the reference array for the encoding check, + // size metric, and canonicalization verification. Extra runs stabilize timings. + let mut compressed = candidate.compress(&input, &mut ctx)?; + for _ in 1..warmup.max(1) { + let mut warm_ctx = SESSION.create_execution_ctx(); + compressed = candidate.compress(&input, &mut warm_ctx)?; + } + + if !candidate.family().matches(&compressed) { + bail!( + "column {} did not compress to {} (got {}); its data may be unsupported \ + by the encoder", + column.name, + candidate.label(), + compressed.encoding_id(), + ); + } + let encoded_bytes = compressed.nbytes(); + + // One-time correctness check before timing: a byte-wrong canonicalization + // must not be reported as a fast one. + if verify { + let canonicalized = compressed.execute::(&mut ctx)?; + verify_canonicalized( + &format!("{} [{}]", column.name, candidate.label()), + &canonical, + &canonicalized, + &mut ctx, + )?; + } + + // Each timed run re-compresses the source array from scratch; the fresh + // context is cheap per-run isolation, not a cache reset (`ExecutionCtx` + // holds no cross-run result cache). `black_box` keeps the result live. + let mut compression_runs = Vec::with_capacity(iterations); + for _ in 0..iterations { + let mut ctx = SESSION.create_execution_ctx(); + let start = Instant::now(); + let compressed = candidate.compress(&input, &mut ctx)?; + compression_runs.push(start.elapsed()); + black_box(compressed); + } + + Ok(ColumnResult { + name: column.name.clone(), + encoder: candidate.label(), + rows, + uncompressed_bytes, + encoded_bytes, + compression_runs, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn direct_benchmark_smoke_tests_each_encoder() -> Result<()> { + let (column, expected_uncompressed_bytes) = crate::repeated_fixture(); + + for (candidate, expected_label) in [ + (DirectCandidate::on_pair(12)?, "onpair-12"), + (DirectCandidate::Fsst, "fsst"), + ] { + let result = bench_column(&column, 1, 0, &candidate, true)?; + + assert_eq!(result.encoder, expected_label); + assert_eq!(result.rows, 128); + assert_eq!(result.uncompressed_bytes, expected_uncompressed_bytes); + assert!(result.encoded_bytes > 0); + assert_eq!(result.compression_runs.len(), 1); + } + Ok(()) + } + + #[test] + fn machine_readable_metric_schema_is_stable() { + let result = ColumnResult { + name: "fixture".to_string(), + encoder: "fsst".to_string(), + rows: 1, + uncompressed_bytes: 2, + encoded_bytes: 1, + compression_runs: vec![Duration::from_millis(12)], + }; + + assert_eq!( + crate::measurement_rows(&result.measurements()), + [ + "codec/size/fixture/fsst % 50", + "codec/compress/fixture/fsst ms 12", + ] + ); + } +} diff --git a/benchmarks/string-bench/src/lib.rs b/benchmarks/string-bench/src/lib.rs new file mode 100644 index 00000000000..52fd05f7943 --- /dev/null +++ b/benchmarks/string-bench/src/lib.rs @@ -0,0 +1,497 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +#![cfg(feature = "unstable_encodings")] + +//! String-compression benchmarks for Vortex. +//! +//! Two intentionally separate suites: +//! +//! * `serialized` is the default and the only one CI tracks. It writes a column +//! to an in-memory Vortex file with one encoder forced, then reads it back, +//! reporting `size`, `write`, and `read`. +//! * `codec` is a local diagnostic (`--suite codec`) that times one encoder's +//! array-level train + compress call and measures the encoded array's buffer +//! bytes, with no layout, child compression, or I/O in the way. Use it to tell +//! an encoder-level change apart from a change in the file stack. +//! +//! Both suites force the requested encoder rather than letting the btrblocks +//! selector pick, so neither can silently measure a different encoding. Their +//! size metrics are still not interchangeable: the codec path trains one state +//! over the whole column and leaves children uncompressed, while the file path +//! trains one per compression chunk and compresses children. See the README +//! ("Codec diagnostic"). + +use std::path::PathBuf; +use std::sync::LazyLock; +use std::time::Duration; + +use anyhow::Result; +use anyhow::bail; +use clap::ValueEnum; +use futures::StreamExt; +use futures::TryStreamExt; +use parquet::arrow::ParquetRecordBatchStreamBuilder; +use parquet::arrow::ProjectionMask; +use tokio::fs::File; +use vortex::VortexSessionDefault; +use vortex::aggregate_fn::fns::uncompressed_size_in_bytes::uncompressed_size_in_bytes; +use vortex::array::ArrayRef; +use vortex::array::ExecutionCtx; +use vortex::array::IntoArray; +use vortex::array::arrays::ChunkedArray; +use vortex::array::arrays::StructArray; +use vortex::array::arrays::VarBinViewArray; +use vortex::array::arrays::struct_::StructArrayExt; +use vortex::io::session::RuntimeSessionExt; +use vortex::session::VortexSession; +use vortex_arrow::FromArrowArray; +use vortex_bench::Format; +use vortex_bench::IdempotentPath; +use vortex_bench::datasets::Dataset; +use vortex_bench::datasets::data_downloads::download_data; +use vortex_bench::datasets::tpch_l_comment::TPCHLCommentCanonical; +use vortex_fsst::FSST; +use vortex_onpair::Config; +use vortex_onpair::OnPair; + +mod codec; +pub use codec::ColumnResult; +pub use codec::DirectCandidate; +pub use codec::bench_column; + +const CLICKBENCH_SHARD_COUNT: u32 = 100; +const CLICKBENCH_URL_PREFIX: &str = + "https://pub-3ba949c0f0354ac18db1f0f14f0a2c52.r2.dev/clickbench/parquet_many/"; + +/// Serialized write and read benchmark: write → open → scan, with each row +/// split decoded to canonical form inside its own scan task. +mod serialized; +pub use serialized::*; + +/// Session with the available string encodings and their canonicalize kernels +/// registered for benchmarking. +pub static SESSION: LazyLock = + LazyLock::new(|| VortexSession::default().with_tokio()); + +/// A named string column, canonicalized to a `VarBinViewArray`-backed array, +/// ready for either benchmark path. +pub struct StringColumn { + /// Human-readable column identifier used in measurement names. + pub name: String, + /// Canonical (`VarBinViewArray`) Utf8 data. + pub array: ArrayRef, +} + +/// String encoder selected by the benchmark. +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +pub enum StringEncoder { + /// The OnPair encoder. + #[value(name = "onpair")] + OnPair, + /// The FSST encoder. + #[value(name = "fsst")] + Fsst, +} + +impl StringEncoder { + /// Stable short label used in benchmark output. + pub fn label(self) -> &'static str { + match self { + Self::OnPair => "onpair", + Self::Fsst => "fsst", + } + } + + /// Whether `array` is this encoder's Vortex encoding. Used to confirm a + /// column was compressed (or serialized) with the requested scheme. + pub fn matches(self, array: &ArrayRef) -> bool { + match self { + Self::OnPair => array.as_opt::().is_some(), + Self::Fsst => array.as_opt::().is_some(), + } + } +} + +impl std::fmt::Display for StringEncoder { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.label()) + } +} + +/// Label for an OnPair configuration, e.g. `onpair-12`. The dictionary budget is +/// read from the config that was actually used, so a label can never claim a +/// budget the measured data was not encoded with. +pub(crate) fn onpair_label(config: &Config) -> String { + format!("onpair-{}", config.max_dict_bits.value()) +} + +/// Bytes per second for `bytes` processed in `elapsed`. +pub(crate) fn throughput(bytes: u64, elapsed: Duration) -> f64 { + let secs = elapsed.as_secs_f64(); + if secs <= 0.0 { + 0.0 + } else { + bytes as f64 / secs + } +} + +/// Duration in milliseconds for lower-is-better machine-readable output. +pub(crate) fn duration_ms(duration: Duration) -> f64 { + duration.as_secs_f64() * 1e3 +} + +fn validate_iterations(iterations: usize) -> Result<()> { + if iterations == 0 { + bail!("iterations must be greater than zero"); + } + Ok(()) +} + +/// Median of `runs` (the two middle values are averaged for an even count). +/// Empty → zero. +pub(crate) fn median(runs: &[Duration]) -> Duration { + if runs.is_empty() { + return Duration::ZERO; + } + let mut sorted = runs.to_vec(); + sorted.sort_unstable(); + let n = sorted.len(); + if n % 2 == 1 { + sorted[n / 2] + } else { + (sorted[n / 2 - 1] + sorted[n / 2]) / 2 + } +} + +/// Return the non-empty, all-valid Utf8 input in compact canonical form, plus +/// the canonical baseline every metric normalizes against: one 16-byte view per +/// row plus the bytes of the strings too long to inline. Preparation and size +/// accounting are outside all timed regions. +pub(crate) fn prepare_column( + column: &StringColumn, + ctx: &mut ExecutionCtx, +) -> Result<(VarBinViewArray, u64)> { + let canonical = column.array.clone().execute::(ctx)?; + if !canonical.dtype().is_utf8() { + bail!( + "column {} has dtype {}; string-bench requires Utf8 input", + column.name, + canonical.dtype(), + ); + } + let valid_count = canonical + .validity()? + .execute_mask(canonical.len(), ctx)? + .true_count(); + if valid_count != canonical.len() { + bail!( + "column {} contains {} null strings; string-bench requires all rows to be valid", + column.name, + canonical.len() - valid_count, + ); + } + + // Use an aggressively compacted array so the baseline cannot include + // retained backing-buffer regions that the compressors never see. + let canonical = canonical.compact_with_threshold(1.0, ctx)?; + let uncompressed_bytes = u64::try_from(uncompressed_size_in_bytes(canonical.as_ref(), ctx)?)?; + if uncompressed_bytes == 0 { + bail!("column {} has zero uncompressed bytes", column.name); + } + + Ok((canonical, uncompressed_bytes)) +} + +/// Assert the canonicalized column matches the original row-for-row (dtype, +/// length, validity, and bytes). This runs once before timing. +pub(crate) fn verify_canonicalized( + label: &str, + expected: &VarBinViewArray, + canonicalized: &VarBinViewArray, + ctx: &mut ExecutionCtx, +) -> Result<()> { + if canonicalized.dtype() != expected.dtype() { + bail!( + "{label}: canonicalized dtype {} != original {}", + canonicalized.dtype(), + expected.dtype(), + ); + } + + let len = expected.len(); + if canonicalized.len() != len { + bail!( + "{label}: canonicalized row count {} != original {len}", + canonicalized.len(), + ); + } + // Materialize both validity masks up front so the per-row `value(i)` and + // `bytes_at(i)` below are O(1); this one-shot check stays off the timed path. + let expected_valid = expected.validity()?.execute_mask(len, ctx)?; + let canonicalized_valid = canonicalized.validity()?.execute_mask(len, ctx)?; + for i in 0..len { + let valid = expected_valid.value(i); + if valid != canonicalized_valid.value(i) { + bail!("{label}: validity mismatch at row {i}"); + } + if valid && expected.bytes_at(i).as_slice() != canonicalized.bytes_at(i).as_slice() { + bail!("{label}: canonicalized value differs from input at row {i}"); + } + } + tracing::debug!("{label}: canonicalized output verified ({len} rows)"); + Ok(()) +} + +/// Canonicalize `array` (a struct, possibly chunked) and pull out `field` as a +/// canonical `VarBinViewArray`-backed Utf8 column. +fn to_utf8_column(array: ArrayRef, field: &str, ctx: &mut ExecutionCtx) -> Result { + let structs = array.execute::(ctx)?; + let column = structs.unmasked_field_by_name(field)?.clone(); + Ok(column.execute::(ctx)?.into_array()) +} + +/// Load the ClickBench `hits` `URL` column from shard `shard`, downloading the +/// shard first if it is not already present locally. +pub async fn load_clickbench_url(shard: u32, ctx: &mut ExecutionCtx) -> Result { + let path = download_clickbench_shard(shard).await?; + let struct_array = read_parquet_projected(path, "URL").await?; + Ok(StringColumn { + name: format!("clickbench/URL/shard-{shard}"), + array: to_utf8_column(struct_array, "URL", ctx)?, + }) +} + +async fn download_clickbench_shard(shard: u32) -> Result { + if shard >= CLICKBENCH_SHARD_COUNT { + bail!( + "invalid ClickBench shard {shard} (want 0..{})", + CLICKBENCH_SHARD_COUNT - 1 + ); + } + let filename = format!("hits_{shard}.parquet"); + let path = "clickbench_partitioned" + .to_data_path() + .join(Format::Parquet.name()) + .join(&filename); + download_data(path, format!("{CLICKBENCH_URL_PREFIX}{filename}")).await +} + +/// Load the TPC-H `l_comment` column (from the first `lineitem` parquet shard), +/// generating the TPC-H data if needed. +pub async fn load_tpch_l_comment(ctx: &mut ExecutionCtx) -> Result { + let path = TPCHLCommentCanonical.to_parquet_path().await?; + let struct_array = read_parquet_projected(path, "l_comment").await?; + Ok(StringColumn { + name: "tpch/l_comment".to_string(), + array: to_utf8_column(struct_array, "l_comment", ctx)?, + }) +} + +/// Read a single column from a parquet file (projected at the parquet level to +/// avoid decoding the other columns) into a chunked struct array. +async fn read_parquet_projected(path: PathBuf, column: &str) -> Result { + let file = File::open(&path).await?; + let builder = ParquetRecordBatchStreamBuilder::new(file).await?; + let col_idx = builder.schema().index_of(column)?; + let mask = ProjectionMask::roots(builder.parquet_schema(), [col_idx]); + let reader = builder.with_projection(mask).build()?; + + let chunks: Vec = reader + .map(|batch| { + batch + .map_err(anyhow::Error::from) + .and_then(|rb| ArrayRef::from_arrow(rb, false).map_err(anyhow::Error::from)) + }) + .try_collect() + .await?; + + Ok(ChunkedArray::from_iter(chunks).into_array()) +} + +/// 128 rows cycling over 8 distinct outlined values, plus the canonical +/// uncompressed size they must account for (one 16-byte view per row). +#[cfg(test)] +fn repeated_fixture() -> (StringColumn, u64) { + let values = (0..128) + .map(|i| { + format!( + "https://example.com/path/to/a/repeated/string-value-{}", + i % 8 + ) + }) + .collect::>(); + let outlined_bytes = values.iter().map(|value| value.len() as u64).sum::(); + ( + StringColumn { + name: "fixture".to_string(), + array: VarBinViewArray::from_iter_str(values.iter()).into_array(), + }, + 128 * 16 + outlined_bytes, + ) +} + +/// Compact `(name, unit, value)` rendering of emitted measurements, for the +/// metric-schema tests that guard the tracked CI names. +#[cfg(test)] +pub(crate) fn measurement_rows( + measurements: &[vortex_bench::measurements::CustomUnitMeasurement], +) -> Vec { + measurements + .iter() + .map(|m| format!("{} {} {}", m.name, m.unit, m.value)) + .collect() +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use vortex::array::VortexSessionExecute; + + use super::*; + + /// The size baseline every metric normalizes against: one 16-byte view per + /// row plus the bytes of the strings too long to inline, counting neither + /// buffer regions a slice merely retains nor anything but the live rows. + #[test] + fn canonical_baseline_counts_views_and_live_outlined_bytes() -> Result<()> { + let kept = "second outlined string kept by the slice"; + let chunk_tail = "an outlined string in the second chunk"; + let cases = [ + ( + "inline and outlined", + VarBinViewArray::from_iter_str(["cat", "hello", "abcdefghijklmnop"]).into_array(), + 3, + 3 * 16 + 16, + ), + ( + "sliced", + VarBinViewArray::from_iter_str([ + "first outlined string dropped by the slice", + kept, + ]) + .into_array() + .slice(1..2)?, + 1, + 16 + kept.len() as u64, + ), + ( + "chunked", + ChunkedArray::from_iter([ + VarBinViewArray::from_iter_str(["alpha", ""]).into_array(), + VarBinViewArray::from_iter_str([chunk_tail]).into_array(), + ]) + .into_array(), + 3, + 3 * 16 + chunk_tail.len() as u64, + ), + ]; + let mut ctx = SESSION.create_execution_ctx(); + + for (name, array, rows, expected_bytes) in cases { + let column = StringColumn { + name: name.to_string(), + array, + }; + let (canonical, uncompressed_bytes) = prepare_column(&column, &mut ctx)?; + + assert_eq!(canonical.len(), rows, "{name} row count"); + assert_eq!(uncompressed_bytes, expected_bytes, "{name} size"); + } + Ok(()) + } + + #[test] + fn verify_canonicalized_rejects_mismatches() { + let cases = [ + ( + "dtype", + VarBinViewArray::from_iter_str(["alpha"]), + VarBinViewArray::from_iter_bin(["alpha".as_bytes()]), + ), + ( + "length", + VarBinViewArray::from_iter_str(["alpha", "beta"]), + VarBinViewArray::from_iter_str(["alpha"]), + ), + ( + "validity", + VarBinViewArray::from_iter_nullable_str([Some("alpha"), Some("beta")]), + VarBinViewArray::from_iter_nullable_str([Some("alpha"), None]), + ), + ( + "value", + VarBinViewArray::from_iter_str(["alpha", "beta"]), + VarBinViewArray::from_iter_str(["alpha", "BETA"]), + ), + ]; + let mut ctx = SESSION.create_execution_ctx(); + + for (kind, expected, actual) in cases { + assert!( + verify_canonicalized("fixture", &expected, &actual, &mut ctx).is_err(), + "{kind} mismatch must be rejected" + ); + } + } + + #[test] + fn median_is_stable_for_odd_and_even_runs() { + let runs = [ + Duration::from_nanos(30), + Duration::from_nanos(10), + Duration::from_nanos(20), + ]; + assert_eq!(median(&runs), Duration::from_nanos(20)); + + let runs = [Duration::from_nanos(10), Duration::from_nanos(30)]; + assert_eq!(median(&runs), Duration::from_nanos(20)); + assert_eq!(median(&[]), Duration::ZERO); + } + + #[test] + fn iterations_must_be_positive() -> Result<()> { + assert!(validate_iterations(0).is_err()); + validate_iterations(1) + } + + #[tokio::test] + async fn clickbench_shard_rejects_out_of_range() { + assert!( + download_clickbench_shard(CLICKBENCH_SHARD_COUNT) + .await + .is_err() + ); + } + + #[test] + fn prepare_column_rejects_invalid_input() { + let cases = [ + ( + "null", + VarBinViewArray::from_iter_nullable_str([Some("alpha"), None]).into_array(), + ), + ( + "empty", + VarBinViewArray::from_iter_str(std::iter::empty::<&str>()).into_array(), + ), + ( + "binary", + VarBinViewArray::from_iter_bin([b"alpha".as_slice()]).into_array(), + ), + ]; + let mut ctx = SESSION.create_execution_ctx(); + + for (name, array) in cases { + let column = StringColumn { + name: name.to_string(), + array, + }; + assert!( + prepare_column(&column, &mut ctx).is_err(), + "{name} input must be rejected" + ); + } + } +} diff --git a/benchmarks/string-bench/src/main.rs b/benchmarks/string-bench/src/main.rs new file mode 100644 index 00000000000..d03304d14c7 --- /dev/null +++ b/benchmarks/string-bench/src/main.rs @@ -0,0 +1,395 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use std::io::Write; +use std::num::NonZeroUsize; +use std::path::PathBuf; + +use anyhow::Result; +use anyhow::bail; +use clap::Parser; +use clap::ValueEnum; +use indicatif::ProgressBar; +use regex::Regex; +use string_bench::ColumnResult; +use string_bench::DirectCandidate; +use string_bench::SESSION; +use string_bench::SerializedResult; +use string_bench::StringColumn; +use string_bench::StringEncoder; +use string_bench::bench_column; +use string_bench::bench_serialized_with_session; +use string_bench::load_clickbench_url; +use string_bench::load_tpch_l_comment; +use tabled::builder::Builder; +use tabled::settings::Style; +use vortex::VortexSessionDefault; +use vortex::array::ExecutionCtx; +use vortex::array::VortexSessionExecute; +use vortex::io::runtime::BlockingRuntime; +use vortex::io::runtime::current::CurrentThreadRuntime; +use vortex::io::session::RuntimeSessionExt; +use vortex::session::VortexSession; +use vortex_bench::LogFormat; +use vortex_bench::create_output_writer; +use vortex_bench::display::DisplayFormat; +use vortex_bench::display::print_measurements_json; +use vortex_bench::measurements::CustomUnitMeasurement; +use vortex_bench::setup_logging_and_tracing_with_format; + +/// The benchmark ID used for the output path. +const BENCHMARK_ID: &str = "string"; + +/// Repo-relative path of the suite explainer linked from CI benchmark PR comments. +const DOC_PATH: &str = "benchmarks/string-bench/README.md"; + +const CODEC_ONPAIR_DICT_BITS: [u8; 2] = [12, 16]; + +/// The configured benchmark inputs. Add a variant to extend the catalog. +enum Input { + /// The `URL` column of one ClickBench `hits` shard. + ClickBenchUrl(u32), + /// The TPC-H SF1 `lineitem.l_comment` column. + TpchLComment, +} + +impl Input { + /// Name the `--columns` regex is matched against. + fn name(&self) -> &'static str { + match self { + Self::ClickBenchUrl(_) => "URL", + Self::TpchLComment => "l_comment", + } + } + + async fn load(&self, ctx: &mut ExecutionCtx) -> Result { + match *self { + Self::ClickBenchUrl(shard) => load_clickbench_url(shard, ctx).await, + Self::TpchLComment => load_tpch_l_comment(ctx).await, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +enum BenchmarkSuite { + /// Run only the Vortex file write/read benchmark: the tracked suite. + #[value(name = "vortex")] + Vortex, + /// Run only the direct whole-column codec microbenchmark: a local + /// diagnostic for separating encoder cost from the Vortex file stack. + #[value(name = "codec")] + Codec, + /// Run both benchmark paths. + #[value(name = "both")] + Both, +} + +#[derive(Parser, Debug)] +#[command(version, about, long_about = None)] +struct Args { + /// Benchmark path to run: the tracked Vortex file suite, the direct codec + /// diagnostic, or both. + #[arg(long, value_enum, default_value_t = BenchmarkSuite::Vortex)] + suite: BenchmarkSuite, + /// Timed runs per (column, encoder, config); the median is reported. + #[arg(short, long, default_value = "10")] + iterations: NonZeroUsize, + /// Untimed warm-up runs before timing, per (column, encoder, config). At + /// least one warm-up always runs; correctness validation is also untimed. + #[arg(long, default_value_t = 3)] + warmup: usize, + /// Regex filter matched against configured input-column identifiers. + #[arg(long)] + columns: Option, + /// ClickBench shard index to read the `URL` column from. + #[arg(long, default_value_t = 0)] + clickbench_shard: u32, + /// Skip the one-time canonicalized-output correctness check before timing. + #[arg(long)] + no_verify: bool, + /// Encoders to benchmark (comma-separated). + #[arg(long, value_delimiter = ',', default_values_t = vec![ + StringEncoder::OnPair, + StringEncoder::Fsst, + ])] + encoders: Vec, + /// Output format: `table` for humans, `gh-json` for machine-readable JSONL. + #[arg(short, long, default_value_t, value_enum)] + display_format: DisplayFormat, + /// Write output to this file instead of stdout. + #[arg(short, long)] + output_path: Option, + /// Enable verbose (debug-level) logging. + #[arg(short, long)] + verbose: bool, + /// Enable span tracing output. + #[arg(long)] + tracing: bool, + /// Format for the primary stderr log sink. + #[arg(long, value_enum, default_value_t = LogFormat::Text)] + log_format: LogFormat, +} + +fn contains_duplicates(values: &[T]) -> bool { + values + .iter() + .enumerate() + .any(|(index, value)| values[..index].contains(value)) +} + +fn main() -> Result<()> { + let args = Args::parse(); + setup_logging_and_tracing_with_format(args.verbose, args.tracing, args.log_format)?; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + runtime.block_on(run(args)) +} + +async fn run(args: Args) -> Result<()> { + let iterations = args.iterations.get(); + let run_codec = matches!(args.suite, BenchmarkSuite::Codec | BenchmarkSuite::Both); + let run_vortex = matches!(args.suite, BenchmarkSuite::Vortex | BenchmarkSuite::Both); + + if args.encoders.is_empty() { + bail!("--encoders must select at least one encoder"); + } + if contains_duplicates(&args.encoders) { + bail!("--encoders must not contain duplicates"); + } + // The codec suite sweeps OnPair dictionary budgets; the file suite cannot, + // since btrblocks picks the budget it compresses with. + let mut candidates: Vec = Vec::new(); + if run_codec { + if args.encoders.contains(&StringEncoder::OnPair) { + for bits in CODEC_ONPAIR_DICT_BITS { + candidates.push(DirectCandidate::on_pair(bits)?); + } + } + if args.encoders.contains(&StringEncoder::Fsst) { + candidates.push(DirectCandidate::Fsst); + } + } + + let filter = args.columns.as_deref().map(Regex::new).transpose()?; + let mut ctx = SESSION.create_execution_ctx(); + let mut columns = Vec::new(); + for input in [ + Input::ClickBenchUrl(args.clickbench_shard), + Input::TpchLComment, + ] + .into_iter() + .filter(|input| filter.as_ref().is_none_or(|f| f.is_match(input.name()))) + { + columns.push(input.load(&mut ctx).await?); + } + if columns.is_empty() { + bail!("no input columns matched the --columns filter"); + } + + let verify = !args.no_verify; + + // `candidates` is empty unless the codec suite runs. + let steps_per_column = candidates.len() + if run_vortex { args.encoders.len() } else { 0 }; + let progress = ProgressBar::new((columns.len() * steps_per_column) as u64); + + let mut column_results: Vec = Vec::new(); + let mut serialized_results: Vec = Vec::new(); + + for column in &columns { + if run_codec { + column_results.extend(run_codec_column( + column, + iterations, + args.warmup, + verify, + &candidates, + &progress, + )?); + } + + if run_vortex { + serialized_results.extend(run_vortex_column( + column, + iterations, + args.warmup, + verify, + &args.encoders, + &progress, + )?); + } + } + progress.finish(); + + let mut writer = create_output_writer(&args.display_format, args.output_path, BENCHMARK_ID)?; + match args.display_format { + DisplayFormat::Table => { + if !serialized_results.is_empty() { + render_serialized_table(&mut writer, &serialized_results)?; + } + if !column_results.is_empty() { + render_codec_table(&mut writer, &column_results)?; + } + } + DisplayFormat::GhJson => { + let mut measurements: Vec = Vec::new(); + for result in &serialized_results { + measurements.extend(result.measurements()); + } + for result in &column_results { + measurements.extend(result.measurements()); + } + print_measurements_json(&mut writer, measurements, DOC_PATH)?; + } + } + + Ok(()) +} + +fn run_codec_column( + column: &StringColumn, + iterations: usize, + warmup: usize, + verify: bool, + candidates: &[DirectCandidate], + progress: &ProgressBar, +) -> Result> { + let mut results = Vec::with_capacity(candidates.len()); + for candidate in candidates { + let result = bench_column(column, iterations, warmup, candidate, verify)?; + tracing::info!( + "{} [{}]: {} rows, {:.3} GB canonical, size {:.2}%, compress {:.2} MB/s", + result.name, + result.encoder, + result.rows, + result.uncompressed_bytes as f64 / 1e9, + result.encoded_size_pct(), + result.compression_mbps(), + ); + results.push(result); + progress.inc(1); + } + Ok(results) +} + +fn run_vortex_column( + column: &StringColumn, + iterations: usize, + warmup: usize, + verify: bool, + encoders: &[StringEncoder], + progress: &ProgressBar, +) -> Result> { + let runtime = CurrentThreadRuntime::new(); + let session = VortexSession::default().with_handle(runtime.handle()); + let mut results = Vec::with_capacity(encoders.len()); + for &encoder in encoders { + let result = runtime.block_on(bench_serialized_with_session( + &session, column, iterations, warmup, verify, encoder, + ))?; + tracing::info!( + "{} [{}]: size {:.2}% | write {:.2} MB/s | read {:.2} MB/s", + result.name, + result.encoder, + result.size_pct(), + result.write_mbps(), + result.read_mbps(), + ); + results.push(result); + progress.inc(1); + } + Ok(results) +} + +/// Render one titled table, prefixed by a one-line note on how to read its +/// metrics. +fn render_table( + writer: &mut dyn Write, + title: &str, + note: &str, + header: &[&str], + rows: Vec>, +) -> Result<()> { + let mut builder = Builder::default(); + builder.push_record(header.iter().copied()); + for row in rows { + builder.push_record(row); + } + let mut table = builder.build(); + table.with(Style::modern()); + writeln!(writer, "{title}\n {note}")?; + writeln!(writer, "{table}")?; + Ok(()) +} + +/// Render the three tracked metrics, one row per (column, encoder). +fn render_serialized_table(writer: &mut dyn Write, results: &[SerializedResult]) -> Result<()> { + render_table( + writer, + "Vortex file: size, write, read", + "Size = % of canonical uncompressed bytes; write = repartition + zone stats + compress \ + (string scheme and children) + layout + serialize; read = open + scan, decoding each row \ + split to canonical form in its own task. Single-threaded, in-memory; MB/s over canonical \ + uncompressed bytes.", + &[ + "Column", + "Encoder", + "Size (%)", + "Write (MB/s)", + "Read (MB/s)", + ], + results + .iter() + .map(|r| { + vec![ + r.name.clone(), + r.encoder.clone(), + format!("{:.2}", r.size_pct()), + format!("{:.2}", r.write_mbps()), + format!("{:.2}", r.read_mbps()), + ] + }) + .collect(), + ) +} + +/// Render the direct codec microbenchmark metrics. +fn render_codec_table(writer: &mut dyn Write, results: &[ColumnResult]) -> Result<()> { + render_table( + writer, + "\nDirect codec microbenchmark (diagnostic)", + "One whole-column codec state, no Vortex layout, child compression, or file I/O. \ + Not comparable with the file suite's size.", + &["Column", "Encoder", "Size (%)", "Compress (MB/s)"], + results + .iter() + .map(|r| { + vec![ + r.name.clone(), + r.encoder.clone(), + format!("{:.2}", r.encoded_size_pct()), + format!("{:.2}", r.compression_mbps()), + ] + }) + .collect(), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_zero_iterations() { + assert!(Args::try_parse_from(["string-bench", "--iterations", "0"]).is_err()); + } + + /// CI passes no `--suite`, so the default is what gets tracked. + #[test] + fn suite_defaults_to_vortex() -> Result<()> { + let args = Args::try_parse_from(["string-bench"])?; + + assert_eq!(args.suite, BenchmarkSuite::Vortex); + Ok(()) + } +} diff --git a/benchmarks/string-bench/src/serialized.rs b/benchmarks/string-bench/src/serialized.rs new file mode 100644 index 00000000000..7f38ef33f85 --- /dev/null +++ b/benchmarks/string-bench/src/serialized.rs @@ -0,0 +1,462 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Serialized Vortex benchmarks. Each timed iteration writes a fresh in-memory +//! file with one string encoder forced and then reads it back: +//! +//! * **write** runs the full default write pipeline — repartition into row +//! blocks, zoned statistics, dictionary probe, coalesce, compress with the +//! forced string scheme and its children, layout, serialize into a buffer; +//! * **read** opens that buffer and runs the scan, decoding each row split to +//! canonical `VarBinViewArray` form inside its own scan task and dropping it +//! before the next split runs — the shape production uses, where +//! `into_record_batch_stream` fuses the Arrow conversion into the split task. +//! +//! Both run on a current-thread runtime, so these are single-threaded CPU costs +//! and exclude physical I/O. + +use std::hint::black_box; +use std::io::Cursor; +use std::sync::Arc; +use std::time::Duration; +use std::time::Instant; + +use anyhow::Result; +use anyhow::bail; +use bytes::Bytes; +use futures::TryStreamExt; +use vortex::array::ArrayRef; +use vortex::array::ExecutionCtx; +use vortex::array::IntoArray; +use vortex::array::VortexSessionExecute; +use vortex::array::arrays::ChunkedArray; +use vortex::array::arrays::VarBinViewArray; +use vortex::compressor::BtrBlocksCompressorBuilder; +use vortex::file::OpenOptionsSessionExt; +use vortex::file::WriteOptionsSessionExt; +use vortex::file::WriteStrategyBuilder; +use vortex::layout::LayoutStrategy; +use vortex::session::VortexSession; +use vortex_bench::Format; +use vortex_bench::measurements::CustomUnitMeasurement; +use vortex_btrblocks::SchemeExt; +use vortex_btrblocks::SchemeId; +use vortex_btrblocks::schemes::string::FSSTScheme; +use vortex_btrblocks::schemes::string::NullDominatedSparseScheme; +use vortex_btrblocks::schemes::string::OnPairScheme; +use vortex_btrblocks::schemes::string::StringDictScheme; +use vortex_onpair::DEFAULT_DICT12_CONFIG; + +use crate::StringColumn; +use crate::StringEncoder; +use crate::duration_ms; +use crate::median; +use crate::onpair_label; +use crate::prepare_column; +use crate::throughput; +use crate::verify_canonicalized; + +/// The btrblocks string schemes that `BtrBlocksCompressorBuilder::default()` can +/// choose between. Forcing one encoder excludes every entry except its own +/// scheme, so this list must track the default scheme set: add a row whenever a +/// new string encoder becomes selectable by default (e.g. Zstd). +fn default_string_scheme_ids() -> Vec { + vec![ + StringDictScheme.id(), + FSSTScheme.id(), + OnPairScheme.id(), + NullDominatedSparseScheme.id(), + ] +} + +impl StringEncoder { + /// The btrblocks string scheme that produces this encoder's on-disk arrays. + fn scheme_id(self) -> SchemeId { + match self { + Self::OnPair => OnPairScheme.id(), + Self::Fsst => FSSTScheme.id(), + } + } +} + +/// Label for one encoder on the serialized path, carrying the configuration the +/// file was written with, e.g. `onpair-12` or `fsst`. Changing that config +/// renames the metric and so restarts its benchmark history, which is intended: +/// neither size nor read time is comparable across dictionary budgets. +fn serialized_encoder_label(encoder: StringEncoder) -> String { + match encoder { + // The config `OnPairScheme` compresses with. When btrblocks gains a + // configurable budget, pass the benchmark's own config here. + StringEncoder::OnPair => onpair_label(&DEFAULT_DICT12_CONFIG), + StringEncoder::Fsst => encoder.label().to_string(), + } +} + +/// An in-memory Vortex file produced for one column and encoder. +struct SerializedFile { + file_bytes: u64, + /// Number of row splits the scan will produce, i.e. the number of encoded + /// arrays the read decodes one at a time. + chunk_count: usize, + strategy: Arc, +} + +/// The three core metrics for one serialized write and read of a string column +/// under one encoder: `size`, `write`, and `read`. +pub struct SerializedResult { + /// Column identifier. + pub name: String, + /// Encoder forced for this file, with its configuration, e.g. `onpair-12` + /// or `fsst`. + pub encoder: String, + /// Number of rows. + pub rows: usize, + /// Canonical uncompressed array bytes used to normalize size and + /// throughput: one 16-byte view per row plus the bytes of the strings too + /// long to inline. + pub uncompressed_bytes: u64, + /// Serialized Vortex file size (bytes). + pub file_bytes: u64, + /// Per-iteration write times: the whole default write pipeline, from + /// repartitioning and zoned statistics through compression, layout, and + /// serialization into an in-memory buffer. + pub write_runs: Vec, + /// Per-iteration read times: open the file, then run the scan with each row + /// split decoded to canonical form inside its own task. + pub read_runs: Vec, +} + +impl SerializedResult { + /// Median serialized-write throughput in MB/s of canonical uncompressed + /// array bytes. + pub fn write_mbps(&self) -> f64 { + throughput(self.uncompressed_bytes, median(&self.write_runs)) / 1e6 + } + + /// Median read throughput in MB/s of canonical uncompressed bytes. + pub fn read_mbps(&self) -> f64 { + throughput(self.uncompressed_bytes, median(&self.read_runs)) / 1e6 + } + + /// Complete serialized file bytes as a percentage of canonical + /// uncompressed array bytes. Lower is better. + pub fn size_pct(&self) -> f64 { + self.file_bytes as f64 / self.uncompressed_bytes as f64 * 100.0 + } + + /// Emit the three core metrics as Vortex-format custom-unit metrics, named + /// `//`. + pub fn measurements(&self) -> Vec { + let suffix = format!("{}/{}", self.name, self.encoder); + let ms = |name: String, runs: &[Duration]| CustomUnitMeasurement { + name, + format: Format::OnDiskVortex, + unit: "ms".into(), + value: duration_ms(median(runs)), + }; + vec![ + CustomUnitMeasurement { + name: format!("size/{suffix}"), + format: Format::OnDiskVortex, + unit: "%".into(), + value: self.size_pct(), + }, + ms(format!("write/{suffix}"), &self.write_runs), + ms(format!("read/{suffix}"), &self.read_runs), + ] + } +} + +/// Build the file writer strategy that forces one selected string scheme while +/// leaving non-string child compression enabled. +fn serialized_write_strategy(encoder: StringEncoder) -> Arc { + let forced = encoder.scheme_id(); + let compressor = BtrBlocksCompressorBuilder::default().exclude_schemes( + default_string_scheme_ids() + .into_iter() + .filter(|&id| id != forced), + ); + WriteStrategyBuilder::default() + .with_btrblocks_builder(compressor) + .build() +} + +/// Write one canonical string column to an in-memory Vortex file, forcing the +/// requested string encoder. +async fn write_serialized_file( + session: &VortexSession, + input: &ArrayRef, + strategy: &Arc, +) -> Result { + let mut buf = Vec::new(); + { + let mut cursor = Cursor::new(&mut buf); + session + .write_options() + .with_strategy(Arc::clone(strategy)) + .write(&mut cursor, input.to_array_stream()) + .await?; + } + Ok(Bytes::from(buf)) +} + +/// Time one complete read of a serialized Vortex buffer: open the file, then run +/// the scan with the canonical decode fused into each row split's task, dropping +/// each decoded chunk before the next split runs. +/// +/// The splits are awaited one at a time rather than through +/// `ScanBuilder::into_array_stream`, which spawns +/// `concurrency * available_parallelism()` of them at once. On a current-thread +/// runtime that read-ahead buys no parallelism; it only holds that many chunks in +/// memory and makes the result depend on the host's core count. Awaiting one at a +/// time keeps the per-split work identical to production while making the +/// measurement machine-independent. +async fn read_serialized_buffer(session: &VortexSession, data: Bytes) -> Result { + let decode_session = session.clone(); + + let start = Instant::now(); + let file = session.open_options().open_buffer(data)?; + let splits = file + .scan()? + .map(move |chunk: ArrayRef| { + let mut ctx = decode_session.create_execution_ctx(); + chunk.execute::(&mut ctx) + }) + .build()?; + + let mut rows = 0usize; + for split in splits { + if let Some(canonical) = split.await? { + rows += canonical.len(); + drop(black_box(canonical)); + } + } + + black_box(rows); + + Ok(start.elapsed()) +} + +/// Write one reference file, then check it is uniformly `encoder`-encoded and, +/// when `verify` is set, that it decodes back to the input. Runs once, outside +/// every timed region: its size is the reported `size` metric. +async fn prepare_serialized_file( + session: &VortexSession, + column: &StringColumn, + input: &ArrayRef, + canonical: &VarBinViewArray, + encoder: StringEncoder, + verify: bool, + ctx: &mut ExecutionCtx, +) -> Result { + let strategy = serialized_write_strategy(encoder); + let data = write_serialized_file(session, input, &strategy).await?; + let file_bytes = data.len() as u64; + + let file = session.open_options().open_buffer(data)?; + let chunks: Vec = file.scan()?.into_array_stream()?.try_collect().await?; + if chunks.is_empty() { + bail!("empty scan for column {}", column.name); + } + if let Some(unexpected) = chunks.iter().find(|array| !encoder.matches(array)) { + bail!( + "serialized column {} contains {} instead of {encoder} — the file was not \ + uniformly {encoder}-encoded (is {encoder} the smallest scheme?)", + column.name, + unexpected.encoding_id(), + ); + } + let chunk_count = chunks.len(); + if verify { + let canonicalized = ChunkedArray::from_iter(chunks) + .into_array() + .execute::(ctx)?; + verify_canonicalized( + &format!("{} [serialized read {encoder}]", column.name), + canonical, + &canonicalized, + ctx, + )?; + } + + Ok(SerializedFile { + file_bytes, + chunk_count, + strategy, + }) +} + +/// Time fresh in-memory serialized writes and reads for one string column and +/// encoder, on the given session's runtime. +pub async fn bench_serialized_with_session( + session: &VortexSession, + column: &StringColumn, + iterations: usize, + warmup: usize, + verify: bool, + encoder: StringEncoder, +) -> Result { + crate::validate_iterations(iterations)?; + let mut ctx = session.create_execution_ctx(); + + let (canonical, uncompressed_bytes) = prepare_column(column, &mut ctx)?; + let input = canonical.clone().into_array(); + let rows = canonical.len(); + let serialized = prepare_serialized_file( + session, column, &input, &canonical, encoder, verify, &mut ctx, + ) + .await?; + + let label = serialized_encoder_label(encoder); + tracing::debug!( + "{} [{label}]: {rows} rows, {} file bytes in {} row splits", + column.name, + serialized.file_bytes, + serialized.chunk_count, + ); + + // Warm up the same complete path that will be timed. The reference file + // above is used only for validation and size. + for _ in 0..warmup.max(1) { + let data = write_serialized_file(session, &input, &serialized.strategy).await?; + let _ = read_serialized_buffer(session, data).await?; + } + + let mut write_runs = Vec::with_capacity(iterations); + let mut read_runs = Vec::with_capacity(iterations); + + for _ in 0..iterations { + let start = Instant::now(); + let data = write_serialized_file(session, &input, &serialized.strategy).await?; + write_runs.push(start.elapsed()); + read_runs.push(read_serialized_buffer(session, data).await?); + } + + Ok(SerializedResult { + name: column.name.clone(), + encoder: label, + rows, + uncompressed_bytes, + file_bytes: serialized.file_bytes, + write_runs, + read_runs, + }) +} + +#[cfg(test)] +mod tests { + use vortex::VortexSessionDefault; + use vortex::array::Canonical; + use vortex::array::VortexSessionExecute; + use vortex::io::runtime::BlockingRuntime; + use vortex::io::runtime::current::CurrentThreadRuntime; + use vortex::io::session::RuntimeSessionExt; + use vortex_btrblocks::ALL_SCHEMES; + use vortex_btrblocks::SchemeExt; + + use super::*; + + #[test] + fn forced_scheme_inventory_matches_default_utf8_schemes() { + // Every default scheme whose dtype gate accepts canonical Utf8 must be + // excluded when another root string encoding is forced. + let canonical = Canonical::VarBinView(VarBinViewArray::from_iter_str(["value"])); + let mut actual = ALL_SCHEMES + .iter() + .filter(|scheme| scheme.matches(&canonical)) + .map(|scheme| scheme.id()) + .collect::>(); + let mut expected = default_string_scheme_ids(); + actual.sort_unstable_by_key(|id| id.to_string()); + expected.sort_unstable_by_key(|id| id.to_string()); + + assert_eq!(actual, expected); + } + + /// The three core metric names are the tracked CI schema: renaming one + /// silently restarts its benchmark history. + #[test] + fn machine_readable_metric_schema_is_stable() { + let result = SerializedResult { + name: "fixture".to_string(), + encoder: "fsst".to_string(), + rows: 1, + uncompressed_bytes: 2, + file_bytes: 1, + write_runs: vec![Duration::from_millis(1)], + read_runs: vec![Duration::from_millis(6)], + }; + + assert_eq!( + crate::measurement_rows(&result.measurements()), + [ + "size/fixture/fsst % 50", + "write/fixture/fsst ms 1", + "read/fixture/fsst ms 6", + ] + ); + } + + /// Also pins the encoder labels: a change to the btrblocks OnPair default + /// renames the tracked metrics and restarts their history. + #[test] + fn serialized_benchmark_smoke_tests_each_encoder() -> Result<()> { + let (column, expected_uncompressed_bytes) = crate::repeated_fixture(); + let runtime = CurrentThreadRuntime::new(); + let session = VortexSession::default().with_handle(runtime.handle()); + + for (encoder, expected_label) in [ + (StringEncoder::OnPair, "onpair-12"), + (StringEncoder::Fsst, "fsst"), + ] { + let result = runtime.block_on(bench_serialized_with_session( + &session, &column, 1, 0, true, encoder, + ))?; + + assert_eq!(result.encoder, expected_label); + assert_eq!(result.rows, 128); + assert_eq!(result.uncompressed_bytes, expected_uncompressed_bytes); + assert!(result.file_bytes > 0); + assert_eq!(result.write_runs.len(), 1); + assert_eq!(result.read_runs.len(), 1); + } + Ok(()) + } + + /// Enough poorly-compressible rows to span several row splits, so the + /// pre-timing check covers a file the read decodes one split at a time. + #[test] + fn serialized_verification_handles_multiple_chunks() -> Result<()> { + let values = (0..50_000_u64) + .map(|i| { + let mixed = i.wrapping_mul(0x9e37_79b9_7f4a_7c15); + format!( + "https://example.com/users/{i:016x}/events/{mixed:016x}/\ + common/repeated/string/payload" + ) + }) + .collect::>(); + let column = StringColumn { + name: "multi-chunk".to_string(), + array: VarBinViewArray::from_iter_str(values.iter()).into_array(), + }; + let runtime = CurrentThreadRuntime::new(); + let session = VortexSession::default().with_handle(runtime.handle()); + let mut ctx = session.create_execution_ctx(); + let (canonical, _) = prepare_column(&column, &mut ctx)?; + let input = canonical.clone().into_array(); + + let serialized = runtime.block_on(prepare_serialized_file( + &session, + &column, + &input, + &canonical, + StringEncoder::Fsst, + true, + &mut ctx, + ))?; + + assert!(serialized.chunk_count > 1); + Ok(()) + } +} diff --git a/scripts/compare-benchmark-jsons.py b/scripts/compare-benchmark-jsons.py index d7cd34793db..d226c23cb48 100644 --- a/scripts/compare-benchmark-jsons.py +++ b/scripts/compare-benchmark-jsons.py @@ -43,6 +43,10 @@ Z_SCORE_99 = 2.5758293035489004 CONTROL_FORMAT = "parquet" FILE_SIZE_METRIC = "file_size" +QUERY_TARGET_PATTERN = re.compile(r"_q(\d+)/([^:]+):(.+)$") +FORMAT_DISPLAY_NAMES = { + "vortex": "vortex-file-compressed", +} @dataclass @@ -60,7 +64,7 @@ def extract_dataset_key(df: pd.DataFrame) -> pd.DataFrame: """Normalize dataset metadata into a stable join key.""" if "dataset" not in df.columns: - df["dataset_key"] = pd.NA + df["dataset_key"] = None else: df["dataset_key"] = df["dataset"].apply(dataset_key) return df @@ -92,20 +96,60 @@ def dataset_key(value: Any) -> str | None: return None -def benchmark_identity(row: Any) -> tuple[Any, Any, Any] | None: - """Return the timing-row identity used to find a matching baseline.""" +def normalize_format_name(value: Any) -> str | None: + """Map serialized format identifiers to the reporter's established labels.""" - if row.get("metric") == FILE_SIZE_METRIC or row.get("file_size") is not None: + value = identity_value(value) + if value is None: + return None + value = str(value) + return FORMAT_DISPLAY_NAMES.get(value, value) + + +def comparison_target(name: Any, target: Any = None) -> tuple[str, str, int | None]: + """Return the engine, display format, and optional SQL query number.""" + + target_engine = None + target_format = None + if isinstance(target, dict): + target_engine = identity_value(target.get("engine")) + target_format = normalize_format_name(target.get("format")) + + match = QUERY_TARGET_PATTERN.search(name) if isinstance(name, str) else None + name_engine = match.group(2) if match is not None else None + name_format = match.group(3) if match is not None else None + query = int(match.group(1)) if match is not None else None + + engine = str(target_engine or name_engine or "unknown") + file_format = str(target_format or normalize_format_name(name_format) or "unknown") + return engine, file_format, query + + +def extract_target_fields(name: str, target: Any = None) -> pd.Series: + """Extract target metadata, using the benchmark name when needed.""" + + engine, file_format, query = comparison_target(name, target) + return pd.Series({"engine": engine, "file_format": file_format, "query": query}) + + +def benchmark_identity(row: Any) -> tuple[Any, Any, Any, str, str, Any] | None: + """Return the measurement identity used to find a matching baseline.""" + + if row.get("metric") == FILE_SIZE_METRIC or isinstance(row.get("file_size"), dict): return None name = row.get("name") if name is None: return None + engine, file_format, _query = comparison_target(name, row.get("target")) return ( identity_value(name), identity_value(row.get("storage")), dataset_key(row.get("dataset")), + engine, + file_format, + identity_value(row.get("unit")), ) @@ -117,16 +161,10 @@ def benchmark_identity_rows(df: pd.DataFrame) -> pd.DataFrame: return pd.DataFrame(columns=["commit_id", "benchmark_identity"]) timing_rows = timing_rows.copy() - if "storage" not in timing_rows.columns: - timing_rows["storage"] = pd.NA if "commit_id" not in timing_rows.columns: timing_rows["commit_id"] = pd.NA - timing_rows = extract_dataset_key(timing_rows) - timing_rows["benchmark_identity"] = [ - tuple(identity_value(row[column]) for column in ("name", "storage", "dataset_key")) - for _, row in timing_rows.iterrows() - ] + timing_rows["benchmark_identity"] = [benchmark_identity(row) for _, row in timing_rows.iterrows()] return timing_rows[["commit_id", "benchmark_identity"]] @@ -146,7 +184,12 @@ def read_jsonl_rows_for_commit(path: str, commit_id: str) -> pd.DataFrame: def read_latest_baseline_rows(path: str, pr: pd.DataFrame) -> pd.DataFrame: - """Read rows from the latest history commit matching the PR benchmark.""" + """Read rows from the latest history commit matching the PR benchmark. + + A benchmark can be new to the PR workflow and therefore have no baseline + yet. Return an empty frame with the PR schema in that case so the report + can show the measurements without comparison. + """ pr_identities = set(benchmark_identity_rows(pr)["benchmark_identity"]) if not pr_identities: @@ -164,7 +207,7 @@ def read_latest_baseline_rows(path: str, pr: pd.DataFrame) -> pd.DataFrame: baseline_commit_id = commit_id if baseline_commit_id is None: - raise ValueError("No baseline rows found for the benchmark under test") + return pr.iloc[0:0].copy() return read_jsonl_rows_for_commit(path, baseline_commit_id) @@ -192,29 +235,34 @@ def select_latest_baseline_rows(base: pd.DataFrame, pr: pd.DataFrame) -> pd.Data matches = base_identities[base_identities["benchmark_identity"].isin(pr_identities)] matches = matches[matches["commit_id"].notna()] if matches.empty: - raise ValueError("No baseline rows found for the benchmark under test") + return base.iloc[0:0].copy() baseline_commit_id = matches["commit_id"].iloc[-1] return base[base["commit_id"] == baseline_commit_id].copy() -def extract_target_fields(name: str) -> pd.Series: - """Parse query, engine, and format from the benchmark name.""" +def normalize_measurement_rows(df: pd.DataFrame) -> pd.DataFrame: + """Add canonical comparison keys to benchmark measurement rows.""" - if not isinstance(name, str): - return pd.Series({"engine": "unknown", "file_format": "unknown", "query": pd.NA}) + df = df.copy() + if "storage" not in df.columns: + df["storage"] = None + if "unit" not in df.columns: + df["unit"] = None - match = re.search(r"_q(\d+)/([^:]+):(.+)$", name) - if match is None: - return pd.Series({"engine": "unknown", "file_format": "unknown", "query": pd.NA}) + df["storage"] = df["storage"].apply(identity_value) + df["unit"] = df["unit"].apply(identity_value) + df = extract_dataset_key(df) - return pd.Series( - { - "engine": match.group(2), - "file_format": match.group(3), - "query": int(match.group(1)), - } + targets = df["target"] if "target" in df.columns else pd.Series(None, index=df.index) + fields = [comparison_target(name, target) for name, target in zip(df["name"], targets)] + df[["engine", "file_format", "query"]] = pd.DataFrame( + fields, + columns=["engine", "file_format", "query"], + index=df.index, ) + df["query"] = pd.array(df["query"], dtype="Int64") + return df def positive_samples(values: Any) -> np.ndarray: @@ -487,12 +535,24 @@ def format_performance( return f"{ratio:.3f}x {emoji}" -def format_integer_value(value: float) -> str: - """Render numeric timing values for markdown tables.""" +def format_measurement_value(value: float) -> str: + """Render integral and fractional measurements for a Markdown table.""" if pd.isna(value): - return "" - return str(int(value)) + return "—" + + value = float(value) + if value.is_integer(): + return str(int(value)) + return f"{value:.9g}" + + +def format_comparison_ratio(value: float) -> str: + """Render a PR/base ratio or identify an unmatched PR measurement.""" + + if pd.isna(value): + return "no baseline" + return f"{float(value):.2f}" def format_size(size_bytes: int) -> str: @@ -826,16 +886,28 @@ def format_report_help() -> str: "arrow": 5, } +UNIT_ORDER = { + "ns": 0, + "μs": 1, + "ms": 2, + "bytes": 3, + "MB": 4, + "%": 5, + "ratio": 6, +} + -def group_sort_key(group_key: tuple[str, str]) -> tuple[int, int, str, str]: - """Keep output ordering stable and grouped by likely reader interest.""" +def group_sort_key(group_key: tuple[str, str, str]) -> tuple[int, int, int, str, str, str]: + """Keep output ordering stable.""" - engine, file_format = group_key + engine, file_format, unit = group_key return ( ENGINE_ORDER.get(engine, len(ENGINE_ORDER)), FILE_FORMAT_ORDER.get(file_format, len(FILE_FORMAT_ORDER)), + UNIT_ORDER.get(unit, len(UNIT_ORDER)), engine, file_format, + unit, ) @@ -848,42 +920,42 @@ def main() -> None: title = format_title(benchmark_name, pr) base = read_latest_baseline_rows(sys.argv[1], pr) - base_commit_id = set(base["commit_id"].unique()) + base_commit_ids = set(base["commit_id"].unique()) pr_commit_id = set(pr["commit_id"].unique()) - assert len(base_commit_id) == 1, base_commit_id + assert len(base_commit_ids) <= 1, base_commit_ids assert len(pr_commit_id) == 1, pr_commit_id - base_commit_id = next(iter(base_commit_id)) + base_commit_id = next(iter(base_commit_ids), None) pr_commit_id = next(iter(pr_commit_id)) base_file_sizes, base = split_file_size_rows(base) pr_file_sizes, pr = split_file_size_rows(pr) - if "storage" not in base: - base["storage"] = pd.NA - if "storage" not in pr: - pr["storage"] = pd.NA + base = normalize_measurement_rows(base) + pr = normalize_measurement_rows(pr) - base = extract_dataset_key(base) - pr = extract_dataset_key(pr) - - df3 = pd.merge(base, pr, on=["name", "storage", "dataset_key"], how="right", suffixes=("_base", "_pr")) + comparison_keys = ["name", "storage", "dataset_key", "engine", "file_format", "unit", "query"] + df3 = pd.merge(base, pr, on=comparison_keys, how="right", suffixes=("_base", "_pr")) + df3["unit"] = df3["unit"].fillna("unit") df3["ratio"] = df3["value_pr"] / df3["value_base"] - df3[["engine", "file_format", "query"]] = df3["name"].apply(extract_target_fields) is_s3_benchmark = "s3" in benchmark_name.lower() threshold_pct = 30 if is_s3_benchmark else 10 improvement_threshold = 1.0 - (threshold_pct / 100.0) regression_threshold = 1.0 + (threshold_pct / 100.0) - vortex_df = df3[df3["name"].str.contains("vortex", case=False, na=False)] - parquet_df = df3[df3["name"].str.contains("parquet", case=False, na=False)] + query_df = df3[df3["query"].notna()] + headline_df = query_df + if headline_df.empty and df3["unit"].nunique() == 1: + headline_df = df3 + vortex_df = headline_df[headline_df["file_format"].str.startswith("vortex")] + parquet_df = headline_df[headline_df["file_format"].eq(CONTROL_FORMAT)] vortex_geo_mean_ratio = calculate_geo_mean(vortex_df) parquet_geo_mean_ratio = calculate_geo_mean(parquet_df) - statistical_analysis = build_statistical_analysis(df3, threshold_pct) + statistical_analysis = build_statistical_analysis(query_df, threshold_pct) verdict = build_verdict(statistical_analysis) if statistical_analysis is not None else None - engine_analyses = build_within_engine_statistical_analyses(df3, threshold_pct) + engine_analyses = build_within_engine_statistical_analyses(query_df, threshold_pct) engine_summary = format_within_engine_summary(engine_analyses) summary_fields: list[str] = [] @@ -921,39 +993,45 @@ def main() -> None: print(title) print("") - print("
".join(summary_fields)) - print("") - print(format_report_help()) - print("") + if summary_fields: + print("
".join(summary_fields)) + print("") + if base_commit_id is None: + print("_No baseline is available for this benchmark yet; PR measurements are shown without comparison._") + print("") + if verdict is not None or engine_summary is not None: + print(format_report_help()) + print("") print("---") print("") - grouped_tables = df3.groupby(["engine", "file_format"], dropna=False, sort=False) - for engine, file_format in sorted(grouped_tables.groups.keys(), key=group_sort_key): - group_df = grouped_tables.get_group((engine, file_format)).sort_values("name") + grouped_tables = df3.groupby(["engine", "file_format", "unit"], dropna=False, sort=False) + base_label = str(base_commit_id)[:8] if base_commit_id is not None else "none" + for engine, file_format, unit in sorted(grouped_tables.groups.keys(), key=group_sort_key): + group_df = grouped_tables.get_group((engine, file_format, unit)).sort_values("name") group_performance = format_performance( calculate_geo_mean(group_df), improvement_threshold, regression_threshold, "group", ) - significant_improvements = (group_df["ratio"] < improvement_threshold).sum() - significant_regressions = (group_df["ratio"] > regression_threshold).sum() - unit = group_df["unit_base"].dropna().iloc[0] if group_df["unit_base"].notna().any() else "unit" + significant_improvements = (group_df["ratio"] <= improvement_threshold).sum() + significant_regressions = (group_df["ratio"] >= regression_threshold).sum() display_df = pd.DataFrame( { "name": [ format_name_with_highlight(name, ratio, improvement_threshold, regression_threshold) for name, ratio in zip(group_df["name"], group_df["ratio"]) ], - f"PR {pr_commit_id[:8]} ({unit})": group_df["value_pr"].map(format_integer_value), - f"base {base_commit_id[:8]} ({unit})": group_df["value_base"].map(format_integer_value), - "ratio (PR/base)": group_df["ratio"], + f"PR {pr_commit_id[:8]} ({unit})": group_df["value_pr"].map(format_measurement_value), + f"base {base_label} ({unit})": group_df["value_base"].map(format_measurement_value), + "ratio (PR/base)": group_df["ratio"].map(format_comparison_ratio), } ) print("
") summary_text = ( - f"{engine} / {file_format} ({group_performance}, {significant_improvements}↑ {significant_regressions}↓)" + f"{engine} / {file_format} / {unit} " + f"({group_performance}, {significant_improvements}↑ {significant_regressions}↓)" ) print(f"{summary_text}") print("") @@ -963,7 +1041,8 @@ def main() -> None: display_df.to_markdown( index=False, tablefmt="github", - floatfmt=".2f", + disable_numparse=True, + colalign=("left", "right", "right", "right"), ) ) print("") diff --git a/scripts/tests/test_benchmark_reporting.py b/scripts/tests/test_benchmark_reporting.py index 7925161e91e..4e7bc7c9933 100644 --- a/scripts/tests/test_benchmark_reporting.py +++ b/scripts/tests/test_benchmark_reporting.py @@ -39,6 +39,8 @@ def stored_timing_row( value: int, storage: str | None = None, dataset: dict[str, object] | None = None, + engine: str = "datafusion", + file_format: str = "parquet", ) -> dict[str, object]: row: dict[str, object] = { "name": name, @@ -46,6 +48,7 @@ def stored_timing_row( "value": value, "all_runtimes": [value, value, value], "commit_id": commit, + "target": {"engine": engine, "format": file_format}, } if storage is not None: row["storage"] = storage @@ -54,6 +57,50 @@ def stored_timing_row( return row +def stored_custom_row( + commit: str, + name: str, + unit: str, + value: float, + engine: str = "vortex", + file_format: str = "vortex", +) -> dict[str, object]: + return { + "name": name, + "unit": unit, + "value": value, + "commit_id": commit, + "target": {"engine": engine, "format": file_format}, + } + + +def render_report( + tmp_path: Path, + base_rows: list[dict[str, object]], + pr_rows: list[dict[str, object]], + benchmark_name: str, +) -> str: + base_path = tmp_path / "base.jsonl" + pr_path = tmp_path / "pr.jsonl" + base_path.write_text("".join(f"{json.dumps(row)}\n" for row in base_rows), encoding="utf-8") + pr_path.write_text("".join(f"{json.dumps(row)}\n" for row in pr_rows), encoding="utf-8") + + result = subprocess.run( + [sys.executable, str(COMPARE_SCRIPT), str(base_path), str(pr_path), benchmark_name], + check=False, + capture_output=True, + text=True, + ) + + assert result.returncode == 0, result.stderr + return result.stdout + + +def markdown_row(report: str, name: str) -> list[str]: + line = next(line for line in report.splitlines() if line.startswith(f"| {name} ")) + return [cell.strip() for cell in line.strip("|").split("|")] + + def test_select_latest_baseline_rows_uses_latest_matching_benchmark_commit() -> None: compare = load_compare_module() history = pd.DataFrame( @@ -115,6 +162,22 @@ def test_read_latest_baseline_rows_streams_latest_matching_benchmark_commit(tmp_ {"scale_factor": "1.0"}, ), file_size_record_for("base-current", 120, "tpch", "1.0", "vortex-file-compressed", "part-0.vortex"), + stored_timing_row( + "base-wrong-unit", + "tpch_q01/datafusion:parquet", + 111, + "nvme", + {"scale_factor": "1.0"}, + ) + | {"unit": "ms"}, + stored_timing_row( + "base-wrong-target", + "tpch_q01/datafusion:parquet", + 112, + "nvme", + {"scale_factor": "1.0"}, + file_format="vortex", + ), stored_timing_row("base-other", "clickbench_q01/datafusion:parquet", 200, "nvme"), ] history_path.write_text( @@ -157,6 +220,98 @@ def test_within_engine_analysis_uses_each_engines_own_parquet_control() -> None: assert compare.build_verdict(analyses["duckdb"])["impact"] == "+20.0%" +def test_comparison_report_groups_by_target_and_unit(tmp_path: Path) -> None: + base_rows = [ + stored_custom_row("base-sha", "timing/fixture", "ms", 12.5), + stored_custom_row("base-sha", "size/fixture", "%", 45.25), + stored_custom_row("base-sha", "ratio/fixture", "ratio", 0.75), + stored_custom_row("base-sha", "parquet timing/fixture", "ms", 20.0, file_format="parquet"), + ] + pr_rows = [ + stored_custom_row("pr-sha", "timing/fixture", "ms", 11.75), + stored_custom_row("pr-sha", "new timing/fixture", "ms", 3.125), + stored_custom_row("pr-sha", "size/fixture", "%", 44.5), + stored_custom_row("pr-sha", "ratio/fixture", "ratio", 0.8), + stored_custom_row("pr-sha", "parquet timing/fixture", "ms", 21.0, file_format="parquet"), + ] + + report = render_report(tmp_path, base_rows, pr_rows, "Mixed metrics") + + assert "unknown / unknown" not in report + assert "How to read Verdict and Engines" not in report + assert "vortex / vortex-file-compressed / ms " in report + assert "vortex / vortex-file-compressed / % " in report + assert "vortex / vortex-file-compressed / ratio " in report + assert "vortex / parquet / ms " in report + assert markdown_row(report, "timing/fixture") == ["timing/fixture", "11.75", "12.5", "0.94"] + assert markdown_row(report, "size/fixture") == ["size/fixture", "44.5", "45.25", "0.98"] + assert markdown_row(report, "ratio/fixture") == ["ratio/fixture", "0.8", "0.75", "1.07"] + assert markdown_row(report, "new timing/fixture") == [ + "new timing/fixture", + "3.125", + "—", + "no baseline", + ] + + +def test_comparison_report_handles_missing_benchmark_baseline(tmp_path: Path) -> None: + base_rows = [stored_custom_row("base-sha", "other timing/fixture", "ms", 12.5)] + pr_rows = [stored_custom_row("pr-sha", "new timing/fixture", "ms", 3.125)] + + report = render_report(tmp_path, base_rows, pr_rows, "New benchmark") + + assert "No baseline is available for this benchmark yet" in report + assert "base none (ms)" in report + assert markdown_row(report, "new timing/fixture") == [ + "new timing/fixture", + "3.125", + "—", + "no baseline", + ] + + +def test_comparison_report_handles_mixed_query_types_in_baseline(tmp_path: Path) -> None: + base_rows = [ + stored_custom_row("base-sha", "random-access/fixture", "ms", 12.5), + stored_timing_row("base-sha", "tpch_q01/datafusion:parquet", 100), + ] + pr_rows = [stored_custom_row("pr-sha", "random-access/fixture", "ms", 13.0)] + + report = render_report(tmp_path, base_rows, pr_rows, "Random Access") + + assert markdown_row(report, "random-access/fixture") == [ + "random-access/fixture", + "13", + "12.5", + "1.04", + ] + + +def test_comparison_report_retains_sql_analysis(tmp_path: Path) -> None: + targets = [ + ("parquet", "parquet", 100, 105), + ("vortex-file-compressed", "vortex", 80, 70), + ] + base_rows = [ + stored_timing_row("base-sha", f"tpch_q01/datafusion:{name}", base, file_format=target) + for name, target, base, _pr in targets + ] + pr_rows = [ + stored_timing_row("pr-sha", f"tpch_q01/datafusion:{name}", pr, file_format=target) + for name, target, _base, pr in targets + ] + + report = render_report(tmp_path, base_rows, pr_rows, "TPC-H") + + assert "**Verdict**:" in report + assert "**Vortex (geomean)**:" in report + assert "**Parquet (geomean)**:" in report + assert "How to read Verdict and Engines" in report + assert "datafusion / vortex-file-compressed / ns " in report + assert "datafusion / parquet / ns " in report + assert "unknown / unknown" not in report + + def file_size_record(commit: str, size: int) -> dict[str, object]: return file_size_record_for(commit, size, "tpch", "10", "vortex-file-compressed", "part-0.vortex")