fix(low-code): data feed stop condition with client-side incremental

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1295,11 +1295,11 @@ definitions:
           - "P1D"
       is_data_feed:
         title: Data Feed API
-        description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination.
+        description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.
         type: boolean
       is_client_side_incremental:
         title: Client-side Incremental Filtering
-        description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, and only emit new records from the last sync, hence incremental. This means that all records would be read from the API, but only new records will be emitted to the destination.
+        description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.
         type: boolean
       is_compare_strictly:
         title: Strict Start-End Time Comparison

airbyte_cdk/sources/declarative/extractors/record_filter.py

@@ -72,9 +72,12 @@ def filter_records(
             record
             for record in records
             if self._cursor.should_be_synced(
-                # Record is created on the fly to align with cursors interface; stream name is ignored as we don't need it here
-                # Record stream name is empty because it is not used during the filtering
-                Record(data=record, associated_slice=stream_slice, stream_name="")
+                record
+                if isinstance(record, Record)
+                # Record is created on the fly to align with cursors interface; stream name is empty because it is not
+                # used during the filtering. Callers that already hold records pass them through untouched, so that the
+                # cursor keeps seeing the real stream name and slice.
+                else Record(data=record, associated_slice=stream_slice, stream_name="")
             )
         )
         if self.condition:

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1857,12 +1857,12 @@ class DatetimeBasedCursor(BaseModel):
     )
     is_data_feed: Optional[bool] = Field(
         None,
-        description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination.",
+        description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.",
         title="Data Feed API",
     )
     is_client_side_incremental: Optional[bool] = Field(
         None,
-        description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, and only emit new records from the last sync, hence incremental. This means that all records would be read from the API, but only new records will be emitted to the destination.",
+        description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.",
         title="Client-side Incremental Filtering",
     )
     is_compare_strictly: Optional[bool] = Field(

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3330,6 +3330,7 @@ def create_record_selector(
         transformations: List[RecordTransformation] | None = None,
         decoder: Decoder | None = None,
         client_side_incremental_sync_cursor: Optional[Cursor] = None,
+        is_client_side_incremental_sync: bool = False,
         file_uploader: Optional[DefaultFileUploader] = None,
         **kwargs: Any,
     ) -> RecordSelector:
@@ -3342,8 +3343,16 @@ def create_record_selector(
             else None
         )
 
+        # A client-side incremental stream transforms before filtering by default. That default belongs to the flag,
+        # not to the component that ends up doing the cursor comparison: a data feed does it in the retriever and
+        # receives no cursor here, but its `record_filter` condition must keep running after the transformations.
+        default_transform_before_filtering = bool(
+            client_side_incremental_sync_cursor or is_client_side_incremental_sync
+        )
         transform_before_filtering = (
-            False if model.transform_before_filtering is None else model.transform_before_filtering
+            default_transform_before_filtering
+            if model.transform_before_filtering is None
+            else model.transform_before_filtering
         )
         if client_side_incremental_sync_cursor:
             record_filter = ClientSideIncrementalRecordFilterDecorator(
@@ -3354,11 +3363,6 @@ def create_record_selector(
                 else None,
                 cursor=client_side_incremental_sync_cursor,
             )
-            transform_before_filtering = (
-                True
-                if model.transform_before_filtering is None
-                else model.transform_before_filtering
-            )
 
         if model.schema_normalization is None:
             # default to no schema normalization if not set
@@ -3465,6 +3469,33 @@ def _get_url(req: Requester) -> str:
         if cursor is None:
             cursor = FinalStateCursor(name, None, self._message_repository)
 
+        # A data feed drops the records the cursor considers already synced in the retriever, which sits downstream of
+        # the paginator. Letting the record selector drop them as well would be redundant and would hide them from the
+        # pagination stop condition, so a data feed never delegates that filtering to the record selector, whether
+        # `is_client_side_incremental` is set or not. The `condition` from `record_filter` is intentionally left out of
+        # the post-pagination filter and stays in the record selector, which preserves the existing behaviour: the
+        # selector runs inside the page loop, so the records the condition rejects never reach the paginator's
+        # accounting. Moving it downstream would start counting them.
+        post_pagination_filter = (
+            ClientSideIncrementalRecordFilterDecorator(
+                config=config,
+                parameters=model.parameters or {},
+                condition=None,
+                cursor=cursor,
+            )
+            if has_stop_condition_cursor
+            else None
+        )
+        client_side_incremental_cursor = (
+            cursor if is_client_side_incremental_sync and not post_pagination_filter else None
+        )
+        if post_pagination_filter and is_client_side_incremental_sync:
+            LOGGER.warning(
+                f"Stream {name}: `is_client_side_incremental` adds no record filtering when `is_data_feed` is set, "
+                "as a data feed already filters on the cursor value. It still makes the record selector apply the "
+                "transformations before the `record_filter` condition."
+            )
+
         decoder = (
             self._create_component_from_model(model=model.decoder, config=config)
             if model.decoder
@@ -3476,7 +3507,8 @@ def _get_url(req: Requester) -> str:
             config=config,
             decoder=decoder,
             transformations=transformations,
-            client_side_incremental_sync_cursor=cursor if is_client_side_incremental_sync else None,
+            client_side_incremental_sync_cursor=client_side_incremental_cursor,
+            is_client_side_incremental_sync=is_client_side_incremental_sync,
             file_uploader=file_uploader,
         )
 
@@ -3611,6 +3643,7 @@ def _get_url(req: Requester) -> str:
                 request_option_provider=request_options_provider,
                 config=config,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
+                post_pagination_filter=post_pagination_filter,
                 parameters=model.parameters or {},
             )
 
@@ -3636,6 +3669,7 @@ def _get_url(req: Requester) -> str:
             pagination_tracker_factory=self._create_pagination_tracker_factory(
                 model.pagination_reset, cursor
             ),
+            post_pagination_filter=post_pagination_filter,
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -24,6 +24,9 @@
 from typing_extensions import deprecated
 
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
+from airbyte_cdk.sources.declarative.extractors.record_filter import (
+    ClientSideIncrementalRecordFilterDecorator,
+)
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import (
     SinglePartitionRouter,
@@ -72,6 +75,8 @@ class SimpleRetriever(Retriever):
         paginator (Optional[Paginator]): The paginator
         stream_slicer (Optional[StreamSlicer]): The stream slicer
         parameters (Mapping[str, Any]): Additional runtime parameters to be used for string interpolation
+        post_pagination_filter (Optional[ClientSideIncrementalRecordFilterDecorator]): Set for data feed streams only.
+            Records the cursor considers already synced are dropped once pagination has observed them
     """
 
     requester: Requester
@@ -95,6 +100,7 @@ class SimpleRetriever(Retriever):
     pagination_tracker_factory: Callable[[], PaginationTracker] = field(
         default_factory=lambda: lambda: PaginationTracker()
     )
+    post_pagination_filter: Optional[ClientSideIncrementalRecordFilterDecorator] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._paginator = self.paginator or NoPagination(parameters=parameters)
@@ -457,7 +463,21 @@ def read_records(
             stream_slice=stream_slice,
             records_schema=records_schema,
         )
-        yield from self._read_pages(record_generator, _slice)
+        records: Iterable[Mapping[str, Any]] = self._read_pages(record_generator, _slice)
+        if self.post_pagination_filter:
+            # A data feed paginates until it reaches a record older than the cursor, so the page that triggers the stop
+            # condition still holds already-synced records. Those are filtered here rather than in the record selector
+            # so that the paginator keeps seeing the whole page: the stop condition is evaluated on the last record of
+            # the page, which is precisely one of the records being dropped. Two consequences of filtering this late:
+            # the pagination tracker observes the dropped records, and a `file_uploader` on the record selector has
+            # already uploaded their files by the time they are dropped.
+            records = self.post_pagination_filter.filter_records(
+                records,
+                # the filter is only used for its cursor comparison, which does not read the stream state
+                stream_state={},
+                stream_slice=_slice,
+            )
+        yield from records
 
     def _parse_records(
         self,