diff --git a/dlt/extract/incremental/__init__.py b/dlt/extract/incremental/__init__.py index f8cd0a44d4..8bb63a15ca 100644 --- a/dlt/extract/incremental/__init__.py +++ b/dlt/extract/incremental/__init__.py @@ -53,6 +53,7 @@ from dlt.extract.incremental.exceptions import ( ExternalSchedulerNotAvailable, IncrementalCursorPathMissing, + IncrementalCursorThresholdExceeded, IncrementalPrimaryKeyMissing, JoinSchedulerError, ) @@ -141,6 +142,8 @@ class Incremental( on_cursor_value_missing: OnCursorValueMissing = "raise" lag: Optional[float] = None duplicate_cursor_warning_threshold: ClassVar[int] = 200 + duplicate_cursor_error_threshold: ClassVar[Optional[int]] = None + """When set, raises `IncrementalCursorThresholdExceeded` once more than this many records share the boundary cursor value. Disabled (`None`) by default.""" range_start: TIncrementalRange = "closed" range_end: TIncrementalRange = "open" @@ -721,13 +724,22 @@ def __call__(self, rows: TDataItems, meta: Any = None) -> Optional[TDataItems]: def _check_duplicate_cursor_threshold( self, initial_hash_count: int, final_hash_count: int ) -> None: + error_threshold = Incremental.duplicate_cursor_error_threshold + if error_threshold is not None and final_hash_count > error_threshold: + raise IncrementalCursorThresholdExceeded( + self.resource_name, self.cursor_path, final_hash_count, error_threshold + ) if initial_hash_count <= Incremental.duplicate_cursor_warning_threshold < final_hash_count: logger.warning( f"Large number of records ({final_hash_count}) sharing the same value of cursor" - f" field '{self.cursor_path}' on resource '{self.resource_name}'. This can happen" - " if the cursor field has a low resolution (e.g., only stores dates without" - " times), causing many records to share the same cursor value. Consider using a" - " cursor column with higher resolution to reduce the deduplication state size." + f" field '{self.cursor_path}' on resource '{self.resource_name}'. dlt keeps one" + " deduplication hash per record at the boundary cursor value in the pipeline state" + " and re-writes it on every run, so a low-resolution cursor (e.g. a date without" + " time, or a backfill that stamps every row with the same timestamp) makes the" + " state grow unbounded. To reduce it: use a cursor column with higher resolution," + " or switch to `merge_key` together with `range_start='open'` to disable boundary" + " deduplication, or set `Incremental.duplicate_cursor_error_threshold` to fail" + " instead of silently accumulating state." ) @@ -963,6 +975,7 @@ def incremental_config_to_instance(cfg: TIncrementalConfig) -> Incremental[Any]: "IncrementalResourceWrapper", "IncrementalColumnState", "IncrementalCursorPathMissing", + "IncrementalCursorThresholdExceeded", "IncrementalPrimaryKeyMissing", "IncrementalUnboundError", "TIncrementalconfig", diff --git a/dlt/extract/incremental/exceptions.py b/dlt/extract/incremental/exceptions.py index 9453b8c0dc..e083a947c1 100644 --- a/dlt/extract/incremental/exceptions.py +++ b/dlt/extract/incremental/exceptions.py @@ -66,6 +66,23 @@ def __init__(self, pipe_name: str, primary_key_column: str, item: TDataItem) -> super().__init__(pipe_name, msg) +class IncrementalCursorThresholdExceeded(PipeException): + def __init__(self, pipe_name: str, cursor_path: str, hash_count: int, threshold: int) -> None: + self.cursor_path = cursor_path + self.hash_count = hash_count + self.threshold = threshold + msg = ( + f"Number of records ({hash_count}) sharing the same value of cursor field" + f" `{cursor_path}` exceeded `duplicate_cursor_error_threshold` ({threshold}). dlt keeps" + " one deduplication hash per boundary record in the pipeline state, so this many" + " records at the boundary value would write a large state on every run. Use a cursor" + " column with higher resolution, or switch to `merge_key` together with" + " `range_start='open'` to disable boundary deduplication, or raise" + " `duplicate_cursor_error_threshold` if a large state is acceptable." + ) + super().__init__(pipe_name, msg) + + class JoinSchedulerError(PipeException): def __init__(self, pipe_name: str, msg: str) -> None: super().__init__(pipe_name, msg) diff --git a/docs/website/docs/general-usage/incremental/troubleshooting.md b/docs/website/docs/general-usage/incremental/troubleshooting.md index f43fc798f6..cab651f440 100644 --- a/docs/website/docs/general-usage/incremental/troubleshooting.md +++ b/docs/website/docs/general-usage/incremental/troubleshooting.md @@ -63,6 +63,20 @@ sources: Verify that the `last_value` is updated between pipeline runs. +### Growing pipeline state from boundary deduplication + +To avoid loading the same row twice, dlt keeps a deduplication hash in `unique_hashes` for **every record that shares the current boundary (maximum) cursor value**. This list is stored in the pipeline state and re-written on every run. When many records share the same cursor value — for example a low-resolution cursor (a date without a time) or a backfill that stamps every row with the same timestamp — `unique_hashes` can grow to tens of MB and bloat the `_dlt_pipeline_state` table. + +dlt logs a warning once the number of boundary records passes `Incremental.duplicate_cursor_warning_threshold` (200 by default). To address it: + +- Use a cursor column with higher resolution so fewer records share the same value. +- If you do not need boundary deduplication (for example because a `primary_key` merge already removes duplicates, or rows never overlap between runs), switch to `merge_key` together with `range_start="open"`, which excludes records equal to the last value and disables the deduplication state. +- Set `Incremental.duplicate_cursor_error_threshold` (disabled by default) to fail with `IncrementalCursorThresholdExceeded` instead of silently accumulating state once the boundary record count crosses the threshold: + +```py +dlt.sources.incremental.duplicate_cursor_error_threshold = 100_000 +``` + ### Type mismatch errors If you encounter an `IncrementalCursorInvalidCoercion` error, it typically means the `initial_value` type does not match the data type of the field in your source data. diff --git a/tests/extract/test_incremental.py b/tests/extract/test_incremental.py index 97c804d9a2..c3f9a9dfae 100644 --- a/tests/extract/test_incremental.py +++ b/tests/extract/test_incremental.py @@ -39,6 +39,7 @@ IncrementalCursorInvalidCoercion, IncrementalCursorPathHasValueNone, IncrementalCursorPathMissing, + IncrementalCursorThresholdExceeded, IncrementalPrimaryKeyMissing, ) from dlt.extract.incremental.lag import apply_lag @@ -4826,3 +4827,49 @@ def default_typed_test( list(r) # type from typed default [str] assert r.incremental._incremental.get_incremental_value_type() is str + + +def _boundary_source(n: int) -> DltResource: + """Resource whose `n` records all share the same cursor value (the boundary).""" + + @dlt.resource(name="items", write_disposition="append", primary_key="id") + def items() -> Any: + for i in range(n): + yield {"id": i, "ts": "2024-01-01", "v": i} + + items.apply_hints(incremental=dlt.sources.incremental("ts")) + return items + + +def test_duplicate_cursor_warning_threshold(monkeypatch: pytest.MonkeyPatch) -> None: + # warning fires once boundary records exceed the warning threshold, and points at the mitigations + monkeypatch.setattr(Incremental, "duplicate_cursor_warning_threshold", 5) + p = dlt.pipeline(pipeline_name="dup_cursor_warn", destination="duckdb", dev_mode=True) + with mock.patch("dlt.extract.incremental.logger.warning") as warn: + p.run(_boundary_source(10)) + # logger is a shared module, so other steps may warn too: match on the threshold message + threshold_warnings = [ + call.args[0] + for call in warn.call_args_list + if "duplicate_cursor_error_threshold" in call.args[0] + ] + assert len(threshold_warnings) == 1 + assert "range_start='open'" in threshold_warnings[0] + + +def test_duplicate_cursor_error_threshold(monkeypatch: pytest.MonkeyPatch) -> None: + # default (None): unbounded boundary dedup keeps all hashes and never raises + p = dlt.pipeline(pipeline_name="dup_cursor_off", destination="duckdb", dev_mode=True) + p.run(_boundary_source(50)) + state = p.state["sources"]["dup_cursor_off"]["resources"]["items"]["incremental"]["ts"] + assert len(state["unique_hashes"]) == 50 + + # threshold set: crossing it raises a terminal error instead of growing state + monkeypatch.setattr(Incremental, "duplicate_cursor_error_threshold", 10) + p2 = dlt.pipeline(pipeline_name="dup_cursor_err", destination="duckdb", dev_mode=True) + with pytest.raises(PipelineStepFailed) as py_ex: + p2.run(_boundary_source(50)) + inner = py_ex.value.__cause__ + assert isinstance(inner, IncrementalCursorThresholdExceeded) + assert inner.threshold == 10 + assert inner.hash_count > 10