feat(bench-vortex): Add time-based benchmarking to random access benchmarks

- Add run_timed_with_setup() to bench_run.rs for time-based benchmarking
- Add --time-limit flag to random_access benchmark CLI
- Update TimingMeasurement to store Vec<Duration> runs and calculate mean
- Ensure at least one run even if it exceeds time limit
- Fall back to iteration-based mode if --time-limit is not specified

This allows different benchmark targets (Vortex, Lance, Parquet) to run for the same
time period, collecting as many samples as possible. This is more fair when targets have
significantly different performance characteristics (e.g., Parquet is much slower than
Vortex/Lance). The mean of all runs is reported.

Fixes #5066

Author: claude

bench-vortex/src/bench_run.rs

@@ -41,3 +41,41 @@ where
 
     fastest_result
 }
+
+/// Run a benchmark for a specified time limit, collecting all run durations
+/// At least one run is guaranteed even if it exceeds the time limit
+pub fn run_timed_with_setup<I, O, S, R, F>(
+    runtime: &Runtime,
+    time_limit_secs: u64,
+    mut setup: S,
+    mut routine: R,
+) -> Vec<Duration>
+where
+    S: FnMut() -> I,
+    R: FnMut(I) -> F,
+    F: Future<Output = O>,
+{
+    let time_limit = Duration::from_secs(time_limit_secs);
+    let overall_start = Instant::now();
+    let mut runs = Vec::new();
+
+    // Ensure at least one run
+    loop {
+        let state = black_box(setup());
+        let elapsed = runtime.block_on(async {
+            let start = Instant::now();
+            let output = routine(state).await;
+            let elapsed = start.elapsed();
+            drop(black_box(output));
+            elapsed
+        });
+        runs.push(elapsed);
+
+        // Check if we should continue
+        if overall_start.elapsed() >= time_limit {
+            break;
+        }
+    }
+
+    runs
+}

bench-vortex/src/bin/random_access.rs

@@ -5,7 +5,7 @@ use std::fs::File;
 use std::io::{Write, stdout};
 use std::path::PathBuf;
 
-use bench_vortex::bench_run::run_with_setup;
+use bench_vortex::bench_run::{run_timed_with_setup, run_with_setup};
 use bench_vortex::datasets::taxi_data::*;
 use bench_vortex::display::{DisplayFormat, print_measurements_json, render_table};
 use bench_vortex::measurements::TimingMeasurement;
@@ -34,8 +34,13 @@ struct Args {
         default_values_t = vec![Format::Parquet, Format::OnDiskVortex]
     )]
     formats: Vec<Format>,
+    /// Number of iterations to run (ignored if time_limit is set)
     #[arg(short, long, default_value_t = 10)]
     iterations: usize,
+    /// Time limit in seconds for each benchmark target (e.g., 10 for 10 seconds)
+    /// If set, benchmarks will run for this duration instead of a fixed number of iterations
+    #[arg(long)]
+    time_limit: Option<u64>,
     #[arg(short, long)]
     threads: Option<usize>,
     #[arg(short, long)]
@@ -61,6 +66,7 @@ fn main() -> anyhow::Result<()> {
     random_access(
         args.formats,
         runtime,
+        args.time_limit,
         args.iterations,
         args.display_format,
         indices,
@@ -75,26 +81,38 @@ fn create_timing_measurement<O, B, F>(
     storage: String,
     runtime: &Runtime,
     indices: &Buffer<u64>,
+    time_limit: Option<u64>,
     iterations: usize,
     target: Target,
 ) -> TimingMeasurement
 where
     B: FnMut(Buffer<u64>) -> F,
     F: Future<Output = O>,
 {
-    let benchmark_duration = run_with_setup(runtime, iterations, || indices.clone(), benchmark);
+    let runs = if let Some(time_limit_secs) = time_limit {
+        run_timed_with_setup(runtime, time_limit_secs, || indices.clone(), benchmark)
+    } else {
+        // For iteration-based, run once and collect that single duration
+        vec![run_with_setup(
+            runtime,
+            iterations,
+            || indices.clone(),
+            benchmark,
+        )]
+    };
 
     TimingMeasurement {
         name,
         storage,
         target,
-        time: benchmark_duration,
+        runs,
     }
 }
 
 fn random_access(
     formats: Vec<Format>,
     runtime: Runtime,
+    time_limit: Option<u64>,
     iterations: usize,
     display_format: DisplayFormat,
     indices: Buffer<u64>,
@@ -127,6 +145,7 @@ fn random_access(
                     STORAGE_NVME.to_owned(),
                     &runtime,
                     &indices,
+                    time_limit,
                     iterations,
                     target,
                 )
@@ -143,6 +162,7 @@ fn random_access(
                     STORAGE_NVME.to_owned(),
                     &runtime,
                     &indices,
+                    time_limit,
                     iterations,
                     target,
                 )
@@ -156,6 +176,7 @@ fn random_access(
                     STORAGE_NVME.to_owned(),
                     &runtime,
                     &indices,
+                    time_limit,
                     iterations,
                     target,
                 )
@@ -170,6 +191,7 @@ fn random_access(
                     STORAGE_NVME.to_owned(),
                     &runtime,
                     &indices,
+                    time_limit,
                     iterations,
                     target,
                 )

bench-vortex/src/measurements.rs

@@ -161,7 +161,23 @@ pub struct TimingMeasurement {
     pub name: String,
     pub target: Target,
     pub storage: String,
-    pub time: Duration,
+    pub runs: Vec<Duration>,
+}
+
+impl TimingMeasurement {
+    pub fn mean_time(&self) -> Duration {
+        let len = self.runs.len();
+        if len == 0 {
+            vortex_panic!("cannot have no runs");
+        }
+
+        let total_nanos: u128 = self.runs.iter().map(|d| d.as_nanos()).sum();
+        let mean_nanos = total_nanos / len as u128;
+        Duration::new(
+            u64::try_from(mean_nanos / 1_000_000_000).vortex_unwrap(),
+            u32::try_from(mean_nanos % 1_000_000_000).vortex_unwrap(),
+        )
+    }
 }
 
 impl ToTable for TimingMeasurement {
@@ -171,7 +187,7 @@ impl ToTable for TimingMeasurement {
             name: self.name.clone(),
             target: self.target,
             unit: Cow::from("μs"),
-            value: MeasurementValue::Int(self.time.as_micros()),
+            value: MeasurementValue::Int(self.mean_time().as_micros()),
         }
     }
 }
@@ -182,7 +198,7 @@ impl ToJson for TimingMeasurement {
             name: self.name.clone(),
             storage: Some(self.storage.clone()),
             unit: Some(Cow::from("ns")),
-            value: MeasurementValue::Int(self.time.as_nanos()),
+            value: MeasurementValue::Int(self.mean_time().as_nanos()),
             bytes: None,
             time: None,
             commit_id: Cow::from(GIT_COMMIT_ID.as_str()),