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:
|
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughData-feed retrieval now applies cursor filtering after pagination. Pagination still evaluates complete records for stop conditions. Factory wiring supports standard and lazy retrievers, client-side incremental modes, transformation ordering, and partition-specific cursor behavior. ChangesData-feed pagination
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant SimpleRetriever
participant Paginator
participant ClientSideIncrementalRecordFilterDecorator
SimpleRetriever->>Paginator: paginate complete records
Paginator-->>SimpleRetriever: return page records
SimpleRetriever->>ClientSideIncrementalRecordFilterDecorator: filter records after pagination
ClientSideIncrementalRecordFilterDecorator-->>SimpleRetriever: return records to emit
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.
…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
…ental 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) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 3448-3452: Update the warning in the post-pagination filter and
client-side incremental branch to clarify that is_client_side_incremental only
affects the incremental-sync filtering behavior, while it still defaults
transform_before_filtering to True. Ensure the message does not claim the entire
flag is ignored.
In `@unit_tests/sources/declarative/retrievers/test_data_feed_integration.py`:
- Around line 211-231: Update
test_given_record_dated_in_the_future_then_filter_it_out so page 1 contains only
the future and fresh records, and move already_synced to page 2. Adjust the
assertions to require both pages to be fetched while still asserting that only
fresh is emitted, proving the future record does not stop pagination.
🪄 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: 8956a03a-ecb5-45d6-a060-c9c88378562d
📒 Files selected for processing (4)
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.py
🚧 Files skipped from review as they are similar to previous changes (1)
- airbyte_cdk/sources/declarative/retrievers/simple_retriever.py
Anatolii Yatsuk (tolik0)
left a comment
Anatolii Yatsuk (tolik0)
left a comment
There was a problem hiding this comment.
Approving. The blocking transform_before_filtering regression is fixed, and it's fixed at the right level — the default now follows the is_client_side_incremental flag rather than whichever component happens to receive the cursor, with both directions pinned by tests.
The bigger improvement is post_pagination_filter being a real ClientSideIncrementalRecordFilterDecorator owned by the retriever instead of an inline should_be_synced loop. That collapses the duplication that caused the round-2 bug in the first place: there is now one implementation of cursor filtering, positioned differently, rather than two selected by a factory conditional. Removing the failure mode rather than the instance.
I've withdrawn the two-sided filtering objection — reproduced your state-poisoning argument independently and it's decisive; details in that thread. Two-sided filtering is load-bearing here, not merely conservative, and my "self-heals next sync" note was wrong.
Three comment-level items below, none blocking. The first is one sentence and worth fixing before merge — a wrong rationale in a comment is what a future change gets argued from, which is precisely how the round-2 regression happened.
Also worth filing separately, beyond the main state-poisoning bug: last_page_size is post-filter. The comment thread below measures 2 for a 3-record page, and OffsetIncrement already carries an optional extractor as a manual escape hatch for exactly this (offset_increment.py:80-86) — which is why the same probe paginates correctly under OffsetIncrement but not PageIncrement. Handing the paginator the raw page's size and last record once would remove the need for that hatch and for any future component to work around filtered pages.
Verification on ef4ffc13: ruff and mypy clean; 272 passed / 2 failed across retrievers/, paginators/ and test_model_to_component_factory.py, and those 2 (test_lazy_simple_retriever.py) fail identically on current origin/main, so they're pre-existing.
- 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) <noreply@anthropic.com>
…ent-side-incremental
|
/prerelease
|
Resolve conflicts and repin the CDK to airbytehq/airbyte-python-cdk#1106. - source-github version: master released 2.1.40, so the release candidate moves to 2.1.41-rc.1 (metadata.yaml, pyproject.toml, changelog) - metadata.yaml: keep master's autopilot rolloutConfiguration alongside enableProgressiveRollout - pyproject.toml: airbyte-cdk 7.24.0.post10.dev31496796422, the prerelease built from the head of the data-feed stop-condition branch (it also carries RateLimitedMultipleTokenAuthenticator and UnionPartitionRouter) - poetry.lock regenerated against that CDK
5fc9b62
into
main
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
New Features
Documentation