From df07154ba1e777bd08a2fad89da42065304c7461 Mon Sep 17 00:00:00 2001 From: Ryan Waskewich Date: Sun, 26 Jul 2026 09:12:04 -0400 Subject: [PATCH] fix(declarative): ignore full-refresh sentinel in parent state seeding Full refresh streams checkpoint {"__ab_no_cursor_state_message": true}. When a later connector version converts such a stream to incremental with an incremental_dependency parent, the single-value fallback in _instantiate_parent_stream_state_manager re-keys the sentinel's boolean under the parent's cursor field and cursor initialization crashes with ValueError: No format in [...] matching True, before any records are read. Filter the sentinel out of the child state before seeding parent state so the converted stream starts its first incremental sync from empty state. Co-Authored-By: Claude Fable 5 --- .../parsers/model_to_component_factory.py | 9 ++ .../test_model_to_component_factory.py | 102 ++++++++++++++++++ 2 files changed, 111 insertions(+) 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 9de12ea87..664f178fd 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -633,6 +633,7 @@ NoopMessageRepository, ) 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 ( APIBudget, FixedWindowCallRatePolicy, @@ -4131,6 +4132,14 @@ def create_parent_stream_config_with_substream_wrapper( self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any ) -> Any: child_state = self._connector_state_manager.get_stream_state(stream_name, None) + if NO_CURSOR_STATE_KEY in child_state: + # Full refresh streams checkpoint a `{NO_CURSOR_STATE_KEY: true}` sentinel. When such a + # stream is later converted to incremental with an incremental_dependency parent, + # `_instantiate_parent_stream_state_manager` would treat the sentinel's boolean as a legacy + # cursor value and re-key it under the parent's cursor field, crashing cursor initialization. + child_state = { + key: value for key, value in child_state.items() if key != NO_CURSOR_STATE_KEY + } parent_state: Optional[Mapping[str, Any]] = ( child_state if model.incremental_dependency and child_state else None 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 bdf061a40..0c971822f 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 @@ -3968,6 +3968,108 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations ) +def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_sentinel_state(): + """ + A stream that synced as full refresh checkpoints `{"__ab_no_cursor_state_message": true}`. If a later + connector version converts that stream to incremental with an `incremental_dependency` parent, the + sentinel must not be re-keyed under the parent's cursor field as if it were a legacy cursor value — + doing so crashed cursor initialization with `ValueError: No format in [...] matching True`. + """ + 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({"__ab_no_cursor_state_message": True}), + ), + ) + ] + ) + factory = ModelToComponentFactory( + emit_connector_builder_messages=True, connector_state_manager=connector_state_manager + ) + 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 + assert all(value is not True for value in parent_cursor_state.values()) + + def test_incrementing_count_cursor_with_partition_router_raises_error(): content = """ type: DeclarativeStream