Skip to content

fix(declarative): seed parent state only from a scalar legacy child cursor value #1189

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Baz (bazarnov) merged 1 commit into from
Oct 2, 2026
Merged

fix(declarative): seed parent state only from a scalar legacy child cursor value #1189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
15 changes: 10 additions & 5 deletions airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5144,8 +5144,15 @@ def _instantiate_parent_stream_state_manager(
)

if not extracted_parent_state and not isinstance(extracted_parent_state, dict):
cursor_values = child_state.values()
if cursor_values and len(cursor_values) == 1:
cursor_values = list(child_state.values())
# Only a scalar legacy cursor value can seed the parent. Sentinels such as
# `{"__ab_full_refresh_sync_complete": true}` or legacy per-partition `states`
# would crash the parent cursor initialization.
if (
len(cursor_values) == 1
and isinstance(cursor_values[0], (str, int, float))
and not isinstance(cursor_values[0], bool)
):
incremental_sync_model: Union[
DatetimeBasedCursorModel,
IncrementingCountCursorModel,
Expand All @@ -5166,9 +5173,7 @@ def _instantiate_parent_stream_state_manager(
stream_descriptor=StreamDescriptor(
name=parent_stream_name, namespace=None
),
stream_state=AirbyteStateBlob(
{cursor_field: list(cursor_values)[0]}
),
stream_state=AirbyteStateBlob({cursor_field: cursor_values[0]}),
),
)
return ConnectorStateManager([extracted_parent_state] if extracted_parent_state else [])
Expand Down
79 changes: 69 additions & 10 deletions unit_tests/sources/declarative/parsers/test_model_to_component_factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4972,14 +4972,71 @@ def test_create_concurrent_cursor_from_perpartition_cursor_runs_state_migrations
)


def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_sentinel_state():
@pytest.mark.parametrize(
"child_state, cursor_datetime_formats, expected_parent_cursor_value",
[
pytest.param(
{"__ab_no_cursor_state_message": True},
["%Y-%m-%dT%H:%M:%S.%f%z"],
"2024-01-01T00:00:00.000000+0000",
id="full_refresh_sentinel",
),
pytest.param(
{"__ab_full_refresh_sync_complete": True},
["%Y-%m-%dT%H:%M:%S.%f%z"],
"2024-01-01T00:00:00.000000+0000",
id="resumable_full_refresh_sentinel",
),
pytest.param(
{
"states": [
{
"partition": {"id": "1", "parent_slice": {}},
"cursor": {"updated_at": "2024-02-01T00:00:00.000000+0000"},
}
]
},
["%Y-%m-%dT%H:%M:%S.%f%z"],
"2024-01-01T00:00:00.000000+0000",
id="legacy_per_partition_state_without_parent_state",
),
pytest.param(
{"updated_at": "2024-02-01T00:00:00.000000+0000"},
["%Y-%m-%dT%H:%M:%S.%f%z"],
"2024-02-01T00:00:00.000000+0000",
id="legacy_string_cursor_value",
),
pytest.param(
{"updated_at": 1706745600},
["%Y-%m-%dT%H:%M:%S.%f%z", "%s"],
"2024-02-01T00:00:00.000000+0000",
id="legacy_epoch_cursor_value",
),
pytest.param(
{"updated_at": 1706745600.0},
["%Y-%m-%dT%H:%M:%S.%f%z", "%s"],
"2024-02-01T00:00:00.000000+0000",
id="legacy_float_epoch_cursor_value",
),
pytest.param(
{"updated_at": "2024-02-01T00:00:00.000000+0000", "lookback_window": 0},
["%Y-%m-%dT%H:%M:%S.%f%z"],
"2024-01-01T00:00:00.000000+0000",
id="multiple_values_are_not_a_legacy_cursor",
),
],
)
def test_create_concurrent_cursor_from_perpartition_cursor_seeds_parent_only_from_a_cursor_value(
child_state, cursor_datetime_formats, expected_parent_cursor_value
):
"""
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`.
Without `parent_state`, a legacy child state with a single cursor value seeds the parent cursor. A child
stream can also carry state that is not a cursor value: the full refresh sentinels, or a legacy
per-partition state. If a later connector version adds an `incremental_dependency` parent, that value
must not be re-keyed under the parent's cursor field - doing so crashed cursor initialization with
`ValueError: No format in [...] matching True`.
"""
content = """
content = f"""
type: DeclarativeStream
primary_key: "id"
name: test
Expand All @@ -4995,7 +5052,8 @@ def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_
type: "DatetimeBasedCursor"
cursor_field: "updated_at"
datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
start_datetime: "{{ config['start_time'] }}"
cursor_datetime_formats: {cursor_datetime_formats}
start_datetime: "{{{{ config['start_time'] }}}}"
retriever:
type: SimpleRetriever
name: test
Expand Down Expand Up @@ -5034,7 +5092,8 @@ def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_
type: "DatetimeBasedCursor"
cursor_field: "updated_at"
datetime_format: "%Y-%m-%dT%H:%M:%S.%f%z"
start_datetime: "{{ config['start_time'] }}"
cursor_datetime_formats: {cursor_datetime_formats}
start_datetime: "{{{{ config['start_time'] }}}}"
retriever:
type: SimpleRetriever
requester:
Expand All @@ -5054,7 +5113,7 @@ def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_
type=AirbyteStateType.STREAM,
stream=AirbyteStreamState(
stream_descriptor=StreamDescriptor(name="test"),
stream_state=AirbyteStateBlob({"__ab_no_cursor_state_message": True}),
stream_state=AirbyteStateBlob(child_state),
),
)
]
Expand All @@ -5071,7 +5130,7 @@ def test_create_concurrent_cursor_from_perpartition_cursor_ignores_full_refresh_
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())
assert parent_cursor_state == {"updated_at": expected_parent_cursor_value}


def test_incrementing_count_cursor_with_partition_router_raises_error():
Expand Down
Loading