Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Comment on lines +4067 to +4070

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Assert the required empty parent state exactly.

Could we assert parent_cursor_state == {} instead? all(...) also passes for any non-boolean residual state, so it does not enforce the intended first incremental sync with empty state, wdyt?

Proposed test update
-    assert all(value is not True for value in parent_cursor_state.values())
+    assert parent_cursor_state == {}
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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())
parent_cursor_state = stream.cursor._partition_router.parent_stream_configs[
0
].stream.cursor.state
assert parent_cursor_state == {}
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@unit_tests/sources/declarative/parsers/test_model_to_component_factory.py`
around lines 4067 - 4070, Update the assertion in the parent cursor state test
to require parent_cursor_state to equal an empty dictionary exactly, replacing
the permissive all(...) check while preserving the existing cursor-state setup.



def test_incrementing_count_cursor_with_partition_router_raises_error():
content = """
type: DeclarativeStream
Expand Down
Loading