feat(bench): emit v3 JSONL records and dual-write to bench server

Brings the v3 emitter and CI dual-write plumbing from ct/benchmarks-v3
onto develop without the v3 server/website code. CI continues to write
v2 results to S3 unchanged; v3 ingest is gated on vars.V3_INGEST_URL
and `continue-on-error: true`, so when the variable is unset (or the
server is unreachable) the workflow no-ops.

vortex-bench:
- New `vortex-bench/src/v3.rs` with one record per `kind`
  (`query_measurement`, `compression_time`, `compression_size`,
  `random_access_time`, `vector_search_run`) plus a serde-tagged
  `V3Record` enum, JSONL writer, and snapshot tests.
- `Dataset::v3_dataset_dims()` (default `(name(), None)`) lets
  Public-BI map to `(public-bi, <subset>)`.
- `compress`/`runner` capture per-iteration timings and provide
  `SqlBenchmarkRunner::v3_records()`.

Benchmark binaries (`compress-bench`, `datafusion-bench`,
`duckdb-bench`, `lance-bench`, `random-access-bench`,
`vector-search-bench`) gain `--gh-json-v3 <path>` for JSONL emission
alongside the existing `gh-json` flow.

bench-orchestrator passes `--gh-json-v3` through `vx-bench run`.

`scripts/post-ingest.py` reads JSONL, fills the `commit` envelope
from `git show`, wraps in `{run_meta, commit, records}`, and POSTs
to `/api/ingest`. Stdlib only.

Workflows:
- `.github/workflows/bench.yml` and `sql-benchmarks.yml` add
  `--gh-json-v3 results.v3.jsonl` and a follow-up "Ingest results to
  v3 server" step.
- New `.github/workflows/v3-commit-metadata.yml` POSTs an empty
  envelope on every push to `develop` so the v3 `commits` dim stays
  populated.

Files intentionally NOT brought over: anything under
`benchmarks-website/`, the workspace member additions for the
v3 server, and workflows depending on the v3 server crate. The v3
website ships in a follow-up PR off `ct/benchmarks-v3` once
dual-write is healthy in production.

Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

.github/workflows/bench.yml

@@ -92,7 +92,7 @@ jobs:
           VORTEX_EXPERIMENTAL_PATCHED_ARRAY: "1"
           FLAT_LAYOUT_INLINE_ARRAY_NODE: "1"
         run: |
-          bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} --formats ${{ matrix.benchmark.formats }} -d gh-json -o results.json
+          bash scripts/bench-taskset.sh target/release_debug/${{ matrix.benchmark.id }} --formats ${{ matrix.benchmark.formats }} -d gh-json -o results.json --gh-json-v3 results.v3.jsonl
 
       - name: Setup AWS CLI
         uses: aws-actions/configure-aws-credentials@ec61189d14ec14c8efccab744f656cffd0e33f37  # v6
@@ -105,6 +105,19 @@ jobs:
         run: |
           bash scripts/cat-s3.sh vortex-ci-benchmark-results data.json.gz results.json
 
+      - name: Ingest results to v3 server
+        if: vars.V3_INGEST_URL != ''
+        continue-on-error: true
+        shell: bash
+        env:
+          INGEST_BEARER_TOKEN: ${{ secrets.INGEST_BEARER_TOKEN }}
+        run: |
+          python3 scripts/post-ingest.py results.v3.jsonl \
+            --server "${{ vars.V3_INGEST_URL }}" \
+            --commit-sha "${{ github.sha }}" \
+            --benchmark-id "${{ matrix.benchmark.id }}" \
+            --repo-url "${{ github.server_url }}/${{ github.repository }}"
+
       - name: Alert incident.io
         if: failure()
         uses: ./.github/actions/alert-incident-io

.github/workflows/sql-benchmarks.yml

@@ -376,6 +376,7 @@ jobs:
           bash scripts/bench-taskset.sh uv run --project bench-orchestrator vx-bench run "${{ matrix.subcommand }}" \
             --targets-json '${{ steps.targets.outputs.targets_json }}' \
             --output results.json \
+            --gh-json-v3 results.v3.jsonl \
             --no-build \
             --runner "ec2_${{ inputs.machine_type }}" \
             ${{ matrix.iterations && format('--iterations {0}', matrix.iterations) || '' }} \
@@ -395,6 +396,7 @@ jobs:
           bash scripts/bench-taskset.sh uv run --project bench-orchestrator vx-bench run "${{ matrix.subcommand }}" \
             --targets-json '${{ steps.targets.outputs.targets_json }}' \
             --output results.json \
