From 5dcd503345c2d0d7e2804ffcdee25a21e0234cda Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 6 Aug 2026 14:29:13 +0300 Subject: [PATCH 1/6] fix(low-code): make is_data_feed stop condition work with is_client_side_incremental MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When both flags were set, the client-side incremental filter dropped below-cursor records before the paginator could observe them, so the CursorStopCondition wired by is_data_feed never fired and every sync re-fetched the full listing. The filter already evaluates should_be_synced on every raw record; it now tracks (per thread, since one retriever is shared across concurrently-read partitions) whether the current page contained a record older than the cursor, and a new FilterAwareStopCondition stops pagination as soon as it did — including when the whole page was filtered out. Co-Authored-By: Claude Fable 5 --- .../declarative/extractors/record_filter.py | 38 ++-- .../parsers/model_to_component_factory.py | 23 ++- .../paginators/strategies/__init__.py | 4 + .../paginators/strategies/stop_condition.py | 39 +++- .../extractors/test_record_filter.py | 56 ++++++ .../test_model_to_component_factory.py | 70 +++++++- .../paginators/test_stop_condition.py | 39 +++- .../test_stop_condition_integration.py | 168 ++++++++++++++++++ 8 files changed, 417 insertions(+), 20 deletions(-) create mode 100644 unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 943068f875..74a0d90e31 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -1,6 +1,7 @@ # # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import threading from dataclasses import InitVar, dataclass from typing import Any, Iterable, Mapping, Optional, Union @@ -60,6 +61,18 @@ def __init__( ): super().__init__(**kwargs) self._cursor = cursor + # One retriever (and hence one record filter) may be shared across partitions that are read + # concurrently, so per-page bookkeeping must be tracked per thread + self._thread_local = threading.local() + + @property + def stale_record_seen_on_current_page(self) -> bool: + """ + Whether the page being filtered contained at least one record the cursor considers already + synced. Used by `FilterAwareStopCondition` to stop paginating on data feed streams, since + such records never reach the paginator. Reset on every `filter_records` call. + """ + return getattr(self._thread_local, "stale_record_seen", False) def filter_records( self, @@ -68,15 +81,8 @@ def filter_records( stream_slice: Optional[StreamSlice] = None, next_page_token: Optional[Mapping[str, Any]] = None, ) -> Iterable[Mapping[str, Any]]: - 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="") - ) - ) + self._thread_local.stale_record_seen = False + records = (record for record in records if self._should_be_synced(record, stream_slice)) if self.condition: records = super().filter_records( records=records, @@ -84,4 +90,16 @@ def filter_records( stream_slice=stream_slice, next_page_token=next_page_token, ) - yield from records + return records + + def _should_be_synced( + self, record: Mapping[str, Any], stream_slice: Optional[StreamSlice] + ) -> bool: + 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="") + ): + return True + self._thread_local.stale_record_seen = True + return False diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 6d9531d29e..4a98c15f3d 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -525,8 +525,10 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies import ( CursorPaginationStrategy, CursorStopCondition, + FilterAwareStopCondition, OffsetIncrement, PageIncrement, + PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) from airbyte_cdk.sources.declarative.requesters.query_properties import ( @@ -2383,6 +2385,9 @@ def create_default_paginator( extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None, decoder: Optional[Decoder] = None, cursor_used_for_stop_condition: Optional[Cursor] = None, + record_filter_used_for_stop_condition: Optional[ + ClientSideIncrementalRecordFilterDecorator + ] = None, ) -> Union[DefaultPaginator, PaginatorTestReadDecorator]: if decoder: if self._is_supported_decoder_for_pagination(decoder): @@ -2408,8 +2413,16 @@ def create_default_paginator( extractor_model=extractor_model, ) if cursor_used_for_stop_condition: + # When client-side incremental filtering is enabled, records older than the cursor are + # dropped before the paginator can observe them, so the stop condition must be driven + # by the record filter instead of the last record emitted for the page + stop_condition: PaginationStopCondition = ( + FilterAwareStopCondition(record_filter_used_for_stop_condition) + if record_filter_used_for_stop_condition + else CursorStopCondition(cursor_used_for_stop_condition) + ) pagination_strategy = StopConditionPaginationStrategyDecorator( - pagination_strategy, CursorStopCondition(cursor_used_for_stop_condition) + pagination_strategy, stop_condition ) paginator = DefaultPaginator( decoder=decoder_to_use, @@ -3524,6 +3537,14 @@ def _get_url(req: Requester) -> str: extractor_model=model.record_selector.extractor, decoder=decoder, cursor_used_for_stop_condition=cursor if has_stop_condition_cursor else None, + record_filter_used_for_stop_condition=( + record_selector.record_filter + if has_stop_condition_cursor + and isinstance( + record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator + ) + else None + ), ) if model.paginator else NoPagination(parameters={}) diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py index c1f9ff1052..cee364d5fd 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py @@ -13,13 +13,17 @@ ) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.stop_condition import ( CursorStopCondition, + FilterAwareStopCondition, + PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) __all__ = [ "CursorPaginationStrategy", "CursorStopCondition", + "FilterAwareStopCondition", "OffsetIncrement", "PageIncrement", + "PaginationStopCondition", "StopConditionPaginationStrategyDecorator", ] diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py index 068df72cb8..6707cc1967 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py @@ -3,7 +3,7 @@ # from abc import ABC, abstractmethod -from typing import Any, Optional +from typing import TYPE_CHECKING, Any, Optional import requests @@ -13,14 +13,21 @@ from airbyte_cdk.sources.streams.concurrent.cursor import Cursor from airbyte_cdk.sources.types import Record +if TYPE_CHECKING: + from airbyte_cdk.sources.declarative.extractors.record_filter import ( + ClientSideIncrementalRecordFilterDecorator, + ) + class PaginationStopCondition(ABC): @abstractmethod - def is_met(self, record: Record) -> bool: + def is_met(self, record: Optional[Record]) -> bool: """ Given a condition is met, the pagination will stop - :param record: a record used to evaluate the condition + :param record: the last record yielded for the current page, if any. Records dropped by + record filters are not visible here — a condition that needs to observe them has to get + that signal from the filter itself (see `FilterAwareStopCondition`). """ raise NotImplementedError() @@ -32,8 +39,26 @@ def __init__( ): self._cursor = cursor - def is_met(self, record: Record) -> bool: - return not self._cursor.should_be_synced(record) + def is_met(self, record: Optional[Record]) -> bool: + return record is not None and not self._cursor.should_be_synced(record) + + +class FilterAwareStopCondition(PaginationStopCondition): + """ + Stop condition for streams combining `is_data_feed` with `is_client_side_incremental`. + + The client-side incremental filter drops records older than the cursor before the paginator can + observe them, so `CursorStopCondition` — which only sees the last record that survived + filtering — would never fire. The filter, however, evaluates `should_be_synced` on every raw + record; this condition stops pagination as soon as the filter reports that the current page + contained a record that was filtered out as already synced. + """ + + def __init__(self, record_filter: "ClientSideIncrementalRecordFilterDecorator"): + self._record_filter = record_filter + + def is_met(self, record: Optional[Record]) -> bool: + return self._record_filter.stale_record_seen_on_current_page class StopConditionPaginationStrategyDecorator(PaginationStrategy): @@ -50,7 +75,9 @@ def next_page_token( ) -> Optional[Any]: # We evaluate in reverse order because the assumption is that most of the APIs using data feed structure # will return records in descending order. In terms of performance/memory, we return the records lazily - if last_record and self._stop_condition.is_met(last_record): + # Note: `last_record` may be None even mid-feed when every record of the page was dropped by + # a record filter, so the stop condition is consulted regardless + if self._stop_condition.is_met(last_record): return None return self._delegate.next_page_token( response, last_page_size, last_record, last_page_token_value diff --git a/unit_tests/sources/declarative/extractors/test_record_filter.py b/unit_tests/sources/declarative/extractors/test_record_filter.py index 9f0cf46d90..9fe59cb04b 100644 --- a/unit_tests/sources/declarative/extractors/test_record_filter.py +++ b/unit_tests/sources/declarative/extractors/test_record_filter.py @@ -456,3 +456,59 @@ def date_time_based_cursor_factory(stream_state, runtime_lookback_window) -> Con ) assert [x.get("id") for x in filtered_records] == expected_record_ids + + +def _stale_record_tracking_filter(should_be_synced_per_record: List[bool]): + cursor = Mock() + cursor.should_be_synced.side_effect = should_be_synced_per_record + return ClientSideIncrementalRecordFilterDecorator( + config={}, + condition="", + parameters={}, + cursor=cursor, + ) + + +def test_stale_record_seen_when_record_is_filtered_out(): + record_filter_decorator = _stale_record_tracking_filter([True, False]) + + assert not record_filter_decorator.stale_record_seen_on_current_page + filtered_records = list( + record_filter_decorator.filter_records( + records=[{"id": 1}, {"id": 2}], + stream_state={}, + stream_slice=StreamSlice(partition={}, cursor_slice={}), + ) + ) + + assert [record["id"] for record in filtered_records] == [1] + assert record_filter_decorator.stale_record_seen_on_current_page + + +def test_stale_record_seen_is_reset_on_each_filter_records_call(): + record_filter_decorator = _stale_record_tracking_filter([False, True]) + + list(record_filter_decorator.filter_records(records=[{"id": 1}], stream_state={})) + assert record_filter_decorator.stale_record_seen_on_current_page + + list(record_filter_decorator.filter_records(records=[{"id": 2}], stream_state={})) + assert not record_filter_decorator.stale_record_seen_on_current_page + + +def test_stale_record_seen_is_tracked_per_thread(): + import threading + + record_filter_decorator = _stale_record_tracking_filter([False]) + list(record_filter_decorator.filter_records(records=[{"id": 1}], stream_state={})) + assert record_filter_decorator.stale_record_seen_on_current_page + + stale_record_seen_on_other_thread = [] + other_thread = threading.Thread( + target=lambda: stale_record_seen_on_other_thread.append( + record_filter_decorator.stale_record_seen_on_current_page + ) + ) + other_thread.start() + other_thread.join() + + assert stale_record_seen_on_other_thread == [False] diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 7728075e11..a8806c166c 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -145,6 +145,8 @@ from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator from airbyte_cdk.sources.declarative.requesters.paginators.strategies import ( CursorPaginationStrategy, + CursorStopCondition, + FilterAwareStopCondition, OffsetIncrement, PageIncrement, StopConditionPaginationStrategyDecorator, @@ -1432,9 +1434,73 @@ def test_incremental_data_feed(): model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config ) + pagination_strategy = get_retriever(stream).paginator.pagination_strategy + assert isinstance(pagination_strategy, StopConditionPaginationStrategyDecorator) + assert isinstance(pagination_strategy._stop_condition, CursorStopCondition) + + +def test_incremental_data_feed_with_client_side_incremental(): + content = """ +selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["extractor_path"] +requester: + type: HttpRequester + name: "{{ parameters['name'] }}" + url_base: "https://api.sendgrid.com/v3/" + http_method: "GET" +list_stream: + type: DeclarativeStream + incremental_sync: + type: DatetimeBasedCursor + $parameters: + datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" + start_datetime: "{{ config['start_time'] }}" + cursor_field: "created" + is_data_feed: true + is_client_side_incremental: true + retriever: + type: SimpleRetriever + name: "{{ parameters['name'] }}" + paginator: + type: DefaultPaginator + pagination_strategy: + type: "CursorPagination" + cursor_value: "{{ response._metadata.next }}" + page_size: 10 + requester: + $ref: "#/requester" + path: "/" + record_selector: + $ref: "#/selector" + $parameters: + name: "lists" + """ + + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + stream_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["list_stream"], {} + ) + + stream = factory.create_component( + model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config + ) + + retriever = get_retriever(stream) + pagination_strategy = retriever.paginator.pagination_strategy + assert isinstance(pagination_strategy, StopConditionPaginationStrategyDecorator) + # the client-side filter drops below-cursor records before the paginator sees them, so the stop + # condition must observe the filter rather than the last emitted record + assert isinstance(pagination_strategy._stop_condition, FilterAwareStopCondition) assert isinstance( - get_retriever(stream).paginator.pagination_strategy, - StopConditionPaginationStrategyDecorator, + retriever.record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator + ) + assert ( + pagination_strategy._stop_condition._record_filter + is retriever.record_selector.record_filter ) diff --git a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py index b89baf4430..c48b2a7771 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py @@ -9,8 +9,12 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) +from airbyte_cdk.sources.declarative.extractors.record_filter import ( + ClientSideIncrementalRecordFilterDecorator, +) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.stop_condition import ( CursorStopCondition, + FilterAwareStopCondition, PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) @@ -47,6 +51,23 @@ def test_given_record_should_not_be_synced_when_is_met_return_true(mocked_cursor assert CursorStopCondition(mocked_cursor).is_met(ANY_RECORD) +def test_given_no_record_when_is_met_return_false(mocked_cursor): + assert not CursorStopCondition(mocked_cursor).is_met(NO_RECORD) + mocked_cursor.should_be_synced.assert_not_called() + + +def test_given_stale_record_seen_by_filter_when_is_met_return_true(): + record_filter = Mock(spec=ClientSideIncrementalRecordFilterDecorator) + record_filter.stale_record_seen_on_current_page = True + assert FilterAwareStopCondition(record_filter).is_met(NO_RECORD) + + +def test_given_no_stale_record_seen_by_filter_when_is_met_return_false(): + record_filter = Mock(spec=ClientSideIncrementalRecordFilterDecorator) + record_filter.stale_record_seen_on_current_page = False + assert not FilterAwareStopCondition(record_filter).is_met(ANY_RECORD) + + def test_given_stop_condition_is_met_when_next_page_token_then_return_none( mocked_pagination_strategy, mocked_stop_condition ): @@ -92,9 +113,10 @@ def test_given_stop_condition_is_not_met_when_next_page_token_then_delegate( mocked_stop_condition.is_met.assert_has_calls([call(last_record)]) -def test_given_no_records_when_next_page_token_then_delegate( +def test_given_no_records_and_stop_condition_is_not_met_when_next_page_token_then_delegate( mocked_pagination_strategy, mocked_stop_condition ): + mocked_stop_condition.is_met.return_value = False decorator = StopConditionPaginationStrategyDecorator( mocked_pagination_strategy, mocked_stop_condition ) @@ -105,6 +127,21 @@ def test_given_no_records_when_next_page_token_then_delegate( mocked_pagination_strategy.next_page_token.assert_called_once_with( ANY_RESPONSE, 0, NO_RECORD, None ) + mocked_stop_condition.is_met.assert_called_once_with(NO_RECORD) + + +def test_given_no_records_and_stop_condition_is_met_when_next_page_token_then_return_none( + mocked_pagination_strategy, mocked_stop_condition +): + # A page can yield no records even mid-feed when a record filter dropped all of them, so the + # stop condition must be consulted even without a last record + mocked_stop_condition.is_met.return_value = True + decorator = StopConditionPaginationStrategyDecorator( + mocked_pagination_strategy, mocked_stop_condition + ) + + assert not decorator.next_page_token(ANY_RESPONSE, 0, NO_RECORD) + mocked_pagination_strategy.next_page_token.assert_not_called() def test_when_get_page_size_then_delegate(mocked_pagination_strategy, mocked_stop_condition): diff --git a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py new file mode 100644 index 0000000000..534a40e44f --- /dev/null +++ b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py @@ -0,0 +1,168 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +""" +End-to-end coverage for the pagination stop condition of data feed streams that also use +client-side incremental filtering. The client-side filter drops records older than the cursor +before the paginator can observe them, so the stop condition is driven by the filter itself: +pagination must stop on the first page containing a record older than the cursor, while still +emitting only the records newer than the cursor. +""" + +import json +import logging +from typing import Any, List, Mapping, Optional + +import requests_mock + +from airbyte_cdk.models import ( + AirbyteStateBlob, + AirbyteStateMessage, + AirbyteStateType, + AirbyteStreamState, + ConfiguredAirbyteCatalogSerializer, + StreamDescriptor, + Type, +) +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + ConcurrentDeclarativeSource, +) + +_MANIFEST = { + "version": "7.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["items"]}, + "spec": { + "type": "Spec", + "connection_specification": {"type": "object", "properties": {}}, + }, + "streams": [ + { + "type": "DeclarativeStream", + "name": "items", + "primary_key": ["id"], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "updated_at": {"type": "string"}, + }, + }, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.example.com", + "path": "/items", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": { + "type": "DefaultPaginator", + "pagination_strategy": { + "type": "PageIncrement", + "page_size": 4, + "start_from_page": 1, + "inject_on_first_request": True, + }, + "page_token_option": { + "type": "RequestOption", + "inject_into": "request_parameter", + "field_name": "page", + }, + }, + }, + "incremental_sync": { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + "start_datetime": { + "type": "MinMaxDatetime", + "datetime": "2019-01-01T00:00:00Z", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + }, + "is_data_feed": True, + "is_client_side_incremental": True, + }, + } + ], +} + +# Records are sorted in descending order of updated_at, as expected from a data feed +_PAGE_1 = [ + {"id": 4, "updated_at": "2022-06-01T00:00:00Z"}, + {"id": 3, "updated_at": "2022-05-01T00:00:00Z"}, + {"id": 2, "updated_at": "2020-06-01T00:00:00Z"}, + {"id": 1, "updated_at": "2020-05-01T00:00:00Z"}, +] +_PAGE_2 = [ + {"id": 0, "updated_at": "2020-04-01T00:00:00Z"}, +] + +_CATALOG = ConfiguredAirbyteCatalogSerializer.load( + { + "streams": [ + { + "stream": { + "name": "items", + "json_schema": {}, + "supported_sync_modes": ["full_refresh", "incremental"], + }, + "sync_mode": "incremental", + "destination_sync_mode": "append", + } + ] + } +) + + +def _read(state: Optional[List[AirbyteStateMessage]]) -> tuple[List[str], List[Mapping[str, Any]]]: + pages_fetched = [] + + def paged_response(request: Any, context: Any) -> str: + page = request.qs.get("page", ["1"])[0] + pages_fetched.append(page) + return json.dumps(_PAGE_1 if page == "1" else _PAGE_2) + + source = ConcurrentDeclarativeSource( + source_config=_MANIFEST, config={}, catalog=_CATALOG, state=state + ) + with requests_mock.Mocker() as http_mocker: + http_mocker.get("https://api.example.com/items", text=paged_response) + records = [ + message.record.data + for message in source.read(logging.getLogger("test"), {}, _CATALOG, state) + if message.type == Type.RECORD + ] + return pages_fetched, records + + +def test_given_stale_records_on_page_when_client_side_incremental_then_stop_pagination(): + state = [ + AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name="items"), + stream_state=AirbyteStateBlob({"updated_at": "2021-01-01T00:00:00Z"}), + ), + ) + ] + + pages_fetched, records = _read(state) + + assert sorted(record["id"] for record in records) == [3, 4] + assert pages_fetched == ["1"] + + +def test_given_no_stale_records_when_client_side_incremental_then_paginate_until_the_end(): + pages_fetched, records = _read(None) + + assert sorted(record["id"] for record in records) == [0, 1, 2, 3, 4] + assert pages_fetched == ["1", "2"] From 16d72e9d00c9bf3f598a12e0bb8aa93971fe10f9 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:09:49 +0300 Subject: [PATCH 2/6] refactor(low-code): filter the data feed boundary page in the retriever Replaces the previous approach, which leaked per-page state from ClientSideIncrementalRecordFilterDecorator back to a new FilterAwareStopCondition through a thread-local flag. The underlying problem is one of layering: the client-side filter runs in RecordSelector, upstream of where SimpleRetriever._read_pages computes last_record, so the record that should trigger the stop condition is already gone by the time the paginator is consulted. Moving the cursor filtering downstream of _read_pages fixes it without any shared mutable state: the paginator sees the page exactly as the API returned it (both last_record and last_page_size), and the consumer sees it without the already-synced tail. Because the filtering happens as read_records yields, partitions read concurrently stay independent by construction. This also gives `is_data_feed` complete semantics on its own: it now stops paginating on the first page containing an already-synced record *and* drops those records, so it no longer has to be paired with `is_client_side_incremental`. The schema documents that, and the factory never installs the record selector filter for a data feed, whether `is_client_side_incremental` is set or not. Streams that set `is_data_feed` alone previously re-emitted the already-synced tail of the last page; they no longer do. FilterAwareStopCondition, the stale-record property on the record filter and the Optional[Record] widening of PaginationStopCondition.is_met are all reverted, leaving CursorStopCondition as the single stop condition. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 4 +- .../declarative/extractors/record_filter.py | 38 +-- .../models/declarative_component_schema.py | 4 +- .../parsers/model_to_component_factory.py | 36 +-- .../paginators/strategies/__init__.py | 4 - .../paginators/strategies/stop_condition.py | 39 +-- .../retrievers/simple_retriever.py | 14 +- .../extractors/test_record_filter.py | 56 ---- .../test_model_to_component_factory.py | 50 ++-- .../paginators/test_stop_condition.py | 39 +-- .../test_stop_condition_integration.py | 168 ------------ .../retrievers/test_data_feed_integration.py | 241 ++++++++++++++++++ .../retrievers/test_simple_retriever.py | 80 ++++++ 13 files changed, 396 insertions(+), 377 deletions(-) delete mode 100644 unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py create mode 100644 unit_tests/sources/declarative/retrievers/test_data_feed_integration.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index b50af83a46..da0116741a 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/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 contains records that were synced during a previous sync, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. 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, 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. This is not needed when Data Feed API is enabled, as a data feed already filters out the records that were synced during a previous sync. type: boolean is_compare_strictly: title: Strict Start-End Time Comparison diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 74a0d90e31..943068f875 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -1,7 +1,6 @@ # # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # -import threading from dataclasses import InitVar, dataclass from typing import Any, Iterable, Mapping, Optional, Union @@ -61,18 +60,6 @@ def __init__( ): super().__init__(**kwargs) self._cursor = cursor - # One retriever (and hence one record filter) may be shared across partitions that are read - # concurrently, so per-page bookkeeping must be tracked per thread - self._thread_local = threading.local() - - @property - def stale_record_seen_on_current_page(self) -> bool: - """ - Whether the page being filtered contained at least one record the cursor considers already - synced. Used by `FilterAwareStopCondition` to stop paginating on data feed streams, since - such records never reach the paginator. Reset on every `filter_records` call. - """ - return getattr(self._thread_local, "stale_record_seen", False) def filter_records( self, @@ -81,8 +68,15 @@ def filter_records( stream_slice: Optional[StreamSlice] = None, next_page_token: Optional[Mapping[str, Any]] = None, ) -> Iterable[Mapping[str, Any]]: - self._thread_local.stale_record_seen = False - records = (record for record in records if self._should_be_synced(record, stream_slice)) + 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="") + ) + ) if self.condition: records = super().filter_records( records=records, @@ -90,16 +84,4 @@ def filter_records( stream_slice=stream_slice, next_page_token=next_page_token, ) - return records - - def _should_be_synced( - self, record: Mapping[str, Any], stream_slice: Optional[StreamSlice] - ) -> bool: - 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="") - ): - return True - self._thread_local.stale_record_seen = True - return False + yield from records diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 1ab2de2eb3..66c228fbd3 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/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 contains records that were synced during a previous sync, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field.", 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, 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. This is not needed when Data Feed API is enabled, as a data feed already filters out the records that were synced during a previous sync.", title="Client-side Incremental Filtering", ) is_compare_strictly: Optional[bool] = Field( diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 4a98c15f3d..9c52b87490 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -525,10 +525,8 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies import ( CursorPaginationStrategy, CursorStopCondition, - FilterAwareStopCondition, OffsetIncrement, PageIncrement, - PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) from airbyte_cdk.sources.declarative.requesters.query_properties import ( @@ -2385,9 +2383,6 @@ def create_default_paginator( extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None, decoder: Optional[Decoder] = None, cursor_used_for_stop_condition: Optional[Cursor] = None, - record_filter_used_for_stop_condition: Optional[ - ClientSideIncrementalRecordFilterDecorator - ] = None, ) -> Union[DefaultPaginator, PaginatorTestReadDecorator]: if decoder: if self._is_supported_decoder_for_pagination(decoder): @@ -2413,16 +2408,8 @@ def create_default_paginator( extractor_model=extractor_model, ) if cursor_used_for_stop_condition: - # When client-side incremental filtering is enabled, records older than the cursor are - # dropped before the paginator can observe them, so the stop condition must be driven - # by the record filter instead of the last record emitted for the page - stop_condition: PaginationStopCondition = ( - FilterAwareStopCondition(record_filter_used_for_stop_condition) - if record_filter_used_for_stop_condition - else CursorStopCondition(cursor_used_for_stop_condition) - ) pagination_strategy = StopConditionPaginationStrategyDecorator( - pagination_strategy, stop_condition + pagination_strategy, CursorStopCondition(cursor_used_for_stop_condition) ) paginator = DefaultPaginator( decoder=decoder_to_use, @@ -3435,6 +3422,16 @@ 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. + data_feed_cursor = cursor if has_stop_condition_cursor else None + client_side_incremental_cursor = ( + cursor if is_client_side_incremental_sync and not data_feed_cursor else None + ) + decoder = ( self._create_component_from_model(model=model.decoder, config=config) if model.decoder @@ -3446,7 +3443,7 @@ 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, file_uploader=file_uploader, ) @@ -3537,14 +3534,6 @@ def _get_url(req: Requester) -> str: extractor_model=model.record_selector.extractor, decoder=decoder, cursor_used_for_stop_condition=cursor if has_stop_condition_cursor else None, - record_filter_used_for_stop_condition=( - record_selector.record_filter - if has_stop_condition_cursor - and isinstance( - record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator - ) - else None - ), ) if model.paginator else NoPagination(parameters={}) @@ -3614,6 +3603,7 @@ def _get_url(req: Requester) -> str: pagination_tracker_factory=self._create_pagination_tracker_factory( model.pagination_reset, cursor ), + data_feed_cursor=data_feed_cursor, parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py index cee364d5fd..c1f9ff1052 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/__init__.py @@ -13,17 +13,13 @@ ) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.stop_condition import ( CursorStopCondition, - FilterAwareStopCondition, - PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) __all__ = [ "CursorPaginationStrategy", "CursorStopCondition", - "FilterAwareStopCondition", "OffsetIncrement", "PageIncrement", - "PaginationStopCondition", "StopConditionPaginationStrategyDecorator", ] diff --git a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py index 6707cc1967..068df72cb8 100644 --- a/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py +++ b/airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.py @@ -3,7 +3,7 @@ # from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any, Optional +from typing import Any, Optional import requests @@ -13,21 +13,14 @@ from airbyte_cdk.sources.streams.concurrent.cursor import Cursor from airbyte_cdk.sources.types import Record -if TYPE_CHECKING: - from airbyte_cdk.sources.declarative.extractors.record_filter import ( - ClientSideIncrementalRecordFilterDecorator, - ) - class PaginationStopCondition(ABC): @abstractmethod - def is_met(self, record: Optional[Record]) -> bool: + def is_met(self, record: Record) -> bool: """ Given a condition is met, the pagination will stop - :param record: the last record yielded for the current page, if any. Records dropped by - record filters are not visible here — a condition that needs to observe them has to get - that signal from the filter itself (see `FilterAwareStopCondition`). + :param record: a record used to evaluate the condition """ raise NotImplementedError() @@ -39,26 +32,8 @@ def __init__( ): self._cursor = cursor - def is_met(self, record: Optional[Record]) -> bool: - return record is not None and not self._cursor.should_be_synced(record) - - -class FilterAwareStopCondition(PaginationStopCondition): - """ - Stop condition for streams combining `is_data_feed` with `is_client_side_incremental`. - - The client-side incremental filter drops records older than the cursor before the paginator can - observe them, so `CursorStopCondition` — which only sees the last record that survived - filtering — would never fire. The filter, however, evaluates `should_be_synced` on every raw - record; this condition stops pagination as soon as the filter reports that the current page - contained a record that was filtered out as already synced. - """ - - def __init__(self, record_filter: "ClientSideIncrementalRecordFilterDecorator"): - self._record_filter = record_filter - - def is_met(self, record: Optional[Record]) -> bool: - return self._record_filter.stale_record_seen_on_current_page + def is_met(self, record: Record) -> bool: + return not self._cursor.should_be_synced(record) class StopConditionPaginationStrategyDecorator(PaginationStrategy): @@ -75,9 +50,7 @@ def next_page_token( ) -> Optional[Any]: # We evaluate in reverse order because the assumption is that most of the APIs using data feed structure # will return records in descending order. In terms of performance/memory, we return the records lazily - # Note: `last_record` may be None even mid-feed when every record of the page was dropped by - # a record filter, so the stop condition is consulted regardless - if self._stop_condition.is_met(last_record): + if last_record and self._stop_condition.is_met(last_record): return None return self._delegate.next_page_token( response, last_page_size, last_record, last_page_token_value diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 1f2eb1c668..0dea0d81cd 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -40,6 +40,7 @@ from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer from airbyte_cdk.sources.source import ExperimentalClassWarning +from airbyte_cdk.sources.streams.concurrent.cursor import Cursor from airbyte_cdk.sources.streams.core import StreamData from airbyte_cdk.sources.streams.http.pagination_reset_exception import ( PaginationResetRequiredException, @@ -72,6 +73,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 + data_feed_cursor (Optional[Cursor]): Set for data feed streams only. Records the cursor considers + already synced are dropped after pagination has observed them """ requester: Requester @@ -95,6 +98,7 @@ class SimpleRetriever(Retriever): pagination_tracker_factory: Callable[[], PaginationTracker] = field( default_factory=lambda: lambda: PaginationTracker() ) + data_feed_cursor: Optional[Cursor] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._paginator = self.paginator or NoPagination(parameters=parameters) @@ -457,7 +461,15 @@ def read_records( stream_slice=stream_slice, records_schema=records_schema, ) - yield from self._read_pages(record_generator, _slice) + # 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 dropped 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. + data_feed_cursor = self.data_feed_cursor + for record in self._read_pages(record_generator, _slice): + if data_feed_cursor is None or data_feed_cursor.should_be_synced(record): + yield record def _parse_records( self, diff --git a/unit_tests/sources/declarative/extractors/test_record_filter.py b/unit_tests/sources/declarative/extractors/test_record_filter.py index 9fe59cb04b..9f0cf46d90 100644 --- a/unit_tests/sources/declarative/extractors/test_record_filter.py +++ b/unit_tests/sources/declarative/extractors/test_record_filter.py @@ -456,59 +456,3 @@ def date_time_based_cursor_factory(stream_state, runtime_lookback_window) -> Con ) assert [x.get("id") for x in filtered_records] == expected_record_ids - - -def _stale_record_tracking_filter(should_be_synced_per_record: List[bool]): - cursor = Mock() - cursor.should_be_synced.side_effect = should_be_synced_per_record - return ClientSideIncrementalRecordFilterDecorator( - config={}, - condition="", - parameters={}, - cursor=cursor, - ) - - -def test_stale_record_seen_when_record_is_filtered_out(): - record_filter_decorator = _stale_record_tracking_filter([True, False]) - - assert not record_filter_decorator.stale_record_seen_on_current_page - filtered_records = list( - record_filter_decorator.filter_records( - records=[{"id": 1}, {"id": 2}], - stream_state={}, - stream_slice=StreamSlice(partition={}, cursor_slice={}), - ) - ) - - assert [record["id"] for record in filtered_records] == [1] - assert record_filter_decorator.stale_record_seen_on_current_page - - -def test_stale_record_seen_is_reset_on_each_filter_records_call(): - record_filter_decorator = _stale_record_tracking_filter([False, True]) - - list(record_filter_decorator.filter_records(records=[{"id": 1}], stream_state={})) - assert record_filter_decorator.stale_record_seen_on_current_page - - list(record_filter_decorator.filter_records(records=[{"id": 2}], stream_state={})) - assert not record_filter_decorator.stale_record_seen_on_current_page - - -def test_stale_record_seen_is_tracked_per_thread(): - import threading - - record_filter_decorator = _stale_record_tracking_filter([False]) - list(record_filter_decorator.filter_records(records=[{"id": 1}], stream_state={})) - assert record_filter_decorator.stale_record_seen_on_current_page - - stale_record_seen_on_other_thread = [] - other_thread = threading.Thread( - target=lambda: stale_record_seen_on_other_thread.append( - record_filter_decorator.stale_record_seen_on_current_page - ) - ) - other_thread.start() - other_thread.join() - - assert stale_record_seen_on_other_thread == [False] diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index a8806c166c..14c890a37b 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -145,8 +145,6 @@ from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator from airbyte_cdk.sources.declarative.requesters.paginators.strategies import ( CursorPaginationStrategy, - CursorStopCondition, - FilterAwareStopCondition, OffsetIncrement, PageIncrement, StopConditionPaginationStrategyDecorator, @@ -1434,13 +1432,27 @@ def test_incremental_data_feed(): model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config ) - pagination_strategy = get_retriever(stream).paginator.pagination_strategy - assert isinstance(pagination_strategy, StopConditionPaginationStrategyDecorator) - assert isinstance(pagination_strategy._stop_condition, CursorStopCondition) + retriever = get_retriever(stream) + assert isinstance( + retriever.paginator.pagination_strategy, + StopConditionPaginationStrategyDecorator, + ) + # the stop condition only prevents the next page from being requested; the already-synced records + # of the last page are dropped by the retriever + assert retriever.data_feed_cursor is stream.cursor -def test_incremental_data_feed_with_client_side_incremental(): - content = """ +@pytest.mark.parametrize( + "is_client_side_incremental", + [ + pytest.param(False, id="test_data_feed_only"), + pytest.param(True, id="test_data_feed_with_client_side_incremental"), + ], +) +def test_incremental_data_feed_filters_already_synced_records_in_the_retriever( + is_client_side_incremental, +): + content = f""" selector: type: RecordSelector extractor: @@ -1448,7 +1460,7 @@ def test_incremental_data_feed_with_client_side_incremental(): field_path: ["extractor_path"] requester: type: HttpRequester - name: "{{ parameters['name'] }}" + name: "{{{{ parameters['name'] }}}}" url_base: "https://api.sendgrid.com/v3/" http_method: "GET" list_stream: @@ -1457,18 +1469,18 @@ def test_incremental_data_feed_with_client_side_incremental(): type: DatetimeBasedCursor $parameters: datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" - start_datetime: "{{ config['start_time'] }}" + start_datetime: "{{{{ config['start_time'] }}}}" cursor_field: "created" is_data_feed: true - is_client_side_incremental: true + is_client_side_incremental: {str(is_client_side_incremental).lower()} retriever: type: SimpleRetriever - name: "{{ parameters['name'] }}" + name: "{{{{ parameters['name'] }}}}" paginator: type: DefaultPaginator pagination_strategy: type: "CursorPagination" - cursor_value: "{{ response._metadata.next }}" + cursor_value: "{{{{ response._metadata.next }}}}" page_size: 10 requester: $ref: "#/requester" @@ -1490,18 +1502,12 @@ def test_incremental_data_feed_with_client_side_incremental(): ) retriever = get_retriever(stream) - pagination_strategy = retriever.paginator.pagination_strategy - assert isinstance(pagination_strategy, StopConditionPaginationStrategyDecorator) - # the client-side filter drops below-cursor records before the paginator sees them, so the stop - # condition must observe the filter rather than the last emitted record - assert isinstance(pagination_strategy._stop_condition, FilterAwareStopCondition) - assert isinstance( + assert retriever.data_feed_cursor is stream.cursor + # filtering in the record selector would hide the already-synced records from the paginator and + # therefore silently disable the stop condition + assert not isinstance( retriever.record_selector.record_filter, ClientSideIncrementalRecordFilterDecorator ) - assert ( - pagination_strategy._stop_condition._record_filter - is retriever.record_selector.record_filter - ) def test_given_data_feed_and_incremental_then_raise_error(): diff --git a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py index c48b2a7771..b89baf4430 100644 --- a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py +++ b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition.py @@ -9,12 +9,8 @@ from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import ( PaginationStrategy, ) -from airbyte_cdk.sources.declarative.extractors.record_filter import ( - ClientSideIncrementalRecordFilterDecorator, -) from airbyte_cdk.sources.declarative.requesters.paginators.strategies.stop_condition import ( CursorStopCondition, - FilterAwareStopCondition, PaginationStopCondition, StopConditionPaginationStrategyDecorator, ) @@ -51,23 +47,6 @@ def test_given_record_should_not_be_synced_when_is_met_return_true(mocked_cursor assert CursorStopCondition(mocked_cursor).is_met(ANY_RECORD) -def test_given_no_record_when_is_met_return_false(mocked_cursor): - assert not CursorStopCondition(mocked_cursor).is_met(NO_RECORD) - mocked_cursor.should_be_synced.assert_not_called() - - -def test_given_stale_record_seen_by_filter_when_is_met_return_true(): - record_filter = Mock(spec=ClientSideIncrementalRecordFilterDecorator) - record_filter.stale_record_seen_on_current_page = True - assert FilterAwareStopCondition(record_filter).is_met(NO_RECORD) - - -def test_given_no_stale_record_seen_by_filter_when_is_met_return_false(): - record_filter = Mock(spec=ClientSideIncrementalRecordFilterDecorator) - record_filter.stale_record_seen_on_current_page = False - assert not FilterAwareStopCondition(record_filter).is_met(ANY_RECORD) - - def test_given_stop_condition_is_met_when_next_page_token_then_return_none( mocked_pagination_strategy, mocked_stop_condition ): @@ -113,10 +92,9 @@ def test_given_stop_condition_is_not_met_when_next_page_token_then_delegate( mocked_stop_condition.is_met.assert_has_calls([call(last_record)]) -def test_given_no_records_and_stop_condition_is_not_met_when_next_page_token_then_delegate( +def test_given_no_records_when_next_page_token_then_delegate( mocked_pagination_strategy, mocked_stop_condition ): - mocked_stop_condition.is_met.return_value = False decorator = StopConditionPaginationStrategyDecorator( mocked_pagination_strategy, mocked_stop_condition ) @@ -127,21 +105,6 @@ def test_given_no_records_and_stop_condition_is_not_met_when_next_page_token_the mocked_pagination_strategy.next_page_token.assert_called_once_with( ANY_RESPONSE, 0, NO_RECORD, None ) - mocked_stop_condition.is_met.assert_called_once_with(NO_RECORD) - - -def test_given_no_records_and_stop_condition_is_met_when_next_page_token_then_return_none( - mocked_pagination_strategy, mocked_stop_condition -): - # A page can yield no records even mid-feed when a record filter dropped all of them, so the - # stop condition must be consulted even without a last record - mocked_stop_condition.is_met.return_value = True - decorator = StopConditionPaginationStrategyDecorator( - mocked_pagination_strategy, mocked_stop_condition - ) - - assert not decorator.next_page_token(ANY_RESPONSE, 0, NO_RECORD) - mocked_pagination_strategy.next_page_token.assert_not_called() def test_when_get_page_size_then_delegate(mocked_pagination_strategy, mocked_stop_condition): diff --git a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py b/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py deleted file mode 100644 index 534a40e44f..0000000000 --- a/unit_tests/sources/declarative/requesters/paginators/test_stop_condition_integration.py +++ /dev/null @@ -1,168 +0,0 @@ -# -# Copyright (c) 2026 Airbyte, Inc., all rights reserved. -# - -""" -End-to-end coverage for the pagination stop condition of data feed streams that also use -client-side incremental filtering. The client-side filter drops records older than the cursor -before the paginator can observe them, so the stop condition is driven by the filter itself: -pagination must stop on the first page containing a record older than the cursor, while still -emitting only the records newer than the cursor. -""" - -import json -import logging -from typing import Any, List, Mapping, Optional - -import requests_mock - -from airbyte_cdk.models import ( - AirbyteStateBlob, - AirbyteStateMessage, - AirbyteStateType, - AirbyteStreamState, - ConfiguredAirbyteCatalogSerializer, - StreamDescriptor, - Type, -) -from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - ConcurrentDeclarativeSource, -) - -_MANIFEST = { - "version": "7.0.0", - "type": "DeclarativeSource", - "check": {"type": "CheckStream", "stream_names": ["items"]}, - "spec": { - "type": "Spec", - "connection_specification": {"type": "object", "properties": {}}, - }, - "streams": [ - { - "type": "DeclarativeStream", - "name": "items", - "primary_key": ["id"], - "schema_loader": { - "type": "InlineSchemaLoader", - "schema": { - "type": "object", - "properties": { - "id": {"type": "integer"}, - "updated_at": {"type": "string"}, - }, - }, - }, - "retriever": { - "type": "SimpleRetriever", - "requester": { - "type": "HttpRequester", - "url_base": "https://api.example.com", - "path": "/items", - "http_method": "GET", - }, - "record_selector": { - "type": "RecordSelector", - "extractor": {"type": "DpathExtractor", "field_path": []}, - }, - "paginator": { - "type": "DefaultPaginator", - "pagination_strategy": { - "type": "PageIncrement", - "page_size": 4, - "start_from_page": 1, - "inject_on_first_request": True, - }, - "page_token_option": { - "type": "RequestOption", - "inject_into": "request_parameter", - "field_name": "page", - }, - }, - }, - "incremental_sync": { - "type": "DatetimeBasedCursor", - "cursor_field": "updated_at", - "datetime_format": "%Y-%m-%dT%H:%M:%SZ", - "start_datetime": { - "type": "MinMaxDatetime", - "datetime": "2019-01-01T00:00:00Z", - "datetime_format": "%Y-%m-%dT%H:%M:%SZ", - }, - "is_data_feed": True, - "is_client_side_incremental": True, - }, - } - ], -} - -# Records are sorted in descending order of updated_at, as expected from a data feed -_PAGE_1 = [ - {"id": 4, "updated_at": "2022-06-01T00:00:00Z"}, - {"id": 3, "updated_at": "2022-05-01T00:00:00Z"}, - {"id": 2, "updated_at": "2020-06-01T00:00:00Z"}, - {"id": 1, "updated_at": "2020-05-01T00:00:00Z"}, -] -_PAGE_2 = [ - {"id": 0, "updated_at": "2020-04-01T00:00:00Z"}, -] - -_CATALOG = ConfiguredAirbyteCatalogSerializer.load( - { - "streams": [ - { - "stream": { - "name": "items", - "json_schema": {}, - "supported_sync_modes": ["full_refresh", "incremental"], - }, - "sync_mode": "incremental", - "destination_sync_mode": "append", - } - ] - } -) - - -def _read(state: Optional[List[AirbyteStateMessage]]) -> tuple[List[str], List[Mapping[str, Any]]]: - pages_fetched = [] - - def paged_response(request: Any, context: Any) -> str: - page = request.qs.get("page", ["1"])[0] - pages_fetched.append(page) - return json.dumps(_PAGE_1 if page == "1" else _PAGE_2) - - source = ConcurrentDeclarativeSource( - source_config=_MANIFEST, config={}, catalog=_CATALOG, state=state - ) - with requests_mock.Mocker() as http_mocker: - http_mocker.get("https://api.example.com/items", text=paged_response) - records = [ - message.record.data - for message in source.read(logging.getLogger("test"), {}, _CATALOG, state) - if message.type == Type.RECORD - ] - return pages_fetched, records - - -def test_given_stale_records_on_page_when_client_side_incremental_then_stop_pagination(): - state = [ - AirbyteStateMessage( - type=AirbyteStateType.STREAM, - stream=AirbyteStreamState( - stream_descriptor=StreamDescriptor(name="items"), - stream_state=AirbyteStateBlob({"updated_at": "2021-01-01T00:00:00Z"}), - ), - ) - ] - - pages_fetched, records = _read(state) - - assert sorted(record["id"] for record in records) == [3, 4] - assert pages_fetched == ["1"] - - -def test_given_no_stale_records_when_client_side_incremental_then_paginate_until_the_end(): - pages_fetched, records = _read(None) - - assert sorted(record["id"] for record in records) == [0, 1, 2, 3, 4] - assert pages_fetched == ["1", "2"] diff --git a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py new file mode 100644 index 0000000000..54f544b9e5 --- /dev/null +++ b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py @@ -0,0 +1,241 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +""" +End-to-end coverage for data feed streams (`is_data_feed: true`). + +A data feed returns records in descending cursor order, so the sync must stop paginating on the +first page that contains a record the cursor considers already synced, and must not emit the +already-synced records sitting at the tail of that page. +""" + +import json +import logging +from typing import Any, List, Mapping, Optional, Tuple + +import pytest +import requests_mock + +from airbyte_cdk.models import ( + AirbyteStateBlob, + AirbyteStateMessage, + AirbyteStateType, + AirbyteStreamState, + ConfiguredAirbyteCatalogSerializer, + StreamDescriptor, + Type, +) +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + ConcurrentDeclarativeSource, +) + +_STREAM_NAME = "items" + + +def _manifest( + is_client_side_incremental: bool = False, partition_router: Optional[Mapping[str, Any]] = None +) -> Mapping[str, Any]: + incremental_sync: dict[str, Any] = { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + "start_datetime": { + "type": "MinMaxDatetime", + "datetime": "2019-01-01T00:00:00Z", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + }, + "is_data_feed": True, + } + if is_client_side_incremental: + incremental_sync["is_client_side_incremental"] = True + + retriever: dict[str, Any] = { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.example.com", + "path": "/items" if partition_router is None else "/{{ stream_partition.owner }}/items", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": { + "type": "DefaultPaginator", + "pagination_strategy": { + "type": "PageIncrement", + "page_size": 4, + "start_from_page": 1, + "inject_on_first_request": True, + }, + "page_token_option": { + "type": "RequestOption", + "inject_into": "request_parameter", + "field_name": "page", + }, + }, + } + if partition_router: + retriever["partition_router"] = partition_router + + return { + "version": "7.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": [_STREAM_NAME]}, + "spec": { + "type": "Spec", + "connection_specification": {"type": "object", "properties": {}}, + }, + "streams": [ + { + "type": "DeclarativeStream", + "name": _STREAM_NAME, + "primary_key": ["id"], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "updated_at": {"type": "string"}, + }, + }, + }, + "retriever": retriever, + "incremental_sync": incremental_sync, + } + ], + } + + +# Records are sorted in descending order of updated_at, as expected from a data feed. With a cursor +# of 2021-01-01, page 1 holds two fresh records followed by two already-synced ones. +_PAGE_1 = [ + {"id": "4", "updated_at": "2022-06-01T00:00:00Z"}, + {"id": "3", "updated_at": "2022-05-01T00:00:00Z"}, + {"id": "2", "updated_at": "2020-06-01T00:00:00Z"}, + {"id": "1", "updated_at": "2020-05-01T00:00:00Z"}, +] +_PAGE_2 = [ + {"id": "0", "updated_at": "2020-04-01T00:00:00Z"}, +] + +_CATALOG = ConfiguredAirbyteCatalogSerializer.load( + { + "streams": [ + { + "stream": { + "name": _STREAM_NAME, + "json_schema": {}, + "supported_sync_modes": ["full_refresh", "incremental"], + }, + "sync_mode": "incremental", + "destination_sync_mode": "append", + } + ] + } +) + + +def _state(stream_state: Mapping[str, Any]) -> List[AirbyteStateMessage]: + return [ + AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name=_STREAM_NAME), + stream_state=AirbyteStateBlob(stream_state), + ), + ) + ] + + +def _read( + manifest: Mapping[str, Any], + state: Optional[List[AirbyteStateMessage]], + pages_per_partition: Optional[Mapping[Tuple[str, str], List[Mapping[str, Any]]]] = None, +) -> Tuple[List[str], List[str]]: + pages_fetched = [] + + def paged_response(request: Any, context: Any) -> str: + page = request.qs.get("page", ["1"])[0] + if pages_per_partition is None: + pages_fetched.append(page) + return json.dumps(_PAGE_1 if page == "1" else _PAGE_2) + owner = request.path.strip("/").split("/")[0] + pages_fetched.append(f"{owner}:{page}") + return json.dumps(pages_per_partition.get((owner, page), [])) + + source = ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=_CATALOG, state=state + ) + with requests_mock.Mocker() as http_mocker: + http_mocker.get(requests_mock.ANY, text=paged_response) + records = [ + message.record.data + for message in source.read(logging.getLogger("test"), {}, _CATALOG, state) + if message.type == Type.RECORD + ] + return sorted(pages_fetched), sorted(record["id"] for record in records) + + +@pytest.mark.parametrize("is_client_side_incremental", [False, True]) +def test_given_already_synced_records_on_page_then_stop_paginating_and_filter_them_out( + is_client_side_incremental: bool, +) -> None: + pages_fetched, record_ids = _read( + _manifest(is_client_side_incremental=is_client_side_incremental), + _state({"updated_at": "2021-01-01T00:00:00Z"}), + ) + + assert pages_fetched == ["1"] + assert record_ids == ["3", "4"] + + +@pytest.mark.parametrize("is_client_side_incremental", [False, True]) +def test_given_no_already_synced_records_then_paginate_until_the_end( + is_client_side_incremental: bool, +) -> None: + pages_fetched, record_ids = _read( + _manifest(is_client_side_incremental=is_client_side_incremental), None + ) + + assert pages_fetched == ["1", "2"] + assert record_ids == ["0", "1", "2", "3", "4"] + + +def test_given_multiple_partitions_then_each_partition_stops_on_its_own_cursor() -> None: + """ + A single retriever instance is shared by every partition and partitions are read concurrently, + so the boundary of one partition must not influence another. + """ + pages_per_partition = { + ("a", "1"): [{**record, "id": f"a{record['id']}"} for record in _PAGE_1], + ("a", "2"): [{**record, "id": f"a{record['id']}"} for record in _PAGE_2], + ("b", "1"): [{**record, "id": f"b{record['id']}"} for record in _PAGE_1], + ("b", "2"): [{**record, "id": f"b{record['id']}"} for record in _PAGE_2], + } + manifest = _manifest( + partition_router={ + "type": "ListPartitionRouter", + "values": ["a", "b"], + "cursor_field": "owner", + } + ) + state = _state( + { + "use_global_cursor": False, + "states": [ + # partition "a" has already synced everything before 2021 hence it stops on page 1 + {"partition": {"owner": "a"}, "cursor": {"updated_at": "2021-01-01T00:00:00Z"}}, + # partition "b" has nothing already synced hence it reads both pages + {"partition": {"owner": "b"}, "cursor": {"updated_at": "2019-01-01T00:00:00Z"}}, + ], + } + ) + + pages_fetched, record_ids = _read(manifest, state, pages_per_partition) + + assert pages_fetched == ["a:1", "b:1", "b:2"] + assert record_ids == ["a3", "a4", "b0", "b1", "b2", "b3", "b4"] diff --git a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py index d9585dbd44..67fcfdd7b8 100644 --- a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py +++ b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py @@ -1430,3 +1430,83 @@ def _mock_paginator(): paginator.get_request_body_data.__name__ = "get_request_body_data" paginator.get_request_body_json.__name__ = "get_request_body_json" return paginator + + +def _data_feed_retriever(cursor: Mock, paginator: Paginator) -> SimpleRetriever: + requester = MagicMock() + requester.send_request.return_value = MagicMock() + record_selector = MagicMock() + return SimpleRetriever( + name=A_STREAM_NAME, + primary_key=primary_key, + requester=requester, + paginator=paginator, + record_selector=record_selector, + stream_slicer=SinglePartitionRouter(parameters={}), + data_feed_cursor=cursor, + parameters={}, + config={}, + ) + + +def test_given_data_feed_cursor_when_read_records_then_filter_out_already_synced_records(): + page = [ + Record(data={"id": "1"}, stream_name=A_STREAM_NAME), + Record(data={"id": "2"}, stream_name=A_STREAM_NAME), + Record(data={"id": "3"}, stream_name=A_STREAM_NAME), + ] + cursor = Mock() + cursor.should_be_synced.side_effect = lambda record: record.data["id"] != "3" + paginator = _mock_paginator() + paginator.get_initial_token.return_value = None + paginator.next_page_token.return_value = None + retriever = _data_feed_retriever(cursor, paginator) + + with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)): + actual_records = list( + retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE) + ) + + assert actual_records == page[:2] + + +def test_given_data_feed_cursor_when_read_records_then_paginator_still_sees_the_whole_page(): + """ + The record that stops the pagination is the very one being filtered out, so the paginator must + be given the page as returned by the API rather than the filtered one. + """ + page = [ + Record(data={"id": "1"}, stream_name=A_STREAM_NAME), + Record(data={"id": "2"}, stream_name=A_STREAM_NAME), + Record(data={"id": "3"}, stream_name=A_STREAM_NAME), + ] + cursor = Mock() + cursor.should_be_synced.side_effect = lambda record: record.data["id"] != "3" + paginator = _mock_paginator() + paginator.get_initial_token.return_value = None + paginator.next_page_token.return_value = None + retriever = _data_feed_retriever(cursor, paginator) + + with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)): + list(retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE)) + + assert paginator.next_page_token.call_args.kwargs["last_page_size"] == 3 + assert paginator.next_page_token.call_args.kwargs["last_record"] == page[-1] + + +def test_given_no_data_feed_cursor_when_read_records_then_emit_every_record(): + page = [ + Record(data={"id": "1"}, stream_name=A_STREAM_NAME), + Record(data={"id": "2"}, stream_name=A_STREAM_NAME), + ] + paginator = _mock_paginator() + paginator.get_initial_token.return_value = None + paginator.next_page_token.return_value = None + retriever = _data_feed_retriever(cursor=None, paginator=paginator) + + with patch.object(SimpleRetriever, "_parse_records", return_value=iter(page)): + actual_records = list( + retriever.read_records(records_schema={}, stream_slice=A_STREAM_SLICE) + ) + + assert actual_records == page From c4358f25df69ead13aec52cc1a05da2c2d86cb4e Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Mon, 10 Aug 2026 19:03:04 +0300 Subject: [PATCH 3/6] refactor(low-code): reuse the record filter for the data feed boundary page The retriever held a `Cursor` and called `should_be_synced` itself, duplicating the rule that `ClientSideIncrementalRecordFilterDecorator` already owns. Give that decorator a `Record`-typed entry point, route its mapping-based path through it, and hand the retriever the filter instead of the cursor. The filtering still happens after `_read_pages` so the paginator keeps seeing whole pages, but the retriever no longer carries any cursor semantics. The post-pagination filter is built without `condition`: the `record_filter` condition stays in the record selector so the records it rejects keep counting towards the page size and can still be the record the stop condition reads. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative/extractors/record_filter.py | 23 +++++-- .../parsers/model_to_component_factory.py | 26 ++++--- .../retrievers/simple_retriever.py | 28 ++++---- .../test_model_to_component_factory.py | 67 ++++++++++++++++++- .../retrievers/test_simple_retriever.py | 17 +++-- 5 files changed, 126 insertions(+), 35 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 943068f875..92f04ade55 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -61,6 +61,16 @@ def __init__( super().__init__(**kwargs) self._cursor = cursor + def filter_typed_records(self, records: Iterable[Record]) -> Iterable[Record]: + """ + Drop the records the cursor considers already synced. + + Unlike `filter_records`, this operates on `Record` objects and does not evaluate `condition`. It exists for + callers that already hold records, such as a retriever filtering a data feed's boundary page once the paginator + has observed it. + """ + return (record for record in records if self._cursor.should_be_synced(record)) + def filter_records( self, records: Iterable[Mapping[str, Any]], @@ -68,15 +78,14 @@ def filter_records( stream_slice: Optional[StreamSlice] = None, next_page_token: Optional[Mapping[str, Any]] = None, ) -> Iterable[Mapping[str, Any]]: - records = ( - record + filtered_records = self.filter_typed_records( + # Records are created on the fly to align with the cursor interface; the stream name is empty because it is + # not used during the filtering + Record(data=record, associated_slice=stream_slice, stream_name="") 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="") - ) ) + # The records yielded downstream must be the ones that were passed in, not the wrappers built above + records = (record.data for record in filtered_records) if self.condition: records = super().filter_records( records=records, diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 9c52b87490..72cfc1c693 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3422,14 +3422,24 @@ 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. - data_feed_cursor = cursor if has_stop_condition_cursor else None + # 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: it stays in the record selector so that the records it rejects keep being counted + # by the paginator. + 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 data_feed_cursor else None + cursor if is_client_side_incremental_sync and not post_pagination_filter else None ) decoder = ( @@ -3603,7 +3613,7 @@ def _get_url(req: Requester) -> str: pagination_tracker_factory=self._create_pagination_tracker_factory( model.pagination_reset, cursor ), - data_feed_cursor=data_feed_cursor, + post_pagination_filter=post_pagination_filter, parameters=model.parameters or {}, ) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 0dea0d81cd..5ab0dcdb6b 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/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, @@ -40,7 +43,6 @@ from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer from airbyte_cdk.sources.source import ExperimentalClassWarning -from airbyte_cdk.sources.streams.concurrent.cursor import Cursor from airbyte_cdk.sources.streams.core import StreamData from airbyte_cdk.sources.streams.http.pagination_reset_exception import ( PaginationResetRequiredException, @@ -73,8 +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 - data_feed_cursor (Optional[Cursor]): Set for data feed streams only. Records the cursor considers - already synced are dropped after pagination has observed them + 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 @@ -98,7 +100,7 @@ class SimpleRetriever(Retriever): pagination_tracker_factory: Callable[[], PaginationTracker] = field( default_factory=lambda: lambda: PaginationTracker() ) - data_feed_cursor: Optional[Cursor] = None + post_pagination_filter: Optional[ClientSideIncrementalRecordFilterDecorator] = None def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._paginator = self.paginator or NoPagination(parameters=parameters) @@ -461,15 +463,15 @@ def read_records( stream_slice=stream_slice, records_schema=records_schema, ) - # 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 dropped 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. - data_feed_cursor = self.data_feed_cursor - for record in self._read_pages(record_generator, _slice): - if data_feed_cursor is None or data_feed_cursor.should_be_synced(record): - yield record + records = 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. Note that the pagination tracker, and + # hence the cursor, still observes the dropped records. + records = self.post_pagination_filter.filter_typed_records(records) + yield from records def _parse_records( self, diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 14c890a37b..4779d17c21 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -1439,7 +1439,8 @@ def test_incremental_data_feed(): ) # the stop condition only prevents the next page from being requested; the already-synced records # of the last page are dropped by the retriever - assert retriever.data_feed_cursor is stream.cursor + assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator) + assert retriever.post_pagination_filter._cursor is stream.cursor @pytest.mark.parametrize( @@ -1502,7 +1503,10 @@ def test_incremental_data_feed_filters_already_synced_records_in_the_retriever( ) retriever = get_retriever(stream) - assert retriever.data_feed_cursor is stream.cursor + assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator) + assert retriever.post_pagination_filter._cursor is stream.cursor + # the `record_filter` condition stays in the record selector, so the post-pagination filter must not evaluate it + assert retriever.post_pagination_filter.condition is None # filtering in the record selector would hide the already-synced records from the paginator and # therefore silently disable the stop condition assert not isinstance( @@ -1510,6 +1514,65 @@ def test_incremental_data_feed_filters_already_synced_records_in_the_retriever( ) +def test_given_data_feed_and_record_filter_then_condition_stays_in_the_record_selector(): + content = """ +selector: + type: RecordSelector + record_filter: + type: RecordFilter + condition: "{{ record['id'] > 1 }}" + extractor: + type: DpathExtractor + field_path: ["extractor_path"] +requester: + type: HttpRequester + name: "{{ parameters['name'] }}" + url_base: "https://api.sendgrid.com/v3/" + http_method: "GET" +list_stream: + type: DeclarativeStream + incremental_sync: + type: DatetimeBasedCursor + $parameters: + datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" + start_datetime: "{{ config['start_time'] }}" + cursor_field: "created" + is_data_feed: true + retriever: + type: SimpleRetriever + name: "{{ parameters['name'] }}" + paginator: + type: DefaultPaginator + pagination_strategy: + type: "CursorPagination" + cursor_value: "{{ response._metadata.next }}" + page_size: 10 + requester: + $ref: "#/requester" + path: "/" + record_selector: + $ref: "#/selector" + $parameters: + name: "lists" + """ + + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + stream_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["list_stream"], {} + ) + + stream = factory.create_component( + model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config + ) + + retriever = get_retriever(stream) + # the condition must keep running upstream of the paginator, otherwise the records it rejects would start counting + # towards the page size and could become the record the stop condition is evaluated on + assert retriever.record_selector.record_filter.condition == "{{ record['id'] > 1 }}" + assert retriever.post_pagination_filter.condition is None + + def test_given_data_feed_and_incremental_then_raise_error(): content = """ incremental_sync: diff --git a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py index 67fcfdd7b8..e05c427acb 100644 --- a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py +++ b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py @@ -20,6 +20,9 @@ from airbyte_cdk.sources.declarative.auth.declarative_authenticator import NoAuth from airbyte_cdk.sources.declarative.decoders import JsonDecoder from airbyte_cdk.sources.declarative.extractors import DpathExtractor, HttpSelector, RecordSelector +from airbyte_cdk.sources.declarative.extractors.record_filter import ( + ClientSideIncrementalRecordFilterDecorator, +) from airbyte_cdk.sources.declarative.partition_routers import SinglePartitionRouter from airbyte_cdk.sources.declarative.requesters.paginators import DefaultPaginator, Paginator from airbyte_cdk.sources.declarative.requesters.paginators.strategies import ( @@ -1432,7 +1435,7 @@ def _mock_paginator(): return paginator -def _data_feed_retriever(cursor: Mock, paginator: Paginator) -> SimpleRetriever: +def _data_feed_retriever(cursor: Optional[Mock], paginator: Paginator) -> SimpleRetriever: requester = MagicMock() requester.send_request.return_value = MagicMock() record_selector = MagicMock() @@ -1443,13 +1446,17 @@ def _data_feed_retriever(cursor: Mock, paginator: Paginator) -> SimpleRetriever: paginator=paginator, record_selector=record_selector, stream_slicer=SinglePartitionRouter(parameters={}), - data_feed_cursor=cursor, + post_pagination_filter=ClientSideIncrementalRecordFilterDecorator( + config={}, parameters={}, condition=None, cursor=cursor + ) + if cursor + else None, parameters={}, config={}, ) -def test_given_data_feed_cursor_when_read_records_then_filter_out_already_synced_records(): +def test_given_data_feed_when_read_records_then_filter_out_already_synced_records(): page = [ Record(data={"id": "1"}, stream_name=A_STREAM_NAME), Record(data={"id": "2"}, stream_name=A_STREAM_NAME), @@ -1470,7 +1477,7 @@ def test_given_data_feed_cursor_when_read_records_then_filter_out_already_synced assert actual_records == page[:2] -def test_given_data_feed_cursor_when_read_records_then_paginator_still_sees_the_whole_page(): +def test_given_data_feed_when_read_records_then_paginator_still_sees_the_whole_page(): """ The record that stops the pagination is the very one being filtered out, so the paginator must be given the page as returned by the API rather than the filtered one. @@ -1494,7 +1501,7 @@ def test_given_data_feed_cursor_when_read_records_then_paginator_still_sees_the_ assert paginator.next_page_token.call_args.kwargs["last_record"] == page[-1] -def test_given_no_data_feed_cursor_when_read_records_then_emit_every_record(): +def test_given_no_data_feed_when_read_records_then_emit_every_record(): page = [ Record(data={"id": "1"}, stream_name=A_STREAM_NAME), Record(data={"id": "2"}, stream_name=A_STREAM_NAME), From 3e8c4878d5b8dd93527e33061bf7559ffa6a43a9 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Mon, 10 Aug 2026 19:19:39 +0300 Subject: [PATCH 4/6] refactor(low-code): call the existing record filter from the retriever `filter_typed_records` was not needed: `Record` is a `Mapping`, so the data feed filtering can go through `filter_records` as it stands. Revert `record_filter.py` to its state on main and keep the change to the two files that carry the fix. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative/extractors/record_filter.py | 23 ++++++------------- .../retrievers/simple_retriever.py | 9 ++++++-- 2 files changed, 14 insertions(+), 18 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 92f04ade55..943068f875 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -61,16 +61,6 @@ def __init__( super().__init__(**kwargs) self._cursor = cursor - def filter_typed_records(self, records: Iterable[Record]) -> Iterable[Record]: - """ - Drop the records the cursor considers already synced. - - Unlike `filter_records`, this operates on `Record` objects and does not evaluate `condition`. It exists for - callers that already hold records, such as a retriever filtering a data feed's boundary page once the paginator - has observed it. - """ - return (record for record in records if self._cursor.should_be_synced(record)) - def filter_records( self, records: Iterable[Mapping[str, Any]], @@ -78,14 +68,15 @@ def filter_records( stream_slice: Optional[StreamSlice] = None, next_page_token: Optional[Mapping[str, Any]] = None, ) -> Iterable[Mapping[str, Any]]: - filtered_records = self.filter_typed_records( - # Records are created on the fly to align with the cursor interface; the stream name is empty because it is - # not used during the filtering - Record(data=record, associated_slice=stream_slice, stream_name="") + 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="") + ) ) - # The records yielded downstream must be the ones that were passed in, not the wrappers built above - records = (record.data for record in filtered_records) if self.condition: records = super().filter_records( records=records, diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 5ab0dcdb6b..d9e79c229b 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -463,14 +463,19 @@ def read_records( stream_slice=stream_slice, records_schema=records_schema, ) - records = 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. Note that the pagination tracker, and # hence the cursor, still observes the dropped records. - records = self.post_pagination_filter.filter_typed_records(records) + 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( From ef4ffc134484dda6a0bce5c939ca74d6a8ddf130 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Mon, 10 Aug 2026 20:13:20 +0300 Subject: [PATCH 5/6] fix(low-code): keep transform_before_filtering for client-side incremental data feeds Routing the cursor away from `create_record_selector` also dropped the `transform_before_filtering=True` default that branch carried, so a `record_filter.condition` reading a transformation-produced field started filtering untransformed records and rejected everything. The default now follows `is_client_side_incremental` itself rather than the component that performs the cursor comparison, which leaves data-feed-only streams unchanged. Also pass the post-pagination filter to `LazySimpleRetriever`, which inherits `read_records` but was constructed without it, and warn when both flags are set so the ignored `is_client_side_incremental` shows up in sync logs. Records dated ahead of `now()` are dropped along with the already-synced ones, because `should_be_synced` is bounded on both ends. That matches what `is_client_side_incremental` has always done; a test pins it. Co-Authored-By: Claude Opus 5 (1M context) --- .../parsers/model_to_component_factory.py | 23 ++- .../retrievers/simple_retriever.py | 5 +- .../test_model_to_component_factory.py | 164 +++++++++++++++++- .../retrievers/test_data_feed_integration.py | 26 +++ 4 files changed, 209 insertions(+), 9 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 72cfc1c693..a80c51d2a0 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3287,6 +3287,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: @@ -3299,8 +3300,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( @@ -3311,11 +3320,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 @@ -3441,6 +3445,11 @@ def _get_url(req: Requester) -> str: 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` is ignored when `is_data_feed` is set, " + "as a data feed already filters out the records that were synced during a previous sync." + ) decoder = ( self._create_component_from_model(model=model.decoder, config=config) @@ -3454,6 +3463,7 @@ def _get_url(req: Requester) -> str: decoder=decoder, transformations=transformations, client_side_incremental_sync_cursor=client_side_incremental_cursor, + is_client_side_incremental_sync=is_client_side_incremental_sync, file_uploader=file_uploader, ) @@ -3588,6 +3598,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 {}, ) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index d9e79c229b..6f82b8ebd9 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -468,8 +468,9 @@ def read_records( # 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. Note that the pagination tracker, and - # hence the cursor, still observes the dropped records. + # 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 diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 4779d17c21..cb7a2faf42 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -172,7 +172,11 @@ ) from airbyte_cdk.sources.declarative.requesters.request_path import RequestPath from airbyte_cdk.sources.declarative.requesters.requester import HttpMethod -from airbyte_cdk.sources.declarative.retrievers import AsyncRetriever, SimpleRetriever +from airbyte_cdk.sources.declarative.retrievers import ( + AsyncRetriever, + LazySimpleRetriever, + SimpleRetriever, +) from airbyte_cdk.sources.declarative.schema import InlineSchemaLoader, JsonFileSchemaLoader from airbyte_cdk.sources.declarative.schema.caching_schema_loader_decorator import ( CachingSchemaLoaderDecorator, @@ -1571,6 +1575,164 @@ def test_given_data_feed_and_record_filter_then_condition_stays_in_the_record_se # towards the page size and could become the record the stop condition is evaluated on assert retriever.record_selector.record_filter.condition == "{{ record['id'] > 1 }}" assert retriever.post_pagination_filter.condition is None + # a data feed that is not client-side incremental keeps the `transform_before_filtering` default it had before the + # cursor filtering moved to the retriever + assert retriever.record_selector.transform_before_filtering is False + + +def test_given_data_feed_and_client_side_incremental_then_transform_before_filtering(): + """ + Moving the cursor filtering to the retriever must not change when the `record_filter` condition runs: a + client-side incremental stream evaluates it after the transformations, otherwise a condition reading a + transformation-produced field silently rejects every record. + """ + content = """ +selector: + type: RecordSelector + record_filter: + type: RecordFilter + condition: "{{ record['keep'] == 'yes' }}" + extractor: + type: DpathExtractor + field_path: ["extractor_path"] +requester: + type: HttpRequester + name: "{{ parameters['name'] }}" + url_base: "https://api.sendgrid.com/v3/" + http_method: "GET" +list_stream: + type: DeclarativeStream + transformations: + - type: AddFields + fields: + - path: ["keep"] + value: "yes" + incremental_sync: + type: DatetimeBasedCursor + $parameters: + datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" + start_datetime: "{{ config['start_time'] }}" + cursor_field: "created" + is_data_feed: true + is_client_side_incremental: true + retriever: + type: SimpleRetriever + name: "{{ parameters['name'] }}" + paginator: + type: DefaultPaginator + pagination_strategy: + type: "CursorPagination" + cursor_value: "{{ response._metadata.next }}" + page_size: 10 + requester: + $ref: "#/requester" + path: "/" + record_selector: + $ref: "#/selector" + $parameters: + name: "lists" + """ + + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + stream_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["list_stream"], {} + ) + + stream = factory.create_component( + model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=input_config + ) + + retriever = get_retriever(stream) + assert retriever.record_selector.transform_before_filtering is True + + +def test_given_data_feed_and_lazy_read_then_lazy_retriever_filters_already_synced_records(): + """ + `LazySimpleRetriever` inherits `read_records`, so it only drops the already-synced records of the boundary page + if the factory passes it the filter too. + """ + stream_definition = { + "type": "DeclarativeStream", + "name": "items", + "primary_key": [], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "incremental_sync": { + "type": "DatetimeBasedCursor", + "datetime_format": "%Y-%m-%dT%H:%M:%S.%f%z", + "start_datetime": "{{ config['start_time'] }}", + "cursor_field": "created", + "is_data_feed": True, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": "parent/{{ stream_partition.parent_id }}/items", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": ["data"]}, + }, + "paginator": { + "type": "DefaultPaginator", + "pagination_strategy": { + "type": "CursorPagination", + "cursor_value": '{{ response["data"][-1]["id"] }}', + }, + }, + "partition_router": { + "type": "SubstreamPartitionRouter", + "parent_stream_configs": [ + { + "type": "ParentStreamConfig", + "parent_key": "id", + "partition_field": "parent_id", + "lazy_read_pointer": ["items"], + "stream": { + "type": "DeclarativeStream", + "name": "parent", + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": "/parents", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": { + "type": "DpathExtractor", + "field_path": ["data"], + }, + }, + }, + }, + } + ], + }, + }, + } + + stream = factory.create_component( + model_type=DeclarativeStreamModel, + component_definition=stream_definition, + config=input_config, + ) + + retriever = get_retriever(stream) + assert isinstance(retriever, LazySimpleRetriever) + assert isinstance(retriever.post_pagination_filter, ClientSideIncrementalRecordFilterDecorator) def test_given_data_feed_and_incremental_then_raise_error(): diff --git a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py index 54f544b9e5..f93db5fecc 100644 --- a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py +++ b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py @@ -155,6 +155,7 @@ def _read( manifest: Mapping[str, Any], state: Optional[List[AirbyteStateMessage]], pages_per_partition: Optional[Mapping[Tuple[str, str], List[Mapping[str, Any]]]] = None, + pages: Optional[Mapping[str, List[Mapping[str, Any]]]] = None, ) -> Tuple[List[str], List[str]]: pages_fetched = [] @@ -162,6 +163,8 @@ def paged_response(request: Any, context: Any) -> str: page = request.qs.get("page", ["1"])[0] if pages_per_partition is None: pages_fetched.append(page) + if pages is not None: + return json.dumps(pages.get(page, [])) return json.dumps(_PAGE_1 if page == "1" else _PAGE_2) owner = request.path.strip("/").split("/")[0] pages_fetched.append(f"{owner}:{page}") @@ -205,6 +208,29 @@ def test_given_no_already_synced_records_then_paginate_until_the_end( assert record_ids == ["0", "1", "2", "3", "4"] +def test_given_record_dated_in_the_future_then_filter_it_out() -> None: + """ + The retriever drops the records the cursor would not sync, and `should_be_synced` is bounded on both ends: with no + `end_datetime` the upper bound is `now()`, so records dated ahead of the connector's clock are dropped too. This is + the behaviour `is_client_side_incremental` has always had, and a data feed now matches it. + """ + pages_fetched, record_ids = _read( + _manifest(), + _state({"updated_at": "2021-01-01T00:00:00Z"}), + pages={ + "1": [ + {"id": "future", "updated_at": "2099-01-01T00:00:00Z"}, + {"id": "fresh", "updated_at": "2022-06-01T00:00:00Z"}, + {"id": "already_synced", "updated_at": "2020-06-01T00:00:00Z"}, + ] + }, + ) + + # the forward-dated record does not stop the pagination, it is only left out of the emitted records + assert pages_fetched == ["1"] + assert record_ids == ["fresh"] + + def test_given_multiple_partitions_then_each_partition_stops_on_its_own_cursor() -> None: """ A single retriever instance is shared by every partition and partitions are read concurrently, From 44165b9902c82bacce9022c4bdbf93251e38c090 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Tue, 11 Aug 2026 12:33:26 +0300 Subject: [PATCH 6/6] fix(low-code): address review on the data feed boundary page filtering - Reword the `is_data_feed` and `is_client_side_incremental` descriptions to describe the cursor window, since the filtering is bounded on both ends and the previous wording only mentioned previously-synced records. - Correct the factory comment on where the `record_filter` condition runs: the record selector sits inside the page loop, so the records the condition rejects are the ones the paginator never counts. Keeping it there preserves existing behaviour; moving it downstream would start counting them. - Make the warning describe what `is_client_side_incremental` still does on a data feed instead of calling it ignored, since it keeps defaulting the record selector to transform before filtering. - Stop re-wrapping records that already are `Record` in the client-side incremental filter, so the cursor sees the real stream name and slice. - Move the already-synced record to a second, full page in the forward-dated test so that reaching page 2 proves the forward-dated record does not stop the pagination. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative/declarative_component_schema.yaml | 4 ++-- .../declarative/extractors/record_filter.py | 9 ++++++--- .../models/declarative_component_schema.py | 4 ++-- .../parsers/model_to_component_factory.py | 10 ++++++---- .../retrievers/test_data_feed_integration.py | 15 +++++++++++---- 5 files changed, 27 insertions(+), 15 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index da0116741a..bd78c72365 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/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. The last page fetched still contains records that were synced during a previous sync, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. + 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. This is not needed when Data Feed API is enabled, as a data feed already filters out the records that were synced during a previous sync. + 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 diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 943068f875..cacee83727 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/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: diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 66c228fbd3..ac8733ee02 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/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. The last page fetched still contains records that were synced during a previous sync, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field.", + 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. This is not needed when Data Feed API is enabled, as a data feed already filters out the records that were synced during a previous sync.", + 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( diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index a80c51d2a0..2148e79652 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3430,8 +3430,9 @@ def _get_url(req: Requester) -> str: # 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: it stays in the record selector so that the records it rejects keep being counted - # by the paginator. + # 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, @@ -3447,8 +3448,9 @@ def _get_url(req: Requester) -> str: ) if post_pagination_filter and is_client_side_incremental_sync: LOGGER.warning( - f"Stream {name}: `is_client_side_incremental` is ignored when `is_data_feed` is set, " - "as a data feed already filters out the records that were synced during a previous sync." + 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 = ( diff --git a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py index f93db5fecc..d4b71b3ee1 100644 --- a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py +++ b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py @@ -218,17 +218,24 @@ def test_given_record_dated_in_the_future_then_filter_it_out() -> None: _manifest(), _state({"updated_at": "2021-01-01T00:00:00Z"}), pages={ + # a full page, so that pagination is only stopped by the already-synced record on page 2 and not by a + # short page. Page 1 holds nothing already synced, so reaching page 2 proves the forward-dated record did + # not stop the pagination. "1": [ {"id": "future", "updated_at": "2099-01-01T00:00:00Z"}, - {"id": "fresh", "updated_at": "2022-06-01T00:00:00Z"}, + {"id": "fresh_1", "updated_at": "2022-06-01T00:00:00Z"}, + {"id": "fresh_2", "updated_at": "2022-05-01T00:00:00Z"}, + {"id": "fresh_3", "updated_at": "2022-04-01T00:00:00Z"}, + ], + "2": [ {"id": "already_synced", "updated_at": "2020-06-01T00:00:00Z"}, - ] + ], }, ) # the forward-dated record does not stop the pagination, it is only left out of the emitted records - assert pages_fetched == ["1"] - assert record_ids == ["fresh"] + assert pages_fetched == ["1", "2"] + assert record_ids == ["fresh_1", "fresh_2", "fresh_3"] def test_given_multiple_partitions_then_each_partition_stops_on_its_own_cursor() -> None: