Skip to content

Commit 282740a

Browse files
Add fingerprint pre-check to Recon (Redshift)
## 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).
1 parent 1e02597 commit 282740a

43 files changed

Lines changed: 6941 additions & 35 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

docs/lakebridge/docs/reconcile/configuration.mdx

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,83 @@ Place the file in `.lakebridge/` in your Databricks workspace home folder.
9191
9292
---
9393
94+
## Fingerprint Pre-check (Experimental)
95+
96+
The fingerprint pre-check is an opt-in optimisation that, when source and target are
97+
already in sync, replaces the per-row hash-and-join pipeline with a sub-bucket-level
98+
aggregate comparison. On a 1 M-row Redshift fixture in MATCH state, the precheck
99+
short-circuits the recon at roughly **30–45 % of the v3 baseline wall-clock**; on
100+
MISMATCH it surgically fetches only the differing rows instead of streaming the full
101+
column set across JDBC. When neither the MATCH nor the surgical-fetch path applies the
102+
recon falls open to the full pipeline so correctness is never compromised.
103+
104+
### When to enable
105+
106+
- The source / target tables are expected to be **mostly identical** (post-migration,
107+
ongoing CDC) — the precheck pays off most when MATCH is the common outcome.
108+
- The runtime targets **DBR 17.3 or later**. The Stage-2 source-side fetch uses
109+
Databricks' `remote_query()` table-valued function which requires DBR 17.3+. On
110+
earlier runtimes, leave the flag off — the eligibility gate doesn't yet check DBR
111+
version, so the JDBC call would fail mid-fetch and trigger the fail-open path.
112+
- The source dialect has a registered fingerprint query builder. Today only
113+
`redshift` is wired. Other dialects fall through to the full pipeline silently.
114+
115+
### Configuration
116+
117+
Add to `recon_config_*.json` at the top level:
118+
119+
```yaml
120+
fingerprint_precheck: true
121+
# Optional. False (default) keeps '' distinct from NULL in fingerprint hashing,
122+
# matching the row-hash compare path in expression_generator. Flip to True only if
123+
# your data treats '' and NULL as the same value AND you have audited the impact;
124+
# the flag is wired symmetrically across both source-side Redshift SQL and the
125+
# target-side Spark Stage-1 / Stage-2 serialisers so the two cannot drift.
126+
fingerprint_treat_empty_as_null: false
127+
# Optional. When set, overrides the target Delta DESCRIBE DETAIL numRecords
128+
# lookup used to pick the sub-bucket tier. Use this when the target is non-Delta
129+
# (DESCRIBE DETAIL returns no numRecords) or when Delta stats are stale and the
130+
# heuristic lands on a tier that is too coarse / too fine for your workload.
131+
# Values <= 0 are treated as unset.
132+
fingerprint_row_count_override: 250000000
133+
```
134+
135+
### Eligibility rules
136+
137+
The pre-check declines (and the recon proceeds via the full pipeline) when **any** of
138+
the conditions below hold. These reasons are recorded in
139+
`recon_metrics.fingerprint_metrics.ineligibility_reason` so adoption queries can
140+
distinguish "feature off" from "feature on but table ineligible".
141+
142+
| Reason value | Meaning |
143+
| --- | --- |
144+
| `flag_disabled` | `fingerprint_precheck` is false (default). |
145+
| `unsupported_dialect` | The source dialect has no registered fingerprint query builder (today: anything other than `redshift`). |
146+
| `report_type_not_data` | `report_type` is `schema` (precheck operates on data-level reconciles). |
147+
| `no_join_columns` | `join_columns` is empty. The precheck needs primary-key columns to disambiguate culprit rows during Stage-2. |
148+
| `filters_configured` | `filters.source` or `filters.target` is set. The precheck does not project filter predicates into Stage-1 aggregates yet. |
149+
| `transforms_configured` | `transformations` is set. Custom transformations are not supported on the fingerprint hash path. |
150+
| `column_thresholds_configured` | `column_thresholds` is set. Threshold semantics conflict with exact-hash comparison. |
151+
| `table_thresholds_configured` | `table_thresholds` is set. Same rationale as column thresholds. |
152+
153+
A separate runtime gate validates that every `column_mapping.target_name` resolves to
154+
a real target column before issuing the source-side scan — a typo in the mapping is
155+
caught at eligibility time, not after Stage-1 has already pulled across JDBC.
156+
157+
### Tuning and observability
158+
159+
- The pre-check selects a sub-bucket / bucket count adaptively from the target's
160+
Delta `numRecords` (DESCRIBE DETAIL). On a non-Delta target, or when the metric
161+
is missing, it falls back to a static `(1 048 576, 32 768)` pair. Override
162+
explicitly via `ReconcileConfig.fingerprint_row_count_override` (an
163+
approximate target row count) when the heuristic picks a tier that you can
164+
show is wrong for your workload.
165+
- Every recon writes a `fingerprint_metrics` named-struct to
166+
`recon_metrics.fingerprint_metrics` regardless of eligibility, so adoption,
167+
fall-open rate, and verdict distribution can be tracked from one query.
168+
169+
---
170+
94171
## TABLE Config Schema
95172

