🐛 low-code: Fix incremental substreams #35471

girarda · 2024-02-21T01:55:05Z

What

Fix incremental substreams so per-partition cursors are updated instead of being appended to the list. This is done by only considering the partition fields from the stream_state and excluding the cursor ones.
Resolves [connector builder] BUG: Incremental substream state is appended rather than replaced (and start_time is not updated) #33854

How

Rename PerPartitionStreamSlice to StreamSlice and move it to types.py
- Also update the partition and cursor_slice methods to recursively look up the nested PerPartitionStreamSlice when needed
Update all stream slicers so they always return a PerPartitionStreamSlice instead of any mappings
Update substream partition router so it only considers the partition part of the parent stream slice when setting the parent_slice field
- This is the bulk of the change.
- Technically a breaking change: SubstreamPartitionRouter expects the parent streams to be declarative to ensure they return PerPartitionStreamSlice
Fix mypy issues

🚨 User Impact 🚨

SubstreamPartitionRouter require the parent streams to be DeclarativeStreams
Incremental substreams will work in incremental mode instead of always reading all the data and appending a new state to the blob

Pre-merge Actions

Expand the relevant checklist and delete the others.

New Connector

Community member or Airbyter

Community member? Grant edit access to maintainers (instructions)
Unit & integration tests added and passing. Community members, please provide proof of success locally e.g: screenshot or copy-paste unit, integration, and acceptance test output. To run acceptance tests for a Python connector, follow instructions in the README. For java connectors run ./gradlew :airbyte-integrations:connectors:<name>:integrationTest.
Connector version is set to 0.0.1
- Dockerfile has version 0.0.1
Documentation updated
- Connector's README.md
- Connector's bootstrap.md. See description and examples
- docs/integrations/<source or destination>/<name>.md including changelog with an entry for the initial version. See changelog example
- docs/integrations/README.md

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.

Updating a connector

Community member or Airbyter

Grant edit access to maintainers (instructions)
Unit & integration tests added

Airbyter

If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.

Create a non-forked branch based on this PR and test the below items on it
Build is successful
If new credentials are required for use in CI, add them to GSM. Instructions.

Connector Generator

Issue acceptance criteria met
PR name follows PR naming conventions
If adding a new generator, add it to the list of scaffold modules being tested
The generator test modules (all connectors with -scaffold in their name) have been updated with the latest scaffold by running ./gradlew :airbyte-integrations:connector-templates:generator:generateScaffolds then checking in your changes
Documentation which references the generator is updated as needed

Updating the Python CDK

Airbyter

Before merging:

Pull Request description explains what problem it is solving
Code change is unit tested
Build and my-py check pass
Smoke test the change on at least one affected connector
- On Github: Run this workflow, passing --use-local-cdk --name=source-<connector> as options
- Locally: airbyte-ci connectors --use-local-cdk --name=source-<connector> test
PR is reviewed and approved

After merging:

Publish the CDK
- The CDK does not follow proper semantic versioning. Choose minor if this the change has significant user impact or is a breaking change. Choose patch otherwise.
- Write a thoughtful changelog message so we know what was updated.
Merge the platform PR that was auto-created for updating the Connector Builder's CDK version
- This step is optional if the change does not affect the connector builder or declarative connectors.

vercel · 2024-02-21T01:55:06Z

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Mar 6, 2024 2:43am

github-actions · 2024-02-21T01:55:26Z

Before Merging a Connector Pull Request

Wow! What a great pull request you have here! 🎉

To merge this PR, ensure the following has been done/considered for each connector added or updated:

PR name follows PR naming conventions
Breaking changes are considered. If a Breaking Change is being introduced, ensure an Airbyte engineer has created a Breaking Change Plan.
Connector version has been incremented in the Dockerfile and metadata.yaml according to our Semantic Versioning for Connectors guidelines
You've updated the connector's metadata.yaml file any other relevant changes, including a breakingChanges entry for major version bumps. See metadata.yaml docs
Secrets in the connector's spec are annotated with airbyte_secret
All documentation files are up to date. (README.md, bootstrap.md, docs.md, etc...)
Changelog updated in docs/integrations/<source or destination>/<name>.md with an entry for the new version. See changelog example
Migration guide updated in docs/integrations/<source or destination>/<name>-migrations.md with an entry for the new version, if the version is a breaking change. See migration guide example
If set, you've ensured the icon is present in the platform-internal repo. (Docs)

If the checklist is complete, but the CI check is failing,

Check for hidden checklists in your PR description
Toggle the github label checklist-action-run on/off to re-run the checklist CI.

…hq/airbyte into alex/explore_substream_state_bug

brianjlai

