fix(ingestion): reconcile pre-existing bronze tables to the DDL snapshot

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

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`

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 ==="

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"]