diff --git a/scripts/ci/components.py b/scripts/ci/components.py index 7298e2ef0..57537e761 100755 --- a/scripts/ci/components.py +++ b/scripts/ci/components.py @@ -203,6 +203,18 @@ "cov_package": "source_github_copilot", "paths": ["src/ingestion/connectors/ai/github-copilot"], }, + # Deploy-time ClickHouse schema tooling (the migration Job's Python half: + # reconcile_bronze_schema, which heals warm-cluster bronze drift — #1991). + # Owning the whole scripts/ tree means a connectors-ddl snapshot regen also + # re-runs these tests, which is the point: the reconciler's contract is with + # that snapshot. Shell scripts in the same tree have no measured lines. + { + "name": "ingestion-scripts", + "lang": "python", + "root": "src/ingestion/scripts", + "cov_package": "reconcile_bronze_schema", + "paths": ["src/ingestion/scripts"], + }, # Mock-server test rig for NOCODE connectors (feature-connector-mock-tests), # split into two CI jobs for clean results (review ask): the harness's own # unit tests (meta/) and the per-connector mock suites. Both measure the diff --git a/src/ingestion/scripts/apply-ch-migrations.sh b/src/ingestion/scripts/apply-ch-migrations.sh index 7a2358b62..b10ee9568 100755 --- a/src/ingestion/scripts/apply-ch-migrations.sh +++ b/src/ingestion/scripts/apply-ch-migrations.sh @@ -183,115 +183,6 @@ heal_task_id_column staging jira__task_comments comment_id heal_task_id_column silver class_task_worklogs worklog_id heal_task_id_column silver class_task_comments comment_id -echo "=== Reconciling legacy Bitbucket bronze placeholders (warm clusters) ===" -# Warm clusters still hold the pre-rewrite FLAT bronze_bitbucket_cloud.{commits, -# pull_requests}; the snapshot applicator's CREATE TABLE IF NOT EXISTS never -# upgrades an existing table, so the git staging models would fail on the missing -# envelope columns (record_type, generation_id, entity_key, ...). Add them -# idempotently (ADD COLUMN IF NOT EXISTS), mirroring the other warm-cluster heals -# above. On a fresh cluster the snapshot already created the full schema, so these -# are no-ops. The sort key of an existing table cannot be altered in place; -# commits/pull_requests carry no snapshot markers so their dedup key is not -# load-bearing. Relocated here from create-bronze-placeholders (which became the -# connectors-ddl snapshot applicator in #1831); originally added in #1880. -heal_bitbucket_commits() { - ch_table_exists bronze_bitbucket_cloud commits || return 0 - echo " bronze_bitbucket_cloud.commits" - run_ch <<'SQL' -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS tenant_id String; -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS source_id String; -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS unique_key String; -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS entity_key Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS data_source Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS collected_at Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS record_type Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS generation_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS bucket_id Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS snapshot_item_count Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS snapshot_available Nullable(Bool); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS repository_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS workspace_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS hash Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS message Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS date Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_raw Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_email Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_display_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS author_account_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_raw Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_email Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_display_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS committer_account_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS parent_hashes Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS workspace Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS repo_slug Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS branch_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS head_sha Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS _airbyte_raw_id String DEFAULT toString(generateUUIDv4()); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS _airbyte_extracted_at DateTime64(3) DEFAULT now64(3); -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS _airbyte_meta String DEFAULT '{}'; -ALTER TABLE bronze_bitbucket_cloud.commits ADD COLUMN IF NOT EXISTS _airbyte_generation_id UInt32 DEFAULT 0; -SQL -} - -heal_bitbucket_pull_requests() { - ch_table_exists bronze_bitbucket_cloud pull_requests || return 0 - echo " bronze_bitbucket_cloud.pull_requests" - run_ch <<'SQL' -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS tenant_id String; -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS source_id String; -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS unique_key String; -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS entity_key Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS data_source Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS collected_at Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS record_type Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS generation_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS bucket_id Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS snapshot_item_count Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS snapshot_available Nullable(Bool); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS repository_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS workspace_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS id Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS title Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS description Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS state Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS created_on Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS updated_on Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS author_display_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS author_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS author_account_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS closed_by_display_name Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS closed_by_uuid Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS closed_by_account_id Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS source_branch Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS destination_branch Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS source_commit_hash Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS destination_commit_hash Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS merge_commit_hash Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS task_count Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS draft Nullable(Bool); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS queued Nullable(Bool); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS close_source_branch Nullable(Bool); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS reason Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS reviewers Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS comment_count Nullable(Int64); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS participants Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS workspace Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS repo_slug Nullable(String); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS _airbyte_raw_id String DEFAULT toString(generateUUIDv4()); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS _airbyte_extracted_at DateTime64(3) DEFAULT now64(3); -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS _airbyte_meta String DEFAULT '{}'; -ALTER TABLE bronze_bitbucket_cloud.pull_requests ADD COLUMN IF NOT EXISTS _airbyte_generation_id UInt32 DEFAULT 0; -SQL -} - -heal_bitbucket_commits -heal_bitbucket_pull_requests - # SKIP_DBT_GOLD=1 (set by bootstrap-db snapshot generation) skips this step: # generation already built every tag:gold model with the pinned dbt venv # (run-dbt.sh) BEFORE the migrations ran, and re-running here would need a `dbt` diff --git a/src/ingestion/scripts/create-bronze-placeholders.sh b/src/ingestion/scripts/create-bronze-placeholders.sh index 281eefac7..1a8d96109 100644 --- a/src/ingestion/scripts/create-bronze-placeholders.sh +++ b/src/ingestion/scripts/create-bronze-placeholders.sh @@ -89,3 +89,15 @@ while true; do done echo "=== connectors-ddl snapshot applied ===" + +# CREATE TABLE IF NOT EXISTS above is a no-op against a table that already +# exists, so a warm cluster keeps the schema its bronze tables had when the +# connector last synced. When a connector adds columns, the staging models that +# read them fail with UNKNOWN_IDENTIFIER and the whole downstream domain is +# skipped (#1991). Add the snapshot's missing columns to every pre-existing +# bronze table; see reconcile_bronze_schema.py for the guarantees (bronze only, +# ADD COLUMN only, idempotent, safe alongside a running sync). Shared with the +# e2e rig so the two cannot drift. +echo "=== Reconciling existing bronze tables to snapshot ===" +python3 "${SCRIPT_DIR}/reconcile_bronze_schema.py" "${DDL_DIR}" +echo "=== bronze reconcile complete ===" diff --git a/src/ingestion/scripts/pyproject.toml b/src/ingestion/scripts/pyproject.toml new file mode 100644 index 000000000..02efff488 --- /dev/null +++ b/src/ingestion/scripts/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=68.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "insight-ingestion-scripts" +version = "0.1.0" +description = "Deploy-time ClickHouse schema tooling shipped in the ingestion toolbox image" +requires-python = ">=3.10" +# Deliberately dependency-free: this runs in the migration Job, which talks to +# ClickHouse over HTTP with nothing but the standard library. +dependencies = [] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.setuptools] +py-modules = ["reconcile_bronze_schema"] diff --git a/src/ingestion/scripts/reconcile_bronze_schema.py b/src/ingestion/scripts/reconcile_bronze_schema.py new file mode 100644 index 000000000..f493f1fca --- /dev/null +++ b/src/ingestion/scripts/reconcile_bronze_schema.py @@ -0,0 +1,325 @@ +#!/usr/bin/env python3 +"""Reconcile pre-existing bronze tables to the connectors-ddl snapshot. + +The snapshot applicator (create-bronze-placeholders.sh) only ever issues +`CREATE TABLE IF NOT EXISTS`, which is a no-op against a table that already +exists. A warm cluster therefore keeps whatever schema its tables had when the +connector last synced: when a connector adds columns, the live table never gains +them, the staging models that read those columns fail with UNKNOWN_IDENTIFIER, +and every downstream silver/gold model is skipped (issue #1991 — an upgrade +froze the git metrics domain because five bitbucket bronze tables predated the +completeness-tracking envelope). + +This module closes that gap generically: for every table in the snapshot that +already exists, add the columns the snapshot declares and the live table lacks. +The snapshot is regenerated from real connector output (bootstrap-db + +dump-ddl.sh), so it is the same source of truth that creates fresh tables — a +hand-maintained per-table ALTER list is exactly what drifted before. + +Schema introspection is delegated to ClickHouse rather than parsed here: each +snapshot statement is replayed as an empty `__ddl_probe`, and the missing +columns are the set difference against `system.columns`. That way column types +are whatever ClickHouse itself normalises them to, so no type-spelling +comparison can go wrong. + +Scope and guarantees: + +* bronze_* databases only. silver/insight/staging/identity/person are owned by + dbt or by the numbered migrations, carry DEFAULT clauses and views, and have + their own heal semantics in apply-ch-migrations.sh. +* ADD COLUMN only. A column whose type differs from the snapshot is reported and + left alone: MODIFY COLUMN rewrites data, which is an operator decision. Live + columns absent from the snapshot are never dropped (legacy or operator columns + keep their data). +* Idempotent. A reconciled cluster produces an empty diff on the next run. +* Safe against a concurrent sync. `ADD COLUMN IF NOT EXISTS` is a metadata-only + mutation, and the probes are separate tables — nothing here rewrites or swaps + a live table. + +Adding a Nullable column is metadata-only in ClickHouse, so this stays fast even +on the largest bronze tables. + +Used by both callers so the logic cannot diverge: + * create-bronze-placeholders.sh runs it as a CLI (prod deploy Job). + * tests/e2e/lib/migration_applier.py imports reconcile() (test rig). +""" + +from __future__ import annotations + +import logging +import os +import re +import sys +import urllib.request +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field +from pathlib import Path + +LOG = logging.getLogger("reconcile_bronze_schema") + +PROBE_SUFFIX = "__ddl_probe" + +# Snapshot statements start with the unquoted `CREATE TABLE IF NOT EXISTS +# db.table` that dump-ddl.sh writes (it rewrites line 1 of SHOW CREATE TABLE). +# Backticks are tolerated in case a future dump quotes identifiers, and table +# names keep their case (bronze_salesforce.OpportunityContactRole). +_CREATE_TABLE_RE = re.compile( + r"^\s*CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(?P\w+)`?\.`?(?P
\w+)`?", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class SnapshotTable: + """One `CREATE TABLE` statement from the snapshot.""" + + database: str + table: str + create_sql: str + + @property + def probe(self) -> str: + return f"{self.table}{PROBE_SUFFIX}" + + def probe_sql(self) -> str: + """The same statement retargeted at the probe table. + + Only the matched `CREATE TABLE ... db.table` prefix is rewritten, so the + column list, ENGINE, ORDER BY and SETTINGS stay byte-identical to the + snapshot. `IF NOT EXISTS` is dropped: the caller drops the probe first, + so a surviving probe should surface rather than be silently reused. + """ + match = _CREATE_TABLE_RE.match(self.create_sql) + if match is None: # pragma: no cover — constructed only from a match + raise ValueError(f"not a CREATE TABLE statement: {self.create_sql[:80]!r}") + return ( + f"CREATE TABLE `{self.database}`.`{self.probe}`" + + self.create_sql[match.end() :] + ) + + +@dataclass +class ReconcileResult: + added: list[tuple[str, str, str]] = field(default_factory=list) + type_drift: list[tuple[str, str, str, str]] = field(default_factory=list) + tables_reconciled: int = 0 + tables_examined: int = 0 + + @property + def columns_added(self) -> int: + return len(self.added) + + +def parse_snapshot_tables(sql: str) -> list[SnapshotTable]: + """Extract every `CREATE TABLE` statement from one snapshot file. + + Statements are separated by a blank line (dump-ddl.sh terminates each with + `printf ';\\n\\n'`), which is also how the shell applicator splits them. + Anything that is not a CREATE TABLE — the leading CREATE DATABASE, and the + views and refreshable MVs in insight.sql — yields no match and is skipped. + """ + tables: list[SnapshotTable] = [] + for block in re.split(r"\n\s*\n", sql): + statement = block.strip() + if not statement: + continue + match = _CREATE_TABLE_RE.match(statement) + if match is None: + continue + tables.append( + SnapshotTable( + database=match.group("db"), + table=match.group("table"), + create_sql=statement, + ) + ) + return tables + + +def load_snapshot_tables(ddl_dir: Path) -> list[SnapshotTable]: + """Every CREATE TABLE across the snapshot, in stable file order.""" + tables: list[SnapshotTable] = [] + for path in sorted(ddl_dir.glob("*.sql")): + tables.extend(parse_snapshot_tables(path.read_text(encoding="utf-8"))) + return tables + + +def is_reconcilable(table: SnapshotTable) -> bool: + """Bronze tables only, and never a probe left over from an aborted run.""" + return table.database.startswith("bronze_") and not table.table.endswith(PROBE_SUFFIX) + + +def _lit(value: str) -> str: + """Single-quoted SQL literal (identifiers here are \\w+, so this is belt-only).""" + return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" + + +def reconcile( + tables: Iterable[SnapshotTable], + *, + execute: Callable[[str], None], + fetch_rows: Callable[[str], Sequence[Sequence[str]]], +) -> ReconcileResult: + """Add snapshot columns missing from each existing bronze table. + + `execute` runs one statement and raises on error; `fetch_rows` runs one + SELECT and returns its rows as string cells. Injecting both keeps this + testable and lets the prod CLI and the e2e rig share one implementation. + """ + result = ReconcileResult() + candidates = [t for t in tables if is_reconcilable(t)] + + live = _existing_tables(candidates, fetch_rows=fetch_rows) + + for table in candidates: + if (table.database, table.table) not in live: + # Absent: the snapshot's own CREATE already made it with the full + # schema, so there is nothing to reconcile. + continue + result.tables_examined += 1 + try: + _reconcile_one(table, execute=execute, fetch_rows=fetch_rows, result=result) + finally: + execute(f"DROP TABLE IF EXISTS `{table.database}`.`{table.probe}`") + + return result + + +def _existing_tables( + tables: Sequence[SnapshotTable], + *, + fetch_rows: Callable[[str], Sequence[Sequence[str]]], +) -> set[tuple[str, str]]: + """One round-trip to learn which candidate tables already exist.""" + if not tables: + return set() + databases = ", ".join(sorted({_lit(t.database) for t in tables})) + rows = fetch_rows( + "SELECT database, name FROM system.tables " + f"WHERE database IN ({databases})" + ) + return {(row[0], row[1]) for row in rows if len(row) >= 2} + + +def _reconcile_one( + table: SnapshotTable, + *, + execute: Callable[[str], None], + fetch_rows: Callable[[str], Sequence[Sequence[str]]], + result: ReconcileResult, +) -> None: + qualified = f"{table.database}.{table.table}" + execute(f"DROP TABLE IF EXISTS `{table.database}`.`{table.probe}`") + execute(table.probe_sql()) + + db, tbl, probe = _lit(table.database), _lit(table.table), _lit(table.probe) + + missing = fetch_rows( + "SELECT name, type FROM system.columns " + f"WHERE database = {db} AND table = {probe} " + f"AND name NOT IN (SELECT name FROM system.columns WHERE database = {db} AND table = {tbl}) " + "ORDER BY position" + ) + for row in missing: + name, ch_type = row[0], row[1] + execute( + f"ALTER TABLE `{table.database}`.`{table.table}` " + f"ADD COLUMN IF NOT EXISTS `{name}` {ch_type}" + ) + result.added.append((qualified, name, ch_type)) + LOG.info(" + %s.%s %s", qualified, name, ch_type) + + # Report-only: a differing type means the live table would need a data + # rewrite (MODIFY COLUMN), which is never done unattended. + drift = fetch_rows( + "SELECT s.name, s.type, l.type FROM " + f"(SELECT name, type FROM system.columns WHERE database = {db} AND table = {probe}) AS s " + "INNER JOIN " + f"(SELECT name, type FROM system.columns WHERE database = {db} AND table = {tbl}) AS l " + "USING (name) WHERE s.type != l.type ORDER BY s.name" + ) + for row in drift: + name, snapshot_type, live_type = row[0], row[1], row[2] + result.type_drift.append((qualified, name, snapshot_type, live_type)) + LOG.warning( + " ! %s.%s type differs — snapshot=%s live=%s (left unchanged)", + qualified, + name, + snapshot_type, + live_type, + ) + + if missing: + result.tables_reconciled += 1 + + +def _http_client() -> tuple[Callable[[str], None], Callable[[str], Sequence[Sequence[str]]]]: + """Executors over the ClickHouse HTTP interface, mirroring lib/ch-exec.sh. + + ClickHouse is always external to the release, so HTTP is the only path. The + password travels in a header (not argv) exactly as ch-exec.sh does. + """ + url = os.environ.get("CLICKHOUSE_URL") + user = os.environ.get("CLICKHOUSE_USER") + password = os.environ.get("CLICKHOUSE_PASSWORD") + missing = [ + name + for name, value in ( + ("CLICKHOUSE_URL", url), + ("CLICKHOUSE_USER", user), + ("CLICKHOUSE_PASSWORD", password), + ) + if not value + ] + if missing: + raise SystemExit(f"{', '.join(missing)} must be set") + + endpoint = url.rstrip("/") + "/" + + def _post(sql: str) -> str: + request = urllib.request.Request( # noqa: S310 — fixed http(s) endpoint from config + endpoint, + data=sql.encode("utf-8"), + headers={"X-ClickHouse-User": user, "X-ClickHouse-Key": password}, + method="POST", + ) + with urllib.request.urlopen(request) as response: # noqa: S310 + return response.read().decode("utf-8") + + def execute(sql: str) -> None: + _post(sql) + + def fetch_rows(sql: str) -> list[list[str]]: + body = _post(sql) + return [line.split("\t") for line in body.splitlines() if line] + + return execute, fetch_rows + + +def main(argv: Sequence[str] | None = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(message)s") + args = list(argv if argv is not None else sys.argv[1:]) + ddl_dir = Path(args[0]) if args else Path(__file__).resolve().parent / "connectors-ddl" + if not ddl_dir.is_dir(): + raise SystemExit(f"DDL snapshot directory not found: {ddl_dir}") + + tables = load_snapshot_tables(ddl_dir) + execute, fetch_rows = _http_client() + result = reconcile(tables, execute=execute, fetch_rows=fetch_rows) + + LOG.info( + " reconciled %d column(s) across %d of %d existing bronze table(s)", + result.columns_added, + result.tables_reconciled, + result.tables_examined, + ) + if result.type_drift: + LOG.warning( + " %d column(s) differ in type from the snapshot and were left unchanged", + len(result.type_drift), + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py b/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py new file mode 100644 index 000000000..c5bc0a39f --- /dev/null +++ b/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py @@ -0,0 +1,348 @@ +"""Opt-in integration test: reconcile a real warm ClickHouse (issue #1991). + +The unit tests pin the algorithm against a stand-in; this one pins it against +ClickHouse itself — the probe replay, the `system.columns` diff, and the ALTERs +all run for real, so it also proves the generated SQL is valid for the pinned +server version. + +Skipped unless a server is offered, so CI and local `pytest` stay dependency-free: + + docker run -d --rm --name ch -p 38200:8123 \\ + -e CLICKHOUSE_PASSWORD=x clickhouse/clickhouse-server:25.7.5.34 + RECONCILE_TEST_CH_URL=http://localhost:38200 \\ + RECONCILE_TEST_CH_PASSWORD=x .venv/bin/python -m pytest tests -q + +It reproduces the reported failure shape rather than asserting on a fixture: for +each table the issue names, the committed snapshot DDL is replayed with the +post-upgrade columns withheld, which is exactly what an upgraded install holds +(the table exists, so `CREATE TABLE IF NOT EXISTS` never widens it). +""" + +from __future__ import annotations + +import os +import re +import sys +import urllib.error +import urllib.request +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import reconcile_bronze_schema as rbs # noqa: E402 + +CH_URL = os.environ.get("RECONCILE_TEST_CH_URL") +CH_USER = os.environ.get("RECONCILE_TEST_CH_USER", "default") +CH_PASSWORD = os.environ.get("RECONCILE_TEST_CH_PASSWORD", "") + +pytestmark = pytest.mark.skipif( + not CH_URL, + reason="set RECONCILE_TEST_CH_URL (and RECONCILE_TEST_CH_PASSWORD) to run against a live ClickHouse", +) + +DDL_DIR = Path(__file__).resolve().parent.parent / "connectors-ddl" + +# The tables issue #1991 lists, with the columns their staging models could not +# find. bronze_bitbucket_cloud gained the completeness-tracking envelope; the +# outline user snapshot gained its own set. +ENVELOPE = { + "record_type", "generation_id", "bucket_id", "snapshot_item_count", + "snapshot_available", "entity_key", "repository_uuid", "workspace_uuid", + "data_source", "collected_at", +} +OUTLINE_COLUMNS = { + "name", "role", "is_suspended", "data_source", "last_active_at", + "created_at", "collected_at", +} +WITHHELD = { + ("bronze_bitbucket_cloud", table): ENVELOPE + for table in ( + "repositories", "branches", "commits", "file_changes", + "pull_requests", "pull_request_comments", "pull_request_commits", + ) +} +WITHHELD[("bronze_outline", "wiki_users")] = OUTLINE_COLUMNS + +_COLUMN = re.compile(r"^\s*`(?P[^`]+)`\s+.+?,?\s*$") + + +def post(sql: str) -> str: + request = urllib.request.Request( # noqa: S310 — operator-supplied test endpoint + CH_URL.rstrip("/") + "/", + data=sql.encode("utf-8"), + headers={"X-ClickHouse-User": CH_USER, "X-ClickHouse-Key": CH_PASSWORD}, + ) + try: + with urllib.request.urlopen(request) as response: # noqa: S310 + return response.read().decode("utf-8") + except urllib.error.HTTPError as exc: # surface ClickHouse's own message + raise AssertionError(f"{exc.code}: {exc.read().decode('utf-8', 'replace')}\n{sql[:200]}") from exc + + +def rows(sql: str) -> list[list[str]]: + return [line.split("\t") for line in post(sql).splitlines() if line] + + +def withhold_columns(create_sql: str, drop: set[str]) -> str: + """The snapshot statement minus `drop`, i.e. the table's pre-upgrade shape. + + Columns named by ENGINE/ORDER BY/SETTINGS are kept regardless, so the + reduced statement stays valid. + """ + head, rest = create_sql.split("(\n", 1) + close = rest.index("\n)") + body, tail = rest[:close], rest[close + 2 :] + protected = set(re.findall(r"\b(\w+)\b", tail)) + kept = [ + line.rstrip().rstrip(",") + for line in body.splitlines() + if line.strip() + and not ((m := _COLUMN.match(line)) and m.group("name") in drop and m.group("name") not in protected) + ] + return (head + "(\n" + ",\n".join(kept) + "\n)" + tail).replace("IF NOT EXISTS ", "").rstrip(";") + + +@pytest.fixture +def warm_cluster(): + """A cluster whose tables predate the columns their staging models read.""" + snapshot = {(t.database, t.table): t for t in rbs.load_snapshot_tables(DDL_DIR)} + created = [] + for key, drop in WITHHELD.items(): + table = snapshot.get(key) + assert table is not None, f"{key} missing from the snapshot — update this test" + database, name = key + post(f"CREATE DATABASE IF NOT EXISTS `{database}`") + post(f"DROP TABLE IF EXISTS `{database}`.`{name}`") + post(withhold_columns(table.create_sql, drop)) + first = rows( + f"SELECT name FROM system.columns WHERE database='{database}' AND table='{name}' ORDER BY position" + )[0][0] + post(f"INSERT INTO `{database}`.`{name}` (`{first}`) VALUES ('legacy-row')") + created.append(key) + yield snapshot + for database, name in created: + post(f"DROP TABLE IF EXISTS `{database}`.`{name}`") + + +def reconcile_all(snapshot): + return rbs.reconcile(snapshot.values(), execute=post, fetch_rows=rows) + + +def columns_of(database: str, table: str) -> set[str]: + return { + row[0] + for row in rows(f"SELECT name FROM system.columns WHERE database='{database}' AND table='{table}'") + } + + +def test_warm_cluster_starts_without_the_columns(warm_cluster): + """Guards the fixture: without this the healing assertions prove nothing.""" + for (database, table), withheld in WITHHELD.items(): + live = columns_of(database, table) + assert withheld - live, f"{database}.{table} already has {withheld} — fixture is not reproducing #1991" + + +def test_reconcile_adds_every_missing_column(warm_cluster): + result = reconcile_all(warm_cluster) + + assert result.columns_added > 0 + for (database, table), withheld in WITHHELD.items(): + live = columns_of(database, table) + missing = withheld - live + assert not missing, f"{database}.{table} still missing {sorted(missing)}" + + +def test_reconcile_preserves_existing_rows(warm_cluster): + reconcile_all(warm_cluster) + + for database, table in WITHHELD: + count = rows(f"SELECT count() FROM `{database}`.`{table}`")[0][0] + assert count == "1", f"{database}.{table} lost its row" + + +def test_reconcile_is_idempotent(warm_cluster): + reconcile_all(warm_cluster) + + second = reconcile_all(warm_cluster) + + assert second.columns_added == 0 + assert second.tables_reconciled == 0 + + +def test_reconcile_leaves_no_probe_tables(warm_cluster): + reconcile_all(warm_cluster) + + leftovers = rows( + "SELECT database, name FROM system.tables " + f"WHERE name LIKE '%{rbs.PROBE_SUFFIX}' FORMAT TSV" + ) + assert leftovers == [] + + +# --------------------------------------------------------------------------- +# Whole-snapshot sweep +# --------------------------------------------------------------------------- +# The tests above cover the tables issue #1991 named. These cover EVERY bronze +# table in the snapshot — 172 across 25 connector databases — each stripped to +# the bare minimum ClickHouse will accept, which is far more drift than a real +# upgrade produces. The assertion is the strong one: after reconcile, a stripped +# table's columns match what a fresh install of that same snapshot statement +# creates, compared by ClickHouse itself rather than by parsing DDL text. + +REFERENCE_SUFFIX = "__ddl_reference" + + +def _split_columns(create_sql: str) -> tuple[str, str, str]: + head, rest = create_sql.split("(\n", 1) + close = rest.index("\n)") + return head, rest[:close], rest[close + 2 :] + + +def strip_to_minimum(create_sql: str) -> tuple[str, set[str]]: + """Withhold every column the ENGINE/ORDER BY/SETTINGS tail does not require. + + Whatever the tail names must stay or the CREATE is invalid; at least one + column is always kept. Returns the reduced statement and the withheld names. + """ + head, body, tail = _split_columns(create_sql) + required = set(re.findall(r"\b(\w+)\b", tail)) + kept, withheld = [], set() + for line in body.splitlines(): + match = _COLUMN.match(line) + if not match: + continue + if match.group("name") in required: + kept.append(line) + else: + withheld.add(match.group("name")) + if not kept: # ORDER BY tuple() protects nothing — keep the first column + first = next(line for line in body.splitlines() if _COLUMN.match(line)) + kept.append(first) + withheld.discard(_COLUMN.match(first).group("name")) + reduced = head + "(\n" + ",\n".join(line.rstrip().rstrip(",") for line in kept) + "\n)" + tail + return reduced.replace("IF NOT EXISTS ", "").rstrip(";"), withheld + + +def retarget(create_sql: str, database: str, table: str) -> str: + """The snapshot statement pointed at a different table name.""" + match = rbs._CREATE_TABLE_RE.match(create_sql) + return f"CREATE TABLE `{database}`.`{table}`" + create_sql[match.end() :].rstrip(";") + + +def columns_with_types(database: str, table: str) -> set[tuple[str, str]]: + return { + (row[0], row[1]) + for row in rows( + f"SELECT name, type FROM system.columns WHERE database='{database}' AND table='{table}' FORMAT TSV" + ) + } + + +@pytest.fixture(scope="module") +def stripped_snapshot(): + """Every bronze table in the snapshot, created stripped to its minimum.""" + tables = rbs.load_snapshot_tables(DDL_DIR) + bronze = [t for t in tables if rbs.is_reconcilable(t)] + assert len(bronze) > 100, f"expected the full snapshot, got {len(bronze)} bronze tables" + + withheld_total = 0 + for table in bronze: + reduced, withheld = strip_to_minimum(table.create_sql) + post(f"CREATE DATABASE IF NOT EXISTS `{table.database}`") + post(f"DROP TABLE IF EXISTS `{table.database}`.`{table.table}`") + post(reduced) + first = rows( + f"SELECT name FROM system.columns WHERE database='{table.database}' " + f"AND table='{table.table}' ORDER BY position" + )[0][0] + post(f"INSERT INTO `{table.database}`.`{table.table}` (`{first}`) VALUES (DEFAULT)") + withheld_total += len(withheld) + + yield {"tables": tables, "bronze": bronze, "withheld_total": withheld_total} + + for table in bronze: + post(f"DROP TABLE IF EXISTS `{table.database}`.`{table.table}`") + post(f"DROP TABLE IF EXISTS `{table.database}`.`{table.table}{REFERENCE_SUFFIX}`") + + +def test_sweep_actually_withholds_columns(stripped_snapshot): + """Guards the sweep fixture — otherwise the healing assertion is vacuous.""" + assert stripped_snapshot["withheld_total"] > 1000, stripped_snapshot["withheld_total"] + + +def test_every_bronze_table_matches_a_fresh_install(stripped_snapshot): + """The core generality claim, across all 25 connector databases.""" + result = rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + assert result.columns_added == stripped_snapshot["withheld_total"] + + mismatched = [] + for table in stripped_snapshot["bronze"]: + reference = f"{table.table}{REFERENCE_SUFFIX}" + post(f"DROP TABLE IF EXISTS `{table.database}`.`{reference}`") + post(retarget(table.create_sql, table.database, reference)) + expected = columns_with_types(table.database, reference) + actual = columns_with_types(table.database, table.table) + if actual != expected: + mismatched.append( + f"{table.database}.{table.table}: missing={sorted(expected - actual)} extra={sorted(actual - expected)}" + ) + assert not mismatched, "\n".join(mismatched) + + +def test_sweep_preserves_every_row(stripped_snapshot): + rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + + empty = [ + f"{t.database}.{t.table}" + for t in stripped_snapshot["bronze"] + if rows(f"SELECT count() FROM `{t.database}`.`{t.table}`")[0][0] != "1" + ] + assert not empty, f"rows lost in: {empty}" + + +def test_sweep_is_idempotent(stripped_snapshot): + rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + + second = rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + + assert second.columns_added == 0 + assert second.type_drift == [] + + +def test_sweep_never_touches_non_bronze_tables(stripped_snapshot): + """silver/insight/identity/person/staging are owned by dbt and the migrations.""" + non_bronze = [t for t in stripped_snapshot["tables"] if not rbs.is_reconcilable(t)] + assert non_bronze, "snapshot should contain non-bronze tables" + created = [] + try: + for table in non_bronze: + reduced, withheld = strip_to_minimum(table.create_sql) + if not withheld: + continue + post(f"CREATE DATABASE IF NOT EXISTS `{table.database}`") + post(f"DROP TABLE IF EXISTS `{table.database}`.`{table.table}`") + post(reduced) + created.append((table, withheld)) + + rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + + widened = [ + f"{t.database}.{t.table}" + for t, withheld in created + if withheld & {name for name, _ in columns_with_types(t.database, t.table)} + ] + assert not widened, f"non-bronze tables were modified: {widened}" + finally: + for table, _ in created: + post(f"DROP TABLE IF EXISTS `{table.database}`.`{table.table}`") + + +def test_sweep_leaves_no_scratch_tables(stripped_snapshot): + rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) + + leftovers = rows( + f"SELECT database, name FROM system.tables WHERE name LIKE '%{rbs.PROBE_SUFFIX}' FORMAT TSV" + ) + assert leftovers == [] diff --git a/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py b/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py new file mode 100644 index 000000000..130502623 --- /dev/null +++ b/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py @@ -0,0 +1,384 @@ +"""Unit tests for the bronze snapshot reconciler (issue #1991). + +The reconciler's I/O is injected, so the whole algorithm is exercised here +against an in-memory stand-in for ClickHouse — no container, no network. The +stand-in answers exactly the three statements the reconciler issues (existence +probe, missing-column diff, type-drift diff) and applies DROP/CREATE/ALTER to +its own state, so assertions are about the schema that results rather than about +SQL strings. +""" + +from __future__ import annotations + +import importlib.util +import re +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import reconcile_bronze_schema as rbs # noqa: E402 +from reconcile_bronze_schema import ( # noqa: E402 + PROBE_SUFFIX, + SnapshotTable, + is_reconcilable, + load_snapshot_tables, + parse_snapshot_tables, + reconcile, +) + +LEGACY_COMMITS = { + "hash": "String", + "date": "String", + "author_email": "Nullable(String)", + "project_key": "Nullable(String)", +} + +SNAPSHOT_COMMITS = """CREATE TABLE IF NOT EXISTS bronze_bitbucket_cloud.commits +( + `hash` String, + `date` String, + `author_email` Nullable(String), + `record_type` Nullable(String), + `generation_id` Nullable(String), + `bucket_id` Nullable(Int64), + `repository_uuid` Nullable(String) +) +ENGINE = ReplacingMergeTree(_airbyte_extracted_at) +ORDER BY hash +SETTINGS allow_nullable_key = 1, index_granularity = 8192;""" + + +class FakeClickHouse: + """Minimal ClickHouse stand-in: tracks {(db, table): {column: type}}.""" + + _COLUMN_RE = re.compile(r"^\s*`(?P[^`]+)`\s+(?P.+?),?\s*$") + + def __init__(self, tables: dict[tuple[str, str], dict[str, str]]) -> None: + self.tables = {key: dict(value) for key, value in tables.items()} + self.executed: list[str] = [] + + # -- injected callables ------------------------------------------------- + def execute(self, sql: str) -> None: + self.executed.append(sql) + if sql.startswith("DROP TABLE IF EXISTS"): + self.tables.pop(self._target(sql), None) + elif sql.startswith("CREATE TABLE"): + self.tables[self._target(sql)] = self._columns_of(sql) + elif sql.startswith("ALTER TABLE"): + key = self._target(sql) + match = re.search(r"ADD COLUMN IF NOT EXISTS `(?P[^`]+)` (?P.+)$", sql) + assert match, sql + self.tables[key].setdefault(match.group("name"), match.group("type")) + else: # pragma: no cover — guards against a silently ignored statement + raise AssertionError(f"unexpected statement: {sql}") + + def fetch_rows(self, sql: str) -> list[list[str]]: + if "FROM system.tables" in sql: + wanted = set(re.findall(r"'([^']+)'", sql)) + return [[db, tbl] for (db, tbl) in sorted(self.tables) if db in wanted] + probe, live = self._diff_targets(sql) + if "NOT IN" in sql: + return [ + [name, ch_type] + for name, ch_type in self.tables.get(probe, {}).items() + if name not in self.tables.get(live, {}) + ] + return [ + [name, ch_type, self.tables[live][name]] + for name, ch_type in sorted(self.tables.get(probe, {}).items()) + if name in self.tables.get(live, {}) and self.tables[live][name] != ch_type + ] + + # -- helpers ------------------------------------------------------------ + @staticmethod + def _target(sql: str) -> tuple[str, str]: + match = re.search(r"`(?P[^`]+)`\.`(?P
[^`]+)`", sql) + assert match, sql + return match.group("db"), match.group("table") + + def _diff_targets(self, sql: str) -> tuple[tuple[str, str], tuple[str, str]]: + names = re.findall(r"table = '([^']+)'", sql) + db = re.findall(r"database = '([^']+)'", sql)[0] + probe = next(n for n in names if n.endswith(PROBE_SUFFIX)) + live = next(n for n in names if not n.endswith(PROBE_SUFFIX)) + return (db, probe), (db, live) + + @classmethod + def _columns_of(cls, create_sql: str) -> dict[str, str]: + columns: dict[str, str] = {} + body = create_sql[create_sql.index("(") + 1 : create_sql.rindex(")")] + for line in body.splitlines(): + match = cls._COLUMN_RE.match(line) + if match: + columns[match.group("name")] = match.group("type").rstrip(",").strip() + return columns + + # -- assertions --------------------------------------------------------- + def alters(self) -> list[str]: + return [s for s in self.executed if s.startswith("ALTER TABLE")] + + +def run(tables, snapshot=(SNAPSHOT_COMMITS,)): + fake = FakeClickHouse(tables) + parsed = [t for sql in snapshot for t in parse_snapshot_tables(sql)] + result = reconcile(parsed, execute=fake.execute, fetch_rows=fake.fetch_rows) + return fake, result + + +class TestSnapshotParsing: + def test_extracts_create_table_statements(self): + tables = parse_snapshot_tables(SNAPSHOT_COMMITS) + assert [(t.database, t.table) for t in tables] == [("bronze_bitbucket_cloud", "commits")] + + def test_skips_databases_views_and_materialized_views(self): + sql = ( + "CREATE DATABASE IF NOT EXISTS `bronze_x`;\n\n" + f"{SNAPSHOT_COMMITS}\n\n" + "CREATE OR REPLACE VIEW insight.v AS SELECT 1;\n\n" + "CREATE MATERIALIZED VIEW IF NOT EXISTS insight.mv REFRESH EVERY 1 HOUR AS SELECT 1;" + ) + assert [t.table for t in parse_snapshot_tables(sql)] == ["commits"] + + def test_preserves_mixed_case_table_names(self): + sql = "CREATE TABLE IF NOT EXISTS bronze_salesforce.OpportunityContactRole\n(\n `Id` String\n)\nENGINE = MergeTree\nORDER BY Id;" + assert parse_snapshot_tables(sql)[0].table == "OpportunityContactRole" + + def test_reads_every_snapshot_file(self, tmp_path): + (tmp_path / "a.sql").write_text(SNAPSHOT_COMMITS) + (tmp_path / "b.sql").write_text( + "CREATE TABLE IF NOT EXISTS bronze_other.things\n(\n `id` String\n)\nENGINE = MergeTree\nORDER BY id;" + ) + assert {t.table for t in load_snapshot_tables(tmp_path)} == {"commits", "things"} + + +class TestScope: + def test_only_bronze_databases(self): + assert is_reconcilable(SnapshotTable("bronze_x", "t", "")) is True + for database in ("silver", "insight", "staging", "identity", "person"): + assert is_reconcilable(SnapshotTable(database, "t", "")) is False + + def test_never_reconciles_a_probe_leftover(self): + assert is_reconcilable(SnapshotTable("bronze_x", f"t{PROBE_SUFFIX}", "")) is False + + def test_non_bronze_tables_are_untouched(self): + snapshot = SNAPSHOT_COMMITS.replace("bronze_bitbucket_cloud.commits", "silver.class_git_commits") + fake, result = run({("silver", "class_git_commits"): {"hash": "String"}}, snapshot=(snapshot,)) + assert fake.alters() == [] + assert result.tables_examined == 0 + + +class TestProbeStatement: + def test_retargets_name_and_preserves_the_rest(self): + table = parse_snapshot_tables(SNAPSHOT_COMMITS)[0] + probe_sql = table.probe_sql() + assert probe_sql.startswith("CREATE TABLE `bronze_bitbucket_cloud`.`commits__ddl_probe`") + assert "IF NOT EXISTS" not in probe_sql.splitlines()[0] + assert "ENGINE = ReplacingMergeTree(_airbyte_extracted_at)" in probe_sql + assert "SETTINGS allow_nullable_key = 1, index_granularity = 8192" in probe_sql + assert "`bucket_id` Nullable(Int64)" in probe_sql + + +class TestReconcile: + def test_adds_columns_missing_from_a_legacy_table(self): + """The #1991 case: a pre-envelope table gains bucket_id and friends.""" + fake, result = run({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + live = fake.tables[("bronze_bitbucket_cloud", "commits")] + for column in ("record_type", "generation_id", "bucket_id", "repository_uuid"): + assert column in live, column + assert live["bucket_id"] == "Nullable(Int64)" + assert result.columns_added == 4 + assert result.tables_reconciled == 1 + + def test_keeps_live_columns_absent_from_the_snapshot(self): + fake, _ = run({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + # project_key is a legacy column the snapshot no longer declares. + assert "project_key" in fake.tables[("bronze_bitbucket_cloud", "commits")] + + def test_absent_table_is_left_to_the_snapshots_own_create(self): + fake, result = run({}) + assert fake.alters() == [] + assert result.tables_examined == 0 + + def test_is_idempotent(self): + first, first_result = run({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + healed = first.tables[("bronze_bitbucket_cloud", "commits")] + second, second_result = run({("bronze_bitbucket_cloud", "commits"): healed}) + assert first_result.columns_added == 4 + assert second_result.columns_added == 0 + assert second.alters() == [] + + def test_reports_type_drift_without_modifying_it(self): + drifted = dict(LEGACY_COMMITS, author_email="String") + fake, result = run({("bronze_bitbucket_cloud", "commits"): drifted}) + assert result.type_drift == [ + ("bronze_bitbucket_cloud.commits", "author_email", "Nullable(String)", "String") + ] + assert fake.tables[("bronze_bitbucket_cloud", "commits")]["author_email"] == "String" + assert not any("MODIFY" in sql for sql in fake.executed) + + def test_drops_the_probe_when_done(self): + fake, _ = run({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + assert ("bronze_bitbucket_cloud", f"commits{PROBE_SUFFIX}") not in fake.tables + assert fake.executed[-1].startswith("DROP TABLE IF EXISTS") + + def test_drops_the_probe_even_when_a_statement_fails(self): + fake = FakeClickHouse({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + real_execute = fake.execute + + def flaky(sql: str) -> None: + if sql.startswith("ALTER TABLE"): + raise RuntimeError("boom") + real_execute(sql) + + with pytest.raises(RuntimeError): + reconcile( + parse_snapshot_tables(SNAPSHOT_COMMITS), + execute=flaky, + fetch_rows=fake.fetch_rows, + ) + assert fake.executed[-1].startswith("DROP TABLE IF EXISTS") + assert ("bronze_bitbucket_cloud", f"commits{PROBE_SUFFIX}") not in fake.tables + + def test_alter_statements_quote_identifiers(self): + fake, _ = run({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + for sql in fake.alters(): + assert sql.startswith("ALTER TABLE `bronze_bitbucket_cloud`.`commits` ADD COLUMN IF NOT EXISTS `") + + +class TestLoadableByPath: + """The e2e rig loads this file by path, not as an installed module. + + Regression test: `module_from_spec` + `exec_module` without registering the + module in sys.modules first raises `AttributeError: 'NoneType' object has no + attribute '__dict__'` on the first @dataclass, because dataclass resolves its + own module through `sys.modules[cls.__module__]`. Importing normally (as the + tests above do) hides that, so this exercises the rig's actual code path. + """ + + def _load(self, name: str): + spec = importlib.util.spec_from_file_location( + name, Path(__file__).resolve().parent.parent / "reconcile_bronze_schema.py" + ) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + except Exception: + sys.modules.pop(spec.name, None) + raise + return module + + def test_executes_and_its_dataclasses_are_usable(self): + module = self._load("reconcile_bronze_schema_pathloaded") + try: + table = module.SnapshotTable("bronze_x", "t", SNAPSHOT_COMMITS) + assert table.probe == f"t{module.PROBE_SUFFIX}" + assert module.ReconcileResult().columns_added == 0 + finally: + sys.modules.pop("reconcile_bronze_schema_pathloaded", None) + + def test_reconciles_when_loaded_by_path(self): + """End-to-end through the path-loaded module, as the rig calls it.""" + module = self._load("reconcile_bronze_schema_pathloaded2") + try: + fake = FakeClickHouse({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + result = module.reconcile( + module.parse_snapshot_tables(SNAPSHOT_COMMITS), + execute=fake.execute, + fetch_rows=fake.fetch_rows, + ) + assert result.columns_added == 4 + assert "bucket_id" in fake.tables[("bronze_bitbucket_cloud", "commits")] + finally: + sys.modules.pop("reconcile_bronze_schema_pathloaded2", None) + + +class TestHttpClient: + """The transport the deploy Job uses — mirrors lib/ch-exec.sh.""" + + ENV = { + "CLICKHOUSE_URL": "http://ch:8123/", + "CLICKHOUSE_USER": "insight", + "CLICKHOUSE_PASSWORD": "secret", + } + + def _stub_urlopen(self, monkeypatch, body: str, seen: list): + class Response: + def __enter__(self_inner): + return self_inner + + def __exit__(self_inner, *exc): + return False + + def read(self_inner): + return body.encode("utf-8") + + def urlopen(request): + seen.append(request) + return Response() + + monkeypatch.setattr(rbs.urllib.request, "urlopen", urlopen) + + def test_requires_every_credential(self, monkeypatch): + for missing in self.ENV: + for key, value in self.ENV.items(): + monkeypatch.setenv(key, value) + monkeypatch.delenv(missing) + with pytest.raises(SystemExit, match=missing): + rbs._http_client() + + def test_posts_credentials_as_headers_not_query(self, monkeypatch): + for key, value in self.ENV.items(): + monkeypatch.setenv(key, value) + seen: list = [] + self._stub_urlopen(monkeypatch, "", seen) + + execute, _ = rbs._http_client() + execute("ALTER TABLE x ADD COLUMN y String") + + request = seen[0] + assert request.full_url == "http://ch:8123/" + assert request.data == b"ALTER TABLE x ADD COLUMN y String" + assert request.get_header("X-clickhouse-user") == "insight" + assert request.get_header("X-clickhouse-key") == "secret" + + def test_fetch_rows_parses_tsv_and_drops_blank_lines(self, monkeypatch): + for key, value in self.ENV.items(): + monkeypatch.setenv(key, value) + self._stub_urlopen(monkeypatch, "bucket_id\tNullable(Int64)\n\nrepo\tString\n", []) + + _, fetch_rows = rbs._http_client() + + assert fetch_rows("SELECT 1") == [["bucket_id", "Nullable(Int64)"], ["repo", "String"]] + + +class TestCli: + def test_reconciles_the_given_snapshot_directory(self, monkeypatch, tmp_path, caplog): + (tmp_path / "bitbucket-cloud.sql").write_text(SNAPSHOT_COMMITS) + fake = FakeClickHouse({("bronze_bitbucket_cloud", "commits"): LEGACY_COMMITS}) + monkeypatch.setattr(rbs, "_http_client", lambda: (fake.execute, fake.fetch_rows)) + + with caplog.at_level("INFO", logger="reconcile_bronze_schema"): + assert rbs.main([str(tmp_path)]) == 0 + + assert "bucket_id" in fake.tables[("bronze_bitbucket_cloud", "commits")] + assert "reconciled 4 column(s)" in caplog.text + + def test_warns_about_type_drift(self, monkeypatch, tmp_path, caplog): + (tmp_path / "bitbucket-cloud.sql").write_text(SNAPSHOT_COMMITS) + fake = FakeClickHouse( + {("bronze_bitbucket_cloud", "commits"): dict(LEGACY_COMMITS, author_email="String")} + ) + monkeypatch.setattr(rbs, "_http_client", lambda: (fake.execute, fake.fetch_rows)) + + with caplog.at_level("WARNING", logger="reconcile_bronze_schema"): + rbs.main([str(tmp_path)]) + + assert "differ in type" in caplog.text + + def test_missing_snapshot_directory_is_fatal(self, tmp_path): + with pytest.raises(SystemExit, match="not found"): + rbs.main([str(tmp_path / "nope")]) diff --git a/src/ingestion/tests/e2e/lib/migration_applier.py b/src/ingestion/tests/e2e/lib/migration_applier.py index 9f545d2c9..3615c52d9 100644 --- a/src/ingestion/tests/e2e/lib/migration_applier.py +++ b/src/ingestion/tests/e2e/lib/migration_applier.py @@ -21,8 +21,11 @@ from __future__ import annotations +import importlib.util import logging import re +import sys +from functools import lru_cache from pathlib import Path from lib import clickhouse as ch @@ -112,9 +115,61 @@ def apply_bronze_placeholders(cfg: SessionConfig) -> int: summary = "\n".join(f" {s[:120]!r}: {e}" for s, e in failed[:5]) raise RuntimeError(f"DDL snapshot stuck; {len(failed)} statement(s) keep failing:\n{summary}") pending = [s for s, _ in failed] + + reconcile_bronze_schema(cfg, ddl_dir) return applied +def reconcile_bronze_schema(cfg: SessionConfig, ddl_dir: Path) -> int: + """Add snapshot columns missing from pre-existing bronze tables. + + Mirrors the phase prod runs at the end of create-bronze-placeholders.sh, by + importing the same module rather than reimplementing it — the rig's + ClickHouse outlives a single run (compose volume, and CI reuses the service + across fixtures), so it accumulates exactly the schema drift #1991 is about. + """ + reconciler = _reconciler(cfg.repo_root / "src/ingestion/scripts/reconcile_bronze_schema.py") + result = reconciler.reconcile( + reconciler.load_snapshot_tables(ddl_dir), + execute=lambda sql: ch.execute(cfg, sql), + fetch_rows=lambda sql: [[str(cell) for cell in row] for row in ch.query(cfg, sql)], + ) + if result.columns_added: + LOG.info( + "reconciled %d bronze column(s) across %d table(s)", + result.columns_added, + result.tables_reconciled, + ) + for qualified, name, snapshot_type, live_type in result.type_drift: + LOG.warning( + "%s.%s type differs — snapshot=%s live=%s (left unchanged)", + qualified, + name, + snapshot_type, + live_type, + ) + return result.columns_added + + +@lru_cache(maxsize=1) +def _reconciler(path: Path): + """Load scripts/reconcile_bronze_schema.py, which lives outside the rig's package root. + + The module must be registered in sys.modules BEFORE exec_module: dataclass + resolves its own module via `sys.modules[cls.__module__]`, so executing an + unregistered module raises AttributeError on the first @dataclass. Loading + the file by path (rather than putting scripts/ on sys.path) keeps the rig's + own `tests` package from being shadowed by the one next to the script. + """ + spec = importlib.util.spec_from_file_location("reconcile_bronze_schema", path) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot load the bronze reconciler from {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + def discover_refreshable_views(cfg: SessionConfig) -> list[str]: """Auto-discover every refreshable MV via `system.view_refreshes`.