feat(bench): add Arrow IPC benchmark baselines (#9624)

Add `arrow-ipc` as an on-disk format in the compression-size,
compression-throughput, and random-access benchmark suites.

Compression benchmark runs emit Arrow IPC file sizes, serialization and
deserialization timings, and decoded in-memory Arrow bytes as
`uncompressed_bytes`. Random-access runs read row-addressable Arrow IPC
files.

The decoded Arrow byte count supplies logical throughput and
compression-ratio estimates. The Arrow IPC rows remain file-format
results. The develop workflow emits these records across the existing
dataset matrix.

---------

Signed-off-by: Will Manning <will@willmanning.io>

Author: lwwmanning

.github/workflows/develop-bench.yml

@@ -156,7 +156,7 @@ jobs:
           VORTEX_EXPERIMENTAL_PATCHED_ARRAY: "1"
           FLAT_LAYOUT_INLINE_ARRAY_NODE: "1"
         run: |
-          python3 scripts/compress-split.py --formats parquet,lance,vortex --emit-ingest-records
+          python3 scripts/compress-split.py --formats arrow-ipc,parquet,lance,vortex --emit-ingest-records
 
       - name: Run ${{ matrix.benchmark.name }} benchmark (per-column-encoder)
         if: matrix.benchmark.id == 'string-bench'

Cargo.lock

@@ -17,6 +17,7 @@ publish = false
 [dependencies]
 anyhow = { workspace = true }
 arrow-array = { workspace = true }
+arrow-ipc = { workspace = true }
 arrow-schema = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }

benchmarks/compress-bench/Cargo.toml

@@ -1,7 +1,15 @@
 # Compression benchmark
 
-Measures compression and decompression throughput, plus resulting file sizes, for Vortex
-versus Parquet (and optionally Lance) across a range of datasets: NYC taxi data, several
+Measures compression and decompression throughput, plus resulting file sizes, for Vortex,
+Parquet, uncompressed Arrow IPC, and optionally Lance.
+
+[Arrow IPC](https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format) is Apache
+Arrow's built-in file format, formerly called Feather V2. This suite writes it without
+optional buffer compression. Its timings therefore isolate serialization and deserialization
+without codec cost. Its file size provides the approximately 1x baseline for compression
+ratios. Parquet provides the established reference for a compressed columnar representation.
+
+The suite covers NYC taxi data and several
 [Public BI](https://github.com/cwida/public_bi_benchmark) tables (Arade, Bimbo,
 CMSprovider, Euro2016, Food, HashTags), TPC-H `l_comment` variants, and synthetic nested
 data. This is the workload behind the `Compression` PR comment.

benchmarks/compress-bench/README.md

