feat: add pages_per_checkpoint_interval to declarative schema for incremental sync cursors

Co-Authored-By: gl_anatolii.yatsuk@airbyte.io <gl_anatolii.yatsuk@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -876,6 +876,10 @@ definitions:
         title: Inject Start Value Into Outgoing HTTP Request
         description: Optionally configures how the start value will be sent in requests to the source API.
         "$ref": "#/definitions/RequestOption"
+      pages_per_checkpoint_interval:
+        title: Pages Per Checkpoint Interval
+        description: The number of pages to fetch before emitting an intermediate state checkpoint during pagination. This enables resuming long-running syncs closer to where they failed rather than from the beginning of the current slice. Only effective when records are returned in ascending cursor order. Defaults to disabled.
+        type: integer
       $parameters:
         type: object
         additionalProperties: true
@@ -1160,6 +1164,10 @@ definitions:
           - "P1W"
           - "P1M"
           - "P1Y"
+      pages_per_checkpoint_interval:
+        title: Pages Per Checkpoint Interval
+        description: The number of pages to fetch before emitting an intermediate state checkpoint during pagination. This enables resuming long-running syncs closer to where they failed rather than from the beginning of the current slice. Only effective when records are returned in ascending cursor order. Defaults to disabled.
+        type: integer
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1604,6 +1604,11 @@ class IncrementingCountCursor(BaseModel):
         description="Optionally configures how the start value will be sent in requests to the source API.",
         title="Inject Start Value Into Outgoing HTTP Request",
     )
+    pages_per_checkpoint_interval: Optional[int] = Field(
+        None,
+        description="The number of pages to fetch before emitting an intermediate state checkpoint during pagination. This enables resuming long-running syncs closer to where they failed rather than from the beginning of the current slice. Only effective when records are returned in ascending cursor order. Defaults to disabled.",
+        title="Pages Per Checkpoint Interval",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -1720,6 +1725,11 @@ class DatetimeBasedCursor(BaseModel):
         examples=["P1W", "{{ config['step_increment'] }}"],
         title="Step",
     )
+    pages_per_checkpoint_interval: Optional[int] = Field(
+        None,
+        description="The number of pages to fetch before emitting an intermediate state checkpoint during pagination. This enables resuming long-running syncs closer to where they failed rather than from the beginning of the current slice. Only effective when records are returned in ascending cursor order. Defaults to disabled.",
+        title="Pages Per Checkpoint Interval",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -546,7 +546,6 @@
     NoopFileWriter,
 )
 from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import (
-    DEFAULT_PAGES_PER_CHECKPOINT_INTERVAL,
     PaginationTracker,
 )
 from airbyte_cdk.sources.declarative.schema import (
@@ -3466,24 +3465,28 @@ def _get_url(req: Requester) -> str:
             additional_query_properties=query_properties,
             log_formatter=self._get_log_formatter(log_formatter, name),
             pagination_tracker_factory=self._create_pagination_tracker_factory(
-                model.pagination_reset, cursor
+                model.pagination_reset,
+                cursor,
+                incremental_sync.pages_per_checkpoint_interval if incremental_sync else None,
             ),
             parameters=model.parameters or {},
         )
 
     def _create_pagination_tracker_factory(
-        self, model: Optional[PaginationResetModel], cursor: Cursor
+        self,
+        model: Optional[PaginationResetModel],
+        cursor: Cursor,
+        pages_per_checkpoint_interval: int | None = None,
     ) -> Callable[[], PaginationTracker]:
         checkpoint_cursor: Optional[ConcurrentCursor] = (
             cursor if isinstance(cursor, ConcurrentCursor) else None
         )
+        effective_interval = pages_per_checkpoint_interval if checkpoint_cursor else None
 
         if model is None:
             return lambda: PaginationTracker(
                 checkpoint_cursor=checkpoint_cursor,
-                pages_per_checkpoint_interval=DEFAULT_PAGES_PER_CHECKPOINT_INTERVAL
-                if checkpoint_cursor
-                else None,
+                pages_per_checkpoint_interval=effective_interval,
             )
 
         # Until we figure out a way to use any cursor for PaginationTracker, we will have to have this cursor selector logic
@@ -3510,9 +3513,7 @@ def _create_pagination_tracker_factory(
             cursor_factory(),
             limit,
             checkpoint_cursor=checkpoint_cursor,
-            pages_per_checkpoint_interval=DEFAULT_PAGES_PER_CHECKPOINT_INTERVAL
-            if checkpoint_cursor
-            else None,
+            pages_per_checkpoint_interval=effective_interval,
         )
 
     def _get_log_formatter(