One cursor per PaginationTracker

Author: maxime.c

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3361,11 +3361,11 @@ def _create_pagination_tracker_factory(
             return lambda: PaginationTracker()
 
         # Until we figure out a way to use any cursor for PaginationTracker, we will have to have this cursor selector logic
-        cursor_for_pagination_tracking = None
+        cursor_factory: Callable[[], Optional[ConcurrentCursor]] = lambda: None
         if isinstance(cursor, ConcurrentCursor):
-            cursor_for_pagination_tracking = cursor
+            cursor_factory = lambda: cursor.copy_without_state()  # type: ignore  # the if condition validates that it is a ConcurrentCursor
         elif isinstance(cursor, ConcurrentPerPartitionCursor):
-            cursor_for_pagination_tracking = cursor._cursor_factory.create(  # type: ignore  # if this becomes a problem, we would need to extract the cursor_factory instantiation logic and make it accessible here
+            cursor_factory = lambda: cursor._cursor_factory.create(  # type: ignore  # if this becomes a problem, we would need to extract the cursor_factory instantiation logic and make it accessible here
                 {}, datetime.timedelta(0)
             )
         elif not isinstance(cursor, FinalStateCursor):
@@ -3374,7 +3374,7 @@ def _create_pagination_tracker_factory(
             )
 
         limit = model.limits.number_of_records if model and model.limits else None
-        return lambda: PaginationTracker(cursor_for_pagination_tracking, limit)
+        return lambda: PaginationTracker(cursor_factory(), limit)
 
     def _get_log_formatter(
         self, log_formatter: Callable[[Response], Any] | None, name: str

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -19,7 +19,7 @@
 )
 
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
-from airbyte_cdk.sources.message import MessageRepository
+from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 from airbyte_cdk.sources.streams.concurrent.clamping import ClampingStrategy, NoClamping
 from airbyte_cdk.sources.streams.concurrent.cursor_types import CursorValueType, GapType
@@ -136,6 +136,24 @@ class ConcurrentCursor(Cursor):
     _START_BOUNDARY = 0
     _END_BOUNDARY = 1
 
+    def copy_without_state(self) -> "ConcurrentCursor":
+        return self.__class__(
+            stream_name=self._stream_name,
+            stream_namespace=self._stream_namespace,
+            stream_state={},
+            message_repository=NoopMessageRepository(),
+            connector_state_manager=ConnectorStateManager(),
+            connector_state_converter=self._connector_state_converter,
+            cursor_field=self._cursor_field,
+            slice_boundary_fields=self._slice_boundary_fields,
+            start=self._start,
+            end_provider=self._end_provider,
+            lookback_window=self._lookback_window,
+            slice_range=self._slice_range,
+            cursor_granularity=self._cursor_granularity,
+            clamping_strategy=self._clamping_strategy,
+        )
+
     def __init__(
         self,
         stream_name: str,