From fd4bdd446228f437d94d233c77be94f053b45b81 Mon Sep 17 00:00:00 2001 From: Serhii Lazebnyi Date: Fri, 31 Jul 2026 23:58:34 +0100 Subject: [PATCH] Warn when incremental_dependency is set on a stream without incremental_sync --- .../parsers/model_to_component_factory.py | 31 +++++ .../test_model_to_component_factory.py | 106 ++++++++++++++++++ 2 files changed, 137 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 e0ee55df45..d24f9f878a 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -2034,6 +2034,7 @@ def create_default_stream( ) -> AbstractStream: primary_key = model.primary_key.__root__ if model.primary_key else None self._migrate_state(model, config) + self._warn_on_ineffective_incremental_dependency(model) partition_router = self._build_stream_slicer_from_partition_router( model.retriever, @@ -2190,6 +2191,36 @@ def create_default_stream( supports_file_transfer=hasattr(model, "file_uploader") and bool(model.file_uploader), ) + def _warn_on_ineffective_incremental_dependency(self, model: DeclarativeStreamModel) -> None: + """ + `incremental_dependency: true` only takes effect when the substream defines its own + `incremental_sync`: the parent cursor is persisted under the `parent_state` key of the + substream's state, which is only emitted by incremental substreams. On a stream without + `incremental_sync`, the setting is silently ignored and all parent records are re-read on + every sync, so we warn about the misconfiguration instead. + """ + if model.incremental_sync: + return + + partition_router = getattr(model.retriever, "partition_router", None) + if not partition_router: + return + + routers = partition_router if isinstance(partition_router, list) else [partition_router] + for router in routers: + if isinstance(router, GroupingPartitionRouterModel): + router = router.underlying_partition_router + if isinstance(router, SubstreamPartitionRouterModel) and any( + parent_stream_config.incremental_dependency + for parent_stream_config in router.parent_stream_configs + ): + LOGGER.warning( + f"Stream `{model.name}` has `incremental_dependency: true` in its parent stream configuration but does not define `incremental_sync`. " + "The parent stream's cursor is only persisted in the state of an incremental substream, so this setting has no effect and all parent records will be re-read on every sync. " + "Define `incremental_sync` on this stream or remove `incremental_dependency`." + ) + return + def _migrate_state(self, model: DeclarativeStreamModel, config: Config) -> None: stream_name = model.name or "" stream_state = self._connector_state_manager.get_stream_state( 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 7728075e11..89fee961b7 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 @@ -2,6 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # import json +import logging from copy import deepcopy # mypy: ignore-errors @@ -5678,3 +5679,108 @@ def test_create_response_to_file_extractor_preserve_na_values(): ResponseToFileExtractorModel(type="ResponseToFileExtractor", preserve_na_values=True) ) assert enabled_component.preserve_na_values is True + + +def _stream_definition_with_incremental_dependency_parent( + child_has_incremental_sync: bool, +) -> dict: + incremental_sync = { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + "cursor_datetime_formats": ["%Y-%m-%dT%H:%M:%SZ"], + "start_datetime": { + "type": "MinMaxDatetime", + "datetime": "2024-01-01T00:00:00Z", + "datetime_format": "%Y-%m-%dT%H:%M:%SZ", + }, + } + parent_stream = { + "type": "DeclarativeStream", + "name": "parents", + "primary_key": ["id"], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.example.com", + "path": "/parents", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + "incremental_sync": incremental_sync, + } + child_stream = { + "type": "DeclarativeStream", + "name": "children", + "primary_key": ["id"], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.example.com", + "path": "/parents/{{ stream_partition.parent_id }}/children", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "partition_router": { + "type": "SubstreamPartitionRouter", + "parent_stream_configs": [ + { + "type": "ParentStreamConfig", + "stream": parent_stream, + "parent_key": "id", + "partition_field": "parent_id", + "incremental_dependency": True, + } + ], + }, + }, + } + if child_has_incremental_sync: + child_stream["incremental_sync"] = incremental_sync + return child_stream + + +@pytest.mark.parametrize( + "child_has_incremental_sync, expected_warning", + [ + pytest.param(False, True, id="full_refresh_child_warns"), + pytest.param(True, False, id="incremental_child_does_not_warn"), + ], +) +def test_incremental_dependency_without_incremental_sync_warns( + caplog, child_has_incremental_sync, expected_warning +): + stream_definition = _stream_definition_with_incremental_dependency_parent( + child_has_incremental_sync + ) + + with caplog.at_level(logging.WARNING, logger="airbyte.model_to_component_factory"): + stream = factory.create_component( + model_type=DeclarativeStreamModel, + component_definition=stream_definition, + config=input_config, + ) + + assert isinstance(stream, DefaultStream) + warning_emitted = any( + "`incremental_dependency: true`" in record.message and "children" in record.message + for record in caplog.records + ) + assert warning_emitted == expected_warning