From 6030b8a1caeb423225a9675947e1bb748d2dbe9d Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 21:43:13 +0800 Subject: [PATCH 01/11] ci: compare PR and base vector index benchmarks --- .github/workflows/benchmark-pr.yml | 111 ++++++++++ tools/README.md | 50 +++++ tools/benchmark_pr.py | 321 +++++++++++++++++++++++++++++ tools/tests/test_benchmark_pr.py | 107 ++++++++++ 4 files changed, 589 insertions(+) create mode 100644 .github/workflows/benchmark-pr.yml create mode 100644 tools/benchmark_pr.py create mode 100644 tools/tests/test_benchmark_pr.py diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml new file mode 100644 index 00000000..d9c6ed45 --- /dev/null +++ b/.github/workflows/benchmark-pr.yml @@ -0,0 +1,111 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +name: PR benchmark + +on: + pull_request: + branches: [main, 'release-*'] + paths: + - 'core/**' + - 'Cargo.toml' + - 'Cargo.lock' + - '.cargo/**' + - 'tools/benchmark_pr.py' + - 'tools/tests/test_benchmark_pr.py' + - '.github/workflows/benchmark-pr.yml' + +permissions: + contents: read + +concurrency: + group: benchmark-pr-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + compare: + name: Compare base and PR + runs-on: ubuntu-24.04 + timeout-minutes: 25 + steps: + - name: Check out PR merge result + uses: actions/checkout@v4 + with: + ref: ${{ github.sha }} + path: candidate + persist-credentials: false + + - name: Check out exact base revision + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.base.sha }} + path: base + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Set up fixed Rust toolchain + run: | + rustup toolchain install 1.94.1 --profile minimal + rustup default 1.94.1 + env: + RUSTUP_TOOLCHAIN: 1.94.1 + + - name: Cache Rust downloads + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + key: benchmark-${{ runner.os }}-rust-1.94.1-${{ hashFiles('candidate/Cargo.lock', 'base/Cargo.lock') }} + restore-keys: | + benchmark-${{ runner.os }}-rust-1.94.1- + + - name: Test comparison and failure handling + run: python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v + + - name: Build and compare both revisions + run: python3 candidate/tools/benchmark_pr.py --base base --candidate candidate --output results + env: + RUSTUP_TOOLCHAIN: 1.94.1 + + - name: Publish comparison summary + if: always() + run: | + if [ -f results/summary.md ]; then + cat results/summary.md >> "$GITHUB_STEP_SUMMARY" + else + echo 'Benchmark did not produce a report. Check the setup and build steps.' >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Upload report and raw samples + if: always() + uses: actions/upload-artifact@v4 + with: + name: pr-${{ github.event.pull_request.number }}-benchmark-${{ github.run_attempt }} + path: | + results/summary.md + results/summary.json + results/metadata.json + results/build-*.log + results/build-*.jsonl + results/raw/ + if-no-files-found: warn + retention-days: 30 diff --git a/tools/README.md b/tools/README.md index 17cac1cf..3fb78e88 100644 --- a/tools/README.md +++ b/tools/README.md @@ -21,6 +21,56 @@ This directory contains helper scripts used by release managers and committers. +## PR / base benchmark + +The `PR benchmark` workflow compares the exact PR base SHA with GitHub's PR merge +commit on the same Ubuntu runner. It runs when the core, Cargo configuration, or +benchmark tooling changes. Open **Checks → Compare base and PR → Summary** for +the comparison; the workflow artifact includes raw CSVs, build/sample logs, +environment metadata, and machine-readable results. No PR comment or external +service is required. + +`benchmark_pr.py` builds both revisions in release mode with separate target +directories, then runs four samples per version per index. Each index is tested +in a fresh process, alternating base/candidate and candidate/base pairs. Both +versions use the candidate's `ann_bench.rs` and its support module, so a change +to the benchmark itself cannot silently change the measurement between sides. +An incompatible driver/API combination fails explicitly and needs a compatible +shared driver before results can be compared. + +The initial workload is deliberately small: 10,000 synthetic 64D vectors, 4,096 +training vectors, 2,048 queries, top-10, seed 42, and two Rayon threads. It covers +IVF-FLAT, IVF-SQ, IVF-PQ, IVF-RQ and DiskANN on local warm page cache. Sequential +queries follow reader optimization and one first query; batch queries use a +separate optimized reader. It does not measure cold storage or real object-store +performance. Recall is currently measured for batch results. Process peak RSS +is sampled through build and includes dataset/ground-truth allocations, not +search peak memory. + +The report shows medians and min/max ranges, with relative changes for timing, +throughput, I/O, memory, and size. Recall changes use percentage points. A recall +drop is highlighted alongside performance. This first version is informational: +performance/recall changes do not fail CI, but build failures, timeouts, invalid +metrics, missing samples, and mismatched workload parameters do. With only four +samples, a reported percentage is not a statistical significance claim. + +To reproduce locally, create two **disposable** checkouts, use the same Rust +toolchain as the workflow, and run from the candidate checkout: + +```bash +python3 tools/benchmark_pr.py \ + --base /path/to/disposable-base \ + --candidate /path/to/disposable-candidate \ + --output /path/to/new-results-directory +``` + +The script overwrites the two benchmark-driver files in the base checkout. The +output directory must not exist. `--rounds 2` provides a shorter smoke run; +`--timeout` sets the per-process timeout in seconds (default 180). Dataset and +index options are pinned in the script; inherited `ANN_*` settings are removed. +CI pins Rust 1.94.1 and the x86-64 CPU target, caches dependency downloads only, +and uploads reports even when a comparison fails. + ## ANN-Benchmarks dataset conversion `convert_ann_benchmarks.py` converts a dense diff --git a/tools/benchmark_pr.py b/tools/benchmark_pr.py new file mode 100644 index 00000000..6753016c --- /dev/null +++ b/tools/benchmark_pr.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Compare two disposable checkouts using the candidate's ANN benchmark driver.""" + +import argparse +import csv +import hashlib +import json +import math +import os +from pathlib import Path +import platform +import shutil +import statistics +import subprocess +import time + + +INDEXES = ("IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_RQ", "DISKANN") +WORKLOAD = { + "ANN_DATASET_NAME": "ci-clustered-v1", + "ANN_N": "10000", + "ANN_TRAIN_N": "4096", + "ANN_NQ": "2048", + "ANN_D": "64", + "ANN_K": "10", + "ANN_NLIST": "64", + "ANN_NPROBE": "8", + "ANN_PQ_M": "8", + "ANN_RQ_BITS": "4", + "ANN_CLUSTERS": "32", + "ANN_NOISE_DIMENSIONS": "64", + "ANN_SEED": "42", + "ANN_DISKANN_L_SEARCHES": "100", + "ANN_DISKANN_MEMORY_BUDGET_BYTES": str(256 * 1024 * 1024), + "ANN_DISKANN_BUILD_DISTANCE": "full_precision", + "ANN_DISKANN_RAW_VECTOR_ENCODING": "f32", + "ANN_STORAGE_CASES": "local_ssd_warm_cache", + "RAYON_NUM_THREADS": "2", +} +# These identify the workload, not measured results. Never compare unlike cases. +CASE_FIELDS = ( + "dataset", "index", "storage", "n", "train_n", "nq", "d", "k", "nlist", + "nprobe", "pq_m", "rq_bits", "diskann_build_distance", + "diskann_raw_vector_encoding", "l_search", +) +METRICS = ( + ("recall_at_10", "Batch Recall@10", "recall"), + ("sequential_qps", "Sequential QPS", "higher"), + ("batch_qps", "Batch QPS", "higher"), + ("p95_query_us", "Sequential P95 (µs)", "lower"), + ("first_query_us", "First query (µs)", "lower"), + ("sequential_pread_rounds", "Read rounds / sequential query", "lower"), + ("sequential_pread_bytes", "Read bytes / sequential query", "lower"), + ("batch_pread_rounds", "Read rounds / batch query", "lower"), + ("batch_pread_bytes", "Read bytes / batch query", "lower"), + ("build_ms", "Build (ms)", "lower"), + ("peak_rss_bytes", "Process peak RSS through build (MiB)", "lower"), + ("file_bytes", "Index size (MiB)", "lower"), +) + + +def command_output(command, cwd=None): + return subprocess.check_output(command, cwd=cwd, text=True).strip() + + +def read_sample(path, index): + with path.open(newline="") as source: + rows = list(csv.DictReader(source)) + if len(rows) != 1: + raise ValueError(f"{path}: expected exactly one case, found {len(rows)}") + row = rows[0] + if row.get("index") != index: + raise ValueError(f"{path}: expected index {index}") + if any(not row.get(field) for field in CASE_FIELDS): + raise ValueError(f"{path}: missing workload fields") + for field, _, _ in METRICS: + value = float(row[field]) + if not math.isfinite(value) or value < 0: + raise ValueError(f"{path}: invalid {field}: {value}") + if not 0 <= float(row["recall_at_10"]) <= 1 or int(row["nq"]) <= 0: + raise ValueError(f"{path}: invalid recall or query count") + if float(row["sequential_qps"]) <= 0 or float(row["batch_qps"]) <= 0: + raise ValueError(f"{path}: throughput must be positive") + return row + + +def metric_value(row, field): + value = float(row[field]) + if "pread_" in field: + return value / int(row["nq"]) + if field in ("peak_rss_bytes", "file_bytes"): + return value / (1024 * 1024) + return value + + +def delta_text(base, candidate, direction): + if direction == "recall": + return f"{(candidate - base) * 100:+.2f} pp" + if base == 0: + return "0.0%" if candidate == 0 else "n/a (base=0)" + return f"{(candidate / base - 1) * 100:+.1f}%" + + +def summarize(samples, rounds): + results = {} + for index in INDEXES: + sides = samples[index] + if any(len(sides[side]) != rounds for side in ("base", "candidate")): + raise ValueError(f"{index}: incomplete samples") + cases = { + tuple(row[field] for field in CASE_FIELDS) + for rows in sides.values() for row in rows + } + if len(cases) != 1: + raise ValueError(f"{index}: workload differs between samples") + result = {} + for field, label, direction in METRICS: + entry = {"label": label, "direction": direction} + for side in ("base", "candidate"): + values = [metric_value(row, field) for row in sides[side]] + entry[side] = { + "median": statistics.median(values), + "min": min(values), "max": max(values), "samples": values, + } + entry["delta"] = delta_text( + entry["base"]["median"], entry["candidate"]["median"], direction + ) + result[field] = entry + results[index] = result + return results + + +def render_report(metadata, results): + lines = [ + "# PR / base benchmark", "", + f"- Base: `{metadata['base_sha']}`", + f"- Candidate (merge result in PR CI): `{metadata['candidate_sha']}`", + f"- Shared benchmark driver SHA-256: `{metadata['driver_sha256']}`", + f"- Rust: `{metadata['rustc'].splitlines()[0]}`", + f"- Runner: {metadata['platform']}; {metadata['cpu']}", + f"- {metadata['rounds']} fresh processes per version per index; " + "alternating base→candidate / candidate→base pairs; 2 Rayon threads.", + "- Fixed synthetic L2 workload: 10,000 × 64D, 4,096 training vectors, " + "2,048 queries, top-10, seed 42, nlist=64, nprobe=8, PQ m=8, DiskANN L=100.", + "- Local warm page cache. Each process builds its own index. Reader optimization " + "and one first query precede sequential queries; batch uses a separate optimized reader.", + "- Values are medians [min, max]. ↑ means higher is better, ↓ means lower is better. " + "Delta is PR/base − 1; recall delta is in percentage points (pp).", + "- Timing changes are observations, not a merge gate " + "or a statistical significance claim. Recall is measured on batch results.", + "- Peak RSS includes dataset/ground truth and is sampled after build; " + "it is not search peak memory. First query is not a cold-disk measurement.", + "", + ] + lines += ["| Index | Recall base → PR | Sequential QPS Δ | Batch QPS Δ | P95 Δ | Build Δ |", + "|---|---:|---:|---:|---:|---:|"] + for index, metrics in results.items(): + recall = metrics["recall_at_10"] + lines.append( + f"| {index} | {recall['base']['median'] * 100:.2f}% → " + f"{recall['candidate']['median'] * 100:.2f}% | " + f"{metrics['sequential_qps']['delta']} | {metrics['batch_qps']['delta']} | " + f"{metrics['p95_query_us']['delta']} | {metrics['build_ms']['delta']} |" + ) + lines += [""] + for index, metrics in results.items(): + lines += [f"
{index}: measurements and sample ranges", ""] + recall = metrics["recall_at_10"] + if recall["candidate"]["median"] < recall["base"]["median"]: + lines += ["**Recall decreased. Do not interpret faster queries as a quality-preserving improvement.**", ""] + lines += ["| Metric | Base [min, max] | PR [min, max] | Δ PR/base |", + "|---|---:|---:|---:|"] + for field, entry in metrics.items(): + values = [] + for side in ("base", "candidate"): + summary = entry[side] + scale = 100 if entry["direction"] == "recall" else 1 + unit = "%" if scale == 100 else "" + precision = 4 if field.endswith("pread_rounds") else 2 + values.append( + f"{summary['median'] * scale:,.{precision}f}{unit} " + f"[{summary['min'] * scale:,.{precision}f}, {summary['max'] * scale:,.{precision}f}]" + ) + arrow = "↑" if entry["direction"] in ("higher", "recall") else "↓" + lines.append(f"| {entry['label']} {arrow} | {values[0]} | {values[1]} | {entry['delta']} |") + lines += ["", "
", ""] + lines += ["Raw CSVs, stderr logs, build logs, environment metadata and summary.json " + "are available in the workflow artifact.", ""] + return "\n".join(lines) + + +def build(checkout, output, side, env): + # Separate target directories prevent artifacts from one revision leaking into the other. + command = ["cargo", "bench", "--locked", "-p", "paimon-vindex-core", "--bench", + "ann_bench", "--no-run", "--message-format=json", "--target-dir", + str(output / "build" / side)] + print(f"Building {side}", flush=True) + messages_path = output / f"build-{side}.jsonl" + with messages_path.open("w") as messages, (output / f"build-{side}.log").open("w") as log: + subprocess.run(command, cwd=checkout, env=env, stdout=messages, stderr=log, + check=True, timeout=900) + executables = [] + for line in messages_path.read_text().splitlines(): + message = json.loads(line) + if (message.get("reason") == "compiler-artifact" + and message.get("target", {}).get("name") == "ann_bench" + and message.get("executable")): + executables.append(message["executable"]) + if len(executables) != 1: + raise ValueError(f"{side}: expected one benchmark executable") + return executables[0] + + +def run(args): + output = args.output.resolve() + base, candidate = args.base.resolve(), args.candidate.resolve() + if base == candidate: + raise ValueError("base and candidate must be separate disposable checkouts") + if command_output(["git", "status", "--porcelain"], base): + raise ValueError("base checkout must be clean before copying the shared driver") + metadata = { + "base_sha": command_output(["git", "rev-parse", "HEAD"], base), + "candidate_sha": command_output(["git", "rev-parse", "HEAD"], candidate), + "rustc": command_output(["rustc", "--version", "--verbose"]), + "platform": platform.platform(), + "cpu": platform.processor(), + "rounds": args.rounds, "workload": WORKLOAD, + "candidate_dirty": bool(command_output(["git", "status", "--porcelain"], candidate)), + "execution_order": [], + "rustflags": "-C target-cpu=x86-64" if platform.machine() == "x86_64" else "", + } + if shutil.which("lscpu"): + metadata["cpu"] = next((line.split(":", 1)[1].strip() + for line in command_output(["lscpu"]).splitlines() + if line.startswith("Model name:")), metadata["cpu"]) + # Use candidate driver on both versions, including its support module. API incompatibility + # fails the run explicitly instead of silently comparing different benchmark programs. + digest = hashlib.sha256() + for relative in ("core/benches/ann_bench.rs", "core/benches/support/ann_bench_support.rs"): + data = (candidate / relative).read_bytes() + digest.update(relative.encode() + b"\0" + data) + (base / relative).parent.mkdir(parents=True, exist_ok=True) + (base / relative).write_bytes(data) + metadata["driver_sha256"] = digest.hexdigest() + (output / "metadata.json").write_text(json.dumps(metadata, indent=2) + "\n") + env = {key: value for key, value in os.environ.items() + if not key.startswith(("ANN_", "DISKANN_BENCH_", "CARGO_PROFILE_")) + and key not in ("RUSTFLAGS", "CARGO_ENCODED_RUSTFLAGS", "CARGO_BUILD_RUSTFLAGS")} + env.update(WORKLOAD) + env.update({"RUSTFLAGS": metadata["rustflags"], "LC_ALL": "C", "CARGO_INCREMENTAL": "0"}) + executables = {side: build(checkout, output, side, env) + for side, checkout in (("base", base), ("candidate", candidate))} + samples = {index: {"base": [], "candidate": []} for index in INDEXES} + raw = output / "raw" + raw.mkdir() + for index in INDEXES: + for repetition in range(args.rounds): + order = ("base", "candidate") if repetition % 2 == 0 else ("candidate", "base") + for side in order: + prefix = raw / f"{index}-{repetition + 1}-{side}" + print(f"Running {prefix.name}", flush=True) + sample_env = dict(env, ANN_INDEXES=index, ANN_OUTPUT_DIR=str(output / "indexes")) + started = time.monotonic() + with prefix.with_suffix(".csv").open("w") as csv_file, prefix.with_suffix(".log").open("w") as log: + subprocess.run([executables[side]], env=sample_env, cwd=output, + stdout=csv_file, stderr=log, timeout=args.timeout, check=True) + samples[index][side].append(read_sample(prefix.with_suffix(".csv"), index)) + metadata["execution_order"].append({ + "index": index, "round": repetition + 1, "side": side, + "wall_seconds": round(time.monotonic() - started, 3), + }) + (output / "metadata.json").write_text(json.dumps(metadata, indent=2) + "\n") + results = summarize(samples, args.rounds) + (output / "summary.json").write_text(json.dumps(results, indent=2) + "\n") + (output / "summary.md").write_text(render_report(metadata, results)) + print(f"Report: {output / 'summary.md'}", flush=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base", type=Path, required=True, help="Disposable base checkout (driver is overwritten)") + parser.add_argument("--candidate", type=Path, required=True, help="Candidate checkout providing the shared driver") + parser.add_argument("--output", type=Path, required=True, help="New output directory") + parser.add_argument("--rounds", type=int, default=4, help="Samples per version/index; even number >= 2") + parser.add_argument("--timeout", type=int, default=180, help="Timeout in seconds per sample") + args = parser.parse_args() + if args.rounds < 2 or args.rounds % 2 or args.timeout <= 0: + parser.error("rounds must be even and >= 2; timeout must be positive") + args.output = args.output.resolve() + args.output.mkdir(parents=True, exist_ok=False) + try: + run(args) + except Exception as error: + (args.output / "summary.md").write_text( + "# PR / base benchmark failed\n\n" + "The comparison is incomplete; no performance conclusion is available. " + "See the uploaded build/sample logs for details.\n\n" + f"```text\n{error}\n```\n" + ) + raise + + +if __name__ == "__main__": + main() diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py new file mode 100644 index 00000000..a721ff2d --- /dev/null +++ b/tools/tests/test_benchmark_pr.py @@ -0,0 +1,107 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import copy +import csv +import importlib.util +from pathlib import Path +import tempfile +import unittest + + +SPEC = importlib.util.spec_from_file_location( + "benchmark_pr", Path(__file__).resolve().parents[1] / "benchmark_pr.py" +) +bench = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(bench) + + +def row(index="IVF_FLAT"): + values = dict.fromkeys(bench.CASE_FIELDS, "1") + values.update({field: "100" for field, _, _ in bench.METRICS}) + values.update(index=index, nq="10", recall_at_10="0.95") + return values + + +def samples(): + return {index: {side: [row(index), row(index)] for side in ("base", "candidate")} + for index in bench.INDEXES} + + +class BenchmarkComparisonTest(unittest.TestCase): + def test_medians_units_and_recall_percentage_points(self): + values = samples() + values["IVF_FLAT"]["base"][1]["sequential_qps"] = "300" + values["IVF_FLAT"]["candidate"][0]["sequential_qps"] = "200" + values["IVF_FLAT"]["candidate"][1]["sequential_qps"] = "400" + for value in values["IVF_FLAT"]["candidate"]: + value["recall_at_10"] = "0.90" + result = bench.summarize(values, 2)["IVF_FLAT"] + self.assertEqual(result["sequential_qps"]["base"]["median"], 200) + self.assertEqual(result["sequential_qps"]["delta"], "+50.0%") + self.assertEqual(result["recall_at_10"]["delta"], "-5.00 pp") + self.assertEqual(result["sequential_pread_bytes"]["base"]["median"], 10) + metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", + rustc="rust", platform="os", cpu="cpu", rounds=2) + report = bench.render_report(metadata, bench.summarize(values, 2)) + self.assertIn("Recall decreased", report) + self.assertIn("[100.00, 300.00]", report) + + def test_incomplete_or_mismatched_workloads_fail(self): + for change in ("missing", "shape", "parameters"): + with self.subTest(change=change): + values = samples() + if change == "missing": + values["IVF_FLAT"]["candidate"].pop() + else: + field = "nq" if change == "shape" else "nprobe" + values["IVF_FLAT"]["candidate"][0][field] = "999" + with self.assertRaises(ValueError): + bench.summarize(values, 2) + + def test_csv_rejects_missing_duplicate_and_invalid_results(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "sample.csv" + good = row() + bad_values = [] + for field, value in (("batch_qps", "nan"), ("build_ms", "-1"), + ("recall_at_10", "1.1"), ("nq", "0"), + ("batch_qps", "0"), ("index", "DISKANN")): + bad = copy.copy(good) + bad[field] = value + bad_values.append([bad]) + for rows in ([], [good, good], *bad_values): + with self.subTest(rows=rows): + with path.open("w", newline="") as file: + writer = csv.DictWriter(file, fieldnames=good) + writer.writeheader() + writer.writerows(rows) + with self.assertRaises(ValueError): + bench.read_sample(path, "IVF_FLAT") + with path.open("w", newline="") as file: + writer = csv.DictWriter(file, fieldnames=good) + writer.writeheader() + writer.writerow(good) + self.assertEqual(bench.read_sample(path, "IVF_FLAT"), good) + + def test_zero_baseline_is_not_a_false_percentage(self): + self.assertEqual(bench.delta_text(0, 1, "lower"), "n/a (base=0)") + self.assertEqual(bench.delta_text(0, 0, "lower"), "0.0%") + + +if __name__ == "__main__": + unittest.main() From 278b85a54f46b6eeef1e1d8083b5cd4b0de58c50 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 21:48:06 +0800 Subject: [PATCH 02/11] ci: publish benchmark results as a sticky PR comment --- .github/workflows/benchmark-pr.yml | 53 ++++++++++++++++++++++++++++++ tools/README.md | 12 ++++--- 2 files changed, 61 insertions(+), 4 deletions(-) diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index d9c6ed45..0f841060 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -109,3 +109,56 @@ jobs: results/raw/ if-no-files-found: warn retention-days: 30 + + comment: + name: Update PR benchmark comment + needs: compare + # Same-repository PRs have a writable token. Fork PRs retain the read-only + # benchmark/summary; never run candidate code with a privileged trigger. + if: ${{ always() && !cancelled() && github.event.pull_request.head.repo.full_name == github.repository }} + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: + actions: read + pull-requests: write + steps: + # This job does not check out or execute PR code. Only read the report as text. + - name: Download benchmark report + id: report + continue-on-error: true + uses: actions/download-artifact@v4 + with: + name: pr-${{ github.event.pull_request.number }}-benchmark-${{ github.run_attempt }} + path: report + + - name: Create or update one PR comment + uses: actions/github-script@v7 + env: + BENCHMARK_RESULT: ${{ needs.compare.result }} + with: + script: | + const fs = require('fs'); + const marker = ''; + const { owner, repo } = context.repo; + const issue_number = context.payload.pull_request.number; + const { data: pr } = await github.rest.pulls.get({ owner, repo, pull_number: issue_number }); + if (pr.state !== 'open' || pr.head.sha !== context.payload.pull_request.head.sha) { + core.info('Skipping stale benchmark result: PR closed or head changed.'); + return; + } + const runUrl = `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`; + const path = 'report/summary.md'; + const report = fs.existsSync(path) + ? fs.readFileSync(path, 'utf8') + : '# PR / base benchmark failed\n\nNo comparison was produced. See the workflow logs.'; + const body = `${marker}\n[Benchmark run](${runUrl}) · ${process.env.BENCHMARK_RESULT} · attempt ${process.env.GITHUB_RUN_ATTEMPT}\n\n${report}`; + const comments = await github.paginate(github.rest.issues.listComments, { + owner, repo, issue_number, per_page: 100, + }); + const existing = comments.find(comment => + comment.user.login === 'github-actions[bot]' && comment.body.startsWith(marker)); + if (existing) { + await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body }); + } else { + await github.rest.issues.createComment({ owner, repo, issue_number, body }); + } diff --git a/tools/README.md b/tools/README.md index 3fb78e88..e4f041be 100644 --- a/tools/README.md +++ b/tools/README.md @@ -25,10 +25,14 @@ This directory contains helper scripts used by release managers and committers. The `PR benchmark` workflow compares the exact PR base SHA with GitHub's PR merge commit on the same Ubuntu runner. It runs when the core, Cargo configuration, or -benchmark tooling changes. Open **Checks → Compare base and PR → Summary** for -the comparison; the workflow artifact includes raw CSVs, build/sample logs, -environment metadata, and machine-readable results. No PR comment or external -service is required. +benchmark tooling changes. For branches in the same repository, the result is +posted directly on the PR as one bot comment that is updated on subsequent runs. +The comment includes the overview and expandable measurements; no download is +needed. A separate job with comment permission reads the report as text without +checking out or executing PR code. Results for an outdated PR head are ignored. +Fork PRs with read-only tokens still publish **Checks → Compare base and PR → +Summary**. The workflow artifact retains raw CSVs, build/sample logs, environment +metadata, and machine-readable results for further investigation. `benchmark_pr.py` builds both revisions in release mode with separate target directories, then runs four samples per version per index. Each index is tested From 91487f6c4b6e6bd02e7f184147bf8e0ee3958d90 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 21:48:39 +0800 Subject: [PATCH 03/11] ci: retain benchmark report for comment-only reruns --- .github/workflows/benchmark-pr.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index 0f841060..b8980360 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -99,7 +99,8 @@ jobs: if: always() uses: actions/upload-artifact@v4 with: - name: pr-${{ github.event.pull_request.number }}-benchmark-${{ github.run_attempt }} + name: pr-${{ github.event.pull_request.number }}-benchmark + overwrite: true path: | results/summary.md results/summary.json @@ -128,7 +129,7 @@ jobs: continue-on-error: true uses: actions/download-artifact@v4 with: - name: pr-${{ github.event.pull_request.number }}-benchmark-${{ github.run_attempt }} + name: pr-${{ github.event.pull_request.number }}-benchmark path: report - name: Create or update one PR comment From bf4a99e3e8c9da7f712ebd47161badd157d78fa5 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 22:03:09 +0800 Subject: [PATCH 04/11] ci: improve benchmark timing and bound failure handling --- .github/workflows/benchmark-pr.yml | 3 +- core/benches/ann_bench.rs | 51 ++++++++++- tools/README.md | 37 +++++--- tools/benchmark_pr.py | 130 +++++++++++++++++++++-------- tools/tests/test_benchmark_pr.py | 74 ++++++++++++++-- 5 files changed, 239 insertions(+), 56 deletions(-) diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index b8980360..4ad24607 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -82,7 +82,8 @@ jobs: run: python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v - name: Build and compare both revisions - run: python3 candidate/tools/benchmark_pr.py --base base --candidate candidate --output results + # Stop the script before the job deadline, leaving time to publish failures. + run: python3 candidate/tools/benchmark_pr.py --base base --candidate candidate --output results --total-timeout 1200 env: RUSTUP_TOOLCHAIN: 1.94.1 diff --git a/core/benches/ann_bench.rs b/core/benches/ann_bench.rs index a62566b1..9183dc46 100644 --- a/core/benches/ann_bench.rs +++ b/core/benches/ann_bench.rs @@ -999,8 +999,48 @@ fn run_query_case( ("DISKANN", DiskAnnRawVectorEncoding::F16) => "f16", _ => "none", }; + // The original passes above retain their first-pass I/O and recall semantics. + // Optional CI timing starts after those complete sweeps have warmed each reader. + let steady_min_ms: u64 = read_env("ANN_STEADY_MIN_MS", 0)?; + if steady_min_ms > 0 && !storage.latency.is_zero() { + return Err("ANN_STEADY_MIN_MS requires local_ssd_warm_cache".into()); + } + let mut steady_sequential_queries = 0usize; + let mut steady_batch_queries = 0usize; + let mut steady_latencies = Vec::new(); + let mut steady_sequential_elapsed = Duration::ZERO; + let mut steady_batch_elapsed = Duration::ZERO; + if steady_min_ms > 0 { + let minimum = Duration::from_millis(steady_min_ms); + let started = Instant::now(); + while started.elapsed() < minimum { + for query in dataset.queries.chunks_exact(config.d) { + let query_started = Instant::now(); + std::hint::black_box(reader.search(query, search)?); + steady_latencies.push(query_started.elapsed()); + } + steady_sequential_queries += config.nq; + } + steady_sequential_elapsed = started.elapsed(); + let started = Instant::now(); + while started.elapsed() < minimum { + std::hint::black_box(batch_reader.search_batch(&dataset.queries, config.nq, search)?); + steady_batch_queries += config.nq; + } + steady_batch_elapsed = started.elapsed(); + } + let steady_sequential_qps = if steady_sequential_queries == 0 { + 0.0 + } else { + steady_sequential_queries as f64 / steady_sequential_elapsed.as_secs_f64() + }; + let steady_batch_qps = if steady_batch_queries == 0 { + 0.0 + } else { + steady_batch_queries as f64 / steady_batch_elapsed.as_secs_f64() + }; println!( - "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks}", + "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks},{steady_min_ms},{steady_sequential_queries},{steady_batch_queries},{steady_sequential_ms},{steady_batch_ms},{steady_sequential_qps:.2},{steady_batch_qps:.2},{steady_sequential_p95_us}", dataset = config.dataset_name, index = index.name, storage = storage.name, @@ -1059,6 +1099,9 @@ fn run_query_case( rq_scalar_blocks = rq_stats.scalar_blocks, rq_seeded_lists = rq_stats.seeded_lists, rq_parallel_list_tasks = rq_stats.parallel_list_tasks, + steady_sequential_ms = steady_sequential_elapsed.as_millis(), + steady_batch_ms = steady_batch_elapsed.as_millis(), + steady_sequential_p95_us = percentile(&steady_latencies, 95).as_micros(), ); Ok(()) } @@ -1067,7 +1110,7 @@ struct CsvRow; impl CsvRow { fn header() -> &'static str { - "dataset,index,storage,n,train_n,raw_dataset_bytes,nq,d,k,nlist,nprobe,pq_m,rq_bits,diskann_build_distance,diskann_raw_vector_encoding,l_search,build_ms,train_ms,add_ms,write_ms,peak_rss_bytes,optimize_ms,optimize_pread_rounds,optimize_pread_ranges,optimize_pread_bytes,file_bytes,recall_at_10,first_query_us,p50_query_us,p95_query_us,sequential_qps,sequential_pread_rounds,sequential_pread_ranges,sequential_pread_bytes,batch_ms,batch_qps,batch_pread_rounds,batch_pread_ranges,batch_pread_bytes,rq_sequential_scanned_vectors,rq_sequential_refined_vectors,rq_sequential_final_distance_evaluations,rq_sequential_seeded_lists,rq_sequential_parallel_list_tasks,rq_scanned_vectors,rq_eligible_vectors,rq_refined_vectors,rq_refine_ratio,rq_final_distance_evaluations,rq_refined_coarse_byte_lookups,rq_extra_plane_byte_lookups,rq_fastscan_blocks,rq_scalar_blocks,rq_seeded_lists,rq_parallel_list_tasks" + "dataset,index,storage,n,train_n,raw_dataset_bytes,nq,d,k,nlist,nprobe,pq_m,rq_bits,diskann_build_distance,diskann_raw_vector_encoding,l_search,build_ms,train_ms,add_ms,write_ms,peak_rss_bytes,optimize_ms,optimize_pread_rounds,optimize_pread_ranges,optimize_pread_bytes,file_bytes,recall_at_10,first_query_us,p50_query_us,p95_query_us,sequential_qps,sequential_pread_rounds,sequential_pread_ranges,sequential_pread_bytes,batch_ms,batch_qps,batch_pread_rounds,batch_pread_ranges,batch_pread_bytes,rq_sequential_scanned_vectors,rq_sequential_refined_vectors,rq_sequential_final_distance_evaluations,rq_sequential_seeded_lists,rq_sequential_parallel_list_tasks,rq_scanned_vectors,rq_eligible_vectors,rq_refined_vectors,rq_refine_ratio,rq_final_distance_evaluations,rq_refined_coarse_byte_lookups,rq_extra_plane_byte_lookups,rq_fastscan_blocks,rq_scalar_blocks,rq_seeded_lists,rq_parallel_list_tasks,steady_min_ms,steady_sequential_queries,steady_batch_queries,steady_sequential_ms,steady_batch_ms,steady_sequential_qps,steady_batch_qps,steady_sequential_p95_us" } } @@ -1096,7 +1139,9 @@ fn exact_ground_truth(dataset: &Dataset, k: usize) -> Vec> { .then_with(|| left.1.cmp(&right.1)) }); } - distances.into_iter().map(|(_, row)| row).collect() + // Borrow the retained top-k slice: consuming the Vec can reuse its + // original N-vector allocation for every ground-truth row. + distances.iter().map(|&(_, row)| row).collect() }) .collect() } diff --git a/tools/README.md b/tools/README.md index e4f041be..87e17688 100644 --- a/tools/README.md +++ b/tools/README.md @@ -35,7 +35,7 @@ Summary**. The workflow artifact retains raw CSVs, build/sample logs, environmen metadata, and machine-readable results for further investigation. `benchmark_pr.py` builds both revisions in release mode with separate target -directories, then runs four samples per version per index. Each index is tested +directories, then runs six samples per version per index. Each index is tested in a fresh process, alternating base/candidate and candidate/base pairs. Both versions use the candidate's `ann_bench.rs` and its support module, so a change to the benchmark itself cannot silently change the measurement between sides. @@ -44,19 +44,31 @@ shared driver before results can be compared. The initial workload is deliberately small: 10,000 synthetic 64D vectors, 4,096 training vectors, 2,048 queries, top-10, seed 42, and two Rayon threads. It covers -IVF-FLAT, IVF-SQ, IVF-PQ, IVF-RQ and DiskANN on local warm page cache. Sequential -queries follow reader optimization and one first query; batch queries use a -separate optimized reader. It does not measure cold storage or real object-store -performance. Recall is currently measured for batch results. Process peak RSS -is sampled through build and includes dataset/ground-truth allocations, not -search peak memory. +IVF-FLAT, IVF-SQ, IVF-PQ, IVF-RQ and DiskANN on local warm page cache. First-pass +Recall and I/O preserve the original benchmark semantics: sequential queries +follow reader optimization and one first query; batch uses a separate optimized +reader. After those complete passes warm each reader, the timing phase repeats +full sequential and batch sweeps for at least one second each. The PR report's +QPS and P95 use this warm timing phase; first-pass metrics remain in the CSV. +Warm timing is enabled by `ANN_STEADY_MIN_MS=1000` and is local-storage only. +Its elapsed times and query counts are included in the CSV for verification. + +This does not measure cold storage or real object-store performance. Recall is +currently measured on the first batch results. RSS is the process lifetime peak +up to build completion, including dataset and ground-truth allocations, not +index-only or search peak memory. Ground-truth IDs are copied from the retained +top-k slice to avoid retaining an N-vector allocation per query. The report shows medians and min/max ranges, with relative changes for timing, throughput, I/O, memory, and size. Recall changes use percentage points. A recall drop is highlighted alongside performance. This first version is informational: performance/recall changes do not fail CI, but build failures, timeouts, invalid -metrics, missing samples, and mismatched workload parameters do. With only four -samples, a reported percentage is not a statistical significance claim. +metrics, missing samples, and mismatched workload parameters do. Sample ranges +are not confidence intervals, and a reported percentage is not a statistical +significance claim. When core sources and Cargo inputs are identical, the report +explicitly labels the run as A/A repeatability calibration. Such deltas cannot +demonstrate an index-code improvement. Repeated calibration runs are needed +before choosing any timing gate. To reproduce locally, create two **disposable** checkouts, use the same Rust toolchain as the workflow, and run from the candidate checkout: @@ -70,7 +82,12 @@ python3 tools/benchmark_pr.py \ The script overwrites the two benchmark-driver files in the base checkout. The output directory must not exist. `--rounds 2` provides a shorter smoke run; -`--timeout` sets the per-process timeout in seconds (default 180). Dataset and +`--timeout` sets the per-sample timeout in seconds (default 180). +`--total-timeout` bounds all builds and samples together (default 1,200 seconds). +Each subprocess receives the smaller of its own timeout and the remaining total +budget. On timeout, its process group is killed and a failure report is written. +The 25-minute CI job reserves time beyond this 20-minute script budget for setup +and report upload. Dataset and index options are pinned in the script; inherited `ANN_*` settings are removed. CI pins Rust 1.94.1 and the x86-64 CPU target, caches dependency downloads only, and uploads reports even when a comparison fails. diff --git a/tools/benchmark_pr.py b/tools/benchmark_pr.py index 6753016c..b3ed851e 100644 --- a/tools/benchmark_pr.py +++ b/tools/benchmark_pr.py @@ -27,6 +27,7 @@ from pathlib import Path import platform import shutil +import signal import statistics import subprocess import time @@ -53,31 +54,67 @@ "ANN_DISKANN_RAW_VECTOR_ENCODING": "f32", "ANN_STORAGE_CASES": "local_ssd_warm_cache", "RAYON_NUM_THREADS": "2", + "ANN_STEADY_MIN_MS": "1000", } # These identify the workload, not measured results. Never compare unlike cases. CASE_FIELDS = ( "dataset", "index", "storage", "n", "train_n", "nq", "d", "k", "nlist", "nprobe", "pq_m", "rq_bits", "diskann_build_distance", - "diskann_raw_vector_encoding", "l_search", + "diskann_raw_vector_encoding", "l_search", "steady_min_ms", ) METRICS = ( ("recall_at_10", "Batch Recall@10", "recall"), - ("sequential_qps", "Sequential QPS", "higher"), - ("batch_qps", "Batch QPS", "higher"), - ("p95_query_us", "Sequential P95 (µs)", "lower"), + ("steady_sequential_qps", "Warm sequential QPS", "higher"), + ("steady_batch_qps", "Warm batch QPS", "higher"), + ("steady_sequential_p95_us", "Warm sequential P95 (µs)", "lower"), ("first_query_us", "First query (µs)", "lower"), - ("sequential_pread_rounds", "Read rounds / sequential query", "lower"), - ("sequential_pread_bytes", "Read bytes / sequential query", "lower"), - ("batch_pread_rounds", "Read rounds / batch query", "lower"), - ("batch_pread_bytes", "Read bytes / batch query", "lower"), + ("sequential_pread_rounds", "First-pass read rounds / sequential query", "lower"), + ("sequential_pread_bytes", "First-pass read bytes / sequential query", "lower"), + ("batch_pread_rounds", "First-pass read rounds / batch query", "lower"), + ("batch_pread_bytes", "First-pass read bytes / batch query", "lower"), ("build_ms", "Build (ms)", "lower"), - ("peak_rss_bytes", "Process peak RSS through build (MiB)", "lower"), + ("peak_rss_bytes", "Process peak RSS up to build completion (MiB)", "lower"), ("file_bytes", "Index size (MiB)", "lower"), ) def command_output(command, cwd=None): - return subprocess.check_output(command, cwd=cwd, text=True).strip() + return subprocess.check_output(command, cwd=cwd, text=True, timeout=30).strip() + + +def run_checked(command, *, deadline, timeout, **kwargs): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Benchmark total time budget exhausted; stopping before next process") + limit = min(timeout, remaining) + process = subprocess.Popen(command, start_new_session=os.name == "posix", **kwargs) + try: + returncode = process.wait(timeout=limit) + except subprocess.TimeoutExpired as error: + # Cargo can leave rustc children alive if only the parent is killed. + if os.name == "posix": + os.killpg(process.pid, signal.SIGKILL) + else: + process.kill() + process.wait() + raise TimeoutError( + f"{command[0]} exceeded {limit:.1f}s process/remaining total budget" + ) from error + if returncode: + raise subprocess.CalledProcessError(returncode, command) + + +def numeric_field(row, field, path, *, integer=False): + raw = row.get(field) + if raw is None or not raw.strip(): + raise ValueError(f"{path}: missing or empty metric/parameter {field}") + try: + value = int(raw) if integer else float(raw) + except (ValueError, OverflowError) as error: + raise ValueError(f"{path}: invalid {field}={raw!r}; expected a number") from error + if not math.isfinite(value) or value < 0: + raise ValueError(f"{path}: invalid {field}={raw!r}; expected finite nonnegative value") + return value def read_sample(path, index): @@ -88,16 +125,23 @@ def read_sample(path, index): row = rows[0] if row.get("index") != index: raise ValueError(f"{path}: expected index {index}") - if any(not row.get(field) for field in CASE_FIELDS): - raise ValueError(f"{path}: missing workload fields") + for field in CASE_FIELDS: + if not row.get(field): + raise ValueError(f"{path}: missing workload field {field}") for field, _, _ in METRICS: - value = float(row[field]) - if not math.isfinite(value) or value < 0: - raise ValueError(f"{path}: invalid {field}: {value}") - if not 0 <= float(row["recall_at_10"]) <= 1 or int(row["nq"]) <= 0: + numeric_field(row, field, path) + if not 0 <= float(row["recall_at_10"]) <= 1 or numeric_field(row, "nq", path, integer=True) <= 0: raise ValueError(f"{path}: invalid recall or query count") - if float(row["sequential_qps"]) <= 0 or float(row["batch_qps"]) <= 0: + if float(row["steady_sequential_qps"]) <= 0 or float(row["steady_batch_qps"]) <= 0: raise ValueError(f"{path}: throughput must be positive") + minimum = numeric_field(row, "steady_min_ms", path, integer=True) + if minimum <= 0: + raise ValueError(f"{path}: steady_min_ms must be positive") + for mode in ("sequential", "batch"): + elapsed = numeric_field(row, f"steady_{mode}_ms", path) + queries = numeric_field(row, f"steady_{mode}_queries", path, integer=True) + if elapsed < minimum or queries < int(row["nq"]) or queries % int(row["nq"]): + raise ValueError(f"{path}: incomplete steady_{mode} measurement") return row @@ -159,25 +203,29 @@ def render_report(metadata, results): "alternating base→candidate / candidate→base pairs; 2 Rayon threads.", "- Fixed synthetic L2 workload: 10,000 × 64D, 4,096 training vectors, " "2,048 queries, top-10, seed 42, nlist=64, nprobe=8, PQ m=8, DiskANN L=100.", - "- Local warm page cache. Each process builds its own index. Reader optimization " - "and one first query precede sequential queries; batch uses a separate optimized reader.", + "- Local warm page cache. Each process builds its own index. First-pass Recall/I/O " + "are recorded before repeated timing. The complete sequential and batch passes " + "warm their separate readers; each timed mode then repeats full sweeps for at least 1 second.", "- Values are medians [min, max]. ↑ means higher is better, ↓ means lower is better. " "Delta is PR/base − 1; recall delta is in percentage points (pp).", "- Timing changes are observations, not a merge gate " "or a statistical significance claim. Recall is measured on batch results.", - "- Peak RSS includes dataset/ground truth and is sampled after build; " - "it is not search peak memory. First query is not a cold-disk measurement.", + "- RSS is the process lifetime peak up to build completion, including dataset/ground truth; " + "it is not index-only or search peak memory. First query is not a cold-disk measurement.", "", ] - lines += ["| Index | Recall base → PR | Sequential QPS Δ | Batch QPS Δ | P95 Δ | Build Δ |", + if metadata.get("calibration"): + lines += ["**A/A calibration: core sources and Cargo inputs are identical. " + "These timing deltas measure repeatability, not an index-code improvement.**", ""] + lines += ["| Index | Recall base → PR | Warm sequential QPS Δ | Warm batch QPS Δ | Warm P95 Δ | Build Δ |", "|---|---:|---:|---:|---:|---:|"] for index, metrics in results.items(): recall = metrics["recall_at_10"] lines.append( f"| {index} | {recall['base']['median'] * 100:.2f}% → " f"{recall['candidate']['median'] * 100:.2f}% | " - f"{metrics['sequential_qps']['delta']} | {metrics['batch_qps']['delta']} | " - f"{metrics['p95_query_us']['delta']} | {metrics['build_ms']['delta']} |" + f"{metrics['steady_sequential_qps']['delta']} | {metrics['steady_batch_qps']['delta']} | " + f"{metrics['steady_sequential_p95_us']['delta']} | {metrics['build_ms']['delta']} |" ) lines += [""] for index, metrics in results.items(): @@ -206,7 +254,7 @@ def render_report(metadata, results): return "\n".join(lines) -def build(checkout, output, side, env): +def build(checkout, output, side, env, deadline): # Separate target directories prevent artifacts from one revision leaking into the other. command = ["cargo", "bench", "--locked", "-p", "paimon-vindex-core", "--bench", "ann_bench", "--no-run", "--message-format=json", "--target-dir", @@ -214,8 +262,8 @@ def build(checkout, output, side, env): print(f"Building {side}", flush=True) messages_path = output / f"build-{side}.jsonl" with messages_path.open("w") as messages, (output / f"build-{side}.log").open("w") as log: - subprocess.run(command, cwd=checkout, env=env, stdout=messages, stderr=log, - check=True, timeout=900) + run_checked(command, cwd=checkout, env=env, stdout=messages, stderr=log, + deadline=deadline, timeout=900) executables = [] for line in messages_path.read_text().splitlines(): message = json.loads(line) @@ -228,7 +276,18 @@ def build(checkout, output, side, env): return executables[0] +def library_fingerprint(checkout): + paths = list((checkout / "core/src").rglob("*")) + paths += [checkout / name for name in ("Cargo.toml", "Cargo.lock", "core/Cargo.toml", "core/build.rs")] + paths += list((checkout / ".cargo").rglob("*")) + digest = hashlib.sha256() + for path in sorted(path for path in paths if path.is_file()): + digest.update(str(path.relative_to(checkout)).encode() + b"\0" + path.read_bytes()) + return digest.hexdigest() + + def run(args): + deadline = time.monotonic() + args.total_timeout output = args.output.resolve() base, candidate = args.base.resolve(), args.candidate.resolve() if base == candidate: @@ -244,8 +303,12 @@ def run(args): "rounds": args.rounds, "workload": WORKLOAD, "candidate_dirty": bool(command_output(["git", "status", "--porcelain"], candidate)), "execution_order": [], + "total_timeout_seconds": args.total_timeout, + "base_library_sha256": library_fingerprint(base), + "candidate_library_sha256": library_fingerprint(candidate), "rustflags": "-C target-cpu=x86-64" if platform.machine() == "x86_64" else "", } + metadata["calibration"] = metadata["base_library_sha256"] == metadata["candidate_library_sha256"] if shutil.which("lscpu"): metadata["cpu"] = next((line.split(":", 1)[1].strip() for line in command_output(["lscpu"]).splitlines() @@ -265,7 +328,7 @@ def run(args): and key not in ("RUSTFLAGS", "CARGO_ENCODED_RUSTFLAGS", "CARGO_BUILD_RUSTFLAGS")} env.update(WORKLOAD) env.update({"RUSTFLAGS": metadata["rustflags"], "LC_ALL": "C", "CARGO_INCREMENTAL": "0"}) - executables = {side: build(checkout, output, side, env) + executables = {side: build(checkout, output, side, env, deadline) for side, checkout in (("base", base), ("candidate", candidate))} samples = {index: {"base": [], "candidate": []} for index in INDEXES} raw = output / "raw" @@ -279,8 +342,8 @@ def run(args): sample_env = dict(env, ANN_INDEXES=index, ANN_OUTPUT_DIR=str(output / "indexes")) started = time.monotonic() with prefix.with_suffix(".csv").open("w") as csv_file, prefix.with_suffix(".log").open("w") as log: - subprocess.run([executables[side]], env=sample_env, cwd=output, - stdout=csv_file, stderr=log, timeout=args.timeout, check=True) + run_checked([executables[side]], env=sample_env, cwd=output, + stdout=csv_file, stderr=log, timeout=args.timeout, deadline=deadline) samples[index][side].append(read_sample(prefix.with_suffix(".csv"), index)) metadata["execution_order"].append({ "index": index, "round": repetition + 1, "side": side, @@ -298,11 +361,12 @@ def main(): parser.add_argument("--base", type=Path, required=True, help="Disposable base checkout (driver is overwritten)") parser.add_argument("--candidate", type=Path, required=True, help="Candidate checkout providing the shared driver") parser.add_argument("--output", type=Path, required=True, help="New output directory") - parser.add_argument("--rounds", type=int, default=4, help="Samples per version/index; even number >= 2") + parser.add_argument("--rounds", type=int, default=6, help="Samples per version/index; even number >= 2") parser.add_argument("--timeout", type=int, default=180, help="Timeout in seconds per sample") + parser.add_argument("--total-timeout", type=int, default=1200, help="Total build/sample budget in seconds") args = parser.parse_args() - if args.rounds < 2 or args.rounds % 2 or args.timeout <= 0: - parser.error("rounds must be even and >= 2; timeout must be positive") + if args.rounds < 2 or args.rounds % 2 or args.timeout <= 0 or args.total_timeout <= 0: + parser.error("rounds must be even and >= 2; timeouts must be positive") args.output = args.output.resolve() args.output.mkdir(parents=True, exist_ok=False) try: diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py index a721ff2d..544ace84 100644 --- a/tools/tests/test_benchmark_pr.py +++ b/tools/tests/test_benchmark_pr.py @@ -20,6 +20,9 @@ import importlib.util from pathlib import Path import tempfile +import subprocess +import sys +import time import unittest @@ -33,7 +36,9 @@ def row(index="IVF_FLAT"): values = dict.fromkeys(bench.CASE_FIELDS, "1") values.update({field: "100" for field, _, _ in bench.METRICS}) - values.update(index=index, nq="10", recall_at_10="0.95") + values.update(index=index, nq="10", recall_at_10="0.95", + steady_sequential_ms="100", steady_batch_ms="100", + steady_sequential_queries="100", steady_batch_queries="100") return values @@ -45,21 +50,22 @@ def samples(): class BenchmarkComparisonTest(unittest.TestCase): def test_medians_units_and_recall_percentage_points(self): values = samples() - values["IVF_FLAT"]["base"][1]["sequential_qps"] = "300" - values["IVF_FLAT"]["candidate"][0]["sequential_qps"] = "200" - values["IVF_FLAT"]["candidate"][1]["sequential_qps"] = "400" + values["IVF_FLAT"]["base"][1]["steady_sequential_qps"] = "300" + values["IVF_FLAT"]["candidate"][0]["steady_sequential_qps"] = "200" + values["IVF_FLAT"]["candidate"][1]["steady_sequential_qps"] = "400" for value in values["IVF_FLAT"]["candidate"]: value["recall_at_10"] = "0.90" result = bench.summarize(values, 2)["IVF_FLAT"] - self.assertEqual(result["sequential_qps"]["base"]["median"], 200) - self.assertEqual(result["sequential_qps"]["delta"], "+50.0%") + self.assertEqual(result["steady_sequential_qps"]["base"]["median"], 200) + self.assertEqual(result["steady_sequential_qps"]["delta"], "+50.0%") self.assertEqual(result["recall_at_10"]["delta"], "-5.00 pp") self.assertEqual(result["sequential_pread_bytes"]["base"]["median"], 10) metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", - rustc="rust", platform="os", cpu="cpu", rounds=2) + rustc="rust", platform="os", cpu="cpu", rounds=2, calibration=True) report = bench.render_report(metadata, bench.summarize(values, 2)) self.assertIn("Recall decreased", report) self.assertIn("[100.00, 300.00]", report) + self.assertIn("A/A calibration", report) def test_incomplete_or_mismatched_workloads_fail(self): for change in ("missing", "shape", "parameters"): @@ -78,9 +84,9 @@ def test_csv_rejects_missing_duplicate_and_invalid_results(self): path = Path(directory) / "sample.csv" good = row() bad_values = [] - for field, value in (("batch_qps", "nan"), ("build_ms", "-1"), + for field, value in (("steady_batch_qps", "nan"), ("build_ms", "-1"), ("recall_at_10", "1.1"), ("nq", "0"), - ("batch_qps", "0"), ("index", "DISKANN")): + ("steady_batch_qps", "0"), ("index", "DISKANN")): bad = copy.copy(good) bad[field] = value bad_values.append([bad]) @@ -98,6 +104,56 @@ def test_csv_rejects_missing_duplicate_and_invalid_results(self): writer.writerow(good) self.assertEqual(bench.read_sample(path, "IVF_FLAT"), good) + def test_parse_errors_identify_file_field_and_value(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "broken.csv" + for field, value in (("steady_batch_qps", None), ("steady_batch_qps", ""), + ("steady_batch_qps", "oops"), ("nq", "2.5")): + with self.subTest(field=field, value=value): + sample = row() + if value is None: + del sample[field] + else: + sample[field] = value + with path.open("w", newline="") as file: + writer = csv.DictWriter(file, fieldnames=sample) + writer.writeheader() + writer.writerow(sample) + with self.assertRaises(ValueError) as caught: + bench.read_sample(path, "IVF_FLAT") + self.assertIn(str(path), str(caught.exception)) + self.assertIn(field, str(caught.exception)) + if value: + self.assertIn(value, str(caught.exception)) + + def test_total_budget_stops_before_start_and_limits_running_process(self): + with tempfile.TemporaryDirectory() as directory: + marker = Path(directory) / "must-not-exist" + with self.assertRaisesRegex(TimeoutError, "total time budget"): + bench.run_checked([sys.executable, "-c", + "from pathlib import Path; Path(__import__('sys').argv[1]).touch()", + str(marker)], deadline=time.monotonic() - 1, timeout=10) + self.assertFalse(marker.exists()) + started = time.monotonic() + with self.assertRaisesRegex(TimeoutError, "remaining total budget"): + bench.run_checked([sys.executable, "-c", "import time; time.sleep(10)"], + deadline=started + 0.1, timeout=10, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + self.assertLess(time.monotonic() - started, 3) + + def test_short_or_partial_steady_measurements_fail(self): + for field, value in (("steady_batch_ms", "0"), ("steady_sequential_queries", "11")): + with self.subTest(field=field), tempfile.TemporaryDirectory() as directory: + sample = row() + sample[field] = value + path = Path(directory) / "incomplete.csv" + with path.open("w", newline="") as file: + writer = csv.DictWriter(file, fieldnames=sample) + writer.writeheader() + writer.writerow(sample) + with self.assertRaisesRegex(ValueError, "incomplete steady_"): + bench.read_sample(path, "IVF_FLAT") + def test_zero_baseline_is_not_a_false_percentage(self): self.assertEqual(bench.delta_text(0, 1, "lower"), "n/a (base=0)") self.assertEqual(bench.delta_text(0, 0, "lower"), "0.0%") From 42d61898fce019b9f1b9cc1b79813330809efbcf Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 22:08:29 +0800 Subject: [PATCH 05/11] bench: preserve fractional microseconds in warm P95 --- core/benches/ann_bench.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/benches/ann_bench.rs b/core/benches/ann_bench.rs index 9183dc46..3baf0649 100644 --- a/core/benches/ann_bench.rs +++ b/core/benches/ann_bench.rs @@ -1040,7 +1040,7 @@ fn run_query_case( steady_batch_queries as f64 / steady_batch_elapsed.as_secs_f64() }; println!( - "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks},{steady_min_ms},{steady_sequential_queries},{steady_batch_queries},{steady_sequential_ms},{steady_batch_ms},{steady_sequential_qps:.2},{steady_batch_qps:.2},{steady_sequential_p95_us}", + "{dataset},{index},{storage},{n},{train_n},{raw_dataset_bytes},{nq},{d},{k},{nlist},{nprobe},{pq_m},{rq_bits},{build_distance},{raw_vector_encoding},{l_search},{build_ms},{train_ms},{add_ms},{write_ms},{peak_rss_bytes},{optimize_ms},{optimize_rounds},{optimize_ranges},{optimize_bytes},{file_bytes},{recall:.4},{first_us},{p50_us},{p95_us},{sequential_qps:.2},{seq_rounds},{seq_ranges},{seq_bytes},{batch_ms},{batch_qps:.2},{batch_rounds},{batch_ranges},{batch_bytes},{rq_seq_scanned},{rq_seq_refined},{rq_seq_final},{rq_seq_seeded_lists},{rq_seq_parallel_list_tasks},{rq_scanned},{rq_eligible},{rq_refined},{rq_refine_ratio:.6},{rq_final},{rq_refined_coarse_lookups},{rq_extra_plane_lookups},{rq_fastscan_blocks},{rq_scalar_blocks},{rq_seeded_lists},{rq_parallel_list_tasks},{steady_min_ms},{steady_sequential_queries},{steady_batch_queries},{steady_sequential_ms},{steady_batch_ms},{steady_sequential_qps:.2},{steady_batch_qps:.2},{steady_sequential_p95_us:.3}", dataset = config.dataset_name, index = index.name, storage = storage.name, @@ -1101,7 +1101,7 @@ fn run_query_case( rq_parallel_list_tasks = rq_stats.parallel_list_tasks, steady_sequential_ms = steady_sequential_elapsed.as_millis(), steady_batch_ms = steady_batch_elapsed.as_millis(), - steady_sequential_p95_us = percentile(&steady_latencies, 95).as_micros(), + steady_sequential_p95_us = percentile(&steady_latencies, 95).as_secs_f64() * 1_000_000.0, ); Ok(()) } From 5f2eb7d26c2875b20ef604a29c88d9d683a2aebc Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 22:14:24 +0800 Subject: [PATCH 06/11] ci: summarize benchmark results with colored alerts --- tools/README.md | 8 +++- tools/benchmark_pr.py | 64 +++++++++++++++++++++++--------- tools/tests/test_benchmark_pr.py | 20 ++++++++++ 3 files changed, 72 insertions(+), 20 deletions(-) diff --git a/tools/README.md b/tools/README.md index 87e17688..5c748696 100644 --- a/tools/README.md +++ b/tools/README.md @@ -27,8 +27,12 @@ The `PR benchmark` workflow compares the exact PR base SHA with GitHub's PR merg commit on the same Ubuntu runner. It runs when the core, Cargo configuration, or benchmark tooling changes. For branches in the same repository, the result is posted directly on the PR as one bot comment that is updated on subsequent runs. -The comment includes the overview and expandable measurements; no download is -needed. A separate job with comment permission reads the report as text without +The comment opens with a five-row Recall/single-QPS/batch-QPS table and colored +status markers. QPS drops greater than 10%/20% or recall drops greater than +1/3 percentage points are yellow/red; these are loose visual reminders, not CI +gates or significance tests. All other metrics, samples, and environment details +are inside one collapsed section. No download is needed. +A separate job with comment permission reads the report as text without checking out or executing PR code. Results for an outdated PR head are ignored. Fork PRs with read-only tokens still publish **Checks → Compare base and PR → Summary**. The workflow artifact retains raw CSVs, build/sample logs, environment diff --git a/tools/benchmark_pr.py b/tools/benchmark_pr.py index b3ed851e..d912c052 100644 --- a/tools/benchmark_pr.py +++ b/tools/benchmark_pr.py @@ -191,9 +191,51 @@ def summarize(samples, rounds): return results +def alert_level(entry): + base, candidate = entry["base"]["median"], entry["candidate"]["median"] + if entry["direction"] == "recall": + loss = round((base - candidate) * 100, 6) + yellow, red = 1, 3 + else: + loss = round((1 - candidate / base) * 100, 6) if base else 0 + yellow, red = 10, 20 + return 2 if loss > red else 1 if loss > yellow else 0 + + def render_report(metadata, results): + focus = (("recall_at_10", "Recall"), ("steady_sequential_qps", "single QPS"), + ("steady_batch_qps", "batch QPS")) + levels = {index: max(alert_level(metrics[field]) for field, _ in focus) + for index, metrics in results.items()} + flagged = sum(level > 0 for level in levels.values()) + icon = ("🟢", "🟡", "🔴")[max(levels.values(), default=0)] + headline = (f"{flagged}/{len(results)} indexes need a look" if flagged + else f"No alerts ({len(results)}/{len(results)} indexes)") + lines = ["## Vector index benchmark", "", f"**{icon} {headline}**", ""] + if metadata.get("calibration"): + lines += ["A/A calibration — identical core code; deltas show measurement variation.", ""] + lines += ["| Index | Recall@10 (Δ) | Single QPS Δ | Batch QPS Δ | Status |", + "|---|---:|---:|---:|---|"] + for index, metrics in results.items(): + recall = metrics["recall_at_10"] + reasons = [label for field, label in focus if alert_level(metrics[field])] + status = ("🟢 OK" if not reasons else + f"{('🟢', '🟡', '🔴')[levels[index]]} {', '.join(reasons)}") + lines.append( + f"| {index} | {recall['candidate']['median'] * 100:.2f}% ({recall['delta']}) | " + f"{metrics['steady_sequential_qps']['delta']} | " + f"{metrics['steady_batch_qps']['delta']} | {status} |" + ) + lines += ["", "🟡 QPS ↓ >10% or Recall ↓ >1pp · 🔴 QPS ↓ >20% or Recall ↓ >3pp. " + "Advisory only; QPS is measured after warmup.", "", + "
All metrics, samples & environment", ""] + lines += render_details(metadata, results) + lines += ["", "
", ""] + return "\n".join(lines) + + +def render_details(metadata, results): lines = [ - "# PR / base benchmark", "", f"- Base: `{metadata['base_sha']}`", f"- Candidate (merge result in PR CI): `{metadata['candidate_sha']}`", f"- Shared benchmark driver SHA-256: `{metadata['driver_sha256']}`", @@ -214,22 +256,8 @@ def render_report(metadata, results): "it is not index-only or search peak memory. First query is not a cold-disk measurement.", "", ] - if metadata.get("calibration"): - lines += ["**A/A calibration: core sources and Cargo inputs are identical. " - "These timing deltas measure repeatability, not an index-code improvement.**", ""] - lines += ["| Index | Recall base → PR | Warm sequential QPS Δ | Warm batch QPS Δ | Warm P95 Δ | Build Δ |", - "|---|---:|---:|---:|---:|---:|"] - for index, metrics in results.items(): - recall = metrics["recall_at_10"] - lines.append( - f"| {index} | {recall['base']['median'] * 100:.2f}% → " - f"{recall['candidate']['median'] * 100:.2f}% | " - f"{metrics['steady_sequential_qps']['delta']} | {metrics['steady_batch_qps']['delta']} | " - f"{metrics['steady_sequential_p95_us']['delta']} | {metrics['build_ms']['delta']} |" - ) - lines += [""] for index, metrics in results.items(): - lines += [f"
{index}: measurements and sample ranges", ""] + lines += [f"### {index}", ""] recall = metrics["recall_at_10"] if recall["candidate"]["median"] < recall["base"]["median"]: lines += ["**Recall decreased. Do not interpret faster queries as a quality-preserving improvement.**", ""] @@ -248,10 +276,10 @@ def render_report(metadata, results): ) arrow = "↑" if entry["direction"] in ("higher", "recall") else "↓" lines.append(f"| {entry['label']} {arrow} | {values[0]} | {values[1]} | {entry['delta']} |") - lines += ["", "
", ""] + lines += [""] lines += ["Raw CSVs, stderr logs, build logs, environment metadata and summary.json " "are available in the workflow artifact.", ""] - return "\n".join(lines) + return lines def build(checkout, output, side, env, deadline): diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py index 544ace84..f8f0b470 100644 --- a/tools/tests/test_benchmark_pr.py +++ b/tools/tests/test_benchmark_pr.py @@ -158,6 +158,26 @@ def test_zero_baseline_is_not_a_false_percentage(self): self.assertEqual(bench.delta_text(0, 1, "lower"), "n/a (base=0)") self.assertEqual(bench.delta_text(0, 0, "lower"), "0.0%") + def test_loose_alert_boundaries_and_recall_precedence(self): + for candidate, expected in ((110, 0), (90, 0), (89, 1), (80, 1), (79, 2)): + entry = dict(direction="higher", base={"median": 100}, candidate={"median": candidate}) + self.assertEqual(bench.alert_level(entry), expected) + for candidate, expected in ((0.94, 0), (0.939, 1), (0.92, 1), (0.919, 2)): + entry = dict(direction="recall", base={"median": 0.95}, candidate={"median": candidate}) + self.assertEqual(bench.alert_level(entry), expected) + values = samples() + for sample in values["IVF_FLAT"]["candidate"]: + sample.update(recall_at_10="0.90", steady_batch_qps="200") + metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", + rustc="rust", platform="os", cpu="cpu", rounds=2) + report = bench.render_report(metadata, bench.summarize(values, 2)) + visible = report.split("
")[0] + self.assertIn("🔴 Recall", visible) + self.assertIn("1/5 indexes need a look", visible) + self.assertNotIn("Runner:", visible) + self.assertNotIn("RSS", visible) + self.assertEqual(report.count("
"), 1) + if __name__ == "__main__": unittest.main() From 28218779e6af3323b8ea6ada3630412d61360a88 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 22:27:41 +0800 Subject: [PATCH 07/11] ci: simplify benchmark details and skip stale base results --- .github/workflows/benchmark-pr.yml | 5 ++- tools/README.md | 10 ++--- tools/benchmark_pr.py | 67 ++++++++---------------------- tools/tests/test_benchmark_pr.py | 7 ++-- 4 files changed, 30 insertions(+), 59 deletions(-) diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index 4ad24607..aad4f9e0 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -144,8 +144,9 @@ jobs: const { owner, repo } = context.repo; const issue_number = context.payload.pull_request.number; const { data: pr } = await github.rest.pulls.get({ owner, repo, pull_number: issue_number }); - if (pr.state !== 'open' || pr.head.sha !== context.payload.pull_request.head.sha) { - core.info('Skipping stale benchmark result: PR closed or head changed.'); + if (pr.state !== 'open' || pr.head.sha !== context.payload.pull_request.head.sha + || pr.base.sha !== context.payload.pull_request.base.sha) { + core.info('Skipping stale benchmark result: PR closed or head/base changed.'); return; } const runUrl = `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`; diff --git a/tools/README.md b/tools/README.md index 5c748696..c1e071dd 100644 --- a/tools/README.md +++ b/tools/README.md @@ -30,10 +30,10 @@ posted directly on the PR as one bot comment that is updated on subsequent runs. The comment opens with a five-row Recall/single-QPS/batch-QPS table and colored status markers. QPS drops greater than 10%/20% or recall drops greater than 1/3 percentage points are yellow/red; these are loose visual reminders, not CI -gates or significance tests. All other metrics, samples, and environment details -are inside one collapsed section. No download is needed. +gates or significance tests. One collapsed table adds absolute QPS, P95, build +time, process RSS and index size. The key results need no download. A separate job with comment permission reads the report as text without -checking out or executing PR code. Results for an outdated PR head are ignored. +checking out or executing PR code. Results for an outdated PR head or base are ignored. Fork PRs with read-only tokens still publish **Checks → Compare base and PR → Summary**. The workflow artifact retains raw CSVs, build/sample logs, environment metadata, and machine-readable results for further investigation. @@ -63,8 +63,8 @@ up to build completion, including dataset and ground-truth allocations, not index-only or search peak memory. Ground-truth IDs are copied from the retained top-k slice to avoid retaining an N-vector allocation per query. -The report shows medians and min/max ranges, with relative changes for timing, -throughput, I/O, memory, and size. Recall changes use percentage points. A recall +The report shows medians, with relative QPS changes and recall changes in +percentage points. Sample ranges are kept in the JSON artifact. A recall drop is highlighted alongside performance. This first version is informational: performance/recall changes do not fail CI, but build failures, timeouts, invalid metrics, missing samples, and mismatched workload parameters do. Sample ranges diff --git a/tools/benchmark_pr.py b/tools/benchmark_pr.py index d912c052..efba2ef3 100644 --- a/tools/benchmark_pr.py +++ b/tools/benchmark_pr.py @@ -67,11 +67,6 @@ ("steady_sequential_qps", "Warm sequential QPS", "higher"), ("steady_batch_qps", "Warm batch QPS", "higher"), ("steady_sequential_p95_us", "Warm sequential P95 (µs)", "lower"), - ("first_query_us", "First query (µs)", "lower"), - ("sequential_pread_rounds", "First-pass read rounds / sequential query", "lower"), - ("sequential_pread_bytes", "First-pass read bytes / sequential query", "lower"), - ("batch_pread_rounds", "First-pass read rounds / batch query", "lower"), - ("batch_pread_bytes", "First-pass read bytes / batch query", "lower"), ("build_ms", "Build (ms)", "lower"), ("peak_rss_bytes", "Process peak RSS up to build completion (MiB)", "lower"), ("file_bytes", "Index size (MiB)", "lower"), @@ -147,8 +142,6 @@ def read_sample(path, index): def metric_value(row, field): value = float(row[field]) - if "pread_" in field: - return value / int(row["nq"]) if field in ("peak_rss_bytes", "file_bytes"): return value / (1024 * 1024) return value @@ -228,57 +221,33 @@ def render_report(metadata, results): ) lines += ["", "🟡 QPS ↓ >10% or Recall ↓ >1pp · 🔴 QPS ↓ >20% or Recall ↓ >3pp. " "Advisory only; QPS is measured after warmup.", "", - "
All metrics, samples & environment", ""] + "
Absolute metrics & environment", ""] lines += render_details(metadata, results) lines += ["", "
", ""] return "\n".join(lines) def render_details(metadata, results): + fields = ("steady_sequential_qps", "steady_batch_qps", "steady_sequential_p95_us", + "build_ms", "peak_rss_bytes", "file_bytes") lines = [ - f"- Base: `{metadata['base_sha']}`", - f"- Candidate (merge result in PR CI): `{metadata['candidate_sha']}`", - f"- Shared benchmark driver SHA-256: `{metadata['driver_sha256']}`", - f"- Rust: `{metadata['rustc'].splitlines()[0]}`", - f"- Runner: {metadata['platform']}; {metadata['cpu']}", - f"- {metadata['rounds']} fresh processes per version per index; " - "alternating base→candidate / candidate→base pairs; 2 Rayon threads.", - "- Fixed synthetic L2 workload: 10,000 × 64D, 4,096 training vectors, " - "2,048 queries, top-10, seed 42, nlist=64, nprobe=8, PQ m=8, DiskANN L=100.", - "- Local warm page cache. Each process builds its own index. First-pass Recall/I/O " - "are recorded before repeated timing. The complete sequential and batch passes " - "warm their separate readers; each timed mode then repeats full sweeps for at least 1 second.", - "- Values are medians [min, max]. ↑ means higher is better, ↓ means lower is better. " - "Delta is PR/base − 1; recall delta is in percentage points (pp).", - "- Timing changes are observations, not a merge gate " - "or a statistical significance claim. Recall is measured on batch results.", - "- RSS is the process lifetime peak up to build completion, including dataset/ground truth; " - "it is not index-only or search peak memory. First query is not a cold-disk measurement.", - "", + "Medians, base → PR. QPS/P95 are measured after warmup.", "", + "| Index | Single QPS | Batch QPS | P95 (µs) | Build (ms) | Process RSS (MiB) | Size (MiB) |", + "|---|---:|---:|---:|---:|---:|---:|", ] for index, metrics in results.items(): - lines += [f"### {index}", ""] - recall = metrics["recall_at_10"] - if recall["candidate"]["median"] < recall["base"]["median"]: - lines += ["**Recall decreased. Do not interpret faster queries as a quality-preserving improvement.**", ""] - lines += ["| Metric | Base [min, max] | PR [min, max] | Δ PR/base |", - "|---|---:|---:|---:|"] - for field, entry in metrics.items(): - values = [] - for side in ("base", "candidate"): - summary = entry[side] - scale = 100 if entry["direction"] == "recall" else 1 - unit = "%" if scale == 100 else "" - precision = 4 if field.endswith("pread_rounds") else 2 - values.append( - f"{summary['median'] * scale:,.{precision}f}{unit} " - f"[{summary['min'] * scale:,.{precision}f}, {summary['max'] * scale:,.{precision}f}]" - ) - arrow = "↑" if entry["direction"] in ("higher", "recall") else "↓" - lines.append(f"| {entry['label']} {arrow} | {values[0]} | {values[1]} | {entry['delta']} |") - lines += [""] - lines += ["Raw CSVs, stderr logs, build logs, environment metadata and summary.json " - "are available in the workflow artifact.", ""] + values = [f"{metrics[field]['base']['median']:,.2f} → " + f"{metrics[field]['candidate']['median']:,.2f}" for field in fields] + lines.append(f"| {index} | " + " | ".join(values) + " |") + lines += [ + "", + f"Base `{metadata['base_sha']}` · PR merge `{metadata['candidate_sha']}`", + f"{metadata['rustc'].splitlines()[0]} · {metadata['platform']} · {metadata['cpu']}", + f"{metadata['rounds']} samples/version/index, alternating execution order; " + "2 threads; 10k × 64D vectors; 2,048 queries; ≥1s per timed mode.", + "RSS is the process lifetime peak up to build completion, including dataset/ground truth.", + "Raw CSVs, sample ranges, workload settings and logs are in the workflow artifact.", + ] return lines diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py index f8f0b470..8819c352 100644 --- a/tools/tests/test_benchmark_pr.py +++ b/tools/tests/test_benchmark_pr.py @@ -59,12 +59,12 @@ def test_medians_units_and_recall_percentage_points(self): self.assertEqual(result["steady_sequential_qps"]["base"]["median"], 200) self.assertEqual(result["steady_sequential_qps"]["delta"], "+50.0%") self.assertEqual(result["recall_at_10"]["delta"], "-5.00 pp") - self.assertEqual(result["sequential_pread_bytes"]["base"]["median"], 10) + self.assertAlmostEqual(result["file_bytes"]["base"]["median"], 100 / (1024 * 1024)) metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", rustc="rust", platform="os", cpu="cpu", rounds=2, calibration=True) report = bench.render_report(metadata, bench.summarize(values, 2)) - self.assertIn("Recall decreased", report) - self.assertIn("[100.00, 300.00]", report) + self.assertIn("🔴 Recall", report) + self.assertIn("200.00 → 300.00", report) self.assertIn("A/A calibration", report) def test_incomplete_or_mismatched_workloads_fail(self): @@ -177,6 +177,7 @@ def test_loose_alert_boundaries_and_recall_precedence(self): self.assertNotIn("Runner:", visible) self.assertNotIn("RSS", visible) self.assertEqual(report.count("
"), 1) + self.assertEqual(report.count("| Index |"), 2) if __name__ == "__main__": From 0a9316611ddd4cb21270dae4cc85f63982ec75a5 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 23:03:43 +0800 Subject: [PATCH 08/11] test: keep benchmark checks focused on calculations and failures --- tools/tests/test_benchmark_pr.py | 21 +-------------------- 1 file changed, 1 insertion(+), 20 deletions(-) diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py index 8819c352..c348f7fc 100644 --- a/tools/tests/test_benchmark_pr.py +++ b/tools/tests/test_benchmark_pr.py @@ -60,12 +60,6 @@ def test_medians_units_and_recall_percentage_points(self): self.assertEqual(result["steady_sequential_qps"]["delta"], "+50.0%") self.assertEqual(result["recall_at_10"]["delta"], "-5.00 pp") self.assertAlmostEqual(result["file_bytes"]["base"]["median"], 100 / (1024 * 1024)) - metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", - rustc="rust", platform="os", cpu="cpu", rounds=2, calibration=True) - report = bench.render_report(metadata, bench.summarize(values, 2)) - self.assertIn("🔴 Recall", report) - self.assertIn("200.00 → 300.00", report) - self.assertIn("A/A calibration", report) def test_incomplete_or_mismatched_workloads_fail(self): for change in ("missing", "shape", "parameters"): @@ -158,26 +152,13 @@ def test_zero_baseline_is_not_a_false_percentage(self): self.assertEqual(bench.delta_text(0, 1, "lower"), "n/a (base=0)") self.assertEqual(bench.delta_text(0, 0, "lower"), "0.0%") - def test_loose_alert_boundaries_and_recall_precedence(self): + def test_loose_alert_boundaries(self): for candidate, expected in ((110, 0), (90, 0), (89, 1), (80, 1), (79, 2)): entry = dict(direction="higher", base={"median": 100}, candidate={"median": candidate}) self.assertEqual(bench.alert_level(entry), expected) for candidate, expected in ((0.94, 0), (0.939, 1), (0.92, 1), (0.919, 2)): entry = dict(direction="recall", base={"median": 0.95}, candidate={"median": candidate}) self.assertEqual(bench.alert_level(entry), expected) - values = samples() - for sample in values["IVF_FLAT"]["candidate"]: - sample.update(recall_at_10="0.90", steady_batch_qps="200") - metadata = dict(base_sha="base", candidate_sha="pr", driver_sha256="driver", - rustc="rust", platform="os", cpu="cpu", rounds=2) - report = bench.render_report(metadata, bench.summarize(values, 2)) - visible = report.split("
")[0] - self.assertIn("🔴 Recall", visible) - self.assertIn("1/5 indexes need a look", visible) - self.assertNotIn("Runner:", visible) - self.assertNotIn("RSS", visible) - self.assertEqual(report.count("
"), 1) - self.assertEqual(report.count("| Index |"), 2) if __name__ == "__main__": From fd186ad0c384125108f89f07d966009064a8d257 Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 23:08:30 +0800 Subject: [PATCH 09/11] ci: use stable Rust for PR benchmarks --- .github/workflows/benchmark-pr.yml | 14 +++++++------- tools/README.md | 5 +++-- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index aad4f9e0..04cbc3c0 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -61,12 +61,12 @@ jobs: with: python-version: '3.12' - - name: Set up fixed Rust toolchain + - name: Set up stable Rust toolchain run: | - rustup toolchain install 1.94.1 --profile minimal - rustup default 1.94.1 + rustup toolchain install stable --profile minimal + rustup default stable env: - RUSTUP_TOOLCHAIN: 1.94.1 + RUSTUP_TOOLCHAIN: stable - name: Cache Rust downloads uses: actions/cache@v4 @@ -74,9 +74,9 @@ jobs: path: | ~/.cargo/registry ~/.cargo/git - key: benchmark-${{ runner.os }}-rust-1.94.1-${{ hashFiles('candidate/Cargo.lock', 'base/Cargo.lock') }} + key: benchmark-${{ runner.os }}-rust-stable-${{ hashFiles('candidate/Cargo.lock', 'base/Cargo.lock') }} restore-keys: | - benchmark-${{ runner.os }}-rust-1.94.1- + benchmark-${{ runner.os }}-rust-stable- - name: Test comparison and failure handling run: python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v @@ -85,7 +85,7 @@ jobs: # Stop the script before the job deadline, leaving time to publish failures. run: python3 candidate/tools/benchmark_pr.py --base base --candidate candidate --output results --total-timeout 1200 env: - RUSTUP_TOOLCHAIN: 1.94.1 + RUSTUP_TOOLCHAIN: stable - name: Publish comparison summary if: always() diff --git a/tools/README.md b/tools/README.md index c1e071dd..c2be3ed5 100644 --- a/tools/README.md +++ b/tools/README.md @@ -93,8 +93,9 @@ budget. On timeout, its process group is killed and a failure report is written. The 25-minute CI job reserves time beyond this 20-minute script budget for setup and report upload. Dataset and index options are pinned in the script; inherited `ANN_*` settings are removed. -CI pins Rust 1.94.1 and the x86-64 CPU target, caches dependency downloads only, -and uploads reports even when a comparison fails. +CI uses the same Rust stable toolchain for both revisions and records the actual +compiler version in the report. It pins the x86-64 CPU target, caches dependency +downloads only, and uploads reports even when a comparison fails. ## ANN-Benchmarks dataset conversion From 9546cab470f86839c4f4bdd71ba8323a4f9438fe Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Mon, 7 Sep 2026 23:40:57 +0800 Subject: [PATCH 10/11] ci: publish fork PR benchmark reports with workflow_run --- .github/workflows/benchmark-comment.yml | 90 +++++++++++++++++++++++++ .github/workflows/benchmark-pr.yml | 76 +++++++-------------- tools/README.md | 29 ++++---- tools/tests/test_benchmark_comment.cjs | 76 +++++++++++++++++++++ 4 files changed, 206 insertions(+), 65 deletions(-) create mode 100644 .github/workflows/benchmark-comment.yml create mode 100644 tools/tests/test_benchmark_comment.cjs diff --git a/.github/workflows/benchmark-comment.yml b/.github/workflows/benchmark-comment.yml new file mode 100644 index 00000000..fe3b0b5a --- /dev/null +++ b/.github/workflows/benchmark-comment.yml @@ -0,0 +1,90 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +name: PR benchmark comment + +on: + workflow_run: + workflows: ['PR benchmark'] + types: [completed] + +permissions: + actions: read + pull-requests: write + +concurrency: + group: benchmark-comment-${{ github.event.workflow_run.head_repository.id }}-${{ github.event.workflow_run.head_branch }} + cancel-in-progress: false + +jobs: + comment: + if: ${{ github.event.workflow_run.event == 'pull_request' && github.event.workflow_run.conclusion != 'cancelled' }} + runs-on: ubuntu-24.04 + timeout-minutes: 5 + steps: + # Never check out PR code, restore its caches, or execute artifact contents. + - name: Download this run's comment input + uses: actions/download-artifact@v4 + with: + name: benchmark-comment-${{ github.event.workflow_run.run_attempt }} + run-id: ${{ github.event.workflow_run.id }} + github-token: ${{ github.token }} + path: report + + - name: Create or update benchmark comment + uses: actions/github-script@v7 + with: + script: | + const fs = require('fs'); + const { owner, repo } = context.repo; + const run = context.payload.workflow_run; + if (run.path !== '.github/workflows/benchmark-pr.yml') { + throw new Error('Unexpected source workflow'); + } + // Artifact fields are untrusted: bind the requested PR to GitHub's run data. + const input = JSON.parse(fs.readFileSync('report/benchmark-context.json', 'utf8')); + if (!Number.isSafeInteger(input.number) || input.number <= 0) { + throw new Error('Invalid PR number'); + } + const { data: pr } = await github.rest.pulls.get({ owner, repo, pull_number: input.number }); + if (pr.state !== 'open' || pr.base.repo.full_name !== `${owner}/${repo}` + || pr.head.repo?.id !== run.head_repository.id + || pr.head.ref !== run.head_branch || pr.head.sha !== run.head_sha + || input.head !== pr.head.sha || input.base !== pr.base.sha) { + core.info('Skipping unrelated or stale benchmark result'); + return; + } + // A delayed delivery from an earlier rerun must not replace the latest result. + const { data: current } = await github.rest.actions.getWorkflowRun({ owner, repo, run_id: run.id }); + if (current.run_attempt !== run.run_attempt) return; + const marker = ''; + const path = 'report/results/summary.md'; + const report = fs.existsSync(path) + ? fs.readFileSync(path, 'utf8') + : 'Benchmark did not produce a report. See the workflow logs.'; + if (Buffer.byteLength(report, 'utf8') > 60000) throw new Error('Report too large'); + const body = `${marker}\n[Benchmark run](${run.html_url}) · ${run.conclusion} · attempt ${run.run_attempt}\n\n${report}`; + const comments = await github.paginate(github.rest.issues.listComments, { + owner, repo, issue_number: pr.number, per_page: 100, + }); + const existing = comments.find(comment => + comment.user.login === 'github-actions[bot]' && comment.body?.startsWith(marker)); + if (existing) { + await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body }); + } else { + await github.rest.issues.createComment({ owner, repo, issue_number: pr.number, body }); + } diff --git a/.github/workflows/benchmark-pr.yml b/.github/workflows/benchmark-pr.yml index 04cbc3c0..184e1c2d 100644 --- a/.github/workflows/benchmark-pr.yml +++ b/.github/workflows/benchmark-pr.yml @@ -27,7 +27,9 @@ on: - '.cargo/**' - 'tools/benchmark_pr.py' - 'tools/tests/test_benchmark_pr.py' + - 'tools/tests/test_benchmark_comment.cjs' - '.github/workflows/benchmark-pr.yml' + - '.github/workflows/benchmark-comment.yml' permissions: contents: read @@ -42,6 +44,16 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 25 steps: + - name: Record comparison revisions + uses: actions/github-script@v7 + with: + script: | + const fs = require('fs'); + const pr = context.payload.pull_request; + fs.writeFileSync('benchmark-context.json', JSON.stringify({ + number: pr.number, head: pr.head.sha, base: pr.base.sha, + })); + - name: Check out PR merge result uses: actions/checkout@v4 with: @@ -79,7 +91,9 @@ jobs: benchmark-${{ runner.os }}-rust-stable- - name: Test comparison and failure handling - run: python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v + run: | + python3 -m unittest discover -s candidate/tools/tests -p 'test_benchmark_pr.py' -v + node candidate/tools/tests/test_benchmark_comment.cjs - name: Build and compare both revisions # Stop the script before the job deadline, leaving time to publish failures. @@ -112,56 +126,12 @@ jobs: if-no-files-found: warn retention-days: 30 - comment: - name: Update PR benchmark comment - needs: compare - # Same-repository PRs have a writable token. Fork PRs retain the read-only - # benchmark/summary; never run candidate code with a privileged trigger. - if: ${{ always() && !cancelled() && github.event.pull_request.head.repo.full_name == github.repository }} - runs-on: ubuntu-24.04 - timeout-minutes: 5 - permissions: - actions: read - pull-requests: write - steps: - # This job does not check out or execute PR code. Only read the report as text. - - name: Download benchmark report - id: report - continue-on-error: true - uses: actions/download-artifact@v4 - with: - name: pr-${{ github.event.pull_request.number }}-benchmark - path: report - - - name: Create or update one PR comment - uses: actions/github-script@v7 - env: - BENCHMARK_RESULT: ${{ needs.compare.result }} + - name: Upload comment input + if: always() + uses: actions/upload-artifact@v4 with: - script: | - const fs = require('fs'); - const marker = ''; - const { owner, repo } = context.repo; - const issue_number = context.payload.pull_request.number; - const { data: pr } = await github.rest.pulls.get({ owner, repo, pull_number: issue_number }); - if (pr.state !== 'open' || pr.head.sha !== context.payload.pull_request.head.sha - || pr.base.sha !== context.payload.pull_request.base.sha) { - core.info('Skipping stale benchmark result: PR closed or head/base changed.'); - return; - } - const runUrl = `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`; - const path = 'report/summary.md'; - const report = fs.existsSync(path) - ? fs.readFileSync(path, 'utf8') - : '# PR / base benchmark failed\n\nNo comparison was produced. See the workflow logs.'; - const body = `${marker}\n[Benchmark run](${runUrl}) · ${process.env.BENCHMARK_RESULT} · attempt ${process.env.GITHUB_RUN_ATTEMPT}\n\n${report}`; - const comments = await github.paginate(github.rest.issues.listComments, { - owner, repo, issue_number, per_page: 100, - }); - const existing = comments.find(comment => - comment.user.login === 'github-actions[bot]' && comment.body.startsWith(marker)); - if (existing) { - await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body }); - } else { - await github.rest.issues.createComment({ owner, repo, issue_number, body }); - } + name: benchmark-comment-${{ github.run_attempt }} + path: | + benchmark-context.json + results/summary.md + retention-days: 7 diff --git a/tools/README.md b/tools/README.md index c2be3ed5..5076c255 100644 --- a/tools/README.md +++ b/tools/README.md @@ -25,18 +25,23 @@ This directory contains helper scripts used by release managers and committers. The `PR benchmark` workflow compares the exact PR base SHA with GitHub's PR merge commit on the same Ubuntu runner. It runs when the core, Cargo configuration, or -benchmark tooling changes. For branches in the same repository, the result is -posted directly on the PR as one bot comment that is updated on subsequent runs. -The comment opens with a five-row Recall/single-QPS/batch-QPS table and colored -status markers. QPS drops greater than 10%/20% or recall drops greater than -1/3 percentage points are yellow/red; these are loose visual reminders, not CI -gates or significance tests. One collapsed table adds absolute QPS, P95, build -time, process RSS and index size. The key results need no download. -A separate job with comment permission reads the report as text without -checking out or executing PR code. Results for an outdated PR head or base are ignored. -Fork PRs with read-only tokens still publish **Checks → Compare base and PR → -Summary**. The workflow artifact retains raw CSVs, build/sample logs, environment -metadata, and machine-readable results for further investigation. +benchmark tooling changes. The result is +published in the Actions summary and posted as one updated bot comment, including +for external fork PRs. The comment opens with a five-row Recall/QPS table and +colored status markers. QPS drops greater than 10%/20% or recall drops greater +than 1/3 percentage points are yellow/red; these are advisory, not CI gates. +One collapsed table adds absolute QPS, P95, build time, process RSS and index size. + +A separate `workflow_run` publisher runs from the default branch with comment +permission. It downloads only the triggering run/attempt's report artifact, +checks the PR against GitHub's source repository, branch and revision data, and +skips closed PRs or changed head/base revisions. It never checks out PR code, +restores PR caches, or executes artifact contents. Raw CSVs, logs, environment +metadata and machine-readable results remain available as workflow artifacts. + +The publisher must first be merged into the target repository's default branch +before automatic comments can run. The PR introducing it still has its Actions +summary; subsequent runs can publish comments once the publisher is installed. `benchmark_pr.py` builds both revisions in release mode with separate target directories, then runs six samples per version per index. Each index is tested diff --git a/tools/tests/test_benchmark_comment.cjs b/tools/tests/test_benchmark_comment.cjs new file mode 100644 index 00000000..baf8352f --- /dev/null +++ b/tools/tests/test_benchmark_comment.cjs @@ -0,0 +1,76 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const workflow = fs.readFileSync(path.join(__dirname, '../../.github/workflows/benchmark-comment.yml'), 'utf8'); +const script = workflow.split(' script: |\n')[1].replace(/^ /gm, ''); +const execute = new (Object.getPrototypeOf(async function () {}).constructor)( + 'require', 'github', 'context', 'core', script); + +async function check({ mutate = () => {}, existing = false, missing = false, posts = 0 } = {}) { + const run = { id: 1, run_attempt: 1, path: '.github/workflows/benchmark-pr.yml', + head_repository: { id: 2 }, head_branch: 'feature', head_sha: 'head', + conclusion: 'success', html_url: 'https://github.com/owner/repo/actions/runs/1' }; + // Distinct repository IDs represent a fork PR; no workflow_run.pull_requests needed. + const pr = { number: 95, state: 'open', base: { sha: 'base', repo: { full_name: 'owner/repo' } }, + head: { sha: 'head', ref: 'feature', repo: { id: 2 } } }; + const input = { number: 95, head: 'head', base: 'base' }; + const current = { run_attempt: 1 }; + mutate({ run, pr, input, current }); + const calls = []; + const github = { rest: { + pulls: { get: async () => ({ data: pr }) }, + actions: { getWorkflowRun: async () => ({ data: current }) }, + issues: { listComments() {}, + updateComment: async value => calls.push(['update', value]), + createComment: async value => calls.push(['create', value]) }, + }, paginate: async () => existing ? [{ id: 7, user: { login: 'github-actions[bot]' }, + body: 'old' }] : [] }; + const files = { existsSync: () => !missing, readFileSync: name => { + if (name === 'report/benchmark-context.json') return JSON.stringify(input); + assert.equal(name, 'report/results/summary.md'); + return '🟢 Benchmark report'; + } }; + await execute(name => { assert.equal(name, 'fs'); return files; }, github, + { repo: { owner: 'owner', repo: 'repo' }, payload: { workflow_run: run } }, { info() {} }); + assert.equal(calls.length, posts); + if (posts) { + assert.equal(calls[0][0], existing ? 'update' : 'create'); + assert.match(calls[0][1].body, missing ? /did not produce/ : /🟢 Benchmark report/); + } +} + +(async () => { + await check({ posts: 1 }); + await check({ existing: true, posts: 1 }); + await check({ missing: true, posts: 1 }); + for (const mutate of [ + ({ pr }) => pr.state = 'closed', + ({ pr }) => pr.head.sha = 'new', + ({ pr }) => pr.base.sha = 'new', + ({ pr }) => pr.head.repo.id = 3, + ({ pr }) => pr.head.ref = 'other', + ({ pr }) => pr.base.repo.full_name = 'other/repo', + ({ input }) => input.head = 'other', + ({ current }) => current.run_attempt = 2, + ]) await check({ mutate }); + await assert.rejects(check({ mutate: ({ input }) => input.number = '../95' }), /Invalid PR/); + await assert.rejects(check({ mutate: ({ run }) => run.path = 'other.yml' }), /Unexpected source/); + console.log('Publisher checks passed: fork routing, create/update, missing report and stale/unrelated runs'); +})().catch(error => { console.error(error); process.exitCode = 1; }); From 2a76386e640a75ffd5c7289d9c101e78e78254fa Mon Sep 17 00:00:00 2001 From: wangzhigang Date: Tue, 8 Sep 2026 00:12:52 +0800 Subject: [PATCH 11/11] bench: bound warm latency samples and clarify query alerts --- core/benches/ann_bench.rs | 9 ++++++--- tools/README.md | 4 +++- tools/benchmark_pr.py | 11 +++++++---- tools/tests/test_benchmark_pr.py | 7 +++++-- 4 files changed, 21 insertions(+), 10 deletions(-) diff --git a/core/benches/ann_bench.rs b/core/benches/ann_bench.rs index 3baf0649..5f685439 100644 --- a/core/benches/ann_bench.rs +++ b/core/benches/ann_bench.rs @@ -1017,7 +1017,10 @@ fn run_query_case( for query in dataset.queries.chunks_exact(config.d) { let query_started = Instant::now(); std::hint::black_box(reader.search(query, search)?); - steady_latencies.push(query_started.elapsed()); + // Keep one complete warm sweep for P95; bound storage to nq samples. + if steady_sequential_queries == 0 { + steady_latencies.push(query_started.elapsed()); + } } steady_sequential_queries += config.nq; } @@ -1139,8 +1142,8 @@ fn exact_ground_truth(dataset: &Dataset, k: usize) -> Vec> { .then_with(|| left.1.cmp(&right.1)) }); } - // Borrow the retained top-k slice: consuming the Vec can reuse its - // original N-vector allocation for every ground-truth row. + // Borrow to collect a fresh top-k Vec. Consuming the iterator can + // reuse the much larger N-vector allocation via in-place collect. distances.iter().map(|&(_, row)| row).collect() }) .collect() diff --git a/tools/README.md b/tools/README.md index 5076c255..39375020 100644 --- a/tools/README.md +++ b/tools/README.md @@ -58,7 +58,9 @@ Recall and I/O preserve the original benchmark semantics: sequential queries follow reader optimization and one first query; batch uses a separate optimized reader. After those complete passes warm each reader, the timing phase repeats full sequential and batch sweeps for at least one second each. The PR report's -QPS and P95 use this warm timing phase; first-pass metrics remain in the CSV. +QPS uses the full warm timing phase. P95 uses only its first complete sequential +sweep, bounding latency storage to one sample per query. First-pass metrics +remain in the CSV. Warm timing is enabled by `ANN_STEADY_MIN_MS=1000` and is local-storage only. Its elapsed times and query counts are included in the CSV for verification. diff --git a/tools/benchmark_pr.py b/tools/benchmark_pr.py index efba2ef3..a052b2d2 100644 --- a/tools/benchmark_pr.py +++ b/tools/benchmark_pr.py @@ -184,7 +184,10 @@ def summarize(samples, rounds): return results -def alert_level(entry): +def query_alert_level(entry): + """Advisory levels for recall and QPS only; other metrics have no thresholds.""" + if entry["direction"] not in ("recall", "higher"): + raise ValueError("Query alerts only support recall and QPS") base, candidate = entry["base"]["median"], entry["candidate"]["median"] if entry["direction"] == "recall": loss = round((base - candidate) * 100, 6) @@ -198,7 +201,7 @@ def alert_level(entry): def render_report(metadata, results): focus = (("recall_at_10", "Recall"), ("steady_sequential_qps", "single QPS"), ("steady_batch_qps", "batch QPS")) - levels = {index: max(alert_level(metrics[field]) for field, _ in focus) + levels = {index: max(query_alert_level(metrics[field]) for field, _ in focus) for index, metrics in results.items()} flagged = sum(level > 0 for level in levels.values()) icon = ("🟢", "🟡", "🔴")[max(levels.values(), default=0)] @@ -211,7 +214,7 @@ def render_report(metadata, results): "|---|---:|---:|---:|---|"] for index, metrics in results.items(): recall = metrics["recall_at_10"] - reasons = [label for field, label in focus if alert_level(metrics[field])] + reasons = [label for field, label in focus if query_alert_level(metrics[field])] status = ("🟢 OK" if not reasons else f"{('🟢', '🟡', '🔴')[levels[index]]} {', '.join(reasons)}") lines.append( @@ -231,7 +234,7 @@ def render_details(metadata, results): fields = ("steady_sequential_qps", "steady_batch_qps", "steady_sequential_p95_us", "build_ms", "peak_rss_bytes", "file_bytes") lines = [ - "Medians, base → PR. QPS/P95 are measured after warmup.", "", + "Medians, base → PR. QPS is timed after warmup; P95 uses the first warm sweep.", "", "| Index | Single QPS | Batch QPS | P95 (µs) | Build (ms) | Process RSS (MiB) | Size (MiB) |", "|---|---:|---:|---:|---:|---:|---:|", ] diff --git a/tools/tests/test_benchmark_pr.py b/tools/tests/test_benchmark_pr.py index c348f7fc..86df1813 100644 --- a/tools/tests/test_benchmark_pr.py +++ b/tools/tests/test_benchmark_pr.py @@ -155,10 +155,13 @@ def test_zero_baseline_is_not_a_false_percentage(self): def test_loose_alert_boundaries(self): for candidate, expected in ((110, 0), (90, 0), (89, 1), (80, 1), (79, 2)): entry = dict(direction="higher", base={"median": 100}, candidate={"median": candidate}) - self.assertEqual(bench.alert_level(entry), expected) + self.assertEqual(bench.query_alert_level(entry), expected) for candidate, expected in ((0.94, 0), (0.939, 1), (0.92, 1), (0.919, 2)): entry = dict(direction="recall", base={"median": 0.95}, candidate={"median": candidate}) - self.assertEqual(bench.alert_level(entry), expected) + self.assertEqual(bench.query_alert_level(entry), expected) + + with self.assertRaises(ValueError): + bench.query_alert_level(dict(direction="lower", base={"median": 100}, candidate={"median": 120})) if __name__ == "__main__":