From 321488ed9c60dca20070d24a27b5f0bf0e777dae Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Mon, 7 Sep 2026 04:26:20 +0530 Subject: [PATCH 01/14] feat(ingest): additive schema-v2 write path for test_cases/test_case_runs v1 tables keep being written exactly as before; this adds the two v2 tables alongside them, guarded on both existing so the script is safe to deploy before the migration lands. One v2 table pair serves all three products, discriminated by a `component` column that replaces the v1 table-name prefix. Both ids are DERIVED, never minted, so this script and the orchestrator agree on the join key with no threading contract -- which is the only thing that makes the tables joinable at all: v1 minted four unrelated identity schemes and artifact_results.run_id consequently joined test_runs.run_id in 2 of 1,266 rows. The recipe must stay byte-identical to spyre-frameworks pipelines/lib/run_identity.py, because an orphaned row is indistinguishable from "no tests ran"; that module's header carries the full rule list and the golden values, and its test suite pins them. run_uid is hashed from the leg's own CI coordinate -- --gha-run-id when GHA dispatched it, else the new --jenkins-run-key -- which is the same value the orchestrator hashes on its side. tags is an Array, not a Map: testtype carries up to 5 values on 91.7% of cases, so a Map would silently keep one and drop the rest. Sorted before hashing. Ingest is idempotent by a skip-if-run_uid-already-present check. That is a hard requirement, not a nicety: test_case_runs is a plain MergeTree with no dedup key and v2 dropped the stored counters because they are derived from these rows, so a double ingest would silently double a leg's counts. Dropped from the v2 path: filename, suite_name, runner_run_id and every stored counter -- all derivable. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 232 ++++++++++++++++++++++ 1 file changed, 232 insertions(+) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 52dd7f1b..7931deba 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -326,6 +326,209 @@ def _threaded_run_id(args) -> str: return "" +# --------------------------------------------------------------------------- +# ── SCHEMA v2: test_cases + test_case_runs ───────────────────────────────── +# +# ADDITIVE. Everything above still writes the v1 tables exactly as before; this +# path writes the two v2 tables alongside them and is skipped entirely if they do +# not exist, so the script is safe to deploy before the v2 migration lands. +# +# Both ids are DERIVED, never minted. Four writers (this script, the two sibling +# product ingests, and the orchestrator's pushToClickhouse.pushArtifactResult) +# compute them independently with no threading contract -- which is the only thing +# that makes the tables joinable: v1 minted four unrelated identity schemes and +# artifact_results.run_id consequently joined test_runs.run_id in 2 of 1,266 rows. +# +# BYTE-EXACTNESS IS THE CONTRACT. Disagree about the namespace, the separator, the +# field order or the normalisation and you mint a different uuid for the same row -- +# and an orphaned row is indistinguishable from "no tests ran", so the failure is +# silent. The reference implementation, its rule list and the golden values every +# port must reproduce live in spyre-frameworks pipelines/lib/run_identity.py and +# pipelines/lib/test_run_identity.py. Keep this block in sync with it. +# --------------------------------------------------------------------------- + +# The product this script ingests for. Replaces v1's hf_/si_ table-name prefixes: one +# v2 table pair serves all three products, discriminated by this column. It is also a +# test_case_id hash input, so it cannot drift from the identity it is stamped on. +V2_COMPONENT = "hf-adapters" + +V2_NAMESPACE = uuid.uuid5(uuid.NAMESPACE_DNS, "clickhouse-v2.spyre.ibm.com") +V2_SEP = "|" + + +def _v2_norm(value) -> str: + """Canonical scalar form. Lowercasing is not cosmetic: the same tier arrives as + 'Regression' from a Jenkins parameter and 'regression' from a GHA input.""" + return ("" if value is None else str(value)).strip().lower() + + +def v2_canonical_arch(arch) -> str: + """amd64/x86/x86-64 all mean x86_64 -- a leg labelled 'amd64' by Jenkins and + 'x86_64' by GHA is ONE leg, and must hash as one.""" + a = _v2_norm(arch) + return "x86_64" if a in ("amd64", "x86", "x86-64", "x86_64") else a + + +def v2_run_uid(source: str, external_run_id: str, arch: str, test_type: str) -> str: + """Identity of one TEST-EXECUTION LEG: (source, external_run_id, arch, test_type). + + arch and test_type are IN the key because the real execution grain measured + (run, arch, tier) at 19,867 legs under 16,381 CI runs. external_run_id is a + STRING: typed numerically, every Jenkins leg would be 0 and collide into one id. + Returns '' when a field is missing -- an all-defaults hash is a real uuid that + every incomplete leg would share, which is worse than a blank. + """ + fields = (source, external_run_id, arch, test_type) + if not all(_v2_norm(f) for f in fields): + return "" + return str( + uuid.uuid5( + V2_NAMESPACE, + V2_SEP.join( + ( + _v2_norm(source), + _v2_norm(external_run_id), + v2_canonical_arch(arch), + _v2_norm(test_type), + ) + ), + ) + ) + + +def v2_test_case_id(component: str, classname: str, name: str, tags) -> str: + """Content identity of a TEST, so the same test reconciles across runs. v1 minted + uuid4 per row: 37,322,701 identities for 58,711 distinct (classname, name) pairs. + + `tags` are deduped and SORTED -- they are a set and source order is incidental, + so an unsorted join makes two writers disagree about the same test. They are + INSIDE the hash, so re-tagging mints a new identity; trend queries must + therefore group on (component, classname, name), never on test_case_id. + """ + norm = sorted({t for t in (_v2_norm(x) for x in (tags or [])) if t}) + return str( + uuid.uuid5( + V2_NAMESPACE, + V2_SEP.join( + (_v2_norm(component), _v2_norm(classname), _v2_norm(name), ",".join(norm)) + ), + ) + ) + + +def v2_tags_for_case(case: dict) -> list: + """The case's tags as an ARRAY of `namespace__value` strings. + + Array, not Map: `testtype` carries up to 5 values on 91.7% of cases, so a Map + would silently keep one and drop the rest. The v1 shape is a (prop_name, + prop_value) list where the only prop_name is literally 'tag' and the real + key is encoded inside the value -- so the VALUE is the tag. + """ + tags = set() + for pname, pvalue in case.get("properties", []) or []: + if pname == "tag": + if pvalue: + tags.add(pvalue) + elif "__" in pname: + # Some emitters put the namespace__value in the property NAME instead. + tags.add(pname) + return sorted(tags) + + +def v2_source_and_external_run_id(args, run_id: str): + """(source, external_run_id) for this leg, from whichever CI dispatched it. + + A numeric --gha-run-id means GHA dispatched it. Otherwise the leg is + Jenkins-dispatched and its own externalizable id ('folder/job#123') is the run + coordinate -- the SAME value the orchestrator hashes on its side of the join, so + neither side has to thread a minted uuid. + `source` is required precisely because a GHA run id and a Jenkins build number + share a number space. + """ + gha = (getattr(args, "gha_run_id", "") or "").strip() + if gha: + try: + int(gha) + return "gha", gha + except (ValueError, TypeError): + pass + jenkins_key = (getattr(args, "jenkins_run_key", "") or "").strip() + if jenkins_key: + return "jenkins", jenkins_key + # No CI coordinate at all: fall back to the run uuid so the rows are still + # self-consistent and joinable WITHIN this ingest, just not to an artifact. + return "local", run_id + + +def v2_tables_present(client) -> bool: + """v2 write path is skipped unless BOTH tables exist, so this script can be + deployed before the migration without erroring on every run.""" + return all( + bool(client.command(f"EXISTS TABLE {t}")) + for t in ("test_cases", "test_case_runs") + ) + + +def v2_already_ingested(client, run_uid: str, component: str) -> bool: + """test_case_runs is a plain MergeTree with no dedup key, so a double ingest of one + leg DOUBLES its counts -- and the v2 schema dropped the stored counters precisely + because they are derived from these rows. This check is what keeps that correct. + Scoped by component as well as run_uid to hit the ORDER BY prefix.""" + rows = client.query( + "SELECT count() FROM test_case_runs " + "WHERE component = {component:String} AND run_uid = {run_uid:UUID}", + parameters={"component": component, "run_uid": run_uid}, + ).result_rows + return bool(rows and rows[0][0] > 0) + + +def insert_v2(client, component: str, run_uid: str, cases: list) -> int: + """Write test_cases (identity) + test_case_runs (outcome) for one leg. + + Dropped from v2 deliberately: filename, suite_name, runner_run_id, and every + stored counter -- all derivable, and a stored counter invites drift. + """ + if not cases: + return 0 + ident_rows, run_rows = {}, [] + for c in cases: + tags = v2_tags_for_case(c) + classname, name = c.get("classname", ""), c.get("name", "") + tcid = v2_test_case_id(component, classname, name, tags) + # Deduped by id within the leg: identical identity rows are one fact, and + # test_cases is a plain MergeTree (a content hash re-writes an identical row, + # so collapsing would only be cosmetic -- but writing it N times is not). + ident_rows[tcid] = [tcid, component, classname, name, tags] + run_rows.append( + [ + run_uid, + tcid, + component, + c.get("status", ""), + float(c.get("duration_s", 0) or 0), + (c.get("fail_message") or "")[:8192], + ] + ) + client.insert( + "test_cases", + list(ident_rows.values()), + column_names=["test_case_id", "component", "classname", "name", "tags"], + ) + client.insert( + "test_case_runs", + run_rows, + column_names=[ + "run_uid", + "test_case_id", + "component", + "status", + "duration_s", + "fail_message", + ], + ) + return len(run_rows) + + def main(): parser = argparse.ArgumentParser() parser.add_argument("--xml-dir", default=None) @@ -337,6 +540,13 @@ def main(): parser.add_argument("--gha-run-id", default="") parser.add_argument("--triggered-at", default="") parser.add_argument("--pr-number", default="") + parser.add_argument( + "--jenkins-run-key", + default="", + help="This leg's own Jenkins externalizable id, e.g. 'Spyre/component-build#417'. " + "Hashed into the schema-v2 run_uid, which is how the orchestrator's " + "artifact_results row and these per-case rows join without threading a uuid.", + ) parser.add_argument( "--trigger-type", default="", @@ -444,6 +654,28 @@ def main(): insert_run(client, run_id, run, args) insert_cases(client, run_id, cases, workflow=args.workflow) insert_properties(client, run_id, cases) + # v2 tables, alongside v1. Guarded so this script still runs against a + # database where the migration has not landed. + if v2_tables_present(client): + _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) + _v2_tier = (getattr(args, "trigger_type", "") or "").strip() + _v2_arch = (args.platform or run.get("platform") or "").strip() + _v2_run_uid = v2_run_uid(_v2_source, _v2_ext, _v2_arch, _v2_tier) + if not _v2_run_uid: + # Loud, because a blank run_uid means these cases reach v2 unjoinable + # to any artifact -- and that reads downstream as "no tests ran". + print( + f" [warn] v2 skipped: run_uid not derivable " + f"(source={_v2_source} ext={_v2_ext!r} arch={_v2_arch!r} " + f"tier={_v2_tier!r}); --trigger-type is the field usually missing", + file=sys.stderr, + ) + elif v2_already_ingested(client, _v2_run_uid, V2_COMPONENT): + print(f" v2: already ingested run_uid={_v2_run_uid} — skipping") + else: + _n = insert_v2(client, V2_COMPONENT, _v2_run_uid, cases) + print(f" v2: {_n} test_case_runs under run_uid={_v2_run_uid}") + total_cases += len(cases) print( From c4303f7586e090b20a404d1bfeea5282bd0a7c88 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Mon, 7 Sep 2026 05:01:03 +0530 Subject: [PATCH 02/14] fix(ingest): refuse a test_case_id derived from incomplete input v2_run_uid already guarded all four of its fields, but v2_test_case_id did not: an empty component or name still hashed to a real, stable uuid, so every such case minted the SAME id and would have collided into one identity rather than merely being orphaned. Rows that collide land looking joined, which is the failure the derived-id design exists to prevent. v2_test_case_id now returns '' without component and name -- the v2 table's CONSTRAINTs reject those as '' regardless -- and insert_v2 skips such a case with a warning instead of writing it. classname stays optional: a module-level test legitimately has none. Matches the same fix to the reference implementation in spyre-frameworks pipelines/lib/run_identity.py, where negative tests are now pinned per field. Goldens unchanged. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 7931deba..988347a1 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -405,6 +405,12 @@ def v2_test_case_id(component: str, classname: str, name: str, tags) -> str: INSIDE the hash, so re-tagging mints a new identity; trend queries must therefore group on (component, classname, name), never on test_case_id. """ + if not (_v2_norm(component) and _v2_norm(name)): + # Same collision hazard as v2_run_uid: an empty field still hashes to a real, + # stable uuid that every other such case shares. The v2 table's CONSTRAINTs + # reject component='' / name='' anyway. classname is legitimately empty for a + # module-level test, so it is NOT required. + return "" norm = sorted({t for t in (_v2_norm(x) for x in (tags or [])) if t}) return str( uuid.uuid5( @@ -491,10 +497,17 @@ def insert_v2(client, component: str, run_uid: str, cases: list) -> int: if not cases: return 0 ident_rows, run_rows = {}, [] + + skipped_unidentifiable = 0 for c in cases: tags = v2_tags_for_case(c) classname, name = c.get("classname", ""), c.get("name", "") tcid = v2_test_case_id(component, classname, name, tags) + if not tcid: + # Refused identity: writing the row anyway would collide it with every + # other unidentifiable case rather than merely orphaning it. + skipped_unidentifiable += 1 + continue # Deduped by id within the leg: identical identity rows are one fact, and # test_cases is a plain MergeTree (a content hash re-writes an identical row, # so collapsing would only be cosmetic -- but writing it N times is not). @@ -526,6 +539,12 @@ def insert_v2(client, component: str, run_uid: str, cases: list) -> int: "fail_message", ], ) + if skipped_unidentifiable: + print( + f" [warn] v2: {skipped_unidentifiable} case(s) skipped -- no derivable " + f"test_case_id (empty name?); they would have collided, not merely orphaned", + file=sys.stderr, + ) return len(run_rows) From 6fc37cff6f2fc8ea364aa24f49036a03243f8aca Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 01:49:56 +0530 Subject: [PATCH 03/14] refactor(ingest): run_uid -> run_id for the v2 identity Brings this writer in line with the schema. The v2 column is run_id -- _id is the single rule for every v2 identity (artifact_id, run_id, test_case_id, benchmark_id) -- so writing run_uid would have inserted a key no table has, silently dropped by input_format_skip_unknown_fields, leaving every row unjoinable to its per-case detail. Renamed by word-boundary token, NOT a blind substitution: this file carries BOTH identities. v1's run_id is a live column and a parameter of insert_run/_runner_run_id (36 occurrences), and all of those had to survive untouched while the 14 v2 ones changed. Both paths verified present afterwards. Verified the hash did not move, only the name: v2_run_id('gha','12345','amd64','integration') returns dab2a67f-14bf-53be-b6e4-fc9642086e47, byte-identical to spyre-frameworks pipelines/lib/run_identity.py. A writer that disagrees about the namespace or field order orphans every row, which is the whole reason these recipes are pinned. arch still folds (amd64 == x86_64) and an incomplete input is still refused rather than hashed. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 36 +++++++++++------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 988347a1..17f86d1f 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -369,7 +369,7 @@ def v2_canonical_arch(arch) -> str: return "x86_64" if a in ("amd64", "x86", "x86-64", "x86_64") else a -def v2_run_uid(source: str, external_run_id: str, arch: str, test_type: str) -> str: +def v2_run_id(source: str, external_run_id: str, arch: str, test_type: str) -> str: """Identity of one TEST-EXECUTION LEG: (source, external_run_id, arch, test_type). arch and test_type are IN the key because the real execution grain measured @@ -406,7 +406,7 @@ def v2_test_case_id(component: str, classname: str, name: str, tags) -> str: therefore group on (component, classname, name), never on test_case_id. """ if not (_v2_norm(component) and _v2_norm(name)): - # Same collision hazard as v2_run_uid: an empty field still hashes to a real, + # Same collision hazard as v2_run_id: an empty field still hashes to a real, # stable uuid that every other such case shares. The v2 table's CONSTRAINTs # reject component='' / name='' anyway. classname is legitimately empty for a # module-level test, so it is NOT required. @@ -475,20 +475,20 @@ def v2_tables_present(client) -> bool: ) -def v2_already_ingested(client, run_uid: str, component: str) -> bool: +def v2_already_ingested(client, run_id: str, component: str) -> bool: """test_case_runs is a plain MergeTree with no dedup key, so a double ingest of one leg DOUBLES its counts -- and the v2 schema dropped the stored counters precisely because they are derived from these rows. This check is what keeps that correct. - Scoped by component as well as run_uid to hit the ORDER BY prefix.""" + Scoped by component as well as run_id to hit the ORDER BY prefix.""" rows = client.query( "SELECT count() FROM test_case_runs " - "WHERE component = {component:String} AND run_uid = {run_uid:UUID}", - parameters={"component": component, "run_uid": run_uid}, + "WHERE component = {component:String} AND run_id = {run_id:UUID}", + parameters={"component": component, "run_id": run_id}, ).result_rows return bool(rows and rows[0][0] > 0) -def insert_v2(client, component: str, run_uid: str, cases: list) -> int: +def insert_v2(client, component: str, run_id: str, cases: list) -> int: """Write test_cases (identity) + test_case_runs (outcome) for one leg. Dropped from v2 deliberately: filename, suite_name, runner_run_id, and every @@ -514,7 +514,7 @@ def insert_v2(client, component: str, run_uid: str, cases: list) -> int: ident_rows[tcid] = [tcid, component, classname, name, tags] run_rows.append( [ - run_uid, + run_id, tcid, component, c.get("status", ""), @@ -531,7 +531,7 @@ def insert_v2(client, component: str, run_uid: str, cases: list) -> int: "test_case_runs", run_rows, column_names=[ - "run_uid", + "run_id", "test_case_id", "component", "status", @@ -563,7 +563,7 @@ def main(): "--jenkins-run-key", default="", help="This leg's own Jenkins externalizable id, e.g. 'Spyre/component-build#417'. " - "Hashed into the schema-v2 run_uid, which is how the orchestrator's " + "Hashed into the schema-v2 run_id, which is how the orchestrator's " "artifact_results row and these per-case rows join without threading a uuid.", ) parser.add_argument( @@ -679,21 +679,21 @@ def main(): _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) _v2_tier = (getattr(args, "trigger_type", "") or "").strip() _v2_arch = (args.platform or run.get("platform") or "").strip() - _v2_run_uid = v2_run_uid(_v2_source, _v2_ext, _v2_arch, _v2_tier) - if not _v2_run_uid: - # Loud, because a blank run_uid means these cases reach v2 unjoinable + _v2_run_id = v2_run_id(_v2_source, _v2_ext, _v2_arch, _v2_tier) + if not _v2_run_id: + # Loud, because a blank run_id means these cases reach v2 unjoinable # to any artifact -- and that reads downstream as "no tests ran". print( - f" [warn] v2 skipped: run_uid not derivable " + f" [warn] v2 skipped: run_id not derivable " f"(source={_v2_source} ext={_v2_ext!r} arch={_v2_arch!r} " f"tier={_v2_tier!r}); --trigger-type is the field usually missing", file=sys.stderr, ) - elif v2_already_ingested(client, _v2_run_uid, V2_COMPONENT): - print(f" v2: already ingested run_uid={_v2_run_uid} — skipping") + elif v2_already_ingested(client, _v2_run_id, V2_COMPONENT): + print(f" v2: already ingested run_id={_v2_run_id} — skipping") else: - _n = insert_v2(client, V2_COMPONENT, _v2_run_uid, cases) - print(f" v2: {_n} test_case_runs under run_uid={_v2_run_uid}") + _n = insert_v2(client, V2_COMPONENT, _v2_run_id, cases) + print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") total_cases += len(cases) From 680113fe4aafd73194021eeaea4091229a7a0f29 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 14:47:10 +0530 Subject: [PATCH 04/14] feat(ingest): --schema flag to write v1, v2, or both Mirrors torch-spyre. v2 replaces this repo's hf_test_runs/hf_test_cases/hf_run_properties outright -- run_properties becomes test_cases.tags, and the per-product runs/cases collapse into the SHARED test_cases + test_case_runs keyed by component, which is why the spyre-frameworks v2 DDL deliberately defines neither. This script must keep writing both while the product images turn over, since it runs from a BAKED image and old and new images coexist until every one is rebuilt. --schema v1 (default) | v2 | both, also settable via INGEST_SCHEMA so a workflow sets it once for every leg. Default v1 means an un-updated caller behaves exactly as before; our own invocations pass both. No data is ported: v1 rows cannot produce a v2 run_id, since the hash inputs were never recorded per row. New data only. v2 gets its OWN connection via CLICKHOUSE_DB_V2, not v1's with a different database: test_cases exists in both generations with incompatible shapes, so one database binding cannot serve both. Unset means no second connection is opened, so the default path costs nothing. Gating the inserts alone was not enough -- the v1 dedup READS had to be gated too, since they query the per-product runs table, which need not exist in a v2-only target. Also fixed a pre-existing bug the flag exposed: --platform defaulted to '' here while torch-spyre defaults to the host arch. arch is an INPUT to the run_id hash, so an empty value made v2_run_id refuse to derive an id and every v2 row would have landed unjoinable -- observed as '[warn] v2 skipped: run_id not derivable ... arch=""'. Now defaults to the ingest host's arch, which is the machine the suite ran on. The 'Inserted N test cases' line is gated on write_v1: it reports the PARSE count, so under --schema v2 it claimed inserts that never happened. Verified all three modes against real v1 and v2 databases: default writes v1 only, both writes both, and v2 leaves v1 frozen at 2 runs / 4 cases. The shared v2 tables correctly separate the two producers -- component='hf-adapters' 4 rows, 'spyre-inference' 4 rows. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 120 +++++++++++++++------- 1 file changed, 85 insertions(+), 35 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 17f86d1f..265d01b9 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -18,6 +18,7 @@ import argparse import os +import platform as _platform import sys import uuid from collections import Counter @@ -163,17 +164,31 @@ def parse_test_xml(xml_path: Path): # --------------------------------------------------------------------------- -def get_client(): +def get_client(database: str | None = None): return clickhouse_connect.get_client( host=os.environ["CLICKHOUSE_HOST"], port=int(os.environ.get("CLICKHOUSE_PORT", 443)), user=os.environ.get("CLICKHOUSE_USER", "default"), password=os.environ["CLICKHOUSE_PASS"], - database=os.environ.get("CLICKHOUSE_DB", "spyre"), + database=database or os.environ.get("CLICKHOUSE_DB", "spyre"), secure=True, ) +def get_v2_client(): + """A SECOND connection, bound to the v2 database, or None when none is configured. + + v2 is a different DATABASE, not different table names: test_cases exists in both + generations with incompatible shapes, so one `database=` cannot serve both. Returns None + when CLICKHOUSE_DB_V2 is unset, which is what makes --schema v1 (the default) cost + nothing -- no second connection is opened. + """ + db = os.environ.get("CLICKHOUSE_DB_V2", "").strip() + if not db: + return None + return get_client(database=db) + + def insert_run(client, run_id: str, run: dict, args): client.insert( "hf_test_runs", @@ -573,15 +588,39 @@ def main(): ) parser.add_argument( "--platform", - default="", - help="Hardware platform the suite ran on, e.g. x86_64 | s390x | ppc64le", + default=_platform.machine() or "", + help="Hardware platform the suite ran on, e.g. x86_64 | s390x | ppc64le. " + "Defaults to the ingest host's arch, which is the machine the suite ran on. NOT " + "optional for v2: arch is an input to the run_id hash, so an empty value makes " + "v2_run_id refuse to derive an id and every row lands unjoinable.", ) parser.add_argument( "--img-digest", default="", help="Digest of the runner image the suite ran against, if known", ) + # Which schema generation to write. Defaults to v1 ONLY, so an un-updated caller behaves + # exactly as before -- this script runs from a BAKED image, so old and new images coexist + # until every product image is rebuilt. + # + # v1 is not a permanent home: v2 replaces these tables outright. run_properties becomes + # test_cases.tags, and the per-product test_runs/test_cases collapse into the shared + # test_cases + test_case_runs keyed by component -- which is why the spyre-frameworks v2 + # DDL deliberately does not define them. No data is ported: v1 rows cannot produce a v2 + # run_id, since the hash inputs were never recorded per row. New data only. + parser.add_argument( + "--schema", + choices=["v1", "v2", "both"], + default=os.environ.get("INGEST_SCHEMA", "v1"), + help="Which schema generation to write: v1 (default, the legacy per-product tables), " + "v2 (the shared replacement tables only), or both (the migration window). Also " + "settable via INGEST_SCHEMA so a workflow can set it once for every leg.", + ) args = parser.parse_args() + # Resolved once, so the two paths cannot drift into disagreeing about what was asked for. + args.write_v1 = args.schema in ("v1", "both") + args.write_v2 = args.schema in ("v2", "both") + print(f" schema={args.schema} (v1={args.write_v1} v2={args.write_v2})") if args.xml_file: xml_root = Path(args.xml_file).parent @@ -602,6 +641,12 @@ def main(): f"{os.environ['CLICKHOUSE_HOST']}:{os.environ.get('CLICKHOUSE_PORT', 443)} ..." ) client = get_client() + # Separate connection for the v2 tables -- see get_v2_client(). None when + # CLICKHOUSE_DB_V2 is unset, which every v2 site treats as "v2 not configured". + v2client = get_v2_client() if args.write_v2 else None + if args.write_v2 and v2client is None: + print(" WARN --schema asked for v2 but CLICKHOUSE_DB_V2 is unset — v2 rows skipped", + file=sys.stderr) client.command("SELECT 1") print("Connected.\n") @@ -640,29 +685,32 @@ def main(): # Dedup on (run_id, filename): a re-ingest of the SAME test run must be idempotent, # but two distinct runs must never collapse. runner_run_id mirrors run_id for a Jenkins/standalone leg, so it's only an independent signal for a GHA numeric id. runner_run_id = _runner_run_id(args, run_id) - existing = client.query( - "SELECT count() FROM hf_test_runs " - "WHERE run_id = {run_id:String} AND filename = {filename:String}", - parameters={"run_id": run_id, "filename": run["filename"]}, - ) - if ( - existing.result_rows[0][0] == 0 - and runner_run_id - and runner_run_id != run_id - ): - # A GHA re-ingest mints a fresh uuid4, so fall back to the numeric - # run id to keep that path idempotent. + # v1-table reads, so gated on v1 being written. v2 dedups against its own + # table via v2_already_ingested(run_id, component). + if args.write_v1: existing = client.query( - "SELECT count() FROM hf_test_runs WHERE " - "runner_run_id = {runner_run_id:String} AND filename = {filename:String}", - parameters={ - "runner_run_id": runner_run_id, - "filename": run["filename"], - }, + "SELECT count() FROM hf_test_runs " + "WHERE run_id = {run_id:String} AND filename = {filename:String}", + parameters={"run_id": run_id, "filename": run["filename"]}, ) - if existing.result_rows[0][0] > 0: - print(f" Already ingested — skipping {run['filename']}") - continue + if ( + existing.result_rows[0][0] == 0 + and runner_run_id + and runner_run_id != run_id + ): + # A GHA re-ingest mints a fresh uuid4, so fall back to the numeric + # run id to keep that path idempotent. + existing = client.query( + "SELECT count() FROM hf_test_runs WHERE " + "runner_run_id = {runner_run_id:String} AND filename = {filename:String}", + parameters={ + "runner_run_id": runner_run_id, + "filename": run["filename"], + }, + ) + if existing.result_rows[0][0] > 0: + print(f" Already ingested — skipping {run['filename']}") + continue print( f" run_id={run_id} tests={run['total_tests']} " @@ -670,12 +718,13 @@ def main(): f"xpass={run['xpass']} xfail={run['xfail']} skipped={run['skipped']}" ) - insert_run(client, run_id, run, args) - insert_cases(client, run_id, cases, workflow=args.workflow) - insert_properties(client, run_id, cases) + if args.write_v1: + insert_run(client, run_id, run, args) + insert_cases(client, run_id, cases, workflow=args.workflow) + insert_properties(client, run_id, cases) # v2 tables, alongside v1. Guarded so this script still runs against a # database where the migration has not landed. - if v2_tables_present(client): + if v2client is not None and v2_tables_present(v2client): _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) _v2_tier = (getattr(args, "trigger_type", "") or "").strip() _v2_arch = (args.platform or run.get("platform") or "").strip() @@ -689,18 +738,19 @@ def main(): f"tier={_v2_tier!r}); --trigger-type is the field usually missing", file=sys.stderr, ) - elif v2_already_ingested(client, _v2_run_id, V2_COMPONENT): + elif v2_already_ingested(v2client, _v2_run_id, V2_COMPONENT): print(f" v2: already ingested run_id={_v2_run_id} — skipping") else: - _n = insert_v2(client, V2_COMPONENT, _v2_run_id, cases) + _n = insert_v2(v2client, V2_COMPONENT, _v2_run_id, cases) print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") total_cases += len(cases) - print( - f" Inserted {len(cases)} test cases + " - f"{sum(len(c['properties']) for c in cases)} properties" - ) + if args.write_v1: + print( + f" Inserted {len(cases)} test cases + " + f"{sum(len(c['properties']) for c in cases)} properties" + ) print(f"\nDone. {len(xml_files)} file(s) processed.") print(f" Test cases ingested: {total_cases}") From a79f02ee64c8d04a9c98715afc3c40a7f824cc41 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 21:37:32 +0530 Subject: [PATCH 05/14] ci(clickhouse): enable the v2 dual-write in the ingest workflow Turns on the --schema flag this repo already carries. INGEST_SCHEMA=both, so v1 stays authoritative while v2 accumulates the same runs -- the only way v2 gets history, since no data is ported (v1 rows cannot produce a v2 run_id, the hash inputs were never recorded per row). Set at JOB level rather than on the step: the ingest step has no env: block of its own, and neither step in this workflow shadows CLICKHOUSE_DB_V2, INGEST_SCHEMA or CLICKHOUSE_DB -- checked, not assumed. Safe before the secret exists. An unset CLICKHOUSE_DB_V2 makes get_v2_client() return None and open no second connection at all, so the v2 write is a no-op and this workflow behaves exactly as it does today until the secret is populated. Verified end to end against the real dev database, driven by ENV VARS ONLY with no CLI flag, which is exactly how the workflow invokes it: both repos wrote 2 test_case_runs each into spyre_v2_unified, correctly attributed by component (hf-adapters 2, spyre-inference 2). Probe rows removed afterwards. Signed-off-by: Ashok Pon Kumar --- .github/workflows/push-test-results-to-clickhouse.yaml | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/.github/workflows/push-test-results-to-clickhouse.yaml b/.github/workflows/push-test-results-to-clickhouse.yaml index f42d933e..29d5553c 100644 --- a/.github/workflows/push-test-results-to-clickhouse.yaml +++ b/.github/workflows/push-test-results-to-clickhouse.yaml @@ -66,6 +66,15 @@ jobs: CLICKHOUSE_USER: ${{ secrets.CLICKHOUSE_USER }} CLICKHOUSE_PASS: ${{ secrets.CLICKHOUSE_PASS }} CLICKHOUSE_DB: ${{ secrets.CLICKHOUSE_DB }} + # Schema-v2 database. A SEPARATE connection from CLICKHOUSE_DB, because test_cases + # exists in both generations with incompatible shapes, so one database binding cannot + # serve both. Unset/empty makes the v2 write a no-op -- get_v2_client() opens no second + # connection -- so this is safe before the secret exists. + CLICKHOUSE_DB_V2: ${{ secrets.CLICKHOUSE_DB_V2 }} + # Dual-write during the migration window: v1 stays authoritative while v2 accumulates + # the same runs. No data is ported (v1 rows cannot produce a v2 run_id, the hash inputs + # were never recorded per row), so `both` is how v2 gets any history at all. + INGEST_SCHEMA: both # Metadata about the triggering run TRIGGERING_RUN_ID: ${{ github.event.workflow_run.id || inputs.gha_run_id }} From 7b1d1f8d528b52e78b8d13954abb5f7cb1b6387f Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 21:59:12 +0530 Subject: [PATCH 06/14] fix(ingest): contain v2 write failures so they cannot cost a v1 row The v2 block was in no try/except. v1 is still authoritative and v2 is the experiment, yet any v2-side failure (schema drift, a connection reset, a constraint trip) propagated out of the per-file loop -- aborting it and dropping the v1 insert for every REMAINING xml file in the run. The experiment could take down the source of truth. Wraps the block and continues. The v1 calls stay outside the try, so they are never caught up in a v2 failure. Prerequisite for pointing CLICKHOUSE_DB_V2 at a prod database: with an empty secret the v2 path never runs and the gap is unreachable, so this only becomes load-bearing at the moment the secret is populated. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 47 ++++++++++++----------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 265d01b9..b88ef273 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -722,28 +722,31 @@ def main(): insert_run(client, run_id, run, args) insert_cases(client, run_id, cases, workflow=args.workflow) insert_properties(client, run_id, cases) - # v2 tables, alongside v1. Guarded so this script still runs against a - # database where the migration has not landed. - if v2client is not None and v2_tables_present(v2client): - _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) - _v2_tier = (getattr(args, "trigger_type", "") or "").strip() - _v2_arch = (args.platform or run.get("platform") or "").strip() - _v2_run_id = v2_run_id(_v2_source, _v2_ext, _v2_arch, _v2_tier) - if not _v2_run_id: - # Loud, because a blank run_id means these cases reach v2 unjoinable - # to any artifact -- and that reads downstream as "no tests ran". - print( - f" [warn] v2 skipped: run_id not derivable " - f"(source={_v2_source} ext={_v2_ext!r} arch={_v2_arch!r} " - f"tier={_v2_tier!r}); --trigger-type is the field usually missing", - file=sys.stderr, - ) - elif v2_already_ingested(v2client, _v2_run_id, V2_COMPONENT): - print(f" v2: already ingested run_id={_v2_run_id} — skipping") - else: - _n = insert_v2(v2client, V2_COMPONENT, _v2_run_id, cases) - print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") - + # v2 tables, alongside v1. Failure here must never cost a v1 row: v1 is still + # authoritative, so the experimental write is contained rather than allowed to + # abort the loop and drop every remaining file's v1 insert. + try: + if v2client is not None and v2_tables_present(v2client): + _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) + _v2_tier = (getattr(args, "trigger_type", "") or "").strip() + _v2_arch = (args.platform or run.get("platform") or "").strip() + _v2_run_id = v2_run_id(_v2_source, _v2_ext, _v2_arch, _v2_tier) + if not _v2_run_id: + # Loud, because a blank run_id means these cases reach v2 unjoinable + # to any artifact -- and that reads downstream as "no tests ran". + print( + f" [warn] v2 skipped: run_id not derivable " + f"(source={_v2_source} ext={_v2_ext!r} arch={_v2_arch!r} " + f"tier={_v2_tier!r}); --trigger-type is the field usually missing", + file=sys.stderr, + ) + elif v2_already_ingested(v2client, _v2_run_id, V2_COMPONENT): + print(f" v2: already ingested run_id={_v2_run_id} — skipping") + else: + _n = insert_v2(v2client, V2_COMPONENT, _v2_run_id, cases) + print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") + except Exception as _v2_err: + print(f" [warn] v2 write failed, v1 unaffected: {_v2_err!r}", file=sys.stderr) total_cases += len(cases) if args.write_v1: From 70ac6a160fad76492e3bca34310012c4a3ea2a72 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 22:15:01 +0530 Subject: [PATCH 07/14] style(ingest): ruff format Formatting only, from this repo's own pinned ruff. No behaviour change: the v2 write guard is unchanged (verified by AST -- insert_v2 still inside try/except Exception with the v1 calls outside it). Line length differs per repo (88 here vs 100 in spyre-inference), which is why code that passes in one repo fails in another. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index b88ef273..e9e4e316 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -431,7 +431,12 @@ def v2_test_case_id(component: str, classname: str, name: str, tags) -> str: uuid.uuid5( V2_NAMESPACE, V2_SEP.join( - (_v2_norm(component), _v2_norm(classname), _v2_norm(name), ",".join(norm)) + ( + _v2_norm(component), + _v2_norm(classname), + _v2_norm(name), + ",".join(norm), + ) ), ) ) @@ -645,16 +650,17 @@ def main(): # CLICKHOUSE_DB_V2 is unset, which every v2 site treats as "v2 not configured". v2client = get_v2_client() if args.write_v2 else None if args.write_v2 and v2client is None: - print(" WARN --schema asked for v2 but CLICKHOUSE_DB_V2 is unset — v2 rows skipped", - file=sys.stderr) + print( + " WARN --schema asked for v2 but CLICKHOUSE_DB_V2 is unset — v2 rows skipped", + file=sys.stderr, + ) client.command("SELECT 1") print("Connected.\n") db = os.environ.get("CLICKHOUSE_DB", "spyre") if not tables_exist(client, db): print( - f"{db}.hf_test_runs does not exist — nothing to ingest into. " - "Silent no-op." + f"{db}.hf_test_runs does not exist — nothing to ingest into. Silent no-op." ) sys.exit(0) @@ -746,7 +752,9 @@ def main(): _n = insert_v2(v2client, V2_COMPONENT, _v2_run_id, cases) print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") except Exception as _v2_err: - print(f" [warn] v2 write failed, v1 unaffected: {_v2_err!r}", file=sys.stderr) + print( + f" [warn] v2 write failed, v1 unaffected: {_v2_err!r}", file=sys.stderr + ) total_cases += len(cases) if args.write_v1: From b9da64ed63f743b8f0dbd80247d03bfe63c93ded Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Wed, 9 Sep 2026 23:43:19 +0530 Subject: [PATCH 08/14] refactor(ingest): insert v2 rows through a schema model, not positional lists Every v2 insert was a positional list paired with a separate column_names list, and the same four tables were assembled independently in three repos. Nothing tied a row's field order to the column list except the author reading both. Rows are now dicts keyed by column name and ordered by v2_schema, so the order lives in exactly one place and a field cannot land in the wrong column. This is not a bug fix -- the audited state was correct (20/20 inserts column-named, 17/17 arity right, all four v2 column lists matching the DDL). It removes a class of mistake, and it removes the divergence that let a real defect live in two repos and not the third: hf-adapters and spyre-inference never deduped identity rows ACROSS runs, so a case seen in N runs became N rows in test_cases. insert_identities() is now the only path, so all three get it. Also enforced ahead of the server, where a violation is otherwise a rejected insert that pushJUnitXml swallows into a green build: unknown/missing columns, empty NOT-NULL columns, and the DDL's CONSTRAINT chk_status value set. A v1 column set can no longer be handed to the v2 table -- spyre.test_cases has 14 columns and spyre_v2.test_cases has 6, told apart only by which connection is used. A TABLE model, not an ingester class hierarchy: per-repo variation is one constant (COMPONENT), and the file is COPIED rather than imported because the baked-image path runs `uv run --no-project`, leaving no sys.path beyond the script's own directory and no package to install. check_v2_schema_drift.py keeps the copies honest by comparing parsed ASTs, not bytes -- the repos pin different ruff line lengths, so byte equality is unachievable while semantic equality is what matters. Identity functions are deliberately untouched: run_id and test_case_id arrive already computed. Changing them would re-key the warehouse and silently break v2_already_ingested dedup, producing duplicate rows rather than an error. Verified: the model emits BYTE-IDENTICAL (column_names, rows) pairs to the pre-refactor writer for all three repos on a fixture exercising dedup, tag reordering and a failure -- since only the pair construction changed, pair equality is a complete proof for the insert layer. The pinned identity goldens are unmoved (namespace cb0af9bf..., run_id dab2a67f... and all three per-component test_case_ids). 20 new unit tests pass in each repo; v2_schema imports from the script's own directory, stdlib only. Signed-off-by: Ashok Pon Kumar --- .github/scripts/check_v2_schema_drift.py | 47 ++++ .github/scripts/ingest_xml_hf_adapters.py | 63 +++-- .github/scripts/test_v2_schema.py | 268 ++++++++++++++++++++++ .github/scripts/v2_schema.py | 158 +++++++++++++ 4 files changed, 501 insertions(+), 35 deletions(-) create mode 100755 .github/scripts/check_v2_schema_drift.py create mode 100644 .github/scripts/test_v2_schema.py create mode 100644 .github/scripts/v2_schema.py diff --git a/.github/scripts/check_v2_schema_drift.py b/.github/scripts/check_v2_schema_drift.py new file mode 100755 index 00000000..191bdb8f --- /dev/null +++ b/.github/scripts/check_v2_schema_drift.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""Fail if this repo's copy of v2_schema.py has drifted from the others in MEANING. + +The file is copied, not imported: the baked-image ingest runs under +`uv run --no-project`, so there is no sys.path beyond the script's own directory and no +package to install. This check is what keeps the copies honest. + +It compares parsed ASTs, not bytes. The repos pin different ruff line lengths (88 here, 100 in +spyre-inference), so byte equality is unachievable while semantic equality is exactly what +matters -- and a formatting-only difference is precisely what made one earlier fix need two +different patches. + +Usage: check_v2_schema_drift.py [ ...] +""" + +import ast +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent / "v2_schema.py" + + +def fingerprint(path: Path) -> str: + return ast.dump(ast.parse(path.read_text(encoding="utf-8"))) + + +def main(argv): + if not HERE.exists(): + print(f"ERROR: {HERE} not found", file=sys.stderr) + return 2 + mine = fingerprint(HERE) + bad = 0 + for other in argv: + p = Path(other) + if not p.exists(): + print(f"SKIP {p} (not present)") + continue + if fingerprint(p) == mine: + print(f"OK {p} matches") + else: + print(f"DRIFT {p} differs in meaning from {HERE}", file=sys.stderr) + bad += 1 + return 1 if bad else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index e9e4e316..6a411e72 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -20,6 +20,8 @@ import os import platform as _platform import sys + +import v2_schema import uuid from collections import Counter from datetime import datetime, timezone @@ -511,8 +513,11 @@ def v2_already_ingested(client, run_id: str, component: str) -> bool: def insert_v2(client, component: str, run_id: str, cases: list) -> int: """Write test_cases (identity) + test_case_runs (outcome) for one leg. - Dropped from v2 deliberately: filename, suite_name, runner_run_id, and every - stored counter -- all derivable, and a stored counter invites drift. + Rows are built as dicts and ordered by v2_schema, so a field cannot be assigned to the + wrong column and the column order lives in exactly one place. + + Dropped from v2 deliberately: filename, suite_name, runner_run_id, and every stored + counter -- all derivable, and a stored counter invites drift. """ if not cases: return 0 @@ -528,46 +533,34 @@ def insert_v2(client, component: str, run_id: str, cases: list) -> int: # other unidentifiable case rather than merely orphaning it. skipped_unidentifiable += 1 continue - # Deduped by id within the leg: identical identity rows are one fact, and - # test_cases is a plain MergeTree (a content hash re-writes an identical row, - # so collapsing would only be cosmetic -- but writing it N times is not). - ident_rows[tcid] = [tcid, component, classname, name, tags] + # Keyed by id: identical identity rows within a leg are one fact. + ident_rows[tcid] = { + "test_case_id": tcid, + "component": component, + "classname": classname, + "name": name, + "tags": tags, + } run_rows.append( - [ - run_id, - tcid, - component, - c.get("status", ""), - float(c.get("duration_s", 0) or 0), - (c.get("fail_message") or "")[:8192], - ] + { + "run_id": run_id, + "test_case_id": tcid, + "component": component, + "status": c.get("status", ""), + "duration_s": float(c.get("duration_s", 0) or 0), + "fail_message": (c.get("fail_message") or "")[:8192], + } ) - client.insert( - "test_cases", - list(ident_rows.values()), - column_names=["test_case_id", "component", "classname", "name", "tags"], - ) - client.insert( - "test_case_runs", - run_rows, - column_names=[ - "run_id", - "test_case_id", - "component", - "status", - "duration_s", - "fail_message", - ], - ) + # Cross-run dedup, not just in-leg: test_cases is a plain MergeTree, so re-inserting a + # known identity appends a duplicate instead of collapsing it. + v2_schema.insert_identities(client, v2_schema.TEST_CASES, ident_rows) + v2_schema.insert(client, v2_schema.TEST_CASE_RUNS, run_rows) if skipped_unidentifiable: print( - f" [warn] v2: {skipped_unidentifiable} case(s) skipped -- no derivable " - f"test_case_id (empty name?); they would have collided, not merely orphaned", + f" [warn] v2: {skipped_unidentifiable} case(s) skipped -- identity not derivable", file=sys.stderr, ) return len(run_rows) - - def main(): parser = argparse.ArgumentParser() parser.add_argument("--xml-dir", default=None) diff --git a/.github/scripts/test_v2_schema.py b/.github/scripts/test_v2_schema.py new file mode 100644 index 00000000..63ac4acb --- /dev/null +++ b/.github/scripts/test_v2_schema.py @@ -0,0 +1,268 @@ +"""Pins the insert layer's contract. The model only changes HOW a (column_names, row) pair is +built, so pair equality against the pre-refactor output is a complete correctness proof.""" + +import pytest +from v2_schema import ( + BENCHMARKS, + BENCHMARK_RUNS, + TEST_CASES, + TEST_CASE_RUNS, + TABLES, + STATUS_VALUES, + SchemaError, + insert, + insert_identities, +) + + +class FakeClient: + def __init__(self, known=()): + self.known = list(known) + self.inserts = [] + + def insert(self, table, rows, column_names=None): + self.inserts.append((table, rows, column_names)) + + def query(self, sql, parameters=None): + asked = set(parameters["ids"]) + + class R: + result_rows = [(k,) for k in self.known if str(k) in asked] + + return R() + + +# ── column order is the pre-refactor order, exactly ───────────────────────────────────── + + +def test_column_order_matches_the_pre_refactor_lists(): + # These are the literal column_names lists the three scripts passed before the refactor. + assert list(TEST_CASES.columns) == [ + "test_case_id", + "component", + "classname", + "name", + "tags", + ] + assert list(TEST_CASE_RUNS.columns) == [ + "run_id", + "test_case_id", + "component", + "status", + "duration_s", + "fail_message", + ] + assert list(BENCHMARKS.columns) == ["benchmark_id", "name", "tags", "props"] + assert list(BENCHMARK_RUNS.columns) == [ + "run_id", + "benchmark_id", + "backend", + "measurements", + "iterations", + "props", + ] + + +def test_row_is_ordered_by_columns_not_by_dict_insertion(): + # A dict built in a different order must still produce the DDL-ordered row; this is the + # whole point of the model. + scrambled = { + "name": "test_y", + "tags": ["a"], + "component": "c", + "classname": "k", + "test_case_id": "u", + } + assert TEST_CASES.row(scrambled) == ["u", "c", "k", "test_y", ["a"]] + + +# ── the mistakes it now makes impossible ──────────────────────────────────────────────── + + +def test_unknown_column_is_refused(): + with pytest.raises(SchemaError, match="no such column"): + TEST_CASES.row( + { + "test_case_id": "u", + "component": "c", + "classname": "k", + "name": "n", + "tags": [], + "run_id": "oops", + } + ) + + +def test_missing_column_is_refused_not_silently_shifted(): + with pytest.raises(SchemaError, match="missing column"): + TEST_CASES.row({"test_case_id": "u", "component": "c", "name": "n", "tags": []}) + + +def test_v1_column_set_cannot_be_written_to_the_v2_table(): + # The live hazard: spyre.test_cases has 14 columns, spyre_v2.test_cases has 6, and the two + # are told apart only by which connection is used. Naming a v1 column now fails loudly here + # instead of reaching the server. + with pytest.raises(SchemaError, match="no such column"): + TEST_CASES.row( + { + "test_case_id": "u", + "component": "c", + "classname": "k", + "name": "n", + "tags": [], + "op_name": "matmul", + "dtype": "fp16", + } + ) + + +def test_empty_required_column_is_refused(): + with pytest.raises(SchemaError, match="must be non-empty"): + TEST_CASES.row( + { + "test_case_id": "u", + "component": "", + "classname": "k", + "name": "n", + "tags": [], + } + ) + + +@pytest.mark.parametrize("status", sorted(STATUS_VALUES)) +def test_every_ddl_allowed_status_is_accepted(status): + row = TEST_CASE_RUNS.row( + { + "run_id": "r", + "test_case_id": "t", + "component": "c", + "status": status, + "duration_s": 1.0, + "fail_message": "", + } + ) + assert row[3] == status + + +def test_status_outside_the_ddl_check_is_refused_before_the_server_sees_it(): + with pytest.raises(SchemaError, match="violates the DDL CHECK"): + TEST_CASE_RUNS.row( + { + "run_id": "r", + "test_case_id": "t", + "component": "c", + "status": "PASSED", + "duration_s": 1.0, + "fail_message": "", + } + ) + + +# ── insert() ──────────────────────────────────────────────────────────────────────────── + + +def test_insert_passes_column_names_and_ordered_rows(): + c = FakeClient() + n = insert( + c, + TEST_CASE_RUNS, + [ + { + "run_id": "r", + "test_case_id": "t", + "component": "c", + "status": "passed", + "duration_s": 0.5, + "fail_message": "", + } + ], + ) + assert n == 1 + table, rows, cols = c.inserts[0] + assert table == "test_case_runs" + assert cols == list(TEST_CASE_RUNS.columns) + assert rows == [["r", "t", "c", "passed", 0.5, ""]] + + +def test_insert_of_nothing_does_not_call_the_client(): + c = FakeClient() + assert insert(c, TEST_CASES, []) == 0 + assert c.inserts == [] + + +# ── identity dedup: the defect that lived in two repos and not the third ───────────────── + + +def test_identity_dedup_skips_rows_the_table_already_holds(): + c = FakeClient(known=["known-id"]) + n = insert_identities( + c, + TEST_CASES, + { + "known-id": { + "test_case_id": "known-id", + "component": "c", + "classname": "k", + "name": "a", + "tags": [], + }, + "new-id": { + "test_case_id": "new-id", + "component": "c", + "classname": "k", + "name": "b", + "tags": [], + }, + }, + ) + assert n == 1 + assert c.inserts[0][1] == [["new-id", "c", "k", "b", []]] + + +def test_identity_dedup_writes_nothing_when_all_are_known(): + c = FakeClient(known=["a", "b"]) + assert ( + insert_identities( + c, + TEST_CASES, + { + "a": { + "test_case_id": "a", + "component": "c", + "classname": "k", + "name": "1", + "tags": [], + }, + "b": { + "test_case_id": "b", + "component": "c", + "classname": "k", + "name": "2", + "tags": [], + }, + }, + ) + == 0 + ) + assert c.inserts == [] + + +def test_identity_dedup_on_a_fact_table_is_a_programming_error(): + with pytest.raises(SchemaError, match="no identity column"): + insert_identities(FakeClient(), TEST_CASE_RUNS, {"x": {}}) + + +def test_fact_tables_declare_no_identity_and_dimensions_do(): + assert TEST_CASES.identity == "test_case_id" + assert BENCHMARKS.identity == "benchmark_id" + assert TEST_CASE_RUNS.identity is None + assert BENCHMARK_RUNS.identity is None + + +def test_registry_covers_exactly_the_four_v2_tables(): + assert set(TABLES) == { + "test_cases", + "test_case_runs", + "benchmarks", + "benchmark_runs", + } diff --git a/.github/scripts/v2_schema.py b/.github/scripts/v2_schema.py new file mode 100644 index 00000000..42fedf76 --- /dev/null +++ b/.github/scripts/v2_schema.py @@ -0,0 +1,158 @@ +"""The v2 ClickHouse schema, as data — one module, byte-identical in all three product repos. + +WHY THIS EXISTS. Every v2 insert used to be a positional list paired with a separate +column_names list, and the same four tables were assembled independently in three repos. +Nothing tied a row's field order to the column list except the author reading both. The +audited state was correct (20/20 inserts column-named, 17/17 arity right), so this is not a +bug fix -- it makes a whole class of mistake unrepresentable, and it removes the divergence +that let one real defect live in two repos and not the third: hf-adapters and spyre-inference +never dedup identity rows across runs, so a case seen in N runs became N rows in test_cases. + +WHY A TABLE MODEL AND NOT AN INGESTER CLASS. The three ingest scripts run two ways -- +directly from a checkout by GitHub Actions, and from inside a baked test image via +`uv run --no-project --with lxml --with clickhouse-connect --with regex`. `--no-project` is +deliberate (uv otherwise tries to sync the torch-spyre project and exits 2, dropping the +ingest), so there is no sys.path beyond the script's own directory and those three wheels. +Nothing here may be imported from another repo or installed as a package: this file is COPIED, +and a drift check keeps the copies honest. Per-repo variation is one constant, COMPONENT, +which is why a class hierarchy would have been the wrong shape. + +WHAT IT DELIBERATELY DOES NOT DO. No runtime type coercion (duration_s typed Float32 accepts +a str), no ClickHouse type mapping, and no identity computation -- run_id and test_case_id +arrive already computed by the uuid5 helpers, which are untouched by design: changing them +re-keys the warehouse and silently breaks v2_already_ingested dedup, producing duplicate rows +rather than an error. +""" + +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple + +# The DDL's CONSTRAINT chk_status, re-expressed. It cannot be read from the server at ingest +# time, so it is duplicated here -- keep in step with functional_tests_v2.sql. +STATUS_VALUES = frozenset({"passed", "failed", "error", "skipped", "xfail", "xpass"}) + + +class SchemaError(ValueError): + """A row that the DDL would reject, or that names a column the table does not have.""" + + +@dataclass(frozen=True) +class Table: + """One v2 table: its columns in DDL order, and how a row is built. + + `columns` is the single place the order lives. Rows are built from a dict keyed by + column name, so a field can never be assigned to the wrong column by position. + """ + + name: str + columns: Tuple[str, ...] + # Columns that must be non-empty, mirroring the DDL's CHECK constraints. + required: Tuple[str, ...] = () + # id column for cross-run identity dedup; None for fact tables, which append freely. + identity: Optional[str] = None + + def row(self, values: Dict[str, Any]) -> List[Any]: + """Order one row by `columns`. Raises on an unknown or missing column. + + The raise is the point: an inserted or renamed column shows up here, at the call + site, instead of shifting every later value into the wrong column. + """ + unknown = set(values) - set(self.columns) + if unknown: + raise SchemaError( + f"{self.name}: no such column(s) {sorted(unknown)}; " + f"table has {list(self.columns)}" + ) + missing = set(self.columns) - set(values) + if missing: + raise SchemaError(f"{self.name}: missing column(s) {sorted(missing)}") + for col in self.required: + if values[col] in ("", None): + raise SchemaError(f"{self.name}: column '{col}' must be non-empty") + if "status" in self.columns and values["status"] not in STATUS_VALUES: + raise SchemaError( + f"{self.name}: status {values['status']!r} violates the DDL CHECK " + f"(allowed: {sorted(STATUS_VALUES)})" + ) + return [values[c] for c in self.columns] + + +# ── the four v2 tables, columns in DDL order ──────────────────────────────────────────── +# `ts` is omitted from every one: it is DEFAULT now() and letting the server set it keeps the +# ingest clock out of the data. props is omitted from TEST_CASE_RUNS for the same reason it is +# absent from the writer today -- nothing populates it yet. + +TEST_CASES = Table( + name="test_cases", + columns=("test_case_id", "component", "classname", "name", "tags"), + required=("component", "name"), + identity="test_case_id", +) + +TEST_CASE_RUNS = Table( + name="test_case_runs", + columns=( + "run_id", + "test_case_id", + "component", + "status", + "duration_s", + "fail_message", + ), + required=("component",), +) + +BENCHMARKS = Table( + name="benchmarks", + columns=("benchmark_id", "name", "tags", "props"), + required=("name",), + identity="benchmark_id", +) + +BENCHMARK_RUNS = Table( + name="benchmark_runs", + columns=( + "run_id", + "benchmark_id", + "backend", + "measurements", + "iterations", + "props", + ), +) + +TABLES = {t.name: t for t in (TEST_CASES, TEST_CASE_RUNS, BENCHMARKS, BENCHMARK_RUNS)} + + +def insert(client, table: Table, rows: Sequence[Dict[str, Any]]) -> int: + """Insert dicts into `table`, ordering every row through the one column list.""" + if not rows: + return 0 + ordered = [table.row(r) for r in rows] + client.insert(table.name, ordered, column_names=list(table.columns)) + return len(ordered) + + +def insert_identities(client, table: Table, rows: Dict[Any, Dict[str, Any]]) -> int: + """Insert only the identity rows the dimension does not already hold. + + Both dimensions are plain MergeTree, so re-inserting a known identity APPENDS a duplicate + rather than collapsing it -- one case seen in 36 runs became 36 rows. Deduping within a run + is not enough because the collision is across runs. Centralising it here is what stops the + check from being present in one repo and missing in the other two. + """ + if not rows: + return 0 + if not table.identity: + raise SchemaError(f"{table.name} has no identity column") + ids = [str(k) for k in rows] + known = { + str(r[0]) + for r in client.query( + f"SELECT {table.identity} FROM {table.name} " + f"WHERE {table.identity} IN {{ids:Array(UUID)}}", + parameters={"ids": ids}, + ).result_rows + } + fresh = [v for k, v in rows.items() if str(k) not in known] + return insert(client, table, fresh) From ae6a77c5a8ae85adce81042f34a20a3316154c10 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 00:29:23 +0530 Subject: [PATCH 09/14] fix(ingest): license headers and per-repo lint on the v2 schema model Three separate failures, all in the new files: license-eye (spyre-inference only -- torch-spyre and hf-adapters have no .licenserc.yaml and no license workflow, verified rather than assumed) rejected all three new files. Added the Apache header its .licenserc.yaml specifies, below the shebang where one exists. ruff in spyre-inference reported 11 errors the other two repos do not: it enforces builtin generics and PEP 604 unions over the typing aliases. Applied the strictest form, so the file now satisfies all three configs at once rather than each repo carrying a different variant. check_v2_schema_drift.py sorts import statements before fingerprinting. Each repo pins its own ruff and their isort rules order `typing` against `collections.abc` differently, so the raw AST differed on files that are otherwise the same code and the check reported false drift. It now compares meaning, insensitive to both formatting and import order -- which is the only thing it was ever meant to assert. Re-verified after the changes: the model still emits BYTE-IDENTICAL (column_names, rows) pairs to the pre-refactor writer in all three repos, the pinned identity goldens are unmoved, 20 tests pass per repo, and the drift check passes across all three copies. Signed-off-by: Ashok Pon Kumar --- .github/scripts/check_v2_schema_drift.py | 12 +++++++++++- .github/scripts/test_v2_schema.py | 8 ++++---- .github/scripts/v2_schema.py | 17 +++++++++-------- 3 files changed, 24 insertions(+), 13 deletions(-) diff --git a/.github/scripts/check_v2_schema_drift.py b/.github/scripts/check_v2_schema_drift.py index 191bdb8f..a8bcacf4 100755 --- a/.github/scripts/check_v2_schema_drift.py +++ b/.github/scripts/check_v2_schema_drift.py @@ -21,7 +21,17 @@ def fingerprint(path: Path) -> str: - return ast.dump(ast.parse(path.read_text(encoding="utf-8"))) + """A signature of the module's MEANING, insensitive to formatting and import order. + + Import statements are sorted before dumping: each repo pins its own ruff, and their + isort rules order `typing` against `collections.abc` differently, so the raw AST differs + on files that are otherwise the same code. Comparing raw dumps reported those as drift. + """ + tree = ast.parse(path.read_text(encoding="utf-8")) + imports = [n for n in tree.body if isinstance(n, (ast.Import, ast.ImportFrom))] + rest = [n for n in tree.body if not isinstance(n, (ast.Import, ast.ImportFrom))] + tree.body = sorted(imports, key=ast.dump) + rest + return ast.dump(tree) def main(argv): diff --git a/.github/scripts/test_v2_schema.py b/.github/scripts/test_v2_schema.py index 63ac4acb..14590d32 100644 --- a/.github/scripts/test_v2_schema.py +++ b/.github/scripts/test_v2_schema.py @@ -3,12 +3,12 @@ import pytest from v2_schema import ( - BENCHMARKS, BENCHMARK_RUNS, - TEST_CASES, - TEST_CASE_RUNS, - TABLES, + BENCHMARKS, STATUS_VALUES, + TABLES, + TEST_CASE_RUNS, + TEST_CASES, SchemaError, insert, insert_identities, diff --git a/.github/scripts/v2_schema.py b/.github/scripts/v2_schema.py index 42fedf76..465e647a 100644 --- a/.github/scripts/v2_schema.py +++ b/.github/scripts/v2_schema.py @@ -24,8 +24,9 @@ rather than an error. """ -from dataclasses import dataclass, field -from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any # The DDL's CONSTRAINT chk_status, re-expressed. It cannot be read from the server at ingest # time, so it is duplicated here -- keep in step with functional_tests_v2.sql. @@ -45,13 +46,13 @@ class Table: """ name: str - columns: Tuple[str, ...] + columns: tuple[str, ...] # Columns that must be non-empty, mirroring the DDL's CHECK constraints. - required: Tuple[str, ...] = () + required: tuple[str, ...] = () # id column for cross-run identity dedup; None for fact tables, which append freely. - identity: Optional[str] = None + identity: str | None = None - def row(self, values: Dict[str, Any]) -> List[Any]: + def row(self, values: dict[str, Any]) -> list[Any]: """Order one row by `columns`. Raises on an unknown or missing column. The raise is the point: an inserted or renamed column shows up here, at the call @@ -124,7 +125,7 @@ def row(self, values: Dict[str, Any]) -> List[Any]: TABLES = {t.name: t for t in (TEST_CASES, TEST_CASE_RUNS, BENCHMARKS, BENCHMARK_RUNS)} -def insert(client, table: Table, rows: Sequence[Dict[str, Any]]) -> int: +def insert(client, table: Table, rows: Sequence[dict[str, Any]]) -> int: """Insert dicts into `table`, ordering every row through the one column list.""" if not rows: return 0 @@ -133,7 +134,7 @@ def insert(client, table: Table, rows: Sequence[Dict[str, Any]]) -> int: return len(ordered) -def insert_identities(client, table: Table, rows: Dict[Any, Dict[str, Any]]) -> int: +def insert_identities(client, table: Table, rows: dict[Any, dict[str, Any]]) -> int: """Insert only the identity rows the dimension does not already hold. Both dimensions are plain MergeTree, so re-inserting a known identity APPENDS a duplicate From e4baa6e73412b1f14c17c6254d8514003cf2b428 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 00:39:16 +0530 Subject: [PATCH 10/14] style(ingest): sort the v2_schema import into the third-party group ruff I001: inserting `import v2_schema` next to the stdlib block left the import block un-sorted. Same one-line fix already applied in torch-spyre and spyre-inference. Scoped to this file deliberately. A repo-wide ruff --fix here also reformats 24 files this branch never touches (hf_adapters/, tests/, evals/, scripts/) -- those findings pre-date this work, this repo's lint was already green, and sweeping them in would hide the v2 change inside an unrelated reformat. They are left alone. Verified: import SET unchanged (only ordering), the model still emits byte-identical (column_names, rows) pairs in all three repos, and the pinned identity goldens are unmoved. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index 6a411e72..f28fef02 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -20,14 +20,13 @@ import os import platform as _platform import sys - -import v2_schema import uuid from collections import Counter from datetime import datetime, timezone from pathlib import Path import clickhouse_connect +import v2_schema from lxml import etree # --------------------------------------------------------------------------- @@ -561,6 +560,8 @@ def insert_v2(client, component: str, run_id: str, cases: list) -> int: file=sys.stderr, ) return len(run_rows) + + def main(): parser = argparse.ArgumentParser() parser.add_argument("--xml-dir", default=None) From 05d9338d4e04152d2c037be1b8a4786a9dd28370 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 00:53:16 +0530 Subject: [PATCH 11/14] fix(ingest): anchor the v2_schema sibling import to the script's own directory The ingest tests failed at collection: all 7 in test_ingest_xml and all 12 in test_ingest_kernel_xml errored with ModuleNotFoundError on `import v2_schema`. v2_schema is a SIBLING FILE, not an installed package -- it is copied into three repos and the script is run by path (`uv run --no-project .../ingest_xml*.py`), so the script's own directory is on sys.path only when it is the ENTRY POINT. The tests load it with spec_from_file_location, which does not add that directory, so the import could never resolve there. Fixed in the script rather than the tests: any caller importing this as a module hits the same wall, and the entry-point-only assumption was the real defect. Also records why this file does not use a driver-provided schema layer: clickhouse-connect has none to use. Its ColumnDef is what DESCRIBE TABLE returns -- it reads a live table's schema and cannot declare one or validate a row -- and Client.insert takes "a sequence of sequences" plus an ordered column-name list, so positional rows are the native API shape, not a style choice. column_names='*' only rebinds the order to whatever the server currently reports, coupling every row to the live DDL instead of removing the hazard. clickhouse-sqlalchemy would subsume most of this file, but adding a dependency is what the uv --no-project runtime rules out. Verified: all three scripts now load via spec_from_file_location from an unrelated cwd; 7 + 12 previously-failing ingest tests pass; the model still emits byte-identical (column_names, rows) pairs in all three repos and the pinned identity goldens are unmoved. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 6 ++++++ .github/scripts/v2_schema.py | 9 +++++++++ 2 files changed, 15 insertions(+) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index f28fef02..c7a91220 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -25,6 +25,12 @@ from datetime import datetime, timezone from pathlib import Path +# v2_schema is a SIBLING file, not an installed package: this script is copied into three +# repos and run by path (`uv run --no-project .../ingest_xml*.py`), so its own directory is only +# on sys.path when it is the entry point. A caller that loads it via spec_from_file_location -- +# as the ingest tests do -- would otherwise fail at this import. +sys.path.insert(0, str(Path(__file__).resolve().parent)) + import clickhouse_connect import v2_schema from lxml import etree diff --git a/.github/scripts/v2_schema.py b/.github/scripts/v2_schema.py index 465e647a..00227ab7 100644 --- a/.github/scripts/v2_schema.py +++ b/.github/scripts/v2_schema.py @@ -17,6 +17,15 @@ and a drift check keeps the copies honest. Per-repo variation is one constant, COMPONENT, which is why a class hierarchy would have been the wrong shape. +WHY NOT THE DRIVER'S OWN SCHEMA SUPPORT. clickhouse-connect has none to use. Its `ColumnDef` +is what DESCRIBE TABLE returns -- it reads a live table's schema, it cannot declare one or check +a row against it -- and `Client.insert` takes "a sequence of sequences" plus an ordered column-name +list, so positional rows are the native API shape rather than a style choice here. Passing +`column_names='*'` only moves the order to whatever the server currently reports, which couples +every row to the live DDL instead of removing the hazard. A declarative layer does exist in +clickhouse-sqlalchemy, and it would subsume most of this file -- but adding a dependency is what +the `uv run --no-project` runtime above rules out. Revisit if that constraint ever lifts. + WHAT IT DELIBERATELY DOES NOT DO. No runtime type coercion (duration_s typed Float32 accepts a str), no ClickHouse type mapping, and no identity computation -- run_id and test_case_id arrive already computed by the uuid5 helpers, which are untouched by design: changing them From ab8170663f548ec421b08dded553f1efa05450d3 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 01:05:27 +0530 Subject: [PATCH 12/14] style(ingest): satisfy black on the v2 schema tests This repo runs black (line-length 88) in ADDITION to ruff; torch-spyre and spyre-inference do not, which is why only this PR failed. I had formatted all three repos with ruff alone, so test_v2_schema.py was left in a shape ruff accepts and black rewrites. Verified black --check, ruff check, ruff format and the hygiene hooks (trailing-whitespace, end-of-file-fixer, debug-statements) all pass together on the new files -- black and ruff can disagree, so satisfying one is not evidence for the other. Cosmetic: 20 tests still pass, the byte-identical (column_names, rows) equivalence holds in all three repos, the pinned identity goldens are unmoved, and the drift check still matches. Signed-off-by: Ashok Pon Kumar --- .github/scripts/test_v2_schema.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/scripts/test_v2_schema.py b/.github/scripts/test_v2_schema.py index 14590d32..d0a954ad 100644 --- a/.github/scripts/test_v2_schema.py +++ b/.github/scripts/test_v2_schema.py @@ -1,5 +1,6 @@ """Pins the insert layer's contract. The model only changes HOW a (column_names, row) pair is -built, so pair equality against the pre-refactor output is a complete correctness proof.""" +built, so pair equality against the pre-refactor output is a complete correctness proof. +""" import pytest from v2_schema import ( From 5262627afcdad855830d0d001e37b2a55546ba6c Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 02:46:35 +0530 Subject: [PATCH 13/14] fix(ingest): dedup per source file so a sharded run keeps every shard Found by ingesting a REAL Spyre-Next run (Spyre-Next/product-test #104, spyre-inference, 7 junit shards) into a dev copy of the prod v2 schema: 9 of 10 cases were silently dropped. v2_already_ingested keyed on (component, run_id). But a sharded run is MANY xml files under ONE run_id -- verified in pushToClickhouse.groovy, which invokes the ingest with --xml-dir and every shard in a single call -- so the first shard's rows made the check true and shards 2..N were skipped as "already ingested". The guard was doing its job for a re-ingest and destroying data for a shard. Now keyed on (component, run_id, props['source_file']). props is in the DDL already and sits outside every key, so recording the filename changes no sort order. Declared on the model's TEST_CASE_RUNS, which is what makes the column writable at all. Double-ingest protection is unchanged: re-running the same 7 shards skips all 7 and leaves exactly 10 rows. Also gated the v1 tables_exist() guard on args.write_v1 (hf/si only). It checks a V1 table, so under --schema v2 an absent v1 schema aborted the whole ingest before the v2 write -- a v2-only database was a silent no-op. Verified against the real data, not a fixture: 7 shards -> 1 run_id / 10 rows / 7 source files; re-ingest -> 7 skipped, still 10 rows; v_run_tier_counters and v_tier_report_completeness both read it correctly. Model still emits byte-identical (column_names, rows) pairs in all three repos, identity goldens unmoved, 22 tests pass per repo. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 61 +++++++++++++++++------ .github/scripts/test_v2_schema.py | 41 ++++++++++++++- .github/scripts/v2_schema.py | 3 ++ 3 files changed, 90 insertions(+), 15 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index c7a91220..cc8103dd 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -502,20 +502,44 @@ def v2_tables_present(client) -> bool: ) -def v2_already_ingested(client, run_id: str, component: str) -> bool: - """test_case_runs is a plain MergeTree with no dedup key, so a double ingest of one - leg DOUBLES its counts -- and the v2 schema dropped the stored counters precisely - because they are derived from these rows. This check is what keeps that correct. - Scoped by component as well as run_id to hit the ORDER BY prefix.""" - rows = client.query( - "SELECT count() FROM test_case_runs " - "WHERE component = {component:String} AND run_id = {run_id:UUID}", - parameters={"component": component, "run_id": run_id}, - ).result_rows +def v2_already_ingested( + client, run_id: str, component: str, source_file: str = "" +) -> bool: + """Has THIS source file's rows for this run already landed? + + test_case_runs is a plain MergeTree with no dedup key, so a double ingest of one leg + DOUBLES its counts -- and v2 dropped the stored counters precisely because they are + derived from these rows. This check is what keeps that correct. + + Scoped by source file, not just run_id: a sharded run is MANY xml files under ONE + run_id (the pipeline passes --xml-dir with every shard in a single invocation), so a + run-level check lets the first shard block all the others. Measured on a real + Spyre-Next run: 9 of 10 cases silently dropped across 7 shards. + + `props['source_file']` carries the discriminator. props is a Map outside every key, so + recording it costs no sort-order change. + """ + if source_file: + rows = client.query( + "SELECT count() FROM test_case_runs " + "WHERE component = {component:String} AND run_id = {run_id:UUID} " + "AND props['source_file'] = {sf:String}", + parameters={"component": component, "run_id": run_id, "sf": source_file}, + ).result_rows + else: + # No discriminator given: fall back to the run-level check rather than skip + # dedup entirely, so a caller that cannot name the file is still protected. + rows = client.query( + "SELECT count() FROM test_case_runs " + "WHERE component = {component:String} AND run_id = {run_id:UUID}", + parameters={"component": component, "run_id": run_id}, + ).result_rows return bool(rows and rows[0][0] > 0) -def insert_v2(client, component: str, run_id: str, cases: list) -> int: +def insert_v2( + client, component: str, run_id: str, cases: list, source_file: str = "" +) -> int: """Write test_cases (identity) + test_case_runs (outcome) for one leg. Rows are built as dicts and ordered by v2_schema, so a field cannot be assigned to the @@ -554,6 +578,9 @@ def insert_v2(client, component: str, run_id: str, cases: list) -> int: "status": c.get("status", ""), "duration_s": float(c.get("duration_s", 0) or 0), "fail_message": (c.get("fail_message") or "")[:8192], + # Names the xml this row came from, so a sharded run dedups per + # file instead of the first shard blocking the rest. + "props": ({"source_file": source_file} if source_file else {}), } ) # Cross-run dedup, not just in-leg: test_cases is a plain MergeTree, so re-inserting a @@ -658,7 +685,9 @@ def main(): print("Connected.\n") db = os.environ.get("CLICKHOUSE_DB", "spyre") - if not tables_exist(client, db): + # Gated on write_v1: this checks a V1 table, so under --schema v2 an absent v1 schema + # is expected, not a reason to abort before the v2 write runs. + if args.write_v1 and not tables_exist(client, db): print( f"{db}.hf_test_runs does not exist — nothing to ingest into. Silent no-op." ) @@ -746,10 +775,14 @@ def main(): f"tier={_v2_tier!r}); --trigger-type is the field usually missing", file=sys.stderr, ) - elif v2_already_ingested(v2client, _v2_run_id, V2_COMPONENT): + elif v2_already_ingested( + v2client, _v2_run_id, V2_COMPONENT, xml_path.name + ): print(f" v2: already ingested run_id={_v2_run_id} — skipping") else: - _n = insert_v2(v2client, V2_COMPONENT, _v2_run_id, cases) + _n = insert_v2( + v2client, V2_COMPONENT, _v2_run_id, cases, xml_path.name + ) print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") except Exception as _v2_err: print( diff --git a/.github/scripts/test_v2_schema.py b/.github/scripts/test_v2_schema.py index d0a954ad..f9124120 100644 --- a/.github/scripts/test_v2_schema.py +++ b/.github/scripts/test_v2_schema.py @@ -52,6 +52,7 @@ def test_column_order_matches_the_pre_refactor_lists(): "status", "duration_s", "fail_message", + "props", ] assert list(BENCHMARKS.columns) == ["benchmark_id", "name", "tags", "props"] assert list(BENCHMARK_RUNS.columns) == [ @@ -140,6 +141,7 @@ def test_every_ddl_allowed_status_is_accepted(status): "status": status, "duration_s": 1.0, "fail_message": "", + "props": {}, } ) assert row[3] == status @@ -155,6 +157,7 @@ def test_status_outside_the_ddl_check_is_refused_before_the_server_sees_it(): "status": "PASSED", "duration_s": 1.0, "fail_message": "", + "props": {}, } ) @@ -175,6 +178,7 @@ def test_insert_passes_column_names_and_ordered_rows(): "status": "passed", "duration_s": 0.5, "fail_message": "", + "props": {"source_file": "a.xml"}, } ], ) @@ -182,7 +186,7 @@ def test_insert_passes_column_names_and_ordered_rows(): table, rows, cols = c.inserts[0] assert table == "test_case_runs" assert cols == list(TEST_CASE_RUNS.columns) - assert rows == [["r", "t", "c", "passed", 0.5, ""]] + assert rows == [["r", "t", "c", "passed", 0.5, "", {"source_file": "a.xml"}]] def test_insert_of_nothing_does_not_call_the_client(): @@ -267,3 +271,38 @@ def test_registry_covers_exactly_the_four_v2_tables(): "benchmarks", "benchmark_runs", } + + +# ── sharded runs: many xml files under ONE run_id ──────────────────────────────────────── + + +def test_props_carries_the_source_file_discriminator(): + # The dedup keys on it, so it must survive row assembly in the right column. + row = TEST_CASE_RUNS.row( + { + "run_id": "r", + "test_case_id": "t", + "component": "c", + "status": "passed", + "duration_s": 0.1, + "fail_message": "", + "props": {"source_file": "junit__shard_3.xml"}, + } + ) + assert row[-1] == {"source_file": "junit__shard_3.xml"} + assert TEST_CASE_RUNS.columns[-1] == "props" + + +def test_props_may_be_empty_when_no_source_file_is_known(): + row = TEST_CASE_RUNS.row( + { + "run_id": "r", + "test_case_id": "t", + "component": "c", + "status": "passed", + "duration_s": 0.1, + "fail_message": "", + "props": {}, + } + ) + assert row[-1] == {} diff --git a/.github/scripts/v2_schema.py b/.github/scripts/v2_schema.py index 00227ab7..760b476a 100644 --- a/.github/scripts/v2_schema.py +++ b/.github/scripts/v2_schema.py @@ -101,6 +101,8 @@ def row(self, values: dict[str, Any]) -> list[Any]: TEST_CASE_RUNS = Table( name="test_case_runs", + # props carries source_file, the per-XML discriminator the dedup checks: a sharded run is + # many files under ONE run_id, so a run-level check would let the first shard block the rest. columns=( "run_id", "test_case_id", @@ -108,6 +110,7 @@ def row(self, values: dict[str, Any]) -> list[Any]: "status", "duration_s", "fail_message", + "props", ), required=("component",), ) From 20f4e556d63d2f7017c5ddb963e6c118961f0b35 Mon Sep 17 00:00:00 2001 From: Ashok Pon Kumar Date: Thu, 10 Sep 2026 19:22:06 +0530 Subject: [PATCH 14/14] refactor(ingest): one ClickHouse client for both schema generations get_v2_client() opened a SECOND connection bound to the v2 database. It is the same instance and the same credentials, so the second connection bought nothing that qualifying the statements does not. What made it look necessary: test_cases and benchmark_runs exist in BOTH generations with incompatible shapes (v1 benchmark_runs is run_id UInt64 + source_file, v2 is run_id UUID + component), so an UNQUALIFIED name resolves against whichever database the connection holds and silently hits the wrong table. The fix for that is to qualify, not to open a connection per database. v2_database() now returns the database NAME, and Table.qualified(db) puts it in front of the table in every v2 statement. "which database" becomes a property of the CALL rather than of the connection, which is also why _table_exists() takes an explicit db instead of asking currentDatabase(). Two things fall out: - get_client() loses its `database` parameter, which now has no caller passing one. - v2_new_identity_rows() is deleted; v2_schema.insert_identities() already did the same cross-run dedup, and the local copy was the only remaining hand-rolled one. insert_benchmarks_v2() now builds dicts and goes through the schema model like every other v2 write, so the benchmark tables' column order lives in v2_schema and not in a positional list paired with its own column_names. BENCHMARKS/BENCHMARK_RUNS gain the `component` column the DDL already has. Verified on the live dev instance: a client whose own database is `spyre` wrote and read spyre_v2_trial through the qualifier alone, v1's table was untouched, and the identity re-insert was a no-op. The three copies of v2_schema.py pass the drift check. Signed-off-by: Ashok Pon Kumar --- .github/scripts/ingest_xml_hf_adapters.py | 54 +++++++------ .github/scripts/test_v2_schema.py | 77 +++++++++++++++++-- .github/scripts/v2_schema.py | 36 ++++++--- .../push-test-results-to-clickhouse.yaml | 8 +- 4 files changed, 128 insertions(+), 47 deletions(-) diff --git a/.github/scripts/ingest_xml_hf_adapters.py b/.github/scripts/ingest_xml_hf_adapters.py index cc8103dd..77e152c1 100644 --- a/.github/scripts/ingest_xml_hf_adapters.py +++ b/.github/scripts/ingest_xml_hf_adapters.py @@ -171,29 +171,26 @@ def parse_test_xml(xml_path: Path): # --------------------------------------------------------------------------- -def get_client(database: str | None = None): +def get_client(): return clickhouse_connect.get_client( host=os.environ["CLICKHOUSE_HOST"], port=int(os.environ.get("CLICKHOUSE_PORT", 443)), user=os.environ.get("CLICKHOUSE_USER", "default"), password=os.environ["CLICKHOUSE_PASS"], - database=database or os.environ.get("CLICKHOUSE_DB", "spyre"), + database=os.environ.get("CLICKHOUSE_DB", "spyre"), secure=True, ) -def get_v2_client(): - """A SECOND connection, bound to the v2 database, or None when none is configured. +def v2_database() -> str: + """The v2 database name, or "" when v2 is not configured. - v2 is a different DATABASE, not different table names: test_cases exists in both - generations with incompatible shapes, so one `database=` cannot serve both. Returns None - when CLICKHOUSE_DB_V2 is unset, which is what makes --schema v1 (the default) cost - nothing -- no second connection is opened. + A NAME rather than a second connection: the same instance holds both generations, so one + client serves both provided every v2 statement is QUALIFIED. Qualifying is not optional -- + test_cases exists in both with incompatible shapes, so an unqualified name resolves + against whichever database the connection holds and silently hits the wrong table. """ - db = os.environ.get("CLICKHOUSE_DB_V2", "").strip() - if not db: - return None - return get_client(database=db) + return os.environ.get("CLICKHOUSE_DB_V2", "").strip() def insert_run(client, run_id: str, run: dict, args): @@ -493,17 +490,17 @@ def v2_source_and_external_run_id(args, run_id: str): return "local", run_id -def v2_tables_present(client) -> bool: +def v2_tables_present(client, db: str) -> bool: """v2 write path is skipped unless BOTH tables exist, so this script can be deployed before the migration without erroring on every run.""" return all( - bool(client.command(f"EXISTS TABLE {t}")) - for t in ("test_cases", "test_case_runs") + bool(client.command(f"EXISTS TABLE {t.qualified(db)}")) + for t in (v2_schema.TEST_CASES, v2_schema.TEST_CASE_RUNS) ) def v2_already_ingested( - client, run_id: str, component: str, source_file: str = "" + client, db: str, run_id: str, component: str, source_file: str = "" ) -> bool: """Has THIS source file's rows for this run already landed? @@ -519,9 +516,10 @@ def v2_already_ingested( `props['source_file']` carries the discriminator. props is a Map outside every key, so recording it costs no sort-order change. """ + table = v2_schema.TEST_CASE_RUNS.qualified(db) if source_file: rows = client.query( - "SELECT count() FROM test_case_runs " + f"SELECT count() FROM {table} " "WHERE component = {component:String} AND run_id = {run_id:UUID} " "AND props['source_file'] = {sf:String}", parameters={"component": component, "run_id": run_id, "sf": source_file}, @@ -530,7 +528,7 @@ def v2_already_ingested( # No discriminator given: fall back to the run-level check rather than skip # dedup entirely, so a caller that cannot name the file is still protected. rows = client.query( - "SELECT count() FROM test_case_runs " + f"SELECT count() FROM {table} " "WHERE component = {component:String} AND run_id = {run_id:UUID}", parameters={"component": component, "run_id": run_id}, ).result_rows @@ -538,7 +536,7 @@ def v2_already_ingested( def insert_v2( - client, component: str, run_id: str, cases: list, source_file: str = "" + client, db: str, component: str, run_id: str, cases: list, source_file: str = "" ) -> int: """Write test_cases (identity) + test_case_runs (outcome) for one leg. @@ -585,8 +583,8 @@ def insert_v2( ) # Cross-run dedup, not just in-leg: test_cases is a plain MergeTree, so re-inserting a # known identity appends a duplicate instead of collapsing it. - v2_schema.insert_identities(client, v2_schema.TEST_CASES, ident_rows) - v2_schema.insert(client, v2_schema.TEST_CASE_RUNS, run_rows) + v2_schema.insert_identities(client, v2_schema.TEST_CASES, ident_rows, db=db) + v2_schema.insert(client, v2_schema.TEST_CASE_RUNS, run_rows, db=db) if skipped_unidentifiable: print( f" [warn] v2: {skipped_unidentifiable} case(s) skipped -- identity not derivable", @@ -673,10 +671,10 @@ def main(): f"{os.environ['CLICKHOUSE_HOST']}:{os.environ.get('CLICKHOUSE_PORT', 443)} ..." ) client = get_client() - # Separate connection for the v2 tables -- see get_v2_client(). None when - # CLICKHOUSE_DB_V2 is unset, which every v2 site treats as "v2 not configured". - v2client = get_v2_client() if args.write_v2 else None - if args.write_v2 and v2client is None: + # One client, both generations: v2 is reached by QUALIFYING every statement with this + # database name (see v2_database). "" means v2 is not configured. + v2db = v2_database() if args.write_v2 else "" + if args.write_v2 and not v2db: print( " WARN --schema asked for v2 but CLICKHOUSE_DB_V2 is unset — v2 rows skipped", file=sys.stderr, @@ -761,7 +759,7 @@ def main(): # authoritative, so the experimental write is contained rather than allowed to # abort the loop and drop every remaining file's v1 insert. try: - if v2client is not None and v2_tables_present(v2client): + if v2db and v2_tables_present(client, v2db): _v2_source, _v2_ext = v2_source_and_external_run_id(args, run_id) _v2_tier = (getattr(args, "trigger_type", "") or "").strip() _v2_arch = (args.platform or run.get("platform") or "").strip() @@ -776,12 +774,12 @@ def main(): file=sys.stderr, ) elif v2_already_ingested( - v2client, _v2_run_id, V2_COMPONENT, xml_path.name + client, v2db, _v2_run_id, V2_COMPONENT, xml_path.name ): print(f" v2: already ingested run_id={_v2_run_id} — skipping") else: _n = insert_v2( - v2client, V2_COMPONENT, _v2_run_id, cases, xml_path.name + client, v2db, V2_COMPONENT, _v2_run_id, cases, xml_path.name ) print(f" v2: {_n} test_case_runs under run_id={_v2_run_id}") except Exception as _v2_err: diff --git a/.github/scripts/test_v2_schema.py b/.github/scripts/test_v2_schema.py index f9124120..c4e31713 100644 --- a/.github/scripts/test_v2_schema.py +++ b/.github/scripts/test_v2_schema.py @@ -20,11 +20,13 @@ class FakeClient: def __init__(self, known=()): self.known = list(known) self.inserts = [] + self.queries = [] - def insert(self, table, rows, column_names=None): - self.inserts.append((table, rows, column_names)) + def insert(self, table, rows, column_names=None, database=None): + self.inserts.append((table, rows, column_names, database)) def query(self, sql, parameters=None): + self.queries.append(sql) asked = set(parameters["ids"]) class R: @@ -36,8 +38,8 @@ class R: # ── column order is the pre-refactor order, exactly ───────────────────────────────────── -def test_column_order_matches_the_pre_refactor_lists(): - # These are the literal column_names lists the three scripts passed before the refactor. +def test_column_order_matches_the_ddl(): + # The DDL's column order, which is what an insert without column_names would rely on. assert list(TEST_CASES.columns) == [ "test_case_id", "component", @@ -54,10 +56,17 @@ def test_column_order_matches_the_pre_refactor_lists(): "fail_message", "props", ] - assert list(BENCHMARKS.columns) == ["benchmark_id", "name", "tags", "props"] + assert list(BENCHMARKS.columns) == [ + "benchmark_id", + "component", + "name", + "tags", + "props", + ] assert list(BENCHMARK_RUNS.columns) == [ "run_id", "benchmark_id", + "component", "backend", "measurements", "iterations", @@ -183,7 +192,7 @@ def test_insert_passes_column_names_and_ordered_rows(): ], ) assert n == 1 - table, rows, cols = c.inserts[0] + table, rows, cols, _db = c.inserts[0] assert table == "test_case_runs" assert cols == list(TEST_CASE_RUNS.columns) assert rows == [["r", "t", "c", "passed", 0.5, "", {"source_file": "a.xml"}]] @@ -306,3 +315,59 @@ def test_props_may_be_empty_when_no_source_file_is_known(): } ) assert row[-1] == {} + + +# ── the db qualifier: one client, two generations ──────────────────────────────────────── + + +def test_qualified_prefixes_the_database_when_given(): + assert TEST_CASE_RUNS.qualified("spyre_v2") == "spyre_v2.test_case_runs" + + +def test_qualified_stays_bare_without_a_database(): + # "" and None both mean "the connection's own database" -- v1's callers pass neither. + assert TEST_CASE_RUNS.qualified("") == "test_case_runs" + assert TEST_CASE_RUNS.qualified(None) == "test_case_runs" + + +def test_insert_routes_rows_to_the_named_database(): + """The whole point of the single-client refactor: `benchmark_runs` exists in v1 AND v2 + with incompatible shapes, so the database must travel with the CALL.""" + c = FakeClient() + insert( + c, + BENCHMARK_RUNS, + [ + { + "run_id": "r", + "benchmark_id": "b", + "component": "torch-spyre", + "backend": "cpu", + "measurements": {"m": 1.0}, + "iterations": 1, + "props": {}, + } + ], + db="spyre_v2", + ) + assert c.inserts[0][3] == "spyre_v2" + + +def test_identity_dedup_reads_the_named_database(): + c = FakeClient() + insert_identities( + c, + TEST_CASES, + { + "i": { + "test_case_id": "i", + "component": "c", + "classname": "k", + "name": "n", + "tags": [], + } + }, + db="spyre_v2", + ) + assert "spyre_v2.test_cases" in c.queries[0] + assert c.inserts[0][3] == "spyre_v2" diff --git a/.github/scripts/v2_schema.py b/.github/scripts/v2_schema.py index 760b476a..caaf8fa7 100644 --- a/.github/scripts/v2_schema.py +++ b/.github/scripts/v2_schema.py @@ -86,11 +86,19 @@ def row(self, values: dict[str, Any]) -> list[Any]: ) return [values[c] for c in self.columns] + def qualified(self, db: str | None) -> str: + """`db.table` when a database is given, bare table otherwise. + + Every v2 statement is qualified because one client now serves both generations: + `benchmark_runs` exists in v1 AND v2 with incompatible shapes, so an unqualified + name would resolve against whichever database the connection happens to hold. + """ + return f"{db}.{self.name}" if db else self.name + # ── the four v2 tables, columns in DDL order ──────────────────────────────────────────── # `ts` is omitted from every one: it is DEFAULT now() and letting the server set it keeps the -# ingest clock out of the data. props is omitted from TEST_CASE_RUNS for the same reason it is -# absent from the writer today -- nothing populates it yet. +# ingest clock out of the data. TEST_CASES = Table( name="test_cases", @@ -117,36 +125,46 @@ def row(self, values: dict[str, Any]) -> list[Any]: BENCHMARKS = Table( name="benchmarks", - columns=("benchmark_id", "name", "tags", "props"), - required=("name",), + columns=("benchmark_id", "component", "name", "tags", "props"), + required=("component", "name"), identity="benchmark_id", ) BENCHMARK_RUNS = Table( name="benchmark_runs", + # component leads the identity hash and the sort key, so two repos writing the same + # benchmark name stay distinct rows rather than colliding on one benchmark_id. columns=( "run_id", "benchmark_id", + "component", "backend", "measurements", "iterations", "props", ), + required=("component",), ) TABLES = {t.name: t for t in (TEST_CASES, TEST_CASE_RUNS, BENCHMARKS, BENCHMARK_RUNS)} -def insert(client, table: Table, rows: Sequence[dict[str, Any]]) -> int: +def insert( + client, table: Table, rows: Sequence[dict[str, Any]], db: str | None = None +) -> int: """Insert dicts into `table`, ordering every row through the one column list.""" if not rows: return 0 ordered = [table.row(r) for r in rows] - client.insert(table.name, ordered, column_names=list(table.columns)) + client.insert( + table.name, ordered, column_names=list(table.columns), database=db or None + ) return len(ordered) -def insert_identities(client, table: Table, rows: dict[Any, dict[str, Any]]) -> int: +def insert_identities( + client, table: Table, rows: dict[Any, dict[str, Any]], db: str | None = None +) -> int: """Insert only the identity rows the dimension does not already hold. Both dimensions are plain MergeTree, so re-inserting a known identity APPENDS a duplicate @@ -162,10 +180,10 @@ def insert_identities(client, table: Table, rows: dict[Any, dict[str, Any]]) -> known = { str(r[0]) for r in client.query( - f"SELECT {table.identity} FROM {table.name} " + f"SELECT {table.identity} FROM {table.qualified(db)} " f"WHERE {table.identity} IN {{ids:Array(UUID)}}", parameters={"ids": ids}, ).result_rows } fresh = [v for k, v in rows.items() if str(k) not in known] - return insert(client, table, fresh) + return insert(client, table, fresh, db=db) diff --git a/.github/workflows/push-test-results-to-clickhouse.yaml b/.github/workflows/push-test-results-to-clickhouse.yaml index 29d5553c..1f79c7d6 100644 --- a/.github/workflows/push-test-results-to-clickhouse.yaml +++ b/.github/workflows/push-test-results-to-clickhouse.yaml @@ -66,10 +66,10 @@ jobs: CLICKHOUSE_USER: ${{ secrets.CLICKHOUSE_USER }} CLICKHOUSE_PASS: ${{ secrets.CLICKHOUSE_PASS }} CLICKHOUSE_DB: ${{ secrets.CLICKHOUSE_DB }} - # Schema-v2 database. A SEPARATE connection from CLICKHOUSE_DB, because test_cases - # exists in both generations with incompatible shapes, so one database binding cannot - # serve both. Unset/empty makes the v2 write a no-op -- get_v2_client() opens no second - # connection -- so this is safe before the secret exists. + # Schema-v2 database. Reached by QUALIFYING every v2 statement rather than by a second + # connection: test_cases exists in both generations with incompatible shapes, so an + # unqualified name would hit the wrong one. Unset/empty makes the v2 write a no-op, so + # this is safe before the secret exists. CLICKHOUSE_DB_V2: ${{ secrets.CLICKHOUSE_DB_V2 }} # Dual-write during the migration window: v1 stays authoritative while v2 accumulates # the same runs. No data is ported (v1 rows cannot produce a v2 run_id, the hash inputs