[claude] feat(bench): add --gh-json-v3 emitter and post-ingest script (#7638)

## Summary

Implements the alpha **emitter** component for `bench.vortex.dev` v3,
per

[`benchmarks-website/planning/components/emitter.md`](https://github.com/vortex-data/vortex/blob/ct/benchmarks-v3/benchmarks-website/planning/components/emitter.md).

**Purely additive** to v2's emission path — the existing `-d gh-json -o
...`
form is untouched.

### Rust emitter (`vortex-bench`)

- New `vortex-bench/src/v3.rs` module with one record type per `kind`
  (`query_measurement`, `compression_time`, `compression_size`,
  `random_access_time`, `vector_search_run`) plus serde-tagged
  `V3Record` enum. Field shapes match

[`02-contracts.md`](https://github.com/vortex-data/vortex/blob/ct/benchmarks-v3/benchmarks-website/planning/02-contracts.md);
  dataset/variant/scale-factor mapping follows

[`benchmark-mapping.md`](https://github.com/vortex-data/vortex/blob/ct/benchmarks-v3/benchmarks-website/planning/benchmark-mapping.md).
- Each benchmark binary gains a `--gh-json-v3 <PATH>` flag that writes
  bare records as JSONL (no envelope), alongside the legacy
  `--display-format gh-json -o ...` flow:
  - `compress-bench` — `compression_time` (encode/decode) +
    `compression_size`. Cross-format ratios are **not** emitted; ratios
    are computed read-side per `decisions.md`.
  - `datafusion-bench`, `duckdb-bench`, `lance-bench` —
    `query_measurement`, with optional memory fields populated when
    `--track-memory` is on. `QueryMeasurement` and the paired
`MemoryMeasurement` collapse into one record
(`SqlBenchmarkRunner::v3_records`).
  - `random-access-bench` — `random_access_time`, with the dataset name
    plumbed alongside `TimingMeasurement`.
- `vector-search-bench` — `vector_search_run`, with `dataset`, `layout`,
`threshold`, `iterations` plumbed in (they don't live on `ScanTiming`).
- `insta` snapshot tests cover one record per `kind`, scrubbing
  `commit_sha` and `env_triple`.

### Post-ingest script

`scripts/post-ingest.py` (Python 3, stdlib only — `urllib`, `json`,
`subprocess`):
- reads JSONL of records,
- fills the `commit` envelope from `git show` for the SHA passed in,
- wraps in `{run_meta, commit, records}` per the contract,
- POSTs to `<server>/api/ingest` with `Authorization: Bearer ...` from
  `INGEST_BEARER_TOKEN`,
- exits non-zero on 4xx/5xx. **No retries, no spool, no S3 outbox** —
  deferred per the alpha plan.

### Out of scope (deferred)

CI workflow integration, dual-write, `bench-orchestrator` updates,
retry/spool/outbox, replacing the v2 CLI form. All listed in

[`deferred.md`](https://github.com/vortex-data/vortex/blob/ct/benchmarks-v3/benchmarks-website/planning/deferred.md).

## Test plan

- [x] `cargo test -p vortex-bench --lib` — 48 passed (7 new `v3` tests,
one snapshot per kind plus a JSONL round-trip).
- [x] `cargo build -p vortex-bench -p compress-bench -p datafusion-bench
-p duckdb-bench -p lance-bench -p random-access-bench -p
vector-search-bench` — all clean.
- [x] `cargo clippy --all-targets` on changed crates (skipping
`duckdb-bench`, blocked by an unrelated pre-existing
`cognitive_complexity` lint in `vortex-duckdb` on `ct/benchmarks-v3`).
- [x] `cargo +nightly fmt --all`.
- [x] End-to-end smoke: `scripts/post-ingest.py` against a Python
`http.server` mock — 200 → exit 0 with `{"inserted":1,"updated":0}`; 400
→ exit 1 with the server body on stderr.
- [ ] Real round-trip against an actual alpha server — blocked on the
server component landing (acceptance criterion 3 in the emitter plan;
verifiable once the server PR exists).

https://claude.ai/code/session_017qh4ju4FtkizW6s67JEhPW

---
_Generated by [Claude
Code](https://claude.ai/code/session_017qh4ju4FtkizW6s67JEhPW)_

---------

Signed-off-by: Claude <noreply@anthropic.com>
Co-authored-by: Claude <noreply@anthropic.com>
Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: connortsui20

Cargo.lock

@@ -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)?;

benchmarks/duckdb-bench/src/main.rs

@@ -28,6 +28,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;
 
 /// Lance benchmark tool - runs SQL queries against Lance format data using DataFusion
 #[derive(Parser)]
@@ -59,6 +60,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)]
     hide_progress_bar: bool,
 
@@ -124,6 +130,10 @@ async fn main() -> anyhow::Result<()> {
         )
         .await?;
 
+    if let Some(path) = args.gh_json_v3.as_ref() {
+        v3::write_jsonl_to_path(path, &runner.v3_records())?;
+    }
+
     let benchmark_id = format!("lance-{}", benchmark.dataset_name());
     let writer = create_output_writer(&args.display_format, args.output_path, &benchmark_id)?;
     runner.export_to(&args.display_format, writer)?;

benchmarks/lance-bench/src/main.rs

@@ -32,6 +32,7 @@ use vortex_bench::random_access::RandomAccessor;
 use vortex_bench::random_access::VortexRandomAccessor;
 use vortex_bench::setup_logging_and_tracing;
 use vortex_bench::utils::constants::STORAGE_NVME;
+use vortex_bench::v3;
 
 // ---------------------------------------------------------------------------
 // Access patterns
@@ -173,6 +174,10 @@ struct Args {
     display_format: DisplayFormat,
     #[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>,
     /// Which datasets to benchmark random access on.
     #[arg(
         long,
@@ -205,6 +210,7 @@ async fn main() -> Result<()> {
         args.open_mode,
         args.display_format,
         args.output_path,
+        args.gh_json_v3,
     )
     .await
 }
@@ -340,6 +346,7 @@ async fn run_random_access(
     open_mode: OpenMode,
     display_format: DisplayFormat,
     output_path: Option<PathBuf>,
+    gh_json_v3: Option<PathBuf>,
 ) -> Result<()> {
     let reopen_variants: &[bool] = match open_mode {
         OpenMode::Cached => &[false],
@@ -358,6 +365,7 @@ async fn run_random_access(
 
     let mut targets = Vec::new();
     let mut measurements = Vec::new();
+    let mut v3_records: Vec<v3::V3Record> = Vec::new();
 
     for dataset in datasets {
         for format in &formats {
@@ -380,6 +388,7 @@ async fn run_random_access(
                     )
                     .await?;
 
+                    v3_records.push(v3::random_access_record(&measurement, dataset.name()));
                     targets.push(measurement.target);
                     measurements.push(measurement);
                     progress.inc(1);
@@ -406,6 +415,7 @@ async fn run_random_access(
                     )
                     .await?;
 
+                    v3_records.push(v3::random_access_record(&measurement, dataset.name()));
                     targets.push(measurement.target);
                     measurements.push(measurement);
                     progress.inc(1);
@@ -416,6 +426,10 @@ async fn run_random_access(
 
     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 {

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

@@ -28,6 +28,7 @@ use vector_search_bench::scan::ScanConfig;
 use vector_search_bench::scan::ScanTiming;
 use vector_search_bench::scan::run_search_scan;
 use vortex_bench::setup_logging_and_tracing;
+use vortex_bench::v3;
 use vortex_bench::vector_dataset;
 use vortex_bench::vector_dataset::TrainLayout;
 use vortex_bench::vector_dataset::VectorDataset;
@@ -71,6 +72,11 @@ struct Args {
     #[arg(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>,
+
     /// Emit verbose tracing.
     #[arg(short, long)]
     verbose: bool,
@@ -143,6 +149,36 @@ async fn main() -> Result<()> {
         vortex_results: &pairs,
     };
 
+    // Emit v3 JSONL if requested. The records carry the per-scan dimensions that
+    // ScanTiming itself does not (dataset, layout, threshold).
+    if let Some(path) = args.gh_json_v3.as_ref() {
+        let records: Vec<v3::V3Record> = scan_timings
+            .iter()
+            .map(|scan| {
+                let all_runs_ns: Vec<u64> = scan
+                    .all_runs
+                    .iter()
+                    .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
+                    .collect();
+                let median_ns = u64::try_from(scan.median.as_nanos()).unwrap_or(u64::MAX);
+                v3::vector_search_record(
+                    v3::VectorSearchDims {
+                        dataset: dataset.name(),
+                        layout: layout.label(),
+                        flavor: scan.flavor.label(),
+                        threshold: f64::from(args.threshold),
+                    },
+                    median_ns,
+                    all_runs_ns,
+                    scan.matches,
+                    scan.rows_scanned,
+                    scan.bytes_scanned,
+                )
+            })
+            .collect();
+        v3::write_jsonl_to_path(path, &records)?;
+    }
+
     // Print the results.
     if let Some(path) = args.output_path {
         let mut file =