Store hot and cold random access benchmark results

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

@@ -235,12 +235,9 @@ fn v3_random_access_dataset_name(dataset: &str, pattern: Option<AccessPattern>)
 }
 
 fn push_v3_random_access_record(records: &mut Vec<v3::V3Record>, run: &RandomAccessRun) {
-    if run.reopen {
-        return;
-    }
-
     let dataset = v3_random_access_dataset_name(&run.dataset, run.pattern);
-    records.push(v3::random_access_record(&run.timing, &dataset));
+    let open_mode = if run.reopen { "reopen" } else { "cached" };
+    records.push(v3::random_access_record(&run.timing, &dataset, open_mode));
 }
 
 /// Open a random accessor for any supported format.
@@ -462,7 +459,7 @@ mod tests {
     }
 
     #[test]
-    fn v3_random_access_records_skip_reopen_variants() {
+    fn v3_random_access_records_include_open_modes() {
         let mut records = Vec::new();
 
         push_v3_random_access_record(&mut records, &fake_run("taxi", None, false));
@@ -475,13 +472,26 @@ mod tests {
             &fake_run("taxi", Some(AccessPattern::Correlated), true),
         );
 
-        assert_eq!(records.len(), 2);
+        assert_eq!(records.len(), 3);
         match &records[0] {
-            v3::V3Record::RandomAccessTime(record) => assert_eq!(record.dataset, "taxi"),
+            v3::V3Record::RandomAccessTime(record) => {
+                assert_eq!(record.dataset, "taxi");
+                assert_eq!(record.open_mode, "cached");
+            }
             other => panic!("expected random-access record, got {other:?}"),
         }
         match &records[1] {
-            v3::V3Record::RandomAccessTime(record) => assert_eq!(record.dataset, "taxi/uniform"),
+            v3::V3Record::RandomAccessTime(record) => {
+                assert_eq!(record.dataset, "taxi/uniform");
+                assert_eq!(record.open_mode, "cached");
+            }
+            other => panic!("expected random-access record, got {other:?}"),
+        }
+        match &records[2] {
+            v3::V3Record::RandomAccessTime(record) => {
+                assert_eq!(record.dataset, "taxi/correlated");
+                assert_eq!(record.open_mode, "reopen");
+            }
             other => panic!("expected random-access record, got {other:?}"),
         }
     }

scripts/_measurement_id.py

@@ -178,13 +178,17 @@ def measurement_id_random_access(
     commit_sha: str,
     dataset: str,
     format: str,
+    open_mode: str = "cached",
 ) -> int:
     """`measurement_id` for a `random_access_times` row. Mirrors
-    `db.rs::measurement_id_random_access`. Note: no `dataset_variant`."""
+    `db.rs::measurement_id_random_access`. Cached IDs preserve the historical
+    `(commit_sha, dataset, format)` hash. Reopen IDs append `open_mode`."""
     buf = _hasher_buf(_TAG_RANDOM_ACCESS_TIMES)
     _write_str(buf, commit_sha)
     _write_str(buf, dataset)
     _write_str(buf, format)
+    if open_mode != "cached":
+        _write_str(buf, open_mode)
     return _finish(buf)
 
 

scripts/post-ingest.py

@@ -36,7 +36,7 @@
 # MUST equal `benchmarks-website/web/lib/schema-version.ts::SCHEMA_VERSION`.
 # Bumping this is a coordinated change across the website contract, v3.rs, and
 # this script.
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 
 
 def parse_args() -> argparse.Namespace:
@@ -191,7 +191,7 @@ def build_commit(sha: str, repo_url: str, git_dir: Path | None) -> dict:
         frozenset({"dataset_variant"}),
     ),
     "random_access_time": (
-        frozenset({"commit_sha", "dataset", "format", "value_ns", "all_runtimes_ns"}),
+        frozenset({"commit_sha", "dataset", "format", "open_mode", "value_ns", "all_runtimes_ns"}),
         frozenset({"env_triple"}),
     ),
     "vector_search_run": (
@@ -394,6 +394,7 @@ def _memory_quartet_consistent(r: dict) -> bool:
         ("commit_sha", "str"),
         ("dataset", "str"),
         ("format", "str"),
+        ("open_mode", "str"),
         ("value_ns", "i64"),
         ("all_runtimes_ns", "i64_list"),
         ("env_triple", "opt_str"),
@@ -448,6 +449,10 @@ def _validate_record_values(record: dict, kind: str, index: int) -> None:
             raise SystemExit(
                 f"record {index} (query_measurement): memory fields must be populated together (all four or none)"
             )
+    elif kind == "random_access_time" and record["open_mode"] not in ("cached", "reopen"):
+        raise SystemExit(
+            f"record {index} (random_access_time): open_mode must be 'cached' or 'reopen', got {record['open_mode']!r}"
+        )
 
 
 def _upsert_returning_was_update(conn, sql: str, params: tuple) -> bool:
@@ -608,16 +613,18 @@ def _insert_random_access(conn, mid_mod, r: dict) -> bool:
         commit_sha=r["commit_sha"],
         dataset=r["dataset"],
         format=r["format"],
+        open_mode=r["open_mode"],
     )
     return _upsert_returning_was_update(
         conn,
         """
         INSERT INTO random_access_times (
-            measurement_id, commit_sha, dataset, format,
+            measurement_id, commit_sha, dataset, format, open_mode,
             value_ns, all_runtimes_ns, env_triple
-        ) VALUES (%s, %s, %s, %s, %s, %s::bigint[], %s)
+        ) VALUES (%s, %s, %s, %s, %s, %s, %s::bigint[], %s)
         ON CONFLICT (measurement_id) DO UPDATE SET
             commit_sha      = excluded.commit_sha,
