Add Redshift connector to Recon - #2339
Merged
Merged
Conversation
bishwajit-db
force-pushed
the
feature/redshift-recon
branch
from
March 18, 2026 11:19
4a9181b to
611e267
Compare
bishwajit-db
force-pushed
the
feature/redshift-recon
branch
from
March 18, 2026 11:31
611e267 to
fb2d451
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #2339 +/- ##
==========================================
+ Coverage 64.62% 64.77% +0.15%
==========================================
Files 103 104 +1
Lines 9419 9460 +41
Branches 992 992
==========================================
+ Hits 6087 6128 +41
Misses 3156 3156
Partials 176 176 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
✅ 151/151 passed, 5 skipped, 24m11s total Running from acceptance #4348 |
bishwajit-db
force-pushed
the
feature/redshift-recon
branch
from
March 25, 2026 12:59
fb2d451 to
e952d15
Compare
bishwajit-db
force-pushed
the
feature/redshift-recon
branch
from
March 25, 2026 13:01
e952d15 to
b095af4
Compare
m-abulazm
reviewed
Mar 25, 2026
m-abulazm
left a comment
m-abulazm
left a comment
Contributor
There was a problem hiding this comment.
Looks good. we need to run the integration tests to be sure
bishwajit-db
changed the title
BesikiML
self-requested a review
April 8, 2026 13:03
Contributor
SummaryThanks for adding Redshift support to Recon. ScopeThe change set is large relative to the PR title (“Add Redshift connector to Recon”): it includes many unrelated areas (e.g. profiler/Synapse, workflows, broader config/telemetry). For reviewability and release clarity, consider updating the title/description to reflect the full scope. Verify before merge
Suggestions (non-blocking)
|
Contributor
Author
Scope: The PR doesn't seem to affect any other part of the code apart from recon. Please elaborate.
|
Contributor
Thanks for the follow-up — a few clarifications on my earlier review. ScopeLooking at the current branch again, the changes are reconcile-scoped (Redshift connector, adapter wiring, hash/query bits, constants, docs, tests). My comment about the PR being much broader than the title does not apply to this revision; I may have been thinking of an older diff or I mixed it up. No action needed from you on that unless you still have unrelated commits that are not on this PR.
|
m-abulazm
reviewed
May 11, 2026
m-abulazm
requested changes
May 11, 2026
# Conflicts: # tests/integration/reconcile/connectors/test_read_schema.py
…ture/redshift-recon # Conflicts: # tests/integration/reconcile/conftest.py
m-abulazm
approved these changes
May 11, 2026
m-abulazm
enabled auto-merge
May 11, 2026 17:51
10 tasks
ameersalman33
pushed a commit
to ameersalman33/lakebridge
that referenced
this pull request
Jun 2, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon. When `fingerprint_precheck=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Three new fields on `ReconcileConfig`: `fingerprint_precheck`, `fingerprint_treat_empty_as_null`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds two legacy spellings (`redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR databrickslabs#2339) Two correctness bugs in the upstream Redshift connector MR (databrickslabs#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. Both crash the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. Both fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests in `test_expression_generator.py`. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. ### Hardening from the internal review round After the initial internal review on the contributor's fork, three substantive code changes landed before pushing upstream: - **Pin TZ-aware Spark target columns to UTC** before formatting in `fingerprint/spark_target.py`. The Redshift side already pinned UTC via `TO_CHAR(_ AT TIME ZONE 'UTC', _)`; the Spark side was using `DATE_FORMAT(ts, _)` which renders in `spark.sql.session.timeZone`. On a non-UTC cluster the same instant rendered different bytes on the two sides. Fix splits LTZ vs NTZ handling and routes LTZ through `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`. NTZ behaviour unchanged. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; the `build_mismatch_output` exception path was the one inconsistency. Metadata still records `fallback_to_full_pipeline=True` for observability. - **Cast Redshift strings to `VARCHAR(65535)`** in `fingerprint/query_builders/redshift.py` instead of bare `VARCHAR`, whose default 256-byte width truncated long text. `VARCHAR(65535)` is Redshift's maximum and matches Spark's unbounded string semantics. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` (instead of substring-checking two different patterns) and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR databrickslabs#2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass; `tests/unit/reconcile/` runs 282 tests in <1 s. The 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated. - Regression tests added for every review-round fix (UTC pin, fallback path, VARCHAR(65535)); `test_expression_generator.py` pins the exact rendered SQL on each dialect for the two pre-existing fixes. - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (databrickslabs#2453).
ameersalman33
pushed a commit
to ameersalman33/lakebridge
that referenced
this pull request
Jun 2, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon. When `fingerprint_precheck=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Three new fields on `ReconcileConfig`: `fingerprint_precheck`, `fingerprint_treat_empty_as_null`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds two legacy spellings (`redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR databrickslabs#2339) Two correctness bugs in the upstream Redshift connector MR (databrickslabs#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. Both crash the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. Both fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests in `test_expression_generator.py`. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. ### Hardening from the internal review round After the initial internal review on the contributor's fork, three substantive code changes landed before pushing upstream: - **Pin TZ-aware Spark target columns to UTC** before formatting in `fingerprint/spark_target.py`. The Redshift side already pinned UTC via `TO_CHAR(_ AT TIME ZONE 'UTC', _)`; the Spark side was using `DATE_FORMAT(ts, _)` which renders in `spark.sql.session.timeZone`. On a non-UTC cluster the same instant rendered different bytes on the two sides. Fix splits LTZ vs NTZ handling and routes LTZ through `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`. NTZ behaviour unchanged. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; the `build_mismatch_output` exception path was the one inconsistency. Metadata still records `fallback_to_full_pipeline=True` for observability. - **Cast Redshift strings to `VARCHAR(65535)`** in `fingerprint/query_builders/redshift.py` instead of bare `VARCHAR`, whose default 256-byte width truncated long text. `VARCHAR(65535)` is Redshift's maximum and matches Spark's unbounded string semantics. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` (instead of substring-checking two different patterns) and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR databrickslabs#2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass; `tests/unit/reconcile/` runs 282 tests in <1 s. The 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated. - Regression tests added for every review-round fix (UTC pin, fallback path, VARCHAR(65535)); `test_expression_generator.py` pins the exact rendered SQL on each dialect for the two pre-existing fixes. - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (databrickslabs#2453).
10 tasks
ameersalman33
pushed a commit
to ameersalman33/lakebridge
that referenced
this pull request
Jun 3, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon. When `fingerprint_precheck=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Three new fields on `ReconcileConfig`: `fingerprint_precheck`, `fingerprint_treat_empty_as_null`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds two legacy spellings (`redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR databrickslabs#2339) Two correctness bugs in the upstream Redshift connector MR (databrickslabs#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. Both crash the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. Both fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests in `test_expression_generator.py`. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. ### Hardening from the internal review round After the initial internal review on the contributor's fork, three substantive code changes landed before pushing upstream: - **Pin TZ-aware Spark target columns to UTC** before formatting in `fingerprint/spark_target.py`. The Redshift side already pinned UTC via `TO_CHAR(_ AT TIME ZONE 'UTC', _)`; the Spark side was using `DATE_FORMAT(ts, _)` which renders in `spark.sql.session.timeZone`. On a non-UTC cluster the same instant rendered different bytes on the two sides. Fix splits LTZ vs NTZ handling and routes LTZ through `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`. NTZ behaviour unchanged. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; the `build_mismatch_output` exception path was the one inconsistency. Metadata still records `fallback_to_full_pipeline=True` for observability. - **Cast Redshift strings to `VARCHAR(65535)`** in `fingerprint/query_builders/redshift.py` instead of bare `VARCHAR`, whose default 256-byte width truncated long text. `VARCHAR(65535)` is Redshift's maximum and matches Spark's unbounded string semantics. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` (instead of substring-checking two different patterns) and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR databrickslabs#2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass; `tests/unit/reconcile/` runs 282 tests in <1 s. The 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated. - Regression tests added for every review-round fix (UTC pin, fallback path, VARCHAR(65535)); `test_expression_generator.py` pins the exact rendered SQL on each dialect for the two pre-existing fixes. - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (databrickslabs#2453).
ameersalman-db
added a commit
that referenced
this pull request
Jul 14, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon. When `fingerprint_precheck=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `fingerprint_precheck`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds two legacy spellings (`redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Two correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. Both crash the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. Both fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests in `test_expression_generator.py`. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-side UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, so timezone handling matches the compare path on both sides (divergence is impossible rather than guarded) and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 373 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no session-timezone-dependent function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`); the fallback path is pinned; `test_expression_generator.py` pins the exact rendered SQL on each dialect for the two pre-existing fixes. - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
12 tasks
ameersalman-db
added a commit
that referenced
this pull request
Jul 15, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon, exposed as the user-facing flag `reconcile_optimizer`. When `reconcile_optimizer=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `reconcile_optimizer`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds the legacy spellings (`fingerprint_precheck`, `redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new `reconcile_optimizer` flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Three correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. All corrupt the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. All fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. - **Both blocks missing DOUBLE handler.** DOUBLE had no override on either dialect, so both fell through to `TRIM(CAST(col AS string))`. Redshift renders `double precision` at full 17-digit precision (`0.28999999999999998`) while Spark emits the shortest round-trip (`0.29`), so every double-bearing row false-mismatched on a Redshift -> Databricks reconcile. In the fingerprint path this trips the systemic-mismatch guard, so the pre-check defers on any table with a DOUBLE column (a `transformations` override can't rescue it — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to `COALESCE(CAST(CAST(col AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_')`, the same normalization the Teradata recon fixture applied by hand; as a dialect default it fixes row-hash and fingerprint at once. ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-side UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, so timezone handling matches the compare path on both sides (divergence is impossible rather than guarded) and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. - **Typed NULLs in the persisted `fingerprint_metrics` struct.** Optional fields (`verdict`, `target_row_count`, `row_count_source`, `fetch_path`) render as `cast(NULL as string|bigint)` rather than a bare `NULL`. A bare `NULL` makes Spark infer `NullType`, which the vectorized Parquet reader cannot read back and which breaks schema equality against the typed `recon_metrics` table; `IS NULL` dashboard semantics are preserved. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 375 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no session-timezone-dependent function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`); the fallback path is pinned; the fingerprint serialization suites pin the exact rendered SQL on each dialect for the three pre-existing fixes (including the `DOUBLE` -> `DECIMAL(38,10)` normalization on source and target). - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
ameersalman-db
added a commit
that referenced
this pull request
Jul 17, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon, exposed as the user-facing flag `reconcile_optimizer`. When `reconcile_optimizer=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `reconcile_optimizer`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds the legacy spellings (`fingerprint_precheck`, `redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new `reconcile_optimizer` flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Three correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. All corrupt the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. All fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. - **Both blocks missing DOUBLE handler.** DOUBLE had no override on either dialect, so both fell through to `TRIM(CAST(col AS string))`. Redshift renders `double precision` at full 17-digit precision (`0.28999999999999998`) while Spark emits the shortest round-trip (`0.29`), so every double-bearing row false-mismatched on a Redshift -> Databricks reconcile. In the fingerprint path this trips the systemic-mismatch guard, so the pre-check defers on any table with a DOUBLE column (a `transformations` override can't rescue it — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to `COALESCE(CAST(CAST(col AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_')`, the same normalization the Teradata recon fixture applied by hand; as a dialect default it fixes row-hash and fingerprint at once. NaN / +-Infinity are rendered as strings (the numeric cast would otherwise hard-fail on Redshift). ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-column UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Session-level UTC pin for cross-engine timestamp determinism.** The Databricks target renders timestamps via `DATE_FORMAT`, which depends on `spark.sql.session.timeZone`, so `pin_utc_session` pins the session to UTC for the recon. It is gated on the source dialect (`redshift`) — a row-hash correctness concern shared by the plain compare and fingerprint paths, not gated on `reconcile_optimizer` — and the original value is restored once the recon completes, so a shared/interactive cluster sees no lasting change. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. - **Typed NULLs in the persisted `fingerprint_metrics` struct.** Optional fields (`verdict`, `target_row_count`, `row_count_source`, `fetch_path`) render as `cast(NULL as string|bigint)` rather than a bare `NULL`. A bare `NULL` makes Spark infer `NullType`, which the vectorized Parquet reader cannot read back and which breaks schema equality against the typed `recon_metrics` table; `IS NULL` dashboard semantics are preserved. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 375 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no explicit session-timezone function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`), the implicit `DATE_FORMAT` dependence being pinned once by `pin_utc_session` (Redshift-scoped, restored after the recon); the fallback path is pinned; the fingerprint serialization suites pin the exact rendered SQL on each dialect for the three pre-existing fixes (including the `DOUBLE` -> `DECIMAL(38,10)` normalization on source and target). - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
ameersalman-db
added a commit
that referenced
this pull request
Jul 19, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon, exposed as the user-facing flag `reconcile_optimizer`. When `reconcile_optimizer=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `reconcile_optimizer`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds the legacy spellings (`fingerprint_precheck`, `redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new `reconcile_optimizer` flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Three correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. All corrupt the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. All fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. - **Both blocks missing DOUBLE handler.** DOUBLE had no override on either dialect, so both fell through to `TRIM(CAST(col AS string))`. Redshift renders `double precision` at full 17-digit precision (`0.28999999999999998`) while Spark emits the shortest round-trip (`0.29`), so every double-bearing row false-mismatched on a Redshift -> Databricks reconcile. In the fingerprint path this trips the systemic-mismatch guard, so the pre-check defers on any table with a DOUBLE column (a `transformations` override can't rescue it — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to `COALESCE(CAST(CAST(col AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_')`, the same normalization the Teradata recon fixture applied by hand; as a dialect default it fixes row-hash and fingerprint at once. NaN / +-Infinity are rendered as strings (the numeric cast would otherwise hard-fail on Redshift). ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-column UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Session-level UTC pin for cross-engine timestamp determinism.** The Databricks target renders timestamps via `DATE_FORMAT`, which depends on `spark.sql.session.timeZone`, so `pin_utc_session` pins the session to UTC for the recon. It is gated on the source dialect (`redshift`) — a row-hash correctness concern shared by the plain compare and fingerprint paths, not gated on `reconcile_optimizer` — and the original value is restored once the recon completes, so a shared/interactive cluster sees no lasting change. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. - **Typed NULLs in the persisted `fingerprint_metrics` struct.** Optional fields (`verdict`, `target_row_count`, `row_count_source`, `fetch_path`) render as `cast(NULL as string|bigint)` rather than a bare `NULL`. A bare `NULL` makes Spark infer `NullType`, which the vectorized Parquet reader cannot read back and which breaks schema equality against the typed `recon_metrics` table; `IS NULL` dashboard semantics are preserved. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 375 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no explicit session-timezone function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`), the implicit `DATE_FORMAT` dependence being pinned once by `pin_utc_session` (Redshift-scoped, restored after the recon); the fallback path is pinned; the fingerprint serialization suites pin the exact rendered SQL on each dialect for the three pre-existing fixes (including the `DOUBLE` -> `DECIMAL(38,10)` normalization on source and target). - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
ameersalman-db
added a commit
that referenced
this pull request
Jul 19, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon, exposed as the user-facing flag `reconcile_optimizer`. When `reconcile_optimizer=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `reconcile_optimizer`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds the legacy spellings (`fingerprint_precheck`, `redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new `reconcile_optimizer` flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Three correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. All corrupt the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. All fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. - **Both blocks missing DOUBLE handler.** DOUBLE had no override on either dialect, so both fell through to `TRIM(CAST(col AS string))`. Redshift renders `double precision` at full 17-digit precision (`0.28999999999999998`) while Spark emits the shortest round-trip (`0.29`), so every double-bearing row false-mismatched on a Redshift -> Databricks reconcile. In the fingerprint path this trips the systemic-mismatch guard, so the pre-check defers on any table with a DOUBLE column (a `transformations` override can't rescue it — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to `COALESCE(CAST(CAST(col AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_')`, the same normalization the Teradata recon fixture applied by hand; as a dialect default it fixes row-hash and fingerprint at once. NaN / +-Infinity are rendered as strings (the numeric cast would otherwise hard-fail on Redshift). ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-column UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Session-level UTC pin for cross-engine timestamp determinism.** The Databricks target renders timestamps via `DATE_FORMAT`, which depends on `spark.sql.session.timeZone`, so `pin_utc_session` pins the session to UTC for the recon. It is gated on the source dialect (`redshift`) — a row-hash correctness concern shared by the plain compare and fingerprint paths, not gated on `reconcile_optimizer` — and the original value is restored once the recon completes, so a shared/interactive cluster sees no lasting change. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. - **Typed NULLs in the persisted `fingerprint_metrics` struct.** Optional fields (`verdict`, `target_row_count`, `row_count_source`, `fetch_path`) render as `cast(NULL as string|bigint)` rather than a bare `NULL`. A bare `NULL` makes Spark infer `NullType`, which the vectorized Parquet reader cannot read back and which breaks schema equality against the typed `recon_metrics` table; `IS NULL` dashboard semantics are preserved. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 375 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no explicit session-timezone function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`), the implicit `DATE_FORMAT` dependence being pinned once by `pin_utc_session` (Redshift-scoped, restored after the recon); the fallback path is pinned; the fingerprint serialization suites pin the exact rendered SQL on each dialect for the three pre-existing fixes (including the `DOUBLE` -> `DECIMAL(38,10)` normalization on source and target). - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
ameersalman-db
added a commit
that referenced
this pull request
Jul 19, 2026
## Changes ### What does this PR do? Adds an opt-in fingerprint pre-check to Recon, exposed as the user-facing flag `reconcile_optimizer`. When `reconcile_optimizer=True` and the source has a registered query builder, Recon runs a sketch-based detection pass (MD5-sub-bucketed aggregates over both sides) before the row-hash compare pipeline. - MATCH -> Recon short-circuits in seconds; no full table scan, no JOIN. - MISMATCH -> an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows and feeds them into the existing `compare.reconcile_data` flow. If the mismatch is systemic (>15% of sub-buckets), the precheck defers to the existing pipeline. - Ineligible -> falls through silently. The flag defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the dataprint sketch-based reconciliation library; this is the first dataprint-into-lakebridge integration. Redshift is the first dialect — adding Snowflake / Oracle / TSQL is one `FingerprintQueryBuilder` subclass plus one registry entry. ### Relevant implementation details - `trigger_recon_service._run_fingerprint_or_reconcile_data` is the single decision point. Static eligibility centralised in `classify_ineligibility`; the schema-dependent `unmapped_target_column_mapping` reason is raised by `align_columns` as a typed exception and routed through `FingerprintRunMetadata.ineligible(...)`. Every reason maps to an `IneligibilityReason` enum value and is recorded on `recon_metrics.fingerprint_metrics.ineligibility_reason`. - Source-side reads use upstream's `RemoteQueryReader` / `remote_query()` TVF unmodified; Stage-1 aggregation pushdown verified empirically on a 1 M-row Redshift fixture (DBR 17.3). - Per-column hash serialization is shared with the row-hash compare path: both the Redshift source SQL and the Databricks target SQL render each column through `DataType_transform_mapping` via `serialize_column_for_hash` (`reconcile/query_builder/expression_generator.py`), the same lookup the row-hash `_default_transformer` uses. The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 -> sub-bucket/bucket arithmetic is fingerprint-specific. - Stage-1 detection is parallelised across source / target via a 2-thread pool; failure semantics match the serial version. - Two new fields on `ReconcileConfig`: `reconcile_optimizer`, `fingerprint_row_count_override`. - Config version bumps 2 -> 3 with a `v2_migrate` that folds the legacy spellings (`fingerprint_precheck`, `redshift_fingerprint_precheck`, `use_fingerprint_precheck`) into the new `reconcile_optimizer` flag. Existing deployments upgrade automatically. ### Pre-existing fixes that ride along (upstream PR #2339) Three correctness bugs in the upstream Redshift connector MR (#2339) surfaced during the dataprint integration P0 / P1 runs against a real cluster. All corrupt the existing row-hash recon path on real customer schemas and are unrelated to dataprint, but they sat in the integration path so they are fixed inline. All fixes live in `reconcile/query_builder/expression_generator.py` and are pinned by regression tests. - **Databricks block missing TIMESTAMP / TIMESTAMPTZ handler.** Redshift's source-side transform emits `COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_')` (always 6 fractional digits), but the Databricks block had no override, so the target side fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))` — Spark emits a variable-length fractional component, omitted entirely for zero-microsecond timestamps. The byte-width drift made per-row SHA2 disagree for every TIMESTAMP / TIMESTAMPTZ row in any Redshift -> Databricks reconcile. Fix: add `COALESCE(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_')` so source and target are byte-identical. - **Redshift block missing BOOLEAN handler.** The Redshift block defined overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and had no dialect-level `default`. BOOLEAN columns fell through to the universal default `TRIM(COALESCE(col, '_null_recon_'))`, which Redshift rejects during output schema resolution with `function pg_catalog.btrim(boolean) does not exist`. Any customer schema containing a single BOOLEAN column crashes row-hash recon end-to-end. Fix: explicit `COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_')` so the rendered string matches Spark's `cast(boolean AS string)` byte-for-byte. - **Both blocks missing DOUBLE handler.** DOUBLE had no override on either dialect, so both fell through to `TRIM(CAST(col AS string))`. Redshift renders `double precision` at full 17-digit precision (`0.28999999999999998`) while Spark emits the shortest round-trip (`0.29`), so every double-bearing row false-mismatched on a Redshift -> Databricks reconcile. In the fingerprint path this trips the systemic-mismatch guard, so the pre-check defers on any table with a DOUBLE column (a `transformations` override can't rescue it — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to `COALESCE(CAST(CAST(col AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_')`, the same normalization the Teradata recon fixture applied by hand; as a dialect default it fixes row-hash and fingerprint at once. NaN / +-Infinity are rendered as strings (the numeric cast would otherwise hard-fail on Redshift). ### Code-review hardening - **Serialization consolidated onto the shared transform map.** Earlier revisions hand-wrote a per-column serializer on each of the three paths (Redshift source SQL, Spark Stage-1 `Column`, Spark Stage-2 SQL), kept byte-aligned by tests — including a per-column UTC pin (`TO_CHAR(_ AT TIME ZONE 'UTC', _)` / `TO_UTC_TIMESTAMP(_, CURRENT_TIMEZONE())`) and a `CAST(_ AS VARCHAR(65535))` to dodge Redshift's 256-byte default. Both are removed: routing through `DataType_transform_mapping` makes the fingerprint serialization identical to the row-hash path by construction, and the default `TRIM(col)` does not truncate, making the width cast unnecessary. - **Session-level UTC pin for cross-engine timestamp determinism.** The Databricks target renders timestamps via `DATE_FORMAT`, which depends on `spark.sql.session.timeZone`, so `pin_utc_session` pins the session to UTC for the recon. It is gated on the source dialect (`redshift`) — a row-hash correctness concern shared by the plain compare and fingerprint paths, not gated on `reconcile_optimizer` — and the original value is restored once the recon completes, so a shared/interactive cluster sees no lasting change. - **Hash-column ordering de-duplicated** into `HashQueryBuilder.ordered_hash_columns()`, reused by the fingerprint pre-check (the standalone `fingerprint_hash_columns` module was deleted). - **Null-safe column diff moved into the compare layer.** The per-column `<=>` recompute + per-row `mismatch_columns` annotation is now the shared `compare.annotate_mismatch_columns(...)` helper instead of being hand-rolled in `fingerprint/orchestrator.py`. - **Table-placeholder substitution moved behind the builder.** `HashQueryBuilder.substitute_table(...)` owns its `:tbl` placeholder and resolves every dialect-rendered form (`:tbl` on Spark, `%(tbl)s` on Postgres-family); the orchestrator no longer hard-codes placeholder syntax. - **Stage-2 build failures fall through to the full pipeline** in `trigger_recon_service.py` instead of marking the table failed. Every other non-MATCH branch already does this; metadata records `fallback_to_full_pipeline=True` for observability. - **Typed NULLs in the persisted `fingerprint_metrics` struct.** Optional fields (`verdict`, `target_row_count`, `row_count_source`, `fetch_path`) render as `cast(NULL as string|bigint)` rather than a bare `NULL`. A bare `NULL` makes Spark infer `NullType`, which the vectorized Parquet reader cannot read back and which breaks schema equality against the typed `recon_metrics` table; `IS NULL` dashboard semantics are preserved. Smaller cleanups: dropped unused `ColumnAlignment.exclude_columns`; reverted a no-op reorder in `connectors/source_adapter.py`; replaced a flaky wall-clock assertion in `test_fetch_parallel.py` with a deterministic distinct-thread-id assertion; pinned the exact rendered SQL on each dialect in `test_expression_generator.py` and added a regression test for the Redshift `BOOLEAN` handler. ### Caveats - DBR 17.3+ required for source-side reads via `remote_query()` (inherited from upstream's `RemoteQueryReader` adoption). - MISMATCH-state cost at 1 M scale currently exceeds row-hash-only mode by 16-94 s because Stage-2 still feeds the existing JOIN. MATCH is the headline win (38.7% on 1 M rows); billion-row scale is the production motivation. Stage-1 hash persistence as Stage-2 input is filed as a follow-up. - Pre-existing `success_count` formula in `verify_successful_reconciliation` (upstream PR #2259, commit `e56c79c3d`) is mathematically wrong; sits next to fingerprint code in `trigger_recon_service.py`. Not fixed here to keep scope contained; filed separately. ### Tests - All unit tests on the touched surface pass: 1551 / 1557 (the 6 `test_cli_analyze.py` failures are pre-existing on main and unrelated). `tests/unit/reconcile/` runs 375 tests in ~1 s. - Parity tests assert the fingerprint source/target serializers are byte-identical to the shared row-hash transform map, and that the target timestamp serializer renders no explicit session-timezone function (`TO_UTC_TIMESTAMP` / `CURRENT_TIMEZONE`), the implicit `DATE_FORMAT` dependence being pinned once by `pin_utc_session` (Redshift-scoped, restored after the recon); the fallback path is pinned; the fingerprint serialization suites pin the exact rendered SQL on each dialect for the three pre-existing fixes (including the `DOUBLE` -> `DECIMAL(38,10)` normalization on source and target). - Correctness validated end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39/40 cells PASS, 1 scenario shows a known fingerprint-solver fallback edge with verdict agreement on both sides — only the cap-bounded `mismatch` count differs (fingerprint reports the true 10000, normal reports the cap-50 sample). - Linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green. - Integration coverage to follow alongside the recon e2e cluster fixture (#2453).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add Redshift connector to Recon