Random access for lance

Signed-off-by: Adam Gutglick <[email protected]>

Author: AdamGS

.github/workflows/bench.yml

@@ -85,7 +85,7 @@ jobs:
         env:
           RUST_BACKTRACE: full
         run: |
-          target/release_debug/${{ matrix.benchmark.id }} -d gh-json --formats parquet,lance,vortex
+          target/release_debug/${{ matrix.benchmark.id }} --formats parquet,lance,vortex -o results.json
 
       - name: Setup AWS CLI
         uses: aws-actions/configure-aws-credentials@v5
@@ -96,7 +96,7 @@ jobs:
       - name: Upload Benchmark Results
         shell: bash
         run: |
-          bash scripts/cat-s3.sh vortex-benchmark-results-database data.json.gz target/vortex-bench/${{ matrix.benchmark.id }}/results.json
+          bash scripts/cat-s3.sh vortex-benchmark-results-database data.json.gz results.json
 
   sql:
     uses: ./.github/workflows/sql-benchmarks.yml

.github/workflows/nightly-bench.yml

@@ -39,7 +39,7 @@ jobs:
             "subcommand": "tpch",
             "name": "TPC-H on NVME",
             "targets": "datafusion:parquet,datafusion:vortex,datafusion:lance,duckdb:parquet,duckdb:vortex,duckdb:duckdb",
-            "scale_factor": "--scale-factor 10.0",
+            "scale_factor": "10.0",
             "build_args": "--features lance"
           },
           {
@@ -49,15 +49,15 @@ jobs:
             "local_dir": "vortex-bench/data/tpch/10.0",
             "remote_storage": "s3://vortex-bench-dev-eu/${{github.ref_name}}/${{github.run_id}}/tpch/10.0/",
             "targets": "datafusion:parquet,datafusion:vortex,datafusion:lance,duckdb:parquet,duckdb:vortex",
-            "scale_factor": "--scale-factor 10.0",
+            "scale_factor": "10.0",
             "build_args": "--features lance"
           },
           {
             "id": "tpch-nvme",
             "subcommand": "tpch",
             "name": "TPC-H on NVME",
             "targets": "datafusion:parquet,duckdb:parquet,duckdb:vortex",
-            "scale_factor": "--scale-factor 100"
+            "scale_factor": "100"
           },
           {
             "id": "tpch-s3",
@@ -66,7 +66,7 @@ jobs:
             "local_dir": "vortex-bench/data/tpch/100.0",
             "remote_storage": "s3://vortex-bench-dev-eu/${{github.ref_name}}/${{github.run_id}}/tpch/100.0/",
             "targets": "datafusion:parquet,duckdb:parquet,duckdb:vortex",
-            "scale_factor": "--scale-factor 100.0"
+            "scale_factor": "100.0"
           },
         ]
     strategy:

Cargo.lock

@@ -169,10 +169,6 @@ pub async fn benchmark_decompress(
     })
 }
 
