fix issues

Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

bench-orchestrator/bench_orchestrator/cli.py

@@ -7,6 +7,7 @@
 from contextlib import contextmanager
 from datetime import datetime, timedelta
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import Annotated
 
 import pandas as pd
@@ -115,6 +116,38 @@ def open_results_output(path: Path | None):
         yield handle
 
 
+@contextmanager
+def temporary_v3_output_dir(enabled: bool):
+    """Create a temporary directory for per-backend v3 JSONL files."""
+    if not enabled:
+        yield None
+        return
+
+    with TemporaryDirectory(prefix="vx-bench-v3-") as temp_dir:
+        yield Path(temp_dir)
+
+
+def backend_v3_output_path(temp_dir: Path | None, index: int, backend: Engine) -> Path | None:
+    """Return the v3 JSONL path a backend should write, if v3 output is enabled."""
+    if temp_dir is None:
+        return None
+    return temp_dir / f"{index:02d}-{backend.value}.jsonl"
+
+
+def write_combined_v3_output(output_path: Path, input_paths: list[Path]) -> None:
+    """Concatenate successful per-backend v3 JSONL files into the requested output."""
+    if output_path.parent != Path():
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with output_path.open("w", encoding="utf-8") as output:
+        for input_path in input_paths:
+            if not input_path.exists():
+                raise RuntimeError(f"v3 output was not written by benchmark backend: {input_path}")
+            with input_path.open("r", encoding="utf-8") as input_file:
+                for line in input_file:
+                    output.write(line)
+
+
 def write_result_line(line: str, store_writer, compatibility_file) -> None:
     """Write a raw result line to the run store and optional compatibility output."""
     store_writer(line)
