✨ low-code: Add Incremental Parent State Handling to SubstreamPartitionRouter #38211

tolik0 · 2024-05-15T12:03:57Z

What

This PR resolves the issue with incremental substream reads in the low-code. It ensures that the state is passed to the parent stream when generating partitions to avoid reading the parent stream in full refresh mode. Resolves: https://github.com/airbytehq/airbyte-internal-issues/issues/7369

Connector's changes:

How

The code changes introduce a new field incremental_dependency in the ParentStreamConfig class. The SubstreamPartitionRouter was updated to read parent state incrementally and save the parent state when the stream slice is processed. Additionally, new default methods were added to StreamSlicer for cases when a stream slicer does not have parent streams.

Review guide

User Impact

The end result is an optimized incremental substream reading process, which improves sync performance by reducing the need to read the parent stream in full refresh mode.
There are no negative side effects anticipated.

Can this PR be safely reverted and rolled back?

YES 💚
NO ❌

vercel · 2024-05-15T12:04:01Z

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		Jun 7, 2024 6:24pm

artem1205 · 2024-05-23T13:04:15Z

Please split CDK and connectors changes into different PRs

tolik0 · 2024-05-23T13:08:24Z

@artem1205 I will create separate PRs for connectors; I just added these changes here for now for easier testing. This way we can execute regression tests from this branch.

girarda · 2024-05-26T22:19:11Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/stream_slicer.py

@@ -28,3 +28,21 @@ def stream_slices(self) -> Iterable[StreamSlice]:

        :return: List of stream slices
        """
+
+    def set_parent_state(self, stream_state: StreamState) -> None:


this method is only used on partition routers so I think it would be worth creating a new PartitionRouter interface that would be more specific.

We should also leave the two methods as abstract to ensure developers explicitly define them instead of relying on a default behavior that may or may not make sense for their use case

Any stream slicer can handle parent states. For example CartesianProductStreamSlicer.

DeclarativeCursor doesn't, right?

I think it makes sense for CartesianProductStreamSlicer for also being a PartitionRouter

Changed CartesianProductStreamSlicer to PartitionRouter

girarda · 2024-05-26T22:20:48Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/stream_slicer.py

+
+    def set_parent_state(self, stream_state: StreamState) -> None:
+        """
+        Set the state of the parent streams.


the docstring should be more specific about the parameter expectations. stream_state must have parent_state which is not obvious from the signature. It might even be worth creating a new class to codify the expectation (kind of like how StreamSlice codifies the presence of a partition and a cursor_slice)

Updated docs for all classes

girarda · 2024-05-26T22:22:20Z

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

+
+                    yield from stream_slices_for_parent
+
+    def set_parent_state(self, stream_state: Optional[StreamState]) -> None:


why is stream_state an Optional?

girarda · 2024-05-26T22:24:04Z

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

@@ -86,6 +86,8 @@ def set_initial_state(self, stream_state: StreamState) -> None:
        for state in stream_state["states"]:
            self._cursor_per_partition[self._to_partition_key(state["partition"])] = self._create_cursor(state["cursor"])

+        self._partition_router.set_parent_state(stream_state)


this deserves a comment as it's not obvious why we need to update the router's parent state

Added comment

reading through the code again. We're passing the full state, not just the parent_state. Is there a reason this can't just call the router's set_initial_state method?

Stream slicers and partition routers don't keep track of the cursor and don't have a set_initial_state method. set_parent_state sets the state to the parent streams that partition routers operate on, if such parent streams exist.

seems to me this means we want to add a set_initial_state method to the routers - from the caller's perspective, it doesn't really matter that it's setting a parent state. It's an implementation detail.

Furthermore, not all stream slicers know or care about parents (eg list partition router)

edit: I do think the state being set is somewhat different, but the interface feels very strange to me for two reasons

Not every stream slicers should know or care about parent streams

even from the caller, the fact that it's updating a parent stream state is irrelevant. It's just passing the whole state object

The same goes with set_initial_state for stream slicers, as not every stream slicer cares about the state. Introducing a set_initial_state method could create confusion, implying that partition routers are somehow dependent on the state. set_parent_state clearly articulates that it is only connected to the parent streams, as the stream interacts with the parent stream through the partition router.

yeah, that's exactly the point - we're now saying that SubstreamPartitionRouter and CartesianProductStreamSlicer are incremental and have a concept of state they need to keep track of.

I'm not sure if it'll make the code simpler or more complicated, but conceptually, this makes me think the SubstreamPartitionRouter should have a Cursor when incremental_dependency is True.

This would essentially mean that instead of keeping the parent state in a dict, it could pass the parent records to its cursor's observe method while iterating over them.

I feel 4/10 strongly on this issue so not worth blocking on.

I changed the methods' names. Regarding using a Cursor to keep track of the state, it is already managed by the cursor of parent streams. Partition routers just retrieve the state from the internal parent streams.

girarda · 2024-05-30T00:07:40Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/stream_slicer.py

@@ -28,3 +28,21 @@ def stream_slices(self) -> Iterable[StreamSlice]:

        :return: List of stream slices
        """
+
+    def set_parent_state(self, stream_state: StreamState) -> None:


DeclarativeCursor doesn't, right?

I think it makes sense for CartesianProductStreamSlicer for also being a PartitionRouter

girarda · 2024-05-30T00:08:15Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/stream_slicer.py

@@ -28,3 +28,45 @@ def stream_slices(self) -> Iterable[StreamSlice]:

        :return: List of stream slices
        """
+
+    def set_parent_state(self, stream_state: StreamState) -> None:


let's leave the method as abstract to force the concrete class to implement it (or intentionally implement a no-op)

girarda · 2024-05-30T00:08:19Z

airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/stream_slicer.py

