|
| 1 | +# SPDX-License-Identifier: Apache-2.0 |
| 2 | +# SPDX-FileCopyrightText: Copyright the Vortex contributors |
| 3 | + |
| 4 | +"""Compare V1, previous V2, and current V2 with full Arrow materialization.""" |
| 5 | + |
| 6 | +import argparse |
| 7 | +import csv |
| 8 | +import hashlib |
| 9 | +import itertools |
| 10 | +import json |
| 11 | +import os |
| 12 | +import platform |
| 13 | +import statistics |
| 14 | +import subprocess |
| 15 | +from pathlib import Path |
| 16 | + |
| 17 | + |
| 18 | +def digest(path: Path) -> str: |
| 19 | + checksum = hashlib.sha256() |
| 20 | + with path.open("rb") as source: |
| 21 | + for chunk in iter(lambda: source.read(1024 * 1024), b""): |
| 22 | + checksum.update(chunk) |
| 23 | + return checksum.hexdigest() |
| 24 | + |
| 25 | + |
| 26 | +def change(current: float, baseline: float) -> float: |
| 27 | + return 100 * (current / baseline - 1) |
| 28 | + |
| 29 | + |
| 30 | +def measure( |
| 31 | + case: dict, |
| 32 | + versions: dict[str, tuple[Path, str]], |
| 33 | + files: dict[str, Path], |
| 34 | + cpus: list[int], |
| 35 | + output: Path, |
| 36 | + iterations: int, |
| 37 | + rounds: int, |
| 38 | +) -> dict: |
| 39 | + env = {k: v for k, v in os.environ.items() if not k.startswith(("VORTEX_", "TOKIO_"))} |
| 40 | + env["RUST_LOG"] = "warn" |
| 41 | + seeds = [] |
| 42 | + for seed in [17, 42, 73]: |
| 43 | + medians = {v: [] for v in versions} |
| 44 | + expected_indices = None |
| 45 | + for repeat in range(rounds): |
| 46 | + order = list(versions)[:: 1 if repeat % 2 == 0 else -1] |
| 47 | + for version in order: |
| 48 | + binary, flag = versions[version] |
| 49 | + stem = "-".join(str(v) for v in (*case.values(), seed, repeat, version)) |
| 50 | + command = [ |
| 51 | + "taskset", |
| 52 | + "-c", |
| 53 | + ",".join(map(str, cpus[: case["cores"]])), |
| 54 | + str(binary), |
| 55 | + str(files[case["dataset"]]), |
| 56 | + str(iterations), |
| 57 | + "100", |
| 58 | + case["pattern"], |
| 59 | + case["mode"], |
| 60 | + "disk", |
| 61 | + "decode", |
| 62 | + str(seed), |
| 63 | + ] |
| 64 | + with (output / f"{stem}.json").open("w") as target, (output / f"{stem}.log").open("w") as log: |
| 65 | + subprocess.run( |
| 66 | + command, env=dict(env, VORTEX_SCAN_V2=flag), stdout=target, stderr=log, check=True, timeout=180 |
| 67 | + ) |
| 68 | + result = json.loads((output / f"{stem}.json").read_text()) |
| 69 | + if result["v2"] != flag or len(result["runs"]) != iterations: |
| 70 | + raise RuntimeError(f"Wrong scan mode or sample count: {stem}") |
| 71 | + if any(run["instrumented"] for run in result["runs"]): |
| 72 | + raise RuntimeError(f"Instrumented binary used for timing: {stem}") |
| 73 | + if expected_indices is None: |
| 74 | + expected_indices = result["indices"] |
| 75 | + if result["indices"] != expected_indices: |
| 76 | + raise RuntimeError(f"Versions selected different rows: {stem}") |
| 77 | + samples = result["runs"][iterations // 8 :] |
| 78 | + medians[version].append(statistics.median(x["elapsed_ns"] for x in samples) / 1e6) |
| 79 | + times = {v: statistics.median(values) for v, values in medians.items()} |
| 80 | + seeds.append(dict(seed=seed, **times, rounds=medians)) |
| 81 | + row = dict(case, **{v + "_ms": statistics.median(s[v] for s in seeds) for v in versions}, seeds=seeds) |
| 82 | + row["vs_old_v2_pct"] = change(row["new_v2_ms"], row["old_v2_ms"]) |
| 83 | + if "v1" in versions: |
| 84 | + row["vs_v1_pct"] = change(row["new_v2_ms"], row["v1_ms"]) |
| 85 | + print(json.dumps({k: v for k, v in row.items() if k != "seeds"}), flush=True) |
| 86 | + return row |
| 87 | + |
| 88 | + |
| 89 | +def save(output: Path, metadata: dict, summary: list[dict], confirmations: list[dict]): |
| 90 | + (output / "results.json").write_text( |
| 91 | + json.dumps(dict(metadata, summary=summary, confirmations=confirmations), indent=2) + "\n" |
| 92 | + ) |
| 93 | + with (output / "results.csv").open("w") as target: |
| 94 | + writer = csv.DictWriter(target, fieldnames=[k for k in summary[0] if k != "seeds"]) |
| 95 | + writer.writeheader() |
| 96 | + writer.writerows({k: v for k, v in row.items() if k != "seeds"} for row in summary) |
| 97 | + |
| 98 | + |
| 99 | +def report(output: Path, metadata: dict, summary: list[dict], confirmations: list[dict]): |
| 100 | + lines = [ |
| 101 | + "# Scan V2 regression comparison", |
| 102 | + "", |
| 103 | + f"Candidate: `{metadata['candidate_commit']}`; previous V2: `{metadata['baseline_commit']}`.", |
| 104 | + "", |
| 105 | + "100 selected rows; full Arrow decode/drop; warm local disk. " |
| 106 | + "Cached reuses file preparation; reopen includes opening/metadata.", |
| 107 | + "Three seeds, two alternating rounds of 64 queries, first 8 discarded. Positive percentages mean slower.", |
| 108 | + "", |
| 109 | + f"Cases above +3% vs previous V2: {sum(r['vs_old_v2_pct'] > 3 for r in summary)}/{len(summary)}; " |
| 110 | + f"above +5% vs V1: {sum(r['vs_v1_pct'] > 5 for r in summary)}/{len(summary)}.", |
| 111 | + "", |
| 112 | + "| Dataset | Cores | Selection | Open | V1 ms | Previous V2 ms | Current V2 ms | vs V1 | vs previous V2 |", |
| 113 | + "|---|---:|---|---|---:|---:|---:|---:|---:|", |
| 114 | + ] |
| 115 | + for r in summary: |
| 116 | + lines.append( |
| 117 | + f"| {r['dataset']} | {r['cores']} | {r['pattern']} | {r['mode']} | {r['v1_ms']:.4f} | " |
| 118 | + f"{r['old_v2_ms']:.4f} | {r['new_v2_ms']:.4f} | {r['vs_v1_pct']:+.1f}% | {r['vs_old_v2_pct']:+.1f}% |" |
| 119 | + ) |
| 120 | + if confirmations: |
| 121 | + lines += [ |
| 122 | + "", |
| 123 | + "## Longer reruns", |
| 124 | + "", |
| 125 | + "Rerun when the first-pass aggregate exceeds +2%, or any seed exceeds +3%, versus previous V2.", |
| 126 | + "Four alternating rounds of 128 queries, first 16 discarded, same three seeds.", |
| 127 | + "", |
| 128 | + "| Dataset | Cores | Selection | Open | Previous V2 ms | Current V2 ms | Change |", |
| 129 | + "|---|---:|---|---|---:|---:|---:|", |
| 130 | + ] |
| 131 | + for r in confirmations: |
| 132 | + lines.append( |
| 133 | + f"| {r['dataset']} | {r['cores']} | {r['pattern']} | {r['mode']} | " |
| 134 | + f"{r['old_v2_ms']:.4f} | {r['new_v2_ms']:.4f} | {r['vs_old_v2_pct']:+.1f}% |" |
| 135 | + ) |
| 136 | + text = "\n".join(lines) + "\n" |
| 137 | + (output / "summary.md").write_text(text) |
| 138 | + if step_summary := os.environ.get("GITHUB_STEP_SUMMARY"): |
| 139 | + with Path(step_summary).open("a") as target: |
| 140 | + target.write(text) |
| 141 | + |
| 142 | + |
| 143 | +def main(): |
| 144 | + parser = argparse.ArgumentParser(description=__doc__) |
| 145 | + parser.add_argument("--baseline", type=Path, required=True) |
| 146 | + parser.add_argument("--candidate", type=Path, required=True) |
| 147 | + parser.add_argument("--baseline-commit", required=True) |
| 148 | + parser.add_argument("--candidate-commit", required=True) |
| 149 | + parser.add_argument("--data-dir", type=Path, default=Path("vortex-bench/data")) |
| 150 | + parser.add_argument("--list-dir", type=Path, default=Path("scan-v2-data")) |
| 151 | + parser.add_argument("--output", type=Path, default=Path("scan-v2-results")) |
| 152 | + args = parser.parse_args() |
| 153 | + args.output.mkdir(parents=True, exist_ok=True) |
| 154 | + raw = args.output / "raw" |
| 155 | + raw.mkdir(exist_ok=True) |
| 156 | + cpus = sorted(os.sched_getaffinity(0)) |
| 157 | + if len(cpus) < 30: |
| 158 | + raise RuntimeError(f"This comparison requires 30 available CPUs; found {len(cpus)}") |
| 159 | + files = { |
| 160 | + "taxi": args.data_dir / "taxi/taxi.vortex", |
| 161 | + "feature-vectors": args.data_dir / "random_access/feature_vectors/feature_vectors.vortex", |
| 162 | + "nested-structs": args.data_dir / "random_access/nested_structs/nested_structs.vortex", |
| 163 | + "flat-lists": args.list_dir / "flat-control.vortex", |
| 164 | + "list-1MiB": args.list_dir / "elements-default.vortex", |
| 165 | + "list-64KiB": args.list_dir / "elements-64k.vortex", |
| 166 | + } |
| 167 | + versions = { |
| 168 | + "v1": (args.candidate.resolve(), "0"), |
| 169 | + "old_v2": (args.baseline.resolve(), "1"), |
| 170 | + "new_v2": (args.candidate.resolve(), "1"), |
| 171 | + } |
| 172 | + metadata = dict( |
| 173 | + baseline_commit=args.baseline_commit, |
| 174 | + candidate_commit=args.candidate_commit, |
| 175 | + platform=platform.platform(), |
| 176 | + cpus=cpus[:30], |
| 177 | + method=dict( |
| 178 | + rows=100, |
| 179 | + seeds=[17, 42, 73], |
| 180 | + iterations=64, |
| 181 | + warmups=8, |
| 182 | + rounds=2, |
| 183 | + output="full Arrow decode and drop", |
| 184 | + storage="warm local disk", |
| 185 | + aggregation="median of round medians, then median across seeds", |
| 186 | + confirmation_iterations=128, |
| 187 | + confirmation_warmups=16, |
| 188 | + confirmation_rounds=4, |
| 189 | + ), |
| 190 | + files={k: dict(path=str(p), sha256=digest(p), bytes=p.stat().st_size) for k, p in files.items()}, |
| 191 | + versions={k: dict(binary=str(p), sha256=digest(p), scan_v2=f) for k, (p, f) in versions.items()}, |
| 192 | + ) |
| 193 | + (args.output / "metadata.json").write_text(json.dumps(metadata, indent=2) + "\n") |
| 194 | + summary = [] |
| 195 | + confirmations = [] |
| 196 | + for cores, dataset, pattern, mode in itertools.product( |
| 197 | + [1, 30], files, ["clustered", "uniform"], ["cached", "reopen"] |
| 198 | + ): |
| 199 | + case = dict(dataset=dataset, cores=cores, pattern=pattern, mode=mode) |
| 200 | + summary.append(measure(case, versions, files, cpus, raw, 64, 2)) |
| 201 | + save(args.output, metadata, summary, confirmations) |
| 202 | + reruns = args.output / "reruns" |
| 203 | + reruns.mkdir(exist_ok=True) |
| 204 | + for row in summary: |
| 205 | + if row["vs_old_v2_pct"] <= 2 and all(change(s["new_v2"], s["old_v2"]) <= 3 for s in row["seeds"]): |
| 206 | + continue |
| 207 | + case = {k: row[k] for k in ["dataset", "cores", "pattern", "mode"]} |
| 208 | + confirmations.append( |
| 209 | + measure(case, {k: v for k, v in versions.items() if k != "v1"}, files, cpus, reruns, 128, 4) |
| 210 | + ) |
| 211 | + save(args.output, metadata, summary, confirmations) |
| 212 | + report(args.output, metadata, summary, confirmations) |
| 213 | + |
| 214 | + |
| 215 | +if __name__ == "__main__": |
| 216 | + main() |
0 commit comments