fix(ingestion): reconcile pre-existing bronze tables to the DDL snapshot - #1993
Conversation
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 <table>__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 #1991 Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
📝 WalkthroughWalkthroughAdds a ClickHouse bronze schema reconciler that compares existing tables with connector DDL snapshots, adds missing columns, reports type drift, integrates into migration flows, and adds unit plus opt-in live ClickHouse tests. ChangesBronze schema reconciliation
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant MigrationApplier
participant DDLApplicator
participant Reconciler
participant ClickHouse
MigrationApplier->>DDLApplicator: Apply connectors-ddl snapshot
DDLApplicator-->>MigrationApplier: Complete DDL application
MigrationApplier->>Reconciler: Load snapshot and reconcile bronze tables
Reconciler->>ClickHouse: Create probe and query system.columns
ClickHouse-->>Reconciler: Return live and probe schemas
Reconciler->>ClickHouse: Add missing columns and drop probe
Reconciler-->>MigrationApplier: Return reconciliation counts and type drift
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| headers={"X-ClickHouse-User": user, "X-ClickHouse-Key": password}, | ||
| method="POST", | ||
| ) | ||
| with urllib.request.urlopen(request) as response: # noqa: S310 |
| headers={"X-ClickHouse-User": CH_USER, "X-ClickHouse-Key": CH_PASSWORD}, | ||
| ) | ||
| try: | ||
| with urllib.request.urlopen(request) as response: # noqa: S310 |
|
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 <pm@aleks.bar>
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 <pm@aleks.bar>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/ingestion/scripts/pyproject.toml (1)
1-24: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAdd the required
cf-studio-pathsetting.This TOML configuration omits
cf-studio-path = ".cf-studio". As per coding guidelines,**/*.toml: Setcf-studio-pathto.cf-studioin TOML configuration.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/scripts/pyproject.toml` around lines 1 - 24, Add the required cf-studio-path setting with the value ".cf-studio" to the pyproject.toml configuration, preserving the existing build, project, pytest, and setuptools settings.Source: Coding guidelines
🧹 Nitpick comments (1)
src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py (1)
119-122: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePrefer
VALUES (DEFAULT)for the seed row, as the sweep fixture does.Line 122 assumes the first ordinal column accepts a String literal.
stripped_snapshot(Line 260) already usesDEFAULTfor the same purpose; matching it removes the type assumption if the snapshot's column order ever changes.♻️ Proposed change
- post(f"INSERT INTO `{database}`.`{name}` (`{first}`) VALUES ('legacy-row')") + post(f"INSERT INTO `{database}`.`{name}` (`{first}`) VALUES (DEFAULT)")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py` around lines 119 - 122, Update the seed INSERT in the reconciliation test to use VALUES (DEFAULT) instead of selecting the first column and inserting a String literal. Remove the unnecessary first-column lookup while preserving the existing database/table targeting and legacy-row seed behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/ingestion/scripts/reconcile_bronze_schema.py`:
- Around line 279-287: Update the _post helper’s urllib.request.urlopen call to
pass an explicit finite timeout for the ClickHouse HTTP request, using the
module’s existing timeout configuration or an appropriate constant. Preserve the
current request construction and response decoding behavior.
In `@src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py`:
- Around line 275-278: Remove the order-dependent exact-equality assertion on
stripped_snapshot["withheld_total"] from
test_every_bronze_table_matches_a_fresh_install, while retaining the per-table
comparisons as the order-independent generality check. If the total must still
be validated, move that guard into the stripped_snapshot fixture before shared
reconciliation can mutate the tables.
---
Outside diff comments:
In `@src/ingestion/scripts/pyproject.toml`:
- Around line 1-24: Add the required cf-studio-path setting with the value
".cf-studio" to the pyproject.toml configuration, preserving the existing build,
project, pytest, and setuptools settings.
---
Nitpick comments:
In `@src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py`:
- Around line 119-122: Update the seed INSERT in the reconciliation test to use
VALUES (DEFAULT) instead of selecting the first column and inserting a String
literal. Remove the unnecessary first-column lookup while preserving the
existing database/table targeting and legacy-row seed behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 636091af-c8c3-45dc-afe1-fa8bd9ef89d3
📒 Files selected for processing (8)
scripts/ci/components.pysrc/ingestion/scripts/apply-ch-migrations.shsrc/ingestion/scripts/create-bronze-placeholders.shsrc/ingestion/scripts/pyproject.tomlsrc/ingestion/scripts/reconcile_bronze_schema.pysrc/ingestion/scripts/tests/test_reconcile_against_clickhouse.pysrc/ingestion/scripts/tests/test_reconcile_bronze_schema.pysrc/ingestion/tests/e2e/lib/migration_applier.py
💤 Files with no reviewable changes (1)
- src/ingestion/scripts/apply-ch-migrations.sh
| 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") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate file =="
fd -a 'reconcile_bronze_schema.py' . || true
echo "== relevant file context =="
if [ -f src/ingestion/scripts/reconcile_bronze_schema.py ]; then
wc -l src/ingestion/scripts/reconcile_bronze_schema.py
sed -n '240,310p' src/ingestion/scripts/reconcile_bronze_schema.py
echo "== timeout/http usage =="
rg -n "urlopen|timeout|HTTP|clickhouse|endpoint|_post" src/ingestion/scripts/reconcile_bronze_schema.py || true
fi
echo "== any timeout constants nearby =="
rg -n "timeout\s*=" . -g '*.py' || trueRepository: constructorfabric/insight
Length of output: 8621
🌐 Web query:
Python urllib.request.urlopen default timeout None blocks indefinitely
💡 Result:
In Python's urllib.request.urlopen, the timeout parameter defaults to socket._GLOBAL_DEFAULT_TIMEOUT [1]. While the official documentation states that "if not specified, the global default timeout setting will be used," [2][3][4] this refers to the setting established by socket.setdefaulttimeout [5]. If socket.setdefaulttimeout has not been called in your application, the global default timeout is None, which means sockets will block indefinitely [5][6]. To ensure your request does not block indefinitely, you should always explicitly provide a timeout value in seconds to the urlopen function [5]. Example: import urllib.request # Explicitly setting a 10-second timeout try: with urllib.request.urlopen('http://example.com', timeout=10) as response: print(response.read) except urllib.error.URLError as e: print(f"Request failed: {e}")
Citations:
- 1: https://github.com/python/cpython/blob/bff98fe5364452775d8273137c9087ed004deb96/Lib/urllib/request.py
- 2: https://docs.python.org/3/library/urllib.request.html
- 3: https://docs.python.org/3.11/library/urllib.request.html
- 4: https://docs.python.org/3.14/library/urllib.request.html
- 5: https://docs.python.org/3.11/howto/urllib2.html
- 6: https://bugs.python.org/issue18417
Bound ClickHouse HTTP calls with an explicit timeout.
urlopen(request) can use the global socket timeout, and if no timeout is set it blocks indefinitely; this can stall reconciliation and prevent deployment completion.
Proposed fix
+HTTP_TIMEOUT_SECONDS = 30
+
- with urllib.request.urlopen(request) as response: # noqa: S310
+ with urllib.request.urlopen(
+ request, timeout=HTTP_TIMEOUT_SECONDS
+ ) as response: # noqa: S310📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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") | |
| HTTP_TIMEOUT_SECONDS = 30 | |
| 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, timeout=HTTP_TIMEOUT_SECONDS | |
| ) as response: # noqa: S310 | |
| return response.read().decode("utf-8") |
🧰 Tools
🪛 ast-grep (0.44.1)
[warning] 285-285: Request-controlled URL passed to urlopen; validate against an allowlist to prevent SSRF.
Context: urllib.request.urlopen(request)
Note: [CWE-918] Server-Side Request Forgery (SSRF).
(urlopen-unsanitized-data)
🪛 GitHub Check: Semgrep OSS
[warning] 286-286: Semgrep Finding: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
Detected a dynamic value being used with urllib. urllib supports 'file://' schemes, so a dynamic value controlled by a malicious actor may allow them to read arbitrary files. Audit uses of urllib calls to ensure user data cannot control the URLs, or consider using the 'requests' library instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/scripts/reconcile_bronze_schema.py` around lines 279 - 287,
Update the _post helper’s urllib.request.urlopen call to pass an explicit finite
timeout for the ClickHouse HTTP request, using the module’s existing timeout
configuration or an appropriate constant. Preserve the current request
construction and response decoding behavior.
| 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"] |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Line 278 silently depends on this being the first reconciling test in the module.
stripped_snapshot is module-scoped, and every other sweep test calls rbs.reconcile(...) on the same shared tables. Once any of them runs first, the tables are already healed and columns_added is 0, so the exact-equality assertion fails. It passes today only because of declaration order — reordering (e.g. pytest-randomly, or --dist load splitting the module across xdist workers) breaks it with a confusing failure.
The per-table comparison on Lines 280-291 is the real generality claim and is order-independent; consider keeping the total as a fixture-level guard instead of an in-test assertion.
♻️ Proposed change
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"]
+ # Order-independent: earlier sweep tests may already have healed the tables,
+ # so only the resulting schema is asserted here. `test_sweep_actually_withholds_columns`
+ # guards the withheld total.
+ rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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"] | |
| def test_every_bronze_table_matches_a_fresh_install(stripped_snapshot): | |
| """The core generality claim, across all 25 connector databases.""" | |
| # Order-independent: earlier sweep tests may already have healed the tables, | |
| # so only the resulting schema is asserted here. `test_sweep_actually_withholds_columns` | |
| # guards the withheld total. | |
| rbs.reconcile(stripped_snapshot["tables"], execute=post, fetch_rows=rows) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/ingestion/scripts/tests/test_reconcile_against_clickhouse.py` around
lines 275 - 278, Remove the order-dependent exact-equality assertion on
stripped_snapshot["withheld_total"] from
test_every_bronze_table_matches_a_fresh_install, while retaining the per-table
comparisons as the order-independent generality check. If the total must still
be validated, move that guard into the stripped_snapshot fixture before shared
reconciliation can mutate the tables.
3 participants
Problem
Upgrading a warm installation freezes a whole metrics domain. The snapshot applicator only issues
CREATE TABLE IF NOT EXISTS, which is a no-op against an existing table, 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:and every downstream
class_git_*/fct_git_*/ gold model is skipped.The existing warm-cluster heal covered 2 of the 7 affected bitbucket tables. That is the flaw: a hand-maintained per-table ALTER list, which drifted.
bronze_outline.wiki_usershad no heal at all.Fix
A generic reconcile phase: for every table in the snapshot that already exists, add the columns the snapshot declares and the live table lacks.
Schema introspection is delegated to ClickHouse rather than parsed — each snapshot statement is replayed as an empty
<table>__ddl_probeand the missing set comes fromsystem.columns, so no type spelling is ever compared by hand. The snapshot is regenerated from real connector output, so it is the same source of truth that creates fresh tables — it cannot drift from what a new install gets.This fixes all 6 items on the issue's checklist and the identical latent exposure in every other connector (hubspot 19 tables, jira 14, gitlab 12, salesforce 10 all sit behind the same no-op CREATE today).
Guarantees
bronze_*only — silver/insight/staging/identity/person are owned by dbt or the numbered migrations and have their own heal semanticsADD COLUMN IF NOT EXISTSis metadata-only; probes are separate tables — nothing rewrites or swaps a live table, so a mid-deploy Airbyte sync is fineSemantics worth knowing: healed legacy rows hold NULL in the new envelope columns, and the staging models'
record_type = 'item'gating naturally excludes them — so a stream's silver may stay empty until the next sync writes new-envelope rows. Same semantics as the heal this replaces; no data is lost.heal_bitbucket_commits/heal_bitbucket_pull_requests(109 lines of hand-written ALTERs) are removed as superseded.One implementation, shared by the deploy Job (as a CLI) and the e2e rig (as an import) — the toolbox image is already
python:3.12. Two parallel implementations would have been the same bug class this PR fixes.Testing
Three layers, because the bug is about real schema state:
RECONCILE_TEST_CH_URLso CI stays dependency-free; they skip otherwise.ingestion-scriptscoverage component — runs the unit tests on every PR at 98% line coverage. It owns the wholescripts/tree, so aconnectors-ddlregeneration re-runs them too: the reconciler's contract is with that snapshot.Plus: every e2e lane now exercises the reconcile via
apply_all().Verified end-to-end on ClickHouse 25.7.5
Replaying the committed snapshot with the post-upgrade columns withheld reproduces an upgraded install exactly, and:
dbt buildfails with the reportedUnknown expression identifier bucket_id1/1for all 20 bitbucket tables, seeded rows preservedreconciled 0 column(s)(idempotent)dbt build --select tag:bitbucket-cloud+— 89/89 PASS, the graph that was frozenoutline__users_snapshotbuilds (its remainingnot_nullfailure is my synthetic seed row leavingunique_keyNULL, not a product issue)Run it yourself:
Out of scope
Closes #1991
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Tests