diff --git a/.github/actions/fetch-bench-baseline/action.yml b/.github/actions/fetch-bench-baseline/action.yml
new file mode 100644
index 00000000000..7141c6f655b
--- /dev/null
+++ b/.github/actions/fetch-bench-baseline/action.yml
@@ -0,0 +1,74 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+name: "Fetch benchmark baseline"
+description: >-
+ Write the develop baseline for a PR benchmark run, read from the benchmarks Postgres: assume the
+ read-only benchmarks role over OIDC and sign in as bench_ci_read with an RDS IAM token. If the
+ baseline cannot be read, it writes no baseline file and the report says the baseline could not be
+ read. The calling job needs `id-token: write`. Composite actions cannot read `vars`, so callers
+ pass the repository variables in.
+
+inputs:
+ pr-records:
+ description: "The PR run's --ingest-jsonl output"
+ default: results.ingest.jsonl
+ output:
+ description: "Where to write the baseline records"
+ default: base.ingest.jsonl
+ role-arn:
+ description: "The read-only benchmarks role (vars.GH_BENCH_READ_ROLE_ARN)"
+ required: true
+ region:
+ description: "AWS region of the benchmarks RDS instance (vars.RDS_BENCH_REGION)"
+ required: true
+ endpoint:
+ description: "Hostname of the benchmarks RDS instance (vars.RDS_BENCH_INSTANCE_ENDPOINT)"
+ required: true
+ database:
+ description: "Name of the benchmarks database (vars.RDS_BENCH_DB_NAME)"
+ required: true
+
+runs:
+ using: "composite"
+ steps:
+ # The credentials stay in step outputs so the job's existing AWS credentials, which later
+ # steps such as the S3 cleanup still use, are not replaced.
+ - name: Configure AWS credentials for the benchmarks database (OIDC)
+ id: aws
+ if: inputs.role-arn != ''
+ continue-on-error: true
+ uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6
+ with:
+ role-to-assume: ${{ inputs.role-arn }}
+ aws-region: ${{ inputs.region }}
+ retry-max-attempts: 1
+ output-credentials: true
+ output-env-credentials: false
+ - name: Fetch baseline
+ shell: bash
+ env:
+ PR_RECORDS: ${{ inputs.pr-records }}
+ OUTPUT: ${{ inputs.output }}
+ AWS_OUTCOME: ${{ steps.aws.outcome }}
+ AWS_ACCESS_KEY_ID: ${{ steps.aws.outputs.aws-access-key-id }}
+ AWS_SECRET_ACCESS_KEY: ${{ steps.aws.outputs.aws-secret-access-key }}
+ AWS_SESSION_TOKEN: ${{ steps.aws.outputs.aws-session-token }}
+ AWS_REGION: ${{ inputs.region }}
+ RDS_BENCH_INSTANCE_ENDPOINT: ${{ inputs.endpoint }}
+ RDS_BENCH_DB_NAME: ${{ inputs.database }}
+ run: |
+ set -Eeu -o pipefail
+ if [[ "$AWS_OUTCOME" != "success" ]]; then
+ # No baseline file tells the report that the baseline could not be read.
+ echo "::warning::Could not assume the read-only benchmarks role; reporting without a baseline."
+ exit 0
+ fi
+ curl -fsSL https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem \
+ -o "${RUNNER_TEMP}/rds-global-bundle.pem"
+ DSN="postgresql://bench_ci_read@${RDS_BENCH_INSTANCE_ENDPOINT}:5432/${RDS_BENCH_DB_NAME}?sslmode=verify-full&sslrootcert=${RUNNER_TEMP}/rds-global-bundle.pem"
+ if ! uv run --no-project scripts/fetch-benchmark-baseline.py "$PR_RECORDS" "$OUTPUT" \
+ --postgres "$DSN" --region "$AWS_REGION"; then
+ echo "::warning::Could not read the baseline from the benchmarks database; reporting without a baseline."
+ rm -f "$OUTPUT"
+ fi
diff --git a/.github/workflows/commit-metadata.yml b/.github/workflows/commit-metadata.yml
index 6e3ed76c3c8..a412b22d6fc 100644
--- a/.github/workflows/commit-metadata.yml
+++ b/.github/workflows/commit-metadata.yml
@@ -59,4 +59,5 @@ jobs:
scripts/post-ingest.py empty.jsonl \
--postgres "${DSN}" \
--commit-sha "${{ github.sha }}" \
+ --branch "${{ github.ref_name }}" \
--region "${AWS_REGION}"
diff --git a/.github/workflows/develop-bench.yml b/.github/workflows/develop-bench.yml
index 8006b420540..b422ad9697d 100644
--- a/.github/workflows/develop-bench.yml
+++ b/.github/workflows/develop-bench.yml
@@ -263,6 +263,7 @@ jobs:
scripts/post-ingest.py results.ingest.jsonl \
--postgres "${DSN}" \
--commit-sha "${{ github.sha }}" \
+ --branch "${{ github.ref_name }}" \
--region "${AWS_REGION}"
- name: Alert incident.io
diff --git a/.github/workflows/pr-bench-gpu-compress.yml b/.github/workflows/pr-bench-gpu-compress.yml
index 4c9676b59a1..3f8b8d0f4f7 100644
--- a/.github/workflows/pr-bench-gpu-compress.yml
+++ b/.github/workflows/pr-bench-gpu-compress.yml
@@ -14,6 +14,7 @@ on:
permissions:
contents: read
pull-requests: write
+ id-token: write # enables AWS-GitHub OIDC
jobs:
gpu-compress-bench:
@@ -48,9 +49,11 @@ jobs:
- name: Install cuDF
# The GPU Parquet number is a cuDF `read_parquet`. cuDF ships prebuilt manylinux wheels
# on NVIDIA's index, so it stays a runtime dependency and never enters the Rust build.
+ # `--no-config` keeps the workspace's `exclude-newer` off this standalone venv: NVIDIA's
+ # index publishes no upload dates, so that cutoff would filter out every cuDF wheel.
run: |
uv venv --python 3.12 .venv-cudf
- uv pip install --python .venv-cudf \
+ uv pip install --no-config --python .venv-cudf \
--extra-index-url https://pypi.nvidia.com \
cudf-cu12 pandas pyarrow
echo "$PWD/.venv-cudf/bin" >> "$GITHUB_PATH"
@@ -78,18 +81,22 @@ jobs:
# (cuda.yaml), which also sets only FLAT_LAYOUT_INLINE_ARRAY_NODE.
FLAT_LAYOUT_INLINE_ARRAY_NODE: "1"
run: |
- python3 scripts/compress-split.py --gpu-decompress
+ python3 scripts/compress-split.py --gpu-decompress --emit-ingest-records
+ - name: Fetch baseline from the benchmarks database
+ uses: ./.github/actions/fetch-bench-baseline
+ with:
+ role-arn: ${{ vars.GH_BENCH_READ_ROLE_ARN }}
+ region: ${{ vars.RDS_BENCH_REGION }}
+ endpoint: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }}
+ database: ${{ vars.RDS_BENCH_DB_NAME }}
+
- name: Publish results
shell: bash
run: |
set -Eeu -o pipefail -x
- python3 scripts/s3-download.py s3://vortex-ci-benchmark-results/data.json.gz \
- data.json.gz --no-sign-request
- gzip -d -c data.json.gz > base.json
-
- uv run --no-project scripts/compare-benchmark-jsons.py base.json results.json \
- "GPU Compression" > comment.md
+ uv run --no-project scripts/compare-benchmark-jsons.py base.ingest.jsonl results.json \
+ "GPU Compression" --pr-records results.ingest.jsonl > comment.md
cat comment.md >> "$GITHUB_STEP_SUMMARY"
- name: Comment PR
if: github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork == false
diff --git a/.github/workflows/pr-bench-runner.yml b/.github/workflows/pr-bench-runner.yml
index 0f9347da61c..9a97c65682e 100644
--- a/.github/workflows/pr-bench-runner.yml
+++ b/.github/workflows/pr-bench-runner.yml
@@ -189,7 +189,7 @@ jobs:
FLAT_LAYOUT_INLINE_ARRAY_NODE: "1"
AWS_REGION: "us-east-1"
run: |
- python3 scripts/random-access-split.py \
+ python3 scripts/random-access-split.py --emit-ingest-records \
${{ inputs.remote_data_dir != '' && format('--remote-data-dir {0}', inputs.remote_data_dir) || '' }}
- name: Run ${{ inputs.benchmark_name }} benchmark (per-dataset)
@@ -199,7 +199,7 @@ jobs:
RUST_BACKTRACE: full
FLAT_LAYOUT_INLINE_ARRAY_NODE: "1"
run: |
- python3 scripts/compress-split.py
+ python3 scripts/compress-split.py --emit-ingest-records
- name: Run ${{ inputs.benchmark_name }} benchmark (per-column-encoder)
if: inputs.benchmark_id == 'string-bench'
@@ -210,16 +210,21 @@ jobs:
run: |
python3 scripts/string-split.py
+ - name: Fetch baseline from the benchmarks database
+ uses: ./.github/actions/fetch-bench-baseline
+ with:
+ role-arn: ${{ vars.GH_BENCH_READ_ROLE_ARN }}
+ region: ${{ vars.RDS_BENCH_REGION }}
+ endpoint: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }}
+ database: ${{ vars.RDS_BENCH_DB_NAME }}
+
- name: Compare results
shell: bash
run: |
set -Eeu -o pipefail -x
- python3 scripts/s3-download.py s3://vortex-ci-benchmark-results/data.json.gz data.json.gz --no-sign-request
- gzip -d -c data.json.gz > base.json
-
- uv run --no-project scripts/compare-benchmark-jsons.py base.json results.json "${{ inputs.benchmark_name }}" \
- > comment.md
+ uv run --no-project scripts/compare-benchmark-jsons.py base.ingest.jsonl results.json \
+ "${{ inputs.benchmark_name }}" --pr-records results.ingest.jsonl > comment.md
cat comment.md >> "$GITHUB_STEP_SUMMARY"
- name: Comment PR
diff --git a/.github/workflows/sql-bench-matrix.yml b/.github/workflows/sql-bench-matrix.yml
index ab6c60fc660..f1661284475 100644
--- a/.github/workflows/sql-bench-matrix.yml
+++ b/.github/workflows/sql-bench-matrix.yml
@@ -242,17 +242,23 @@ jobs:
-o sizes.json
cat sizes.json >> results.json
+ - name: Fetch baseline from the benchmarks database
+ if: inputs.mode == 'pr'
+ uses: ./.github/actions/fetch-bench-baseline
+ with:
+ role-arn: ${{ vars.GH_BENCH_READ_ROLE_ARN }}
+ region: ${{ vars.RDS_BENCH_REGION }}
+ endpoint: ${{ vars.RDS_BENCH_INSTANCE_ENDPOINT }}
+ database: ${{ vars.RDS_BENCH_DB_NAME }}
+
- name: Compare results
if: inputs.mode == 'pr'
shell: bash
run: |
set -Eeu -o pipefail -x
- python3 scripts/s3-download.py s3://vortex-ci-benchmark-results/data.json.gz data.json.gz --no-sign-request
- gzip -d -c data.json.gz > base.json
-
- uv run --no-project scripts/compare-benchmark-jsons.py base.json results.json "${{ matrix.name }}" \
- > comment.md
+ uv run --no-project scripts/compare-benchmark-jsons.py base.ingest.jsonl results.json \
+ "${{ matrix.name }}" --pr-records results.ingest.jsonl > comment.md
cat comment.md >> "$GITHUB_STEP_SUMMARY"
- name: Comment PR
@@ -314,6 +320,7 @@ jobs:
scripts/post-ingest.py results.ingest.jsonl \
--postgres "${DSN}" \
--commit-sha "${{ github.sha }}" \
+ --branch "${{ github.ref_name }}" \
--region "${AWS_REGION}"
- name: Alert incident.io
diff --git a/scripts/compare-benchmark-jsons.py b/scripts/compare-benchmark-jsons.py
index 4e2fe966b2f..a519be616f5 100644
--- a/scripts/compare-benchmark-jsons.py
+++ b/scripts/compare-benchmark-jsons.py
@@ -11,11 +11,10 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright the Vortex contributors
+import argparse
import math
import os
import re
-import subprocess
-import sys
from collections.abc import Callable
from dataclasses import dataclass
from io import StringIO
@@ -165,137 +164,77 @@ def extract_target_fields(name: str, target: object = None) -> pd.Series:
return pd.Series({"engine": engine, "file_format": file_format, "query": query})
-def benchmark_identity(row: object) -> tuple[Any, Any, Any, str, str, Any] | None:
- """Return the measurement identity used to find a matching baseline."""
+def ingest_dataset_label(record: dict[str, Any]) -> str:
+ """Join a record's dataset and variant into one label, skipping an absent variant."""
- if row.get("metric") == FILE_SIZE_METRIC or isinstance(row.get("file_size"), dict):
- return None
-
- name = row.get("name")
- if name is None:
- return None
-
- engine, file_format, _query = comparison_target(name, row.get("target"))
- return (
- identity_value(name),
- identity_value(row.get("storage")),
- dataset_key(row.get("dataset")),
- engine,
- file_format,
- identity_value(row.get("unit")),
- )
+ label = str(record["dataset"])
+ if record.get("dataset_variant") is not None:
+ label += f"/{record['dataset_variant']}"
+ return label
-def benchmark_identity_rows(df: pd.DataFrame) -> pd.DataFrame:
- """Return timing rows with the identity used to match a PR benchmark."""
+def ingest_record_row(record: dict[str, Any]) -> dict[str, Any] | None:
+ """Convert one `--ingest-jsonl` record into a comparison row.
- _file_size_rows, timing_rows = split_file_size_rows(df)
- if timing_rows.empty or "name" not in timing_rows.columns:
- return pd.DataFrame(columns=["commit_id", "benchmark_identity"])
-
- timing_rows = timing_rows.copy()
- if "commit_id" not in timing_rows.columns:
- timing_rows["commit_id"] = pd.NA
-
- timing_rows["benchmark_identity"] = [benchmark_identity(row) for _, row in timing_rows.iterrows()]
-
- return timing_rows[["commit_id", "benchmark_identity"]]
-
-
-def read_jsonl_rows_for_commit(path: str, commit_id: str) -> pd.DataFrame:
- """Read the latest copy of each row matching a history commit.
-
- Re-running a develop workflow appends a second result block for the same
- commit. Keeping the last copy of each logical row prevents a many-to-one
- merge from weighting that baseline multiple times.
+ The names keep the shapes `comparison_target` parses: `_qNN/engine:format` for SQL queries
+ and `random-access/.../ext-variant` for random access, so both sides of the comparison get
+ the same engine, format, and query attribution.
"""
- rows_by_identity: dict[tuple[Any, ...], dict[str, Any]] = {}
- with open(path, encoding="utf-8") as lines:
- for line in lines:
- if '"commit_id"' not in line or f'"{commit_id}"' not in line:
- continue
- record = orjson.loads(line)
- if record.get("commit_id") != commit_id:
- continue
-
- file_size = record.get("file_size")
- if isinstance(file_size, dict):
- identity = (
- FILE_SIZE_METRIC,
- file_size.get("benchmark"),
- file_size.get("scale_factor"),
- file_size.get("format"),
- file_size.get("file"),
- )
- else:
- identity = ("timing", benchmark_identity(record))
- rows_by_identity[identity] = record
- return pd.DataFrame(rows_by_identity.values())
-
-
-def git_tree_commit_ids() -> set[str]:
- """Return every commit reachable from the checked-out branch head.
+ kind = record.get("kind")
+ file_format = record.get("format")
+ row: dict[str, Any] = {"commit_id": record.get("commit_sha"), "storage": None, "unit": "ns"}
+ if kind == "query_measurement":
+ engine = record["engine"]
+ row["name"] = f"{record['dataset']}_q{record['query_idx']:02d}/{engine}:{file_format}"
+ row["storage"] = record["storage"]
+ # Like the results JSON, the scale factor and variant stay out of the name and in the
+ # dataset join key.
+ dims = {dim: record.get(dim) for dim in ("dataset_variant", "scale_factor") if record.get(dim) is not None}
+ if dims:
+ row["dataset"] = dims
+ row["target"] = {"engine": engine, "format": file_format}
+ elif kind == "random_access_time":
+ ext = "vortex" if str(file_format).startswith("vortex") else file_format
+ row["name"] = f"random-access/{record['dataset']}/{ext}-{record['open_mode']}"
+ row["target"] = {"engine": "vortex", "format": file_format}
+ elif kind == "compression_time":
+ op = {"encode": "compress", "decode": "decompress"}.get(record["op"], record["op"])
+ row["name"] = f"{op} time/{ingest_dataset_label(record)}"
+ row["target"] = {"engine": "vortex", "format": file_format}
+ elif kind == "compression_size":
+ row["name"] = f"{file_format} size/{ingest_dataset_label(record)}"
+ row["target"] = {"engine": "vortex", "format": file_format}
+ row["unit"] = "bytes"
+ row["value"] = record["value_bytes"]
+ return row
+ else:
+ return None
+ row["value"] = record["value_ns"]
+ row["all_runtimes"] = record["all_runtimes_ns"]
+ return row
- A shallow checkout reaches only its own tip, so every recorded baseline
- commit looks unreachable and the report silently degrades into "no
- baseline is available for this benchmark yet" — indistinguishable from a
- genuinely new benchmark, and wrong. Refuse the guess and say what to fix.
- """
- is_shallow = subprocess.run(
- ["git", "rev-parse", "--is-shallow-repository"],
- check=True,
- capture_output=True,
- text=True,
- ).stdout.strip()
- if is_shallow == "true":
- raise SystemExit(
- "Cannot resolve a baseline from a shallow checkout: the baseline is the newest "
- "recorded commit reachable from HEAD, and a shallow clone reaches only its own "
- "tip. Check the repository out with `fetch-depth: 0`."
- )
+def read_benchmark_rows(path: str) -> pd.DataFrame:
+ """Read comparison rows from a JSONL file; a missing file reads as no rows.
- commits = subprocess.run(
- ["git", "rev-list", "HEAD"],
- check=True,
- capture_output=True,
- text=True,
- ).stdout
- return set(commits.splitlines())
-
-
-def read_latest_baseline_rows(
- path: str,
- pr: pd.DataFrame,
- reachable_commit_ids: set[str],
-) -> pd.DataFrame:
- """Read rows from the latest reachable commit matching the PR benchmark.
-
- A benchmark can be new to the PR workflow and therefore have no baseline
- yet. Return an empty frame with the PR schema in that case so the report
- can show the measurements without comparison.
+ Lines carrying a `kind` are `--ingest-jsonl` records and are converted; any other line is
+ already a results-JSON row.
"""
- pr_identities = set(benchmark_identity_rows(pr)["benchmark_identity"])
- if not pr_identities:
- return pd.read_json(path, lines=True)
-
- baseline_commit_id = None
- with open(path, encoding="utf-8") as lines:
- for line in lines:
- if '"name"' not in line or '"commit_id"' not in line:
- continue
- record = orjson.loads(line)
- if benchmark_identity(record) in pr_identities:
- commit_id = record.get("commit_id")
- if commit_id is not None and commit_id in reachable_commit_ids:
- baseline_commit_id = commit_id
-
- if baseline_commit_id is None:
- return pr.iloc[0:0].copy()
-
- return read_jsonl_rows_for_commit(path, baseline_commit_id)
+ rows = []
+ if os.path.exists(path):
+ with open(path, encoding="utf-8") as lines:
+ for line in lines:
+ if not line.strip():
+ continue
+ record = orjson.loads(line)
+ row = ingest_record_row(record) if "kind" in record else record
+ if row is not None:
+ rows.append(row)
+ if not rows:
+ return pd.DataFrame(columns=["commit_id", "name", "storage", "unit", "value", "all_runtimes", "target"])
+ return pd.DataFrame(rows)
def normalize_measurement_rows(df: pd.DataFrame) -> pd.DataFrame:
@@ -1126,11 +1065,27 @@ def group_sort_key(group_key: tuple[str, str, str]) -> tuple[int, int, int, str,
def main() -> None:
"""Render the benchmark comparison markdown used in CI PR comments."""
- benchmark_name = sys.argv[3] if len(sys.argv) > 3 else ""
-
- pr = pd.read_json(sys.argv[2], lines=True)
- title = format_title(benchmark_name, pr)
- base = read_latest_baseline_rows(sys.argv[1], pr, git_tree_commit_ids())
+ parser = argparse.ArgumentParser(description=main.__doc__)
+ parser.add_argument("base_records", help="Baseline records written by fetch-benchmark-baseline.py.")
+ parser.add_argument("pr_results", help="The PR run's results JSON, for file sizes and suite metadata.")
+ parser.add_argument("benchmark_name", nargs="?", default="")
+ parser.add_argument(
+ "--pr-records",
+ help="The PR run's --ingest-jsonl records. Without them, PR rows come from the results JSON "
+ "and have no baseline.",
+ )
+ args = parser.parse_args()
+ benchmark_name = args.benchmark_name
+
+ pr_results = pd.read_json(args.pr_results, lines=True)
+ title = format_title(benchmark_name, pr_results)
+ pr_file_sizes, pr = split_file_size_rows(pr_results)
+ if args.pr_records:
+ pr_records = read_benchmark_rows(args.pr_records)
+ if not pr_records.empty:
+ pr = pr_records
+ base_fetched = os.path.exists(args.base_records)
+ base_file_sizes, base = split_file_size_rows(read_benchmark_rows(args.base_records))
base_commit_ids = set(base["commit_id"].unique())
pr_commit_id = set(pr["commit_id"].unique())
@@ -1139,9 +1094,6 @@ def main() -> None:
base_commit_id = next(iter(base_commit_ids), None)
pr_commit_id = next(iter(pr_commit_id))
- base_file_sizes, base = split_file_size_rows(base)
- pr_file_sizes, pr = split_file_size_rows(pr)
-
base = normalize_measurement_rows(base)
pr = normalize_measurement_rows(pr)
@@ -1213,7 +1165,13 @@ def main() -> None:
if summary_fields:
print("
".join(summary_fields))
print("")
- if base_commit_id is None:
+ if not base_fetched:
+ print(
+ "_The baseline could not be read from the benchmarks database; "
+ "PR measurements are shown without comparison._"
+ )
+ print("")
+ elif base_commit_id is None:
print("_No baseline is available for this benchmark yet; PR measurements are shown without comparison._")
print("")
if verdict is not None or engine_summary is not None:
@@ -1287,7 +1245,8 @@ def delta_cell(
print("")
print("")
- file_size_report = format_file_size_report(base_file_sizes, pr_file_sizes)
+ # File sizes are not stored in the benchmarks database, so they have no baseline to compare.
+ file_size_report = "" if base_file_sizes.empty else format_file_size_report(base_file_sizes, pr_file_sizes)
if file_size_report:
print("")
print("---")
diff --git a/scripts/fetch-benchmark-baseline.py b/scripts/fetch-benchmark-baseline.py
new file mode 100644
index 00000000000..613a80459f0
--- /dev/null
+++ b/scripts/fetch-benchmark-baseline.py
@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+# /// script
+# requires-python = ">=3.11"
+# dependencies = ["psycopg[binary]", "boto3"]
+# ///
+
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+"""Fetch the develop baseline for a PR benchmark run from the benchmarks Postgres.
+
+Reads the PR's `--ingest-jsonl` records, finds the newest commit reachable from
+`HEAD` that recorded any of the same measurements, and writes that commit's
+matching rows in the same JSONL record shape. `compare-benchmark-jsons.py` then
+compares the two record files.
+
+It connects as the read-only `bench_ci_read` Postgres user, with an RDS IAM token minted
+(by `post-ingest.py`'s token helper) from the AWS credentials of the assumed
+`GH_BENCH_READ_ROLE_ARN` role, over verify-full TLS. The session is read-only.
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import subprocess
+from pathlib import Path
+from types import ModuleType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from psycopg import Connection
+
+# Per record kind: the fact table and the natural-key columns other than
+# `commit_sha`. Rows from different commits match when these columns are equal.
+# `None` compares equal to `None`, as the ingest path stores absent dims as NULL.
+_KINDS: dict[str, tuple[str, tuple[str, ...], tuple[str, ...]]] = {
+ "query_measurement": (
+ "query_measurements",
+ ("dataset", "dataset_variant", "scale_factor", "query_idx", "storage", "engine", "format"),
+ ("value_ns", "all_runtimes_ns"),
+ ),
+ "compression_time": (
+ "compression_times",
+ ("dataset", "dataset_variant", "format", "op"),
+ ("value_ns", "all_runtimes_ns"),
+ ),
+ "compression_size": (
+ "compression_sizes",
+ ("dataset", "dataset_variant", "format"),
+ ("value_bytes", "uncompressed_bytes"),
+ ),
+ "random_access_time": (
+ "random_access_times",
+ ("dataset", "format", "open_mode"),
+ ("value_ns", "all_runtimes_ns"),
+ ),
+}
+
+# Rows read per query while looking for an identity's newest reachable commit.
+_PAGE_SIZE = 200
+
+
+def parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+ parser.add_argument("pr_records", type=Path, help="The PR run's --ingest-jsonl output.")
+ parser.add_argument("output", type=Path, help="Where to write the baseline records.")
+ parser.add_argument(
+ "--postgres",
+ metavar="DSN",
+ required=True,
+ help="Libpq DSN of the benchmarks database. Without a password, an RDS IAM token is minted.",
+ )
+ parser.add_argument("--region", default=None, help="AWS region for RDS IAM token minting.")
+ return parser.parse_args()
+
+
+def read_records(path: Path) -> list[dict]:
+ if not path.exists():
+ return []
+ with path.open(encoding="utf-8") as lines:
+ return [json.loads(line) for line in lines if line.strip()]
+
+
+def identities_by_kind(records: list[dict]) -> dict[str, list[tuple]]:
+ """Return the distinct natural keys of the PR records, grouped by kind."""
+
+ identities: dict[str, set[tuple]] = {}
+ for record in records:
+ kind = record.get("kind")
+ if kind not in _KINDS:
+ continue
+ _table, dims, _values = _KINDS[kind]
+ identities.setdefault(kind, set()).add(tuple(record.get(dim) for dim in dims))
+ return {kind: sorted(keys, key=repr) for kind, keys in identities.items()}
+
+
+def reachable_commits() -> list[str]:
+ """Return every commit reachable from `HEAD`; a shallow checkout cannot pick a baseline."""
+
+ is_shallow = subprocess.run(
+ ["git", "rev-parse", "--is-shallow-repository"], check=True, capture_output=True, text=True
+ ).stdout.strip()
+ if is_shallow == "true":
+ raise SystemExit(
+ "Cannot resolve a baseline from a shallow checkout: the baseline is the newest "
+ "recorded commit reachable from HEAD. Check the repository out with `fetch-depth: 0`."
+ )
+ commits = subprocess.run(["git", "rev-list", "HEAD"], check=True, capture_output=True, text=True).stdout
+ return commits.split()
+
+
+def identity_condition(kind: str, identity: tuple) -> tuple[str, list]:
+ """Return a `WHERE` condition on `t` matching one identity, and its parameters.
+
+ Absent dims match with `IS NULL` rather than `IS NOT DISTINCT FROM`, which
+ no index can serve, so each lookup is a range scan of the table's chart or
+ summary index instead of a scan of the whole table.
+ """
+
+ _table, dims, _values = _KINDS[kind]
+ conditions = []
+ params = []
+ for dim, value in zip(dims, identity):
+ if value is None:
+ conditions.append(f"t.{dim} IS NULL")
+ else:
+ conditions.append(f"t.{dim} = %s")
+ params.append(value)
+ return " AND ".join(conditions), params
+
+
+def newest_baseline_commit(
+ conn: Connection, identities: dict[str, list[tuple]], reachable: set[str]
+) -> tuple[str, object] | None:
+ """Return the newest reachable commit that recorded any PR identity, with its timestamp.
+
+ Each identity's rows are read newest first, a page at a time, until one is from a commit
+ reachable from `HEAD`. That is usually the first page: the baseline is the PR's merge base
+ or close to it. One query per identity, as Postgres plans a `UNION ALL` of them far worse.
+ """
+
+ newest = None
+ for kind, keys in identities.items():
+ table, _dims, _values = _KINDS[kind]
+ # `query_measurements` carries its commit's timestamp at the end of its summary index,
+ # so its rows come back newest first straight from an index range scan.
+ if table == "query_measurements":
+ source, timestamp = "query_measurements t", "t.commit_timestamp"
+ else:
+ source, timestamp = f"{table} t JOIN commits c ON c.commit_sha = t.commit_sha", "c.timestamp"
+ for identity in keys:
+ where, params = identity_condition(kind, identity)
+ before = None
+ while True:
+ page = conn.execute(
+ f"SELECT t.commit_sha, {timestamp} FROM {source}"
+ f" WHERE {where} AND {timestamp} IS NOT NULL"
+ + (f" AND {timestamp} < %s" if before is not None else "")
+ + f" ORDER BY {timestamp} DESC LIMIT {_PAGE_SIZE}",
+ [*params, before] if before is not None else params,
+ ).fetchall()
+ found = next((row for row in page if row[0] in reachable), None)
+ if found is not None or len(page) < _PAGE_SIZE:
+ break
+ before = page[-1][1]
+ if found is not None and (newest is None or (found[1], found[0]) > (newest[1], newest[0])):
+ newest = (found[0], found[1])
+ return newest
+
+
+def baseline_records(
+ conn: Connection, identities: dict[str, list[tuple]], commit_sha: str, commit_timestamp: object
+) -> list[dict]:
+ """Return the rows `commit_sha` recorded for the PR identities, as ingest records."""
+
+ records = []
+ for kind, keys in identities.items():
+ table, dims, values = _KINDS[kind]
+ columns = ("commit_sha", *dims, *values)
+ for identity in keys:
+ where, params = identity_condition(kind, identity)
+ where += " AND t.commit_sha = %s"
+ params.append(commit_sha)
+ if table == "query_measurements":
+ where += " AND t.commit_timestamp = %s"
+ params.append(commit_timestamp)
+ rows = conn.execute(f"SELECT {', '.join(f't.{c}' for c in columns)} FROM {table} t WHERE {where}", params)
+ records.extend({"kind": kind, **dict(zip(columns, row))} for row in rows.fetchall())
+ return records
+
+
+def post_ingest_module() -> ModuleType:
+ """Load `scripts/post-ingest.py`, whose RDS IAM token helper this script reuses."""
+
+ path = Path(__file__).resolve().parent / "post-ingest.py"
+ spec = importlib.util.spec_from_file_location("post_ingest", path)
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+def connect_postgres(dsn: str, region: str | None) -> Connection:
+ """Open a read-only, verify-full TLS connection, minting an RDS IAM token if the DSN has no password."""
+
+ import psycopg
+ from psycopg import conninfo
+
+ params = conninfo.conninfo_to_dict(dsn)
+ if params.setdefault("sslmode", "verify-full") != "verify-full":
+ raise SystemExit(f"--postgres requires sslmode=verify-full; DSN specified {params['sslmode']!r}.")
+ if not params.get("password"):
+ params["password"] = post_ingest_module()._rds_iam_token(
+ host=params["host"], port=int(params.get("port", 5432)), user=params["user"], region=region
+ )
+ params["connect_timeout"] = 10
+ params["options"] = "-c statement_timeout=30000"
+ conn = psycopg.connect(**params)
+ conn.read_only = True
+ return conn
+
+
+def main() -> None:
+ args = parse_args()
+ identities = identities_by_kind(read_records(args.pr_records))
+ records: list[dict] = []
+ if identities:
+ commits = reachable_commits()
+ with connect_postgres(args.postgres, args.region) as conn:
+ newest = newest_baseline_commit(conn, identities, set(commits))
+ if newest is not None:
+ commit_sha, commit_timestamp = newest
+ records = baseline_records(conn, identities, commit_sha, commit_timestamp)
+ print(f"Baseline commit {commit_sha}: {len(records)} records")
+ if not records:
+ print("No baseline found; the report shows PR measurements without comparison.")
+ with args.output.open("w", encoding="utf-8") as output:
+ for record in records:
+ output.write(json.dumps(record) + "\n")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/post-ingest.py b/scripts/post-ingest.py
index 5dbd260e068..3c4115e7b7a 100755
--- a/scripts/post-ingest.py
+++ b/scripts/post-ingest.py
@@ -74,6 +74,11 @@ def parse_args() -> argparse.Namespace:
required=True,
help="40-hex commit SHA. Usually ${{ github.sha }} in CI.",
)
+ parser.add_argument(
+ "--branch",
+ required=True,
+ help="Branch the results were measured on, stored as commits.branch. Usually ${{ github.ref_name }} in CI.",
+ )
parser.add_argument(
"--repo-url",
default="https://github.com/vortex-data/vortex",
@@ -125,10 +130,12 @@ def git_show_field(sha: str, fmt: str, cwd: Path | None) -> str:
return result.stdout.strip()
-def build_commit(sha: str, repo_url: str, git_dir: Path | None) -> dict:
+def build_commit(sha: str, branch: str, repo_url: str, git_dir: Path | None) -> dict:
sha = sha.strip().lower()
if len(sha) != 40 or any(c not in "0123456789abcdef" for c in sha):
raise SystemExit(f"commit SHA must be 40-hex lowercase, got: {sha!r}")
+ if not branch:
+ raise SystemExit("--branch must not be empty")
timestamp = git_show_field(sha, "%cI", git_dir)
message = git_show_field(sha, "%s", git_dir)
@@ -140,6 +147,7 @@ def build_commit(sha: str, repo_url: str, git_dir: Path | None) -> dict:
return {
"sha": sha,
+ "branch": branch,
"timestamp": timestamp,
"message": message,
"author_name": author_name,
@@ -739,10 +747,11 @@ def _upsert_commit(conn, commit: dict) -> None:
conn.execute(
"""
INSERT INTO commits (
- commit_sha, timestamp, message, author_name, author_email,
+ commit_sha, branch, timestamp, message, author_name, author_email,
committer_name, committer_email, tree_sha, url
- ) VALUES (%s, %s::timestamptz, %s, %s, %s, %s, %s, %s, %s)
+ ) VALUES (%s, %s, %s::timestamptz, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (commit_sha) DO UPDATE SET
+ branch = excluded.branch,
timestamp = excluded.timestamp,
message = excluded.message,
author_name = excluded.author_name,
@@ -754,6 +763,7 @@ def _upsert_commit(conn, commit: dict) -> None:
""",
(
commit["sha"],
+ commit["branch"],
commit["timestamp"],
commit["message"],
commit["author_name"],
@@ -981,7 +991,7 @@ def _main_postgres(args: argparse.Namespace) -> int:
records = read_records(args.jsonl_path)
# `build_commit` runs `git show `, so the SHA must be in the runner's local git
# history. A shallow checkout missing the SHA fails before opening a database connection.
- commit = build_commit(args.commit_sha, args.repo_url, args.git_dir)
+ commit = build_commit(args.commit_sha, args.branch, args.repo_url, args.git_dir)
conn = connect_postgres(args.postgres, args.region)
try:
inserted, updated = ingest_postgres(conn, commit, records)
diff --git a/scripts/tests/test_benchmark_reporting.py b/scripts/tests/test_benchmark_reporting.py
index 87a8f1370e9..0b55fe1e3ca 100644
--- a/scripts/tests/test_benchmark_reporting.py
+++ b/scripts/tests/test_benchmark_reporting.py
@@ -15,6 +15,16 @@
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"
+FETCH_BASELINE_SCRIPT = REPO_ROOT / "scripts" / "fetch-benchmark-baseline.py"
+
+
+def load_module(name: str, path: Path) -> ModuleType:
+ spec = importlib.util.spec_from_file_location(name, path)
+ 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 load_compare_module() -> ModuleType:
@@ -100,6 +110,34 @@ def stored_timing_row(
return row
+def query_record(commit: str, query_idx: int, engine: str, file_format: str, value: int) -> dict[str, object]:
+ return {
+ "kind": "query_measurement",
+ "commit_sha": commit,
+ "dataset": "tpch",
+ "dataset_variant": None,
+ "scale_factor": "1",
+ "query_idx": query_idx,
+ "storage": "nvme",
+ "engine": engine,
+ "format": file_format,
+ "value_ns": value,
+ "all_runtimes_ns": [value, value, value],
+ }
+
+
+def random_access_record(commit: str, file_format: str, open_mode: str, value: int) -> dict[str, object]:
+ return {
+ "kind": "random_access_time",
+ "commit_sha": commit,
+ "dataset": "taxi/uniform",
+ "format": file_format,
+ "open_mode": open_mode,
+ "value_ns": value,
+ "all_runtimes_ns": [value, value, value],
+ }
+
+
def stored_custom_row(
commit: str,
name: str,
@@ -139,28 +177,19 @@ def render_report(
base_rows: list[dict[str, object]],
pr_rows: list[dict[str, object]],
benchmark_name: str,
+ pr_records: list[dict[str, object]] | None = None,
) -> str:
- repo = git_repo_with_one_commit(tmp_path / "repo")
- head_commit = subprocess.run(
- ["git", "rev-parse", "HEAD"],
- cwd=repo,
- check=True,
- capture_output=True,
- text=True,
- ).stdout.strip()
- base_rows = [row | {"commit_id": head_commit} for row in base_rows]
base_path = tmp_path / "base.jsonl"
pr_path = tmp_path / "pr.jsonl"
base_path.write_text("".join(f"{json.dumps(row)}\n" for row in base_rows), encoding="utf-8")
pr_path.write_text("".join(f"{json.dumps(row)}\n" for row in pr_rows), encoding="utf-8")
+ command = [sys.executable, str(COMPARE_SCRIPT), str(base_path), str(pr_path), benchmark_name]
+ if pr_records is not None:
+ records_path = tmp_path / "pr.ingest.jsonl"
+ records_path.write_text("".join(f"{json.dumps(row)}\n" for row in pr_records), encoding="utf-8")
+ command += ["--pr-records", str(records_path)]
- result = subprocess.run(
- [sys.executable, str(COMPARE_SCRIPT), str(base_path), str(pr_path), benchmark_name],
- cwd=repo,
- check=False,
- capture_output=True,
- text=True,
- )
+ result = subprocess.run(command, check=False, capture_output=True, text=True)
assert result.returncode == 0, result.stderr
return result.stdout
@@ -171,139 +200,6 @@ def markdown_row(report: str, name: str) -> list[str]:
return [cell.strip() for cell in line.strip("|").split("|")]
-def test_read_latest_baseline_rows_streams_latest_matching_benchmark_commit(tmp_path: Path) -> None:
- compare = load_compare_module()
- history_path = tmp_path / "history.jsonl"
- history_rows = [
- stored_timing_row(
- "base-old",
- "tpch_q01/datafusion:parquet",
- 100,
- "nvme",
- {"scale_factor": "1.0"},
- ),
- file_size_record_for("base-old", 100, "tpch", "1.0", "vortex-file-compressed", "part-0.vortex"),
- stored_timing_row(
- "base-current",
- "tpch_q01/datafusion:parquet",
- 110,
- "nvme",
- {"scale_factor": "1.0"},
- ),
- file_size_record_for("base-current", 120, "tpch", "1.0", "vortex-file-compressed", "part-0.vortex"),
- stored_timing_row(
- "base-wrong-unit",
- "tpch_q01/datafusion:parquet",
- 111,
- "nvme",
- {"scale_factor": "1.0"},
- )
- | {"unit": "ms"},
- stored_timing_row(
- "base-wrong-target",
- "tpch_q01/datafusion:parquet",
- 112,
- "nvme",
- {"scale_factor": "1.0"},
- file_format="vortex",
- ),
- stored_timing_row("base-other", "clickbench_q01/datafusion:parquet", 200, "nvme"),
- ]
- history_path.write_text(
- "".join(f"{json.dumps(row)}\n" for row in history_rows),
- encoding="utf-8",
- )
- pr = pd.DataFrame(
- [
- stored_timing_row(
- "pr-sha",
- "tpch_q01/datafusion:parquet",
- 115,
- "nvme",
- {"scale_factor": "1.0"},
- ),
- ]
- )
-
- selected = compare.read_latest_baseline_rows(
- history_path,
- pr,
- {"base-old", "base-current", "base-wrong-unit", "base-wrong-target", "base-other"},
- )
-
- pd.testing.assert_frame_equal(selected, pd.DataFrame(history_rows[2:4]))
-
-
-def test_read_latest_baseline_rows_skips_commits_outside_git_tree(tmp_path: Path) -> None:
- compare = load_compare_module()
- history_path = tmp_path / "history.jsonl"
- history_rows = [
- stored_timing_row("base-reachable", "tpch_q01/datafusion:parquet", 100),
- file_size_record_for(
- "base-reachable",
- 100,
- "tpch",
- "1.0",
- "vortex-file-compressed",
- "part-0.vortex",
- ),
- stored_timing_row("base-off-tree", "tpch_q01/datafusion:parquet", 110),
- file_size_record_for(
- "base-off-tree",
- 120,
- "tpch",
- "1.0",
- "vortex-file-compressed",
- "part-0.vortex",
- ),
- ]
- history_path.write_text(
- "".join(f"{json.dumps(row)}\n" for row in history_rows),
- encoding="utf-8",
- )
- pr = pd.DataFrame([stored_timing_row("pr", "tpch_q01/datafusion:parquet", 105)])
-
- selected = compare.read_latest_baseline_rows(history_path, pr, {"base-reachable"})
-
- pd.testing.assert_frame_equal(selected, pd.DataFrame(history_rows[:2]))
-
-
-def test_read_latest_baseline_rows_returns_empty_without_reachable_match(tmp_path: Path) -> None:
- compare = load_compare_module()
- history_path = tmp_path / "history.jsonl"
- history_path.write_text(
- f"{json.dumps(stored_timing_row('base-off-tree', 'tpch_q01/datafusion:parquet', 100))}\n",
- encoding="utf-8",
- )
- pr = pd.DataFrame([stored_timing_row("pr", "tpch_q01/datafusion:parquet", 105)])
-
- selected = compare.read_latest_baseline_rows(history_path, pr, set())
-
- assert selected.empty
-
-
-def test_read_latest_baseline_rows_uses_last_result_from_rerun(tmp_path: Path) -> None:
- compare = load_compare_module()
- history_path = tmp_path / "history.jsonl"
- history_path.write_text(
- "".join(
- f"{json.dumps(row)}\n"
- for row in [
- stored_timing_row("base", "tpch_q01/datafusion:parquet", 100),
- stored_timing_row("base", "tpch_q01/datafusion:parquet", 110),
- ]
- ),
- encoding="utf-8",
- )
- pr = pd.DataFrame([stored_timing_row("pr", "tpch_q01/datafusion:parquet", 105)])
-
- selected = compare.read_latest_baseline_rows(history_path, pr, {"base"})
-
- pd.testing.assert_frame_equal(
- selected, pd.DataFrame([stored_timing_row("base", "tpch_q01/datafusion:parquet", 110)])
- )
-
-
def test_within_engine_analysis_uses_each_engines_own_parquet_control() -> None:
compare = load_compare_module()
rows = [
@@ -405,10 +301,9 @@ def test_comparison_report_groups_by_target_and_unit(tmp_path: Path) -> None:
def test_comparison_report_handles_missing_benchmark_baseline(tmp_path: Path) -> None:
- base_rows = [stored_custom_row("base-sha", "other timing/fixture", "ms", 12.5)]
pr_rows = [stored_custom_row("pr-sha", "new timing/fixture", "ms", 3.125)]
- report = render_report(tmp_path, base_rows, pr_rows, "New benchmark")
+ report = render_report(tmp_path, [], pr_rows, "New benchmark")
assert "No baseline is available for this benchmark yet" in report
assert "vs base `none`" in report
@@ -731,7 +626,7 @@ def test_capture_file_sizes_emits_shared_benchmark_rows(tmp_path: Path) -> None:
]
-def test_shallow_checkout_fails_instead_of_reporting_no_baseline(tmp_path: Path) -> None:
+def test_baseline_fetch_refuses_a_shallow_checkout(tmp_path: Path) -> None:
"""A shallow clone must fail loudly, not render a baseline-free report.
`git rev-list HEAD` reaches only the tip of a shallow clone, so every
@@ -746,19 +641,11 @@ def test_shallow_checkout_fails_instead_of_reporting_no_baseline(tmp_path: Path)
check=True,
capture_output=True,
)
- base_path = tmp_path / "base.jsonl"
- pr_path = tmp_path / "pr.jsonl"
- base_path.write_text(
- f"{json.dumps(stored_timing_row('recorded', 'tpch_q01/datafusion:parquet', 100))}\n",
- encoding="utf-8",
- )
- pr_path.write_text(
- f"{json.dumps(stored_timing_row('pr', 'tpch_q01/datafusion:parquet', 105))}\n",
- encoding="utf-8",
- )
+ pr_path = tmp_path / "pr.ingest.jsonl"
+ pr_path.write_text(f"{json.dumps(query_record('pr', 1, 'datafusion', 'parquet', 105))}\n", encoding="utf-8")
result = subprocess.run(
- [sys.executable, str(COMPARE_SCRIPT), str(base_path), str(pr_path), "TPC-H"],
+ [sys.executable, str(FETCH_BASELINE_SCRIPT), str(pr_path), str(tmp_path / "base.jsonl"), "--postgres", ""],
cwd=shallow,
check=False,
capture_output=True,
@@ -768,7 +655,122 @@ def test_shallow_checkout_fails_instead_of_reporting_no_baseline(tmp_path: Path)
assert result.returncode != 0
assert "shallow checkout" in result.stderr
assert "fetch-depth: 0" in result.stderr
- assert "No baseline is available" not in result.stdout
+
+
+def test_baseline_fetch_groups_distinct_pr_identities_by_kind() -> None:
+ fetch = load_module("fetch_benchmark_baseline", FETCH_BASELINE_SCRIPT)
+ records = [
+ query_record("pr", 1, "datafusion", "parquet", 100),
+ query_record("pr", 1, "datafusion", "parquet", 100),
+ random_access_record("pr", "parquet", "cached", 100),
+ {"kind": "vector_search_run", "dataset": "cohere"},
+ ]
+
+ assert fetch.identities_by_kind(records) == {
+ "query_measurement": [("tpch", None, "1", 1, "nvme", "datafusion", "parquet")],
+ "random_access_time": [("taxi/uniform", "parquet", "cached")],
+ }
+
+
+def test_baseline_fetch_matches_absent_dims_as_equal() -> None:
+ fetch = load_module("fetch_benchmark_baseline", FETCH_BASELINE_SCRIPT)
+
+ absent = fetch.identity_condition("compression_size", ("taxi", None, "parquet"))
+ present = fetch.identity_condition("compression_size", ("taxi", "gpu", "lance"))
+
+ assert absent == ("t.dataset = %s AND t.dataset_variant IS NULL AND t.format = %s", ["taxi", "parquet"])
+ assert present == (
+ "t.dataset = %s AND t.dataset_variant = %s AND t.format = %s",
+ ["taxi", "gpu", "lance"],
+ )
+
+
+def test_report_compares_ingest_records(tmp_path: Path) -> None:
+ targets = [("parquet", 100, 105), ("vortex-file-compressed", 80, 70)]
+ base_records = [query_record("base-sha", 1, "datafusion", target, base) for target, base, _pr in targets]
+ base_records += [random_access_record("base-sha", target, "cached", base) for target, base, _pr in targets]
+ pr_records = [query_record("pr-sha", 1, "datafusion", target, pr) for target, _base, pr in targets]
+ pr_records += [random_access_record("pr-sha", target, "cached", pr) for target, _base, pr in targets]
+ pr_records.append(random_access_record("pr-sha", "lance", "cached", 300))
+ pr_results = [
+ stored_timing_row("pr-sha", "ignored when records exist", 1) | {"doc": "benchmarks/tpch.md"},
+ file_size_record("pr-sha", 125),
+ ]
+
+ report = render_report(tmp_path, base_records, pr_results, "TPC-H", pr_records)
+
+ assert "vs base `base-sha`" in report
+ assert "**Verdict**:" in report
+ assert "**Engines**: DataFusion " in report
+ assert "random-access " in report
+ assert "ignored when records exist" not in report
+ assert markdown_row(report, "tpch_q01/datafusion:vortex-file-compressed")[1] == "70 / 80 / -12.5% 🟢"
+ assert markdown_row(report, "random-access/taxi/uniform/parquet-cached")[1] == "105 / 100 / +5.0%"
+ assert markdown_row(report, "random-access/taxi/uniform/lance-cached")[1] == "300 / — / no baseline"
+ assert "random-access / lance / ns " in report
+ assert "File Size" not in report
+
+
+def test_report_says_when_the_baseline_could_not_be_read(tmp_path: Path) -> None:
+ pr_records = [query_record("pr-sha", 1, "datafusion", "parquet", 100)]
+ records_path = tmp_path / "pr.ingest.jsonl"
+ records_path.write_text(f"{json.dumps(pr_records[0])}\n", encoding="utf-8")
+ results_path = tmp_path / "results.json"
+ results_path.write_text(f"{json.dumps(stored_timing_row('pr-sha', 'unused', 1))}\n", encoding="utf-8")
+
+ result = subprocess.run(
+ [
+ sys.executable,
+ str(COMPARE_SCRIPT),
+ str(tmp_path / "missing.jsonl"),
+ str(results_path),
+ "TPC-H",
+ "--pr-records",
+ str(records_path),
+ ],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+
+ assert "The baseline could not be read from the benchmarks database" in result.stdout
+ assert "No baseline is available for this benchmark yet" not in result.stdout
+
+
+def test_query_records_keep_scale_factors_apart() -> None:
+ compare = load_compare_module()
+ sf1 = compare.ingest_record_row(query_record("sha", 1, "duckdb", "parquet", 5))
+ sf10 = compare.ingest_record_row(query_record("sha", 1, "duckdb", "parquet", 5) | {"scale_factor": "10"})
+
+ assert sf1 is not None and sf10 is not None
+ assert sf1["name"] == sf10["name"] == "tpch_q01/duckdb:parquet"
+ assert compare.dataset_key(sf1["dataset"]) != compare.dataset_key(sf10["dataset"])
+
+
+def test_report_converts_compression_records() -> None:
+ compare = load_compare_module()
+ common = {"commit_sha": "sha", "dataset": "taxi", "dataset_variant": None, "format": "parquet"}
+
+ encode = compare.ingest_record_row(
+ common | {"kind": "compression_time", "op": "encode", "value_ns": 5, "all_runtimes_ns": [6, 5]}
+ )
+ decode = compare.ingest_record_row(
+ common
+ | {
+ "kind": "compression_time",
+ "dataset_variant": "gpu",
+ "op": "decode",
+ "value_ns": 5,
+ "all_runtimes_ns": [6, 5],
+ }
+ )
+ size = compare.ingest_record_row(common | {"kind": "compression_size", "value_bytes": 10, "uncompressed_bytes": 20})
+
+ assert encode is not None and decode is not None and size is not None
+ assert encode["name"] == "compress time/taxi"
+ assert decode["name"] == "decompress time/taxi/gpu"
+ assert (size["name"], size["unit"], size["value"]) == ("parquet size/taxi", "bytes", 10)
+ assert compare.ingest_record_row({"kind": "vector_search_run"}) is None
def test_summary_reports_hot_alone_when_no_row_recorded_its_runs(tmp_path: Path) -> None:
diff --git a/scripts/tests/test_post_ingest.py b/scripts/tests/test_post_ingest.py
index 05d2fbf6990..203f606e38a 100644
--- a/scripts/tests/test_post_ingest.py
+++ b/scripts/tests/test_post_ingest.py
@@ -28,6 +28,7 @@ def writer() -> ModuleType:
def commit() -> dict[str, str]:
return {
"sha": "a" * 40,
+ "branch": "develop",
"timestamp": "2026-09-08T12:00:00Z",
"message": "benchmark",
"author_name": "Author",