feat(airbyte-cdk): add global_state => per_partition transformation #45122

artem1205 · 2024-09-04T10:58:32Z

What

Resolve https://github.com/airbytehq/airbyte-internal-issues/issues/9576

How

add global_state => per partition transformation

Review guide

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

User Impact

Can this PR be safely reverted and rolled back?

YES 💚
NO ❌

Signed-off-by: Artem Inzhyyants <[email protected]>

vercel · 2024-09-04T10:58:36Z

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

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Sep 6, 2024 1:02pm

Signed-off-by: Artem Inzhyyants <[email protected]>

artem1205 · 2024-09-04T12:54:58Z

Tested on source-klaviyo:
run read with state:

"stream": {
      "stream_state": {
        "updated": "2120-10-10T00:00:00+00:00"
      },
      "stream_descriptor": { "name": "lists_detailed" }
    }

got transformed output per partition:

{
        "states": [
          {
            "partition": { "id": "R2p3ry", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "R4ZhCr", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "RPfQMj", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "RgS4w6", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "RnsiHB", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "RwKPyg", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "S7aBY2", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "S8nmQ9", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "SBYgiK", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "SYEFFb", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "Seq8wh", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "SmDD4y", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "TDGJsj", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "TWcKFn", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "TaSce6", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "TjbH4K", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "TpNXq9", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "UeGLUr", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "UzdNhZ", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "VDZnQt", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "VJCDbR", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "VmvmBq", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "WBxsQE", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "WJLXnV", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "X7UeXn", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "XGj3p8", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "XUbNgM", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "XpP2a5", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          },
          {
            "partition": { "id": "Ya5ziX", "parent_slice": {} },
            "cursor": { "updated": "2120-10-10T00:00:00+00:00" }
          }
        ]
      }

Signed-off-by: Artem Inzhyyants <[email protected]>

maxi297

I would like the test to be improved. The rest is nit. I'll approve even then because it is a pretty solid PR

maxi297 · 2024-09-05T12:52:13Z

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

@@ -56,7 +55,10 @@ def stream_slices(self) -> Iterable[StreamSlice]:
        slices = self._partition_router.stream_slices()
        for partition in slices:
            cursor = self._cursor_per_partition.get(self._to_partition_key(partition.partition))
-            if not cursor:
+            if not cursor and self._state_to_migrate_from:


nit: would it avoid duplication to do something like:

partition_state = self._state_to_migrate_from if self._state_to_migrate_from else self._NO_CURSOR_STATE cursor = self._create_cursor(partition_state) self._cursor_per_partition[self._to_partition_key(partition.partition)] = cursor

The reasoning is that the if part is only to determine the state to use, not to build the cursor

agreed, ref

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

maxi297 · 2024-09-05T13:06:24Z

airbyte-cdk/python/unit_tests/sources/declarative/incremental/test_per_partition_cursor.py

-        cursor.set_initial_state({"invalid_state": 1})
-
-    assert exception.value.failure_type == FailureType.config_error
+    cursor.set_initial_state({"global_state_format_key": "global_state_format_value"})


I'm not sure I understand this test. It feels like if we were to change cursor.set_initial_state({"global_state_format_key": "global_state_format_value"}) to cursor.set_initial_state({"states": []}), the test would still pass which makes me think it isn't testing global state. Should we assert mocked_cursor_factory.create.calls == [calls(<global state>)] * <number of times we expect it to be called>?

refactored, added test for call_args

Has this change been pushed? I don't see it

just pushed
fb54832

…dk-state-mgiration

Signed-off-by: Artem Inzhyyants <[email protected]>

maxi297

LGTM! Thanks for tackling the comments and being diligent about this change

Signed-off-by: Artem Inzhyyants <[email protected]>

…45122 (`artem1205/airbyte-cdk-state-mgiration`) Here is a more optimized version of the given Python program to improve runtime efficiency by minimizing redundant computations and reducing the complexity of operations. ### Key Optimizations. 1. **Re-assigned state_dict to avoid multiple dictionary access:** `stream_state.get("states")` 2. **Used set comprehension** enhances readability and avoids multiple calls. 3. **Combined dictionary updates directly** to use a single pass to populate `_cursor_per_partition`.

girarda

this is great thanks @artem1205 !

@lazebnyi can you use this to avoid the breaking change in the linkedin ads migration?

lazebnyi · 2024-09-09T20:21:33Z

I tested it locally with #44370, and everything looks good.

artem1205 · 2024-09-09T20:28:25Z

/approve-regression-tests

Check job output.

✅ Approving regression tests

Airbyte CDK: add global_state => per partition transformation

625574d

Signed-off-by: Artem Inzhyyants <[email protected]>

artem1205 self-assigned this Sep 4, 2024

octavia-squidington-iii added the CDK Connector Development Kit label Sep 4, 2024

Airbyte CDK: fmt

a19686c

Signed-off-by: Artem Inzhyyants <[email protected]>

Airbyte CDK: add comment

ce1d896

Signed-off-by: Artem Inzhyyants <[email protected]>

artem1205 marked this pull request as ready for review September 4, 2024 13:48

tolik0 approved these changes Sep 4, 2024

View reviewed changes

artem1205 requested a review from maxi297 September 5, 2024 12:07

maxi297 approved these changes Sep 5, 2024

View reviewed changes

artem1205 added 2 commits September 6, 2024 14:01

Merge remote-tracking branch 'origin/master' into artem1205/airbyte-c…

3c250e9

…dk-state-mgiration

Airbyte CDK: fix test

fb54832

Signed-off-by: Artem Inzhyyants <[email protected]>

maxi297 approved these changes Sep 6, 2024

View reviewed changes

Airbyte CDK: fmt

2b5c531

Signed-off-by: Artem Inzhyyants <[email protected]>

codeflash-ai bot mentioned this pull request Sep 6, 2024

⚡️ Speed up method PerPartitionCursor.set_initial_state by 8% in PR #45122 (artem1205/airbyte-cdk-state-mgiration) #45189

Closed

airbytehq deleted a comment from codeflash-ai bot Sep 6, 2024

girarda approved these changes Sep 7, 2024

View reviewed changes

lazebnyi approved these changes Sep 9, 2024

View reviewed changes

artem1205 merged commit e3ce82e into master Sep 9, 2024
34 of 36 checks passed

artem1205 deleted the artem1205/airbyte-cdk-state-mgiration branch September 9, 2024 20:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(airbyte-cdk): add global_state => per_partition transformation #45122

feat(airbyte-cdk): add global_state => per_partition transformation #45122

artem1205 commented Sep 4, 2024

vercel bot commented Sep 4, 2024 •

edited

Loading

artem1205 commented Sep 4, 2024

maxi297 left a comment

maxi297 Sep 5, 2024

artem1205 Sep 6, 2024

maxi297 Sep 5, 2024

artem1205 Sep 6, 2024

maxi297 Sep 6, 2024

artem1205 Sep 6, 2024

maxi297 left a comment

girarda left a comment

lazebnyi commented Sep 9, 2024

artem1205 commented Sep 9, 2024 •

edited by github-actions bot

Loading

feat(airbyte-cdk): add global_state => per_partition transformation #45122

feat(airbyte-cdk): add global_state => per_partition transformation #45122

Conversation

artem1205 commented Sep 4, 2024

What

How

Review guide

User Impact

Can this PR be safely reverted and rolled back?

vercel bot commented Sep 4, 2024 • edited Loading

artem1205 commented Sep 4, 2024

maxi297 left a comment

Choose a reason for hiding this comment

maxi297 Sep 5, 2024

Choose a reason for hiding this comment

artem1205 Sep 6, 2024

Choose a reason for hiding this comment

maxi297 Sep 5, 2024

Choose a reason for hiding this comment

artem1205 Sep 6, 2024

Choose a reason for hiding this comment

maxi297 Sep 6, 2024

Choose a reason for hiding this comment

artem1205 Sep 6, 2024

Choose a reason for hiding this comment

maxi297 left a comment

Choose a reason for hiding this comment

girarda left a comment

Choose a reason for hiding this comment

lazebnyi commented Sep 9, 2024

artem1205 commented Sep 9, 2024 • edited by github-actions bot Loading

vercel bot commented Sep 4, 2024 •

edited

Loading

artem1205 commented Sep 9, 2024 •

edited by github-actions bot

Loading