feat: add CombinedExtractor for combining multiple record extractors

[Anatolii Yatsuk (tolik0)]

What

Adds CombinedExtractor, a declarative RecordExtractor that combines the output of several
sub-extractors. A single DpathExtractor can only describe one path into a response; connectors
that need more than that drop into a custom components.py, which makes them ineligible for the
Connector Builder and for manifest-only packaging.

- type: CombinedExtractor
  mode: union # union | first_match (default: union)
  extractors:
    - type: DpathExtractor
      field_path: ["data", "repository", "issues", "nodes"]
    - type: DpathExtractor
      field_path: ["data", "repository", "pullRequests", "nodes"]

Why - connectors hand-roll this today

Every claim below was checked against the connector's components.py in the monorepo, and the replacement was run against fixtures where it mattered rather than compared by shape. Two of the claims in the first revision of this description were too strong and are corrected here; a third mode, zip_merge, was dropped after that check (see below).

1. source-monday -
   components.py:75-124,
   MondayIncrementalItemsExtractor. It does not concatenate: it tries field_path, tracks
   has_records, and only falls back to field_path_pagination when the first path produced
   nothing (lines 113-124). That is first_match, and it is the reason that mode exists.
   The first revision of this description said monday was not a drop-in adopter, because its
   _try_extract_records (lines 106-111) also drops falsy entries and does so before deciding
   whether the primary path produced anything, while DpathExtractor yields list items verbatim.
   Migrating it as-is would have started emitting None records and - on a page where every item
   is None - counted those as records, locked in the primary extractor and never fallen back.
   skip_empty_records closes that gap, so monday is a drop-in adopter with the flag on. One divergence
   stays, deliberately: monday's warning names body.get("errors"), which assumes a GraphQL error
   envelope, so the component logs the dropped count and the sub-extractor instead. monday keeps
   six other custom classes, so it does not become manifest-only.
2. source-zendesk-chat -
   components.py:19,
   ZendeskChatBansRecordExtractor, concatenates ip_address and visitor from one response.
   That is union. It then sorts by created_at, which union does not do; the sort exists because
   the bans paginator uses {{ last_record['id'] + 1 }} as since_id, so the connector needs the
   last emitted record to carry the highest id. A union adoption therefore comes with a manifest
   change to cursor_value: "{{ (response.ip_address + response.visitor) | map(attribute='id') | max + 1 }}",
   which is also more correct than the sort, since it does not assume created_at order equals id order.
   With that change zendesk-chat loses its components.py.
3. source-github - union is the wrapper that picks between the listing document
   (data.repository.pullRequests.nodes[*]) and the drill-down document (data.repository.pullRequest,
   or data.node) of the GraphQL streams on the unmerged tolik0/source-github/graphql-streams branch.
   It cannot express those streams on its own, because the children also need fields copied from their
   parent; that half is DpathExtractor.record_expander with the parent_fields option from feat(RecordExpander): add parent_fields to copy named parent values onto expanded items #1165.
   union over DpathExtractors with a RecordExpander reproduces both NestedGraphQLRecordExtractor
   and DeepNestedGraphQLRecordExtractor record-for-record on seven fixture shapes, including the
   four-level pull_request_comment_reactions traversal. The connector still keeps two custom
   pagination strategies and a transformation, so it does not become manifest-only either.

Checked and found not to be adopters:

• source-google-analytics-data-api CombinedExtractor (components.py:17-44) was zip_merge
  character for character, but both of its sub-extractors are a custom KeyValueExtractor this
  component cannot express, so only the outer layer could ever migrate and the connector keeps its
  components.py regardless. It was the only zip_merge adopter in the fleet, which is why the mode
  was dropped.
• source-zendesk-support ZendeskSupportAttributeDefinitionsExtractor (components.py:32-46)
  reads two paths and stamps a per-path constant (condition: all / any). After a union nothing
  tells AddFields which branch a record came from.
• source-pinterest AdAccountRecordExtractor (components.py:60-82) returns items when the
  key is present and otherwise wraps the whole response as one record. first_match falls through
  when items is present but empty, where pinterest returns nothing, and the wrap is a
  DpathExtractor gap, not a combine mode.

Count: union 2 adopters, first_match 1, connectors that lose components.py 1 (zendesk-chat).

Dropped: zip_merge

An earlier revision had a third mode, zip_merge, merging the i-th record of every sub-extractor
into one record and stopping at the shortest sub-extractor. It had exactly one adopter
(google-analytics-data-api, above), that connector stays Python either way, and the mode needed
more documentation of its failure modes - shortest-wins truncation, alignment shift under
skip_empty_records, object-only records - than of its behaviour. Five connectors do zip
column headers with row values (amplitude, google-analytics-data-api, google-search-console,
google-sheets, us-census), but four of them zip inside a single response array, which sub-extractor
zipping does not express. If a columnar extractor is wanted later it should be its own component
over one array, not a combine mode. Commit bbfd864c removes the mode, its nine tests, and the
OffsetIncrement counting rule for it.

