feat(bench): add Arrow IPC benchmark baselines

.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'

benchmarks/compress-bench/Cargo.toml

@@ -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/README.md

@@ -1,7 +1,8 @@
 # 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.
+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/src/arrow.rs

@@ -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/lib.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/main.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/parquet.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/datafusion-bench/src/lib.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/random-access-bench/README.md

@@ -11,8 +11,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

benchmarks/random-access-bench/src/lib.rs

@@ -20,6 +20,7 @@ use vortex_bench::create_output_writer;
 use vortex_bench::display::DisplayFormat;
 use vortex_bench::display::print_measurements_json;
 use vortex_bench::measurements::TimingMeasurement;
+use vortex_bench::random_access::ArrowIpcRandomAccessor;
 use vortex_bench::random_access::BenchDataset;
 use vortex_bench::random_access::ParquetRandomAccessor;
 use vortex_bench::random_access::RandomAccessor;
@@ -240,6 +241,10 @@ async fn open_accessor(
         format.ext()
     );
     match format {
+        Format::ArrowIpc => {
+            let path = dataset.path(format).await?;
+            Ok(Box::new(ArrowIpcRandomAccessor::open(path, name)?))
+        }
         Format::OnDiskVortex | Format::VortexCompact => {
             let path = dataset.path(format).await?;
             Ok(Box::new(

benchmarks/random-access-bench/src/main.rs

@@ -49,7 +49,7 @@ struct Args {
         long,
         value_delimiter = ',',
         value_parser = Format::parse_allowed,
-        default_values = ["parquet", "vortex", "lance"]
+        default_values = ["arrow-ipc", "parquet", "vortex", "lance"]
     )]
     formats: Vec<Format>,
     /// Time limit in seconds for each benchmark target (e.g., 10 for 10 seconds).

scripts/compress-split.py

@@ -83,7 +83,7 @@ def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
         "--formats",
-        default="parquet,vortex",
+        default="arrow-ipc,parquet,vortex",
         help="comma-separated formats to forward to compress-bench",
     )
     parser.add_argument(

scripts/post-ingest.py

@@ -36,7 +36,7 @@
 # MUST equal `benchmarks-website/web/lib/schema-version.ts::SCHEMA_VERSION`.
 # Bumping this is a coordinated change across the website contract, v3.rs, and
 # this script.
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
 
 def parse_args() -> argparse.Namespace:
@@ -187,7 +187,7 @@ def build_commit(sha: str, repo_url: str, git_dir: Path | None) -> dict:
         frozenset({"dataset_variant", "env_triple"}),
     ),
     "compression_size": (
-        frozenset({"commit_sha", "dataset", "format", "value_bytes"}),
+        frozenset({"commit_sha", "dataset", "format", "value_bytes", "uncompressed_bytes"}),
         frozenset({"dataset_variant"}),
     ),
     "random_access_time": (
@@ -388,6 +388,7 @@ def _memory_quartet_consistent(r: dict) -> bool:
         ("dataset_variant", "opt_str"),
         ("format", "str"),
         ("value_bytes", "i64"),
+        ("uncompressed_bytes", "i64"),
     ),
     "random_access_time": (
         ("commit_sha", "str"),
@@ -581,11 +582,12 @@ def _insert_compression_size(conn, mid_mod, r: dict) -> bool:
         """
         INSERT INTO compression_sizes (
             measurement_id, commit_sha, dataset, dataset_variant,
-            format, value_bytes
-        ) VALUES (%s, %s, %s, %s, %s, %s)
+            format, value_bytes, uncompressed_bytes
+        ) VALUES (%s, %s, %s, %s, %s, %s, %s)
         ON CONFLICT (measurement_id) DO UPDATE SET
-            commit_sha   = excluded.commit_sha,
-            value_bytes  = excluded.value_bytes
+            commit_sha          = excluded.commit_sha,
+            value_bytes         = excluded.value_bytes,
+            uncompressed_bytes  = excluded.uncompressed_bytes
         RETURNING (xmax = 0) AS inserted
         """,
         (
@@ -595,6 +597,7 @@ def _insert_compression_size(conn, mid_mod, r: dict) -> bool:
             r.get("dataset_variant"),
             r["format"],
             r["value_bytes"],
+            r["uncompressed_bytes"],
         ),
     )
 

scripts/random-access-split.py

@@ -18,7 +18,7 @@
 PARTS_DIR = Path("parts")
 
 DATASETS = ["taxi", "feature-vectors", "nested-lists", "nested-structs"]
-FORMATS = ["parquet", "lance", "vortex"]
+FORMATS = ["arrow-ipc", "parquet", "lance", "vortex"]
 PATTERNS = ["correlated", "uniform"]
 OPEN_MODES = ["cached", "reopen"]
 

vortex-bench/Cargo.toml

@@ -33,6 +33,7 @@ vortex-tensor = { workspace = true } # TODO(connor): In the future, this might b
 
 anyhow = { workspace = true }
 arrow-array = { workspace = true }
+arrow-ipc = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 async-trait = { workspace = true }
@@ -84,6 +85,7 @@ wkb = { workspace = true }
 [dev-dependencies]
 insta = { workspace = true }
 rstest = { workspace = true }
+tempfile = { workspace = true }
 
 [features]
 unstable_encodings = ["vortex/unstable_encodings"]