feat: add CombinedExtractor for combining multiple record extractors

airbyte_cdk/__init__.py

@@ -103,7 +103,12 @@
 from .sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from .sources.declarative.decoders import Decoder, JsonDecoder
 from .sources.declarative.exceptions import ReadException
-from .sources.declarative.extractors import DpathExtractor, RecordSelector
+from .sources.declarative.extractors import (
+    CombinedExtractor,
+    CombineMode,
+    DpathExtractor,
+    RecordSelector,
+)
 from .sources.declarative.extractors.record_extractor import RecordExtractor
 from .sources.declarative.extractors.record_filter import RecordFilter
 from .sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
@@ -234,6 +239,8 @@
     "BasicHttpAuthenticator",
     "BearerAuthenticator",
     "CartesianProductStreamSlicer",
+    "CombineMode",
+    "CombinedExtractor",
     "CursorPaginationStrategy",
     "DatetimeBasedCursor",
     "DeclarativeAuthenticator",

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1911,6 +1911,7 @@ definitions:
             anyOf:
               - "$ref": "#/definitions/DpathExtractor"
               - "$ref": "#/definitions/CustomRecordExtractor"
+              - "$ref": "#/definitions/CombinedExtractor"
           file_extractor:
             description: Responsible for fetching the content of the file. If not defined, the assumption is that the whole response body is the file content
             anyOf:
@@ -2183,6 +2184,66 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  CombineMode:
+    title: Combine Mode
+    description: 'How the records of the sub-extractors of a CombinedExtractor are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record.'
+    type: string
+    enum:
+      - union
+      - first_match
+    examples:
+      - union
+      - first_match
+  CombinedExtractor:
+    title: Combined Extractor
+    description: Record extractor that combines the output of several sub-extractors. Use it when a single response holds records under more than one path, when the records can be found under one of several alternative paths, or when a record has to be assembled from several parallel lists.
+    type: object
+    required:
+      - type
+      - extractors
+    properties:
+      type:
+        type: string
+        enum: [CombinedExtractor]
+      extractors:
+        title: Extractors
+        description: The record extractors to combine. At least one is required. Each sub-extractor is given the same HTTP response and decodes it independently, so the response is parsed once per sub-extractor. Streaming decoders (CsvDecoder, JsonlDecoder, JsonItemsDecoder, GzipDecoder, IterableDecoder) can only read the response body once and are rejected.
+        type: array
+        minItems: 1
+        items:
+          anyOf:
+            - "$ref": "#/definitions/DpathExtractor"
+            - "$ref": "#/definitions/CustomRecordExtractor"
+            - "$ref": "#/definitions/CombinedExtractor"
+      mode:
+        title: Combine Mode
+        description: 'How the records of the sub-extractors are combined. "union" (default) yields every record of every sub-extractor, in the order the extractors are declared. "first_match" yields the records of the first sub-extractor that produces at least one record and skips the remaining ones; nothing is yielded if none of them produces a record. 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, which is usually the number of records the API returned for the page.'
+        "$ref": "#/definitions/CombineMode"
+        default: union
+      skip_empty_records:
+        title: Skip Empty Records
+        description: 'Whether to drop empty records - null, {}, [], "" - yielded by a sub-extractor before the records are combined. Off by default, so the output of a sub-extractor is passed through as it is. Turn it on when the API can return nulls in the middle of a record list, which a GraphQL API does when it answers a partial response and reports the failure in a sibling error field. Under "first_match" this also changes which sub-extractor wins: a sub-extractor whose records are all empty no longer counts as a match, so the next one is tried. The dropped records still count toward the page size of an OffsetIncrement or PageIncrement paginator, so a full page holding a null does not end pagination early.'
+        type: boolean
+        default: false
+      $parameters:
+        type: object
+        additionalProperties: true
+    examples:
+      - type: CombinedExtractor
+        mode: union
+        extractors:
+          - type: DpathExtractor
+            field_path: ["data", "repository", "issues", "nodes"]
+          - type: DpathExtractor
+            field_path: ["data", "repository", "pullRequests", "nodes"]
+      - 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"]
   DpathExtractor:
     title: Dpath Extractor
     description: Record extractor that searches a decoded response over a path defined as an array of fields.
