feat(cdk): add schema filter to DynamicSchemaLoader (#550)

Co-authored-by: octavia-squidington-iii <[email protected]>

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2318,6 +2318,12 @@ definitions:
           - "$ref": "#/definitions/AsyncRetriever"
           - "$ref": "#/definitions/CustomRetriever"
           - "$ref": "#/definitions/SimpleRetriever"
+      schema_filter:
+        title: Schema Filter
+        description: Responsible for filtering fields to be added to json schema.
+        anyOf:
+          - "$ref": "#/definitions/RecordFilter"
+          - "$ref": "#/definitions/CustomRecordFilter"
       schema_transformations:
         title: Schema Transformations
         description: A list of transformations to be applied to the schema.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1,5 +1,3 @@
-# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
-
 # generated by datamodel-codegen:
 #   filename:  declarative_component_schema.yaml
 
@@ -2582,6 +2580,11 @@ class DynamicSchemaLoader(BaseModel):
         description="Component used to coordinate how records are extracted across stream slices and request pages.",
         title="Retriever",
     )
+    schema_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field(
+        None,
+        description="Responsible for filtering fields to be added to json schema.",
+        title="Schema Filter",
+    )
     schema_transformations: Optional[
         List[
             Union[

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2544,10 +2544,19 @@ def create_dynamic_schema_loader(
         schema_type_identifier = self._create_component_from_model(
             model.schema_type_identifier, config=config, parameters=model.parameters or {}
         )
+        schema_filter = (
+            self._create_component_from_model(
+                model.schema_filter, config=config, parameters=model.parameters or {}
+            )
+            if model.schema_filter is not None
+            else None
+        )
+
         return DynamicSchemaLoader(
             retriever=retriever,
             config=config,
             schema_transformations=schema_transformations,
+            schema_filter=schema_filter,
             schema_type_identifier=schema_type_identifier,
             parameters=model.parameters or {},
         )

airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py

@@ -10,6 +10,7 @@
 import dpath
 from typing_extensions import deprecated
 
+from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
@@ -126,6 +127,7 @@ class DynamicSchemaLoader(SchemaLoader):
     parameters: InitVar[Mapping[str, Any]]
     schema_type_identifier: SchemaTypeIdentifier
     schema_transformations: List[RecordTransformation] = field(default_factory=lambda: [])
+    schema_filter: Optional[RecordFilter] = None
 
     def get_json_schema(self) -> Mapping[str, Any]:
         """
@@ -151,20 +153,18 @@ def get_json_schema(self) -> Mapping[str, Any]:
             )
             properties[key] = value
 
-        transformed_properties = self._transform(properties, {})
+        filtered_transformed_properties = self._transform(self._filter(properties))
 
         return {
             "$schema": "https://json-schema.org/draft-07/schema#",
             "type": "object",
             "additionalProperties": True,
-            "properties": transformed_properties,
+            "properties": filtered_transformed_properties,
         }
 
     def _transform(
         self,
         properties: Mapping[str, Any],
-        stream_state: StreamState,
-        stream_slice: Optional[StreamSlice] = None,
     ) -> Mapping[str, Any]:
         for transformation in self.schema_transformations:
             transformation.transform(
@@ -173,6 +173,21 @@ def _transform(
             )
         return properties
 
+    def _filter(
+        self,
+        properties: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        if not self.schema_filter:
+            return properties
+
+        filtered_properties: MutableMapping[str, Any] = {}
+        for item in self.schema_filter.filter_records(
+            ({k: v} for k, v in properties.items()),
+            {},
+        ):
+            filtered_properties.update(item)
+        return filtered_properties
+
     def _get_key(
         self,
         raw_schema: MutableMapping[str, Any],

unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py

@@ -68,6 +68,10 @@
                     },
                     "paginator": {"type": "NoPagination"},
                 },
+                "schema_filter": {
+                    "type": "RecordFilter",
+                    "condition": "{{ 'filtered_field' not in record }}",
+                },
                 "schema_transformations": [
                     {
                         "type": "AddFields",
@@ -390,6 +394,7 @@ def test_dynamic_schema_loader_with_type_conditions():
                 body=json.dumps(
                     {
                         "fields": [
+                            {"name": "filtered_field", "type": "string"},
                             {"name": "Id", "type": "integer"},
                             {"name": "FirstName", "type": "string"},
                             {"name": "Description", "type": "singleLineText"},