@@ -0,0 +1,84 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fs::File;
+use std::io::Cursor;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::Instant;
+
+use arrow_array::RecordBatch;
+use arrow_ipc::reader::FileReader;
+use arrow_ipc::writer::FileWriter;
+use arrow_schema::Schema;
+use async_trait::async_trait;
+use bytes::Bytes;
+use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+use vortex_bench::Format;
+use vortex_bench::compress::Compressor;
+use vortex_bench::compress::read_projection;
+
+/// Uncompressed Arrow IPC file baseline.
+pub struct ArrowIpcCompressor;
+
+#[async_trait]
+impl Compressor for ArrowIpcCompressor {
+    fn format(&self) -> Format {
+        Format::ArrowIpc
+    }
+
+    async fn compress(&self, parquet_path: &Path) -> anyhow::Result<(u64, Duration)> {
+        let file = File::open(parquet_path)?;
+        let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
+        let schema = Arc::clone(builder.schema());
+        let batches = builder.build()?.collect::<Result<Vec<_>, _>>()?;
+
+        let mut buf = Vec::new();
+        let start = Instant::now();
+        arrow_file_write(&mut buf, &schema, &batches)?;
+        let elapsed = start.elapsed();
+        Ok((buf.len() as u64, elapsed))
+    }
+
+    async fn decompress(&self, parquet_path: &Path) -> anyhow::Result<Duration> {
+        let file = File::open(parquet_path)?;
+        let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
+        let schema = Arc::clone(builder.schema());
+        let batches = builder.build()?.collect::<Result<Vec<_>, _>>()?;
+
+        let mut buf = Vec::new();
+        arrow_file_write(&mut buf, &schema, &batches)?;
+
+        let start = Instant::now();
+        arrow_file_read(Bytes::from(buf), schema.fields().len())?;
+        Ok(start.elapsed())
+    }
+}
+
+#[inline(never)]
+fn arrow_file_write(
+    buf: &mut Vec<u8>,
+    schema: &Schema,
+    batches: &[RecordBatch],
+) -> anyhow::Result<()> {
+    let mut writer = FileWriter::try_new(buf, schema)?;
+    for batch in batches {
+        writer.write(batch)?;
+    }
+    writer.finish()?;
+    Ok(())
+}
+
+#[inline(never)]
+fn arrow_file_read(buf: Bytes, root_columns: usize) -> anyhow::Result<usize> {
+    let cursor = Cursor::new(buf);
+    let projection = read_projection(root_columns).map(<[usize]>::to_vec);
+    let reader = FileReader::try_new(cursor, projection)?;
+
+    let mut nbytes = 0;
+    for batch in reader {
+        nbytes += batch?.get_array_memory_size();
+    }
+    Ok(nbytes)
+}

benchmarks/compress-bench/src/arrow.rs

@@ -3,6 +3,7 @@
 
 #[cfg(feature = "lance")]
 pub use lance_bench::compress::LanceCompressor;
+pub mod arrow;
 pub mod gpu;
 pub mod parquet;
 pub mod vortex;

benchmarks/compress-bench/src/lib.rs

@@ -10,10 +10,12 @@ use anyhow::Context;
 use clap::Parser;
 #[cfg(feature = "lance")]
 use compress_bench::LanceCompressor;
+use compress_bench::arrow::ArrowIpcCompressor;
 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::parquet::arrow_uncompressed_size;
 use compress_bench::vortex::VortexCompressor;
 use futures::FutureExt;
 use indicatif::ProgressBar;
@@ -58,7 +60,7 @@ struct Args {
         long,
         value_delimiter = ',',
         value_enum,
-        default_values_t = vec![Format::Parquet, Format::OnDiskVortex]
+        default_values_t = vec![Format::ArrowIpc, Format::Parquet, Format::OnDiskVortex]
     )]
     formats: Vec<Format>,
     #[arg(short, long, default_value_t = 5)]
