Running global cursors in concurrent CDK

Author: maxi297

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -19,7 +19,10 @@
 from airbyte_cdk.sources.declarative.extractors.record_filter import (
     ClientSideIncrementalRecordFilterDecorator,
 )
-from airbyte_cdk.sources.declarative.incremental import ConcurrentPerPartitionCursor
+from airbyte_cdk.sources.declarative.incremental import (
+    ConcurrentPerPartitionCursor,
+    GlobalSubstreamCursor,
+)
 from airbyte_cdk.sources.declarative.incremental.datetime_based_cursor import DatetimeBasedCursor
 from airbyte_cdk.sources.declarative.incremental.per_partition_with_global import (
     PerPartitionWithGlobalCursor,
@@ -361,7 +364,7 @@ def _group_streams(
                     == DatetimeBasedCursorModel.__name__
                     and hasattr(declarative_stream.retriever, "stream_slicer")
                     and isinstance(
-                        declarative_stream.retriever.stream_slicer, PerPartitionWithGlobalCursor
+                        declarative_stream.retriever.stream_slicer, (GlobalSubstreamCursor, PerPartitionWithGlobalCursor)
                     )
                 ):
                     stream_state = self._connector_state_manager.get_stream_state(

unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py

@@ -3456,6 +3456,7 @@ def test_given_global_state_when_read_then_state_is_not_per_partition() -> None:
     manifest["definitions"]["post_comments_stream"]["incremental_sync"][
         "global_substream_cursor"
     ] = True
+    manifest["streams"].remove({"$ref": "#/definitions/post_comment_votes_stream"})
     record = {
         "id": 9,
         "post_id": 1,
@@ -3478,6 +3479,7 @@ def test_given_global_state_when_read_then_state_is_not_per_partition() -> None:
             },
         ),
     ]
+    
     run_mocked_test(
         mock_requests,
         manifest,
@@ -3489,5 +3491,6 @@ def test_given_global_state_when_read_then_state_is_not_per_partition() -> None:
             "lookback_window": 1,
             "parent_state": {"posts": {"updated_at": "2024-01-30T00:00:00Z"}},
             "state": {"updated_at": "2024-01-25T00:00:00Z"},
+            "use_global_cursor": True  # ensures that it is running the Concurrent CDK version as this is not populated in the declarative implementation
         },  # this state does have per partition which would be under `states`
     )