introduce a JsonSchemaPropertySelector component to allow for more flexibility to modify selected properties

Author: brianjlai

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2520,6 +2520,34 @@ definitions:
       type:
         type: string
         enum: [JsonlDecoder]
+  JsonSchemaPropertySelector:
+    title: Json Schema Property Selector
+    description: When configured, the JSON schema supplied in the catalog containing which columns are selected for the current stream will be used to reduce which query properties will be included in the outbound API request. This can improve the performance of API requests, especially for those requiring multiple requests to get a complete record.
+    type: object
+    required:
+      - type
+    properties:
+      type:
+        type: string
+        enum: [JsonSchemaPropertySelector]
+      transformations:
+        title: Transformations
+        description: A list of transformations to be applied on the customer configured schema that will be used to filter out unselected fields when specifying query properties for API requests.
+        linkable: true
+        type: array
+        items:
+          anyOf:
+            - "$ref": "#/definitions/AddFields"
+            - "$ref": "#/definitions/RemoveFields"
+            - "$ref": "#/definitions/KeysToLower"
+            - "$ref": "#/definitions/KeysToSnakeCase"
+            - "$ref": "#/definitions/FlattenFields"
+            - "$ref": "#/definitions/DpathFlattenFields"
+            - "$ref": "#/definitions/KeysReplace"
+            - "$ref": "#/definitions/CustomTransformation"
+      $parameters:
+        type: object
+        additionalProperties: true
   KeysToLower:
     title: Keys to Lower Case
     description: A transformation that renames all keys to lower case.
@@ -3410,6 +3438,10 @@ definitions:
         title: Property Chunking
         description: Defines how query properties will be grouped into smaller sets for APIs with limitations on the number of properties fetched per API request.
         "$ref": "#/definitions/PropertyChunking"
+      property_selector:
+        title: Property Selector
+        description: Defines where to look for and which query properties that should be sent in outbound API requests. For example, you can specify that only the selected columns of a stream should be in the request.
+        "$ref": "#/definitions/JsonSchemaPropertySelector"
       $parameters:
         type: object
         additionalProperties: true
@@ -3746,7 +3778,7 @@ definitions:
     properties:
       type:
         type: string
-        enum: [ PaginationReset ]
+        enum: [PaginationReset]
       action:
         type: string
         enum:
@@ -3763,7 +3795,7 @@ definitions:
     properties:
       type:
         type: string
-        enum: [ PaginationResetLimits ]
+        enum: [PaginationResetLimits]
       number_of_records:
         type: integer
   GzipDecoder:

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,3 +1,5 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2029,6 +2031,29 @@ class SessionTokenRequestApiKeyAuthenticator(BaseModel):
     )
 
 