+            --gh-json-v3 results.v3.jsonl \
             --no-build \
             --runner "ec2_${{ inputs.machine_type }}" \
             ${{ matrix.iterations && format('--iterations {0}', matrix.iterations) || '' }} \
@@ -499,6 +501,19 @@ jobs:
         run: |
           bash scripts/cat-s3.sh vortex-ci-benchmark-results data.json.gz results.json
 
+      - name: Ingest results to v3 server
+        if: inputs.mode == 'develop' && vars.V3_INGEST_URL != ''
+        continue-on-error: true
+        shell: bash
+        env:
+          INGEST_BEARER_TOKEN: ${{ secrets.INGEST_BEARER_TOKEN }}
+        run: |
+          python3 scripts/post-ingest.py results.v3.jsonl \
+            --server "${{ vars.V3_INGEST_URL }}" \
+            --commit-sha "${{ github.sha }}" \
+            --benchmark-id "${{ matrix.id }}" \
+            --repo-url "${{ github.server_url }}/${{ github.repository }}"
+
       - name: Upload File Sizes
         if: inputs.mode == 'develop' && matrix.remote_storage == null
         shell: bash

.github/workflows/v3-commit-metadata.yml

@@ -0,0 +1,35 @@
+# Posts a v3 ingest envelope with no records on every push to develop, so the
+# `commits` dim stays populated even when no benchmark ran.
+
+name: v3 commit metadata
+
+on:
+  push:
+    branches: [develop]
+  workflow_dispatch: { }
+
+permissions:
+  contents: read
+
+jobs:
+  commit-metadata:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 2
+
+      - name: Ingest commit metadata to v3 server
+        if: vars.V3_INGEST_URL != ''
+        continue-on-error: true
+        shell: bash
+        env:
+          INGEST_BEARER_TOKEN: ${{ secrets.INGEST_BEARER_TOKEN }}
+        run: |
+          echo -n > empty.jsonl
+          python3 scripts/post-ingest.py empty.jsonl \
+            --server "${{ vars.V3_INGEST_URL }}" \
+            --commit-sha "${{ github.sha }}" \
+            --benchmark-id "commit-metadata" \
+            --repo-url "${{ github.server_url }}/${{ github.repository }}"

Cargo.lock

@@ -210,6 +210,10 @@ def run(
         Path | None,
         typer.Option("--output", help="Optional path for compatibility JSONL output"),
     ] = None,
+    gh_json_v3: Annotated[
+        Path | None,
+        typer.Option("--gh-json-v3", help="Optional path for v3 JSONL records emitted by the benchmark binary"),
+    ] = None,
     options: Annotated[list[str] | None, typer.Option("--opt", help="Engine or benchmark specific options")] = None,
 ) -> None:
     """Run benchmarks with specified configuration."""
