From d64b7d1516de9dc92627069cf101c289ae27c913 Mon Sep 17 00:00:00 2001 From: Aleksandr Barkhatov Date: Tue, 28 Jul 2026 12:07:50 +0200 Subject: [PATCH 1/3] fix(ingestion): reconcile pre-existing bronze tables to the DDL snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The snapshot applicator only issues CREATE TABLE IF NOT EXISTS, which is a no-op against a table that already exists, so a warm cluster keeps whatever 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 every downstream model is skipped — an upgrade froze the whole git metrics domain because five bitbucket tables and bronze_outline.wiki_users predated their current schema. Add a reconcile phase: for every snapshot table that already exists, add the columns the snapshot declares and the live table lacks. Schema introspection is delegated to ClickHouse — each statement is replayed as an empty __ddl_probe and the diff comes from system.columns — so no type spelling is compared by hand. The snapshot is regenerated from real connector output, so it is the same source of truth that creates fresh tables; the hand-maintained per-table ALTER list it replaces is what drifted and caused this. Scope: bronze_* only (silver/insight/staging/identity/person are owned by dbt or the numbered migrations). ADD COLUMN only — a differing type is reported and left alone, and live columns absent from the snapshot are never dropped. Idempotent, and safe alongside a running sync since ADD COLUMN IF NOT EXISTS is metadata-only and the probes are separate tables. This supersedes heal_bitbucket_commits/heal_bitbucket_pull_requests, which covered two of the seven affected tables; both are removed. One implementation, shared by the deploy Job (as a CLI) and the e2e rig (as an import), so the two cannot diverge — the toolbox image is already python:3.12. Tests: 22 unit tests drive the algorithm through an in-memory ClickHouse stand-in, plus 5 integration tests that run it against a real server, opt-in via RECONCILE_TEST_CH_URL so CI stays dependency-free. A new ingestion-scripts coverage component runs them on every PR (98% line coverage), and it owns the whole scripts/ tree so a connectors-ddl regeneration re-runs them too. Verified on ClickHouse 25.7.5: replaying the snapshot with the post-upgrade columns withheld reproduces the reported failure, the phase heals all 77 columns across the 8 tables while preserving their rows, a second run is a no-op, and dbt build --select tag:bitbucket-cloud+ then completes 89/89. Closes constructorfabric/insight#1991 Signed-off-by: Aleksandr Barkhatov --- scripts/ci/components.py | 12 + src/ingestion/scripts/apply-ch-migrations.sh | 109 ------ .../scripts/create-bronze-placeholders.sh | 12 + src/ingestion/scripts/pyproject.toml | 23 ++ .../scripts/reconcile_bronze_schema.py | 325 +++++++++++++++++ .../test_reconcile_against_clickhouse.py | 181 ++++++++++ .../tests/test_reconcile_bronze_schema.py | 335 ++++++++++++++++++ .../tests/e2e/lib/migration_applier.py | 46 +++ 8 files changed, 934 insertions(+), 109 deletions(-) create mode 100644 src/ingestion/scripts/pyproject.toml create mode 100644 src/ingestion/scripts/reconcile_bronze_schema.py create mode 100644 src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py create mode 100644 src/ingestion/scripts/tests/test_reconcile_bronze_schema.py 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..ac8226608 --- /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}) FORMAT TSV" + ) + 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 FORMAT TSV" + ) + 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 FORMAT TSV" + ) + 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..07639cd23 --- /dev/null +++ b/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py @@ -0,0 +1,181 @@ +"""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 == [] 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..37b97c230 --- /dev/null +++ b/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py @@ -0,0 +1,335 @@ +"""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 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 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..6dd6f89b9 100644 --- a/src/ingestion/tests/e2e/lib/migration_applier.py +++ b/src/ingestion/tests/e2e/lib/migration_applier.py @@ -21,8 +21,10 @@ from __future__ import annotations +import importlib.util import logging import re +from functools import lru_cache from pathlib import Path from lib import clickhouse as ch @@ -112,9 +114,53 @@ 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.""" + 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) + spec.loader.exec_module(module) + return module + + def discover_refreshable_views(cfg: SessionConfig) -> list[str]: """Auto-discover every refreshable MV via `system.view_refreshes`. From dde9f650f172ca4a57fc12ee3fd24c465e38b9b7 Mon Sep 17 00:00:00 2001 From: Aleksandr Barkhatov Date: Tue, 28 Jul 2026 12:22:21 +0200 Subject: [PATCH 2/3] test(ingestion): sweep every bronze table in the snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The integration tests covered the eight tables issue #1991 named. Extend them to all 172 bronze tables across the 25 connector databases: each is created stripped to the minimum ClickHouse accepts (only what the ENGINE/ORDER BY tail requires), which withholds 3,784 columns — far more drift than a real upgrade produces. The assertion is the general one: after reconcile, a stripped table's columns and types match what the same snapshot statement creates on a fresh install, compared via system.columns rather than by parsing DDL, so column ordering (ADD COLUMN appends) is correctly ignored while names and types must match exactly. This exercises every column type bronze actually uses — String, Bool, Int64, Decimal, UInt32, DateTime64, Date — and the mixed-case Salesforce table names, none of which the eight-table fixture reached. Added alongside: rows preserved across all 172, a second pass adding nothing, no scratch tables left behind, and a check that the 62 non-bronze snapshot tables (silver/insight/identity/person) are left alone even when they are missing columns too. A guard test asserts the sweep really withholds columns, so the healing assertions cannot pass vacuously. Signed-off-by: Aleksandr Barkhatov --- .../test_reconcile_against_clickhouse.py | 167 ++++++++++++++++++ 1 file changed, 167 insertions(+) diff --git a/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py b/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py index 07639cd23..c5bc0a39f 100644 --- a/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py +++ b/src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py @@ -179,3 +179,170 @@ def test_reconcile_leaves_no_probe_tables(warm_cluster): 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 == [] From 51b27240b713c40620f524efd132f62b7ad0fc8f Mon Sep 17 00:00:00 2001 From: Aleksandr Barkhatov Date: Tue, 28 Jul 2026 13:13:26 +0200 Subject: [PATCH 3/3] fix(ingestion): make the reconciler work through the e2e rig's client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in the rig integration, both caught by CI's api/metrics lanes failing at session start. Neither was reachable from the unit tests or the raw-HTTP integration tests, which is why they shipped. 1. The by-path module load did not register the module in sys.modules before exec_module. dataclass resolves its own module via sys.modules[__module__], so the first @dataclass raised "AttributeError: 'NoneType' object has no attribute '__dict__'". Importing the module normally — what every existing test did — hides this. 2. The introspection queries carried an explicit FORMAT TSV. clickhouse-connect appends its own FORMAT Native, so the rig sent "FORMAT TSV FORMAT Native" and ClickHouse rejected it (code 62). The suffix was never needed: the HTTP interface already defaults to TabSeparated, which is what the CLI's fetch_rows parses. Regression cover for the first: a test that loads the file by path exactly as the rig does and then exercises a dataclass and a full reconcile, so the sys.modules requirement is pinned rather than implicit. The second is covered by the e2e lanes themselves — they exercise the rig client on every run, and a FORMAT mismatch cannot pass them. Verified against ClickHouse 25.7.5 through the rig's own clickhouse-connect client: lib.migration_applier.reconcile_bronze_schema heals a legacy table (39 columns), preserves its rows, and is a no-op on a second call; the full apply_all() session bootstrap — the call that failed in CI — completes 245 migration statements with the drift healed. The 172-table sweep and the unit suite still pass (35 with a live server, 24 + 11 skipped without). Signed-off-by: Aleksandr Barkhatov --- .../scripts/reconcile_bronze_schema.py | 6 +-- .../tests/test_reconcile_bronze_schema.py | 49 +++++++++++++++++++ .../tests/e2e/lib/migration_applier.py | 11 ++++- 3 files changed, 62 insertions(+), 4 deletions(-) diff --git a/src/ingestion/scripts/reconcile_bronze_schema.py b/src/ingestion/scripts/reconcile_bronze_schema.py index ac8226608..f493f1fca 100644 --- a/src/ingestion/scripts/reconcile_bronze_schema.py +++ b/src/ingestion/scripts/reconcile_bronze_schema.py @@ -196,7 +196,7 @@ def _existing_tables( databases = ", ".join(sorted({_lit(t.database) for t in tables})) rows = fetch_rows( "SELECT database, name FROM system.tables " - f"WHERE database IN ({databases}) FORMAT TSV" + f"WHERE database IN ({databases})" ) return {(row[0], row[1]) for row in rows if len(row) >= 2} @@ -218,7 +218,7 @@ def _reconcile_one( "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 FORMAT TSV" + "ORDER BY position" ) for row in missing: name, ch_type = row[0], row[1] @@ -236,7 +236,7 @@ def _reconcile_one( 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 FORMAT TSV" + "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] diff --git a/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py b/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py index 37b97c230..130502623 100644 --- a/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py +++ b/src/ingestion/scripts/tests/test_reconcile_bronze_schema.py @@ -10,6 +10,7 @@ from __future__ import annotations +import importlib.util import re import sys from pathlib import Path @@ -247,6 +248,54 @@ def test_alter_statements_quote_identifiers(self): 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.""" diff --git a/src/ingestion/tests/e2e/lib/migration_applier.py b/src/ingestion/tests/e2e/lib/migration_applier.py index 6dd6f89b9..3615c52d9 100644 --- a/src/ingestion/tests/e2e/lib/migration_applier.py +++ b/src/ingestion/tests/e2e/lib/migration_applier.py @@ -24,6 +24,7 @@ import importlib.util import logging import re +import sys from functools import lru_cache from pathlib import Path @@ -152,11 +153,19 @@ def reconcile_bronze_schema(cfg: SessionConfig, ddl_dir: Path) -> int: @lru_cache(maxsize=1) def _reconciler(path: Path): - """Load scripts/reconcile_bronze_schema.py, which lives outside the rig's package root.""" + """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