+        """
+        pass
+
+    def get_parent_state(self) -> Optional[Mapping[str, StreamState]]:


let's leave the method as abstract to force the concrete class to implement it (or intentionally implement a no-op)

girarda · 2024-05-30T00:17:40Z

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

@@ -86,6 +86,8 @@ def set_initial_state(self, stream_state: StreamState) -> None:
        for state in stream_state["states"]:
            self._cursor_per_partition[self._to_partition_key(state["partition"])] = self._create_cursor(state["cursor"])

+        self._partition_router.set_parent_state(stream_state)


yeah, that's exactly the point - we're now saying that SubstreamPartitionRouter and CartesianProductStreamSlicer are incremental and have a concept of state they need to keep track of.

I'm not sure if it'll make the code simpler or more complicated, but conceptually, this makes me think the SubstreamPartitionRouter should have a Cursor when incremental_dependency is True.

This would essentially mean that instead of keeping the parent state in a dict, it could pass the parent records to its cursor's observe method while iterating over them.

I feel 4/10 strongly on this issue so not worth blocking on.

girarda · 2024-05-30T00:21:21Z

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

                    parent_partition = parent_stream_slice.partition if parent_stream_slice else {}

+                    # we need to read all records for slice to update the parent stream cursor
+                    stream_slices_for_parent = []
                    for parent_record in parent_stream.read_records(


let's add a comment here explaining that the sync mode and the stream state passed are irrelevant since we set the initial state from set_parent_state

artem1205

LGTM! could you please add more tests to ensure that changes and stream state is handled correctly.
e.g. test_incremental_parent_state

create test_source
create stream_A (parent)
create stream_B (child)
run test_source.read() with/without stream_state
assert:
- requests
- records
- stream_state (in output)

octavia-squidington-iii added area/connectors Connector related issues CDK Connector Development Kit connectors/source/slack labels May 15, 2024

tolik0 self-assigned this May 15, 2024

tolik0 added 4 commits May 16, 2024 18:02

Update substream partition router to pass state to parent streams

8909f49

Migrate incremental substream for Slack

03dba07

Add new ParentStreamConfig parameter - incremental_dependency

5172a32

Fix other partition routers

11851ee

tolik0 force-pushed the tolik0/airbyte-cdk/add-incremental-substreams branch from 611acee to 11851ee Compare May 16, 2024 15:03

Migrated streams for Jira and ZendeskSupport

a553710

octavia-squidington-iii added connectors/source/jira connectors/source/zendesk-support labels May 16, 2024

tolik0 added 3 commits May 16, 2024 20:14

Update poetry.lock for Slack

9043883

Add incremental_dependency for Jira

24f9990

Add unit tests

84b40a0

tolik0 marked this pull request as ready for review May 23, 2024 12:59

tolik0 requested a review from a team as a code owner May 23, 2024 12:59

tolik0 requested a review from artem1205 May 23, 2024 13:00

Add type annotations

d3b2591

Fix more mypy errors

afea860

Fix minor errors

47a6f21

girarda reviewed May 26, 2024

View reviewed changes

girarda requested a review from brianjlai May 26, 2024 22:25

Add parent state handling to CartesianProductStreamSlicer

86d26b5

Fix mypy errors

5021d0c

tolik0 requested a review from girarda May 28, 2024 15:03

girarda reviewed May 30, 2024

View reviewed changes

artem1205 requested changes May 31, 2024

View reviewed changes

tolik0 added 6 commits June 6, 2024 19:43

Create PartitionRouter abstract class

46fed95

Move CartesianProductStreamSlicer

c745075

Update other partition routers

8875233

Add comment for reading parent stream

776b974

Update comment

ad0d49b

Add unit test for reading incremental substream with state

20085a8

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

✨ low-code: Add Incremental Parent State Handling to SubstreamPartitionRouter #38211

✨ low-code: Add Incremental Parent State Handling to SubstreamPartitionRouter #38211

tolik0 commented May 15, 2024 •

edited

vercel bot commented May 15, 2024 •

edited

artem1205 commented May 23, 2024

tolik0 commented May 23, 2024

girarda May 26, 2024

tolik0 May 27, 2024 •

edited

girarda May 30, 2024

tolik0 Jun 6, 2024

girarda May 26, 2024

tolik0 May 27, 2024

girarda May 26, 2024

tolik0 May 27, 2024

girarda May 26, 2024

tolik0 May 27, 2024

girarda May 29, 2024

tolik0 May 29, 2024

girarda May 29, 2024

girarda May 29, 2024

tolik0 May 29, 2024

girarda May 30, 2024

tolik0 Jun 6, 2024

girarda May 30, 2024

girarda May 30, 2024

tolik0 Jun 6, 2024

girarda May 30, 2024

tolik0 Jun 6, 2024

girarda May 30, 2024

girarda May 30, 2024

tolik0 Jun 6, 2024

artem1205 left a comment


		yield from stream_slices_for_parent

		def set_parent_state(self, stream_state: Optional[StreamState]) -> None:

✨ low-code: Add Incremental Parent State Handling to SubstreamPartitionRouter #38211

Are you sure you want to change the base?

✨ low-code: Add Incremental Parent State Handling to SubstreamPartitionRouter #38211

Conversation

tolik0 commented May 15, 2024 • edited

What

How

Review guide

User Impact

Can this PR be safely reverted and rolled back?

vercel bot commented May 15, 2024 • edited

artem1205 commented May 23, 2024

tolik0 commented May 23, 2024

Choose a reason for hiding this comment

tolik0 May 27, 2024 • edited

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

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

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

artem1205 left a comment

Choose a reason for hiding this comment

tolik0 commented May 15, 2024 •

edited

vercel bot commented May 15, 2024 •

edited

tolik0 May 27, 2024 •

edited