+            open_mode       = excluded.open_mode,
             value_ns        = excluded.value_ns,
             all_runtimes_ns = excluded.all_runtimes_ns,
             env_triple      = excluded.env_triple
@@ -628,6 +635,7 @@ def _insert_random_access(conn, mid_mod, r: dict) -> bool:
             r["commit_sha"],
             r["dataset"],
             r["format"],
+            r["open_mode"],
             r["value_ns"],
             r["all_runtimes_ns"],
             r.get("env_triple"),

scripts/random-access-split.py

@@ -93,6 +93,15 @@ def merge(pattern: str, key: Callable[[dict], object], out_path: str) -> None:
     Path(out_path).write_text("".join(line + "\n" for line in lines), encoding="utf-8")
 
 
+def ingest_identity(record: dict) -> tuple[object, object, object, object]:
+    return (
+        record["kind"],
+        record["dataset"],
+        record["format"],
+        record["open_mode"],
+    )
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -107,7 +116,7 @@ def main() -> None:
     if args.emit_ingest_records:
         merge(
             f"{PARTS_DIR}/*.ingest.jsonl",
-            lambda record: (record["kind"], record["dataset"], record["format"]),
+            ingest_identity,
             "results.ingest.jsonl",
         )
 

scripts/tests/test_benchmark_reporting.py

@@ -13,6 +13,7 @@
 REPO_ROOT = Path(__file__).resolve().parents[2]
 COMPARE_SCRIPT = REPO_ROOT / "scripts" / "compare-benchmark-jsons.py"
 CAPTURE_SCRIPT = REPO_ROOT / "scripts" / "capture-file-sizes.py"
+RANDOM_ACCESS_SPLIT_SCRIPT = REPO_ROOT / "scripts" / "random-access-split.py"
 
 
 def load_compare_module():
@@ -24,6 +25,39 @@ def load_compare_module():
     return module
 
 
