fix(incremental): bound boundary-dedup state growth (#4030)

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",

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)

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.

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