@@ -188,6 +190,7 @@ fn get_compressor(format: Format, mode: BenchMode) -> Box<dyn Compressor> {
     }
 
     match format {
+        Format::ArrowIpc => Box::new(ArrowIpcCompressor),
         Format::OnDiskVortex => Box::new(VortexCompressor),
         Format::Parquet => Box::new(ParquetCompressor::new()),
         #[cfg(feature = "lance")]
@@ -217,7 +220,11 @@ async fn run_compress(
     output_path: Option<PathBuf>,
     ingest_output: Option<PathBuf>,
 ) -> anyhow::Result<()> {
-    let targets = formats
+    let timing_targets = formats
+        .iter()
+        .map(|f| Target::new(Engine::default(), *f))
+        .collect_vec();
+    let size_targets = formats
         .iter()
         .map(|f| Target::new(Engine::default(), *f))
         .collect_vec();
@@ -360,15 +367,15 @@ async fn run_compress(
     // publishes the numbers for the datasets that did decode.
     match display_format {
         DisplayFormat::Table => {
-            render_table(&mut writer, measurements.timings, &targets)?;
+            render_table(&mut writer, measurements.timings, &timing_targets)?;
             render_table(
                 &mut writer,
-                measurements.ratios,
-                &if formats.contains(&Format::OnDiskVortex) {
-                    vec![Target::new(Engine::default(), Format::OnDiskVortex)]
-                } else {
-                    vec![]
-                },
+                measurements
+                    .ratios
+                    .into_iter()
+                    .filter(|measurement| measurement.unit == "bytes")
+                    .collect(),
+                &size_targets,
             )?;
         }
         DisplayFormat::GhJson => {
@@ -421,6 +428,11 @@ async fn run_benchmark_for_dataset(
 
     // Get the parquet file path for this dataset
     let parquet_path = dataset_handle.to_parquet_path().await?;
+    let uncompressed_size = ops
+        .contains(&CompressOp::Compress)
+        .then(|| arrow_uncompressed_size(&parquet_path))
+        .transpose()
+        .with_context(|| format!("measuring Arrow memory size for {bench_name}"))?;
 
     let mut ratios = Vec::new();
     let mut timings = Vec::new();
@@ -460,6 +472,7 @@ async fn run_benchmark_for_dataset(
                         v3_variant,
                         *format,
                         result.compressed_size,
+                        uncompressed_size.context("compression size requires Arrow memory size")?,
                     ));
                     ratios.extend(result.ratios);
                     timings.push(result.timing);

benchmarks/compress-bench/src/main.rs

@@ -45,6 +45,23 @@ impl Default for ParquetCompressor {
     }
 }
 
+/// Return the Arrow memory size after decoding the input Parquet file.
+pub fn arrow_uncompressed_size(parquet_path: &Path) -> anyhow::Result<u64> {
+    let file = File::open(parquet_path)?;
+    let reader = ParquetRecordBatchReaderBuilder::try_new(file)?.build()?;
+    let mut total = 0u64;
+
+    for batch in reader {
+        let batch = batch?;
+        let batch_size = u64::try_from(batch.get_array_memory_size())?;
+        total = total
+            .checked_add(batch_size)
+            .ok_or_else(|| anyhow::anyhow!("Arrow memory size exceeds u64"))?;
+    }
+
+    Ok(total)
+}
+
 #[async_trait]
 impl Compressor for ParquetCompressor {
     fn format(&self) -> Format {

benchmarks/compress-bench/src/parquet.rs

@@ -112,7 +112,7 @@ pub fn format_to_df_format(format: Format) -> Arc<dyn FileFormat> {
         Format::OnDiskVortex | Format::VortexCompact | Format::VortexSpatialNative => Arc::new(
             VortexFormat::new_with_options(SESSION.clone(), vortex_table_options()),
         ),
-        Format::OnDiskDuckDB | Format::Lance => {
+        Format::ArrowIpc | Format::OnDiskDuckDB | Format::Lance => {
             unimplemented!("Format {format} cannot be turned into a DataFusion `FileFormat`")
         }
     }

benchmarks/datafusion-bench/src/lib.rs

@@ -3,6 +3,12 @@
 Measures point-lookup latency: fetching individual rows by index from a file, rather than
 scanning it. This is the workload behind the `Random Access` PR comment.
 
+[Arrow IPC](https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format) is Apache
+Arrow's built-in file format, formerly called Feather V2. This suite writes it without
+optional buffer compression, so it provides the established constant-time access reference.
+Parquet provides the established reference for a compressed columnar representation. Together,
+they let the suite compare Vortex and Lance against both ends of the storage trade-off.
+
 Two access patterns are generated with a fixed seed (see [`src/main.rs`](./src/main.rs)):
 
 - **correlated**: several clusters of consecutive indices scattered across the dataset,
@@ -11,8 +17,8 @@ Two access patterns are generated with a fixed seed (see [`src/main.rs`](./src/m
   simulating lookups with no locality.
 
 Each pattern runs over four datasets (`taxi`, `feature-vectors`, `nested-lists`,
-`nested-structs`) in Parquet, Lance, and Vortex, both with a cached open file handle and
-reopening the file per lookup. CI drives the full matrix via
+`nested-structs`) in Arrow IPC, Parquet, Lance, and Vortex. Each format uses a cached open file
+handle and a per-lookup reopen mode. CI drives the full matrix via
 [`scripts/random-access-split.py`](../../scripts/random-access-split.py).
 
 ## Running locally