From f44d3d563f68c135f4a2ddcadacd6d7ee239dfea Mon Sep 17 00:00:00 2001 From: Joe Harris Date: Mon, 12 Oct 2026 10:28:25 -0400 Subject: [PATCH 1/3] fix(validation): require complete stream ids and query counts in clean throughput bundles Clean-claimed bundles with a completed throughput phase must carry contiguous stream ids 1..S, and every stream of a TPC-H or TPC-DS bundle must cover the full canonical query set (22 and 99). Streams referencing unknown ids, empty stream results, and non-integer ids are rejected. Adds eleven rejection tests alongside the existing coverage suite. --- benchbox/validation/bundle.py | 151 +++++++++++++++++- .../unit/scripts/test_validate_submission.py | 115 +++++++++++++ 2 files changed, 258 insertions(+), 8 deletions(-) diff --git a/benchbox/validation/bundle.py b/benchbox/validation/bundle.py index 963227ea3f..6372afb8ad 100644 --- a/benchbox/validation/bundle.py +++ b/benchbox/validation/bundle.py @@ -1036,6 +1036,15 @@ def _validate_compliance_section( ) +def _fold_tpcds_variant_observations(observed: set[str]) -> set[str]: + folded = set(observed) + for base_id in ("14", "23", "24", "39"): + folded.discard(base_id) + if {f"{base_id}a", f"{base_id}b"} <= folded: + folded.add(base_id) + return folded + + def _validate_query_coverage( data: dict[str, Any], vr: ValidationResult, @@ -1066,10 +1075,7 @@ def _validate_query_coverage( else: observed.add(normalized_qid) if normalized_id == "tpcds": - for base_id in ("14", "23", "24", "39"): - observed.discard(base_id) - if {f"{base_id}a", f"{base_id}b"} <= observed: - observed.add(base_id) + observed = _fold_tpcds_variant_observations(observed) variant_hint = "" if normalized_id == "tpchavoc": platform = data.get("platform") @@ -1091,10 +1097,7 @@ def _validate_query_coverage( if len(missing_variants) > 12: shown += f", … (+{len(missing_variants) - 12} more)" variant_hint = f" (missing successful or documented-skip variants: {shown})" - missing = sorted( - canonical - observed, - key=lambda s: [int(part) if part.isdigit() else part for part in re.split(r"(\d+)", s)], - ) + missing = sorted(canonical - observed, key=_canonical_id_sort_key) if missing: covered = len(observed & canonical) shown = ", ".join(missing[:12]) @@ -1658,6 +1661,137 @@ def _validate_queries_section(queries: Any, version: Any, vr: ValidationResult) vr.error("queries must contain at least one positive measurement timing") +_THROUGHPUT_STREAM_QUERY_BENCHMARKS = frozenset({"tpch", "tpcds"}) + + +def _canonical_id_sort_key(value: str) -> list[int | str]: + return [int(part) if part.isdigit() else part for part in re.split(r"(\d+)", value)] + + +def _coerce_stream_id(raw_id: Any) -> int | None: + if isinstance(raw_id, bool): + return None + if isinstance(raw_id, int): + return raw_id + if isinstance(raw_id, float) and raw_id.is_integer(): + return int(raw_id) + if isinstance(raw_id, str) and raw_id.strip().isdigit(): + return int(raw_id.strip()) + return None + + +def _completed_throughput_block(data: dict[str, Any]) -> dict[str, Any] | None: + phases = data.get("phases") + if not isinstance(phases, dict): + return None + block = phases.get("throughput_test") + if not isinstance(block, dict): + return None + if str(block.get("status") or "").upper() != "COMPLETED": + return None + return block + + +def _validated_throughput_stream_ids(streams: Any, vr: ValidationResult) -> list[int] | None: + if not isinstance(streams, list) or not streams: + vr.error("phases.throughput_test.status='COMPLETED' requires a non-empty stream_results list") + return None + stream_ids: list[int] = [] + for entry in streams: + raw_id = entry.get("stream_id") if isinstance(entry, dict) else None + stream_id = _coerce_stream_id(raw_id) + if stream_id is None: + vr.error(f"throughput stream_id must be an integer, got {raw_id!r}") + return None + stream_ids.append(stream_id) + if sorted(stream_ids) != list(range(1, len(stream_ids) + 1)): + vr.error(f"throughput stream_ids must be contiguous 1..{len(stream_ids)}, got {sorted(stream_ids)}") + return None + return stream_ids + + +def _throughput_measurement_queries(queries: Any) -> list[dict[str, Any]]: + if not isinstance(queries, list): + return [] + selected: list[dict[str, Any]] = [] + for query in queries: + if not isinstance(query, dict): + continue + if not isinstance(query.get("status"), str): + continue + if query["status"].upper() not in {"SUCCESS", "PASS"}: + continue + if str(query.get("run_type") or "measurement").strip().lower() != "measurement": + continue + if _normalize_coverage_query_id(query.get("id")) is None: + continue + selected.append(query) + return selected + + +def _grouped_throughput_observations(queries: list[dict[str, Any]], vr: ValidationResult) -> dict[int, set[str]] | None: + observed_by_stream: dict[int, set[str]] = {} + for query in queries: + stream_key = _coerce_stream_id(query.get("stream", query.get("stream_id", 0))) + if stream_key is None: + vr.error(f"throughput query stream must be an integer, got {query.get('stream')!r}") + return None + normalized_query = _normalize_coverage_query_id(query.get("id")) + if normalized_query is None: + continue + observed_by_stream.setdefault(stream_key, set()).add(normalized_query) + return observed_by_stream + + +def _validate_throughput_stream_coverage( + stream_ids: list[int], + observed_by_stream: dict[int, set[str]], + benchmark_id: str, + vr: ValidationResult, +) -> None: + canonical = CANONICAL_LOGICAL_QUERY_IDS.get(benchmark_id) + if not canonical: + return + if benchmark_id == "tpcds": + observed_by_stream = {key: _fold_tpcds_variant_observations(value) for key, value in observed_by_stream.items()} + known_ids = set(stream_ids) + for stream_key in sorted(observed_by_stream): + if stream_key not in known_ids: + vr.error(f"throughput query references unknown stream_id {stream_key}") + for stream_id in sorted(known_ids): + missing = sorted( + canonical - observed_by_stream.get(stream_id, set()), + key=_canonical_id_sort_key, + ) + if not missing: + continue + shown = ", ".join(missing[:12]) + if len(missing) > 12: + shown += f", … (+{len(missing) - 12} more)" + vr.error( + f"throughput stream {stream_id} covers {len(canonical) - len(missing)} of {len(canonical)}" + f" canonical queries (missing: {shown})" + ) + + +def _validate_throughput_stream_completeness(data: dict[str, Any], vr: ValidationResult) -> None: + block = _completed_throughput_block(data) + if block is None: + return + stream_ids = _validated_throughput_stream_ids(block.get("stream_results"), vr) + if stream_ids is None: + return + benchmark = data.get("benchmark") + benchmark_id = benchmark.get("id") if isinstance(benchmark, dict) else None + normalized_benchmark = benchmark_id.strip().casefold() if isinstance(benchmark_id, str) else None + if normalized_benchmark not in _THROUGHPUT_STREAM_QUERY_BENCHMARKS: + return + observed_by_stream = _grouped_throughput_observations(_throughput_measurement_queries(data.get("queries")), vr) + if observed_by_stream is None: + return + _validate_throughput_stream_coverage(stream_ids, observed_by_stream, normalized_benchmark, vr) + + def _validate_execution_consistency(data: dict[str, Any], vr: ValidationResult) -> None: summary = data.get("summary") if not isinstance(summary, dict) or _normalize_status(summary.get("validation")) != PUBLIC_CLEAN_VALIDATION_STATUS: @@ -1680,6 +1814,7 @@ def _validate_execution_consistency(data: dict[str, Any], vr: ValidationResult) vr.error( f"summary.validation='passed' contradicts queries[{i}].row_count_validation.status={rcv_status!r}" ) + _validate_throughput_stream_completeness(data, vr) def _validate_validation_phase_consistency(data: dict[str, Any], vr: ValidationResult) -> None: diff --git a/tests/unit/scripts/test_validate_submission.py b/tests/unit/scripts/test_validate_submission.py index 1b4cb39902..f65932cfa4 100644 --- a/tests/unit/scripts/test_validate_submission.py +++ b/tests/unit/scripts/test_validate_submission.py @@ -11,6 +11,7 @@ SUBMISSION_NOTES_MAX_LEN, ValidationResult, _validate_bundle, + _validate_execution_consistency, _validate_manifest_hash, _validate_manifest_provenance, discover_bundles, @@ -1558,6 +1559,120 @@ def test_canonical_membership_allows_extra_ids(self): _validate_bundle(data, vr) assert vr.ok, vr.errors + class TestThroughputStreamCompleteness: + def _bundle(self, stream_ids, queries_by_stream, benchmark="tpch", status="COMPLETED"): + return { + "summary": {"validation": "passed"}, + "benchmark": {"id": benchmark}, + "phases": { + "throughput_test": { + "status": status, + "stream_results": [{"stream_id": sid, "success": True} for sid in stream_ids], + } + }, + "queries": [ + {"id": qid, "status": "SUCCESS", "run_type": "measurement", "stream": sid} + for sid, qids in queries_by_stream.items() + for qid in qids + ], + } + + def _full_tpch(self): + return [f"Q{i}" for i in range(1, 23)] + + def test_contiguous_streams_with_full_coverage_pass(self): + full = self._full_tpch() + data = self._bundle([1, 2], {1: full, 2: full}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + + def test_non_contiguous_stream_ids_rejected(self): + full = self._full_tpch() + data = self._bundle([1, 3], {1: full, 3: full}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("contiguous 1..2" in error for error in vr.errors) + + def test_stream_missing_queries_rejected(self): + data = self._bundle([1, 2], {1: self._full_tpch(), 2: self._full_tpch()[:21]}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("throughput stream 2 covers 21 of 22" in error for error in vr.errors) + + def test_query_referencing_unknown_stream_rejected(self): + full = self._full_tpch() + data = self._bundle([1, 2], {1: full, 2: full, 9: full}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("unknown stream_id 9" in error for error in vr.errors) + + def test_empty_stream_results_rejected(self): + data = self._bundle([], {}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("non-empty stream_results" in error for error in vr.errors) + + def test_non_integer_stream_id_rejected(self): + data = self._bundle([1, 2], {1: self._full_tpch(), 2: self._full_tpch()}) + data["phases"]["throughput_test"]["stream_results"][1] = {"success": True} + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("must be an integer" in error for error in vr.errors) + + def test_non_clean_bundle_skips_stream_checks(self): + full = self._full_tpch() + data = self._bundle([1, 3], {1: full}) + data["summary"]["validation"] = "failed" + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + + def test_non_completed_phase_skips_stream_checks(self): + data = self._bundle([1, 3], {}, status="FAILED") + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + + def test_other_benchmark_skips_query_counts_but_checks_contiguity(self): + data = self._bundle([1, 2], {1: ["Q1"], 2: ["Q1"]}, benchmark="ssb") + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + data = self._bundle([1, 3], {1: ["Q1"], 3: ["Q1"]}, benchmark="ssb") + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("contiguous 1..2" in error for error in vr.errors) + + def test_tpcds_full_canonical_stream_passes(self): + data = self._bundle([1], {1: self._tpcds_variant_ids()}, benchmark="tpcds") + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + + def test_tpcds_short_stream_rejected(self): + data = self._bundle([1], {1: self._tpcds_variant_ids()[:-9]}, benchmark="TPCDS") + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("of 99" in error and "canonical queries" in error for error in vr.errors) + + def _tpcds_variant_ids(self): + ids: list[str] = [] + canonical = bundle_module.CANONICAL_LOGICAL_QUERY_IDS["tpcds"] + for qid in sorted(canonical, key=int): + if qid in ("14", "23", "24", "39"): + ids += [f"{qid}a", f"{qid}b"] + else: + ids.append(qid) + return ids + def test_tpchavoc_requires_every_declared_variant(self): from benchbox.validation.bundle import TPCHAVOC_CANONICAL_VARIANTS From 4bc5127cfd437ea421f32a792d16c4698aa8700c Mon Sep 17 00:00:00 2001 From: Joe Harris Date: Mon, 12 Oct 2026 18:54:51 -0400 Subject: [PATCH 2/3] fix(validation): address oracle review on throughput stream completeness Ignore power-phase stream-0 queries when grouping throughput observations, require usable measurement timings per stream, reject completed phases with unsuccessful streams, and enforce the two-stream TPC minimum for TPC-H and TPC-DS coverage. --- benchbox/validation/bundle.py | 19 +++++++- .../unit/scripts/test_validate_submission.py | 43 +++++++++++++++++-- 2 files changed, 58 insertions(+), 4 deletions(-) diff --git a/benchbox/validation/bundle.py b/benchbox/validation/bundle.py index 6372afb8ad..3248748001 100644 --- a/benchbox/validation/bundle.py +++ b/benchbox/validation/bundle.py @@ -12,6 +12,8 @@ from pathlib import Path from typing import Any +from benchbox.core.schemas import MIN_THROUGHPUT_STREAMS + def _load_schema_policy_helpers(): try: @@ -1662,6 +1664,7 @@ def _validate_queries_section(queries: Any, version: Any, vr: ValidationResult) _THROUGHPUT_STREAM_QUERY_BENCHMARKS = frozenset({"tpch", "tpcds"}) +_POWER_PHASE_STREAM_ID = 0 def _canonical_id_sort_key(value: str) -> list[int | str]: @@ -1703,6 +1706,9 @@ def _validated_throughput_stream_ids(streams: Any, vr: ValidationResult) -> list if stream_id is None: vr.error(f"throughput stream_id must be an integer, got {raw_id!r}") return None + if not isinstance(entry, dict) or entry.get("success") is not True: + vr.error(f"throughput stream {stream_id} did not succeed") + return None stream_ids.append(stream_id) if sorted(stream_ids) != list(range(1, len(stream_ids) + 1)): vr.error(f"throughput stream_ids must be contiguous 1..{len(stream_ids)}, got {sorted(stream_ids)}") @@ -1719,7 +1725,13 @@ def _throughput_measurement_queries(queries: Any) -> list[dict[str, Any]]: continue if not isinstance(query.get("status"), str): continue - if query["status"].upper() not in {"SUCCESS", "PASS"}: + if query["status"].upper() not in _PASS_TIMING_STATUSES: + continue + try: + usable_ms = float(query.get("ms")) + except (TypeError, ValueError): + continue + if not math.isfinite(usable_ms) or usable_ms < 1.0: continue if str(query.get("run_type") or "measurement").strip().lower() != "measurement": continue @@ -1736,6 +1748,8 @@ def _grouped_throughput_observations(queries: list[dict[str, Any]], vr: Validati if stream_key is None: vr.error(f"throughput query stream must be an integer, got {query.get('stream')!r}") return None + if stream_key == _POWER_PHASE_STREAM_ID: + continue normalized_query = _normalize_coverage_query_id(query.get("id")) if normalized_query is None: continue @@ -1752,6 +1766,9 @@ def _validate_throughput_stream_coverage( canonical = CANONICAL_LOGICAL_QUERY_IDS.get(benchmark_id) if not canonical: return + if len(stream_ids) < MIN_THROUGHPUT_STREAMS: + vr.error(f"throughput requires at least {MIN_THROUGHPUT_STREAMS} streams, got {len(stream_ids)}") + return if benchmark_id == "tpcds": observed_by_stream = {key: _fold_tpcds_variant_observations(value) for key, value in observed_by_stream.items()} known_ids = set(stream_ids) diff --git a/tests/unit/scripts/test_validate_submission.py b/tests/unit/scripts/test_validate_submission.py index f65932cfa4..62b60e0fbe 100644 --- a/tests/unit/scripts/test_validate_submission.py +++ b/tests/unit/scripts/test_validate_submission.py @@ -1571,7 +1571,7 @@ def _bundle(self, stream_ids, queries_by_stream, benchmark="tpch", status="COMPL } }, "queries": [ - {"id": qid, "status": "SUCCESS", "run_type": "measurement", "stream": sid} + {"id": qid, "status": "SUCCESS", "run_type": "measurement", "ms": 100, "stream": sid} for sid, qids in queries_by_stream.items() for qid in qids ], @@ -1651,13 +1651,50 @@ def test_other_benchmark_skips_query_counts_but_checks_contiguity(self): assert any("contiguous 1..2" in error for error in vr.errors) def test_tpcds_full_canonical_stream_passes(self): - data = self._bundle([1], {1: self._tpcds_variant_ids()}, benchmark="tpcds") + variant_ids = self._tpcds_variant_ids() + data = self._bundle([1, 2], {1: variant_ids, 2: variant_ids}, benchmark="tpcds") vr = ValidationResult("test") _validate_execution_consistency(data, vr) assert vr.ok, vr.errors + def test_power_stream_zero_queries_ignored(self): + full = self._full_tpch() + power = [{"id": qid, "status": "SUCCESS", "run_type": "measurement", "ms": 50, "stream": 0} for qid in full] + data = self._bundle([1, 2], {1: full, 2: full}) + data["queries"].extend(power) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert vr.ok, vr.errors + + def test_query_without_usable_timing_not_counted(self): + full = self._full_tpch() + data = self._bundle([1, 2], {1: full, 2: full}) + data["queries"] = [q for q in data["queries"] if not (q["stream"] == 2 and q["id"] == "Q22")] + data["queries"].append({"id": "Q22", "status": "SUCCESS", "run_type": "measurement", "stream": 2}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("throughput stream 2 covers 21 of 22" in error for error in vr.errors) + + def test_unsuccessful_stream_rejected(self): + full = self._full_tpch() + data = self._bundle([1, 2], {1: full, 2: full}) + data["phases"]["throughput_test"]["stream_results"][1]["success"] = False + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("throughput stream 2 did not succeed" in error for error in vr.errors) + + def test_single_stream_rejected_for_tpch(self): + data = self._bundle([1], {1: self._full_tpch()}) + vr = ValidationResult("test") + _validate_execution_consistency(data, vr) + assert not vr.ok + assert any("at least 2 streams" in error for error in vr.errors) + def test_tpcds_short_stream_rejected(self): - data = self._bundle([1], {1: self._tpcds_variant_ids()[:-9]}, benchmark="TPCDS") + variant_ids = self._tpcds_variant_ids() + data = self._bundle([1, 2], {1: variant_ids, 2: variant_ids[:-9]}, benchmark="TPCDS") vr = ValidationResult("test") _validate_execution_consistency(data, vr) assert not vr.ok From 9b75926d336ad5c638de4d91bd8ac89a238c05c7 Mon Sep 17 00:00:00 2001 From: Joe Harris Date: Mon, 12 Oct 2026 21:20:35 -0400 Subject: [PATCH 3/3] fix(validation): lazy-load stream minimum for isolated slim checkouts The publication parity harness replays validation from a slim checkout containing only the validator runtime files. Import the TPC stream minimum defensively with the same value as fallback so parity holds. --- benchbox/validation/bundle.py | 7 +++++-- tests/unit/scripts/test_validate_submission.py | 9 +++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/benchbox/validation/bundle.py b/benchbox/validation/bundle.py index 3248748001..e174ffbe31 100644 --- a/benchbox/validation/bundle.py +++ b/benchbox/validation/bundle.py @@ -12,8 +12,6 @@ from pathlib import Path from typing import Any -from benchbox.core.schemas import MIN_THROUGHPUT_STREAMS - def _load_schema_policy_helpers(): try: @@ -44,6 +42,11 @@ def _load_schema_policy_helpers(): FUNDING_SOURCES = ("employer", "personal", "free-trial", "vendor-sponsored", "grant", "unspecified") RESULT_SOURCES = ("internal", "community", "vendor") +try: + from benchbox.core.schemas import MIN_THROUGHPUT_STREAMS +except ImportError: + MIN_THROUGHPUT_STREAMS = 2 + def _load_bundle_failed_query_count(): helper_path = Path(__file__).resolve().parents[1] / "core" / "results" / "query_status.py" diff --git a/tests/unit/scripts/test_validate_submission.py b/tests/unit/scripts/test_validate_submission.py index 62b60e0fbe..2d8e620282 100644 --- a/tests/unit/scripts/test_validate_submission.py +++ b/tests/unit/scripts/test_validate_submission.py @@ -1692,6 +1692,15 @@ def test_single_stream_rejected_for_tpch(self): assert not vr.ok assert any("at least 2 streams" in error for error in vr.errors) + def test_stream_minimum_falls_back_without_schemas_package(self, monkeypatch): + import importlib + import sys + + monkeypatch.setitem(sys.modules, "benchbox.core.schemas", None) + monkeypatch.delitem(sys.modules, "benchbox.validation.bundle") + reloaded = importlib.import_module("benchbox.validation.bundle") + assert reloaded.MIN_THROUGHPUT_STREAMS == 2 + def test_tpcds_short_stream_rejected(self): variant_ids = self._tpcds_variant_ids() data = self._bundle([1, 2], {1: variant_ids, 2: variant_ids[:-9]}, benchmark="TPCDS")