fix(incremental): bound boundary-dedup state growth (#4030)#4131
Open
AndreaBozzo wants to merge 1 commit into
Open
fix(incremental): bound boundary-dedup state growth (#4030)#4131AndreaBozzo wants to merge 1 commit into
AndreaBozzo wants to merge 1 commit into
Conversation
When many records share the boundary (max) cursor value, dlt stores one deduplication hash per record in `unique_hashes` and re-writes the full list to `_dlt_pipeline_state` on every run, with no upper bound — bloating the pipeline-state table to tens/hundreds of MB on backfills or low-resolution cursors. The hashes are load-bearing (dropping them re-emits boundary rows as duplicates on the next run), so rather than change dedup semantics this: - adds an opt-in `Incremental.duplicate_cursor_error_threshold` (ClassVar, disabled by default) that raises `IncrementalCursorThresholdExceeded` once the boundary-hash count crosses it, so state can't silently balloon - rewrites the existing warning to be actionable (higher-resolution cursor, or `merge_key` + `range_start="open"` to disable boundary dedup) - documents the behavior and mitigations in incremental troubleshooting Default behavior is unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
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
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.
Fixes #4030.
Problem
When many records share the boundary (maximum) cursor value, dlt stores one deduplication hash per record in
unique_hashesand re-writes the full list into_dlt_pipeline_stateon every run, with no upper bound. A backfill that stamps every row with the same timestamp, or a low-resolution cursor (a date without a time), produces hundreds of thousands of hashes — tens of MB written to the state table on each run, growing the destination's pipeline-state table atrun_count * state_size.Why not just drop the hashes
The hashes are load-bearing. The closed range start re-fetches boundary rows on the next run, and
unique_hashesis how dlt filters the ones already emitted. Dropping them (e.g. the issue's suggestion to skip them whenprimary_keyis set) re-appends every boundary row on the next run — i.e. it reintroduces duplicate data. I confirmed this with a repro: a 3-run pipeline stays at exactly 1000 rows because of the hashes; without them it would grow to 2000/3000.So this PR keeps the dedup semantics intact and instead makes the cost visible and boundable:
Incremental.duplicate_cursor_error_threshold(ClassVar,None/disabled by default) raises the new terminalIncrementalCursorThresholdExceededonce the boundary-hash count crosses it, so state can't silently balloon. Fails fast on the batch that crosses the bound.duplicate_cursor_warning_thresholdwarning now names the concrete mitigations: use a higher-resolution cursor, or switch tomerge_key+range_start="open"to disable boundary deduplication, or set the error threshold to fail instead of accumulating.Default behavior is unchanged — the cap is off unless explicitly set.
Tests
test_duplicate_cursor_error_threshold: default keeps all hashes and never raises; with the threshold set, crossing it raisesIncrementalCursorThresholdExceeded(terminal, wrapped inPipelineStepFailed).test_duplicate_cursor_warning_threshold: the warning fires once and includes therange_start='open'/duplicate_cursor_error_thresholdguidance.