From 9911f40c77237efa910cceb7becf851166cc9086 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 9 Jul 2026 05:58:13 +0000 Subject: [PATCH] fix(declarative): ignore NO_CURSOR_STATE_KEY sentinel when reconstructing parent stream state Co-Authored-By: bot_apk --- .../parsers/model_to_component_factory.py | 10 +- .../test_model_to_component_factory.py | 109 ++++++++++++++++++ 2 files changed, 118 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 5a75912b3..fda5d47b7 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -666,6 +666,7 @@ from airbyte_cdk.sources.streams.concurrent.state_converters.incrementing_count_stream_state_converter import ( IncrementingCountStreamStateConverter, ) +from airbyte_cdk.sources.streams.core import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction from airbyte_cdk.sources.types import Config from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer @@ -4164,7 +4165,14 @@ def _instantiate_parent_stream_state_manager( ) if not extracted_parent_state and not isinstance(extracted_parent_state, dict): - cursor_values = child_state.values() + # Exclude the no-cursor sentinel (persisted by a child stream that + # completed without a cursor value, e.g. {NO_CURSOR_STATE_KEY: True}). + # Treating it as a real cursor value would synthesize a bogus parent + # state like {cursor_field: True} and later crash when the parent's + # ConcurrentCursor tries to parse the boolean as a timestamp. + cursor_values = [ + value for key, value in child_state.items() if key != NO_CURSOR_STATE_KEY + ] if cursor_values and len(cursor_values) == 1: incremental_sync_model: Union[ DatetimeBasedCursorModel, diff --git a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py index 5149f46d2..fb53962a9 100644 --- a/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py +++ b/unit_tests/sources/declarative/parsers/test_model_to_component_factory.py @@ -182,6 +182,7 @@ ) from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource from airbyte_cdk.sources.message.repository import StateFilteringMessageRepository +from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.call_rate import HttpAPIBudget, MovingWindowCallRatePolicy from airbyte_cdk.sources.streams.concurrent.clamping import ( ClampingEndProvider, @@ -3923,6 +3924,114 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations ) +def test_create_concurrent_cursor_from_perpartition_cursor_ignores_no_cursor_sentinel_child_state(): + """A child stream that finished with no cursor value persists the NO_CURSOR_STATE_KEY sentinel. + + The parent-state reconstruction fallback must not treat that sentinel as a real cursor value + (which would synthesize a parent state like {cursor_field: True} and crash when the parent's + ConcurrentCursor parses the boolean as a timestamp). Instead the parent cursor should initialize + from its configured start_datetime. + """ + content = """ + type: DeclarativeStream + primary_key: "id" + name: test + schema_loader: + type: InlineSchemaLoader + schema: + $schema: "http://json-schema.org/draft-07/schema" + type: object + properties: + id: + type: string + incremental_sync: + type: "DatetimeBasedCursor" + cursor_field: "updated_at" + datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" + start_datetime: "{{ config['start_time'] }}" + retriever: + type: SimpleRetriever + name: test + requester: + type: HttpRequester + name: "test" + url_base: "https://api.test.com/v3/" + http_method: "GET" + authenticator: + type: NoAuth + record_selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: [] + partition_router: + type: SubstreamPartitionRouter + parent_stream_configs: + - type: ParentStreamConfig + parent_key: id + partition_field: id + incremental_dependency: true + stream: + type: DeclarativeStream + primary_key: id + name: parent_stream + schema_loader: + type: InlineSchemaLoader + schema: + $schema: "http://json-schema.org/draft-07/schema" + type: object + properties: + id: + type: string + incremental_sync: + type: "DatetimeBasedCursor" + cursor_field: "updated_at" + datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z" + start_datetime: "{{ config['start_time'] }}" + retriever: + type: SimpleRetriever + requester: + type: HttpRequester + url_base: "https://api.test.com/v3/parent" + http_method: "GET" + record_selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: [] + """ + + connector_state_manager = ConnectorStateManager( + state=[ + AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name="test"), + stream_state=AirbyteStateBlob({NO_CURSOR_STATE_KEY: True}), + ), + ) + ] + ) + factory = ModelToComponentFactory( + emit_connector_builder_messages=True, connector_state_manager=connector_state_manager + ) + + # Should not raise (previously crashed with "No format ... matching True" while building the + # parent ConcurrentCursor from the synthesized {updated_at: True} state). + stream = factory.create_component( + model_type=DeclarativeStreamModel, + component_definition=YamlDeclarativeSource._parse(content), + config=input_config, + ) + + parent_cursor_state = stream.cursor._partition_router.parent_stream_configs[ + 0 + ].stream.cursor.state + # The sentinel must not become a parent cursor value; the parent initializes from its + # configured start rather than from a low-water-mark derived from the child's sentinel state. + assert parent_cursor_state["updated_at"] == "2024-01-01T00:00:00.000000+0000" + + def test_incrementing_count_cursor_with_partition_router_raises_error(): content = """ type: DeclarativeStream