@@ -294,6 +298,7 @@ def run(
                         sample_rate=sample_rate,
                         tracing=tracing,
                         runner=runner,
+                        gh_json_v3=gh_json_v3,
                         on_result=lambda line, store_writer=ctx.write_raw_json, compatibility=compatibility_file: (
                             write_result_line(
                                 line,

bench-orchestrator/bench_orchestrator/cli.py

@@ -40,6 +40,7 @@ def build_command(
         sample_rate: int | None = None,
         tracing: bool = False,
         runner: str | None = None,
+        gh_json_v3: Path | None = None,
     ) -> list[str]:
         """Build the command used to execute a benchmark binary."""
         cmd = [
@@ -67,6 +68,8 @@ def build_command(
             cmd.append("--tracing")
         if runner:
             cmd.extend(["--runner", runner])
+        if gh_json_v3 is not None:
+            cmd.extend(["--gh-json-v3", str(gh_json_v3)])
         if options:
             for key, value in options.items():
                 cmd.extend(["--opt", f"{key}={value}"])
@@ -98,6 +101,7 @@ def run(
         sample_rate: int | None = None,
         tracing: bool = False,
         runner: str | None = None,
+        gh_json_v3: Path | None = None,
         on_result: Callable[[str], None] | None = None,
     ) -> list[str]:
         """
@@ -128,6 +132,7 @@ def run(
             sample_rate=sample_rate,
             tracing=tracing,
             runner=runner,
+            gh_json_v3=gh_json_v3,
         )
 
         if self.verbose:

bench-orchestrator/bench_orchestrator/runner/executor.py

@@ -48,6 +48,31 @@ def test_build_command_omits_formats_for_lance_backend() -> None:
     assert "1,3" in cmd
 
 
+def test_build_command_includes_gh_json_v3_when_set() -> None:
+    executor = BenchmarkExecutor(Path("/tmp/duckdb-bench"), Engine.DUCKDB)
+
+    cmd = executor.build_command(
+        benchmark=Benchmark.TPCH,
+        formats=[Format.PARQUET],
+        gh_json_v3=Path("results.v3.jsonl"),
+    )
+
+    assert "--gh-json-v3" in cmd
+    flag_idx = cmd.index("--gh-json-v3")
+    assert cmd[flag_idx + 1] == "results.v3.jsonl"
+
+
+def test_build_command_omits_gh_json_v3_when_unset() -> None:
+    executor = BenchmarkExecutor(Path("/tmp/duckdb-bench"), Engine.DUCKDB)
+
+    cmd = executor.build_command(
+        benchmark=Benchmark.TPCH,
+        formats=[Format.PARQUET],
+    )
+
+    assert "--gh-json-v3" not in cmd
+
+
 def test_run_streams_logs_without_counting_them(tmp_path: Path) -> None:
     script = tmp_path / "fake-bench.py"
     script.write_text(

bench-orchestrator/tests/test_executor.py

@@ -41,6 +41,7 @@ use vortex_bench::public_bi::PBIDataset::Euro2016;
 use vortex_bench::public_bi::PBIDataset::Food;
 use vortex_bench::public_bi::PBIDataset::HashTags;
 use vortex_bench::setup_logging_and_tracing_with_format;
+use vortex_bench::v3;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -68,6 +69,10 @@ struct Args {
     display_format: DisplayFormat,
     #[arg(short, long)]
     output_path: Option<PathBuf>,
+    /// Additionally write v3 JSONL records to this path. See
+    /// `benchmarks-website/planning/02-contracts.md`.
+    #[arg(long)]
+    gh_json_v3: Option<PathBuf>,
     #[arg(long)]
     tracing: bool,
     /// Format for the primary stderr log sink. `text` is the default human-readable format;
@@ -89,6 +94,7 @@ async fn main() -> anyhow::Result<()> {
         args.ops,
         args.display_format,
         args.output_path,
+        args.gh_json_v3,
     )
     .await
 }
@@ -114,6 +120,7 @@ async fn run_compress(
     ops: Vec<CompressOp>,
     display_format: DisplayFormat,
     output_path: Option<PathBuf>,
+    gh_json_v3: Option<PathBuf>,
 ) -> anyhow::Result<()> {
     let targets = formats
         .iter()
@@ -163,17 +170,24 @@ async fn run_compress(
     let progress = ProgressBar::new((datasets.len() * formats.len() * ops.len()) as u64);
 
     let mut measurements = vec![];
+    let mut v3_records: Vec<v3::V3Record> = Vec::new();
 
     for dataset_handle in datasets.into_iter() {
-        let m = run_benchmark_for_dataset(&progress, &formats, &ops, iterations, dataset_handle)
-            .await?;
+        let (m, mut records) =
+            run_benchmark_for_dataset(&progress, &formats, &ops, iterations, dataset_handle)
+                .await?;
         measurements.push(m);
+        v3_records.append(&mut records);
     }
 
     let measurements = CompressMeasurements::from_iter(measurements);
 
     progress.finish();
 
+    if let Some(path) = gh_json_v3 {
+        v3::write_jsonl_to_path(&path, &v3_records)?;
+    }
+
     let mut writer = create_output_writer(&display_format, output_path, BENCHMARK_ID)?;
 
     match display_format {
@@ -202,8 +216,9 @@ async fn run_benchmark_for_dataset(
     ops: &[CompressOp],
     iterations: usize,
     dataset_handle: &dyn Dataset,
-) -> anyhow::Result<CompressMeasurements> {
+) -> anyhow::Result<(CompressMeasurements, Vec<v3::V3Record>)> {
     let bench_name = dataset_handle.name();
+    let (v3_dataset, v3_variant) = dataset_handle.v3_dataset_dims();
     tracing::info!("Running {bench_name} benchmark");
 
     // Get the parquet file path for this dataset
@@ -213,6 +228,7 @@ async fn run_benchmark_for_dataset(
     let mut timings = Vec::new();
     let mut measurements_map: HashMap<(Format, CompressOp), Duration> = HashMap::new();
     let mut compressed_sizes: HashMap<Format, u64> = HashMap::new();
+    let mut v3_records: Vec<v3::V3Record> = Vec::new();
 
     for format in formats {
         let compressor = get_compressor(*format);
@@ -228,6 +244,24 @@ async fn run_benchmark_for_dataset(
                     )
                     .await?;
                     compressed_sizes.insert(*format, result.compressed_size);
+                    let all_runs_ns: Vec<u64> = result
+                        .all_runs
+                        .iter()
+                        .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
+                        .collect();
+                    v3_records.push(v3::compression_time_record(
+                        &result.timing,
+                        v3_dataset,
+                        v3_variant,
+                        CompressOp::Compress,
+                        all_runs_ns,
+                    ));
+                    v3_records.push(v3::compression_size_record(
+                        v3_dataset,
+                        v3_variant,
+                        *format,
+                        result.compressed_size,
+                    ));
                     ratios.extend(result.ratios);
                     timings.push(result.timing);
                     result.time
@@ -240,6 +274,18 @@ async fn run_benchmark_for_dataset(
                         bench_name,
                     )
                     .await?;
+                    let all_runs_ns: Vec<u64> = result
+                        .all_runs
+                        .iter()
+                        .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
+                        .collect();
+                    v3_records.push(v3::compression_time_record(
+                        &result.timing,
+                        v3_dataset,
+                        v3_variant,
+                        CompressOp::Decompress,
+                        all_runs_ns,
+                    ));
                     timings.push(result.timing);
                     result.time
                 }
@@ -258,5 +304,5 @@ async fn run_benchmark_for_dataset(
         &mut ratios,
     );
 
-    Ok(CompressMeasurements { timings, ratios })
+    Ok((CompressMeasurements { timings, ratios }, v3_records))
 }

benchmarks/compress-bench/src/main.rs

@@ -44,6 +44,7 @@ use vortex_bench::runner::BenchmarkQueryResult;
 use vortex_bench::runner::SqlBenchmarkRunner;
 use vortex_bench::runner::filter_queries;
 use vortex_bench::setup_logging_and_tracing;
+use vortex_bench::v3;
 use vortex_datafusion::metrics::VortexMetricsFinder;
 
 /// Common arguments shared across benchmarks
@@ -82,6 +83,11 @@ struct Args {
     #[arg(short)]
     output_path: Option<PathBuf>,
 
+    /// Additionally write v3 JSONL records to this path. See
+    /// `benchmarks-website/planning/02-contracts.md`.
+    #[arg(long)]
+    gh_json_v3: Option<PathBuf>,
+
     #[arg(long, default_value_t = false)]
     show_metrics: bool,
 
@@ -226,6 +232,10 @@ async fn main() -> anyhow::Result<()> {
             print_metrics(plans.as_ref());
         }
 
+        if let Some(path) = args.gh_json_v3.as_ref() {
+            v3::write_jsonl_to_path(path, &runner.v3_records())?;
+        }
+
         let benchmark_id = format!("datafusion-{}", benchmark.dataset_name());
         let writer = create_output_writer(&args.display_format, args.output_path, &benchmark_id)?;
         runner.export_to(&args.display_format, writer)?;

benchmarks/datafusion-bench/src/main.rs

@@ -24,6 +24,7 @@ use vortex_bench::runner::BenchmarkMode;
 use vortex_bench::runner::SqlBenchmarkRunner;
 use vortex_bench::runner::filter_queries;
 use vortex_bench::setup_logging_and_tracing;
+use vortex_bench::v3;
 
 /// Common arguments shared across benchmarks
 #[derive(Parser)]
@@ -58,6 +59,11 @@ struct Args {
     #[arg(short)]
     output_path: Option<PathBuf>,
 
+    /// Additionally write v3 JSONL records to this path. See
+    /// `benchmarks-website/planning/02-contracts.md`.
+    #[arg(long)]
+    gh_json_v3: Option<PathBuf>,
+
     #[arg(long, default_value_t = false)]
     track_memory: bool,
 
@@ -190,6 +196,10 @@ fn main() -> anyhow::Result<()> {
     )?;
 
     if !args.explain {
+        if let Some(path) = args.gh_json_v3.as_ref() {
+            v3::write_jsonl_to_path(path, &runner.v3_records())?;
+        }
+
         let benchmark_id = format!("duckdb-{}", benchmark.dataset_name());
         let writer = create_output_writer(&args.display_format, args.output_path, &benchmark_id)?;
         runner.export_to(&args.display_format, writer)?;