no need to block as overall changes look good, but I had some comments around the main part of the changes in substream_partition_router.py that I wanted a little more clarity on

brianjlai · 2024-03-04T20:25:00Z

...ython/airbyte_cdk/sources/declarative/requesters/request_options/request_options_provider.py

@@ -28,7 +28,7 @@ def get_request_params(
        stream_state: Optional[StreamState] = None,
        stream_slice: Optional[StreamSlice] = None,
        next_page_token: Optional[Mapping[str, Any]] = None,
-    ) -> MutableMapping[str, Any]:
+    ) -> Mapping[str, Any]:


This in theory could have an affect if we were relying on manipulating the contents of this return value. Do you by chance have context why this specific return was originally mutable?

brianjlai · 2024-03-04T20:42:34Z

...e-cdk/python/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

-                parent_field = parent_stream_config.parent_key.eval(self.config)
-                stream_state_field = parent_stream_config.partition_field.eval(self.config)
+                parent_field = parent_stream_config.parent_key.eval(self.config)  # type: ignore # parent_key is always casted to an interpolated string
+                partition_field = parent_stream_config.partition_field.eval(self.config)  # type: ignore # partition_field is always casted to an interpolated string
                for parent_stream_slice in parent_stream.stream_slices(
                    sync_mode=SyncMode.full_refresh, cursor_field=None, stream_state=None
                ):
                    empty_parent_slice = True
                    parent_slice = parent_stream_slice


this assignment looks redundant. We don't manipulate either of these variables

brianjlai · 2024-03-04T20:51:57Z

...e-cdk/python/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

                for parent_stream_slice in parent_stream.stream_slices(
                    sync_mode=SyncMode.full_refresh, cursor_field=None, stream_state=None
                ):
                    empty_parent_slice = True
                    parent_slice = parent_stream_slice
+                    if parent_slice:
+                        parent_partition = {k: v for k, v in parent_slice.items() if k in parent_slice.partition.keys()}


Two things:

The if parent_slice: check is to protect against None stream slices that could come from the parent stream right?

How come we do this complicated comprehension statement to create the dictionary if the parent slice item key is in the partition keys? Is there a reason why we can't just call parent_slice.partition().items()?

That's right. I don't really expect this to happen for declarative stream, but it is technically possible for a stream's stream_slice method to return None...

Good catch. Simplified to your suggestion!

natikgadzhi

TIL a few things 👀 thank you for cleaning this up.

natikgadzhi · 2024-03-06T00:34:28Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