+class JsonSchemaPropertySelector(BaseModel):
+    type: Literal["JsonSchemaPropertySelector"]
+    transformations: Optional[
+        List[
+            Union[
+                AddFields,
+                RemoveFields,
+                KeysToLower,
+                KeysToSnakeCase,
+                FlattenFields,
+                DpathFlattenFields,
+                KeysReplace,
+                CustomTransformation,
+            ]
+        ]
+    ] = Field(
+        None,
+        description="A list of transformations to be applied on the customer configured schema that will be used to filter out unselected fields when specifying query properties for API requests.",
+        title="Transformations",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class ListPartitionRouter(BaseModel):
     type: Literal["ListPartitionRouter"]
     cursor_field: str = Field(
@@ -2799,6 +2824,11 @@ class QueryProperties(BaseModel):
         description="Defines how query properties will be grouped into smaller sets for APIs with limitations on the number of properties fetched per API request.",
         title="Property Chunking",
     )
+    property_selector: Optional[JsonSchemaPropertySelector] = Field(
+        None,
+        description="Defines where to look for and which query properties that should be sent in outbound API requests. For example, you can specify that only the selected columns of a stream should be in the request.",
+        title="Property Selector",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -316,6 +316,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     JsonlDecoder as JsonlDecoderModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    JsonSchemaPropertySelector as JsonSchemaPropertySelectorModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     JwtAuthenticator as JwtAuthenticatorModel,
 )
@@ -503,6 +506,9 @@
 from airbyte_cdk.sources.declarative.requesters.query_properties.property_chunking import (
     PropertyLimitType,
 )
+from airbyte_cdk.sources.declarative.requesters.query_properties.property_selector import (
+    JsonSchemaPropertySelector,
+)
 from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import (
     GroupByKey,
 )
@@ -740,6 +746,7 @@ def _init_mappings(self) -> None:
             InlineSchemaLoaderModel: self.create_inline_schema_loader,
             JsonDecoderModel: self.create_json_decoder,
             JsonlDecoderModel: self.create_jsonl_decoder,
+            JsonSchemaPropertySelectorModel: self.create_json_schema_property_selector,
             GzipDecoderModel: self.create_gzip_decoder,
             KeysToLowerModel: self.create_keys_to_lower_transformation,
             KeysToSnakeCaseModel: self.create_keys_to_snake_transformation,
@@ -3003,7 +3010,7 @@ def create_property_chunking(
         )
 
     def create_query_properties(
-        self, model: QueryPropertiesModel, config: Config, **kwargs: Any
+        self, model: QueryPropertiesModel, config: Config, *, stream_name: str, **kwargs: Any
     ) -> QueryProperties:
         if isinstance(model.property_list, list):
             property_list = model.property_list
@@ -3020,10 +3027,43 @@ def create_query_properties(
             else None
         )
 
+        property_selector = (
+            self._create_component_from_model(
+                model=model.property_selector, config=config, stream_name=stream_name, **kwargs
+            )
+            if model.property_selector
+            else None
+        )
+
         return QueryProperties(
             property_list=property_list,
             always_include_properties=model.always_include_properties,
             property_chunking=property_chunking,
+            property_selector=property_selector,
+            config=config,
+            parameters=model.parameters or {},
+        )
+
+    def create_json_schema_property_selector(
+        self,
+        model: JsonSchemaPropertySelectorModel,
+        config: Config,
+        *,
+        stream_name: str,
+        **kwargs: Any,
+    ) -> JsonSchemaPropertySelector:
+        configured_stream = self._stream_name_to_configured_stream.get(stream_name)
+
+        transformations = []
+        if model.transformations:
+            for transformation_model in model.transformations:
+                transformations.append(
+                    self._create_component_from_model(model=transformation_model, config=config)
+                )
+
+        return JsonSchemaPropertySelector(
+            configured_stream=configured_stream,
+            properties_transformations=transformations,
             config=config,
             parameters=model.parameters or {},
         )
@@ -3251,7 +3291,7 @@ def _get_url(req: Requester) -> str:
 
             if len(query_properties_definitions) == 1:
                 query_properties = self._create_component_from_model(
-                    model=query_properties_definitions[0], config=config
+                    model=query_properties_definitions[0], stream_name=name, config=config
                 )
 
             # Removes QueryProperties components from the interpolated mappings because it has been designed
@@ -3277,11 +3317,13 @@ def _get_url(req: Requester) -> str:
 
             query_properties = self.create_query_properties(
                 model=query_properties_definition,
+                stream_name=name,
                 config=config,
             )
         elif hasattr(model.requester, "query_properties") and model.requester.query_properties:
             query_properties = self.create_query_properties(
                 model=model.requester.query_properties,
+                stream_name=name,
                 config=config,
             )
 
@@ -3318,8 +3360,6 @@ def _get_url(req: Requester) -> str:
             model.ignore_stream_slicer_parameters_on_paginated_requests or False
         )
 
-        configured_stream = self._stream_name_to_configured_stream.get(name)
-
         if (
             model.partition_router
             and isinstance(model.partition_router, SubstreamPartitionRouterModel)
@@ -3355,7 +3395,6 @@ def _get_url(req: Requester) -> str:
                 request_option_provider=request_options_provider,
                 cursor=None,
                 config=config,
-                configured_airbyte_stream=configured_stream,
                 ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
                 parameters=model.parameters or {},
             )
@@ -3377,7 +3416,6 @@ def _get_url(req: Requester) -> str:
             request_option_provider=request_options_provider,
             cursor=None,
             config=config,
-            configured_airbyte_stream=configured_stream,
             ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
             additional_query_properties=query_properties,
             log_formatter=self._get_log_formatter(log_formatter, name),

airbyte_cdk/sources/declarative/requesters/query_properties/json_schema_property_selector.py

