Add fingerprint pre-check to Recon (Redshift)

[ameersalman-db]

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 → short-circuits in seconds; no full scan, no JOIN.
• MISMATCH → an algebraic solver returns the differing row hashes; a surgical Stage-2 fetch pulls just those rows into the existing compare.reconcile_data flow. If the mismatch is systemic (>15% of sub-buckets), it defers to the row-hash pipeline.
• Ineligible → falls through to the row-hash pipeline silently.

Defaults to False; existing behaviour is unchanged. The algorithm is byte-identical to the internal dataprint library; this is the first dataprint-into-lakebridge integration. Redshift is the first source dialect — adding Snowflake / Oracle / TSQL is one FingerprintQueryBuilder subclass plus a registry entry, with no orchestrator changes.

Implementation details

• Single decision point: trigger_recon_service._run_fingerprint_or_reconcile_data. Static eligibility (flag off, unsupported dialect, non-data report type, no join columns, filters / transforms / thresholds configured) is centralised in classify_ineligibility (fingerprint/orchestrator.py) and runs before any source compute. One reason — unmapped_target_column_mapping — needs the target schema, so it's detected inside the precheck via the typed UnmappedTargetColumnMappingError from align_columns and routed through FingerprintRunMetadata.ineligible(...). Every reason is an IneligibilityReason enum value recorded on recon_metrics.fingerprint_metrics.ineligibility_reason, so adoption queries need no log-grepping.
• No new connector code: source reads use upstream's RemoteQueryReader / remote_query() TVF unmodified. Stage-1 aggregation pushdown was verified on a 1 M-row Redshift fixture (DBR 17.3): direct JDBC (4.28 s median) and remote_query() (5.44 s median) both push the GROUP BY to Redshift with identical aggregated counts.
• Shared serialization, not re-implemented: both Redshift-source and Databricks-target SQL render each column through the existing DataType_transform_mapping via serialize_column_for_hash — the same lookup the row-hash QueryBuilder._default_transformer uses (refactored to the shared get_transform_for_type helper). The fingerprint byte stream is identical to the row-hash pipeline by construction; only the MD5 → sub-bucket/bucket arithmetic is fingerprint-specific.
• Parallelised Stage-1 across source / target via ThreadPoolExecutor(max_workers=2); failure semantics match the serial version.
• New ReconcileConfig fields: reconcile_optimizer: bool = False; fingerprint_row_count_override: int | None = None (pins the adaptive tier when DESCRIBE DETAIL can't read numRecords).
• Config __version__ 2 → 3: new v2_migrate adds reconcile_optimizer and folds the legacy spellings (fingerprint_precheck and older names) into it; the v1 → v2 → v3 chain runs automatically via Installation.load. Zero user action.

Pre-existing fixes that ride along (upstream PR #2339)

Integrating against real Redshift surfaced three pre-existing correctness bugs in the Redshift connector MR (#2339). None is dataprint-specific — all corrupt the existing row-hash path on real schemas — but they sit in the integration path, so they're fixed inline in expression_generator.py and pinned by regression tests.

#	Fix	What broke before
1	Add TIMESTAMP / TIMESTAMPTZ handlers to the Databricks dialect (target)	The Redshift source emits fixed-width COALESCE(TO_CHAR(ts, 'YYYY-MM-DD HH24:MI:SS.US'), '_null_recon_') (always 6 fractional digits). The Databricks block had no TIMESTAMP / TIMESTAMPTZ override, so the target fell through to TRIM(COALESCE(col, '_null_recon_')), where Spark's cast(timestamp AS string) emits a variable-length fraction (dropped entirely at zero microseconds: '2023-10-02 18:08:43' vs source …43.000000'). The byte-width drift broke the per-row SHA2 for every identical timestamp row in any Redshift → Databricks reconcile. Fix: COALESCE(DATE_FORMAT(_, 'yyyy-MM-dd HH:mm:ss.SSSSSS'), '_null_recon_') for both types.
2	Add BOOLEAN handler to the Redshift dialect (source)	The Redshift block had overrides only for SUPER / DATE / TIMESTAMP / TIMESTAMPTZ and no dialect default, so BOOLEAN fell through to TRIM(...), which Redshift rejects at output-schema resolution (before reading rows) with function pg_catalog.btrim(boolean) does not exist — any schema with one BOOLEAN column crashes row-hash recon on #2339. Fix: COALESCE(CASE WHEN col THEN 'true' WHEN NOT col THEN 'false' ELSE NULL END, '_null_recon_'), matching Spark's cast(boolean AS string). The unit suite missed it because fixtures hand-build query strings and skip the transform-mapping path.
3	Add DOUBLE handler to the Redshift and Databricks dialects (both sides)	DOUBLE had no override on either dialect, so both fell through to TRIM(CAST(col AS string)) — but 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 and the pre-check defers on any table with a DOUBLE column (it can't be rescued with a transformations override — a configured transform makes the pre-check ineligible by design). Fix: pin both sides to a fixed-scale COALESCE(CAST(CAST(_ AS DECIMAL(38,10)) AS STRING/VARCHAR), '_null_recon_') so the byte streams are identical, with NaN / ±Infinity rendered as strings (the numeric cast would otherwise hard-fail on Redshift). This is the same normalization the Teradata recon fixture applied by hand; making it the dialect default fixes row-hash and fingerprint at once.

Code-review hardening

• Serialization consolidated onto the shared transform map. Earlier revisions had a hand-written per-column serializer on each of three paths, including a per-column UTC pin and a CAST(_ AS VARCHAR(65535)). Both removed: routing through DataType_transform_mapping makes fingerprint serialization identical to the row-hash path by construction (the default TRIM(col) doesn't truncate, so no width cast is needed).
• Session-level UTC pin for cross-engine timestamp determinism. The Databricks target renders timestamps via DATE_FORMAT, which is spark.sql.session.timeZone-dependent, so pin_utc_session sets the session timezone to UTC for the recon. It's 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 when the recon completes, so a shared/interactive cluster sees no lasting change.
• Hash-column ordering de-duplicated into HashQueryBuilder.ordered_hash_columns() (standalone fingerprint_hash_columns module deleted).
• Null-safe column diff moved to compare.annotate_mismatch_columns(...) rather than hand-rolled in the orchestrator. (Routing through Reconciliation._get_sample_data would re-issue sampling queries and defeat the surgical fetch, so it lives in compare.py.)
• Table-placeholder substitution moved behind HashQueryBuilder.substitute_table(...), which resolves every dialect form (:tbl, %(tbl)s); the orchestrator no longer hard-codes placeholder syntax.
• Fail-open on Stage-2 build failure: a build_mismatch_output exception falls through to the full pipeline (fallback_to_full_pipeline=True) instead of failing the table.
• Typed NULLs in the persisted fingerprint_metrics struct. Optional fields (verdict, target_row_count, row_count_source, fetch_path) now render as cast(NULL as string|bigint) rather than a bare NULL. A bare NULL makes Spark infer NullType for the struct field, which the vectorized Parquet reader cannot read back (VectorizedParquetRecordReader error) and which breaks schema equality against the typed recon_metrics table. IS NULL semantics for dashboards are preserved.
• Cleanups: unused ColumnAlignment.exclude_columns removed, no-op reorder in source_adapter.py reverted, flaky wall-clock assertion in test_fetch_parallel.py replaced with a distinct-thread-id check.

Caveats for reviewers

• DBR 17.3+: source reads via remote_query() need classic clusters on DBR 17.3+ or SQL warehouses (Pro / Serverless) 2025.35+ — inherited from upstream's RemoteQueryReader, documented in docs/lakebridge/docs/reconcile/index.mdx.
• MISMATCH cost: at 1 M, fingerprint pays ~20 s for Stage-1 unconditionally and Stage-2 still feeds the same JOIN, so MISMATCH costs an extra 16–94 s vs row-hash-only. The MATCH path is the win (38.7%: 55.5 s vs 90.6 s at 1 M); the motivation is billion-row scale, where the full-table JOIN is intractable. Stage-1 hash reuse as Stage-2 input is a filed follow-up.
• Test fixture: the only non-fingerprint / non-expression_generator test change is test_install.py (six version 2 → 3 bumps from the migration we own). Not scope creep.
• success_count bug in verify_successful_reconciliation (yields counts > total) is pre-existing (Improve reconciliation result handling and logging #2259, e56c79c3d), sits next to fingerprint code, and is not fixed here to keep scope contained — filed separately.

Linked issues

Resolves #..

Functionality

• user docs (configuration.mdx Fingerprint Pre-check (Experimental) + index.mdx runtime requirements)
• added new CLI command
• modified existing command
• new opt-in feature behind ReconcileConfig.reconcile_optimizer (default False)

Tests

• unit tests across the fingerprint modules (query builders, solver, eligibility, fetch parallelisation, v1 → v2 → v3 migration, recon-capture schema, dispatch, Stage-1 / Stage-2 symmetry) + parity tests asserting the fingerprint serializers are byte-identical to the shared transform map.
• All 1551 unit tests pass (2 skipped, 3 xfailed); tests/unit/reconcile/ runs 373 tests in ~1 s. The 6 failing test_cli_analyze.py cases (Informatica binary) are pre-existing on main and unrelated.
• regression tests for the three Add Redshift connector to Recon #2339 fixes: handler presence, format string, and source / target byte alignment for TIMESTAMP / TIMESTAMPTZ, the BOOLEAN rendering on Redshift, and the DOUBLE → DECIMAL(38,10) normalization on both dialects.
• timezone determinism: the target timestamp serializer asserts no explicit session-timezone function (TO_UTC_TIMESTAMP / CURRENT_TIMEZONE) is rendered; the implicit DATE_FORMAT dependence is pinned once at the session level by pin_utc_session (Redshift-scoped, restored after the recon).
• end-to-end on a 1 M-row Redshift / Delta fixture across the 20-scenario dual-mode parity matrix: 39 / 40 PASS; the 1 outlier is a known solver-fallback edge with verdict + missing-rows agreement on both sides — only the cap-bounded mismatch count differs (fingerprint reports the true 10000, normal the cap-50 sample).
• solver-sufficiency guards (test_solver_sufficiency.py): a randomized no-false-solve property, clean-≤2 coverage, and a per-tier ≥98.8% solve-rate floor against the real guard constants and tier table. Backed by the three-method proof under experiments/redshift_fingerprint_validation/proof/solver_sufficiency/ (see below).
• linter clean: pylint 10.00/10 on touched src; ruff, black, mypy green across reconcile/ + config.py.
• integration tests (the recon e2e cluster gate landed in create dedicated test cluster for recon e2e tests #2453; coverage follows in a later MR with the cluster fixture)

End-to-end validation (real Redshift → Delta, 1 M rows)

One self-contained demo notebook generates identical deterministic data (MD5(order_id ‖ salt)) in both engines, reconciles the same tables twice (reconcile_optimizer OFF = built-in row-hash, ON = fingerprint), and diffs the audit tables. Source: Redshift perf_test.orders_demo via remote_query(); target: Delta orders_demo.

Dual-mode parity — identical counts (audit metrics):

scenario	flag	mismatch	missing_in_target	missing_in_source	fp verdict	fallback
clean match	OFF	0	0	0	—	—
clean match	ON	0	0	0	MATCH	false
known mutations	OFF	5	3	2	—	—
known mutations	ON	5	3	2	MISMATCH	false

Record-level parity — exact keys, not just counts (audit details). Ground truth: order_id 1–5 value-mutated, 6–8 deleted from target, 9–10 deleted from source. Both modes captured exactly:

recon bucket	built-in (OFF)	fingerprint (ON)
mismatch	1, 2, 3, 4, 5	1, 2, 3, 4, 5
missing_in_target	6, 7, 8	6, 7, 8
missing_in_source	9, 10	9, 10

Live post-mutation state: Redshift = 999,998 rows (9,10 absent → 2 source deletes); Delta = 999,997 (6,7,8 absent → 3 target deletes).

On MISMATCH the solver resolved all affected sub-buckets (15 solved, no fallback) and reached the identical answer as the built-in path — same counts and same exact records, both scenarios, with the pre-check engaged (eligible=true, fallback_to_full_pipeline=false). Re-derivable from the main / metrics / details tables.

Solver sufficiency — why d≤2 is enough

The reasonable reviewer question: the solver only resolves a sub-bucket holding ≤2 differing rows (d=1, d=2-swap, d=2-extras) — how do we know mismatches don't pile 3+ into one bucket? It's exact, not a guess: rows map to sub-buckets uniformly by content (sub_bucket_id = ABS(MOD(rh1, N))), so it's the classic occupancy problem. Proven three ways that agree: closed-form occupancy, Monte-Carlo, and the real engine.py solver on generated deltas.

Two shipped mechanisms bound λ = D / N (differing rows per sub-bucket): adaptive tiering scales N with table size, and the systemic guard bails once > min(50,000, 15%·N) sub-buckets mismatch. So the solver only runs at low λ, which floors the solve rate per tier:

tier N (sub-buckets)	binding guard	worst-case λ	min records solved by d≤2
16,384	15% ratio	0.163	98.81%
262,144	15% ratio	0.163	98.81%
1,048,576	50k count	0.049	99.88%
2,097,152	50k count	0.024	99.97%
4,194,304	50k count	0.012	99.99%
8,388,608	50k count	0.006	99.998%
16,777,216	50k count	0.003	99.9996%

Across the entire operational regime, d≤2 surgically resolves 98.8%–99.999% of mismatch records — the "95%" intuition is a conservative floor. The full 7-tier × 4-density × 3-type (missing / value / mixed) sweep shows ≥99.98% whenever the solver runs; closed-form and Monte-Carlo agree to the fourth decimal.

Correctness does not depend on this coverage. The <2% not solved algebraically (≥3-ball buckets, plus rare 32-bit rh1 collisions) are brute-force fetched, never dropped — d<3 is an efficiency threshold, not a correctness one. Over 12,000,000+ simulated differing rows: 0 false solves, 0 clean-config failures, 0 records lost. All 271 declined ≤2 buckets were rh1 collisions the second (rh2) channel correctly flagged and routed to brute-force — exactly where a single-channel sketch would silently declare a MATCH.

Reproducible in seconds with no workspace or Spark (pure Python + NumPy, imports the real solver and tier table): experiments/redshift_fingerprint_validation/proof/solver_sufficiency/ (solver_sufficiency.py, RESULTS.txt, report.html). The ≥98.8% floor and no-false-solve property are CI-guarded in tests/unit/reconcile/fingerprint/test_solver_sufficiency.py.

[CLAassistant]

All committers have signed the CLA.

[codecov]

Codecov Report

❌ Patch coverage is 82.08431% with 153 lines in your changes missing coverage. Please review.
✅ Project coverage is 71.68%. Comparing base (c1fdb1f) to head (e33c710).
⚠️ Report is 15 commits behind head on main.

Files with missing lines	Patch %	Lines
...ks/labs/lakebridge/reconcile/fingerprint/engine.py	71.83%	29 Missing and 20 partials ⚠️
...s/lakebridge/reconcile/fingerprint/orchestrator.py	79.31%	31 Missing and 11 partials ⚠️
...s/lakebridge/reconcile/fingerprint/spark_target.py	63.15%	28 Missing ⚠️
...labs/lakebridge/reconcile/trigger_recon_service.py	82.55%	13 Missing and 2 partials ⚠️
...abricks/labs/lakebridge/reconcile/recon_capture.py	80.00%	7 Missing ⚠️
...rc/databricks/labs/lakebridge/reconcile/compare.py	66.66%	2 Missing and 3 partials ⚠️
...labs/lakebridge/reconcile/fingerprint/constants.py	93.33%	1 Missing and 1 partial ⚠️
...s/lakebridge/reconcile/query_builder/hash_query.py	86.66%	1 Missing and 1 partial ⚠️
...bricks/labs/lakebridge/reconcile/reconciliation.py	66.66%	2 Missing ⚠️
...labs/lakebridge/reconcile/fingerprint/row_count.py	98.21%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2570      +/-   ##
==========================================
+ Coverage   70.35%   71.68%   +1.32%     
==========================================
  Files         107      116       +9     
  Lines        9658    10474     +816     
  Branches     1069     1187     +118     
==========================================
+ Hits         6795     7508     +713     
- Misses       2655     2719      +64     
- Partials      208      247      +39     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

✅ 176/176 passed, 8 flaky, 2 skipped, 2h22m29s total

Flaky tests:

• 🤪 test_installs_and_runs_local_bladebridge (13.14s)
• 🤪 test_installs_and_runs_pypi_bladebridge (22.219s)
• 🤪 test_recon_for_report_type_is_data (45.958s)
• 🤪 test_transpiles_informatica_to_sparksql_non_interactive[False] (24.042s)
• 🤪 test_transpiles_informatica_to_sparksql (25.726s)
• 🤪 test_transpile_teradata_sql_non_interactive[False] (6.174s)
• 🤪 test_transpile_teradata_sql (6.9s)
• 🤪 test_transpile_teradata_sql_non_interactive[True] (9.463s)

_{Running from acceptance #5103}

[m-abulazm]

@ameersalman-db some initial thoughts given the PR description

Config version 2 → 3: new v2_migrate adds reconcile_optimizer and folds the legacy spellings (fingerprint_precheck and older names) into it; the v1 → v2 → v3 chain runs automatically via Installation.load. Zero user action.

This is hallucinated by the AI. there was never a legacy spelling except for local development which affects no users

New ReconcileConfig fields: reconcile_optimizer: bool = False; fingerprint_row_count_override: int | None = None (pins the adaptive tier when DESCRIBE DETAIL can't read numRecords).

The config changes should be reconsidered

Pre-existing fixes that ride along (upstream PR #2339)
Integrating against real Redshift surfaced three pre-existing correctness bugs in the Redshift connector MR (#2339). None is dataprint-specific — all corrupt the existing row-hash path on real schemas — but they sit in the integration path, so they're fixed inline in expression_generator.py and pinned by regression tests.

This should go on its own PR

Session-level UTC pin for cross-engine timestamp determinism. The Databricks target renders timestamps via DATE_FORMAT, which is spark.sql.session.timeZone-dependent, so pin_utc_session sets the session timezone to UTC for the recon. It's 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 when the recon completes, so a shared/interactive cluster sees no lasting change.

revert please. leave this up to the user

Serialization consolidated onto the shared transform map. Earlier revisions had a hand-written per-column serializer on each of three paths, including a per-column UTC pin and a CAST(_ AS VARCHAR(65535)). Both removed: routing through DataType_transform_mapping makes fingerprint serialization identical to the row-hash path by construction (the default TRIM(col) doesn't truncate, so no width cast is needed).

I dont get this, but should be a new PR

Table-placeholder substitution moved behind HashQueryBuilder.substitute_table(...), which resolves every dialect form (:tbl, %(tbl)s); the orchestrator no longer hard-codes placeholder syntax.

this needs to be reconsidered. we can just build the query completely with table name in the first place

Null-safe column diff moved to compare.annotate_mismatch_columns(...) rather than hand-rolled in the orchestrator. (Routing through Reconciliation._get_sample_data would re-issue sampling queries and defeat the surgical fetch, so it lives in compare.py.)

introducing a third diff is no-go. we should fix the existing diff

success_count bug in verify_successful_reconciliation (yields counts > total) is pre-existing (#2259, e56c79c), sits next to fingerprint code, and is not fixed here to keep scope contained — filed separately.

new PR. we can group some of the fixes together on one PR