96173
<Tabs>

docs/lakebridge/docs/reconcile/index.mdx

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,18 @@ The User configuring reconcile must have permission to:
8282
- `USE CATALOG` and `CREATE SCHEMA` on the target catalog
8383
- `CREATE VOLUME` if using a pre-existing schema on a serverless cluster
8484

85+
### Runtime requirements
86+
87+
- The job / interactive cluster running reconcile must be on **DBR 15.4 LTS or later**
88+
for the standard data-comparison path.
89+
- If `fingerprint_precheck` is enabled (see
90+
[Configuration Reference → Fingerprint Pre-check](/docs/reconcile/configuration#fingerprint-pre-check-experimental)),
91+
the cluster must be on **DBR 17.3 or later**. The Stage-2 source-side fetch uses
92+
Databricks' `remote_query()` table-valued function, which became available on DBR
93+
17.3. On earlier runtimes, leave the flag off; otherwise the JDBC call fails
94+
mid-fetch and the recon falls open to the full pipeline (correct, but the precheck
95+
buys you nothing while paying for itself in cluster time).
96+
8597
### Serverless cluster support
8698

8799
Reconcile automatically detects the cluster type and optimizes intermediate data persistence accordingly:

src/databricks/labs/lakebridge/config.py

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -280,13 +280,27 @@ class ReconcileJobConfig:
280280
@dataclass
281281
class ReconcileConfig:
282282
__file__ = "reconcile.yml"
283-
__version__ = 2
283+
__version__ = 3
284284

285285
report_type: str
286286
source: SourceConnectionConfig
287287
target: TargetConnectionConfig
288288
metadata_config: ReconcileMetadataConfig
289289
job_overrides: ReconcileJobConfig | None = None
290+
fingerprint_precheck: bool = False
291+
# When True, fingerprint hashing collapses '' to NULL on BOTH source and target
292+
# sides — flipped here so the Stage-1 / Stage-2 serialisers cannot drift apart
293+
# (without this knob, target call sites silently kept the function default of
294+
# False while source picked up a constant override). Staying False matches the
295+
# row-hash compare path in ``expression_generator``.
296+
fingerprint_treat_empty_as_null: bool = False
297+
# Optional explicit row count for fingerprint tier selection. When set,
298+
# overrides Delta ``DESCRIBE DETAIL`` numRecords lookup so customers whose
299+
# target is non-Delta (or whose Delta stats are stale) can pick the right
300+
# sub-bucket tier without waiting for a full COUNT(*). ``None`` keeps the
301+
# default heuristic. Values ``<= 0`` are treated as "unset" by
302+
# ``fetch_target_row_count``.
303+
fingerprint_row_count_override: int | None = None
290304

291305
@classmethod
292306
def v1_migrate(cls, raw: dict[str, Any]) -> dict[str, Any]:
@@ -314,6 +328,25 @@ def v1_migrate(cls, raw: dict[str, Any]) -> dict[str, Any]:
314328
raw["version"] = 2
315329
return raw
316330

331+
@classmethod
332+
def v2_migrate(cls, raw: dict[str, Any]) -> dict[str, Any]:
333+
"""v2 → v3: introduce the source-agnostic ``fingerprint_precheck`` flag.
334+
335+
Older field names (``redshift_fingerprint_precheck``, ``use_fingerprint_precheck``)
336+
from internal pre-v2 deployments are folded into the new flag if present;
337+
otherwise the field defaults to ``False`` so existing v2 configs keep their
338+
current behaviour.
339+
"""
340+
if "fingerprint_precheck" not in raw:
341+
for legacy in ("redshift_fingerprint_precheck", "use_fingerprint_precheck"):
342+
if legacy in raw:
343+
raw["fingerprint_precheck"] = raw.pop(legacy)
344+
break
345+
for legacy in ("redshift_fingerprint_precheck", "use_fingerprint_precheck"):
346+
raw.pop(legacy, None)
347+
raw["version"] = 3
348+
return raw
349+
317350
@property
318351
def database_config(self) -> DatabaseConfig:
319352
"""TODO remove. this was kept for backwards compatibility while migrating to ReconcileConfig v2"""
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
"""Fingerprint-accelerated reconciliation."""
Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
"""Shared constants and SQL helpers for fingerprint detection and row filtering."""
2+
3+
from __future__ import annotations
4+
5+
# Must match the row-hash path's NULL stand-in (``'_null_recon_'`` literal in
6+
# ``reconcile/query_builder/expression_generator.py``). Fingerprint and row-hash
7+
# both encode NULLs into the per-column hash payload before MD5/SHA — picking a
8+
# different stand-in here would alias real data ``'_null_recon_'`` with NULL on
9+
# only one side and produce the inverse alias on the other, so any row that
10+
# happens to carry either literal would be silently misclassified by Stage-1.
11+
# A unit test pins this to the row-hash literal so a future drift fails CI
12+
# rather than the reconcile.
13+
NULL_SENTINEL = "_null_recon_"
14+
15+
# chr(1) — column separator inside the MD5 concat. Rendered three ways:
16+
# Redshift SQL: CHR(1)
17+
# Spark SQL: CHAR(1)
18+
# Python / Spark DataFrame: "\x01"
19+
SEPARATOR_PYTHON = "\x01"
20+
SEPARATOR_REDSHIFT_SQL = "CHR(1)"
21+
SEPARATOR_SPARK_SQL = "CHAR(1)"
22+
23+
# Static defaults retained for backwards compatibility and as the fallback when the
24+
# adaptive selector has no row count to work with.
25+
SUB_BUCKET_COUNT = 1_048_576 # 1M sub-buckets
26+
BUCKET_COUNT = 32_768
27+
28+
# Adaptive tier table. Each entry: (max_row_count_inclusive, sub_bucket_count, bucket_count).
29+
# Last entry's max_row_count is None and clamps everything larger. Sub-bucket counts are
30+
# powers of 2 to keep MOD distribution clean; bucket count = sub_bucket_count / 1024.
31+
SUB_BUCKET_TIERS: tuple[tuple[int | None, int, int], ...] = (
32+
(50_000, 16_384, 128), # < 50K
33+
(500_000, 262_144, 512), # 50K – 500K
34+
(50_000_000, 1_048_576, 1_024), # 500K – 50M
35+
(500_000_000, 2_097_152, 2_048), # 50M – 500M
36+
(5_000_000_000, 4_194_304, 4_096), # 500M – 5B
37+
(50_000_000_000, 8_388_608, 8_192), # 5B – 50B
38+
(None, 16_777_216, 16_384), # 50B+
39+
)
40+
41+
42+
def pick_sub_bucket_count(row_count: int | None) -> tuple[int, int]:
43+
"""Select (sub_bucket_count, bucket_count) for ``row_count``.
44+
45+
Falls back to (SUB_BUCKET_COUNT, BUCKET_COUNT) when the count is unknown or
46+
non-positive, so callers can pass None safely.
47+
48+
>>> pick_sub_bucket_count(10_000)
49+
(16384, 128)
50+
>>> pick_sub_bucket_count(100_000_000)
51+
(2097152, 2048)
52+
>>> pick_sub_bucket_count(None)
53+
(1048576, 32768)
54+
"""
55+
if row_count is None or row_count <= 0:
56+
return SUB_BUCKET_COUNT, BUCKET_COUNT
57+
for max_row_count, sub_buckets, buckets in SUB_BUCKET_TIERS:
58+
if max_row_count is None or row_count <= max_row_count:
59+
return sub_buckets, buckets
60+
return SUB_BUCKET_COUNT, BUCKET_COUNT
61+
62+
63+
def build_fingerprint_where_clause(
64+
sb_expr: str,
65+
rh1_expr: str,
66+
solved_hashes: dict[int, list[int]],
67+
unsolved_sb_ids: list[int],
68+
) -> str:
69+
"""Build the WHERE body (no ``WHERE``, no trailing alias) for a filtered fetch.
70+
71+
Emits the union form ``(sb_expr IN (..) AND rh1_expr IN (..)) [OR sb_expr IN (..)]``.
72+
The form is mathematically equivalent to per-sub-bucket disjuncts because
73+
``sb_id = ABS(MOD(rh1, N))`` is invariant, but stays ``O(|sb_expr| + |IN list|)``
74+
instead of ``O(k · |sb_expr|)`` so it stays under Redshift's 16 MB statement
75+
limit even on workloads with millions of solved sub-buckets.
76+
77+
Raises ``ValueError`` when both filter inputs are empty: callers must gate the
78+
fetch (eligibility check in the orchestrator) before reaching this helper. An
79+
empty result here would interpolate to ``WHERE )`` downstream — fail-loud beats
80+
silently emitting a syntactically broken query that fail-open would mask.
81+
"""
82+
if not solved_hashes and not unsolved_sb_ids:
83+
raise ValueError(
84+
"build_fingerprint_where_clause requires at least one of solved_hashes "
85+
"or unsolved_sb_ids to be non-empty; the empty case must be filtered "
86+
"out by the caller before issuing a fetch."
87+
)
88+
conditions: list[str] = []
89+
# Sort all IN-list operands for deterministic SQL across dict / list iteration
90+
# orders — helps query-plan caching and unit-test diffing.
91+
if solved_hashes:
92+
sb_list = ", ".join(str(sb_id) for sb_id in sorted(solved_hashes))
93+
hash_list = ", ".join(str(h) for h in sorted({h for hs in solved_hashes.values() for h in hs}))
94+
conditions.append(f"({sb_expr} IN ({sb_list}) AND {rh1_expr} IN ({hash_list}))")
95+
if unsolved_sb_ids:
96+
sb_list = ", ".join(str(sb_id) for sb_id in sorted(unsolved_sb_ids))
97+
conditions.append(f"{sb_expr} IN ({sb_list})")
98+
return " OR ".join(conditions)

0 commit comments

Comments
 (0)