+def load_random_access_split_module():
+    spec = importlib.util.spec_from_file_location("random_access_split", RANDOM_ACCESS_SPLIT_SCRIPT)
+    assert spec is not None
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_random_access_merge_preserves_each_open_mode(tmp_path: Path) -> None:
+    split = load_random_access_split_module()
+    parts = tmp_path / "parts"
+    parts.mkdir()
+    cached = {
+        "kind": "random_access_time",
+        "dataset": "taxi",
+        "format": "parquet",
+        "open_mode": "cached",
+    }
+    reopen = cached | {"open_mode": "reopen"}
+    (parts / "0.ingest.jsonl").write_text(f"{json.dumps(cached)}\n", encoding="utf-8")
+    (parts / "1.ingest.jsonl").write_text(
+        f"{json.dumps(cached)}\n{json.dumps(reopen)}\n",
+        encoding="utf-8",
+    )
+    output = tmp_path / "results.ingest.jsonl"
+
+    split.merge(str(parts / "*.ingest.jsonl"), split.ingest_identity, str(output))
+
+    records = [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()]
+    assert records == [cached, reopen]
+
+
 def timing_row(name: str, base: int, pr: int) -> dict[str, object]:
     return {
         "name": name,

scripts/tests/test_measurement_id.py

@@ -60,3 +60,19 @@ def test_every_fact_table_is_covered():
 
     covered_tables = {vector["table"] for vector in golden["vectors"]}
     assert covered_tables == set(module.MEASUREMENT_ID_BY_TABLE)
+
+
+def test_random_access_open_mode_preserves_cached_ids_and_separates_reopen():
+    module = load_measurement_id_module()
+    dimensions = {
+        "commit_sha": "0123456789abcdef0123456789abcdef01234567",
+        "dataset": "taxi",
+        "format": "parquet",
+    }
+
+    historical = module.measurement_id_random_access(**dimensions)
+    cached = module.measurement_id_random_access(**dimensions, open_mode="cached")
+    reopen = module.measurement_id_random_access(**dimensions, open_mode="reopen")
+
+    assert cached == historical
+    assert reopen != cached

vortex-bench/src/snapshots/vortex_bench__v3__tests__snapshot_random_access_time.snap

@@ -7,6 +7,7 @@ expression: render(&record)
   "commit_sha": "<commit-sha>",
   "dataset": "taxi",
   "format": "parquet",
+  "open_mode": "cached",
   "value_ns": 850000,
   "all_runtimes_ns": [
     800000,

vortex-bench/src/v3.rs

@@ -201,6 +201,8 @@ pub struct RandomAccessTimeRecord {
     pub dataset: String,
     /// On-disk format the timing applies to.
     pub format: String,
+    /// File access mode: `cached` reuses an accessor and `reopen` opens one per take.
+    pub open_mode: String,
     /// Median per-iteration wall time in nanoseconds.
     pub value_ns: u64,
     /// Per-iteration wall times in nanoseconds.
@@ -416,13 +418,18 @@ pub fn compression_size_record(
 }
 
 /// Build a `random_access_time` record from a [`TimingMeasurement`].
-pub fn random_access_record(timing: &TimingMeasurement, dataset: &str) -> V3Record {
+pub fn random_access_record(
+    timing: &TimingMeasurement,
+    dataset: &str,
+    open_mode: &str,
+) -> V3Record {
     let value_ns = duration_as_ns(timing.median_time());
     let all_runtimes_ns = timing.runs.iter().copied().map(duration_as_ns).collect();
     V3Record::RandomAccessTime(RandomAccessTimeRecord {
         commit_sha: GIT_COMMIT_ID.clone(),
         dataset: dataset.to_string(),
         format: timing.target.format.name().to_string(),
+        open_mode: open_mode.to_string(),
         value_ns,
         all_runtimes_ns,
         env_triple: Some(ENV_TRIPLE.clone()),
@@ -657,7 +664,7 @@ mod tests {
                 Duration::from_nanos(850_000),
             ],
         };
-        let record = random_access_record(&timing, "taxi");
+        let record = random_access_record(&timing, "taxi", "cached");
         assert_snapshot!(render(&record)?);
         Ok(())
     }