From d72f9a92059013732152a4e11bda061362a682d8 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 16 Sep 2026 17:34:00 +0300 Subject: [PATCH 1/5] feat(RecordExpander): add `parent_fields` to copy named parent values onto expanded items `RecordExpander` can already carry parent context onto each expanded item, but only by way of `remain_original_record: true`, which deep-copies the entire parent record once per item. When the parent is large and the nested list is long - a GitHub pull request node with ten reviews, say - that copies the whole pull request ten times to carry one URL. `parent_fields` names the values to copy instead: ```yaml record_expander: type: RecordExpander expand_records_from_field: ["reviews", "nodes"] parent_fields: - type: ParentFieldPath parent_path: ["url"] record_path: ["pull_request_url"] ``` Both paths are lists, so a value nested in the parent can be written to a nested position on the item; intermediate objects are created as needed. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies `None` - which is what `parent.get(field)` does in the connector classes this replaces. Glob metacharacters are rejected in both paths, matching the existing rule on `truncation_indicator_path`. The copy happens in `_apply_parent_context`, the single point both the embedded-items path and the `truncated_list_retriever` path go through, so fetched items are stamped the same way as embedded ones. `parent_fields` and `remain_original_record` are independent and may both be set. Making one suppress the other would be surprising; the reason to prefer `parent_fields` is that it does not need the deep copy, which the docstring and the schema description both say. Scalar list items previously fell through unwrapped unless `remain_original_record` was set, in which case they were wrapped as `{"value": ..., "original_record": ...}`. They are now wrapped as `{"value": ...}` plus the copied fields whenever either option asks for parent context, and still yielded bare when neither does. The generated model was hand-applied and verified against the pinned generator (`datamodel_code_generator==0.26.3` with the flag set from `bin/generate_component_manifest_files.py`): the `ParentFieldPath` class and the `parent_fields` field are byte-identical to its output after `ruff format`. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 45 ++++ .../sources/declarative/expanders/__init__.py | 8 +- .../declarative/expanders/record_expander.py | 95 +++++++- .../models/declarative_component_schema.py | 22 ++ .../parsers/model_to_component_factory.py | 24 ++ .../expanders/test_record_expander.py | 220 +++++++++++++++++- .../test_model_to_component_factory.py | 94 ++++++++ 7 files changed, 498 insertions(+), 10 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 071f5eeaf4..62e8209660 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2250,6 +2250,12 @@ definitions: description: If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false. type: boolean default: false + parent_fields: + title: Parent Fields + description: 'Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.' + type: array + items: + "$ref": "#/definitions/ParentFieldPath" on_no_records: title: On No Records description: Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged. @@ -2277,6 +2283,45 @@ definitions: $parameters: type: object additionalProperties: true + ParentFieldPath: + title: Parent Field Path + description: One value to copy from the record being expanded onto each expanded item. Both paths are lists, so a value nested inside the parent can be copied into a nested position on the item; intermediate objects on `record_path` are created as needed. Glob characters (`*`, `?`, `[`) are not supported in either path, since each must identify a single field; this is enforced on the interpolated values. + type: object + required: + - type + - parent_path + - record_path + properties: + type: + type: string + enum: [ParentFieldPath] + parent_path: + title: Parent Path + description: Path to the value on the record being expanded. + type: array + items: + type: string + interpolation_context: + - config + examples: + - ["url"] + - ["id"] + - ["repository", "name"] + record_path: + title: Record Path + description: Path on the expanded item to write the value to. An existing value is overwritten. + type: array + items: + type: string + interpolation_context: + - config + examples: + - ["pull_request_url"] + - ["comment_id"] + - ["parent", "id"] + $parameters: + type: object + additionalProperties: true ExponentialBackoffStrategy: title: Exponential Backoff description: Backoff strategy with an exponential backoff interval. The interval is defined as factor * 2^attempt_count. diff --git a/airbyte_cdk/sources/declarative/expanders/__init__.py b/airbyte_cdk/sources/declarative/expanders/__init__.py index 1b10ee82d7..23f59bf1bf 100644 --- a/airbyte_cdk/sources/declarative/expanders/__init__.py +++ b/airbyte_cdk/sources/declarative/expanders/__init__.py @@ -2,6 +2,10 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # -from airbyte_cdk.sources.declarative.expanders.record_expander import OnNoRecords, RecordExpander +from airbyte_cdk.sources.declarative.expanders.record_expander import ( + OnNoRecords, + ParentFieldPath, + RecordExpander, +) -__all__ = ["OnNoRecords", "RecordExpander"] +__all__ = ["OnNoRecords", "ParentFieldPath", "RecordExpander"] diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py index 68c71af693..99125d5ae9 100644 --- a/airbyte_cdk/sources/declarative/expanders/record_expander.py +++ b/airbyte_cdk/sources/declarative/expanders/record_expander.py @@ -50,6 +50,57 @@ class OnNoRecords(Enum): emit_parent = "emit_parent" +@dataclass +class ParentFieldPath: + """One field to copy from the record being expanded onto each expanded item. + + `parent_path` locates the value on the parent, `record_path` says where to put it on the + child. Both are lists, so a value nested inside the parent can be copied into a nested + position on the child. Glob metacharacters are rejected: each path must identify a single + field. + """ + + parent_path: Sequence[str] + record_path: Sequence[str] + config: Config + parameters: InitVar[Mapping[str, Any]] + + def __post_init__(self, parameters: Mapping[str, Any]) -> None: + if not self.parent_path: + raise ValueError("`parent_path` cannot be empty.") + if not self.record_path: + raise ValueError("`record_path` cannot be empty.") + self._parent_path: list[InterpolatedString] = [ + InterpolatedString.create(path, parameters=parameters) for path in self.parent_path + ] + self._record_path: list[InterpolatedString] = [ + InterpolatedString.create(path, parameters=parameters) for path in self.record_path + ] + RecordExpander._reject_globs(self.evaluated_parent_path(), "parent_path") + RecordExpander._reject_globs(self.evaluated_record_path(), "record_path") + + def evaluated_parent_path(self) -> list[Any]: + return [segment.eval(self.config) for segment in self._parent_path] + + def evaluated_record_path(self) -> list[Any]: + return [segment.eval(self.config) for segment in self._record_path] + + def copy_onto( + self, parent_record: Mapping[str, Any], child_record: MutableMapping[str, Any] + ) -> None: + """Copy the parent value onto the child, overwriting whatever was there. + + A `parent_path` that the parent does not have copies `None`, which is what the custom + connector classes this replaces do (`parent.get(field)`). Distinguishing "absent" from + "present and null" would need a third option and no connector needs one. + """ + try: + value = dpath.get(dict(parent_record), self.evaluated_parent_path()) + except (KeyError, ValueError): + value = None + dpath.new(child_record, self.evaluated_record_path(), value) + + @dataclass class RecordExpander: """Expands records by extracting items from a nested array field. @@ -59,6 +110,11 @@ class RecordExpander: to embed the full parent record under `original_record` in each expanded item when you need downstream transformations to access parent context. + When only a few parent fields are needed, prefer `parent_fields` over + `remain_original_record`: it copies the named values onto each expanded item instead of + deep-copying the whole parent once per item, which matters when the parent is large and the + nested list is long. + The expand_records_from_field path supports wildcards (*) for matching multiple arrays. When wildcards are used, items from all matched arrays are extracted and emitted. @@ -72,6 +128,17 @@ class RecordExpander: remain_original_record: true ``` + ``` + record_expander: + type: RecordExpander + expand_records_from_field: + - "reviews" + - "nodes" + parent_fields: + - parent_path: ["url"] + record_path: ["pull_request_url"] + ``` + ``` record_expander: type: RecordExpander @@ -88,6 +155,13 @@ class RecordExpander: Supports wildcards (*). remain_original_record: If True, each expanded record will include the original parent record in an "original_record" field. Defaults to False. + parent_fields: Named values to copy from the record being expanded onto each expanded + item. Each entry has a `parent_path` and a `record_path`; an existing value at + `record_path` is overwritten, and a `parent_path` the parent does not have copies + `None`. Independent of `remain_original_record` - both may be set, though copying + named fields is the cheaper way to carry parent context. Applies to items fetched + through `truncated_list_retriever` as well as to embedded ones. Glob metacharacters + are rejected in both paths. on_no_records: Behavior when expansion produces no records. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged. truncation_indicator_path: Path within each record to a field indicating that the @@ -125,6 +199,7 @@ class RecordExpander: config: Config parameters: InitVar[Mapping[str, Any]] remain_original_record: bool = False + parent_fields: Optional[Sequence[ParentFieldPath]] = None on_no_records: OnNoRecords = OnNoRecords.skip truncation_indicator_path: Optional[Sequence[str]] = None truncated_list_retriever: Optional["Retriever"] = None @@ -216,11 +291,10 @@ def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMap expanded_record = dict(item) self._apply_parent_context(parent_record, expanded_record) yield expanded_record - elif self.remain_original_record: - yield { - "value": item, - "original_record": copy.deepcopy(parent_record), - } + elif self._carries_parent_context(): + wrapped: MutableMapping[str, Any] = {"value": item} + self._apply_parent_context(parent_record, wrapped) + yield wrapped else: yield item @@ -323,8 +397,10 @@ def _fetch_complete_list(self, parent_record: Mapping[str, Any]) -> Iterable[Any expanded_record = dict(data) self._apply_parent_context(parent_record, expanded_record) yield expanded_record - elif self.remain_original_record: - yield {"value": data, "original_record": copy.deepcopy(parent_record)} + elif self._carries_parent_context(): + wrapped: MutableMapping[str, Any] = {"value": data} + self._apply_parent_context(parent_record, wrapped) + yield wrapped else: yield data @@ -334,3 +410,8 @@ def _apply_parent_context( """Apply parent context to a child record.""" if self.remain_original_record: child_record["original_record"] = copy.deepcopy(parent_record) + for parent_field in self.parent_fields or []: + parent_field.copy_onto(parent_record, child_record) + + def _carries_parent_context(self) -> bool: + return bool(self.remain_original_record or self.parent_fields) diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 41265c9db9..4345e40df2 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -522,6 +522,23 @@ class OnNoRecords(Enum): emit_parent = "emit_parent" +class ParentFieldPath(BaseModel): + type: Literal["ParentFieldPath"] + parent_path: List[str] = Field( + ..., + description="Path to the value on the record being expanded.", + examples=[["url"], ["id"], ["repository", "name"]], + title="Parent Path", + ) + record_path: List[str] = Field( + ..., + description="Path on the expanded item to write the value to. An existing value is overwritten.", + examples=[["pull_request_url"], ["comment_id"], ["parent", "id"]], + title="Record Path", + ) + parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters") + + class ExponentialBackoffStrategy(BaseModel): type: Literal["ExponentialBackoffStrategy"] factor: Optional[Union[float, str]] = Field( @@ -2300,6 +2317,11 @@ class RecordExpander(BaseModel): description='If true, each expanded record will include the original parent record in an "original_record" field. Defaults to false.', title="Remain Original Record", ) + parent_fields: Optional[List[ParentFieldPath]] = Field( + None, + description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Parent Fields", + ) on_no_records: Optional[OnNoRecords] = Field( OnNoRecords.skip, description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', 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 bec394f85e..6dab0a84e9 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -109,6 +109,7 @@ ) from airbyte_cdk.sources.declarative.expanders.record_expander import ( OnNoRecords, + ParentFieldPath, RecordExpander, ) from airbyte_cdk.sources.declarative.extractors import ( @@ -390,6 +391,9 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ParametrizedComponentsResolver as ParametrizedComponentsResolverModel, ) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + ParentFieldPath as ParentFieldPathModel, +) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ParentStreamConfig as ParentStreamConfigModel, ) @@ -824,6 +828,7 @@ def _init_mappings(self) -> None: PropertiesFromEndpointModel: self.create_properties_from_endpoint, PropertyChunkingModel: self.create_property_chunking, QueryPropertiesModel: self.create_query_properties, + ParentFieldPathModel: self.create_parent_field_path, RecordExpanderModel: self.create_record_expander, RecordFilterModel: self.create_record_filter, RecordSelectorModel: self.create_record_selector, @@ -2545,6 +2550,12 @@ def create_record_expander( config=config, parameters=model.parameters or {}, remain_original_record=model.remain_original_record or False, + parent_fields=[ + self._create_component_from_model(model=parent_field, config=config) + for parent_field in model.parent_fields + ] + if model.parent_fields + else None, on_no_records=OnNoRecords(model.on_no_records.value) if model.on_no_records else OnNoRecords.skip, @@ -2554,6 +2565,19 @@ def create_record_expander( suppress_incomplete_fetch_warning=suppress_incomplete_fetch_warning, ) + @staticmethod + def create_parent_field_path( + model: ParentFieldPathModel, + config: Config, + **kwargs: Any, + ) -> ParentFieldPath: + return ParentFieldPath( + parent_path=model.parent_path, + record_path=model.record_path, + config=config, + parameters=model.parameters or {}, + ) + @staticmethod def create_response_to_file_extractor( model: ResponseToFileExtractorModel, diff --git a/unit_tests/sources/declarative/expanders/test_record_expander.py b/unit_tests/sources/declarative/expanders/test_record_expander.py index 23731dda9e..42fc37109a 100644 --- a/unit_tests/sources/declarative/expanders/test_record_expander.py +++ b/unit_tests/sources/declarative/expanders/test_record_expander.py @@ -8,7 +8,10 @@ import pytest from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, AirbyteRecordMessage, Level, Type -from airbyte_cdk.sources.declarative.expanders.record_expander import RecordExpander +from airbyte_cdk.sources.declarative.expanders.record_expander import ( + ParentFieldPath, + RecordExpander, +) from airbyte_cdk.sources.message import InMemoryMessageRepository from airbyte_cdk.sources.types import Record, StreamSlice @@ -572,3 +575,218 @@ def test_no_warning_when_retriever_configured(caplog): assert len(records) == 15 assert not [r for r in caplog.records if r.levelname == "WARNING"] + + +def _parent_field(parent_path, record_path): + return ParentFieldPath( + parent_path=parent_path, + record_path=record_path, + config=config, + parameters=parameters, + ) + + +def _reviews_parent(): + return { + "number": 7, + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}, {"id": "PRR_2"}]}, + } + + +def test_parent_fields_copies_the_named_value_onto_every_item(): + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + parent_fields=[_parent_field(["url"], ["pull_request_url"])], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record(_reviews_parent())) == [ + {"id": "PRR_1", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + {"id": "PRR_2", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + ] + + +def test_parent_fields_does_not_embed_the_parent(): + """The reason to use `parent_fields`: no `original_record`, so no deep copy per item.""" + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + parent_fields=[_parent_field(["url"], ["pull_request_url"])], + config=config, + parameters=parameters, + ) + + for record in expander.expand_record(_reviews_parent()): + assert "original_record" not in record + + +def test_parent_fields_and_remain_original_record_are_independent(): + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + remain_original_record=True, + parent_fields=[_parent_field(["number"], ["pull_request_number"])], + config=config, + parameters=parameters, + ) + + records = list(expander.expand_record(_reviews_parent())) + + assert [record["pull_request_number"] for record in records] == [7, 7] + assert all(record["original_record"]["number"] == 7 for record in records) + + +def test_parent_fields_reads_and_writes_nested_paths(): + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["repository", "name"], ["parent", "repository_name"])], + config=config, + parameters=parameters, + ) + + parent = {"repository": {"name": "airbyte"}, "items": [{"id": 1}]} + + assert list(expander.expand_record(parent)) == [ + {"id": 1, "parent": {"repository_name": "airbyte"}} + ] + + +def test_parent_fields_overwrites_an_existing_value_on_the_item(): + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["id"], ["id"])], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"id": "parent", "items": [{"id": "child"}]})) == [ + {"id": "parent"} + ] + + +def test_parent_fields_copies_none_when_the_parent_lacks_the_field(): + """Matches `parent.get(field)` in the connector classes this replaces.""" + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["missing"], ["copied"])], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"items": [{"id": 1}]})) == [{"id": 1, "copied": None}] + + +def test_parent_fields_copies_a_present_null_the_same_way(): + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["maybe"], ["copied"])], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"maybe": None, "items": [{"id": 1}]})) == [ + {"id": 1, "copied": None} + ] + + +def test_several_parent_fields_are_applied_in_order(): + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[ + _parent_field(["a"], ["copied"]), + _parent_field(["b"], ["copied"]), + ], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"a": 1, "b": 2, "items": [{}]})) == [{"copied": 2}] + + +def test_parent_fields_does_not_mutate_the_parent_record(): + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + parent_fields=[_parent_field(["url"], ["pull_request_url"])], + config=config, + parameters=parameters, + ) + parent = _reviews_parent() + + list(expander.expand_record(parent)) + + assert parent == _reviews_parent() + + +def test_parent_fields_wraps_scalar_items(): + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["id"], ["parent_id"])], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"id": 9, "items": ["a", "b"]})) == [ + {"value": "a", "parent_id": 9}, + {"value": "b", "parent_id": 9}, + ] + + +def test_scalar_items_stay_bare_without_parent_context(): + expander = RecordExpander( + expand_records_from_field=["items"], + config=config, + parameters=parameters, + ) + + assert list(expander.expand_record({"items": ["a", "b"]})) == ["a", "b"] + + +def test_parent_fields_apply_to_items_fetched_by_the_truncated_list_retriever(): + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + truncation_indicator_path=["reviews", "has_more"], + truncated_list_retriever=_make_retriever([{"id": "PRR_9"}]), + parent_fields=[_parent_field(["url"], ["pull_request_url"])], + config=config, + parameters=parameters, + ) + parent = { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}], "has_more": True}, + } + + assert list(expander.expand_record(parent)) == [ + {"id": "PRR_9", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"} + ] + + +@pytest.mark.parametrize("path", [["*"], ["a", "?"], ["a[0]"]]) +def test_globs_are_rejected_in_parent_path(path): + with pytest.raises(ValueError, match="Glob characters"): + _parent_field(path, ["copied"]) + + +@pytest.mark.parametrize("path", [["*"], ["a", "?"], ["a[0]"]]) +def test_globs_are_rejected_in_record_path(path): + with pytest.raises(ValueError, match="Glob characters"): + _parent_field(["id"], path) + + +@pytest.mark.parametrize("parent_path,record_path", [([], ["a"]), (["a"], [])]) +def test_empty_paths_are_rejected(parent_path, record_path): + with pytest.raises(ValueError, match="cannot be empty"): + _parent_field(parent_path, record_path) + + +def test_parent_field_paths_interpolate_config(): + parent_field = ParentFieldPath( + parent_path=["{{ config['from'] }}"], + record_path=["{{ config['to'] }}"], + config={"from": "url", "to": "pull_request_url"}, + parameters=parameters, + ) + child = {} + + parent_field.copy_onto({"url": "https://example.com/7"}, child) + + assert child == {"pull_request_url": "https://example.com/7"} 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 39adbde700..101f3105e0 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 @@ -1990,6 +1990,100 @@ def test_client_side_incremental_with_partition_router(): ) +def test_create_record_expander_with_parent_fields(): + """The source-github `reviews` shape: children stamped with a field from the parent node.""" + content = """ + selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["data", "repository", "pullRequests", "nodes"] + record_expander: + type: RecordExpander + expand_records_from_field: ["reviews", "nodes"] + parent_fields: + - type: ParentFieldPath + parent_path: ["url"] + record_path: ["pull_request_url"] + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + expander = selector.extractor.record_expander + assert expander is not None + assert expander.remain_original_record is False + assert len(expander.parent_fields) == 1 + + response = requests.Response() + response._content = json.dumps( + { + "data": { + "repository": { + "pullRequests": { + "nodes": [ + { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}, {"id": "PRR_2"}]}, + }, + { + "url": "https://github.com/airbytehq/airbyte/pull/8", + "reviews": {"nodes": [{"id": "PRR_3"}]}, + }, + ] + } + } + } + } + ).encode("utf-8") + + assert list(selector.extractor.extract_records(response)) == [ + {"id": "PRR_1", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + {"id": "PRR_2", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + {"id": "PRR_3", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/8"}, + ] + + +def test_create_record_expander_without_parent_fields_leaves_them_unset(): + content = """ + selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["data"] + record_expander: + type: RecordExpander + expand_records_from_field: ["items"] + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + assert selector.extractor.record_expander.parent_fields is None + + def test_create_record_expander_with_truncated_list_retriever(): content = """ selector: From 120c1efd2652d46e50429a768edd9ecbbf501401 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 16 Sep 2026 18:22:27 +0300 Subject: [PATCH 2/5] feat(RecordExpander): add `merge_parent` to shallow-merge the parent onto each expanded item With `merge_parent: true` each expanded item becomes the parent record shallow-merged with the item, the item's own keys winning on collision, and the value at `expand_records_from_field` removed from the parent's copy. Only that value is removed: on a multi-segment path the top-level key stays with its other fields, and the parent record itself is never mutated. This is the `{**record, **item}` flatten that source-mailchimp (`components.py:16`) and source-orb (`components.py:16-45`) implement in custom Python today. The order inside `_apply_parent_context` is fixed: the merge first, then `parent_fields` copies, then `original_record` when `remain_original_record` is set, so a named copy can overwrite a merged value. The three options are independent and may be combined. Scalar list items are wrapped as `{"value": item}` before the merge, and items fetched through `truncated_list_retriever` are merged the same way as embedded ones. The stripped parent is built once per parent record rather than once per item. The generated model change was hand-applied in the same style as `parent_fields`. Co-Authored-By: Claude Fable 5.1 --- .../declarative_component_schema.yaml | 5 + .../declarative/expanders/record_expander.py | 112 +++++++++-- .../models/declarative_component_schema.py | 5 + .../parsers/model_to_component_factory.py | 1 + .../expanders/test_record_expander.py | 181 ++++++++++++++++++ .../test_model_to_component_factory.py | 87 +++++++++ 6 files changed, 371 insertions(+), 20 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 62e8209660..4ef4215e5d 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2256,6 +2256,11 @@ definitions: type: array items: "$ref": "#/definitions/ParentFieldPath" + merge_parent: + title: Merge Parent + description: If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it. + type: boolean + default: false on_no_records: title: On No Records description: Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged. diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py index 99125d5ae9..4dfd6b0caa 100644 --- a/airbyte_cdk/sources/declarative/expanders/record_expander.py +++ b/airbyte_cdk/sources/declarative/expanders/record_expander.py @@ -115,6 +115,10 @@ class RecordExpander: deep-copying the whole parent once per item, which matters when the parent is large and the nested list is long. + Set `merge_parent: true` to flatten the parent into each item instead: the parent's fields, + minus the expanded list, are shallow-merged underneath the item's own fields, so the item wins + on any key both have. + The expand_records_from_field path supports wildcards (*) for matching multiple arrays. When wildcards are used, items from all matched arrays are extracted and emitted. @@ -139,6 +143,14 @@ class RecordExpander: record_path: ["pull_request_url"] ``` + ``` + record_expander: + type: RecordExpander + expand_records_from_field: + - "activity" + merge_parent: true + ``` + ``` record_expander: type: RecordExpander @@ -162,6 +174,13 @@ class RecordExpander: named fields is the cheaper way to carry parent context. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. Glob metacharacters are rejected in both paths. + merge_parent: If True, each expanded item is the parent record shallow-merged with the + item, the item's own keys winning on collision, and the expanded list removed from + the parent's copy. Only the value at `expand_records_from_field` is removed: for a + multi-segment path the top-level key stays with its other fields. The merge happens + before `parent_fields` are copied and before `original_record` is embedded, so both + can overwrite merged values. Applies to items fetched through + `truncated_list_retriever` as well as to embedded ones. Defaults to False. on_no_records: Behavior when expansion produces no records. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged. truncation_indicator_path: Path within each record to a field indicating that the @@ -200,6 +219,7 @@ class RecordExpander: parameters: InitVar[Mapping[str, Any]] remain_original_record: bool = False parent_fields: Optional[Sequence[ParentFieldPath]] = None + merge_parent: bool = False on_no_records: OnNoRecords = OnNoRecords.skip truncation_indicator_path: Optional[Sequence[str]] = None truncated_list_retriever: Optional["Retriever"] = None @@ -259,13 +279,18 @@ def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMap expand_path = self._evaluated_expand_path() truncated = bool(self._truncation_indicator_path) and self._is_truncated(parent_record) + # Built once per parent, not once per item: the walk that finds the expanded list costs + # as much as extracting it does. + merge_base = ( + self._without_expanded_list(parent_record, expand_path) if self.merge_parent else None + ) if truncated and self.truncated_list_retriever: # Streamed, so the shortfall is only known once the retriever is exhausted. If the # consumer stops early the fetch was cut short by it, not by the API, and no warning # would be accurate anyway. fetched_count = 0 - for fetched in self._fetch_complete_list(parent_record): + for fetched in self._fetch_complete_list(parent_record, merge_base): fetched_count += 1 yield fetched self._warn_if_fetch_incomplete(parent_record, expand_path, fetched_count) @@ -288,13 +313,9 @@ def expand_record(self, record: MutableMapping[Any, Any]) -> Iterable[MutableMap for items in embedded_lists: for item in items: if isinstance(item, dict): - expanded_record = dict(item) - self._apply_parent_context(parent_record, expanded_record) - yield expanded_record + yield self._apply_parent_context(parent_record, dict(item), merge_base) elif self._carries_parent_context(): - wrapped: MutableMapping[str, Any] = {"value": item} - self._apply_parent_context(parent_record, wrapped) - yield wrapped + yield self._apply_parent_context(parent_record, {"value": item}, merge_base) else: yield item @@ -376,7 +397,9 @@ def _is_truncated(self, parent_record: MutableMapping[Any, Any]) -> bool: except (KeyError, ValueError): return False - def _fetch_complete_list(self, parent_record: Mapping[str, Any]) -> Iterable[Any]: + def _fetch_complete_list( + self, parent_record: Mapping[str, Any], merge_base: Optional[Mapping[str, Any]] + ) -> Iterable[Any]: if not self.truncated_list_retriever: return stream_slice = StreamSlice(partition={"parent_record": parent_record}, cursor_slice={}) @@ -394,24 +417,73 @@ def _fetch_complete_list(self, parent_record: Mapping[str, Any]) -> Iterable[Any else: data = item if isinstance(data, Mapping): - expanded_record = dict(data) - self._apply_parent_context(parent_record, expanded_record) - yield expanded_record + yield self._apply_parent_context(parent_record, dict(data), merge_base) elif self._carries_parent_context(): - wrapped: MutableMapping[str, Any] = {"value": data} - self._apply_parent_context(parent_record, wrapped) - yield wrapped + yield self._apply_parent_context(parent_record, {"value": data}, merge_base) else: yield data def _apply_parent_context( - self, parent_record: Mapping[str, Any], child_record: MutableMapping[str, Any] - ) -> None: - """Apply parent context to a child record.""" - if self.remain_original_record: - child_record["original_record"] = copy.deepcopy(parent_record) + self, + parent_record: Mapping[str, Any], + child_record: MutableMapping[str, Any], + merge_base: Optional[Mapping[str, Any]], + ) -> MutableMapping[str, Any]: + """Return the child record carrying the configured parent context. + + The order is fixed: the parent is merged underneath the child first, then `parent_fields` + are copied on top, then `original_record` is embedded. A named copy can therefore + overwrite a merged value. + """ + if merge_base is not None: + child_record = {**merge_base, **child_record} for parent_field in self.parent_fields or []: parent_field.copy_onto(parent_record, child_record) + if self.remain_original_record: + child_record["original_record"] = copy.deepcopy(parent_record) + return child_record def _carries_parent_context(self) -> bool: - return bool(self.remain_original_record or self.parent_fields) + return bool(self.remain_original_record or self.parent_fields or self.merge_parent) + + @classmethod + def _without_expanded_list( + cls, parent_record: Mapping[str, Any], expand_path: list[Any] + ) -> Mapping[str, Any]: + """Copy of the parent with only the value at `expand_path` removed. + + The containers on the way to the list are copied and the rest of the parent is shared, + so siblings of the list are kept and the parent itself is never mutated. Every match of + the path is removed, which covers glob paths that select several lists. + """ + matched_paths = [ + path + for path, _ in dpath.segments.walk(parent_record) # type: ignore[no-untyped-call] + if dpath.segments.match(path, expand_path) + ] + stripped: Any = parent_record + # Deepest and right-most first, so removing a list element never shifts the index of a + # match still to be removed, and a match nested inside another match goes first. + for path in sorted(matched_paths, key=cls._path_sort_key, reverse=True): + stripped = cls._without_path(stripped, path) + return dict(stripped) + + @staticmethod + def _path_sort_key(path: Sequence[Any]) -> list[tuple[int, Any]]: + # Integers and strings are not mutually comparable, so tag each segment by kind. + return [(0, segment) if isinstance(segment, int) else (1, str(segment)) for segment in path] + + @classmethod + def _without_path(cls, container: Any, path: Sequence[Any]) -> Any: + head, rest = path[0], path[1:] + if isinstance(container, Mapping): + copied: Any = dict(container) + elif isinstance(container, list): + copied = list(container) + else: + return container + if rest: + copied[head] = cls._without_path(copied[head], rest) + else: + del copied[head] + return copied diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 4345e40df2..8aab6fea73 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -2322,6 +2322,11 @@ class RecordExpander(BaseModel): description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", title="Parent Fields", ) + merge_parent: Optional[bool] = Field( + False, + description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + title="Merge Parent", + ) on_no_records: Optional[OnNoRecords] = Field( OnNoRecords.skip, description='Behavior when the expansion path is missing, not a list, or an empty list. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged.', 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 6dab0a84e9..f5242ef700 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -2556,6 +2556,7 @@ def create_record_expander( ] if model.parent_fields else None, + merge_parent=model.merge_parent or False, on_no_records=OnNoRecords(model.on_no_records.value) if model.on_no_records else OnNoRecords.skip, diff --git a/unit_tests/sources/declarative/expanders/test_record_expander.py b/unit_tests/sources/declarative/expanders/test_record_expander.py index 42fc37109a..0a7bdbe1e4 100644 --- a/unit_tests/sources/declarative/expanders/test_record_expander.py +++ b/unit_tests/sources/declarative/expanders/test_record_expander.py @@ -790,3 +790,184 @@ def test_parent_field_paths_interpolate_config(): parent_field.copy_onto({"url": "https://example.com/7"}, child) assert child == {"pull_request_url": "https://example.com/7"} + + +def _mailchimp_parent(): + return { + "email_id": "e1", + "list_id": "l1", + "activity": [ + {"action": "open", "timestamp": "t1"}, + {"action": "click", "timestamp": "t2"}, + ], + } + + +def _merge_expander(expand_records_from_field, **kwargs): + return RecordExpander( + expand_records_from_field=expand_records_from_field, + merge_parent=True, + config=config, + parameters=parameters, + **kwargs, + ) + + +def test_merge_parent_flattens_the_parent_into_every_item(): + """The source-mailchimp `email_activity` shape: `{**record, **activity_item}` per item.""" + expander = _merge_expander(["activity"]) + + records = list(expander.expand_record(_mailchimp_parent())) + + assert records == [ + {"email_id": "e1", "list_id": "l1", "action": "open", "timestamp": "t1"}, + {"email_id": "e1", "list_id": "l1", "action": "click", "timestamp": "t2"}, + ] + assert all("activity" not in record for record in records) + + +def test_merge_parent_lets_the_item_win_on_collision(): + expander = _merge_expander(["items"]) + + assert list(expander.expand_record({"id": "parent", "items": [{"id": "child"}]})) == [ + {"id": "child"} + ] + + +def test_merge_parent_removes_only_the_expanded_list_on_a_multi_segment_path(): + expander = _merge_expander(["reviews", "nodes"]) + parent = { + "number": 7, + "reviews": {"nodes": [{"id": "PRR_1"}], "totalCount": 1}, + } + + assert list(expander.expand_record(parent)) == [ + {"number": 7, "reviews": {"totalCount": 1}, "id": "PRR_1"} + ] + + +def test_merge_parent_removes_every_list_matched_by_a_glob_path(): + expander = _merge_expander(["sections", "*", "items"]) + parent = { + "id": 1, + "sections": { + "a": {"items": [{"n": 1}], "title": "A"}, + "b": {"items": [{"n": 2}], "title": "B"}, + }, + } + + assert list(expander.expand_record(parent)) == [ + {"id": 1, "sections": {"a": {"title": "A"}, "b": {"title": "B"}}, "n": 1}, + {"id": 1, "sections": {"a": {"title": "A"}, "b": {"title": "B"}}, "n": 2}, + ] + + +def test_merge_parent_removes_lists_nested_in_a_list_of_sections(): + expander = _merge_expander(["sections", "*", "items"]) + parent = {"sections": [{"items": [{"n": 1}], "k": "s0"}, {"items": [{"n": 2}], "k": "s1"}]} + + assert list(expander.expand_record(parent)) == [ + {"sections": [{"k": "s0"}, {"k": "s1"}], "n": 1}, + {"sections": [{"k": "s0"}, {"k": "s1"}], "n": 2}, + ] + + +def test_parent_fields_applied_after_merge_parent_overwrite_a_merged_value(): + expander = _merge_expander( + ["items"], + parent_fields=[_parent_field(["id"], ["id"])], + ) + + assert list(expander.expand_record({"id": "parent", "items": [{"id": "child"}]})) == [ + {"id": "parent"} + ] + + +def test_merge_parent_and_remain_original_record_are_independent(): + expander = _merge_expander(["activity"], remain_original_record=True) + + records = list(expander.expand_record(_mailchimp_parent())) + + assert records == [ + { + "email_id": "e1", + "list_id": "l1", + "action": "open", + "timestamp": "t1", + "original_record": _mailchimp_parent(), + }, + { + "email_id": "e1", + "list_id": "l1", + "action": "click", + "timestamp": "t2", + "original_record": _mailchimp_parent(), + }, + ] + + +def test_merge_parent_wraps_scalar_items(): + expander = _merge_expander(["items"]) + + assert list(expander.expand_record({"id": 9, "items": ["a", "b"]})) == [ + {"id": 9, "value": "a"}, + {"id": 9, "value": "b"}, + ] + + +def test_merge_parent_does_not_mutate_the_parent_record(): + expander = _merge_expander(["reviews", "nodes"]) + parent = _reviews_parent() + + records = list(expander.expand_record(parent)) + + assert parent == _reviews_parent() + assert records[0]["reviews"] == {} + + +def test_merge_parent_shares_untouched_nested_values_rather_than_deep_copying_them(): + expander = _merge_expander(["items"]) + metadata = {"k": "v"} + + records = list(expander.expand_record({"metadata": metadata, "items": [{"id": 1}]})) + + assert records[0]["metadata"] is metadata + + +def test_merge_parent_applies_to_items_fetched_by_the_truncated_list_retriever(): + expander = _merge_expander( + ["reviews", "nodes"], + truncation_indicator_path=["reviews", "has_more"], + truncated_list_retriever=_make_retriever([{"id": "PRR_9"}, "scalar"]), + ) + parent = { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}], "has_more": True}, + } + + assert list(expander.expand_record(parent)) == [ + { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"has_more": True}, + "id": "PRR_9", + }, + { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"has_more": True}, + "value": "scalar", + }, + ] + + +def test_merge_parent_unset_leaves_items_unchanged(): + expander = RecordExpander( + expand_records_from_field=["activity"], + config=config, + parameters=parameters, + ) + + assert expander.merge_parent is False + assert list(expander.expand_record(_mailchimp_parent())) == [ + {"action": "open", "timestamp": "t1"}, + {"action": "click", "timestamp": "t2"}, + ] 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 101f3105e0..dbd8b7f4d4 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 @@ -2084,6 +2084,93 @@ def test_create_record_expander_without_parent_fields_leaves_them_unset(): assert selector.extractor.record_expander.parent_fields is None +def test_create_record_expander_with_merge_parent(): + """The source-mailchimp `email_activity` shape: the parent flattened into each activity item.""" + content = """ + selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["emails"] + record_expander: + type: RecordExpander + expand_records_from_field: ["activity"] + merge_parent: true + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + expander = selector.extractor.record_expander + assert expander is not None + assert expander.merge_parent is True + assert expander.remain_original_record is False + assert expander.parent_fields is None + + response = requests.Response() + response._content = json.dumps( + { + "emails": [ + { + "email_id": "e1", + "list_id": "l1", + "activity": [ + {"action": "open", "timestamp": "t1"}, + {"action": "click", "timestamp": "t2"}, + ], + }, + {"email_id": "e2", "list_id": "l1", "activity": [{"action": "bounce"}]}, + ] + } + ).encode("utf-8") + + assert list(selector.extractor.extract_records(response)) == [ + {"email_id": "e1", "list_id": "l1", "action": "open", "timestamp": "t1"}, + {"email_id": "e1", "list_id": "l1", "action": "click", "timestamp": "t2"}, + {"email_id": "e2", "list_id": "l1", "action": "bounce"}, + ] + + +def test_create_record_expander_without_merge_parent_defaults_to_false(): + content = """ + selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["data"] + record_expander: + type: RecordExpander + expand_records_from_field: ["items"] + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + assert selector.extractor.record_expander.merge_parent is False + + def test_create_record_expander_with_truncated_list_retriever(): content = """ selector: From 606c32f3f8aa0b80f30d82cea8017b8151219e95 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 16 Sep 2026 18:53:41 +0300 Subject: [PATCH 3/5] fix(RecordExpander): isolate copied parent context from the parent and sibling items `merge_parent` shallow-merged the parent, so every nested container of the parent was aliased into every expanded item and into the parent itself. The same held for a `parent_fields` entry whose `parent_path` pointed at a dict or a list, because `dpath.get` returns the parent's object by reference. A downstream `AddFields`/`RemoveFields` with a multi-segment path then wrote into records already yielded, and into the record being expanded. `merge_base` is now deep-copied per item, and a container copied by `ParentFieldPath.copy_onto` is deep-copied as well; scalars pass through untouched. `merge_base` still excludes the expanded list and is still built once per parent record, so the copy stays cheaper than `remain_original_record`. Also: - register `RecordExpander.parent_fields` in `DEFAULT_MODEL_TYPES`, so a manifest may omit `type: ParentFieldPath` as it may for `AddFields.fields` - drop the "when truncation handling is configured" clause from the shared glob-rejection message, which `parent_fields` paths also reach - document `dpath.new` behaviour on `record_path` and the per-item copying in the schema descriptions and docstrings Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 6 +-- .../declarative/expanders/record_expander.py | 49 ++++++++++++----- .../models/declarative_component_schema.py | 6 +-- .../parsers/manifest_component_transformer.py | 2 + .../expanders/test_record_expander.py | 39 ++++++++++++-- .../test_model_to_component_factory.py | 52 +++++++++++++++++++ 6 files changed, 131 insertions(+), 23 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 4ef4215e5d..7f3ccd845f 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -2252,13 +2252,13 @@ definitions: default: false parent_fields: title: Parent Fields - description: 'Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.' + description: 'Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.' type: array items: "$ref": "#/definitions/ParentFieldPath" merge_parent: title: Merge Parent - description: If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it. + description: If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it. type: boolean default: false on_no_records: @@ -2314,7 +2314,7 @@ definitions: - ["repository", "name"] record_path: title: Record Path - description: Path on the expanded item to write the value to. An existing value is overwritten. + description: Path on the expanded item to write the value to. An existing value is overwritten. Intermediate objects are created as needed, so a path may not pass through a value the item already holds as a scalar, and a numeric path segment creates an array rather than an object - the same behavior as `AddFields`. type: array items: type: string diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py index 4dfd6b0caa..2f3de73cf6 100644 --- a/airbyte_cdk/sources/declarative/expanders/record_expander.py +++ b/airbyte_cdk/sources/declarative/expanders/record_expander.py @@ -41,6 +41,13 @@ ) +def _detached(value: Any) -> Any: + """Copy of `value` sharing nothing with the record it came from; scalars are returned as-is.""" + if isinstance(value, (Mapping, list, set, tuple)): + return copy.deepcopy(value) + return value + + class OnNoRecords(Enum): """ Behavior when record expansion produces no records. @@ -93,12 +100,16 @@ def copy_onto( A `parent_path` that the parent does not have copies `None`, which is what the custom connector classes this replaces do (`parent.get(field)`). Distinguishing "absent" from "present and null" would need a third option and no connector needs one. + + A container value is deep-copied, so that a downstream transformation writing inside it + cannot reach the parent record or the items expanded from it alongside this one. The copy + costs proportionally to the named value, not to the whole parent. """ try: value = dpath.get(dict(parent_record), self.evaluated_parent_path()) except (KeyError, ValueError): value = None - dpath.new(child_record, self.evaluated_record_path(), value) + dpath.new(child_record, self.evaluated_record_path(), _detached(value)) @dataclass @@ -170,17 +181,20 @@ class RecordExpander: parent_fields: Named values to copy from the record being expanded onto each expanded item. Each entry has a `parent_path` and a `record_path`; an existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies - `None`. Independent of `remain_original_record` - both may be set, though copying - named fields is the cheaper way to carry parent context. Applies to items fetched - through `truncated_list_retriever` as well as to embedded ones. Glob metacharacters - are rejected in both paths. + `None`. A copied container is deep-copied, so writing into it downstream does not + reach the parent or the sibling items. Independent of `remain_original_record` - + both may be set, though copying named fields is the cheaper way to carry parent + context. Applies to items fetched through `truncated_list_retriever` as well as to + embedded ones. Glob metacharacters are rejected in both paths. merge_parent: If True, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed: for a - multi-segment path the top-level key stays with its other fields. The merge happens - before `parent_fields` are copied and before `original_record` is embedded, so both - can overwrite merged values. Applies to items fetched through - `truncated_list_retriever` as well as to embedded ones. Defaults to False. + multi-segment path the top-level key stays with its other fields. Each item gets its + own deep copy of the merged parent, so a downstream transformation writing into a + nested value affects only that item. The merge happens before `parent_fields` are + copied and before `original_record` is embedded, so both can overwrite merged + values. Applies to items fetched through `truncated_list_retriever` as well as to + embedded ones. Defaults to False. on_no_records: Behavior when expansion produces no records. "skip" (default) emits nothing. "emit_parent" emits the original parent record unchanged. truncation_indicator_path: Path within each record to a field indicating that the @@ -254,8 +268,8 @@ def _reject_globs(path: Sequence[Any], field_name: str) -> None: for segment in path ): raise ValueError( - f"Glob characters {_GLOB_METACHARACTERS} are not supported in `{field_name}` when " - "truncation handling is configured: the path must identify a single field." + f"Glob characters {_GLOB_METACHARACTERS} are not supported in `{field_name}`: " + "the path must identify a single field." ) def _evaluated_indicator_path(self) -> list[Any]: @@ -436,7 +450,12 @@ def _apply_parent_context( overwrite a merged value. """ if merge_base is not None: - child_record = {**merge_base, **child_record} + # Deep-copied per item: a shallow merge would alias every nested container of the + # parent into every item, so a downstream transformation writing into one of them + # would reach the parent record and the items already yielded. The expanded list is + # already stripped from `merge_base`, so the copy costs less than + # `remain_original_record` does. + child_record = {**copy.deepcopy(merge_base), **child_record} for parent_field in self.parent_fields or []: parent_field.copy_onto(parent_record, child_record) if self.remain_original_record: @@ -453,8 +472,10 @@ def _without_expanded_list( """Copy of the parent with only the value at `expand_path` removed. The containers on the way to the list are copied and the rest of the parent is shared, - so siblings of the list are kept and the parent itself is never mutated. Every match of - the path is removed, which covers glob paths that select several lists. + so siblings of the list are kept and the parent itself is not mutated here. This base is + built once per parent record and then deep-copied per item in `_apply_parent_context`, + which is what keeps the parent and the sibling items isolated from one another. Every + match of the path is removed, which covers glob paths that select several lists. """ matched_paths = [ path diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 8aab6fea73..6c944d8679 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -532,7 +532,7 @@ class ParentFieldPath(BaseModel): ) record_path: List[str] = Field( ..., - description="Path on the expanded item to write the value to. An existing value is overwritten.", + description="Path on the expanded item to write the value to. An existing value is overwritten. Intermediate objects are created as needed, so a path may not pass through a value the item already holds as a scalar, and a numeric path segment creates an array rather than an object - the same behavior as `AddFields`.", examples=[["pull_request_url"], ["comment_id"], ["parent", "id"]], title="Record Path", ) @@ -2319,12 +2319,12 @@ class RecordExpander(BaseModel): ) parent_fields: Optional[List[ParentFieldPath]] = Field( None, - description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + description="Named values to copy from the record being expanded onto each expanded item. Use this instead of `remain_original_record` when only a few parent fields are needed: it copies the named values rather than deep-copying the whole parent once per item, which matters when the parent record is large and the nested list is long. An existing value at `record_path` is overwritten, and a `parent_path` the parent does not have copies null. A copied object or array is deep-copied, so a transformation that writes inside it affects only that item. Independent of `remain_original_record`; both may be set. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", title="Parent Fields", ) merge_parent: Optional[bool] = Field( False, - description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", + description="If true, each expanded item is the parent record shallow-merged with the item, the item's own keys winning on collision, and the expanded list removed from the parent's copy. Only the value at `expand_records_from_field` is removed, so for a multi-segment path the top-level key stays with its other fields. Each item receives its own deep copy of the merged parent fields, so a transformation that writes into a nested value affects only that item. The merge happens first, then `parent_fields` are copied, then `original_record` is embedded when `remain_original_record` is set; all three may be combined. Applies to items fetched through `truncated_list_retriever` as well as to embedded ones. This field is ignored by CDK versions that predate it, so pin the connector to a CDK version that supports it.", title="Merge Parent", ) on_no_records: Optional[OnNoRecords] = Field( diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py index 00b1a18ffb..e11a0af515 100644 --- a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py +++ b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py @@ -50,6 +50,8 @@ # ParentStreamConfig "ParentStreamConfig.request_option": "RequestOption", "ParentStreamConfig.stream": "DeclarativeStream", + # RecordExpander + "RecordExpander.parent_fields": "ParentFieldPath", # RecordSelector "RecordSelector.extractor": "DpathExtractor", "RecordSelector.record_filter": "RecordFilter", diff --git a/unit_tests/sources/declarative/expanders/test_record_expander.py b/unit_tests/sources/declarative/expanders/test_record_expander.py index 0a7bdbe1e4..26269e9f63 100644 --- a/unit_tests/sources/declarative/expanders/test_record_expander.py +++ b/unit_tests/sources/declarative/expanders/test_record_expander.py @@ -772,6 +772,14 @@ def test_globs_are_rejected_in_record_path(path): _parent_field(["id"], path) +def test_glob_rejection_message_does_not_mention_truncation_handling(): + with pytest.raises(ValueError) as error: + _parent_field(["*"], ["copied"]) + + assert "truncation" not in str(error.value) + assert "`parent_path`" in str(error.value) + + @pytest.mark.parametrize("parent_path,record_path", [([], ["a"]), (["a"], [])]) def test_empty_paths_are_rejected(parent_path, record_path): with pytest.raises(ValueError, match="cannot be empty"): @@ -925,13 +933,38 @@ def test_merge_parent_does_not_mutate_the_parent_record(): assert records[0]["reviews"] == {} -def test_merge_parent_shares_untouched_nested_values_rather_than_deep_copying_them(): +def test_merge_parent_gives_each_item_its_own_copy_of_nested_values(): expander = _merge_expander(["items"]) metadata = {"k": "v"} + parent = {"metadata": metadata, "items": [{"id": 1}, {"id": 2}]} + + records = list(expander.expand_record(parent)) + + assert records[0]["metadata"] is not metadata + assert records[0]["metadata"] is not records[1]["metadata"] + # A downstream transformation writing into one item cannot reach the parent or its siblings. + records[0]["metadata"]["written"] = True + assert records[1]["metadata"] == {"k": "v"} + assert metadata == {"k": "v"} + - records = list(expander.expand_record({"metadata": metadata, "items": [{"id": 1}]})) +def test_parent_fields_gives_each_item_its_own_copy_of_a_copied_container(): + expander = RecordExpander( + expand_records_from_field=["reviews", "nodes"], + parent_fields=[_parent_field(["repository"], ["repository"])], + config=config, + parameters=parameters, + ) + repository = {"name": "airbyte"} + parent = {"repository": repository, "reviews": {"nodes": [{"id": 1}, {"id": 2}]}} + + records = list(expander.expand_record(parent)) - assert records[0]["metadata"] is metadata + assert records[0]["repository"] is not repository + assert records[0]["repository"] is not records[1]["repository"] + records[0]["repository"]["review_id"] = 1 + assert records[1]["repository"] == {"name": "airbyte"} + assert repository == {"name": "airbyte"} def test_merge_parent_applies_to_items_fetched_by_the_truncated_list_retriever(): 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 dbd8b7f4d4..a1f612e479 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 @@ -2055,6 +2055,58 @@ def test_create_record_expander_with_parent_fields(): ] +def test_create_record_expander_with_parent_fields_without_explicit_type(): + """`type: ParentFieldPath` is optional: the transformer fills it in, as it does for AddFields.""" + content = """ + selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["nodes"] + record_expander: + type: RecordExpander + expand_records_from_field: ["reviews", "nodes"] + parent_fields: + - parent_path: ["url"] + record_path: ["pull_request_url"] + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + assert ( + selector_manifest["extractor"]["record_expander"]["parent_fields"][0]["type"] + == "ParentFieldPath" + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + response = requests.Response() + response._content = json.dumps( + { + "nodes": [ + { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}]}, + } + ] + } + ).encode("utf-8") + + assert list(selector.extractor.extract_records(response)) == [ + {"id": "PRR_1", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + ] + + def test_create_record_expander_without_parent_fields_leaves_them_unset(): content = """ selector: From 92daa4ebb93611df582a2188fcd1c751b18e4166 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Mon, 28 Sep 2026 18:42:15 +0300 Subject: [PATCH 4/5] fix(RecordExpander): make parent_fields lookup linear and harden merge_parent stripping - Look up `parent_path` by walking only the containers on the path instead of calling `dpath.get`, which folds over the whole parent (the expanded list included) for every item and every field. The expansion was quadratic in the list length and slower than `remain_original_record`. Paths are evaluated once at construction. - Raise a config-error `AirbyteTracedException` naming the `record_path` when it runs through a value the item holds as a scalar or a list, instead of a raw `PathNotFound`/`TypeError`. - `merge_parent` strips only the list matches of a glob path; non-list matches were never expanded and are kept. - Pin the strip order, `$parameters` propagation into `parent_fields`, and the merge-then-copy order with tests. Co-Authored-By: Claude Opus 5.5 --- .../declarative/expanders/record_expander.py | 71 ++++++++-- .../expanders/test_record_expander.py | 132 +++++++++++++++++- .../test_model_to_component_factory.py | 49 +++++++ 3 files changed, 240 insertions(+), 12 deletions(-) diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py index 2f3de73cf6..19693e2529 100644 --- a/airbyte_cdk/sources/declarative/expanders/record_expander.py +++ b/airbyte_cdk/sources/declarative/expanders/record_expander.py @@ -17,12 +17,14 @@ AirbyteMessage, AirbyteStateMessage, AirbyteTraceMessage, + FailureType, Level, ) from airbyte_cdk.models import Type as MessageType from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString from airbyte_cdk.sources.message import MessageRepository from airbyte_cdk.sources.types import Config, Record, StreamSlice +from airbyte_cdk.utils.traced_exception import AirbyteTracedException if TYPE_CHECKING: from airbyte_cdk.sources.declarative.retrievers import Retriever @@ -32,6 +34,9 @@ # dpath treats these characters as glob metacharacters (fnmatch semantics) inside a path segment. _GLOB_METACHARACTERS = ("*", "?", "[") +# Returned by `_lookup` for a path the record does not have, since `None` is a valid value. +_MISSING = object() + # Protocol payloads a retriever may yield unwrapped (i.e. not inside an `AirbyteMessage` envelope). _BARE_PROTOCOL_MESSAGES = ( AirbyteControlMessage, @@ -41,6 +46,34 @@ ) +def _lookup(record: Any, path: Sequence[Any]) -> Any: + """Value at a glob-free `path`, or `_MISSING`, reading only the containers on the path. + + Matches the way `dpath.get` resolves such a path: a mapping key is compared as a string, and + a segment reaching a list is taken as an index, negative ones counting from the end. + `dpath.get` itself walks the whole record, so on a parent holding a long list each lookup + costs as much as that list. + """ + value = record + for segment in path: + if isinstance(value, Mapping): + key = str(segment) + if key not in value: + return _MISSING + value = value[key] + elif isinstance(value, list): + try: + index = int(segment) + except (TypeError, ValueError): + return _MISSING + if not -len(value) <= index < len(value): + return _MISSING + value = value[index] + else: + return _MISSING + return value + + def _detached(value: Any) -> Any: """Copy of `value` sharing nothing with the record it came from; scalars are returned as-is.""" if isinstance(value, (Mapping, list, set, tuple)): @@ -83,8 +116,11 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._record_path: list[InterpolatedString] = [ InterpolatedString.create(path, parameters=parameters) for path in self.record_path ] - RecordExpander._reject_globs(self.evaluated_parent_path(), "parent_path") - RecordExpander._reject_globs(self.evaluated_record_path(), "record_path") + # The paths only interpolate `config`, so they are evaluated once instead of per item. + self._evaluated_parent_path = self.evaluated_parent_path() + self._evaluated_record_path = self.evaluated_record_path() + RecordExpander._reject_globs(self._evaluated_parent_path, "parent_path") + RecordExpander._reject_globs(self._evaluated_record_path, "record_path") def evaluated_parent_path(self) -> list[Any]: return [segment.eval(self.config) for segment in self._parent_path] @@ -102,14 +138,26 @@ def copy_onto( "present and null" would need a third option and no connector needs one. A container value is deep-copied, so that a downstream transformation writing inside it - cannot reach the parent record or the items expanded from it alongside this one. The copy - costs proportionally to the named value, not to the whole parent. + cannot reach the parent record or the items expanded from it alongside this one. Both the + lookup and the copy cost proportionally to the named value, not to the whole parent. + + A `record_path` running through a value the item holds as a scalar or a list cannot be + written and fails the sync as a config error naming the path. """ + value = _lookup(parent_record, self._evaluated_parent_path) + copied = None if value is _MISSING else _detached(value) try: - value = dpath.get(dict(parent_record), self.evaluated_parent_path()) - except (KeyError, ValueError): - value = None - dpath.new(child_record, self.evaluated_record_path(), _detached(value)) + dpath.new(child_record, self._evaluated_record_path, copied) + except (dpath.exceptions.PathNotFound, TypeError) as error: + message = ( + f"Cannot write parent field to `record_path` {self._evaluated_record_path}: " + "the expanded item already holds a non-object value on that path." + ) + raise AirbyteTracedException( + message=message, + internal_message=f"{message} {error!r}", + failure_type=FailureType.config_error, + ) from error @dataclass @@ -475,12 +523,13 @@ def _without_expanded_list( so siblings of the list are kept and the parent itself is not mutated here. This base is built once per parent record and then deep-copied per item in `_apply_parent_context`, which is what keeps the parent and the sibling items isolated from one another. Every - match of the path is removed, which covers glob paths that select several lists. + list matched by the path is removed, which covers glob paths that select several lists; + a match that is not a list is kept, since nothing is expanded from it. """ matched_paths = [ path - for path, _ in dpath.segments.walk(parent_record) # type: ignore[no-untyped-call] - if dpath.segments.match(path, expand_path) + for path, value in dpath.segments.walk(parent_record) # type: ignore[no-untyped-call] + if isinstance(value, list) and dpath.segments.match(path, expand_path) ] stripped: Any = parent_record # Deepest and right-most first, so removing a list element never shifts the index of a diff --git a/unit_tests/sources/declarative/expanders/test_record_expander.py b/unit_tests/sources/declarative/expanders/test_record_expander.py index 26269e9f63..d73a0f1b9e 100644 --- a/unit_tests/sources/declarative/expanders/test_record_expander.py +++ b/unit_tests/sources/declarative/expanders/test_record_expander.py @@ -5,15 +5,24 @@ import threading from unittest.mock import MagicMock +import dpath import pytest -from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, AirbyteRecordMessage, Level, Type +from airbyte_cdk.models import ( + AirbyteLogMessage, + AirbyteMessage, + AirbyteRecordMessage, + FailureType, + Level, + Type, +) from airbyte_cdk.sources.declarative.expanders.record_expander import ( ParentFieldPath, RecordExpander, ) from airbyte_cdk.sources.message import InMemoryMessageRepository from airbyte_cdk.sources.types import Record, StreamSlice +from airbyte_cdk.utils.traced_exception import AirbyteTracedException config = {} parameters = {} @@ -800,6 +809,82 @@ def test_parent_field_paths_interpolate_config(): assert child == {"pull_request_url": "https://example.com/7"} +@pytest.mark.parametrize( + "parent,parent_path", + [ + ({"a": {"b": 1}}, ["a", "b"]), + ({"a": {"b": None}}, ["a", "b"]), + ({"a": {"b": {"c": [1, 2]}}}, ["a", "b"]), + ({"a": [{"b": 1}, {"b": 2}]}, ["a", "1", "b"]), + ({"a": [{"b": 1}, {"b": 2}]}, ["a", "01", "b"]), + ({"a": [{"b": 1}]}, ["a", "1", "b"]), + ({"a": [{"b": 1}, {"b": 2}]}, ["a", "-1", "b"]), + ({"a": [{"b": 1}]}, ["a", "-2", "b"]), + ({"a": [{"b": 1}]}, ["a", "x"]), + ({"a": {"0": "zero"}}, ["a", "0"]), + ({"a": "text"}, ["a", "0"]), + ({"a": 1}, ["a", "b"]), + ({}, ["a"]), + ], +) +def test_parent_field_lookup_matches_dpath_get(parent, parent_path): + """The direct lookup replacing `dpath.get` resolves every path the way `dpath.get` did.""" + try: + expected = dpath.get(parent, parent_path) + except (KeyError, ValueError): + expected = None + child = {} + + _parent_field(parent_path, ["copied"]).copy_onto(parent, child) + + assert child == {"copied": expected} + + +def test_parent_field_lookup_does_not_walk_the_parent(monkeypatch): + """Walking the whole parent per item made `parent_fields` quadratic in the list length.""" + + def fail(*args, **kwargs): + raise AssertionError("the parent was walked") + + monkeypatch.setattr(dpath, "get", fail) + expander = RecordExpander( + expand_records_from_field=["items"], + parent_fields=[_parent_field(["meta", "url"], ["url"])], + config=config, + parameters=parameters, + ) + parent = {"meta": {"url": "u"}, "items": [{"n": n} for n in range(3)]} + + assert list(expander.expand_record(parent)) == [ + {"n": 0, "url": "u"}, + {"n": 1, "url": "u"}, + {"n": 2, "url": "u"}, + ] + + +def test_parent_field_paths_interpolate_parameters(): + parent_field = ParentFieldPath( + parent_path=["{{ parameters['from'] }}"], + record_path=["{{ parameters['to'] }}"], + config=config, + parameters={"from": "url", "to": "pull_request_url"}, + ) + child = {} + + parent_field.copy_onto({"url": "https://example.com/7"}, child) + + assert child == {"pull_request_url": "https://example.com/7"} + + +@pytest.mark.parametrize("blocking_value", [1, [1]], ids=["scalar", "list"]) +def test_record_path_through_a_non_object_value_raises_a_config_error(blocking_value): + with pytest.raises(AirbyteTracedException) as error: + _parent_field(["id"], ["q", "x"]).copy_onto({"id": 1}, {"q": blocking_value}) + + assert error.value.failure_type == FailureType.config_error + assert "['q', 'x']" in error.value.message + + def _mailchimp_parent(): return { "email_id": "e1", @@ -880,6 +965,51 @@ def test_merge_parent_removes_lists_nested_in_a_list_of_sections(): ] +def test_merge_parent_removes_several_lists_from_the_same_list(): + """Removal runs right-most first; left to right, the second index would have shifted.""" + expander = _merge_expander(["sec", "*"]) + parent = {"id": 1, "sec": [[{"n": 1}], [{"n": 2}], [{"n": 3}]]} + + assert list(expander.expand_record(parent)) == [ + {"id": 1, "sec": [], "n": 1}, + {"id": 1, "sec": [], "n": 2}, + {"id": 1, "sec": [], "n": 3}, + ] + + +def test_merge_parent_removes_a_list_matched_inside_another_matched_list(): + """Removal runs deepest first, so the outer list is still there when the inner one goes.""" + expander = _merge_expander(["**"]) + parent = {"outer": [{"n": 1}, [{"n": 2}]]} + merge_base = expander._without_expanded_list(parent, ["**"]) + + assert merge_base == {} + assert parent == {"outer": [{"n": 1}, [{"n": 2}]]} + + +def test_merge_parent_keeps_glob_matches_that_are_not_lists(): + expander = _merge_expander(["sec", "*"]) + parent = {"sec": [[{"n": 1}], "s", [{"n": 2}]]} + + assert list(expander.expand_record(parent)) == [ + {"sec": ["s"], "n": 1}, + {"sec": ["s"], "n": 2}, + ] + + +def test_parent_fields_write_into_a_merged_nested_value(): + """Merge first, then the named copy: the copy lands inside the merged object, not over it.""" + expander = _merge_expander( + ["items"], + parent_fields=[_parent_field(["id"], ["meta", "parent_id"])], + ) + parent = {"id": 7, "meta": {"a": 1}, "items": [{"n": 1}]} + + assert list(expander.expand_record(parent)) == [ + {"id": 7, "meta": {"a": 1, "parent_id": 7}, "n": 1}, + ] + + def test_parent_fields_applied_after_merge_parent_overwrite_a_merged_value(): expander = _merge_expander( ["items"], 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 71bf059192..af5be7167c 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 @@ -2118,6 +2118,55 @@ def test_create_record_expander_with_parent_fields_without_explicit_type(): ] +def test_create_record_expander_propagates_parameters_into_parent_fields(): + content = """ + selector: + type: RecordSelector + $parameters: + parent_key: "url" + child_key: "pull_request_url" + extractor: + type: DpathExtractor + field_path: ["nodes"] + record_expander: + type: RecordExpander + expand_records_from_field: ["reviews", "nodes"] + parent_fields: + - parent_path: ["{{ parameters['parent_key'] }}"] + record_path: ["{{ parameters['child_key'] }}"] + """ + parsed_manifest = YamlDeclarativeSource._parse(content) + resolved_manifest = resolver.preprocess_manifest(parsed_manifest) + selector_manifest = transformer.propagate_types_and_parameters( + "", resolved_manifest["selector"], {} + ) + + selector = factory.create_component( + model_type=RecordSelectorModel, + name="test_stream", + component_definition=selector_manifest, + decoder=None, + transformations=[], + config=input_config, + ) + + response = requests.Response() + response._content = json.dumps( + { + "nodes": [ + { + "url": "https://github.com/airbytehq/airbyte/pull/7", + "reviews": {"nodes": [{"id": "PRR_1"}]}, + } + ] + } + ).encode("utf-8") + + assert list(selector.extractor.extract_records(response)) == [ + {"id": "PRR_1", "pull_request_url": "https://github.com/airbytehq/airbyte/pull/7"}, + ] + + def test_create_record_expander_without_parent_fields_leaves_them_unset(): content = """ selector: From eba17fbfff503019be846bcde4f40c143d63f6c8 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Mon, 28 Sep 2026 19:36:05 +0300 Subject: [PATCH 5/5] fix(RecordExpander): append parent_fields and merge_parent after the existing fields Inserting them after `remain_original_record` shifted every later positional argument of the `RecordExpander` constructor. No caller constructs it positionally, but appending them keeps the change strictly additive; a test pins the order. Co-Authored-By: Claude Opus 5.5 --- .../sources/declarative/expanders/record_expander.py | 5 +++-- .../declarative/expanders/test_record_expander.py | 10 ++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/expanders/record_expander.py b/airbyte_cdk/sources/declarative/expanders/record_expander.py index 19693e2529..19b9bd383f 100644 --- a/airbyte_cdk/sources/declarative/expanders/record_expander.py +++ b/airbyte_cdk/sources/declarative/expanders/record_expander.py @@ -280,13 +280,14 @@ class RecordExpander: config: Config parameters: InitVar[Mapping[str, Any]] remain_original_record: bool = False - parent_fields: Optional[Sequence[ParentFieldPath]] = None - merge_parent: bool = False on_no_records: OnNoRecords = OnNoRecords.skip truncation_indicator_path: Optional[Sequence[str]] = None truncated_list_retriever: Optional["Retriever"] = None message_repository: Optional[MessageRepository] = None suppress_incomplete_fetch_warning: bool = False + # Appended after the pre-existing fields so positional construction keeps its meaning. + parent_fields: Optional[Sequence[ParentFieldPath]] = None + merge_parent: bool = False def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._expand_path: list[InterpolatedString] = [ diff --git a/unit_tests/sources/declarative/expanders/test_record_expander.py b/unit_tests/sources/declarative/expanders/test_record_expander.py index d73a0f1b9e..84e3259340 100644 --- a/unit_tests/sources/declarative/expanders/test_record_expander.py +++ b/unit_tests/sources/declarative/expanders/test_record_expander.py @@ -17,6 +17,7 @@ Type, ) from airbyte_cdk.sources.declarative.expanders.record_expander import ( + OnNoRecords, ParentFieldPath, RecordExpander, ) @@ -885,6 +886,15 @@ def test_record_path_through_a_non_object_value_raises_a_config_error(blocking_v assert "['q', 'x']" in error.value.message +def test_new_fields_do_not_shift_positional_arguments(): + """`parent_fields` and `merge_parent` come after the fields that predate them.""" + expander = RecordExpander(["items"], config, parameters, False, OnNoRecords.emit_parent) + + assert expander.on_no_records == OnNoRecords.emit_parent + assert expander.parent_fields is None + assert expander.merge_parent is False + + def _mailchimp_parent(): return { "email_id": "e1",