@@ -3978,6 +4039,7 @@ definitions:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       record_filter:
         title: Record Filter
         description: Responsible for filtering records to be emitted by the Source.
@@ -4540,17 +4602,20 @@ definitions:
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       download_target_extractor:
         description: Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       download_extractor:
         description: Responsible for fetching the records from provided urls.
         anyOf:
           - "$ref": "#/definitions/DpathExtractor"
           - "$ref": "#/definitions/CustomRecordExtractor"
           - "$ref": "#/definitions/ResponseToFileExtractor"
+          - "$ref": "#/definitions/CombinedExtractor"
       creation_requester:
         description: Requester component that describes how to prepare HTTP requests to send to the source API to create the async server-side job.
         anyOf:

airbyte_cdk/sources/declarative/extractors/__init__.py

@@ -2,6 +2,10 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+from airbyte_cdk.sources.declarative.extractors.combined_extractor import (
+    CombinedExtractor,
+    CombineMode,
+)
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
 from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
@@ -14,6 +18,8 @@
 __all__ = [
     "TypeTransformer",
     "HttpSelector",
+    "CombinedExtractor",
+    "CombineMode",
     "DpathExtractor",
     "RecordFilter",
     "RecordSelector",

airbyte_cdk/sources/declarative/extractors/combined_extractor.py

@@ -0,0 +1,248 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import logging
+from dataclasses import InitVar, dataclass
+from enum import Enum
+from itertools import chain
+from typing import Any, Iterable, Iterator, List, Mapping, Union
+
+import requests
+
+from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor
+
+logger = logging.getLogger("airbyte")
+
+
+class CombineMode(Enum):
+    """
+    How a `CombinedExtractor` combines the output of its sub-extractors.
+    """
+
+    union = "union"
+    first_match = "first_match"
+
+
+class _NoRecord:
+    """Sentinel telling "this extractor yielded nothing" apart from a falsy or `None` record."""
+
+
+_NO_RECORD = _NoRecord()
+
+
+class _DroppedEmptyRecord:
+    """Stands in for an empty record dropped by `skip_empty_records` when the extractor counts a page.
+
+    A record-counting paginator only takes the length of what the extractor yields, so the
+    placeholder makes the dropped record count toward the page size without ever reaching the
+    record stream.
+    """
+
+
+_DROPPED_EMPTY_RECORD = _DroppedEmptyRecord()
+
+
+@dataclass
+class CombinedExtractor(RecordExtractor):
+    """Combines the output of several record extractors into a single stream of records.
+
+    A single `DpathExtractor` can only describe one path into the response. Several connectors need
+    more than that and reach for a custom `components.py` today, which keeps them out of the
+    Connector Builder. This component covers the two shapes those connectors implement:
+
+    - `union` (default) — yield every record of every sub-extractor, in sub-extractor order. Use it
+      when one response carries records under several paths, for example a GraphQL document that
+      returns `issues.nodes` and `pullRequests.nodes` side by side.
+    - `first_match` — yield the records of the first sub-extractor that produces at least one
+      record, and skip the remaining sub-extractors. If no sub-extractor produces a record, nothing
+      is yielded. Use it when an API answers with one of several alternative shapes. The winning
+      sub-extractor is only peeked at, never restarted: the peeked record is chained back in front
+      of the remaining ones, so large responses are still streamed lazily and no sub-extractor is
+      ever materialized into a list.
+
+    Examples of instantiating this component:
+    ```
+      extractor:
+        type: CombinedExtractor
+        mode: union
+        extractors:
+          - type: DpathExtractor
+            field_path: ["data", "repository", "issues", "nodes"]
+          - type: DpathExtractor
+            field_path: ["data", "repository", "pullRequests", "nodes"]
+    ```
+
+    ```
+      extractor:
+        type: CombinedExtractor
+        mode: first_match
+        extractors:
+          - type: DpathExtractor
+            field_path: ["data", "boards", "*", "items_page", "items", "*"]
+          - type: DpathExtractor
+            field_path: ["data", "next_items_page", "items"]
+    ```
+
+    ## Streaming decoders are rejected
+
+    Every sub-extractor is handed the same `requests.Response`, so the response body must be
+    readable more than once. That holds for the decoders that buffer the whole body, which is the
+    common case: `JsonDecoder` and any `CompositeRawDecoder` built with `stream_response=False`
+    read `response.content`, `XmlDecoder` reads `response.text` and `ZipfileDecoder` reads
+    `response.content` — `requests` caches all of those, so every sub-extractor sees the full body.
+
+    It does NOT hold for streaming decoders, i.e. a `CompositeRawDecoder` with
+    `stream_response=True`, which is what `CsvDecoder`, `JsonlDecoder`, `JsonItemsDecoder` and
+    `GzipDecoder` resolve to outside the Connector Builder, nor for `IterableDecoder`. Those read
+    `response.raw` and close it afterwards, so the first sub-extractor drains the stream and the
+    later ones read a dead body. That loss is SILENT rather than loud: `requests` puts a
+    `urllib3.HTTPResponse` in `response.raw`, and reading a closed `urllib3.HTTPResponse` returns
+    an empty body instead of raising, so `union` emits only the first sub-extractor's records and
+    `first_match` emits nothing at all when the first path misses.
+
+    Because of that, `ModelToComponentFactory.create_combined_extractor` refuses to build this
+    component over a streaming decoder and raises a configuration error, naming the decoder in the
+    internal message. The Connector Builder forces those same decoders to `stream_response=False`,
+    so the rejection is applied there too rather than letting a manifest test-read correctly and
+    lose records once published.
+
+    ## Empty records
+
+    `skip_empty_records` drops falsy records — `None`, `{}`, `[]`, `""` — before any of the modes
+    sees them. It is off by default, so a sub-extractor's output is passed through verbatim.
+
+    Turn it on when an API can answer with nulls in the middle of a record list. A GraphQL API
+    that returns a partial response puts `null` in the `data` array for the fields it failed to
+    resolve and describes the failure in a sibling `errors` field; without this flag those nulls
+    reach the record stream and fail schema validation downstream.
+
+    Under `first_match` the flag also changes which sub-extractor wins: a sub-extractor whose
+    records are all empty no longer counts as a match, so the next sub-extractor is tried. That is
+    the behaviour of `source-monday`'s `MondayIncrementalItemsExtractor`, where a page of nulls
+    must fall through to the pagination path rather than lock in the primary path.
+
+    The dropped records still count toward the size of the page, see "Record-counting paginators"
+    below.
+
+    ## Cost
+
+    Each sub-extractor decodes the response independently — `CompositeRawDecoder` re-parses the
+    cached body on every `decode()` call — so a response is parsed once per sub-extractor.
+    Combining three extractors over a large response costs roughly three times the decode time of
+    a single one.
+
+    Note that an `OffsetIncrement` or `PageIncrement` paginator builds its own copy of the
+    extractor to count the records of a page, which doubles that cost.
+
+    ## Record-counting paginators
+
+    The count those paginators obtain is the combined count, not the API's page size:
+
+    - `union` returns the SUM over all sub-extractors. An `OffsetIncrement` advances the offset by
+      that sum, so the next page starts past the records that were never read — two sub-extractors
+      returning two records each with `page_size: 2` request offsets 0, 4, 8 instead of 0, 2, 4 and
+      silently drop two thirds of the records. `ModelToComponentFactory.create_offset_increment`
+      therefore rejects a `union` `CombinedExtractor` (nested ones included) with a configuration
+      error. A `PageIncrement` only compares the count against `page_size` to decide whether to
+      stop, so it loses nothing, but it issues one extra request whenever the summed count of the
+      last page happens to equal `page_size`.
+    - `first_match` returns the count of the winning sub-extractor, which does not inflate the
+      count, so it works with either paginator as long as the path the count comes from is the one
+      the API paginates over.
+
+    Empty records dropped by `skip_empty_records` are part of that count. The API returned them as
+    slots of the page, so a full page holding a null must still read as full: counting only the
+    surviving records would make both paginators stop at the first page with a null, and would
+    make an `OffsetIncrement` re-request records it already read. The factory builds the
+    paginator's copy of the extractor with `count_dropped_empty_records` set, which makes that copy
+    yield a placeholder for every empty record it drops. The winner under `first_match` is still
+    chosen by the surviving records alone, so the paginator counts the same sub-extractor the
+    record stream reads.
+
+    Attributes:
+        extractors (List[RecordExtractor]): The sub-extractors to combine. At least one is required.
+        mode (CombineMode): How the sub-extractor outputs are combined. Defaults to `union`.
+        skip_empty_records (bool): Whether falsy records are dropped before they are combined.
+            Defaults to `False`.
+        count_dropped_empty_records (bool): Set by the factory on the copy a record-counting
+            paginator uses, never on the one that emits records: yield a placeholder in place of
+            every dropped empty record so it counts toward the page size. Defaults to `False`.
+    """
+
+    extractors: List[RecordExtractor]
+    parameters: InitVar[Mapping[str, Any]]
+    mode: CombineMode = CombineMode.union
+    skip_empty_records: bool = False
+    count_dropped_empty_records: bool = False
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        if not self.extractors:
+            raise ValueError(
+                "CombinedExtractor requires at least one extractor in its `extractors` field."
+            )
+        if not isinstance(self.mode, CombineMode):
+            self.mode = CombineMode(self.mode)
+
+    def extract_records(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
+        if self.mode == CombineMode.first_match:
+            yield from self._extract_first_match(response)
+        else:
+            yield from self._extract_union(response)
+
+    def _extract_union(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
+        for index in range(len(self.extractors)):
+            yield from self._records_of(index, response)
+
+    def _extract_first_match(self, response: requests.Response) -> Iterable[Mapping[str, Any]]:
+        for index in range(len(self.extractors)):
+            records: Iterator[Mapping[str, Any]] = iter(self._records_of(index, response))
+            # Placeholders for dropped empty records do not make a sub-extractor win, but they are
+            # kept so the winner's count still includes the ones that came before its first record.
+            dropped_placeholders: List[Mapping[str, Any]] = []
+            first_record: Union[Mapping[str, Any], _NoRecord] = _NO_RECORD
+            for record in records:
+                if isinstance(record, _DroppedEmptyRecord):
+                    dropped_placeholders.append(record)
+                    continue
+                first_record = record
+                break
+            if isinstance(first_record, _NoRecord):
+                continue
+            # Chain the peeked record back instead of restarting the extractor: the rest of the
+            # records stay lazy and the response is never read twice.
+            yield from chain(dropped_placeholders, [first_record], records)
+            return
+
+    def _records_of(
+        self, extractor_index: int, response: requests.Response
+    ) -> Iterable[Mapping[str, Any]]:
+        records = self.extractors[extractor_index].extract_records(response)
+        if not self.skip_empty_records:
+            return records
+        return self._without_empty_records(records, extractor_index)
+
+    def _without_empty_records(
+        self, records: Iterable[Mapping[str, Any]], extractor_index: int
+    ) -> Iterator[Mapping[str, Any]]:
+        dropped = 0
+        for record in records:
+            if not record:
+                if self.count_dropped_empty_records:
+                    yield _DROPPED_EMPTY_RECORD  # type: ignore[misc]  # only ever counted, never emitted
+                else:
+                    dropped += 1
+                continue
+            yield record
+        # The paginator's copy re-reads the page the record stream already read, so only the
+        # record stream reports the drop.
+        if dropped:
+            logger.warning(
+                "CombinedExtractor dropped %s empty record(s) yielded by sub-extractor %s (%s) "
+                "because `skip_empty_records` is enabled. An API that returns empty records "
+                "alongside real ones usually reports the reason in an error field of the "
+                "response body.",
+                dropped,
+                extractor_index,
+                type(self.extractors[extractor_index]).__name__,
+            )