fix(low-code): data feed stop condition with client-side incremental - #1106
Conversation
…ide_incremental 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 <noreply@anthropic.com>
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. 💡 Show Tips and TricksTesting This CDK VersionYou can test this version of the CDK using the following: # Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@daryna/fix-data-feed-stop-condition-with-client-side-incremental#egg=airbyte-python-cdk[dev]' --help
# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch daryna/fix-data-feed-stop-condition-with-client-side-incrementalPR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
|
|
Warning Review limit reached
Next review available in: 49 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughData-feed streams now pass cursor state to ChangesData-feed pagination
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant DeclarativeSource
participant SimpleRetriever
participant Paginator
participant Cursor
DeclarativeSource->>SimpleRetriever: read records
SimpleRetriever->>Paginator: paginate complete records
Paginator-->>SimpleRetriever: return page records
SimpleRetriever->>Cursor: filter already-synced records
Cursor-->>SimpleRetriever: return records to emit
SimpleRetriever-->>DeclarativeSource: yield filtered records
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py`:
- Line 3601: Update the factory’s LazySimpleRetriever construction to preserve
data_feed_cursor filtering when selector-side cursor filtering is disabled.
Ensure the lazy retrieval path applies the cursor during _read_pages/_paginate
before yielding records, while retaining the existing behavior for non-data-feed
streams.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: dedfef76-6e48-4f61-b788-a7e190f982b8
📒 Files selected for processing (5)
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.pyairbyte_cdk/sources/declarative/retrievers/simple_retriever.pyunit_tests/sources/declarative/parsers/test_model_to_component_factory.pyunit_tests/sources/declarative/retrievers/test_data_feed_integration.pyunit_tests/sources/declarative/retrievers/test_simple_retriever.py
d16e403 to
117fee5
Compare
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) <noreply@anthropic.com>
117fee5 to
16d72e9
Compare
Anatolii Yatsuk (tolik0)
left a comment
Anatolii Yatsuk (tolik0)
left a comment
There was a problem hiding this comment.
The rework in 16d72e9d is a real improvement over the first approach. Filtering downstream of _read_pages means the paginator sees the page exactly as the API returned it, so CursorStopCondition needs no modification at all — no thread-local, no coupling between requesters.paginators and extractors, no widening of a public ABC. test_given_data_feed_cursor_when_read_records_then_paginator_still_sees_the_whole_page asserts the invariant that actually matters (last_page_size == 3, last_record == page[-1]) rather than an observable side effect, and the multi-partition integration test is a genuine concurrency test.
Two problems, one blocking. Both reproduced against 4758b150.
1. Blocking — transform_before_filtering silently flips to False. Routing the cursor away from create_record_selector also drops the True default that branch carried. A data feed stream with is_client_side_incremental and a record_filter.condition over a transformed field now filters against untransformed records. With an AddFields adding keep: "yes" and condition: "{{ record['keep'] == 'yes' }}":
base 4758b150 → transform_before_filtering = True → EMITTED: [{'id': '1', ..., 'keep': 'yes'}]
branch 16d72e9d → transform_before_filtering = False → EMITTED: []
Total record loss, no error. Details inline.
2. is_data_feed alone now drops records above the cursor's end boundary. should_be_synced is two-sided and _end_provider() is now() when there is no end_datetime, so forward-dated records are discarded too — a change in emitted output for every existing data feed connector, not just ones using client-side filtering. Details inline.
Two smaller things that have no line in the diff to hang off:
file_uploadernow runs before the drop.RecordSelector.filter_and_transformcallsfile_uploader.upload(record)while building records, and the data feed drop is downstream in the retriever. A data feed stream with afile_uploaderwill fetch and upload files for already-synced boundary-page records and then discard the records pointing at them. Under the old client-side-incremental path the filter ran before the upload.PaginationTracker.observesees records the sync never emits. Harmless today — I checked, the tracker only holds a cursor forpagination_reset: SPLIT_USING_CURSORand it is acopy_without_state()clone, so real stream state is untouched. But ifpagination_resetandis_data_feedare ever combined, slice reduction would be computed from records that were never emitted. Worth a one-line comment noting the drop happens afterobserve().
For what it's worth on the design question: the "a retriever shouldn't know about cursors" objection doesn't really hold — stream_slicer already is the concurrent cursor for incremental streams and the paginator already holds the same object via CursorStopCondition, so data_feed_cursor makes existing knowledge explicit rather than introducing new coupling. The field is fine.
The thing I would file as a follow-up rather than fix here: cursor-based filtering now has two implementations of the same should_be_synced call in two layers, selected by a factory conditional — and that duplication is precisely what produced problem 1. The underlying invariant is "the paginator must see the raw page", and OffsetIncrement already carries an optional extractor as a manual escape hatch for exactly that (offset_increment.py:80-86), because any record_filter that shortens a page makes offset pagination stop early today. Solving that once — hand the paginator the raw page's last record and size — would cover this case with no new plumbing and fix the latent bug too.
Local verification: ruff and mypy clean; 268 passed / 2 failed across retrievers/, paginators/ and test_model_to_component_factory.py, and those 2 (test_lazy_simple_retriever.py) fail identically on 4758b150, so they are pre-existing. CI's destination-motherduck failure is a connection check, unrelated to this diff, but worth a re-run rather than an assumption.
| # 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 |
There was a problem hiding this comment.
Blocking — this silently flips transform_before_filtering to False.
create_record_selector carries two unrelated things in one branch:
transform_before_filtering = (
False if model.transform_before_filtering is None else model.transform_before_filtering
)
if client_side_incremental_sync_cursor:
record_filter = ClientSideIncrementalRecordFilterDecorator(...)
transform_before_filtering = (
True if model.transform_before_filtering is None else model.transform_before_filtering
)Since a data feed now never enters that branch, it loses the True default along with the decorator. The user's own record_filter.condition is still built as a plain RecordFilter, but it now runs before transformations instead of after.
Any stream with both flags whose record_filter.condition references a transformation-produced field now filters against a record that does not have that field yet. Reproduced with an AddFields adding keep: "yes" and condition: "{{ record['keep'] == 'yes' }}":
base 4758b150 → transform_before_filtering = True → EMITTED: [{'id': '1', ..., 'keep': 'yes'}]
branch 16d72e9d → transform_before_filtering = False → EMITTED: []
Total record loss with no error, and the affected population is exactly the one this PR targets — is_client_side_incremental is the flag those connectors set.
Suggested fix: decouple the default from which component performs the cursor filtering. Default transform_before_filtering to True when the stream is client-side incremental or a data feed, regardless of whether client_side_incremental_sync_cursor is passed. Worth a factory test asserting retriever.record_selector.transform_before_filtering is True for a data feed stream that also sets is_client_side_incremental.
| # 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): |
There was a problem hiding this comment.
should_be_synced is two-sided:
# airbyte_cdk/sources/streams/concurrent/cursor.py:573
return self.start <= record_cursor_value <= self._end_provider()With no end_datetime — the normal data feed shape — _end_provider() is now(). So this drops records dated ahead of the connector's clock as well as the already-synced ones at the tail of the boundary page.
Reproduced with is_data_feed only (no is_client_side_incremental), page 1 leading with a record dated 2099-01-01:
base 4758b150 → PAGES ['1','2'] IDS [1, 2, 3]
branch 16d72e9d → PAGES ['1','2'] IDS [1, 2] ← id 3 dropped
Not severe — pagination is not truncated, only the record is lost, and a few seconds of clock skew self-heals next sync once now() catches up. But for APIs with genuinely forward-dated cursor values (scheduled or pending items) the record stays dropped every sync until its date arrives.
Two things worth settling:
- Whether the drop should be one-sided. The behaviour being modelled — "the boundary page still holds records from a previous sync" — is a lower-bound statement; filtering on the upper bound is a side effect of reusing
should_be_synced. - Either way this changes emitted output for every existing
is_data_feedconnector, not only ones that opted into client-side filtering. That is broader than "make the two flags compose" and deserves an explicit callout in the PR description and release notes, plus a test pinning whichever semantics you pick.
| 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. |
There was a problem hiding this comment.
Documenting the interaction is the right instinct, but a YAML description is the weakest place to put a rule the code enforces silently — the factory now ignores is_client_side_incremental on a data feed without saying so anywhere the connector developer will look.
And the combination is not actually inert: the blocking comment on model_to_component_factory.py shows setting both flags today changes transform_before_filtering. A rule that only exists in prose is exactly how that slipped through.
Suggest backing it with a LOGGER.warning when both flags are set, alongside these descriptions, so it surfaces in sync logs rather than only in the manifest schema:
if has_stop_condition_cursor 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 records synced during a previous sync."
)Separately, the wording here ("those are filtered out as well") describes one-sided filtering, while the implementation is two-sided — see the comment on simple_retriever.py.
…y 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) <noreply@anthropic.com>
`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) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
unit_tests/sources/declarative/parsers/test_model_to_component_factory.py (1)
1505-1514: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe
not isinstance(..., ClientSideIncrementalRecordFilterDecorator)assertion is trivially true here sincerecord_filterisNonein this manifest.Since the manifest doesn't set a
record_filterblock,retriever.record_selector.record_filterisNone, andNoneis never an instance ofClientSideIncrementalRecordFilterDecoratorregardless of the fix. Could we also add a manifest variant with arecord_filter.conditionset, to actually prove the selector's filter stays a plainRecordFilter(or checkretriever.record_selector.transform_before_filteringtoo, tying into thetransform_before_filteringdefault discussed inmodel_to_component_factory.py)? wdyt?🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@unit_tests/sources/declarative/parsers/test_model_to_component_factory.py` around lines 1505 - 1514, Add a manifest variant that defines a record_filter.condition, then assert the resulting record_selector.record_filter remains a plain RecordFilter rather than a ClientSideIncrementalRecordFilterDecorator. Also verify transform_before_filtering if relevant to the manifest’s expected default, while preserving the existing post-pagination filter assertions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py`:
- Around line 3441-3444: Update create_record_selector to default
transform_before_filtering to True for data-feed streams, including when
post_pagination_filter is configured, rather than relying only on
client_side_incremental_sync_cursor. Preserve any explicit
model.transform_before_filtering override, and ensure the caller path around
client_side_incremental_cursor passes the data-feed context needed for this
default.
---
Nitpick comments:
In `@unit_tests/sources/declarative/parsers/test_model_to_component_factory.py`:
- Around line 1505-1514: Add a manifest variant that defines a
record_filter.condition, then assert the resulting record_selector.record_filter
remains a plain RecordFilter rather than a
ClientSideIncrementalRecordFilterDecorator. Also verify
transform_before_filtering if relevant to the manifest’s expected default, while
preserving the existing post-pagination filter assertions.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 288f88cb-71ad-4cbc-b4a7-182a9b127168
📒 Files selected for processing (7)
airbyte_cdk/sources/declarative/declarative_component_schema.yamlairbyte_cdk/sources/declarative/extractors/record_filter.pyairbyte_cdk/sources/declarative/models/declarative_component_schema.pyairbyte_cdk/sources/declarative/parsers/model_to_component_factory.pyairbyte_cdk/sources/declarative/retrievers/simple_retriever.pyunit_tests/sources/declarative/parsers/test_model_to_component_factory.pyunit_tests/sources/declarative/retrievers/test_simple_retriever.py
🚧 Files skipped from review as they are similar to previous changes (1)
- unit_tests/sources/declarative/retrievers/test_simple_retriever.py
| client_side_incremental_cursor = ( | ||
| cursor if is_client_side_incremental_sync and not post_pagination_filter else None | ||
| ) | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
transform_before_filtering no longer defaults to True for data-feed streams with a record_filter.condition.
Since client_side_incremental_cursor is None whenever post_pagination_filter exists, create_record_selector's if client_side_incremental_sync_cursor: branch never runs for a data-feed stream (line 3456 always passes None there). Because of this, transform_before_filtering stays at its default False for any data-feed stream that also configures record_filter.condition on the selector, independent of is_client_side_incremental. Records get filtered before transformations run instead of after.
This mirrors a past "Blocking" review comment that reproduced complete record loss with an AddFields-produced field referenced by record_filter.condition, and it has no "Addressed" marker. Since the cursor-based filter now lives entirely in post_pagination_filter, could create_record_selector default transform_before_filtering to True whenever the stream is a data feed too, not only when it receives client_side_incremental_sync_cursor? wdyt?
🐛 Proposed direction (needs a matching change in create_record_selector)
client_side_incremental_cursor = (
cursor if is_client_side_incremental_sync and not post_pagination_filter else None
)
+ # `record_filter.condition` should keep evaluating after transformations for a data-feed
+ # stream too, even though cursor-based filtering now lives in `post_pagination_filter`.
+ prefer_transform_before_filtering = bool(client_side_incremental_cursor or post_pagination_filter)# create_record_selector (unchanged in this diff, ~line 3302)
transform_before_filtering = (
default_transform_before_filtering
if model.transform_before_filtering is None
else model.transform_before_filtering
)Also applies to: 3456-3456
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py` around
lines 3441 - 3444, Update create_record_selector to default
transform_before_filtering to True for data-feed streams, including when
post_pagination_filter is configured, rather than relying only on
client_side_incremental_sync_cursor. Preserve any explicit
model.transform_before_filtering override, and ensure the caller path around
client_side_incremental_cursor passes the data-feed context needed for this
default.
2 participants
What
When a
DatetimeBasedCursorsets bothis_data_feed: trueandis_client_side_incremental: true, the pagination stop condition silently never fires: incremental syncs re-fetch the stream's complete listing on every run and rely on the client-side filter to drop the stale records. This PR makes the two flags compose, so pagination stops on the first page that contains a record older than the cursor while output stays identical.Found while reviewing the source-github
repositoriesstream migration (airbytehq/airbyte#81428), where the legacy Python stream used the sorted-desc early exit that the manifest equivalent lost.How
Root cause:
SimpleRetriever._read_pagestakeslast_recordfrom the post-filter record pipeline, andClientSideIncrementalRecordFilterDecoratordrops exactly the records whoseshould_be_syncedis false — soCursorStopCondition, which only sees records that survived filtering, can never observe a stale one.The filter already evaluates
should_be_syncedon every raw record, so it now records that fact: it tracks whether the current page contained a record older than the cursor (per thread, because a single retriever — and therefore a single filter — is shared across partitions that are read concurrently; the flag is reset at the start of eachfilter_recordscall). A newFilterAwareStopConditionconsults that flag instead oflast_record, and the factory wires it in place ofCursorStopConditionwhen both flags are set.StopConditionPaginationStrategyDecoratornow consults the stop condition even when the page yielded no records, since a page whose records were all filtered out must still stop the feed;PaginationStopCondition.is_metaccordingly acceptsOptional[Record]andCursorStopConditiontreats "no record" as not met (same behavior as before).Empirically verified on a 2-page sorted-desc stream with the state cursor falling inside page 1: before, both pages were fetched; after, only page 1 is fetched and the emitted records are unchanged. A first sync with no stale records still paginates to the natural end (covered by the new integration test).
Changes
ClientSideIncrementalRecordFilterDecoratortracks a thread-localstale_record_seen_on_current_pageflag, reset on everyfilter_recordscallFilterAwareStopConditionstops pagination when the filter observed a below-cursor record on the current pageStopConditionPaginationStrategyDecoratorevaluates the stop condition even whenlast_recordisNone;CursorStopConditionisNone-safeModelToComponentFactory.create_default_paginatorwiresFilterAwareStopConditionwhen bothis_data_feedandis_client_side_incrementalare set, keepingCursorStopConditionotherwiseRecommended Review Order
airbyte_cdk/sources/declarative/requesters/paginators/strategies/stop_condition.pyairbyte_cdk/sources/declarative/extractors/record_filter.pyairbyte_cdk/sources/declarative/parsers/model_to_component_factory.py🤖 Generated with Claude Code
Summary by CodeRabbit