skip_empty_records

Off by default; existing manifests are unaffected. When on, falsy records - None, {}, [],
"" - are dropped before any mode sees them, so they neither reach the record stream nor count
towards the first_match decision. A sub-extractor whose records are all empty therefore loses,
and the next one is tried. That is what makes source-monday a drop-in adopter, and it is also
what a paginator counts: under first_match the count is the winning sub-extractor's count after
the empty records were dropped.

A warning naming the dropped count and the sub-extractor is logged once per extraction, and only
when something was actually dropped.

Record-counting paginators: union is rejected under OffsetIncrement

OffsetIncrement.next_page_token advances the offset by the number of records its extractor
returns for the page. Under union that is the sum over all sub-extractors, so the offset
overshoots the page the API returned: two sub-extractors returning two records each with
page_size: 2 request offsets 0, 4, 8 instead of 0, 2, 4 and silently lose two thirds of the
records. ModelToComponentFactory.create_offset_increment therefore rejects a union
CombinedExtractor - including a union nested anywhere in the tree - with an
AirbyteTracedException(failure_type=config_error).

first_match (the winning sub-extractor's count) does not inflate the count and is accepted. PageIncrement accepts union too: it only
compares the count against page_size to decide whether to stop, so an inflated count costs one
extra request on the last page rather than records. Both facts are in the docstring and in the
mode schema description.

Modes

union (default) - every record of every sub-extractor, in the order the extractors are
declared. The GraphQL case: one document, records under two sibling paths.

extractor:
  type: CombinedExtractor
  mode: union
  extractors:
    - type: DpathExtractor
      field_path: ["data", "repository", "issues", "nodes"]
    - type: DpathExtractor
      field_path: ["data", "repository", "pullRequests", "nodes"]

first_match - the records of the first sub-extractor that produces at least one record;
the remaining sub-extractors are never touched. Nothing is yielded if none of them produces a
record. The monday case.

extractor:
  type: CombinedExtractor
  mode: first_match
  skip_empty_records: true
  extractors:
    - type: DpathExtractor
      field_path: ["data", "boards", "*", "items_page", "items", "*"]
    - type: DpathExtractor
      field_path: ["data", "next_items_page", "items", "*"]

"Produced at least one record" is decided by peeking a single element and chaining it back with
itertools.chain, never by list()-ing a sub-extractor - large responses stay lazy and the peeked
record is still emitted.

Backward compatibility: additive only - no existing manifest changes behavior

• The component is new. No existing component's class, defaults or runtime behavior is touched.
• The schema change is additive in both directions: one new CombinedExtractor definition, and
  CombinedExtractor appended to five existing anyOf lists. Appending a member to an anyOf
  only widens what validates; every manifest that validated before still validates, and it still
  resolves to the same member because each extractor definition pins a distinct type literal.
• Verified empirically rather than by inspection: all 531 connector manifests in the monorepo were
  jsonschema-validated against the base and head schemas with identical verdicts, and all 5,818
  extractor payloads inside them were parsed through both the base and the head pydantic model
  modules with zero resolution changes (5,677 DpathExtractor, 84 CustomRecordExtractor, the
  same 57 non-literal payloads failing on both sides).
• model_to_component_factory gains one registry entry and two methods. Nothing dispatches
  differently: _create_component_from_model keys on the pydantic model type, and no existing
  manifest can produce a CombinedExtractor model.
• No connector is migrated in this PR, so no released connector's output can change.

Streaming decoders are rejected at parse time

Every sub-extractor is handed the same requests.Response, so the body must be readable more than
once. That holds for the buffering decoders, which is the common case: JsonDecoder and any
CompositeRawDecoder(stream_response=False) read response.content (cached by requests),
XmlDecoder reads response.text, ZipfileDecoder reads response.content.

It does not hold for streaming decoders - CompositeRawDecoder(stream_response=True), which is
what CsvDecoder, JsonlDecoder, JsonItemsDecoder and GzipDecoder resolve to outside the
Connector Builder, plus IterableDecoder. An earlier revision of this description claimed the
failure was loud. It is not. requests puts a urllib3.HTTPResponse in response.raw, and
reading a closed urllib3.HTTPResponse returns an empty body rather than raising, so:

• union emits only the first sub-extractor's records, with no error;
• first_match emits nothing at all when the first path misses - which is precisely the case that
  mode exists for.

The old test_streaming_decoder_is_a_known_limitation only passed because it put an io.BytesIO
in response.raw, and io.BytesIO does raise on a read after close. Since the Connector Builder
forces those decoders to stream_response=False, the manifest would have test-read correctly and
lost records once published, with the cursor advancing over the lost window.

Both halves are fixed here:

• ModelToComponentFactory.create_combined_extractor now raises an AirbyteTracedException with
  FailureType.config_error when the stream's decoder streams the response. The user-facing
  message states the violated condition only; the decoder that was configured and the buffered
  alternatives are in internal_message. It fires for nested CombinedExtractors too, since
  sub-extractors inherit the retriever's decoder, and it fires in the Connector Builder as well:
  _is_decoder_downgraded_by_connector_builder matches the exact CompositeRawDecoder class
  carrying one of the four streaming parsers the Builder downgrades, so the Builder rejects exactly
  what production rejects while a buffered CompositeRawDecoder (or a custom subclass of it) is
  left alone.
• The test is rebuilt on a real urllib3.HTTPResponse and renamed to what it pins:
  test_union_over_a_streaming_decoder_silently_drops_the_later_sub_extractors and
  test_first_match_over_a_streaming_decoder_silently_returns_the_wrong_answer, with a buffered
  control. A parametrized factory test asserts the parse-time rejection for all five decoders, in
  and out of the Builder.

minItems: 1 is now in the schema

An earlier revision left extractors without minItems: 1, on the grounds that pydantic v1 cannot
enforce a length constraint through the ForwardRef created by the self-referencing list. The
pydantic behaviour is real in isolation, but the constraint never reaches pydantic:
datamodel-codegen silently drops minItems on this construct (verified by regenerating the models
with the pinned datamodel_code_generator==0.26.3; the generated file is byte-identical with and
without it, while a non-self-referencing list such as RateLimitedMultipleTokenAuthenticator.quotas
does get min_items=1). Manifest validation is jsonschema, not pydantic, so the constraint is
enforced exactly where a configuration error belongs. The __post_init__ guard stays as the
backstop for direct Python construction.

Other notes

• Sub-extractors must declare their type. I deliberately did not add
  "CombinedExtractor.extractors": "DpathExtractor" to DEFAULT_MODEL_TYPES: inside a
  heterogeneous list, silently defaulting a typo'd entry to DpathExtractor hides errors.
• file_extractor on FileUploader was left alone, deliberately. It is the one record
  extractor reference site that did not get CombinedExtractor. git grep file_extractor -- airbyte_cdk returns only the schema line and the model line: create_file_uploader never reads
  it, nothing under retrievers/file_uploader/ reads it, and no connector in the monorepo sets it.
  Widening a dead field would be noise; deleting it belongs in a separate PR. The other five sites
  all got it: RecordSelector.extractor, FileUploader.download_target_extractor, and
  AsyncRetriever's status_extractor, download_target_extractor and download_extractor.
• Cost. Each sub-extractor decodes the response independently (CompositeRawDecoder re-parses
  the cached body on each decode()), so a response is parsed once per sub-extractor. An
  OffsetIncrement or PageIncrement paginator builds its own copy of the extractor to count a
  page, doubling that.
• Annotations widened. AsyncHttpJobRepository.status_extractor /
  download_target_extractor are now typed RecordExtractor rather than DpathExtractor, and the
  three extractor_model parameters in the factory now include CombinedExtractorModel. No
  runtime change; the old annotations simply understated what those sites already receive.
• Fixture ignore. unit_tests/sources/declarative/extractors/.gitignore now ignores
  test_response.csv, the 62 MB artifact large_event_response_fixture writes and removes in
  teardown, mirroring the existing decoders/test_response.txt rule.
• Exports. CombinedExtractor and CombineMode are exported from airbyte_cdk, matching
  DpathExtractor and the UnionPartitionRouter precedent.

Generated models

airbyte_cdk/sources/declarative/models/declarative_component_schema.py was updated by hand to
match the schema, on purpose. A local poe assemble on this machine does not reproduce what is
committed on main: it rewrites hundreds of unrelated lines, emitting conint(ge=1) /
confloat(ge=0.0) where the committed file uses Field(..., ge=N) (main contains zero
conint/confloat; a fresh local run emits nine), renames OAuthScope to Scope/OptionalScope,
and reorders unrelated generated classes. Rather than bury this component under that churn, the
generated file carries only the delta the schema change implies.

The delta is byte-faithful to what the pinned generator produces for the new block, verified by
regenerating with datamodel_code_generator==0.26.3: class CombineMode(Enum) immediately before
ResponseToFileExtractor, mode: Optional[CombineMode] = Field(CombineMode.union, ...) with no
per-field examples, and CombinedExtractor.update_forward_refs() after CompositeErrorHandler's.

mode is a $ref to a CombineMode definition rather than an inline enum for that reason:
datamodel-codegen derives an inline property enum's class name from the property name and not from
title, so title: Combine Mode produced the very generic top-level Mode. A definitions entry
gives the generated enum the same name as the runtime enum in extractors/combined_extractor.py,
and moves the examples list from the field onto the definition. A maintainer should still run
/poe build on this PR to regenerate the models authoritatively before merge.

Tests

unit_tests/sources/declarative/extractors/test_combined_extractor.py (22 tests):

• union yields every record and preserves sub-extractor order; union is the default mode;
  union skips empty sub-extractors.
• first_match falls through an empty first extractor; does not drop the peeked record; never
  touches the extractors after the winner; yields nothing when every sub-extractor is empty.
• A single sub-extractor behaves identically to that extractor used directly, in both modes.
• An empty extractors list raises a ValueError naming the field.
• mode accepts the raw string from a manifest; an unknown mode is rejected.
• A buffered decoder can be read by every sub-extractor; a streaming decoder silently drops
  records, pinned over a real urllib3.HTTPResponse in both union and first_match.

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py:

• A manifest-level test that builds a RecordSelector whose extractor is a CombinedExtractor
  containing a nested CombinedExtractor, exercising the schema, the generated model, the
  PYDANTIC_MODEL_TO_CONSTRUCTOR registration, $parameters propagation and the extracted records.
• mode defaults to union through the factory.
• The parse-time rejection, parametrized over CsvDecoder, JsonlDecoder, JsonItemsDecoder,
  GzipDecoder and IterableDecoder, with and without emit_connector_builder_messages; plus the
  nested case and a full SimpleRetriever manifest, and the negative case for JsonDecoder /
  XmlDecoder.
• The union + OffsetIncrement rejection, parametrized over the default mode, an explicit
  mode: union and a union nested under a first_match; plus behavioural assertions that
  first_match advances the offset by the winning sub-extractor's count and that PageIncrement
  still accepts union.
• A buffered CompositeRawDecoder and a subclass of it are accepted in Builder mode, since the
  Builder guard matches only the four decoders the Builder downgrades.
• CombinedExtractor at all three AsyncRetriever extractor sites and at
  FileUploader.download_target_extractor.

unit_tests/sources/declarative/test_concurrent_declarative_source.py:

• Three manifest-level tests that actually reach jsonschema validation: a valid nested
  CombinedExtractor manifest constructs, mode: concatenate raises ValidationError, and
  extractors: [] raises ValidationError with the minItems message in the error tree.

No new markers; nothing is marked slow, flaky, super_slow or linting.

CI at this head

bbfd864c: Pytest (Fast) and Pytest (All) on Python 3.10–3.13, MyPy, Ruff Lint and Ruff Format all pass. Check: destination-motherduck fails and is unrelated; it is red on #1149 and #1165 at the same time.

🤖 Generated with Claude Code

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You 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#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

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 495 tests  +107   4 483 ✅ +107   7m 50s ⏱️ - 1m 48s
    1 suites ±  0      12 💤 ±  0 
    1 files   ±  0       0 ❌ ±  0 

Results for commit bbfd864. ± Comparison against base commit 96a7c0a.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 498 tests  +107   4 486 ✅ +107   13m 37s ⏱️ -24s
    1 suites ±  0      12 💤 ±  0 
    1 files   ±  0       0 ❌ ±  0 

Results for commit bbfd864. ± Comparison against base commit 96a7c0a.

♻️ This comment has been updated with latest results.

[devin-ai-integration]

Review of 5c7510c37244cdcb1a695d8cebc7dee205253327

Everything below was executed locally on this SHA (branch checked out, Poetry env from the lockfile). Findings are tagged P0–P3; no P0.

Breaking-change call: NON_BREAKING

git grep on airbytehq/airbyte origin/master (583b2dee3a1e):

Query	Count
type: CombinedExtractor under airbyte-integrations/	0
CombinedExtractor anywhere	4 — all source-google-analytics-data-api (AGENTS.md, CONTRIBUTING.md, components.py, manifest.yaml). It is a connector-local custom component (class_name: source_declarative_manifest.components.CombinedExtractor), resolved via CustomRecordExtractor, so it does not collide with the new declarative type.
CombineMode	0
from airbyte_cdk import *	0
class Mode / import Mode / Mode as from airbyte_cdk.sources.declarative.models in connector .py	0 (class Mode is new on this branch, absent on origin/main)
.py referencing status_extractor / download_target_extractor / AsyncHttpJobRepository	1 — source-salesforce/source_salesforce/streams.py (constructs AsyncHttpJobRepository; the annotation widening DpathExtractor → RecordExtractor is contravariant-safe for callers)
manifests with type: RecordSelector	533 (all unaffected: schema changes are additive anyOf members)
manifests with first_match / zip_merge	0

All changes are additive (new schema type, new anyOf members, widened type hints, new exports). No existing manifest or Python consumer is affected.

Codegen (checked against clean main, per the known ~350-line drift)

Ran bin/generate_component_manifest_files.py (datamodel-codegen 0.26.3 in Dagger) in a clean origin/main worktree and in this branch's worktree, then diffed generator-output vs generator-output. The regen delta is 147 lines and contains exactly this PR's additions (class Mode, class CombinedExtractor, the union widenings, CombinedExtractor.update_forward_refs()) — the committed hand-edit matches generator output modulo line wrapping. min_items=1 for extractors is absent in both committed and regenerated output: datamodel-codegen 0.26.3 drops minItems when items is an anyOf (same as partition_routers), so it's a generator limitation, not a hand-edit deviation. JSON-schema validation still enforces minItems (the new test_concurrent_declarative_source test covers it).

Local checks

• poetry run ruff check . — pass; poetry run ruff format --check . — pass; poetry run mypy --config-file mypy.ini airbyte_cdk — pass (459 files).
• unit_tests/sources/declarative/{extractors,decoders} + unit_tests/connector_builder — 236 passed.
• extractors/test_combined_extractor.py, parsers/test_model_to_component_factory.py, test_concurrent_declarative_source.py, requesters/ — 728 passed, 1 skipped, 2 failed: test_read_with_concurrent_and_synchronous_streams_with_{concurrent,sequential}_state (sqlite3.OperationalError: database table is locked: responses from requests_cache). Same two fail identically on a clean origin/main (96a7c0ac) worktree in this environment, so I'm not attributing them to this PR; flagging for awareness since PR CI reports them green.

CI snapshot at review time

25 passed, destination-motherduck failed, source-shopify pending — matches the stated baseline. As noted, no connector in the matrix declares CombinedExtractor, so CI doesn't exercise this component; the findings below come from a local end-to-end read.

Findings

• P1 — union mode + OffsetIncrement silently skips pages (data loss). Reproduced end-to-end; details inline on combined_extractor.py.
• P2 — schema mode description states the wrong pagination caveat (blames first_match/zip_merge, omits union, which is the harmful one). Inline on declarative_component_schema.yaml.
• P2 — create_offset_increment accepts CombinedExtractorModel without guarding the union case. Inline on model_to_component_factory.py.
• P3 — Builder-mode guard rejects any CompositeRawDecoder subclass regardless of stream_response. Inline.
• P3 — generated enum name Mode is a very generic new top-level export of declarative_component_schema.py. Inline.

Probe script used for the P1 repro: end-to-end ConcurrentDeclarativeSource.read against requests_mock, manifest with CombinedExtractor(union, [a, b]) + OffsetIncrement(page_size=2), API returning 2 items per path per page for 6 items each.

---

Devin session

[devin-ai-integration]

P1 — union + OffsetIncrement skips records. This note covers first_match and zip_merge but omits union, which is the mode where the mismatch actually loses data: OffsetIncrement.next_page_token does len(list(self.extractor.extract_records(response))) and advances the offset by that count. Under union that's the sum across sub-extractors, so the offset overshoots.

Reproduced end-to-end on this SHA (ConcurrentDeclarativeSource.read, requests_mock, CombinedExtractor(union, [a, b]) + OffsetIncrement(page_size=2), API returning 2 items for a and 2 for b per page, 6 each):

requests (qs): [{'offset': ['0'], 'limit': ['2']}, {'offset': ['4'], 'limit': ['2']}, {'offset': ['8'], 'limit': ['2']}]
records: ['a0', 'a1', 'b0', 'b1', 'a4', 'a5', 'b4', 'b5']
expected 12 records; got 8

a2, a3, b2, b3 are silently dropped and there is no warning. first_match behaves correctly in the same harness (offsets 0, 2, 4; all 5 b records). PageIncrement is not lossy (it only uses the count for the < page_size stop check) but will issue one extra request when the combined last page happens to equal page_size.

Suggest either (a) rejecting mode: union under an OffsetIncrement paginator in the factory with a config_error, or (b) documenting here and in the schema that union must not be combined with OffsetIncrement, and making the existing test_combined_extractor_is_accepted_by_an_offset_increment_paginator assert behaviour rather than just construction.

[Anatolii Yatsuk (tolik0)]

Confirmed and fixed in 02f07c6 — I went with option (a), the parse-time rejection.

ModelToComponentFactory.create_offset_increment now raises an AirbyteTracedException with FailureType.config_error when its extractor_model is a CombinedExtractor whose tree contains a union node at any depth (a nested union inflates the count of the node above it, so the whole tree is walked). first_match and zip_merge are untouched.

PageIncrement is left accepting union, since it only compares the count against page_size: the cost is the one extra request you describe, which is now documented on the component rather than rejected.

Tests replacing test_combined_extractor_is_accepted_by_an_offset_increment_paginator:

• test_union_combined_extractor_is_rejected_by_an_offset_increment_paginator, parametrized over the default mode, an explicit mode: union, and a union nested under a first_match.
• test_first_match_combined_extractor_counts_only_the_winning_sub_extractor_for_the_offset — asserts behaviour, not construction: with a empty and b holding two records, next_page_token returns offset 2, not 4.
• test_zip_merge_combined_extractor_counts_the_shortest_sub_extractor_for_the_offset — two merged records out of four extracted, offset advances by 2.
• test_union_combined_extractor_is_accepted_by_a_page_increment_paginator.

[devin-ai-integration]

P2 — pagination caveat is inverted. "under first_match and zip_merge is not the number of records the API returned" — in practice first_match returns the winner's count, which is the API count for the path that had data (verified: offsets 0/2/4 in the repro), and zip_merge returns the shortest count, which is usually the API count too. The mode that misreports is union (sum of all sub-extractors), and with OffsetIncrement that skips pages (see the P1 comment on combined_extractor.py). This text is what Builder users will read; it should name union + OffsetIncrement as the unsupported combination.

[Anatolii Yatsuk (tolik0)]

You are right, the caveat was backwards. Rewritten in 02f07c6; the mode description now reads:

Note that a paginator which counts the records of a page counts the combined records: under "union" that is the sum over all sub-extractors, which overshoots the API page size, so "union" is rejected with an OffsetIncrement paginator because the offset would skip records. Under "first_match" the count is the count of the winning sub-extractor and under "zip_merge" the count of the shortest one, which are usually the number of records the API returned for the page.

The same correction is in the component docstring, under a new "Record-counting paginators" section that also records the extra PageIncrement request.

[devin-ai-integration]

P2 — accepting CombinedExtractorModel here without a guard is what enables the P1 data loss. create_offset_increment builds a second CombinedExtractor whose extract_records count is fed to OffsetIncrement.next_page_token. For mode: union that count is sum(len(sub_i)), not the API page size. Since this PR already rejects unsupported decoder combinations at construction (_reject_combined_extractor_over_streaming_decoder), the same pattern fits here: if isinstance(extractor_model, CombinedExtractorModel) and (extractor_model.mode or Mode.union) == Mode.union, raise an AirbyteTracedException(failure_type=config_error) explaining that union cannot drive offset pagination. Nested CombinedExtractors with a union anywhere in the tree have the same problem.

[Anatolii Yatsuk (tolik0)]

Done in 02f07c6, using exactly the pattern you suggest: create_offset_increment calls a new _reject_union_combined_extractor_for_offset_increment(extractor_model) right after the decoder check, which raises AirbyteTracedException(failure_type=FailureType.config_error) when the extractor model is a CombinedExtractorModel whose tree contains a union.

The nested case is covered by _combined_extractor_tree_contains_union, which recurses through the sub-extractors, so a union under a first_match is rejected too. create_page_increment deliberately keeps accepting union — it only uses the count for the < page_size stop check, so the count inflation costs a request rather than records.

[devin-ai-integration]

P3 — In Builder mode this rejects every CompositeRawDecoder instance, including ones constructed with stream_response=False (e.g. a CustomDecoder subclassing CompositeRawDecoder, or JsonDecoder-style wrappers if they are ever refactored to subclass it). inner_decoder.is_stream_response() already covers the production case; the Builder branch exists only because create_*_decoder forces stream_response=False when _emit_connector_builder_messages is set. Consider checking the model type (CsvDecoderModel, JsonlDecoderModel, GzipDecoderModel, IterableDecoderModel, …) rather than the runtime class, so the rejection matches the message text exactly. Low impact today: 0 connectors on origin/master use CombinedExtractor.

[Anatolii Yatsuk (tolik0)]

Narrowed in 02f07c6. The Builder half of the guard is now _is_decoder_downgraded_by_connector_builder, which matches on the exact class plus the parser rather than on isinstance:

return type(decoder) is CompositeRawDecoder and isinstance(
    decoder.parser, (CsvParser, JsonLineParser, JsonItemsParser, GzipParser)
)

That is exactly the set create_csv_decoder / create_jsonl_decoder / create_json_items_decoder / create_gzip_decoder build with stream_response=False in Builder mode, so the rejection matches the message text. A CustomDecoder subclassing CompositeRawDecoder, or a buffered CompositeRawDecoder around a JsonParser, is no longer rejected — both read response.content in production too. IterableDecoder still lands in the first branch, since is_stream_response() is True for it in the Builder as well.

I kept the runtime check rather than threading decoder models down to create_combined_extractor: the decoder arrives there as an instance through create_record_selector, and passing the model through every call site would be a much larger change than the guard warrants.

New test: test_combined_extractor_over_a_buffered_composite_raw_decoder_is_accepted_in_the_builder, parametrized over a CompositeRawDecoder(JsonParser(), stream_response=False) and a subclass of it.

[devin-ai-integration]

P3 — This is what datamodel-codegen 0.26.3 emits for the inline mode enum (verified by regenerating from clean main and from this branch — the regen delta matches the committed file), so it's not a hand-edit issue. But Mode is a very generic new top-level name in airbyte_cdk.sources.declarative.models.declarative_component_schema, which several connectors import names from (source-github, source-klaviyo, source-stripe, source-sftp-bulk); today no connector imports or defines Mode from it (0 hits on origin/master), so this is not breaking. If you want a stable, descriptive name, give the enum its own definitions entry (e.g. CombineMode) and $ref it — codegen will then emit CombineMode, matching the runtime enum in combined_extractor.py.

[Anatolii Yatsuk (tolik0)]

Took the suggestion in 02f07c6: mode is now a $ref to a new CombineMode definition in declarative_component_schema.yaml (alongside the existing enum-only definitions such as SchemaNormalization and ValueType), so the generated name matches the runtime enum in combined_extractor.py.

Verified with the pinned generator rather than by hand-editing the name: running datamodel-codegen==0.26.3 with the flags from bin/generate_component_manifest_files.py emits

class CombineMode(Enum):
    union = "union"
    first_match = "first_match"
    zip_merge = "zip_merge"

immediately before ResponseToFileExtractor, and mode: Optional[CombineMode] = Field(CombineMode.union, ...). The committed generated block is that output verbatim (the per-field examples list moves onto the definition, which is why it is gone from the field). class Mode is gone from the module. Schema validation is unchanged: the mode: concatenate and extractors: [] ValidationError tests in test_concurrent_declarative_source.py still pass through the $ref.

[devin-ai-integration]

Re-review of 02f07c66d0f8b7bb27262de692ebb0c3e48ab355 (follow-up to my review of 5c7510c3)

Everything below was executed locally on this SHA.

Previous findings

Finding	Status
P1 union + OffsetIncrement skips pages	Fixed. Re-ran the same end-to-end repro (ConcurrentDeclarativeSource.read + requests_mock, union[a,b] + OffsetIncrement(page_size=2)): construction now raises AirbyteTracedException (config_error) before any request is sent. first_match + OffsetIncrement still reads correctly (offsets 0/2/4, all 5 records). OffsetIncrement( is constructed at exactly one site in the factory, so the guard covers every declarative path, including nested unions (_combined_extractor_tree_contains_union).
P2 schema mode description inverted	Fixed — now names union as the rejected combination.
P2 create_offset_increment unguarded	Fixed — _reject_union_combined_extractor_for_offset_increment, 3 rejection cases + first_match/zip_merge/PageIncrement acceptance tests with real next_page_token assertions.
P3 Builder guard rejects any CompositeRawDecoder	Fixed — _is_decoder_downgraded_by_connector_builder matches type(...) is CompositeRawDecoder + one of the four parsers, which is exactly what create_{csv,jsonl,json_items,gzip}_decoder build under _emit_connector_builder_messages (lines 2839/2847/2855/2875).
P3 generic Mode enum name	Fixed — CombineMode is now a definitions entry; regenerated with datamodel-codegen 0.26.3 from clean main and from this SHA, and the regen delta (class CombineMode(Enum) before ResponseToFileExtractor, mode: Optional[CombineMode], no examples=, update_forward_refs position) matches the committed file exactly.

New findings

• P1 — a 62 MB, 2,000,001-line unit_tests/sources/declarative/extractors/test_response.csv is committed in this commit (git log --diff-filter=A → 02f07c66). GitHub won't render a diff for it, so no inline anchor is possible:

  This file (2,000,001 lines, 62,000,016 bytes) is the artifact that large_event_response_fixture in test_response_to_file_extractor.py writes and then os.removes; it survived here because a @pytest.mark.slow run was interrupted before teardown. Once merged it stays in every clone's history forever (a squash merge only helps if the file is gone from the final tree), and unit_tests/sources/declarative/extractors/ has no .gitignore covering it (the only related rule is decoders/.gitignore: test_response.txt).

  Please git rm it in a follow-up commit, and consider adding unit_tests/sources/declarative/extractors/.gitignore with test_response.csv (mirroring the decoders/ one) so the fixture can't be committed again.

• P3 — the new AirbyteTracedException.message embeds remediation ("Use the first_match or zip_merge mode, …"); per the CDK error-message guidelines the user-facing message should state the one violated condition and remediation goes in docs/internal_message. Same pattern as the pre-existing streaming-decoder message from 5c7510c3, so treat as a nit for both. Inline.

Breaking-change call: NON_BREAKING (unchanged)

The new commit only touches the new component, its schema entry, the factory guard, and tests. The Mode → CombineMode rename affects a symbol that was never released (introduced in this PR; 0 references on airbytehq/airbyte origin/master). Counts from the first review stand (0 type: CombinedExtractor, 0 CombineMode, 0 first_match/zip_merge in manifests, 0 star imports).

Local checks

• ruff check / ruff format --check / mypy --config-file mypy.ini airbyte_cdk — all pass (459 files).
• test_combined_extractor.py + test_model_to_component_factory.py + test_concurrent_declarative_source.py: no failures in the first two. test_concurrent_declarative_source.py failed 14 tests in one run and 2 in an immediate rerun — every failure is sqlite3.OperationalError: database table is locked: responses from requests_cache under concurrent reads (no assertion errors). The same 2 fail identically on a clean origin/main worktree in this environment, so I'm not attributing them to the PR, but flagging since CI reports the file green.

CI

Not re-checked for this SHA beyond the snapshot in my previous review; connector matrix still doesn't exercise CombinedExtractor.

---

Devin session

[devin-ai-integration]

P3 (nit) — Per the CDK error-message guidelines, the user-facing message should be one violated condition, ≤~120 chars, no remediation (that belongs in the docs/schema description, which now already have it); internal_message carries the mechanism. Something like message="CombinedExtractor mode \"union\" is not supported with an OffsetIncrement paginator." with the current text moved into internal_message. Same nit applies to the pre-existing _reject_combined_extractor_over_streaming_decoder message — fine to address both or neither together.

[Anatolii Yatsuk (tolik0)]

Taken, for both messages, in 28da58e:

• message='CombinedExtractor mode "union" is not supported with an OffsetIncrement paginator.'
• message="CombinedExtractor is not supported with a streaming decoder."

The mechanism and the remediation (which mode/paginator or which buffered decoder to use instead) moved into internal_message in both cases; the schema description and the component docstring already carry the guidance for users.

[Anatolii Yatsuk (tolik0)]

Both findings from the re-review of 02f07c66 are addressed in 28da58ef.

P1 — the 62 MB test_response.csv. Correct, and thanks for catching it: it was a leftover large_event_response_fixture artifact that a git add on the test directory swept up. Rather than deleting it in a follow-up commit, I amended 02f07c66 and force-pushed, so the blob is no longer in this branch's history at all (02f07c66 → 28da58ef). unit_tests/sources/declarative/extractors/.gitignore now contains test_response.csv, mirroring the decoders/test_response.txt rule, so the fixture cannot be committed again.

P3 — error-message remediation. Applied to both messages, the new one and the pre-existing streaming-decoder one; details in the inline reply.

Local checks on 28da58ef: ruff check, ruff format --check and mypy --config-file mypy.ini airbyte_cdk pass (459 files); the CombinedExtractor tests in test_combined_extractor.py, test_model_to_component_factory.py and test_concurrent_declarative_source.py pass. I see the same sqlite3.OperationalError: database table is locked: responses flakiness in test_concurrent_declarative_source.py locally, on main as well.

[devin-ai-integration]

Re-review of 28da58ef18e964e123a670bbcb9f2e87a93455db

Executed locally on this SHA (git diff 02f07c66 28da58ef: 4 files, +13/−2,000,016).

Previous findings

Finding	Status
P1 62 MB unit_tests/sources/declarative/extractors/test_response.csv committed	Fixed. File is gone from the tree; new unit_tests/sources/declarative/extractors/.gitignore (test_response.csv) — git check-ignore confirms the fixture path is now ignored. Note the blob still exists in the branch's history (added in 02f07c66); a squash merge drops it from main, a merge commit would keep it.
P3 error message embedded remediation	Fixed. Both AirbyteTracedException.messages are now a single condition (CombinedExtractor is not supported with a streaming decoder. / CombinedExtractor mode "union" is not supported with an OffsetIncrement paginator.); remediation and decoder lists moved to internal_message.

All findings from 5c7510c3 and 02f07c66 remain fixed — re-ran the end-to-end repro on this SHA: union + OffsetIncrement is rejected at construction before any request; first_match + OffsetIncrement reads offsets 0/2/4 and all 5 records.

Local checks

• poetry run ruff check . — pass; poetry run ruff format --check . — pass; poetry run mypy --config-file mypy.ini airbyte_cdk — pass (459 files).
• test_combined_extractor.py + test_model_to_component_factory.py — 206 passed, 0 failed (the message-text assertions "union" in message / "OffsetIncrement" in message still hold with the shortened strings).
• Codegen: this commit doesn't touch the schema YAML or generated model, so the 02f07c66 regen comparison (committed block == generator output) still applies.

Breaking-change call: NON_BREAKING (unchanged)

Only error strings, a docstring, a .gitignore and a fixture deletion changed. Monorepo counts from the first review stand: 0 type: CombinedExtractor, 0 CombineMode, 0 first_match/zip_merge manifests, 0 from airbyte_cdk import * on airbytehq/airbyte origin/master.

No open findings. Approving; the only thing to keep in mind is the squash-merge note above so the 62 MB blob doesn't land in main history.

---

Devin session