-        if self.end_datetime and not isinstance(self.end_datetime, MinMaxDatetime):
-            self.end_datetime = MinMaxDatetime(self.end_datetime, parameters)
+        self._start_datetime = (
+            MinMaxDatetime(self.start_datetime, parameters) if not isinstance(self.start_datetime, MinMaxDatetime) else self.start_datetime


I don't know how exactly MinMaxDatetime works, but could we make it's constructor always return an object of MinMaxTime and accept MinMaxtime too so that we can always wrap start_datetime into it, cutting on one if statement?

Sure, that's one more function call, but perhaps more readable?

good idea. Added a factory method for MinMaxDatetime and simplified this call

natikgadzhi · 2024-03-06T00:35:25Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

+        self._end_datetime = (
+            None
+            if not self.end_datetime
+            else MinMaxDatetime(self.end_datetime, parameters)


Here, nested ifs are harder to read — perhaps if MinMaxtime is always called on that, we cut out one if, and it's easier to read instantly.

good idea. Added a factory method for MinMaxDatetime and simplified this call

natikgadzhi · 2024-03-06T00:36:48Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -109,17 +116,20 @@ def set_initial_state(self, stream_state: StreamState) -> None:

        :param stream_state: The state of the stream as returned by get_stream_state
        """
-        self._cursor = stream_state.get(self.cursor_field.eval(self.config)) if stream_state else None
+        self._cursor = stream_state.get(self._cursor_field.eval(self.config)) if stream_state else None

    def close_slice(self, stream_slice: StreamSlice, most_recent_record: Optional[Record]) -> None:


Why narrow it?

natikgadzhi · 2024-03-06T00:38:08Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/per_partition_cursor.py

@@ -179,16 +123,16 @@ def _get_state_for_partition(self, partition: Mapping[str, Any]) -> Optional[Str
        return None

    @staticmethod
-    def _is_new_state(stream_state):
+    def _is_new_state(stream_state: Mapping[str, Any]) -> bool:


Why not be strict on type here? We don't have a special StreamState type as that mapping yet?

that's correct. We have types for StreamSlice and Record, but StreamState is still TBD

…hq/airbyte into alex/explore_substream_state_bug

girarda added 5 commits February 20, 2024 17:00

first hack

10d06bc

Update

7b9af90

Update

82f8915

this looks ok with all the hardcoded stuff...

b7f45eb

less hardcoding

0dc78b7

octavia-squidington-iii added area/connectors Connector related issues CDK Connector Development Kit labels Feb 21, 2024

octavia-squidington-iii added the connectors/source/chargebee label Feb 21, 2024

girarda added 19 commits February 20, 2024 17:57

update

5b90e2b

delete

9f78777

reset to master

c503691

update

624eddc

undo rename

35db4cf

rename

94f2c40

update

e316f5f

Update

3bdc41e

update

ccd0645

remove prints

6dce076

reset files

7360b8c

remove unused file

601d15d

small update

c94ef81

move code around

b8eb8db

format

288e56c

fix about half the mypy issues

c5fac71

fix more type issues

af8f439

more fixes

49b8a26

update

90254dd

girarda mentioned this pull request Feb 22, 2024

[connector builder] BUG: Incremental substream state is appended rather than replaced (and start_time is not updated) #33854

Closed

Merge branch 'alex/explore_substream_state_bug' of github.com:airbyte…

3dd24e3

…hq/airbyte into alex/explore_substream_state_bug

vercel bot deployed to Preview March 1, 2024 17:46 View deployment

girarda requested review from brianjlai and maxi297 March 4, 2024 15:59

brianjlai approved these changes Mar 4, 2024

View reviewed changes

girarda added 5 commits March 4, 2024 18:12

simplify assignment

87ebdbc

handle none stream slice

2edd16c

flake

2aa68d5

reformat

6c3df15

reset to master

728cd62

octavia-squidington-iii removed the area/connectors Connector related issues label Mar 5, 2024

natikgadzhi reviewed Mar 6, 2024

View reviewed changes

girarda added 9 commits March 5, 2024 16:45

format

f1a415e

small cleanup

dc01934

missing spaces

75a851c

fix test

9560422

try fixing test on ci

6dbc44e

try adding another filter

c149d6d

Merge branch 'master' into alex/explore_substream_state_bug

64e408d

stop referring to the legacy state

d3dddfa

Merge branch 'alex/explore_substream_state_bug' of github.com:airbyte…

c7b70e8

…hq/airbyte into alex/explore_substream_state_bug

vercel bot deployed to Preview March 6, 2024 02:31 View deployment

vercel bot deployed to Preview March 6, 2024 02:36 View deployment

airbyte-ci format

641d735

girarda merged commit f55abc1 into master Mar 6, 2024
27 checks passed

girarda deleted the alex/explore_substream_state_bug branch March 6, 2024 02:50

girarda mentioned this pull request Mar 6, 2024

🐛 follow up to #35471: update the cartesian stream slicer #35865

Merged

xiaohansong pushed a commit that referenced this pull request Mar 7, 2024

🐛 low-code: Fix incremental substreams (#35471)

aabd2bd

girarda added a commit that referenced this pull request Mar 7, 2024

🐛 follow up to #35471: update the cartesian stream slicer (#35865)

4a808ee

ChristoGrab mentioned this pull request Mar 13, 2024

Source Greenhouse: unpin CDK #35988

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

🐛 low-code: Fix incremental substreams #35471

🐛 low-code: Fix incremental substreams #35471

girarda commented Feb 21, 2024 •

edited

Loading

vercel bot commented Feb 21, 2024 •

edited

Loading

github-actions bot commented Feb 21, 2024

brianjlai left a comment

brianjlai Mar 4, 2024

brianjlai Mar 4, 2024

brianjlai Mar 4, 2024

girarda Mar 5, 2024

natikgadzhi left a comment

natikgadzhi Mar 6, 2024

girarda Mar 6, 2024

natikgadzhi Mar 6, 2024

girarda Mar 6, 2024

natikgadzhi Mar 6, 2024

natikgadzhi Mar 6, 2024

girarda Mar 6, 2024

🐛 low-code: Fix incremental substreams #35471

🐛 low-code: Fix incremental substreams #35471

Conversation

girarda commented Feb 21, 2024 • edited Loading

What

How

Recommended reading order

🚨 User Impact 🚨

Pre-merge Actions

Community member or Airbyter

Airbyter

Community member or Airbyter

Airbyter

Airbyter

vercel bot commented Feb 21, 2024 • edited Loading

github-actions bot commented Feb 21, 2024

Before Merging a Connector Pull Request

brianjlai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

natikgadzhi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

girarda commented Feb 21, 2024 •

edited

Loading

vercel bot commented Feb 21, 2024 •

edited

Loading