@@ -280,10 +313,16 @@ def run(
     soft_failures: list[str] = []
 
     try:
-        with store.create_run(config, build_config) as ctx, open_results_output(output) as compatibility_file:
-            for backend, backend_targets in backend_groups.items():
+        with (
+            store.create_run(config, build_config) as ctx,
+            open_results_output(output) as compatibility_file,
+            temporary_v3_output_dir(gh_json_v3 is not None) as v3_temp_dir,
+        ):
+            v3_output_parts: list[Path] = []
+            for backend_idx, (backend, backend_targets) in enumerate(backend_groups.items()):
                 executor = BenchmarkExecutor(binary_paths[backend], backend, verbose=verbose)
                 backend_formats = [target.format for target in backend_targets]
+                backend_gh_json_v3 = backend_v3_output_path(v3_temp_dir, backend_idx, backend)
 
                 try:
                     results = executor.run(
@@ -298,7 +337,7 @@ def run(
                         sample_rate=sample_rate,
                         tracing=tracing,
                         runner=runner,
-                        gh_json_v3=gh_json_v3,
+                        gh_json_v3=backend_gh_json_v3,
                         on_result=lambda line, store_writer=ctx.write_raw_json, compatibility=compatibility_file: (
                             write_result_line(
                                 line,
@@ -307,6 +346,8 @@ def run(
                             )
                         ),
                     )
+                    if backend_gh_json_v3 is not None:
+                        v3_output_parts.append(backend_gh_json_v3)
                     console.print(f"[green]{backend.value}: {len(results)} results[/green]")
                 except RuntimeError as exc:
                     ctx.metadata.partial = True
@@ -315,6 +356,9 @@ def run(
                     console.print(f"[red]{backend.value} failed: {exc}[/red]")
                     soft_failures.append(str(exc))
 
+            if gh_json_v3 is not None:
+                write_combined_v3_output(gh_json_v3, v3_output_parts)
+
             ctx.metadata.binaries = {backend.value: str(path) for backend, path in binary_paths.items()}
     except RuntimeError as exc:
         console.print(f"[red]{exc}[/red]")

bench-orchestrator/tests/test_cli.py

@@ -105,3 +105,47 @@ def fake_run(self, **kwargs):
     metadata = json.loads((run_dirs[0] / "metadata.json").read_text(encoding="utf-8"))
     assert metadata["targets"] == [{"engine": "datafusion", "format": "parquet"}]
     assert metadata["binaries"] == {"datafusion": str(binary_path)}
+
+
+def test_run_combines_gh_json_v3_output_per_backend(tmp_path, monkeypatch) -> None:
+    run_store = ResultStore(base_dir=tmp_path / "runs")
+    output_path = tmp_path / "artifacts" / "results.v3.jsonl"
+    binary_paths = {
+        cli_module.Engine.DATAFUSION: tmp_path / "datafusion-bench",
+        cli_module.Engine.DUCKDB: tmp_path / "duckdb-bench",
+    }
+    for binary_path in binary_paths.values():
+        binary_path.write_text("", encoding="utf-8")
+
+    monkeypatch.setattr(cli_module, "ResultStore", lambda: run_store)
+    monkeypatch.setattr(cli_module.BenchmarkBuilder, "get_binary_path", lambda self, backend: binary_paths[backend])
+
+    seen_backend_paths = []
+
+    def fake_run(self, **kwargs):
+        backend_output = kwargs["gh_json_v3"]
+        assert backend_output is not None
+        assert backend_output != output_path
+        backend_output.write_text(f"{self.backend.value}-v3\n", encoding="utf-8")
+        seen_backend_paths.append(backend_output)
+        return []
+
+    monkeypatch.setattr(BenchmarkExecutor, "run", fake_run)
+
+    result = runner.invoke(
+        cli_module.app,
+        [
+            "run",
+            "tpch",
+            "--targets-json",
+            '[{"engine":"datafusion","format":"parquet"},{"engine":"duckdb","format":"parquet"}]',
+            "--no-build",
+            "--gh-json-v3",
+            str(output_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert output_path.read_text(encoding="utf-8") == "datafusion-v3\nduckdb-v3\n"
+    assert len(seen_backend_paths) == 2
+    assert seen_backend_paths[0] != seen_backend_paths[1]

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

@@ -280,6 +280,28 @@ fn measurement_name(dataset: &str, pattern: Option<AccessPattern>, format: Forma
     }
 }
 
+fn v3_random_access_dataset_name(dataset: &str, pattern: Option<AccessPattern>) -> String {
+    match pattern {
+        Some(pattern) => format!("{dataset}/{}", pattern.name()),
+        None => dataset.to_string(),
+    }
+}
+
+fn push_v3_random_access_record(
+    records: &mut Vec<v3::V3Record>,
+    measurement: &TimingMeasurement,
+    dataset: &str,
+    pattern: Option<AccessPattern>,
+    reopen: bool,
+) {
+    if reopen {
+        return;
+    }
+
+    let dataset = v3_random_access_dataset_name(dataset, pattern);
+    records.push(v3::random_access_record(measurement, &dataset));
+}
+
 /// Map format to the appropriate engine for random access benchmarks.
 fn format_to_engine(format: Format) -> Engine {
     match format {
@@ -388,7 +410,13 @@ async fn run_random_access(
                     )
                     .await?;
 
-                    v3_records.push(v3::random_access_record(&measurement, dataset.name()));
+                    push_v3_random_access_record(
+                        &mut v3_records,
+                        &measurement,
+                        dataset.name(),
+                        None,
+                        reopen,
+                    );
                     targets.push(measurement.target);
                     measurements.push(measurement);
                     progress.inc(1);
@@ -415,7 +443,13 @@ async fn run_random_access(
                     )
                     .await?;
 
-                    v3_records.push(v3::random_access_record(&measurement, dataset.name()));
+                    push_v3_random_access_record(
+                        &mut v3_records,
+                        &measurement,
+                        dataset.name(),
+                        Some(*pattern),
+                        reopen,
+                    );
                     targets.push(measurement.target);
                     measurements.push(measurement);
                     progress.inc(1);
@@ -443,3 +477,58 @@ async fn run_random_access(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn v3_random_access_dataset_names_match_schema_dims() {
+        assert_eq!(v3_random_access_dataset_name("taxi", None), "taxi");
+        assert_eq!(
+            v3_random_access_dataset_name("taxi", Some(AccessPattern::Correlated)),
+            "taxi/correlated"
+        );
+        assert_eq!(
+            v3_random_access_dataset_name("feature-vectors", Some(AccessPattern::Uniform)),
+            "feature-vectors/uniform"
+        );
+    }
+
+    #[test]
+    fn v3_random_access_records_skip_reopen_variants() {
+        let measurement = TimingMeasurement {
+            name: "random-access/taxi/uniform/parquet-tokio-local-disk".to_string(),
+            target: Target::new(Engine::Arrow, Format::Parquet),
+            storage: STORAGE_NVME.to_string(),
+            runs: vec![Duration::from_nanos(10)],
+        };
+        let mut records = Vec::new();
+
+        push_v3_random_access_record(&mut records, &measurement, "taxi", None, false);
+        push_v3_random_access_record(
+            &mut records,
+            &measurement,
+            "taxi",
+            Some(AccessPattern::Uniform),
+            false,
+        );
+        push_v3_random_access_record(
+            &mut records,
+            &measurement,
+            "taxi",
+            Some(AccessPattern::Correlated),
+            true,
+        );
+
+        assert_eq!(records.len(), 2);
+        match &records[0] {
+            v3::V3Record::RandomAccessTime(record) => assert_eq!(record.dataset, "taxi"),
+            other => panic!("expected random-access record, got {other:?}"),
+        }
+        match &records[1] {
+            v3::V3Record::RandomAccessTime(record) => assert_eq!(record.dataset, "taxi/uniform"),
+            other => panic!("expected random-access record, got {other:?}"),
+        }
+    }
+}

vortex-bench/src/v3.rs

@@ -201,12 +201,16 @@ pub struct VectorSearchRunRecord {
 /// `benchmarks-website/planning/benchmark-mapping.md`.
 pub fn benchmark_dataset_dims(d: &BenchmarkDataset) -> (String, Option<String>, Option<String>) {
     match d {
-        BenchmarkDataset::TpcH { scale_factor } => {
-            ("tpch".to_string(), None, Some(scale_factor.clone()))
-        }
-        BenchmarkDataset::TpcDS { scale_factor } => {
-            ("tpcds".to_string(), None, Some(scale_factor.clone()))
-        }
+        BenchmarkDataset::TpcH { scale_factor } => (
+            "tpch".to_string(),
+            None,
+            Some(canonical_tpc_scale_factor(scale_factor)),
+        ),
+        BenchmarkDataset::TpcDS { scale_factor } => (
+            "tpcds".to_string(),
+            None,
+            Some(canonical_tpc_scale_factor(scale_factor)),
+        ),
         BenchmarkDataset::ClickBench { flavor } => {
             let variant = match flavor {
                 Flavor::Partitioned => "partitioned",
@@ -237,6 +241,7 @@ pub fn query_measurement_record(
     let (dataset, dataset_variant, scale_factor) = benchmark_dataset_dims(&qm.benchmark_dataset);
     let value_ns = duration_as_ns(qm.median_run());
     let all_runtimes_ns = qm.runs.iter().copied().map(duration_as_ns).collect();
+    let query_idx = v3_query_idx(qm);
     let (peak_physical, peak_virtual, physical_delta, virtual_delta) = match memory {
         Some(m) => (
             Some(m.peak_physical_memory),
@@ -251,7 +256,7 @@ pub fn query_measurement_record(
         dataset,
         dataset_variant,
         scale_factor,
-        query_idx: u32::try_from(qm.query_idx).unwrap_or(u32::MAX),
+        query_idx,
         storage: qm.storage.clone(),
         engine: engine_label(qm.target.engine).to_string(),
         format: qm.target.format.name().to_string(),
@@ -383,6 +388,23 @@ fn duration_as_ns(d: std::time::Duration) -> u64 {
     u64::try_from(d.as_nanos()).unwrap_or(u64::MAX)
 }
 
+fn canonical_tpc_scale_factor(scale_factor: &str) -> String {
+    let trimmed = scale_factor.trim();
+    match trimmed.parse::<f64>() {
+        Ok(value) if value.is_finite() => format!("{value}"),
+        _ => scale_factor.to_string(),
+    }
+}
+
+fn v3_query_idx(qm: &QueryMeasurement) -> u32 {
+    let query_idx = if matches!(&qm.benchmark_dataset, BenchmarkDataset::ClickBench { .. }) {
+        qm.query_idx.saturating_add(1)
+    } else {
+        qm.query_idx
+    };
+    u32::try_from(query_idx).unwrap_or(u32::MAX)
+}
+
 fn engine_label(engine: Engine) -> &'static str {
     match engine {
         Engine::Vortex => "vortex",
@@ -461,7 +483,7 @@ mod tests {
     #[test]
     fn snapshot_query_measurement_clickbench_no_memory() -> anyhow::Result<()> {
         let qm = QueryMeasurement {
-            query_idx: 1,
+            query_idx: 0,
             target: Target::new(Engine::DuckDB, Format::Parquet),
             benchmark_dataset: BenchmarkDataset::ClickBench {
                 flavor: Flavor::Partitioned,
@@ -478,6 +500,22 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn tpc_scale_factors_are_canonicalized_for_query_dims() {
+        assert_eq!(
+            benchmark_dataset_dims(&BenchmarkDataset::TpcH {
+                scale_factor: "1.0".to_string()
+            }),
+            ("tpch".to_string(), None, Some("1".to_string()))
+        );
+        assert_eq!(
+            benchmark_dataset_dims(&BenchmarkDataset::TpcDS {
+                scale_factor: "10.0".to_string()
+            }),
+            ("tpcds".to_string(), None, Some("10".to_string()))
+        );
+    }
+
     #[test]
     fn snapshot_compression_time_encode() -> anyhow::Result<()> {
         let timing = CompressionTimingMeasurement {