Repository navigation
[DO NOT MERGE] prerelease: CombinedExtractor (#1162) + RecordExpander parent_fields (#1165) - #1175
Closed
Anatolii Yatsuk (tolik0) wants to merge 16 commits into
Closed
Anatolii Yatsuk (tolik0) wants to merge 16 commits into
Anatolii Yatsuk (tolik0) wants to merge 16 commits into
Conversation
A DpathExtractor can only describe one path into a response. Three connectors need more than that and hand-roll it in a custom components.py today, which keeps them out of the Connector Builder and out of manifest-only packaging: - source-google-analytics-data-api ships its own class literally named CombinedExtractor (components.py:17) that zip-merges the i-th record of each sub-extractor - the `zip_merge` mode here. - source-monday's MondayIncrementalItemsExtractor (components.py:75) falls back to a second field path only when the first one produced nothing - the `first_match` mode here. - source-github's in-progress GraphQL extractors read records from several sibling paths of one document - the `union` mode here. Their deep traversal stays out of scope; a separate paginator PR covers it. `zip_merge` truncates at the shortest sub-extractor instead of padding, matching GA4's existing zip() so that connector can drop its custom class without a data change. `first_match` decides "produced at least one record" by peeking a single element and chaining it back, so no sub-extractor is materialized and the peeked record is still emitted. All sub-extractors read the same requests.Response. That is safe for the buffered decoders (JsonDecoder, XmlDecoder, ZipfileDecoder) and not for the streaming ones (CompositeRawDecoder with stream_response=True, i.e. CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder outside the Connector Builder). The limitation is documented in the docstring and in the schema, and pinned by a test. The generated pydantic model was updated by hand on purpose: a local `poe assemble` on arm64 rewrites ~340 unrelated lines relative to main (conint/ confloat where main uses Field(..., ge=N), plus unrelated class reordering), so only the delta this schema change implies was applied. A maintainer should run /poe build to regenerate the models authoritatively before merge. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…address review The streaming-decoder hazard was documented as a loud `ValueError`, but that is only what `io.BytesIO` does. In production `requests` puts a `urllib3.HTTPResponse` in `response.raw`, whose read on a closed body returns `b""`, so a `CombinedExtractor` over `JsonlDecoder` or `GzipDecoder` silently drops records: `union` emits only the first sub-extractor's output and `first_match` emits nothing when the first path misses. The Connector Builder downgrades those decoders to `stream_response=False`, so the manifest test-reads green and loses data once published, with the cursor advancing regardless. - `create_combined_extractor` now raises an `AirbyteTracedException` (`config_error`) when the stream's decoder streams the response, naming the decoders and the buffered alternatives. It also rejects the downgraded `CompositeRawDecoder` instances the Builder substitutes, so the Builder rejects exactly what production rejects, and it fires for nested `CombinedExtractor`s, which inherit the retriever's decoder. - `test_streaming_decoder_is_a_known_limitation` is replaced by two tests built on a real `urllib3.HTTPResponse`, pinning the silent loss in `union` and the silent wrong answer in `first_match` against a buffered control, plus parametrized factory tests for the parse-time rejection. `minItems: 1` is added to `extractors`. The stated blocker did not exist: datamodel-codegen silently drops `minItems` on this self-referencing construct at the pinned 0.26.3, so pydantic never sees it, while manifest validation is jsonschema and does enforce it. The `__post_init__` guard stays as the backstop. The hand-written generated enum is renamed `CombineMode` -> `Mode` to match what the pinned generator emits (it derives an inline property enum's name from the property name, not from `title`), so a `/poe build` regen is not a surprise diff. The runtime enum stays `CombineMode`. Also from the review: - Three manifest-level tests through `ConcurrentDeclarativeSource` so the 59 new schema lines are covered by jsonschema validation rather than `parse_obj`. - `zip_merge` logs a warning when the sub-extractors have different lengths, and raises an error naming the mode and sub-extractor index when one yields a non-object instead of a bare `dictionary update sequence` built-in error. - Docstring and schema descriptions rewritten: the limitation is silent, not loud, and is now rejected; the per-sub-extractor decode cost and the wrong paginator record count are documented. The docstring uses headings so pdoc no longer renders the warning as a code block. - `AsyncHttpJobRepository.status_extractor` / `download_target_extractor` and the three factory `extractor_model` parameters widened to what they receive. - `CombinedExtractor` and `CombineMode` exported from `airbyte_cdk`. - Coverage for the four widened unions outside `RecordSelector` and for `CombinedExtractor` behind an `OffsetIncrement` paginator. Backward compatibility re-verified: 531 monorepo manifests validate identically against the base and head schemas, and 5,818 extractor payloads resolve identically through the base and head model modules. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…de enum - Reject a `union` CombinedExtractor (nested ones included) under an `OffsetIncrement` paginator with a config error: the offset is advanced by the summed record count of all sub-extractors, which overshoots the API page size and silently skips records. - Correct the pagination caveat in the `mode` schema description and in the component docstring: `union` is the harmful mode, `first_match` and `zip_merge` report the winning/shortest sub-extractor's count. - Narrow the Connector Builder half of the streaming-decoder guard to the four decoders the Builder actually downgrades (exact `CompositeRawDecoder` class with a streaming parser), so a buffered subclass is no longer rejected. - Move the `mode` enum into its own `CombineMode` schema definition so codegen emits `CombineMode` instead of the very generic top-level `Mode`; the generated model matches `datamodel_code_generator==0.26.3` output. - Keep both new user-facing messages to the violated condition and move the remediation into `internal_message`, per the error-message guidelines. - Ignore the `test_response.csv` fixture that `test_response_to_file_extractor.py` writes under `unit_tests/sources/declarative/extractors/`, mirroring the `decoders/` rule. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…n fall through a page of nulls
`source-monday`'s `MondayIncrementalItemsExtractor` was described as a motivating
connector for this component but was not a drop-in adopter, for one reason: it
drops falsy entries before deciding whether the primary path produced anything
(`components.py:106-124`). A GraphQL partial response puts `null` in the record
list and explains itself in a sibling `errors` field, and monday has to fall
through to `field_path_pagination` when a page is all nulls.
`first_match` counted a `None` as a record, so the primary path won with nothing
usable and the fallback was never tried. `skip_empty_records` drops falsy records
- `None`, `{}`, `[]`, `""` - before any mode sees them, which both keeps them out
of the record stream and takes them out of the match decision. It is off by
default, so existing manifests are unaffected.
The flag applies to all three modes. Under `zip_merge` it shifts the alignment of
the merged records, because dropping the i-th record of one sub-extractor pairs
its next record with the i-th record of the others; the schema description and a
test say so rather than leaving it to be discovered.
One divergence from monday remains and is deliberate: monday's warning names
`body.get("errors")`, which assumes a GraphQL error envelope. The component logs
the dropped count and the sub-extractor that produced them instead.
The generated model was hand-applied and then verified against the pinned
generator: the `CombinedExtractor` block is byte-identical to `poe assemble`
output once `ruff format` is applied.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The architecture review of this PR found one adopter for `zip_merge` (source-google-analytics-data-api), and that connector keeps its `components.py` in every scenario because both of its inner extractors are a custom `KeyValueExtractor`. `union` and `first_match` each have two adopters, so they stay; `zip_merge` was fitted to a single connector that it does not free. Removes the mode from the enum, the schema, the generated models, the component and its tests, along with the shortest-wins truncation warning, the Mapping-only merge check and the `skip_empty_records` alignment caveat that only applied to it. The mode can come back when a second columnar API turns up, ideally alongside a key/value-list-to-object transformation that would free GA4's inner extractor too. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… 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) <noreply@anthropic.com>
…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 <noreply@anthropic.com>
…d 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) <noreply@anthropic.com>
…dk/combined-extractor-with-parent-fields
Contributor
Author
|
/prerelease
|
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. 💡 Show Tips and TricksTesting This CDK VersionYou can test this version of the CDK using the following: # Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@tolik0/cdk/combined-extractor-with-parent-fields#egg=airbyte-python-cdk[dev]' --help
# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch tolik0/cdk/combined-extractor-with-parent-fieldsPR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
|
2 tasks
Anatolii Yatsuk (tolik0)
added a commit
to airbytehq/airbyte
that referenced
this pull request
Sep 23, 2026
…-cdk#1175 prerelease #1175 merges #1162 (`CombinedExtractor`) and #1165 (`RecordExpander.parent_fields`) into one build. Pin `airbyte-cdk` to `7.30.0.post11.dev35881573973` until both are released, and fill in the changelog PR number. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…e_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 <noreply@anthropic.com>
…uard the async download paginator - `skip_empty_records` dropped nulls before an `OffsetIncrement` or `PageIncrement` counted the page, so one null on a full page made the page look short and silently ended pagination. The paginator's copy of the extractor now yields a placeholder for every empty record it drops, so the count is what the API returned. The `first_match` winner is still chosen by the surviving records, so the paginator counts the same sub-extractor the record stream reads, and only the record stream logs the drop. - The `union` + `OffsetIncrement` rejection never reached `AsyncRetriever.download_paginator`, which was built without an extractor model and fell back to the retriever's summed count. A `CombinedExtractor` download extractor is now handed to the download paginator, which applies both the rejection and the dropped-record count. Other download extractors keep the retriever's count, because a paginator copy would re-read a response that a streaming download decoder has already consumed. - The documented `first_match` example lacked the trailing `"*"` on its wildcard path, so it emitted each board's item list as one record. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…-extractor-with-parent-fields
…dk/combined-extractor-with-parent-fields
Contributor
Author
|
/prerelease
|
Contributor
Author
|
Superseded: #1162, #1165 and #1174 are merged and released in 7.31.0, so this combined prerelease branch is no longer needed. The connector PR airbytehq/airbyte#86932 now pins 7.31.0. Closing. |
This branch was successfully deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Do not merge. This branch exists only to publish a prerelease that contains both #1162 (
CombinedExtractor) and #1165 (RecordExpander.parent_fields), each withmainmerged in.The source-github PR that adopts
parent_fieldson top ofCombinedExtractorneeds one CDK build with both features. Close this PR once #1162 and #1165 are merged.🤖 Generated with Claude Code