@@ -0,0 +1,38 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from dataclasses import InitVar, dataclass, field
+from typing import Any, List, Mapping, Optional, Set
+
+from airbyte_protocol_dataclasses.models import ConfiguredAirbyteStream
+
+from airbyte_cdk.sources.declarative.transformations import RecordTransformation
+from airbyte_cdk.sources.types import Config
+
+
+@dataclass
+class JsonSchemaPropertySelector:
+    """
+    A class that contains a list of transformations to apply to properties.
+    """
+
+    configured_stream: ConfiguredAirbyteStream
+    config: Config
+    parameters: InitVar[Mapping[str, Any]]
+    properties_transformations: List[RecordTransformation] = field(default_factory=lambda: [])
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._parameters = parameters
+
+    def select(self) -> Set[str]:
+        properties = set()
+        for schema_property in self.configured_stream.stream.json_schema.get(
+            "properties", {}
+        ).keys():
+            if self.properties_transformations:
+                for transformation in self.properties_transformations:
+                    transformation.transform(
+                        schema_property,  # type: ignore  # record has type Mapping[str, Any], but Dict[str, Any] expected
+                        config=self.config,
+                    )
+            properties.add(schema_property)
+        return properties

airbyte_cdk/sources/declarative/requesters/query_properties/property_selector/__init__.py

@@ -0,0 +1,10 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from airbyte_cdk.sources.declarative.requesters.query_properties.property_selector.json_schema_property_selector import (
+    JsonSchemaPropertySelector,
+)
+from airbyte_cdk.sources.declarative.requesters.query_properties.property_selector.property_selector import (
+    PropertySelector,
+)
+
+__all__ = ["JsonSchemaPropertySelector", "PropertySelector"]

airbyte_cdk/sources/declarative/requesters/query_properties/property_selector/json_schema_property_selector.py

@@ -0,0 +1,52 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from dataclasses import InitVar, dataclass, field
+from typing import Any, List, Mapping, Optional, Set
+
+from airbyte_protocol_dataclasses.models import ConfiguredAirbyteStream
+
+from airbyte_cdk.sources.declarative.requesters.query_properties.property_selector.property_selector import (
+    PropertySelector,
+)
+from airbyte_cdk.sources.declarative.transformations import RecordTransformation
+from airbyte_cdk.sources.types import Config
+
+
+@dataclass
+class JsonSchemaPropertySelector(PropertySelector):
+    """
+    A class that contains a list of transformations to apply to properties.
+    """
+
+    config: Config
+    parameters: InitVar[Mapping[str, Any]]
+    # For other non-read operations, there is no configured catalog and therefore no schema selection
+    configured_stream: Optional[ConfiguredAirbyteStream] = None
+    properties_transformations: List[RecordTransformation] = field(default_factory=lambda: [])
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._parameters = parameters
+
+    def select(self) -> Optional[Set[str]]:
+        """
+        Returns the set of properties that have been selected for the configured stream. The intent being that
+        we should only query for selected properties not all since disabled properties are discarded.
+
+        When configured_stream is None, then there was no incoming catalog and all fields should be retrieved.
+        This is different from the empty set where the json_schema was empty and no schema fields were selected.
+        """
+
+        # For CHECK/DISCOVER operations, there is no catalog and therefor no configured stream or selected
+        # columns. In this case we return None which is interpreted by the QueryProperties component to not
+        # perform any filtering of schema properties and fetch all of them
+        if self.configured_stream is None:
+            return None
+
+        schema_properties = self.configured_stream.stream.json_schema.get("properties", {})
+        if self.properties_transformations:
+            for transformation in self.properties_transformations:
+                transformation.transform(
+                    schema_properties,  # type: ignore  # record has type Mapping[str, Any], but Dict[str, Any] expected
+                    config=self.config,
+                )
+        return set(schema_properties.keys())

airbyte_cdk/sources/declarative/requesters/query_properties/property_selector/property_selector.py

@@ -0,0 +1,24 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Optional, Set
+
+
+@dataclass
+class PropertySelector(ABC):
+    """
+    Describes the interface for selecting and transforming properties from a configured stream's schema
+    to determine which properties should be queried from the API.
+    """
+
+    @abstractmethod
+    def select(self) -> Optional[Set[str]]:
+        """
+        Selects and returns the set of properties that should be queried from the API based on the
+        configured stream's schema and any applicable transformations.
+
+        Returns:
+            Set[str]: The set of property names to query
+        """
+        pass