-// ============================================================================
-// Ratio calculations
-// ============================================================================
-
 /// Calculate cross-format comparison ratios.
 pub fn calculate_ratios(
     measurements: &HashMap<(Format, CompressOp), Duration>,

benchmarks/compress-bench/src/bench.rs

@@ -21,6 +21,7 @@ lance-encoding = { version = "0.39.0" }
 anyhow = { workspace = true }
 arrow-cast = { workspace = true }
 arrow-schema = { workspace = true }
+async-trait = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 futures = { workspace = true }
 parquet = { workspace = true }

benchmarks/lance-bench/Cargo.toml

@@ -3,3 +3,4 @@
 
 pub mod compress;
 pub mod convert;
+pub mod random_access;

benchmarks/lance-bench/src/lib.rs

@@ -0,0 +1,74 @@
+use std::fs::File;
+use std::path::PathBuf;
+
+use anyhow::anyhow;
+use async_trait::async_trait;
+use lance::Dataset;
+use lance::dataset::ProjectionRequest;
+use lance::dataset::WriteParams;
+use lance_encoding::version::LanceFileVersion;
+use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+use vortex_bench::Format;
+use vortex_bench::datasets::taxi_data::taxi_data_parquet;
+use vortex_bench::idempotent_async;
+use vortex_bench::random_access::RandomAccessor;
+
+pub async fn taxi_data_lance() -> anyhow::Result<PathBuf> {
+    idempotent_async("taxi/taxi.lance", |output_fname| async move {
+        let parquet_path = taxi_data_parquet().await?;
+
+        let file = File::open(&parquet_path)?;
+        let builder = ParquetRecordBatchReaderBuilder::try_new(file)?;
+        let reader = builder.build()?;
+
+        let write_params = WriteParams::with_storage_version(LanceFileVersion::V2_1);
+        Dataset::write(
+            reader,
+            output_fname
+                .to_str()
+                .ok_or_else(|| anyhow!("Invalid output file path"))?,
+            Some(write_params),
+        )
+        .await?;
+
+        Ok(output_fname.to_path_buf())
+    })
+    .await
+}
+
+pub struct LanceRandomAccessor {
+    path: PathBuf,
+}
+
+impl LanceRandomAccessor {
+    pub fn new(path: PathBuf) -> Self {
+        Self { path }
+    }
+}
+
+#[async_trait]
+impl RandomAccessor for LanceRandomAccessor {
+    fn format(&self) -> Format {
+        Format::Lance
+    }
+
+    fn name(&self) -> &str {
+        "random-access/lance-tokio-local-disk"
+    }
+
+    fn path(&self) -> &PathBuf {
+        &self.path
+    }
+
+    async fn take(&self, indices: Vec<u64>) -> anyhow::Result<usize> {
+        let dataset = Dataset::open(
+            self.path
+                .to_str()
+                .ok_or_else(|| anyhow!("Invalid dataset path"))?,
+        )
+        .await?;
+        let projection = ProjectionRequest::from_schema(dataset.schema().clone()); // All columns.
+        let result = dataset.take(indices.as_slice(), projection).await?;
+        Ok(result.num_rows())
+    }
+}

benchmarks/lance-bench/src/random_access.rs

@@ -18,9 +18,13 @@ publish = false
 anyhow = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 indicatif = { workspace = true }
+lance-bench = { path = "../lance-bench", optional = true }
 tokio = { workspace = true, features = ["full"] }
 vortex = { workspace = true }
 vortex-bench = { workspace = true }
 
+[features]
+lance = ["dep:lance-bench"]
+
 [lints]
 workspace = true

benchmarks/random-access-bench/Cargo.toml

@@ -7,8 +7,6 @@ use std::time::Instant;
 
 use clap::Parser;
 use indicatif::ProgressBar;
-use vortex::buffer::Buffer;
-use vortex::buffer::buffer;
 use vortex_bench::BenchmarkOutput;
 use vortex_bench::Engine;
 use vortex_bench::Format;
@@ -54,7 +52,7 @@ async fn main() -> anyhow::Result<()> {
     setup_logging_and_tracing(args.verbose, args.tracing)?;
 
     // Row count of the dataset is 3,339,715.
-    let indices = buffer![10u64, 11, 12, 13, 100_000, 3_000_000];
+    let indices = vec![10u64, 11, 12, 13, 100_000, 3_000_000];
 
     run_random_access(
         args.formats,
@@ -81,6 +79,14 @@ async fn get_accessor(format: Format) -> anyhow::Result<Box<dyn RandomAccessor>>
             let path = taxi_data_parquet().await?;
             Ok(Box::new(ParquetRandomAccessor::new(path)))
         }
+        #[cfg(feature = "lance")]
+        Format::Lance => {
+            use lance_bench::random_access::LanceRandomAccessor;
+            use lance_bench::random_access::taxi_data_lance;
+
+            let path = taxi_data_lance().await?;
+            Ok(Box::new(LanceRandomAccessor::new(path)))
+        }
         _ => unimplemented!("Random access bench not implemented for {format}"),
     }
 }
@@ -91,7 +97,7 @@ async fn get_accessor(format: Format) -> anyhow::Result<Box<dyn RandomAccessor>>
 /// collecting timing for each run.
 async fn benchmark_random_access(
     accessor: &dyn RandomAccessor,
-    indices: &Buffer<u64>,
+    indices: &[u64],
     time_limit_secs: u64,
     storage: &str,
 ) -> anyhow::Result<TimingMeasurement> {
@@ -101,9 +107,9 @@ async fn benchmark_random_access(
 
     // Run at least once, then continue until time limit
     loop {
-        let indices_clone = indices.clone();
+        let indices = indices.to_vec();
         let start = Instant::now();
-        let _row_count = accessor.take(indices_clone).await?;
+        let _row_count = accessor.take(indices).await?;
         runs.push(start.elapsed());
 
         if overall_start.elapsed() >= time_limit {
@@ -124,6 +130,8 @@ fn format_to_engine(format: Format) -> Engine {
     match format {
         Format::OnDiskVortex | Format::VortexCompact => Engine::Vortex,
         Format::Parquet => Engine::Arrow,
+        #[cfg(feature = "lance")]
+        Format::Lance => Engine::Arrow, // Is this right here?
         _ => Engine::default(),
     }
 }
@@ -135,7 +143,7 @@ async fn run_random_access(
     formats: Vec<Format>,
     time_limit: u64,
     display_format: DisplayFormat,
-    indices: Buffer<u64>,
+    indices: Vec<u64>,
     output_path: Option<PathBuf>,
 ) -> anyhow::Result<()> {
     let progress = ProgressBar::new(formats.len() as u64);

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

@@ -220,10 +220,6 @@ impl CompactionStrategy {
     }
 }
 
-// ============================================================================
-// Benchmark CLI types
-// ============================================================================
-
 /// CLI argument for selecting which benchmark to run.
 #[derive(clap::ValueEnum, Clone, Copy)]
 pub enum BenchmarkArg {