diff --git a/.github/actions/setup-duckdb/action.yml b/.github/actions/setup-duckdb/action.yml new file mode 100644 index 00000000000..58964dda51d --- /dev/null +++ b/.github/actions/setup-duckdb/action.yml @@ -0,0 +1,17 @@ +name: "Setup DuckDB" +description: "Download the DuckDB CLI and put it on PATH" +inputs: + duckdb_version: + description: "Version of the DuckDB CLI" + default: "1.5.5" +runs: + using: "composite" + steps: + - name: Download DuckDB + shell: bash + run: | + wget -qO- \ + "https://github.com/duckdb/duckdb/releases/download/v${{ inputs.duckdb_version }}/duckdb_cli-linux-amd64.zip" \ + | funzip > duckdb + chmod +x duckdb + echo "$PWD" >> "$GITHUB_PATH" diff --git a/.github/workflows/develop-bench.yml b/.github/workflows/develop-bench.yml index 2066d80257c..d326670d5e8 100644 --- a/.github/workflows/develop-bench.yml +++ b/.github/workflows/develop-bench.yml @@ -67,10 +67,7 @@ jobs: enable-sccache: ${{ github.repository == 'vortex-data/vortex' && 'true' || 'false' }} - name: Install DuckDB - run: | - wget -qO- https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.zip | funzip > duckdb - chmod +x duckdb - echo "$PWD" >> "$GITHUB_PATH" + uses: ./.github/actions/setup-duckdb - uses: ./.github/actions/system-info diff --git a/.github/workflows/pr-bench-gpu-compress.yml b/.github/workflows/pr-bench-gpu-compress.yml index cd97755e636..5ef1693972f 100644 --- a/.github/workflows/pr-bench-gpu-compress.yml +++ b/.github/workflows/pr-bench-gpu-compress.yml @@ -32,6 +32,21 @@ jobs: with: repo-token: ${{ secrets.GITHUB_TOKEN }} enable-sccache: "true" + - name: Install DuckDB + # The Public BI datasets are converted from CSV to Parquet through the DuckDB CLI. + # Without it those datasets cannot build their fixture. + uses: ./.github/actions/setup-duckdb + - name: Install uv + uses: spiraldb/actions/.github/actions/setup-uv@a746510eafaa926484c354541cfc49b2ec06cc63 # 0.18.6 + - name: Install cuDF + # The GPU Parquet number is a cuDF `read_parquet`. cuDF ships prebuilt manylinux wheels + # on NVIDIA's index, so it stays a runtime dependency and never enters the Rust build. + run: | + uv venv --python 3.12 .venv-cudf + uv pip install --python .venv-cudf \ + --extra-index-url https://pypi.nvidia.com \ + cudf-cu12 pandas pyarrow + echo "$PWD/.venv-cudf/bin" >> "$GITHUB_PATH" - uses: ./.github/actions/system-info - name: Display NVIDIA GPU details run: | diff --git a/.github/workflows/pr-bench-runner.yml b/.github/workflows/pr-bench-runner.yml index 22c2c693fd3..0c3ee63b3ba 100644 --- a/.github/workflows/pr-bench-runner.yml +++ b/.github/workflows/pr-bench-runner.yml @@ -49,10 +49,7 @@ jobs: enable-sccache: ${{ github.event.pull_request.head.repo.fork == false && 'true' || 'false' }} - name: Install DuckDB - run: | - wget -qO- https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.zip | funzip > duckdb - chmod +x duckdb - echo "$PWD" >> "$GITHUB_PATH" + uses: ./.github/actions/setup-duckdb - uses: ./.github/actions/system-info diff --git a/.github/workflows/sql-bench-matrix.yml b/.github/workflows/sql-bench-matrix.yml index 52d4e6a545a..3adb478092d 100644 --- a/.github/workflows/sql-bench-matrix.yml +++ b/.github/workflows/sql-bench-matrix.yml @@ -133,10 +133,7 @@ jobs: sync: false - name: Install DuckDB - run: | - wget -qO- https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.zip | funzip > duckdb - chmod +x duckdb - echo "$PWD" >> "$GITHUB_PATH" + uses: ./.github/actions/setup-duckdb - uses: ./.github/actions/system-info diff --git a/Cargo.lock b/Cargo.lock index fee1e3ebd34..a80d6f85f02 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1578,6 +1578,8 @@ dependencies = [ "lance-bench", "parquet 58.4.0", "regex", + "serde", + "serde_json", "tempfile", "tokio", "tracing", diff --git a/benchmarks/compress-bench/Cargo.toml b/benchmarks/compress-bench/Cargo.toml index 4046a12d42e..7b1dadc3209 100644 --- a/benchmarks/compress-bench/Cargo.toml +++ b/benchmarks/compress-bench/Cargo.toml @@ -27,6 +27,8 @@ itertools = { workspace = true } lance-bench = { path = "../lance-bench", optional = true } parquet = { workspace = true } regex = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } tempfile = { workspace = true, optional = true } tokio = { workspace = true, features = ["full"] } tracing = { workspace = true } @@ -45,7 +47,6 @@ name = "compress-bench" test = false [lib] -test = false [lints] workspace = true diff --git a/benchmarks/compress-bench/README.md b/benchmarks/compress-bench/README.md index bf2d3efc1db..01b3e6e7d82 100644 --- a/benchmarks/compress-bench/README.md +++ b/benchmarks/compress-bench/README.md @@ -15,13 +15,109 @@ See [`src/main.rs`](./src/main.rs) for the dataset list and CLI flags (`--format cargo run -p compress-bench --profile release_debug ``` -GPU decompression is opt-in and runs only the existing benchmark names allow-listed in -`src/main.rs`: +## GPU decompression + +`--gpu-decompress` is opt-in, requires the `cuda` feature, and restricts the suite to the +GPU dataset list in `src/main.rs`. It measures decompression only, for two backends: + +- **Vortex** — the file is written with CUDA-compatible BtrBlocks encodings only + (`only_cuda_compatible`) and a CUDA flat layout, then decoded on the device all the way to + canonical arrays. +- **Parquet** — the file is rewritten with GPU-friendly writer settings (see below) and read + back with [cuDF](https://github.com/rapidsai/cudf)'s `read_parquet`, which performs the + whole read on the device: page header decode, codec decompression, dictionary/RLE/plain + decoding and column assembly. + +Both sides therefore decode all the way to device-resident arrays, which is what makes the +`vortex:parquet- gpu ratio decompress time` metric a like-for-like comparison. ```bash cargo run -p compress-bench --profile release_debug \ --features cuda,unstable_encodings -- --gpu-decompress + +# pick the Parquet page codec the GPU file is written with (default: snappy) +cargo run -p compress-bench --profile release_debug \ + --features cuda,unstable_encodings -- --gpu-decompress --gpu-parquet-codec zstd ``` -On Linux, GPU files are read with direct IO (`O_DIRECT`) so repeated iterations measure -storage bandwidth rather than page-cache hits. +### cuDF + +cuDF has no Rust binding, so the benchmark drives it out of process: it spawns `python3` running +`scripts/cudf-parquet-read.py`, which imports cuDF from the prebuilt `cudf-cu12` Python package, +reads the file, and prints its timings back as JSON on stdout. Nothing links against libcudf, so +cuDF is a runtime requirement rather than a Rust build dependency: + +```bash +uv pip install --extra-index-url https://pypi.nvidia.com cudf-cu12 pandas pyarrow +``` + +The clock lives inside that script rather than around the subprocess, so process spawn, +interpreter start, `import cudf` and CUDA context creation are all excluded; a warm-up read runs +first for the same reason. The script performs several timed reads per invocation and reports the +fastest, and the harness then takes its own minimum across `--iterations`. + +Both backends read a warm file by default. cuDF runs an untimed warm-up read before the timed +one, so its timed read hits the page cache; the Vortex reader therefore does **not** use direct +I/O by default, because `O_DIRECT` would bypass the page cache and compare a Vortex read of the +disk against a cuDF read of RAM. `--gpu-direct-io` turns it back on to measure storage bandwidth +instead — a different question, and the resulting ratio is not a decode comparison. + +The remaining asymmetry is the transfer path: the Vortex reader uses pinned buffers, while cuDF +does its own host read and host-to-device copy. + +### GPU-friendly Parquet writer settings + +Set in `src/gpu/writer.rs`: + +| Setting | Value | Why | +| --- | --- | --- | +| writer version | `PARQUET_1_0` | v1 pages compress the whole page body; v2 pages put uncompressed levels ahead of the compressed values in the same body. | +| compression | Snappy (default) or Zstd | Snappy is the Parquet default and has the higher device-side throughput. | +| dictionary | enabled | Keeps the decompressed payload small; the encoding GPU Parquet readers decode fastest. | +| data page size | 1 MiB | Large enough to amortize per-page setup, small enough to keep every SM fed. Matches the page size cuDF targets. | +| data page row limit | 1,000,000 | The 20k-row default caps narrow columns' pages far below 1 MiB. | +| statistics | chunk-level | Page statistics only inflate the headers a reader has to walk. | +| row group size | 1,048,576 rows | Shared with the Vortex side as `GPU_ROW_GROUP_SIZE` — see below. | + +### Matching physical partitions + +A Parquet row group and a Vortex chunk are the same thing for this comparison: the unit the +reader plans and dispatches over. Both formats are pinned to `GPU_ROW_GROUP_SIZE` +(1,048,576 rows, Parquet's `DEFAULT_MAX_ROW_GROUP_ROW_COUNT`). + +Without this the two are not comparable. Parquet reads ~1M-row row groups, while the Vortex +side inherits the Arrow reader's ~8K-row batches — each of which becomes its own chunk, its own +compressed blocks and its own kernel launches, so a single dispatch turns into hundreds. + +Setting the Arrow reader's batch size alone is not enough: the reader also breaks at the source +file's row group boundaries, so short batches survive. `parquet_to_vortex_chunks_with_batch_size` +therefore concatenates the source batches and re-slices them on exact boundaries. Those batches +are written straight through as root chunks via `ChunkedLayoutStrategy`, and read back with +`SplitBy::RowCount(GPU_ROW_GROUP_SIZE)` so a scan batch is one whole partition. + +### Correctness + +`--gpu-verify` cross-checks device output against the CPU decoders on every iteration: + +- Parquet: the cuDF-read frame is compared against a CPU Parquet read of the same file. +- Vortex: each GPU-decoded field is copied back and compared against the same field decoded + on the CPU, through Arrow with a pinned target type. + +The check runs before each timed measurement and is not included in it, so a verifying run +still publishes comparable numbers — it just takes considerably longer: + +```bash +cargo run -p compress-bench --profile release_debug \ + --features cuda,unstable_encodings -- --gpu-decompress --gpu-verify --iterations 1 +``` + +Any `--gpu-decompress` run reports on every dataset rather than stopping at the first failure, so +one run shows which datasets decode correctly on the GPU and which do not. The timing tables are +rendered before the failure summary, so a dataset the GPU cannot decode still leaves the rest of +the matrix with numbers — the process exits non-zero either way. + +The dataset list in `src/main.rs` therefore holds only datasets a `--gpu-verify` run has confirmed. +Several others are waiting on `vortex-cuda` kernel gaps (`u16` in `date_time_parts`, a +`vortex.masked` kernel, and a CPU fallback reached with device-resident buffers); they are listed +with their reasons next to `gpu_datasets`. Add one there once its gap is closed and verification +passes. diff --git a/benchmarks/compress-bench/src/gpu/mod.rs b/benchmarks/compress-bench/src/gpu/mod.rs new file mode 100644 index 00000000000..aba1e4a4dfd --- /dev/null +++ b/benchmarks/compress-bench/src/gpu/mod.rs @@ -0,0 +1,66 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! The GPU decompression mode behind `--gpu-decompress`. +//! +//! This module is the only place in the crate that mentions the `cuda` feature. The two device +//! backends need it, so they are gated here and reached through [`compressor`]; the rest of the +//! crate calls that one function and stays feature-agnostic. +//! +//! [`GpuOptions`] and [`writer`] deliberately sit outside the gate. Neither touches CUDA, and +//! keeping them unconditional is what lets `main` parse the `--gpu-*` flags — and reject them +//! with a clear message — in a build without the feature. + +use vortex_bench::Format; +use vortex_bench::compress::Compressor; + +pub mod writer; + +#[cfg(feature = "cuda")] +mod parquet; +#[cfg(feature = "cuda")] +mod vortex; + +pub use crate::gpu::writer::GpuCodec; + +/// Settings for the GPU decompression mode. +#[derive(Clone, Copy, Debug)] +pub struct GpuOptions { + /// Parquet page codec to write the GPU file with. + pub codec: GpuCodec, + /// Cross-check decompressed output against the CPU decoders. + pub verify: bool, + /// Read the Vortex file with direct IO instead of through the page cache. + pub direct_io: bool, +} + +/// The GPU backend that measures `format`. +#[cfg(feature = "cuda")] +pub fn compressor(format: Format, options: GpuOptions) -> Box { + match format { + Format::OnDiskVortex => Box::new(vortex::GpuVortexCompressor::new( + options.verify, + options.direct_io, + )) as Box, + Format::Parquet => Box::new(parquet::GpuParquetCompressor::new( + options.codec, + options.verify, + )), + _ => unimplemented!("GPU compress bench not implemented for {format}"), + } +} + +/// Stands in for [`compressor`] in a build without the `cuda` feature. +/// +/// `main` rejects `--gpu-decompress` before any compressor is selected, so reaching this is a +/// bug. Destructuring the options is what marks their fields as read: they are only otherwise +/// used by the gated backends, and without this they are dead code in a non-CUDA build. +#[cfg(not(feature = "cuda"))] +pub fn compressor(format: Format, options: GpuOptions) -> Box { + let GpuOptions { + codec: _, + verify: _, + direct_io: _, + } = options; + unreachable!("GPU mode requires the cuda feature, checked before selecting a {format} backend") +} diff --git a/benchmarks/compress-bench/src/gpu/parquet.rs b/benchmarks/compress-bench/src/gpu/parquet.rs new file mode 100644 index 00000000000..2d01a1527eb --- /dev/null +++ b/benchmarks/compress-bench/src/gpu/parquet.rs @@ -0,0 +1,170 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! GPU Parquet decompression backend, timed through cuDF. +//! +//! cuDF's `read_parquet` performs the whole read on the device — page header decode, +//! codec decompression, dictionary/RLE/plain decoding and column assembly — which makes it +//! the like-for-like opponent for the Vortex GPU backend, which likewise decodes all the way +//! to canonical arrays on device. +//! +//! cuDF has no Rust binding, so this backend drives it out of process: it spawns `python3` +//! running [`CUDF_SCRIPT`], which imports cuDF from the prebuilt `cudf-cu12` Python package +//! and prints its timings back as JSON on stdout. Nothing here links against libcudf, so cuDF +//! is a runtime requirement of the benchmark rather than a Rust build dependency — see the +//! README for the install. +//! +//! The clock lives inside that script, not around the subprocess, so process spawn, +//! interpreter start, `import cudf` and CUDA context creation are all excluded; only the reads +//! themselves are timed. + +use std::path::Path; +use std::process::Command; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use anyhow::ensure; +use arrow_array::RecordBatch; +use async_trait::async_trait; +use parquet::arrow::ArrowWriter; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use serde::Deserialize; +use tempfile::NamedTempFile; +use vortex_bench::Format; +use vortex_bench::compress::Compressor; + +use crate::gpu::writer::GpuCodec; +use crate::gpu::writer::gpu_writer_properties; + +/// Repo-relative path of the script that performs and times the cuDF read. +const CUDF_SCRIPT: &str = "scripts/cudf-parquet-read.py"; + +/// Parquet compressor whose decompression measurement is a whole-file cuDF GPU read. +/// +/// "Whole-file" means every row and every column: no projection and no filter is pushed into +/// the read, so the measurement covers decoding the entire table. That matches what the Vortex +/// GPU backend does, which decodes every field of every batch it scans. +pub struct GpuParquetCompressor { + codec: GpuCodec, + /// Cross-check the GPU read against a CPU Parquet read of the same file. + /// + /// cuDF has no verification of its own — a `read_parquet` either succeeds or raises — so + /// [`CUDF_SCRIPT`] does the checking, comparing the frame it read on the device against + /// one pandas read on the host. Off by default: it is a correctness pass, not a benchmark. + verify: bool, +} + +/// Timed reads to ask [`CUDF_SCRIPT`] for per invocation. +/// +/// One read is a noisy sample, and repeating inside the script is nearly free — the cost that +/// dominates a cuDF read is process spawn and `import cudf`, which is paid once either way. +const TIMED_READS: usize = 3; + +/// What the cuDF script reports back. +#[derive(Debug, Deserialize)] +struct CudfReadReport { + /// Fastest of the script's [`TIMED_READS`] reads, in nanoseconds. + /// + /// The outer harness calls `decompress` once per `--iterations` and takes its own minimum, + /// so the published number is the fastest read across every process the run spawned. + min_ns: u64, + rows: u64, + columns: u64, +} + +impl GpuParquetCompressor { + /// Create a backend that writes pages with `codec` and times cuDF reading them back. + /// + /// When `verify` is set, the GPU read is cross-checked against a CPU Parquet read of the + /// same file before the measurement is reported. + pub fn new(codec: GpuCodec, verify: bool) -> Self { + Self { codec, verify } + } + + /// Rewrite the source Parquet file with GPU-friendly writer settings. + fn write_gpu_parquet(&self, parquet_path: &Path) -> Result<(NamedTempFile, u64)> { + let builder = ParquetRecordBatchReaderBuilder::try_new(std::fs::File::open(parquet_path)?)?; + let schema = Arc::clone(builder.schema()); + let batches: Vec = builder.build()?.collect::, _>>()?; + + let output = NamedTempFile::new()?; + let mut writer = ArrowWriter::try_new( + output.reopen()?, + schema, + Some(gpu_writer_properties(self.codec)), + )?; + for batch in batches { + writer.write(&batch)?; + } + writer.flush()?; + let size = writer.bytes_written() as u64; + writer.close()?; + Ok((output, size)) + } +} + +#[async_trait] +impl Compressor for GpuParquetCompressor { + fn format(&self) -> Format { + Format::Parquet + } + + /// Unsupported: GPU mode measures decompression only. + /// + /// `--gpu-decompress` restricts the suite to [`CompressOp::Decompress`], so nothing calls + /// this. It used to time [`Self::write_gpu_parquet`], but that measures the host Parquet + /// writer rather than anything on the device, and the result was never rendered — so it was + /// a number nobody could read and nobody should have compared. The Vortex GPU backend + /// refuses the same way. + /// + /// [`CompressOp::Decompress`]: vortex_bench::compress::CompressOp::Decompress + async fn compress(&self, _parquet_path: &Path) -> Result<(u64, Duration)> { + bail!("GPU compress-bench only supports decompression measurements") + } + + async fn decompress(&self, parquet_path: &Path) -> Result { + let (gpu_file, _) = self.write_gpu_parquet(parquet_path)?; + let report = run_cudf_read(gpu_file.path(), self.verify)?; + + ensure!( + report.rows > 0 && report.columns > 0, + "cuDF read {} rows and {} columns, expected a non-empty table", + report.rows, + report.columns + ); + + Ok(Duration::from_nanos(report.min_ns)) + } +} + +/// Runs the cuDF read script and returns the timing it measured. +fn run_cudf_read(path: &Path, verify: bool) -> Result { + let output = Command::new("python3") + .arg(CUDF_SCRIPT) + .arg(path) + .arg("--iterations") + .arg(TIMED_READS.to_string()) + .args(verify.then_some("--verify")) + .output() + .with_context(|| format!("failed to spawn python3 to run {CUDF_SCRIPT}"))?; + + // A missing cuDF surfaces here rather than above: python3 starts fine and then fails on + // `import cudf`, so the traceback is on stderr and the install hint belongs with it. + if !output.status.success() { + bail!( + "{CUDF_SCRIPT} exited with {}; is cudf-cu12 installed on this host?\n{}", + output.status, + String::from_utf8_lossy(&output.stderr).trim() + ); + } + + serde_json::from_slice(&output.stdout).with_context(|| { + format!( + "could not parse the report from {CUDF_SCRIPT}: {}", + String::from_utf8_lossy(&output.stdout).trim() + ) + }) +} diff --git a/benchmarks/compress-bench/src/gpu/vortex.rs b/benchmarks/compress-bench/src/gpu/vortex.rs new file mode 100644 index 00000000000..bc236b3f939 --- /dev/null +++ b/benchmarks/compress-bench/src/gpu/vortex.rs @@ -0,0 +1,329 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use std::hint::black_box; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; +use std::time::Instant; + +use anyhow::Result; +use anyhow::bail; +use anyhow::ensure; +use arrow_array::ArrayRef as ArrowArrayRef; +use arrow_schema::Field; +use async_trait::async_trait; +use futures::Stream; +use futures::StreamExt; +use tempfile::NamedTempFile; +use vortex::array::ArrayRef; +use vortex::array::ExecutionCtx; +use vortex::array::IntoArray; +use vortex::array::VortexSessionExecute; +use vortex::array::arrays::StructArray; +use vortex::array::arrays::struct_::StructArrayExt; +use vortex::compressor::BtrBlocksCompressorBuilder; +use vortex::error::VortexResult; +use vortex::file::OpenOptionsSessionExt; +use vortex::file::WriteOptionsSessionExt; +use vortex::layout::layouts::chunked::writer::ChunkedLayoutStrategy; +use vortex::layout::layouts::compressed::CompressingStrategy; +use vortex::layout::scan::split_by::SplitBy; +use vortex_arrow::ArrowSessionExt; +use vortex_bench::Format; +use vortex_bench::SESSION; +use vortex_bench::compress::Compressor; +use vortex_bench::conversions::parquet_to_vortex_chunks_with_batch_size; +use vortex_cuda::CanonicalCudaExt; +use vortex_cuda::CudaOpenOptionsExt; +use vortex_cuda::CudaSession; +#[cfg(target_os = "linux")] +use vortex_cuda::PooledFileReadAtOptions; +use vortex_cuda::executor::CudaArrayExt; +use vortex_cuda::layout::CudaFlatLayoutStrategy; +use vortex_cuda::layout::register_cuda_layout; + +use crate::gpu::writer::GPU_ROW_GROUP_SIZE; + +/// Vortex compressor whose decompression measurement executes CUDA-compatible files on the GPU. +pub struct GpuVortexCompressor { + verify: bool, + direct_io: bool, +} + +impl GpuVortexCompressor { + /// Create the backend. + /// + /// When `verify` is set, each GPU-decoded field is copied back to the host and compared + /// against the same field decoded on the CPU before the timed scan runs. The verification is + /// not itself timed, so a verifying run reports the same measurement a plain one would — it + /// just takes longer to get there. + pub fn new(verify: bool, direct_io: bool) -> Self { + Self { verify, direct_io } + } +} + +#[async_trait] +impl Compressor for GpuVortexCompressor { + fn format(&self) -> Format { + Format::OnDiskVortex + } + + async fn compress(&self, _parquet_path: &Path) -> Result<(u64, Duration)> { + anyhow::bail!("GPU compress-bench only supports decompression measurements") + } + + async fn decompress(&self, parquet_path: &Path) -> Result { + register_cuda_layout(&SESSION); + + // Rebatch to the same partition size the GPU Parquet file is written with. Left alone, + // the Arrow reader hands back ~8K-row batches, each of which becomes its own Vortex + // chunk and its own set of kernel launches. + let uncompressed = parquet_to_vortex_chunks_with_batch_size( + parquet_path.to_path_buf(), + Some(GPU_ROW_GROUP_SIZE), + ) + .await?; + let gpu_file = NamedTempFile::new()?; + let mut output = tokio::fs::File::create(gpu_file.path()).await?; + // Write those batches straight through as root chunks, so a chunk on disk is one + // partition rather than whatever the default strategy would regroup them into. + let strategy = Arc::new(ChunkedLayoutStrategy::new(CompressingStrategy::new( + CudaFlatLayoutStrategy::default(), + BtrBlocksCompressorBuilder::default() + .only_cuda_compatible() + .build(), + ))); + SESSION + .write_options() + .with_strategy(strategy) + .write(&mut output, uncompressed.into_array().to_array_stream()) + .await?; + output.sync_all().await?; + drop(output); + + // Verification is a precondition on the measurement below, not a substitute for it. It + // used to return its own elapsed time, which bundled a file copy, a second host scan and + // every Arrow conversion into a number the table then published as a decode time. + if self.verify { + verify_against_host_scan(gpu_file.path(), self.direct_io).await?; + } + + let mut cuda_ctx = CudaSession::create_execution_ctx(&SESSION)?; + let start = Instant::now(); + let file = open_gpu(gpu_file.path(), self.direct_io).await?; + // Split reads on the same boundary the file was written with, so a scan batch is one + // partition instead of a sub-slice of one. + let mut batches = file + .scan()? + .with_split_by(SplitBy::RowCount(GPU_ROW_GROUP_SIZE)) + .into_array_stream()?; + + while let Some(batch) = batches.next().await { + let record = batch?.execute::(cuda_ctx.execution_ctx())?; + for field in record.iter_unmasked_fields() { + black_box(field.clone().execute_cuda(&mut cuda_ctx).await?); + } + } + cuda_ctx.synchronize_stream()?; + + Ok(start.elapsed()) + } +} + +/// Opens a Vortex file for CUDA execution. +/// +/// `direct_io` is off by default so this backend is comparable with the cuDF one: cuDF reads +/// through the page cache after an untimed warm-up read, so leaving direct IO on would compare +/// a Vortex read of the disk against a cuDF read of RAM. Turning it on measures storage +/// bandwidth instead, which is a different question and not comparable across the two. +/// +/// Only the direct-IO path is Linux-only — `O_DIRECT` has no portable equivalent, so +/// [`PooledFileReadAtOptions`] only offers it there. The rest of this module is not: CUDA runs on +/// Windows too, and the whole crate still has to compile on a developer's macOS machine. Asking +/// for `--gpu-direct-io` where it cannot be honoured is an error rather than a silent no-op, +/// because the flag changes what the resulting number means. +async fn open_gpu(path: &Path, direct_io: bool) -> Result { + let open_options = SESSION.open_options().with_cuda(); + + #[cfg(target_os = "linux")] + let open_options = if direct_io { + open_options.with_read_at_options(PooledFileReadAtOptions::default().with_direct_io()) + } else { + open_options + }; + + #[cfg(not(target_os = "linux"))] + anyhow::ensure!( + !direct_io, + "--gpu-direct-io needs O_DIRECT, which is only available on Linux" + ); + + Ok(open_options.open_path(path).await?) +} + +/// Decodes the same file on the GPU and on the CPU and fails on the first difference. +/// +/// The CPU reference comes from a second, host-only scan rather than from re-decoding the +/// GPU scan's arrays: a CUDA scan hands back arrays whose buffers live in device memory, +/// which the host decoders cannot read. +/// +/// This times nothing and reports no measurement: the caller runs its own timed scan afterwards, +/// so a verifying run publishes the same kind of number as a plain one. +async fn verify_against_host_scan(path: &Path, direct_io: bool) -> Result<()> { + let mut cuda_ctx = CudaSession::create_execution_ctx(&SESSION)?; + // Everything on the reference side — the host scan and both Arrow conversions — has to run + // through a plain host context. A CUDA context allocates its outputs in device memory, and + // the Arrow conversion then reads those buffers on the host. + let mut host_ctx = SESSION.create_execution_ctx(); + + // The host scan reads a copy rather than the same path. The session's segment cache is + // keyed by URI, and the CUDA reader deliberately bypasses it because its buffers are + // device-resident; running both scans against one URI risks the two sharing entries. + let host_path = NamedTempFile::new()?; + std::fs::copy(path, host_path.path())?; + + let gpu_file = open_gpu(path, direct_io).await?; + let mut gpu_batches = gpu_file + .scan()? + .with_split_by(SplitBy::RowCount(GPU_ROW_GROUP_SIZE)) + .into_array_stream()?; + let host_file = SESSION.open_options().open_path(host_path.path()).await?; + let mut host_batches = host_file + .scan()? + .with_split_by(SplitBy::RowCount(GPU_ROW_GROUP_SIZE)) + .into_array_stream()?; + + let mut fields_checked = 0usize; + let mut batch_index = 0usize; + while let Some((gpu_batch, host_batch)) = + next_batch_pair(&mut gpu_batches, &mut host_batches).await? + { + let gpu_record = gpu_batch.execute::(cuda_ctx.execution_ctx())?; + let host_record = host_batch.execute::(&mut host_ctx)?; + ensure!( + gpu_record.len() == host_record.len(), + "batch {batch_index} length differs between the GPU and CPU scans: {} vs {}", + gpu_record.len(), + host_record.len() + ); + + let gpu_fields = gpu_record + .iter_unmasked_fields() + .cloned() + .collect::>(); + let host_fields = host_record + .iter_unmasked_fields() + .cloned() + .collect::>(); + ensure!( + gpu_fields.len() == host_fields.len(), + "batch {batch_index} field count differs between the GPU and CPU scans" + ); + + for (field_index, (gpu_field, host_field)) in + gpu_fields.into_iter().zip(host_fields).enumerate() + { + let decoded = gpu_field.execute_cuda(&mut cuda_ctx).await?; + // The decode is enqueued, not complete: make the writes visible before reading + // the buffers back to the host. + cuda_ctx.synchronize_stream()?; + let decoded = decoded.into_host().await?.into_array(); + verify_field( + &host_field, + decoded, + &mut host_ctx, + batch_index, + field_index, + )?; + fields_checked += 1; + } + + batch_index += 1; + } + cuda_ctx.synchronize_stream()?; + + tracing::info!("verified {fields_checked} GPU-decoded Vortex fields against the CPU decode"); + Ok(()) +} + +/// Pulls one batch from each scan, or `None` once both are exhausted. +/// +/// Both streams cover the same file, so one ending before the other is itself a failure rather +/// than a stopping condition — that is why this returns `Result>` instead of letting +/// the caller zip the two streams together. +async fn next_batch_pair( + gpu: &mut (impl Stream> + Unpin), + host: &mut (impl Stream> + Unpin), +) -> Result> { + match (gpu.next().await, host.next().await) { + (Some(gpu_batch), Some(host_batch)) => Ok(Some((gpu_batch?, host_batch?))), + (None, None) => Ok(None), + _ => bail!("the GPU and CPU scans of the same file produced different batch counts"), + } +} + +/// Fails unless a GPU-decoded field matches the same field decoded on the CPU. +fn verify_field( + host: &ArrayRef, + gpu: ArrayRef, + ctx: &mut ExecutionCtx, + batch_index: usize, + field_index: usize, +) -> Result<()> { + let expected = SESSION.arrow().execute_arrow(host.clone(), None, ctx)?; + // Pin the Arrow target type so the two sides cannot land on different but equivalent + // encodings of the same logical values. + let target = Field::new("", expected.data_type().clone(), gpu.dtype().is_nullable()); + let actual = SESSION.arrow().execute_arrow(gpu, Some(&target), ctx)?; + + if expected.to_data() == actual.to_data() { + return Ok(()); + } + + bail!( + "GPU decode of a {} field does not match the CPU decode \ + (batch {batch_index}, field {field_index}){}", + host.encoding_id(), + describe_mismatch(&expected, &actual) + ) +} + +/// Builds a human-readable description of how two Arrow arrays differ. +fn describe_mismatch(expected: &ArrowArrayRef, actual: &ArrowArrayRef) -> String { + let mut description = format!( + "\n cpu: type={:?} len={} nulls={}\n gpu: type={:?} len={} nulls={}", + expected.data_type(), + expected.len(), + expected.null_count(), + actual.data_type(), + actual.len(), + actual.null_count(), + ); + + if expected.data_type() != actual.data_type() || expected.len() != actual.len() { + return description; + } + + // Binary search for the shortest prefix that already differs; its last element is the + // first mismatching row. + let (mut low, mut high) = (0usize, expected.len()); + while low < high { + let mid = low + (high - low) / 2 + 1; + if expected.slice(0, mid).to_data() == actual.slice(0, mid).to_data() { + low = mid; + } else { + high = mid - 1; + } + } + + if low < expected.len() { + description.push_str(&format!( + "\n first difference at row {low}:\n cpu: {:?}\n gpu: {:?}", + expected.slice(low, 1), + actual.slice(low, 1), + )); + } + + description +} diff --git a/benchmarks/compress-bench/src/gpu/writer.rs b/benchmarks/compress-bench/src/gpu/writer.rs new file mode 100644 index 00000000000..49d9493f91f --- /dev/null +++ b/benchmarks/compress-bench/src/gpu/writer.rs @@ -0,0 +1,108 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Parquet writer settings for the GPU benchmark. +//! +//! The GPU backend rewrites each dataset before reading it back with cuDF. The CPU suite's +//! settings are tuned for host reads, and reusing them would measure a device decoder on a file +//! laid out for a different one; these settings size pages and row groups for a GPU read +//! instead. Each choice and its reason is tabulated in the README. + +use clap::ValueEnum; +use parquet::basic::Compression; +use parquet::basic::ZstdLevel; +use parquet::file::properties::DEFAULT_MAX_ROW_GROUP_ROW_COUNT; +use parquet::file::properties::EnabledStatistics; +use parquet::file::properties::WriterProperties; +use parquet::file::properties::WriterVersion; + +/// Target size of a data page written for GPU reads. +/// +/// Pages are the unit a GPU reader decompresses in parallel, so they need to be large enough +/// to amortize per-page setup and numerous enough to fill the device. ~1 MiB is the page size +/// cuDF's Parquet reader is tuned around. +pub const GPU_DATA_PAGE_SIZE: usize = 1024 * 1024; + +/// Row cap per data page. +/// +/// `parquet`'s default caps pages at 20k rows, which produces pages far below +/// [`GPU_DATA_PAGE_SIZE`] for narrow columns and leaves the device underfed. +const GPU_DATA_PAGE_ROW_COUNT_LIMIT: usize = 1_000_000; + +/// Rows per physical partition in both GPU benchmark formats. +/// +/// A Parquet row group and a Vortex chunk are the same thing for this comparison: the unit the +/// reader plans and dispatches over. Pinning both to one value is what makes the two numbers +/// comparable — otherwise Parquet reads ~1M-row row groups while Vortex inherits the Arrow +/// reader's ~8K-row batches, turning one launch into hundreds. +pub const GPU_ROW_GROUP_SIZE: usize = DEFAULT_MAX_ROW_GROUP_ROW_COUNT; + +/// Parquet page codecs the GPU benchmark can write. +/// +/// A deliberate two-codec subset, not a limit of Parquet or of cuDF — both handle more. These +/// two are the ones worth comparing: Snappy is the Parquet default and the fastest to decode on +/// the device, and Zstd is what the CPU Parquet benchmark writes, so picking it answers "what +/// does the GPU do with the file the CPU suite already measures?". Adding a codec here is a +/// two-line change if another becomes interesting. +#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)] +pub enum GpuCodec { + /// The Parquet default, and the codec with the highest device-side throughput. + #[default] + Snappy, + /// Matches the codec used by the CPU Parquet benchmark, at lower device throughput. + Zstd, +} + +impl GpuCodec { + /// The Parquet compression setting for this codec. + pub fn to_parquet(self) -> Compression { + match self { + GpuCodec::Snappy => Compression::SNAPPY, + GpuCodec::Zstd => Compression::ZSTD(ZstdLevel::default()), + } + } + + /// Short lowercase name, used in measurement labels. + pub fn name(self) -> &'static str { + match self { + GpuCodec::Snappy => "snappy", + GpuCodec::Zstd => "zstd", + } + } +} + +/// Writer properties tuned for a GPU read. +pub fn gpu_writer_properties(codec: GpuCodec) -> WriterProperties { + WriterProperties::builder() + // V1 data pages compress the whole page body. V2 pages place uncompressed + // repetition/definition levels ahead of the compressed values in the same body, which + // not every GPU reader path handles. + .set_writer_version(WriterVersion::PARQUET_1_0) + .set_compression(codec.to_parquet()) + // Dictionary encoding keeps the decompressed payload small and is the encoding GPU + // Parquet readers decode fastest. + .set_dictionary_enabled(true) + .set_data_page_size_limit(GPU_DATA_PAGE_SIZE) + .set_data_page_row_count_limit(GPU_DATA_PAGE_ROW_COUNT_LIMIT) + // Stated explicitly rather than left to the default, because the Vortex side is + // rebatched to the same constant and the two have to move together. + .set_max_row_group_row_count(Some(GPU_ROW_GROUP_SIZE)) + // Per-page statistics only inflate the page headers a reader has to walk. + .set_statistics_enabled(EnabledStatistics::Chunk) + .build() +} + +#[cfg(test)] +mod tests { + use parquet::file::properties::WriterVersion; + + use super::*; + + #[test] + fn gpu_properties_use_v1_pages_and_the_requested_codec() { + let properties = gpu_writer_properties(GpuCodec::Snappy); + assert_eq!(properties.writer_version(), WriterVersion::PARQUET_1_0); + assert_eq!(properties.compression(&"x".into()), Compression::SNAPPY); + assert_eq!(properties.data_page_size_limit(), GPU_DATA_PAGE_SIZE); + } +} diff --git a/benchmarks/compress-bench/src/gpu_vortex.rs b/benchmarks/compress-bench/src/gpu_vortex.rs deleted file mode 100644 index 3dbb68bc7a8..00000000000 --- a/benchmarks/compress-bench/src/gpu_vortex.rs +++ /dev/null @@ -1,85 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// SPDX-FileCopyrightText: Copyright the Vortex contributors - -use std::hint::black_box; -use std::path::Path; -use std::sync::Arc; -use std::time::Duration; -use std::time::Instant; - -use anyhow::Result; -use async_trait::async_trait; -use futures::StreamExt; -use tempfile::NamedTempFile; -use vortex::array::IntoArray; -use vortex::array::arrays::StructArray; -use vortex::array::arrays::struct_::StructArrayExt; -use vortex::compressor::BtrBlocksCompressorBuilder; -use vortex::file::OpenOptionsSessionExt; -use vortex::file::WriteOptionsSessionExt; -use vortex::file::WriteStrategyBuilder; -use vortex_bench::Format; -use vortex_bench::SESSION; -use vortex_bench::compress::Compressor; -use vortex_bench::conversions::parquet_to_vortex_chunks; -use vortex_cuda::CudaOpenOptionsExt; -use vortex_cuda::CudaSession; -#[cfg(target_os = "linux")] -use vortex_cuda::PooledFileReadAtOptions; -use vortex_cuda::executor::CudaArrayExt; -use vortex_cuda::layout::CudaFlatLayoutStrategy; -use vortex_cuda::layout::register_cuda_layout; - -/// Vortex compressor whose decompression measurement executes CUDA-compatible files on the GPU. -pub struct GpuVortexCompressor; - -#[async_trait] -impl Compressor for GpuVortexCompressor { - fn format(&self) -> Format { - Format::OnDiskVortex - } - - async fn compress(&self, _parquet_path: &Path) -> Result<(u64, Duration)> { - anyhow::bail!("GPU compress-bench only supports decompression measurements") - } - - async fn decompress(&self, parquet_path: &Path) -> Result { - register_cuda_layout(&SESSION); - - let uncompressed = parquet_to_vortex_chunks(parquet_path.to_path_buf()).await?; - let gpu_file = NamedTempFile::new()?; - let mut output = tokio::fs::File::create(gpu_file.path()).await?; - let strategy = WriteStrategyBuilder::default() - .with_btrblocks_builder(BtrBlocksCompressorBuilder::default().only_cuda_compatible()) - .with_flat_strategy(Arc::new(CudaFlatLayoutStrategy::default())) - .build(); - SESSION - .write_options() - .with_strategy(strategy) - .write(&mut output, uncompressed.into_array().to_array_stream()) - .await?; - output.sync_all().await?; - drop(output); - - let mut cuda_ctx = CudaSession::create_execution_ctx(&SESSION)?; - let start = Instant::now(); - let open_options = SESSION.open_options().with_cuda(); - // Direct IO keeps repeated iterations measuring storage bandwidth rather than - // page-cache hits. It is only available on Linux. - #[cfg(target_os = "linux")] - let open_options = - open_options.with_read_at_options(PooledFileReadAtOptions::default().with_direct_io()); - let file = open_options.open_path(gpu_file.path()).await?; - let mut batches = file.scan()?.into_array_stream()?; - - while let Some(batch) = batches.next().await { - let record = batch?.execute::(cuda_ctx.execution_ctx())?; - for field in record.iter_unmasked_fields() { - black_box(field.clone().execute_cuda(&mut cuda_ctx).await?); - } - } - cuda_ctx.synchronize_stream()?; - - Ok(start.elapsed()) - } -} diff --git a/benchmarks/compress-bench/src/lib.rs b/benchmarks/compress-bench/src/lib.rs index 68039996605..25347300667 100644 --- a/benchmarks/compress-bench/src/lib.rs +++ b/benchmarks/compress-bench/src/lib.rs @@ -3,7 +3,6 @@ #[cfg(feature = "lance")] pub use lance_bench::compress::LanceCompressor; -#[cfg(feature = "cuda")] -pub mod gpu_vortex; +pub mod gpu; pub mod parquet; pub mod vortex; diff --git a/benchmarks/compress-bench/src/main.rs b/benchmarks/compress-bench/src/main.rs index 1b1603e52c8..0660d345879 100644 --- a/benchmarks/compress-bench/src/main.rs +++ b/benchmarks/compress-bench/src/main.rs @@ -1,16 +1,21 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +use std::any::Any; +use std::panic::AssertUnwindSafe; use std::path::PathBuf; use std::time::Duration; +use anyhow::Context; use clap::Parser; #[cfg(feature = "lance")] use compress_bench::LanceCompressor; -#[cfg(feature = "cuda")] -use compress_bench::gpu_vortex::GpuVortexCompressor; +use compress_bench::gpu::GpuCodec; +use compress_bench::gpu::GpuOptions; +use compress_bench::gpu::compressor as gpu_compressor; use compress_bench::parquet::ParquetCompressor; use compress_bench::vortex::VortexCompressor; +use futures::FutureExt; use indicatif::ProgressBar; use itertools::Itertools; use regex::Regex; @@ -35,6 +40,7 @@ use vortex_bench::display::DisplayFormat; use vortex_bench::display::print_measurements_json; use vortex_bench::display::render_table; use vortex_bench::downloadable_dataset::DownloadableDataset; +use vortex_bench::measurements::CustomUnitMeasurement; use vortex_bench::public_bi::PBI_DATASETS; use vortex_bench::public_bi::PBIDataset::Arade; use vortex_bench::public_bi::PBIDataset::Bimbo; @@ -67,11 +73,30 @@ struct Args { ops: Vec, #[arg(long)] datasets: Option, - /// Run GPU decompression for the allow-listed benchmarks. + /// Run GPU decompression for the GPU-supported benchmarks. /// - /// This filters the suite to GPU-supported dataset names and runs only Vortex decompression. + /// Restricts the suite to datasets with verified CUDA decode support and measures + /// decompression only, for both Vortex and Parquet. #[arg(long)] gpu_decompress: bool, + /// Page codec the GPU Parquet file is written with. + /// + /// Snappy is the Parquet default and the codec GPU readers decompress fastest. + #[arg(long, value_enum, default_value_t)] + gpu_parquet_codec: GpuCodec, + /// Cross-check every GPU-decompressed result against the CPU decoder. + /// + /// The check runs before the timed measurement and is not included in it, so the numbers a + /// verifying run publishes are still comparable — it only makes the run slower. + #[arg(long)] + gpu_verify: bool, + /// Read the Vortex GPU file with direct IO, bypassing the page cache. + /// + /// Off by default: cuDF reads through the page cache after an untimed warm-up, so direct + /// IO would compare a Vortex read of the disk against a cuDF read of RAM. Turn it on to + /// measure storage bandwidth instead, and do not read the ratio as a decode comparison. + #[arg(long)] + gpu_direct_io: bool, #[arg(short, long, default_value_t, value_enum)] display_format: DisplayFormat, #[arg(short, long)] @@ -97,8 +122,21 @@ async fn main() -> anyhow::Result<()> { anyhow::bail!("--gpu-decompress requires building compress-bench with --features cuda"); } - let (formats, ops) = if args.gpu_decompress { - (vec![Format::OnDiskVortex], vec![CompressOp::Decompress]) + let mode = if args.gpu_decompress { + BenchMode::Gpu(GpuOptions { + codec: args.gpu_parquet_codec, + verify: args.gpu_verify, + direct_io: args.gpu_direct_io, + }) + } else { + BenchMode::Cpu + }; + + let (formats, ops) = if mode.is_gpu() { + ( + vec![Format::Parquet, Format::OnDiskVortex], + vec![CompressOp::Decompress], + ) } else { (args.formats, args.ops) }; @@ -108,7 +146,7 @@ async fn main() -> anyhow::Result<()> { args.datasets.map(|d| Regex::new(&d)).transpose()?, formats, ops, - args.gpu_decompress, + mode, args.display_format, args.output_path, args.ingest_output, @@ -116,15 +154,31 @@ async fn main() -> anyhow::Result<()> { .await } +/// Which suite a run measures. +/// +/// GPU mode is not a setting on the CPU suite: it selects a different dataset list, restricts the +/// ops to decompression, swaps both compressors and labels its ratio differently. A variant says +/// that; an `Option` only said "maybe some GPU settings", leaving "which suite is +/// this?" and "how is the GPU configured?" to be the same question. +#[derive(Clone, Copy, Debug)] +enum BenchMode { + /// The ordinary host suite. + Cpu, + /// Device decompression, configured by the `--gpu-*` flags. + Gpu(GpuOptions), +} + +impl BenchMode { + /// Whether this run measures the GPU. + fn is_gpu(self) -> bool { + matches!(self, BenchMode::Gpu(_)) + } +} + /// Get a compressor for the given format. -fn get_compressor(format: Format, gpu_decompress: bool) -> Box { - if gpu_decompress { - #[cfg(feature = "cuda")] - { - return Box::new(GpuVortexCompressor); - } - #[cfg(not(feature = "cuda"))] - unreachable!("GPU feature validation happens before selecting compressors"); +fn get_compressor(format: Format, mode: BenchMode) -> Box { + if let BenchMode::Gpu(options) = mode { + return gpu_compressor(format, options); } match format { @@ -151,7 +205,7 @@ async fn run_compress( datasets_filter: Option, formats: Vec, ops: Vec, - gpu_decompress: bool, + mode: BenchMode, display_format: DisplayFormat, output_path: Option, ingest_output: Option, @@ -178,15 +232,28 @@ async fn run_compress( // ), ]; - // Add an existing benchmark name here only after its CUDA-compatible compression and - // decompression kernels have been verified end to end. - #[expect( - clippy::useless_vec, - reason = "this is an intentionally incremental allow-list of benchmark names" - )] - let gpu_decompress_benchmarks = vec!["TPC-H l_comment canonical"]; + // Datasets run in GPU mode. Add one only after a `--gpu-verify` run has confirmed its CUDA + // decode end to end; a dataset here that cannot decode fails the benchmark job. Between them + // these cover FSST strings, bit-packed numerics and columns with nulls. + // + // Not yet listed, each blocked on a `vortex-cuda` gap rather than on the benchmark: + // + // - `taxi` and `Arade`: `Unsupported ptype u16`. The CUDA `date_time_parts` kernel dispatches + // with `match_each_signed_integer_ptype!` where the CPU canonicaliser uses + // `match_each_integer_ptype!`. Widening the fused kernel takes 4³ = 64 PTX instantiations + // to 8³ = 512, so it is not a free change. + // - `Euro2016` and `HashTags`: `No CUDA kernel for encoding vortex.masked`. + // - `CMSprovider`: `expected host buffer` — a CPU fallback is reached with device-resident + // buffers, which `CudaArrayExt::execute_cuda` refuses. + // - `StructListOfInts`: its list layouts have no verified CUDA decode path. + let gpu_datasets: [&dyn Dataset; 4] = [ + &TPCHLCommentCanonical as &dyn Dataset, + &TPCHLCommentChunked, + PBI_DATASETS.get(Bimbo), + PBI_DATASETS.get(Food), + ]; - let datasets: Vec<&dyn Dataset> = [ + let all_datasets: Vec<&dyn Dataset> = [ &TaxiData as &dyn Dataset, PBI_DATASETS.get(Arade), PBI_DATASETS.get(Bimbo), @@ -206,10 +273,15 @@ async fn run_compress( ] .into_iter() .chain(structlistofints.iter().map(|d| d as &dyn Dataset)) + .collect(); + + let datasets: Vec<&dyn Dataset> = if mode.is_gpu() { + gpu_datasets.to_vec() + } else { + all_datasets + } + .into_iter() .filter(|d| { - if gpu_decompress && !gpu_decompress_benchmarks.contains(&d.name()) { - return false; - } if let Some(filter) = datasets_filter.as_ref() { filter.is_match(d.name()) } else { @@ -225,18 +297,37 @@ async fn run_compress( let mut measurements = vec![]; let mut v3_records: Vec = Vec::new(); + // A GPU pass reports on every dataset rather than stopping at the first failure, so one run + // says which datasets decode correctly on the GPU and still yields numbers for the rest. + let survey_all = mode.is_gpu(); + let mut failures: Vec<(String, anyhow::Error)> = Vec::new(); + for dataset_handle in datasets.into_iter() { - let (m, mut records) = run_benchmark_for_dataset( - &progress, - &formats, - &ops, - iterations, - dataset_handle, - gpu_decompress, - ) - .await?; - measurements.push(m); - v3_records.append(&mut records); + let run = + run_benchmark_for_dataset(&progress, &formats, &ops, iterations, dataset_handle, mode); + + // Missing CUDA kernel support surfaces as a panic rather than an error, so the survey + // has to catch those too or the first unsupported dataset ends the run. + let result = if survey_all { + match AssertUnwindSafe(run).catch_unwind().await { + Ok(result) => result, + Err(panic) => Err(anyhow::anyhow!("panicked: {}", panic_message(&panic))), + } + } else { + run.await + }; + + match result { + Ok((m, mut records)) => { + measurements.push(m); + v3_records.append(&mut records); + } + Err(error) if survey_all => { + tracing::error!("{}: {error:#}", dataset_handle.name()); + failures.push((dataset_handle.name().to_string(), error)); + } + Err(error) => return Err(error), + } } let measurements = CompressMeasurements::from_iter(measurements); @@ -249,6 +340,8 @@ async fn run_compress( let mut writer = create_output_writer(&display_format, output_path, BENCHMARK_ID)?; + // The tables render before any failure is reported, so a partially failing GPU matrix still + // publishes the numbers for the datasets that did decode. match display_format { DisplayFormat::Table => { render_table(&mut writer, measurements.timings, &targets)?; @@ -260,13 +353,33 @@ async fn run_compress( } else { vec![] }, - ) + )?; } DisplayFormat::GhJson => { print_measurements_json(&mut writer, measurements.timings, DOC_PATH)?; - print_measurements_json(&mut writer, measurements.ratios, DOC_PATH) + print_measurements_json(&mut writer, measurements.ratios, DOC_PATH)?; + } + } + + if !failures.is_empty() { + eprintln!( + "\nGPU decompression failed for {} dataset(s):", + failures.len() + ); + for (dataset, error) in &failures { + eprintln!(" - {dataset}: {error:#}"); } + anyhow::bail!( + "GPU decompression failed for: {}", + failures + .iter() + .map(|(dataset, _)| dataset.as_str()) + .collect::>() + .join(", ") + ); } + + Ok(()) } async fn run_benchmark_for_dataset( @@ -275,7 +388,7 @@ async fn run_benchmark_for_dataset( ops: &[CompressOp], iterations: usize, dataset_handle: &dyn Dataset, - gpu_decompress: bool, + mode: BenchMode, ) -> anyhow::Result<(CompressMeasurements, Vec)> { let bench_name = dataset_handle.name(); let (v3_dataset, v3_variant) = dataset_handle.v3_dataset_dims(); @@ -291,7 +404,7 @@ async fn run_benchmark_for_dataset( let mut v3_records: Vec = Vec::new(); for format in formats { - let compressor = get_compressor(*format, gpu_decompress); + let compressor = get_compressor(*format, mode); for op in ops { let time = match op { @@ -302,7 +415,8 @@ async fn run_benchmark_for_dataset( iterations, bench_name, ) - .await?; + .await + .with_context(|| format!("compressing {bench_name} as {format}"))?; compressed_sizes.insert(*format, result.compressed_size); let all_runs_ns: Vec = result .all_runs @@ -333,7 +447,8 @@ async fn run_benchmark_for_dataset( iterations, bench_name, ) - .await?; + .await + .with_context(|| format!("decompressing {bench_name} as {format}"))?; let all_runs_ns: Vec = result .all_runs .iter() @@ -342,7 +457,7 @@ async fn run_benchmark_for_dataset( v3_records.push(v3::compression_time_record( &result.timing, v3_dataset, - if gpu_decompress { + if mode.is_gpu() { Some("gpu") } else { v3_variant @@ -361,12 +476,55 @@ async fn run_benchmark_for_dataset( } // Calculate cross-format ratios after all measurements. - calculate_ratios( - &measurements_map, - &compressed_sizes, - bench_name, - &mut ratios, - ); + match mode { + // The shared ratio labels name the CPU suite's codec, which the GPU run does not + // necessarily use, so GPU mode emits its own correctly-labelled ratio. + BenchMode::Gpu(options) => { + push_gpu_ratio(&measurements_map, options, bench_name, &mut ratios) + } + BenchMode::Cpu => calculate_ratios( + &measurements_map, + &compressed_sizes, + bench_name, + &mut ratios, + ), + } Ok((CompressMeasurements { timings, ratios }, v3_records)) } + +/// Emit the Vortex-versus-Parquet decompression ratio for a GPU run. +fn push_gpu_ratio( + measurements: &HashMap<(Format, CompressOp), Duration>, + options: GpuOptions, + bench_name: &str, + ratios: &mut Vec, +) { + let (Some(vortex_time), Some(parquet_time)) = ( + measurements.get(&(Format::OnDiskVortex, CompressOp::Decompress)), + measurements.get(&(Format::Parquet, CompressOp::Decompress)), + ) else { + return; + }; + + ratios.push(CustomUnitMeasurement { + name: format!( + "vortex:parquet-{} gpu ratio decompress time/{bench_name}", + options.codec.name() + ), + format: Format::OnDiskVortex, + unit: std::borrow::Cow::from("ratio"), + value: vortex_time.as_nanos() as f64 / parquet_time.as_nanos() as f64, + }); +} + +/// Extracts the message from a caught panic payload. +fn panic_message(panic: &Box) -> String { + if let Some(message) = panic.downcast_ref::<&str>() { + (*message).to_string() + } else if let Some(message) = panic.downcast_ref::() { + message.clone() + } else { + "non-string panic payload".to_string() + } +} diff --git a/scripts/cudf-parquet-read.py b/scripts/cudf-parquet-read.py new file mode 100644 index 00000000000..6368f20ff66 --- /dev/null +++ b/scripts/cudf-parquet-read.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright the Vortex contributors + +"""Times a full GPU Parquet read with cuDF. + +`cudf.read_parquet` performs the entire read on the device: page header decode, +codec decompression, dictionary/RLE/plain decoding and column assembly. That makes it +the like-for-like opponent for the Vortex GPU backend, which also decodes all the way +to canonical arrays on device. + +Timing excludes interpreter start, `import cudf`, CUDA context creation and any JIT +warm-up, all of which are paid once per process and are not part of a read. A warm-up +read runs first for exactly that reason. + +Emits one JSON object on stdout so the benchmark can parse it. +""" + +import argparse +import json +import sys +import time +from datetime import date + + +def synchronize() -> None: + """Block until queued device work finishes. + + `cudf.read_parquet` returns a materialized DataFrame, but synchronizing explicitly + keeps the measurement honest if that ever stops being true. + """ + try: + import cupy + + cupy.cuda.runtime.deviceSynchronize() + except ImportError: + pass + + +def normalize(frame): + """Collapses representation differences that are not value differences. + + A Parquet DATE column comes back from pyarrow as a column of `datetime.date` + objects but from cuDF as `datetime64[s]`. Those hold the same instants, yet + `check_dtype=False` does not bridge them because one side is `object`, so the + comparison reports every row as different. Coercing both sides to datetime64 + compares the dates themselves. + """ + import pandas as pd + + for name in frame.columns: + column = frame[name] + if column.dtype == object and len(column) and isinstance(column.iloc[0], date): + frame[name] = pd.to_datetime(column) + return frame + + +def verify(path: str, frame) -> None: + """Fails unless the GPU read matches a CPU Parquet read of the same file.""" + import pandas as pd + from pandas.testing import assert_frame_equal + + expected = normalize(pd.read_parquet(path)) + actual = normalize(frame.to_pandas()) + + # cuDF and pyarrow can land on different-but-equivalent dtypes (nullable vs numpy + # backed, for instance), so compare values and leave dtype policy out of it. + assert_frame_equal(actual, expected, check_dtype=False) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("path", help="Parquet file to read") + parser.add_argument("--iterations", type=int, default=1, help="timed reads to perform") + parser.add_argument( + "--verify", + action="store_true", + help="cross-check the GPU read against a CPU Parquet read", + ) + args = parser.parse_args() + + import cudf + + # Warm-up: pays CUDA context creation and any first-call JIT so they stay out of + # the timed reads below. + warmup = cudf.read_parquet(args.path) + synchronize() + + if args.verify: + verify(args.path, warmup) + + rows, columns = warmup.shape + del warmup + + runs_ns = [] + for _ in range(max(args.iterations, 1)): + start = time.perf_counter_ns() + frame = cudf.read_parquet(args.path) + synchronize() + runs_ns.append(time.perf_counter_ns() - start) + del frame + + json.dump( + { + "min_ns": min(runs_ns), + "runs_ns": runs_ns, + "rows": int(rows), + "columns": int(columns), + "verified": bool(args.verify), + }, + sys.stdout, + ) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/vortex-bench/src/conversions.rs b/vortex-bench/src/conversions.rs index 69cba42c6b0..f9c4ee63734 100644 --- a/vortex-bench/src/conversions.rs +++ b/vortex-bench/src/conversions.rs @@ -6,6 +6,8 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use arrow_array::RecordBatch; +use arrow_select::concat::concat_batches; use futures::StreamExt; use futures::TryStreamExt; use parquet::arrow::AsyncArrowWriter; @@ -96,18 +98,73 @@ fn calculate_concurrency() -> usize { /// Note: This loads the entire file into memory. For large files, use the streaming conversion like /// in [`parquet_to_vortex_stream`] instead. pub async fn parquet_to_vortex_chunks(parquet_path: PathBuf) -> anyhow::Result { + parquet_to_vortex_chunks_with_batch_size(parquet_path, None).await +} + +/// Read a Parquet file as a Vortex [`ChunkedArray`] with chunks of exactly `batch_size` rows. +/// +/// With `batch_size` set, the source batches are concatenated and re-sliced on exact boundaries, +/// so every chunk but the last has the requested length. Setting the Arrow reader's batch size +/// is not enough on its own: the reader also breaks at the source file's row group boundaries, +/// so a file whose row groups are not a multiple of the batch size still yields short batches. +/// +/// This matters when comparing against a format whose physical partitioning is explicit. Chunk +/// size becomes the Vortex file's partition size, and small chunks mean many small compressed +/// blocks — and, on the GPU, many small kernel launches. +/// +/// `None` keeps whatever batches the Parquet reader produces. +pub async fn parquet_to_vortex_chunks_with_batch_size( + parquet_path: PathBuf, + batch_size: Option, +) -> anyhow::Result { let file = File::open(parquet_path).await?; let builder = ParquetRecordBatchStreamBuilder::new(file).await?; - let reader = builder.build()?; - let chunks: Vec = parquet_to_vortex_stream(reader) - .map(|r| r.map_err(anyhow::Error::from)) + let Some(batch_size) = batch_size.filter(|size| *size > 0) else { + let chunks: Vec = parquet_to_vortex_stream(builder.build()?) + .map(|r| r.map_err(anyhow::Error::from)) + .try_collect() + .await?; + return Ok(ChunkedArray::from_iter(chunks)); + }; + + let batches: Vec = builder + .with_batch_size(batch_size) + .build()? + .map_err(anyhow::Error::from) .try_collect() .await?; + let schema = batches + .first() + .map(RecordBatch::schema) + .ok_or_else(|| anyhow::anyhow!("cannot convert an empty Parquet file"))?; + let combined = concat_batches(&schema, &batches)?; + + let mut chunks = Vec::with_capacity(combined.num_rows().div_ceil(batch_size)); + for start in (0..combined.num_rows()).step_by(batch_size) { + let len = batch_size.min(combined.num_rows() - start); + chunks.push(record_batch_to_vortex(combined.slice(start, len))?); + } + Ok(ChunkedArray::from_iter(chunks)) } +/// Convert one Arrow [`RecordBatch`] into a canonical Vortex array. +fn record_batch_to_vortex(batch: RecordBatch) -> VortexResult { + let schema = batch.schema(); + let chunk = SESSION.arrow().from_arrow_record_batch(batch, &schema)?; + let mut builder = builder_with_capacity(chunk.dtype(), chunk.len()); + + // Canonicalize the chunk. + chunk.append_to_builder( + builder.as_mut(), + &mut VortexSession::default().create_execution_ctx(), + )?; + + Ok(builder.finish()) +} + /// Create a streaming Vortex array from a Parquet reader. /// /// Streams record batches and converts them to Vortex arrays on-the-fly, avoiding loading the @@ -116,19 +173,9 @@ pub fn parquet_to_vortex_stream( reader: ParquetRecordBatchStream, ) -> impl futures::Stream> { reader.map(move |result| { - result.map_err(|e| vortex_err!(External: e)).and_then(|rb| { - let schema = rb.schema(); - let chunk = SESSION.arrow().from_arrow_record_batch(rb, &schema)?; - let mut builder = builder_with_capacity(chunk.dtype(), chunk.len()); - - // Canonicalize the chunk. - chunk.append_to_builder( - builder.as_mut(), - &mut VortexSession::default().create_execution_ctx(), - )?; - - Ok(builder.finish()) - }) + result + .map_err(|e| vortex_err!(External: e)) + .and_then(record_batch_to_vortex) }) } diff --git a/vortex-cuda/kernels/src/bit_unpack_16.cu b/vortex-cuda/kernels/src/bit_unpack_16.cu index a784df201d3..43acff3d08d 100644 --- a/vortex-cuda/kernels/src/bit_unpack_16.cu +++ b/vortex-cuda/kernels/src/bit_unpack_16.cu @@ -13,11 +13,16 @@ __device__ void _bit_unpack_16_device(const uint16_t *__restrict in, uint16_t *_ } __syncwarp(); - // Step 2: Apply patches to shared memory in parallel + // Step 2: Apply patches to shared memory in parallel. + // + // Patch values are stored in the same frame-of-reference domain as the packed values, so + // they take the same `+ reference` the lane decoder applies to every unpacked value. For a + // plain bit-packed array the reference is zero and this is a no-op; for FoR-over-bit-packed + // it is what keeps patched positions from coming out short by the reference. PatchesCursor cursor(patches, blockIdx.x, thread_idx, 32); auto patch = cursor.next(); while (patch.index != FL_CHUNK) { - shared_out[patch.index] = patch.value; + shared_out[patch.index] = patch.value + reference; patch = cursor.next(); } __syncwarp(); diff --git a/vortex-cuda/kernels/src/bit_unpack_32.cu b/vortex-cuda/kernels/src/bit_unpack_32.cu index 3f8fcb5c227..b9f29c72c3a 100644 --- a/vortex-cuda/kernels/src/bit_unpack_32.cu +++ b/vortex-cuda/kernels/src/bit_unpack_32.cu @@ -13,11 +13,16 @@ __device__ void _bit_unpack_32_device(const uint32_t *__restrict in, uint32_t *_ } __syncwarp(); - // Step 2: Apply patches to shared memory in parallel + // Step 2: Apply patches to shared memory in parallel. + // + // Patch values are stored in the same frame-of-reference domain as the packed values, so + // they take the same `+ reference` the lane decoder applies to every unpacked value. For a + // plain bit-packed array the reference is zero and this is a no-op; for FoR-over-bit-packed + // it is what keeps patched positions from coming out short by the reference. PatchesCursor cursor(patches, blockIdx.x, thread_idx, 32); auto patch = cursor.next(); while (patch.index != FL_CHUNK) { - shared_out[patch.index] = patch.value; + shared_out[patch.index] = patch.value + reference; patch = cursor.next(); } __syncwarp(); diff --git a/vortex-cuda/kernels/src/bit_unpack_64.cu b/vortex-cuda/kernels/src/bit_unpack_64.cu index ebe0b125369..aebe500c653 100644 --- a/vortex-cuda/kernels/src/bit_unpack_64.cu +++ b/vortex-cuda/kernels/src/bit_unpack_64.cu @@ -13,11 +13,16 @@ __device__ void _bit_unpack_64_device(const uint64_t *__restrict in, uint64_t *_ } __syncwarp(); - // Step 2: Apply patches to shared memory in parallel + // Step 2: Apply patches to shared memory in parallel. + // + // Patch values are stored in the same frame-of-reference domain as the packed values, so + // they take the same `+ reference` the lane decoder applies to every unpacked value. For a + // plain bit-packed array the reference is zero and this is a no-op; for FoR-over-bit-packed + // it is what keeps patched positions from coming out short by the reference. PatchesCursor cursor(patches, blockIdx.x, thread_idx, 16); auto patch = cursor.next(); while (patch.index != FL_CHUNK) { - shared_out[patch.index] = patch.value; + shared_out[patch.index] = patch.value + reference; patch = cursor.next(); } __syncwarp(); diff --git a/vortex-cuda/kernels/src/bit_unpack_8.cu b/vortex-cuda/kernels/src/bit_unpack_8.cu index b2fcfd26f04..cabea862e59 100644 --- a/vortex-cuda/kernels/src/bit_unpack_8.cu +++ b/vortex-cuda/kernels/src/bit_unpack_8.cu @@ -13,11 +13,16 @@ __device__ void _bit_unpack_8_device(const uint8_t *__restrict in, uint8_t *__re } __syncwarp(); - // Step 2: Apply patches to shared memory in parallel + // Step 2: Apply patches to shared memory in parallel. + // + // Patch values are stored in the same frame-of-reference domain as the packed values, so + // they take the same `+ reference` the lane decoder applies to every unpacked value. For a + // plain bit-packed array the reference is zero and this is a no-op; for FoR-over-bit-packed + // it is what keeps patched positions from coming out short by the reference. PatchesCursor cursor(patches, blockIdx.x, thread_idx, 32); auto patch = cursor.next(); while (patch.index != FL_CHUNK) { - shared_out[patch.index] = patch.value; + shared_out[patch.index] = patch.value + reference; patch = cursor.next(); } __syncwarp(); diff --git a/vortex-cuda/src/bit_unpack_gen.rs b/vortex-cuda/src/bit_unpack_gen.rs index 2482c0996b8..7b3ce148753 100644 --- a/vortex-cuda/src/bit_unpack_gen.rs +++ b/vortex-cuda/src/bit_unpack_gen.rs @@ -152,11 +152,16 @@ __device__ void _bit_unpack_{bits}_device(const uint{bits}_t *__restrict in, uin }} __syncwarp(); - // Step 2: Apply patches to shared memory in parallel + // Step 2: Apply patches to shared memory in parallel. + // + // Patch values are stored in the same frame-of-reference domain as the packed values, so + // they take the same `+ reference` the lane decoder applies to every unpacked value. For a + // plain bit-packed array the reference is zero and this is a no-op; for FoR-over-bit-packed + // it is what keeps patched positions from coming out short by the reference. PatchesCursor cursor(patches, blockIdx.x, thread_idx, {thread_count}); auto patch = cursor.next(); while (patch.index != FL_CHUNK) {{ - shared_out[patch.index] = patch.value; + shared_out[patch.index] = patch.value + reference; patch = cursor.next(); }} __syncwarp(); diff --git a/vortex-cuda/src/canonical.rs b/vortex-cuda/src/canonical.rs index 9f1fce7e68e..6da059f7cd5 100644 --- a/vortex-cuda/src/canonical.rs +++ b/vortex-cuda/src/canonical.rs @@ -23,6 +23,7 @@ use vortex::array::arrays::varbinview::BinaryView; use vortex::array::arrays::varbinview::VarBinViewDataParts; use vortex::array::buffer::BufferHandle; use vortex::array::legacy_session; +use vortex::array::validity::Validity; use vortex::buffer::BitBuffer; use vortex::buffer::Buffer; use vortex::buffer::ByteBuffer; @@ -36,6 +37,26 @@ pub trait CanonicalCudaExt { Self: Sized; } +/// Copies an array-backed validity mask back to the host. +/// +/// Only [`Validity::Array`] owns a buffer; the other variants are metadata and pass through. +/// Migrating the values of a nullable array without its validity leaves the mask on the +/// device, and the first host read of it — canonicalising to Arrow, say — panics in +/// `BufferHandle::unwrap_host`. +#[allow(clippy::disallowed_methods)] +async fn validity_into_host(validity: Validity) -> VortexResult { + let Validity::Array(array) = validity else { + return Ok(validity); + }; + Ok(Validity::Array( + array + .execute::(&mut legacy_session().create_execution_ctx())? + .into_host() + .await? + .into_array(), + )) +} + #[async_trait] impl CanonicalCudaExt for Canonical { #[allow(clippy::disallowed_methods)] @@ -67,13 +88,11 @@ impl CanonicalCudaExt for Canonical { struct_fields.names().clone(), host_fields, len, - validity, + validity_into_host(validity).await?, ))) } n @ Canonical::Null(_) => Ok(n), Canonical::Bool(bool) => { - // NOTE: update to copy to host when adding buffer handle. - // Also update other method to copy validity to host. let len = bool.len(); let validity = bool.validity()?; let BoolDataParts { bits, meta } = bool.into_data().into_parts(len); @@ -83,7 +102,10 @@ impl CanonicalCudaExt for Canonical { meta.len(), meta.offset(), ); - Ok(Canonical::Bool(BoolArray::new(bits, validity))) + Ok(Canonical::Bool(BoolArray::new( + bits, + validity_into_host(validity).await?, + ))) } Canonical::Primitive(prim) => { let PrimitiveDataParts { @@ -95,7 +117,7 @@ impl CanonicalCudaExt for Canonical { Ok(Canonical::Primitive(PrimitiveArray::from_byte_buffer( buffer.try_into_host()?.await?, ptype, - validity, + validity_into_host(validity).await?, ))) } Canonical::Decimal(decimal) => { @@ -106,6 +128,7 @@ impl CanonicalCudaExt for Canonical { validity, .. } = decimal.into_data_parts(); + let validity = validity_into_host(validity).await?; Ok(Canonical::Decimal(unsafe { DecimalArray::new_unchecked_handle( BufferHandle::new_host(values.try_into_host()?.await?), @@ -136,6 +159,7 @@ impl CanonicalCudaExt for Canonical { let host_buffers = try_join_all(host_buffers).await?; let host_buffers: Arc<[ByteBuffer]> = Arc::from(host_buffers); + let validity = validity_into_host(validity).await?; Ok(Canonical::VarBinView(unsafe { VarBinViewArray::new_unchecked(host_views, host_buffers, dtype, validity) })) diff --git a/vortex-cuda/src/kernel/encodings/for_.rs b/vortex-cuda/src/kernel/encodings/for_.rs index fefa4ff5ef6..cc3ceed2869 100644 --- a/vortex-cuda/src/kernel/encodings/for_.rs +++ b/vortex-cuda/src/kernel/encodings/for_.rs @@ -158,6 +158,7 @@ mod tests { use vortex::buffer::Buffer; use vortex::dtype::NativePType; use vortex::encodings::fastlanes::BitPacked; + use vortex::encodings::fastlanes::BitPackedArrayExt; use vortex::encodings::fastlanes::FoR; use vortex::encodings::fastlanes::FoRArray; use vortex::error::VortexExpect; @@ -228,4 +229,50 @@ mod tests { assert_arrays_eq!(for_array, gpu_result, &mut ctx); } + + /// Patched positions must pick up the frame of reference, exactly like unpacked ones. + /// + /// The bit-packed exceptions are stored reference-relative, so a decoder that writes them + /// straight into the output leaves every patched value short by the reference. A plain + /// bit-packed array cannot catch that: its reference is zero. + #[rstest] + #[case::u32(100_000u32)] + #[case::u64(1_000_000u64)] + #[crate::test] + async fn test_ffor_patched_values_include_reference(#[case] reference: T) -> VortexResult<()> + where + T: NativePType + Into + From, + { + let mut ctx = array_session().create_execution_ctx(); + let mut cuda_ctx = CudaSession::create_execution_ctx(&crate::cuda_session()) + .vortex_expect("failed to create execution context"); + + // Values that fit in 8 bits, with a handful that do not and so become patches. + let mut values = (0..2048u32) + .map(|i| >::from(i % 200)) + .collect::>(); + for index in [7, 1023, 1024, 2047] { + values[index] = >::from(1u32 << 17); + } + + let values = PrimitiveArray::new(Buffer::from(values), NonNullable).into_array(); + let packed = BitPacked::encode(&values, 8, &mut array_session().create_execution_ctx())?; + assert!( + packed.patches().is_some(), + "test setup expects the exceptions to be stored as patches" + ); + let for_array = FoR::try_new(packed.into_array(), reference.into())?; + + let gpu_result = FoRExecutor + .execute(for_array.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(for_array, gpu_result, &mut ctx); + + Ok(()) + } }