feat: Add support for per-partition state and IncrementingCountCursor validation

- Add _extract_cursor_value_from_state helper to handle different state structures
- For per-partition state, use global cursor value from 'state' key
- Raise ValueError when IncrementingCountCursor is used with api_retention_period
- Add unit tests for per-partition state (cursor too old and within retention)
- Add unit test for IncrementingCountCursor error handling

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3599,13 +3599,22 @@ def _is_cursor_older_than_retention_period(
         If the cursor is too old, the incremental API may not have data going back that far,
         so we should fall back to a full refresh to avoid data loss.
 
-        The state could have been produced by either full_refresh_stream (first sync) or
-        incremental_stream (subsequent syncs), so we try parsing with formats from both.
+        This method handles different state structures:
+        - Simple cursor state: {"cursor_field": "value"}
+        - Per-partition state: {"state": {"cursor_field": "value"}, "states": [...]}
 
         Returns True if the cursor is older than the retention period or if the cursor is
         invalid/unparseable (should use full refresh).
         Returns False if the cursor is within the retention period (safe to use incremental).
         """
+        for incremental_sync in incremental_sync_sources:
+            if isinstance(incremental_sync, IncrementingCountCursorModel):
+                raise ValueError(
+                    f"Stream '{stream_name}' uses IncrementingCountCursor which is not supported "
+                    f"with api_retention_period. IncrementingCountCursor does not use datetime-based "
+                    f"cursors, so cursor age validation cannot be performed."
+                )
+
         cursor_field = None
         for incremental_sync in incremental_sync_sources:
             cursor_field = getattr(incremental_sync, "cursor_field", None)
@@ -3615,7 +3624,7 @@ def _is_cursor_older_than_retention_period(
         if not cursor_field:
             return True
 
-        cursor_value = stream_state.get(cursor_field)
+        cursor_value = self._extract_cursor_value_from_state(stream_state, cursor_field)
         if not cursor_value:
             return True
 
@@ -3641,6 +3650,29 @@ def _is_cursor_older_than_retention_period(
 
         return False
 
+    def _extract_cursor_value_from_state(
+        self,
+        stream_state: Mapping[str, Any],
+        cursor_field: str,
+    ) -> Any:
+        """Extract cursor value from state, handling different state structures.
+
+        Supports:
+        - Simple cursor state: {"cursor_field": "value"} -> returns "value"
+        - Per-partition state: {"state": {"cursor_field": "value"}, ...} -> returns "value"
+          (uses global cursor from "state" key)
+
+        Returns None if cursor value cannot be extracted.
+        """
+        if cursor_field in stream_state:
+            return stream_state.get(cursor_field)
+
+        global_state = stream_state.get("state")
+        if isinstance(global_state, dict) and cursor_field in global_state:
+            return global_state.get(cursor_field)
+
+        return None
+
     def _parse_cursor_datetime(
         self,
         cursor_value: str,

unit_tests/sources/declarative/test_state_delegating_stream.py

@@ -8,6 +8,7 @@
 from unittest.mock import MagicMock
 
 import freezegun
+import pytest
 
 from airbyte_cdk.models import (
     AirbyteStateBlob,
@@ -405,3 +406,148 @@ def test_cursor_age_validation_emits_warning_when_falling_back(caplog):
         assert any(
             "TestStream" in msg and "older than" in msg and "P7D" in msg for msg in warning_messages
         ), f"Expected warning about stale cursor not found. Warnings: {warning_messages}"
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_cursor_age_validation_with_per_partition_state_uses_global_cursor():
+    """Test that per-partition state structure uses global cursor for age validation."""
+    manifest = _create_manifest_with_retention_period("P7D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/items"),
+            HttpResponse(
+                body=json.dumps(
+                    [
+                        {"id": 1, "name": "item_1", "updated_at": "2024-07-13"},
+                    ]
+                )
+            ),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                    stream_state=AirbyteStateBlob(
+                        state={"updated_at": "2024-07-01"},
+                        states=[
+                            {
+                                "partition": {"parent_id": "1"},
+                                "cursor": {"updated_at": "2024-07-10"},
+                            },
+                            {
+                                "partition": {"parent_id": "2"},
+                                "cursor": {"updated_at": "2024-07-05"},
+                            },
+                        ],
+                        use_global_cursor=False,
+                    ),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        assert len(records) == 1
+
+
+@freezegun.freeze_time("2024-07-15")
+def test_cursor_age_validation_with_per_partition_state_within_retention():
+    """Test per-partition state with global cursor within retention uses incremental.
+
+    This test verifies that when the global cursor in a per-partition state structure
+    is within the retention period, the incremental stream is selected (not full refresh).
+    We verify this by checking that the incremental endpoint is called, not the full refresh one.
+    """
+    manifest = _create_manifest_with_retention_period("P30D")
+
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(
+                url="https://api.test.com/items_with_filtration",
+                query_params={"start": "2024-07-01", "end": "2024-07-15"},
+            ),
+            HttpResponse(
+                body=json.dumps(
+                    [
+                        {"id": 3, "name": "item_3", "updated_at": "2024-07-14"},
+                    ]
+                )
+            ),
+        )
+
+        state = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream=AirbyteStreamState(
+                    stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                    stream_state=AirbyteStateBlob(
+                        state={"updated_at": "2024-07-10"},
+                        states=[
+                            {
+                                "partition": {"parent_id": "1"},
+                                "cursor": {"updated_at": "2024-07-10"},
+                            },
+                        ],
+                        use_global_cursor=False,
+                    ),
+                ),
+            )
+        ]
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=state
+        )
+        configured_catalog = create_configured_catalog(source, _CONFIG)
+
+        records = get_records(source, _CONFIG, configured_catalog, state)
+        assert len(records) == 1
+
+
+def _create_manifest_with_incrementing_count_cursor(api_retention_period: str) -> dict:
+    """Create a manifest with IncrementingCountCursor and api_retention_period."""
+    manifest = copy.deepcopy(_MANIFEST)
+    manifest["definitions"]["TestStream"]["api_retention_period"] = api_retention_period
+
+    incrementing_cursor = {
+        "type": "IncrementingCountCursor",
+        "cursor_field": "id",
+        "start_value": 0,
+    }
+    manifest["definitions"]["TestStream"]["full_refresh_stream"]["incremental_sync"] = (
+        incrementing_cursor
+    )
+    manifest["definitions"]["TestStream"]["incremental_stream"]["incremental_sync"] = (
+        incrementing_cursor
+    )
+    return manifest
+
+
+def test_cursor_age_validation_raises_error_for_incrementing_count_cursor():
+    """Test that IncrementingCountCursor with api_retention_period raises an error."""
+    manifest = _create_manifest_with_incrementing_count_cursor("P7D")
+
+    state = [
+        AirbyteStateMessage(
+            type=AirbyteStateType.STREAM,
+            stream=AirbyteStreamState(
+                stream_descriptor=StreamDescriptor(name="TestStream", namespace=None),
+                stream_state=AirbyteStateBlob(id=100),
+            ),
+        )
+    ]
+
+    source = ConcurrentDeclarativeSource(
+        source_config=manifest, config=_CONFIG, catalog=None, state=state
+    )
+
+    with pytest.raises(ValueError) as exc_info:
+        source.discover(logger=MagicMock(), config=_CONFIG)
+
+    assert "IncrementingCountCursor" in str(exc_info.value)
+    assert "not supported" in str(exc_info.value)
+    assert "api_retention_period" in str(exc_info.value)