diff --git a/.github/workflows/silver-contract-parity.yml b/.github/workflows/silver-contract-parity.yml new file mode 100644 index 000000000..4bef361d8 --- /dev/null +++ b/.github/workflows/silver-contract-parity.yml @@ -0,0 +1,57 @@ +name: Silver Contract Parity + +# Offline cross-connector silver-contract consistency gate. +# +# Verifies that every connector feeding a `silver:` tag declares exactly +# the columns the silver class contract declares — same names, ORDER and types +# (catches "column added/reordered/retyped in one connector, forgotten in +# another" before merge). Reads the dbt manifest produced by `dbt parse` — NO +# warehouse needed, runs in seconds. +# +# This is the offline declaration-consistency gate (layer 2). The dbt contracts +# themselves (layer 1: SQL output == declared schema) are enforced wherever the +# models are built with `dbt build` (e2e / Argo). +# +# No `paths:` filter on purpose: meant to be a required status check on main — +# a path-filtered required check never reports on PRs outside the filter, which +# leaves them stuck on "Expected". +# +# Script: src/ingestion/scripts/check_silver_contract_parity.py + +on: + pull_request: + branches: [main] + workflow_dispatch: + +jobs: + parity: + name: Silver contract parity + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + + # Pinned so an unrelated dbt-core/adapter release can't break this gate. + - name: Install dbt-clickhouse + run: pip install "dbt-core==1.10.15" "dbt-clickhouse==1.9.6" + + # Self-test the gate's own logic (no warehouse, no manifest needed). + - name: Unit-test the parity gate + run: python3 src/ingestion/scripts/tests/test_check_silver_contract_parity.py + + - name: dbt parse (offline, no warehouse) + working-directory: src/ingestion/dbt + env: + # profiles.yml renders env_var('CLICKHOUSE_PASSWORD'); `dbt parse` + # never opens a connection, so any non-empty value lets it render. + CLICKHOUSE_PASSWORD: parse-only + run: dbt parse --profiles-dir . + + - name: Check silver contract parity + run: | + python3 src/ingestion/scripts/check_silver_contract_parity.py \ + --manifest src/ingestion/dbt/target/manifest.json diff --git a/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commits.sql b/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commits.sql index 3dd84ebc1..b6ffda7c9 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commits.sql +++ b/src/ingestion/connectors/git/bitbucket-cloud/dbt/bitbucket_cloud__commits.sql @@ -3,6 +3,7 @@ materialized='incremental', unique_key='unique_key', order_by=['unique_key'], + on_schema_change='append_new_columns', settings={'allow_nullable_key': 1}, schema='staging', tags=['bitbucket-cloud', 'silver:class_git_commits'] @@ -22,9 +23,9 @@ SELECT '' AS committer_email, COALESCE(c.message, '') AS message, parseDateTimeBestEffortOrNull(c.date) AS date, - COALESCE(fc.files_changed, 0) AS files_changed, - COALESCE(fc.lines_added, 0) AS lines_added, - COALESCE(fc.lines_removed, 0) AS lines_removed, + toInt64(COALESCE(fc.files_changed, 0)) AS files_changed, + toInt64(COALESCE(fc.lines_added, 0)) AS lines_added, + toInt64(COALESCE(fc.lines_removed, 0)) AS lines_removed, if(JSONLength(COALESCE(toString(c.parent_hashes), '[]')) > 1, 1, 0) AS is_merge_commit, 'insight_bitbucket_cloud' AS data_source, toUnixTimestamp64Milli(now64()) AS _version, diff --git a/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml b/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml index 6769f905c..490c74256 100644 --- a/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml +++ b/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml @@ -25,8 +25,56 @@ models: - name: bitbucket_cloud__repository_branches description: "Bitbucket Cloud branches -> staging, feeds class_git_repository_branches" + # CONTRACT must match the `class_git_commits` silver contract exactly — name, + # order and data_type. Enforced by scripts/check_silver_contract_parity.py in + # CI: you cannot add/reorder/retype a column here without doing the same in the + # silver contract (and therefore every other connector feeding this class). - name: bitbucket_cloud__commits description: "Bitbucket Cloud commits -> staging, feeds class_git_commits" + config: + contract: + enforced: true + columns: + - name: tenant_id + data_type: String + - name: source_id + data_type: String + - name: unique_key + data_type: String + - name: project_key + data_type: String + - name: repo_slug + data_type: String + - name: commit_hash + data_type: String + - name: branch + data_type: String + - name: author_name + data_type: String + - name: author_email + data_type: String + - name: committer_name + data_type: String + - name: committer_email + data_type: String + - name: message + data_type: String + - name: date + data_type: Nullable(DateTime) + - name: files_changed + data_type: Int64 + - name: lines_added + data_type: Int64 + - name: lines_removed + data_type: Int64 + - name: is_merge_commit + data_type: UInt8 + - name: data_source + data_type: String + - name: _version + data_type: Int64 + - name: _airbyte_extracted_at + data_type: DateTime64(3) - name: bitbucket_cloud__file_changes description: "Bitbucket Cloud file changes -> staging, feeds class_git_file_changes" diff --git a/src/ingestion/connectors/git/github/dbt/github__commits.sql b/src/ingestion/connectors/git/github/dbt/github__commits.sql index 6d58c2c41..2cbb5c2e9 100644 --- a/src/ingestion/connectors/git/github/dbt/github__commits.sql +++ b/src/ingestion/connectors/git/github/dbt/github__commits.sql @@ -2,6 +2,7 @@ materialized='incremental', unique_key='unique_key', order_by=['unique_key'], + on_schema_change='append_new_columns', settings={'allow_nullable_key': 1}, schema='staging', tags=['github', 'silver:class_git_commits'] @@ -21,9 +22,9 @@ SELECT COALESCE(committer_email, '') AS committer_email, COALESCE(message, '') AS message, parseDateTimeBestEffortOrNull(committed_date) AS date, - COALESCE(changed_files, 0) AS files_changed, - COALESCE(additions, 0) AS lines_added, - COALESCE(deletions, 0) AS lines_removed, + toInt64(COALESCE(changed_files, 0)) AS files_changed, + toInt64(COALESCE(additions, 0)) AS lines_added, + toInt64(COALESCE(deletions, 0)) AS lines_removed, if(length(parent_hashes) > 1, 1, 0) AS is_merge_commit, 'insight_github' AS data_source, toUnixTimestamp64Milli(now64()) AS _version, diff --git a/src/ingestion/connectors/git/github/dbt/schema.yml b/src/ingestion/connectors/git/github/dbt/schema.yml index cd3c5996d..943ea6b7b 100644 --- a/src/ingestion/connectors/git/github/dbt/schema.yml +++ b/src/ingestion/connectors/git/github/dbt/schema.yml @@ -26,8 +26,56 @@ models: - name: github__repository_branches description: "GitHub branches -> staging, feeds class_git_repository_branches" + # CONTRACT must match the `class_git_commits` silver contract exactly — name, + # order and data_type. Enforced by scripts/check_silver_contract_parity.py in + # CI: you cannot add/reorder/retype a column here without doing the same in the + # silver contract (and therefore every other connector feeding this class). - name: github__commits description: "GitHub commits -> staging, feeds class_git_commits" + config: + contract: + enforced: true + columns: + - name: tenant_id + data_type: String + - name: source_id + data_type: String + - name: unique_key + data_type: String + - name: project_key + data_type: String + - name: repo_slug + data_type: String + - name: commit_hash + data_type: String + - name: branch + data_type: String + - name: author_name + data_type: String + - name: author_email + data_type: String + - name: committer_name + data_type: String + - name: committer_email + data_type: String + - name: message + data_type: String + - name: date + data_type: Nullable(DateTime) + - name: files_changed + data_type: Int64 + - name: lines_added + data_type: Int64 + - name: lines_removed + data_type: Int64 + - name: is_merge_commit + data_type: UInt8 + - name: data_source + data_type: String + - name: _version + data_type: Int64 + - name: _airbyte_extracted_at + data_type: DateTime64(3) - name: github__file_changes description: "GitHub file changes (PR + direct push) -> staging, feeds class_git_file_changes" diff --git a/src/ingestion/connectors/git/gitlab/dbt/gitlab__commits.sql b/src/ingestion/connectors/git/gitlab/dbt/gitlab__commits.sql index a7e94f77e..8a01f2af6 100644 --- a/src/ingestion/connectors/git/gitlab/dbt/gitlab__commits.sql +++ b/src/ingestion/connectors/git/gitlab/dbt/gitlab__commits.sql @@ -3,6 +3,7 @@ materialized='incremental', unique_key='unique_key', order_by=['unique_key'], + on_schema_change='append_new_columns', settings={'allow_nullable_key': 1}, schema='staging', tags=['gitlab', 'silver:class_git_commits'] @@ -47,9 +48,9 @@ SELECT COALESCE(c.committer_email, '') AS committer_email, COALESCE(c.message, '') AS message, parseDateTimeBestEffortOrNull(c.committed_date) AS date, - COALESCE(f.files_changed, 0) AS files_changed, - COALESCE(c.stats_additions, 0) AS lines_added, - COALESCE(c.stats_deletions, 0) AS lines_removed, + toInt64(COALESCE(f.files_changed, 0)) AS files_changed, + toInt64(COALESCE(c.stats_additions, 0)) AS lines_added, + toInt64(COALESCE(c.stats_deletions, 0)) AS lines_removed, if(COALESCE(c.parent_count, 0) > 1, 1, 0) AS is_merge_commit, 'insight_gitlab' AS data_source, toUnixTimestamp64Milli(now64()) AS _version, diff --git a/src/ingestion/connectors/git/gitlab/dbt/schema.yml b/src/ingestion/connectors/git/gitlab/dbt/schema.yml index 99c42c81c..88bb908d2 100644 --- a/src/ingestion/connectors/git/gitlab/dbt/schema.yml +++ b/src/ingestion/connectors/git/gitlab/dbt/schema.yml @@ -20,8 +20,56 @@ models: - name: gitlab__repository_branches description: "GitLab branches -> staging, feeds class_git_repository_branches" + # CONTRACT must match the `class_git_commits` silver contract exactly — name, + # order and data_type. Enforced by scripts/check_silver_contract_parity.py in + # CI: you cannot add/reorder/retype a column here without doing the same in the + # silver contract (and therefore every other connector feeding this class). - name: gitlab__commits description: "GitLab commits -> staging, feeds class_git_commits" + config: + contract: + enforced: true + columns: + - name: tenant_id + data_type: String + - name: source_id + data_type: String + - name: unique_key + data_type: String + - name: project_key + data_type: String + - name: repo_slug + data_type: String + - name: commit_hash + data_type: String + - name: branch + data_type: String + - name: author_name + data_type: String + - name: author_email + data_type: String + - name: committer_name + data_type: String + - name: committer_email + data_type: String + - name: message + data_type: String + - name: date + data_type: Nullable(DateTime) + - name: files_changed + data_type: Int64 + - name: lines_added + data_type: Int64 + - name: lines_removed + data_type: Int64 + - name: is_merge_commit + data_type: UInt8 + - name: data_source + data_type: String + - name: _version + data_type: Int64 + - name: _airbyte_extracted_at + data_type: DateTime64(3) - name: gitlab__file_changes description: "GitLab commit file changes -> staging, feeds class_git_file_changes" diff --git a/src/ingestion/scripts/check_silver_contract_parity.py b/src/ingestion/scripts/check_silver_contract_parity.py new file mode 100644 index 000000000..b9855aa7a --- /dev/null +++ b/src/ingestion/scripts/check_silver_contract_parity.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Silver-contract consistency gate (connectors <-> silver class). + +Catches cross-connector schema drift *at PR time*, without a warehouse: + + Connectors (e.g. github, bitbucket-cloud) feed one silver class via + `union_by_tag('silver:class_git_commits')` -> `SELECT * ... UNION ALL`. + A dev adds/removes/reorders/retypes a column in one connector's staging + model and forgets the others. On a cluster the UNION ALL then fails + (ClickHouse `Code: 386 NO_COMMON_TYPE` / column-count mismatch) or — worse, + when two same-typed columns swap positions — silently cross-wires data. + PR CI never catches it because it never materialises every connector at once. + +Opt-in by design. A silver class is checked ONLY when its silver model declares +`contract: enforced: true`. That contract is the single source of truth: every +connector tagged `silver:` MUST emit exactly the columns the silver model +declares — same NAMES, same ORDER, same data_types — and must itself carry an +enforced contract (so dbt build keeps its declared columns honest; this gate +only compares declarations). Classes without an enforced silver contract are +reported and skipped, so contracts roll out one class at a time. + +ORDER matters: `union_by_tag` emits positional `SELECT * ... UNION ALL`, and dbt +contracts do not enforce column order, so two connectors with the same column +SET but a different ORDER would pass a set-only check yet cross-wire columns at +runtime. This gate compares the ordered column list. + +Convention: the silver class model for tag `silver:` is the model named +`` (e.g. tag `silver:class_git_commits` -> model `class_git_commits`). + +Source of truth is the dbt manifest (`dbt parse` -> target/manifest.json), fully +offline. + +Usage: + dbt parse # produces target/manifest.json, no DB connection needed + python scripts/check_silver_contract_parity.py \ + --manifest src/ingestion/dbt/target/manifest.json + +Exit code 0 = all enforced silver classes consistent. 1 = drift found (CI fails). +""" + +from __future__ import annotations + +import argparse +import json +import sys +from collections import defaultdict +from pathlib import Path + +SILVER_TAG_PREFIX = "silver:" + + +def load_manifest(path: Path) -> dict: + if not path.exists(): + sys.exit( + f"manifest not found: {path}\n" + f"Run `dbt parse` first (offline, no warehouse needed)." + ) + with path.open(encoding="utf-8") as fh: + return json.load(fh) + + +def _ordered_columns(node: dict) -> list[tuple[str, str]]: + """[(name, data_type), ...] preserving the manifest's declaration order.""" + return [ + (name, col.get("data_type") or "") + for name, col in node.get("columns", {}).items() + ] + + +def index_manifest(manifest: dict): + """Return (connectors_by_tag, model_index). + + connectors_by_tag: tag -> [model_name, ...] for models carrying a + `silver:*` tag (the per-connector staging models). + model_index: model_name -> {"columns": [(name, type), ...], "enforced": bool} + for every model; used to resolve the silver class model per tag. + """ + connectors_by_tag: dict[str, list[str]] = defaultdict(list) + model_index: dict[str, dict] = {} + for node in manifest.get("nodes", {}).values(): + if node.get("resource_type") != "model": + continue + name = node["name"] + model_index[name] = { + "columns": _ordered_columns(node), + "enforced": bool(node.get("contract", {}).get("enforced", False)), + } + for tag in node.get("tags", []): + if tag.startswith(SILVER_TAG_PREFIX): + connectors_by_tag[tag].append(name) + return connectors_by_tag, model_index + + +def check_connector( + reference: list[tuple[str, str]], connector_name: str, connector: dict +) -> list[str]: + """Compare one connector against the silver reference (name + order + type).""" + problems: list[str] = [] + + if not connector["enforced"]: + problems.append( + f"{connector_name}: contract not enforced — its declared columns are " + f"unverified against its SQL (add `contract: enforced: true`)" + ) + cols = connector["columns"] + if not cols: + problems.append(f"{connector_name}: no typed columns declared in schema.yml") + return problems + + ref_names = [c[0] for c in reference] + col_names = [c[0] for c in cols] + ref_set, col_set = set(ref_names), set(col_names) + + missing = [n for n in ref_names if n not in col_set] + extra = [n for n in col_names if n not in ref_set] + if missing: + problems.append(f"{connector_name}: MISSING columns {missing}") + if extra: + problems.append(f"{connector_name}: UNEXPECTED columns {extra} (not in silver contract)") + + # Type mismatches on shared columns. + ref_types = dict(reference) + for name, dtype in cols: + if name in ref_types and dtype != ref_types[name]: + problems.append( + f"{connector_name}.{name}: type {dtype} != silver {ref_types[name]}" + ) + + # Order mismatch — only meaningful once the sets match (otherwise the + # missing/extra messages already explain the divergence). UNION ALL is + # positional, so identical sets in a different order still cross-wire. + if not missing and not extra and col_names != ref_names: + problems.append( + f"{connector_name}: column ORDER differs from silver contract\n" + f" silver: {ref_names}\n" + f" connector: {col_names}" + ) + + return problems + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--manifest", + type=Path, + default=Path("src/ingestion/dbt/target/manifest.json"), + help="path to dbt manifest.json (run `dbt parse` to produce it)", + ) + args = parser.parse_args() + + connectors_by_tag, model_index = index_manifest(load_manifest(args.manifest)) + if not connectors_by_tag: + print("No `silver:*`-tagged models found — nothing to check.") + return 0 + + failed = False + checked = skipped = 0 + for tag in sorted(connectors_by_tag): + silver_model = tag[len(SILVER_TAG_PREFIX):] # e.g. "class_git_commits" + anchor = model_index.get(silver_model) + connector_names = sorted(connectors_by_tag[tag]) + label = f"{tag} ({len(connector_names)} connector(s): {', '.join(connector_names)})" + + # Opt-in: only enforce classes whose silver model has an enforced contract. + if anchor is None or not anchor["enforced"]: + skipped += 1 + reason = "no silver model found" if anchor is None else "silver contract not enforced" + print(f"· {tag} [skipped — {reason}]") + continue + + checked += 1 + problems: list[str] = [] + for name in connector_names: + problems.extend(check_connector(anchor["columns"], name, model_index[name])) + + if problems: + failed = True + print(f"\n✗ {label} [vs silver contract `{silver_model}`]") + for p in problems: + print(f" - {p}") + else: + print(f"✓ {label} [vs silver contract `{silver_model}`]") + + print(f"\nChecked {checked} enforced class(es); skipped {skipped} without an enforced contract.") + if failed: + print( + "Contract drift detected. Every connector feeding an enforced silver " + "class must emit exactly the columns the silver contract declares — " + "same names, order and types.\nFix: align the `columns:` blocks in " + "each connector's schema.yml to the silver class contract.", + file=sys.stderr, + ) + return 1 + + print("All enforced silver classes are connector-consistent.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ingestion/scripts/tests/test_check_silver_contract_parity.py b/src/ingestion/scripts/tests/test_check_silver_contract_parity.py new file mode 100644 index 000000000..f0736d1e0 --- /dev/null +++ b/src/ingestion/scripts/tests/test_check_silver_contract_parity.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +"""Tests for check_silver_contract_parity.py. + +Dependency-free: runs under pytest OR standalone (`python3 `), so CI +can exercise it without a pytest install. Covers the comparison logic +(check_connector) and the opt-in / exit-code behaviour (end-to-end via the CLI). +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +from pathlib import Path + +_SCRIPT_DIR = Path(__file__).resolve().parents[1] +_SCRIPT = _SCRIPT_DIR / "check_silver_contract_parity.py" +sys.path.insert(0, str(_SCRIPT_DIR)) + +import check_silver_contract_parity as gate # noqa: E402 + + +# --- canonical 3-column reference (name, data_type), order significant -------- +REF = [("tenant_id", "String"), ("unique_key", "String"), ("files_changed", "Int64")] + + +def _connector(columns, *, enforced=True): + return {"columns": list(columns), "enforced": enforced} + + +# --- check_connector: the comparison core ------------------------------------ + +def test_in_sync_passes(): + assert gate.check_connector(REF, "gh", _connector(REF)) == [] + + +def test_missing_column_flagged(): + cols = [c for c in REF if c[0] != "files_changed"] + problems = gate.check_connector(REF, "bb", _connector(cols)) + assert any("MISSING" in p and "files_changed" in p for p in problems) + + +def test_extra_column_flagged(): + cols = REF + [("co_authors", "String")] + problems = gate.check_connector(REF, "gh", _connector(cols)) + assert any("UNEXPECTED" in p and "co_authors" in p for p in problems) + + +def test_type_mismatch_flagged(): + cols = [("tenant_id", "String"), ("unique_key", "String"), ("files_changed", "UInt64")] + problems = gate.check_connector(REF, "gh", _connector(cols)) + assert any("files_changed" in p and "UInt64" in p and "Int64" in p for p in problems) + + +def test_order_mismatch_flagged(): + # Same set + same types, different order — must still fail (UNION ALL is positional). + cols = [("unique_key", "String"), ("tenant_id", "String"), ("files_changed", "Int64")] + problems = gate.check_connector(REF, "gh", _connector(cols)) + assert any("ORDER" in p for p in problems) + # ...and a pure reorder must NOT be reported as missing/extra. + assert not any("MISSING" in p or "UNEXPECTED" in p for p in problems) + + +def test_unenforced_connector_flagged(): + problems = gate.check_connector(REF, "gh", _connector(REF, enforced=False)) + assert any("not enforced" in p for p in problems) + + +def test_no_columns_flagged(): + problems = gate.check_connector(REF, "gh", _connector([])) + assert any("no typed columns" in p for p in problems) + + +# --- end-to-end via the CLI: opt-in + exit codes ----------------------------- + +def _node(name, tags, columns, enforced): + return { + "resource_type": "model", + "name": name, + "tags": tags, + "contract": {"enforced": enforced}, + "columns": {n: {"data_type": t} for n, t in columns}, + } + + +def _run(nodes): + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: + json.dump({"nodes": nodes}, f) + path = f.name + try: + return subprocess.run( + [sys.executable, str(_SCRIPT), "--manifest", path], + capture_output=True, text=True, + ) + finally: + Path(path).unlink(missing_ok=True) + + +def test_e2e_enforced_consistent_exit0(): + r = _run({ + "s": _node("class_git_commits", ["silver"], REF, True), + "gh": _node("github__commits", ["github", "silver:class_git_commits"], REF, True), + "bb": _node("bitbucket_cloud__commits", ["bitbucket-cloud", "silver:class_git_commits"], REF, True), + }) + assert r.returncode == 0, r.stdout + r.stderr + assert "✓ silver:class_git_commits" in r.stdout + + +def test_e2e_drift_exit1(): + bb_missing = [c for c in REF if c[0] != "files_changed"] + r = _run({ + "s": _node("class_git_commits", ["silver"], REF, True), + "gh": _node("github__commits", ["github", "silver:class_git_commits"], REF, True), + "bb": _node("bitbucket_cloud__commits", ["bitbucket-cloud", "silver:class_git_commits"], bb_missing, True), + }) + assert r.returncode == 1 + assert "MISSING" in r.stdout + + +def test_e2e_unenforced_silver_skipped_exit0(): + # Silver model without an enforced contract -> class is skipped, not failed, + # even though the connectors blatantly disagree. + r = _run({ + "s": _node("class_git_commits", ["silver"], REF, False), + "gh": _node("github__commits", ["github", "silver:class_git_commits"], REF, True), + "bb": _node("bitbucket_cloud__commits", ["bitbucket-cloud", "silver:class_git_commits"], [("x", "String")], True), + }) + assert r.returncode == 0, r.stdout + r.stderr + assert "skipped" in r.stdout + + +def _main() -> int: + tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] + failed = 0 + for t in tests: + try: + t() + print(f"ok {t.__name__}") + except AssertionError as e: + failed += 1 + print(f"FAIL {t.__name__}: {e}") + print(f"\n{len(tests) - failed}/{len(tests)} passed") + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/src/ingestion/silver/git/class_git_commits.sql b/src/ingestion/silver/git/class_git_commits.sql index b4a44bbd2..72294f249 100644 --- a/src/ingestion/silver/git/class_git_commits.sql +++ b/src/ingestion/silver/git/class_git_commits.sql @@ -7,6 +7,7 @@ incremental_strategy='delete+insert', engine='ReplacingMergeTree(_version)', order_by=['unique_key'], + on_schema_change='append_new_columns', settings={'allow_nullable_key': 1}, schema='silver', tags=['silver'] diff --git a/src/ingestion/silver/git/schema.yml b/src/ingestion/silver/git/schema.yml index a70bf9a2a..6bae212ce 100644 --- a/src/ingestion/silver/git/schema.yml +++ b/src/ingestion/silver/git/schema.yml @@ -27,18 +27,66 @@ models: - not_null - unique + # CONTRACT = source of truth for the git-commits silver class. Every connector + # tagged `silver:class_git_commits` (github, bitbucket-cloud, github-v2, …) + # MUST emit exactly these columns, in this order, with these types. Enforced: + # - dbt build: this contract validates the UNION output; each connector's + # own contract validates its staging SQL. + # - PR CI: scripts/check_silver_contract_parity.py compares every connector + # against this contract — name, ORDER (UNION ALL is positional) and type. - name: class_git_commits description: "Unified git commits from all sources (GitHub, Bitbucket, GitLab)" + config: + contract: + enforced: true columns: - name: tenant_id description: "Tenant isolation field" + data_type: String tests: - not_null + - name: source_id + data_type: String - name: unique_key description: "Composite deduplication key" + data_type: String tests: - not_null - unique + - name: project_key + data_type: String + - name: repo_slug + data_type: String + - name: commit_hash + data_type: String + - name: branch + data_type: String + - name: author_name + data_type: String + - name: author_email + data_type: String + - name: committer_name + data_type: String + - name: committer_email + data_type: String + - name: message + data_type: String + - name: date + data_type: Nullable(DateTime) + - name: files_changed + data_type: Int64 + - name: lines_added + data_type: Int64 + - name: lines_removed + data_type: Int64 + - name: is_merge_commit + data_type: UInt8 + - name: data_source + data_type: String + - name: _version + data_type: Int64 + - name: _airbyte_extracted_at + data_type: DateTime64(3) - name: class_git_file_changes description: "Unified git file changes from all sources (GitHub, Bitbucket, GitLab)"