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