Airbyte CDK: add filter to RemoveFields

airbyte-cdk/python/airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1847,6 +1847,8 @@ definitions:
       type:
         type: string
         enum: [RemoveFields]
+      condition:
+        type: string
       field_pointers:
         title: Field Paths
         description: Array of paths defining the field to remove. Each item is an array whose field describe the path of a field to remove.

airbyte-cdk/python/airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -585,6 +585,13 @@ class RemoveFields(BaseModel):
         examples=[['tags'], [['content', 'html'], ['content', 'plain_text']]],
         title='Field Paths',
     )
+    condition: Optional[str] = Field(
+        '',
+        description='The predicate to filter a property value. Property will be removed if evaluated to True.',
+        examples=[
+            "{{ property|string == '' }}",
+        ],
+    )
 
 
 class RequestPath(BaseModel):

airbyte-cdk/python/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -909,7 +909,7 @@ def create_record_selector(
 
     @staticmethod
     def create_remove_fields(model: RemoveFieldsModel, config: Config, **kwargs: Any) -> RemoveFields:
-        return RemoveFields(field_pointers=model.field_pointers, parameters={})
+        return RemoveFields(field_pointers=model.field_pointers, condition=model.condition or "", parameters={})
 
     def create_selective_authenticator(self, model: SelectiveAuthenticatorModel, config: Config, **kwargs: Any) -> DeclarativeAuthenticator:
         authenticators = {name: self._create_component_from_model(model=auth, config=config) for name, auth in model.authenticators.items()}

airbyte-cdk/python/airbyte_cdk/sources/declarative/transformations/remove_fields.py

@@ -7,6 +7,7 @@
 
 import dpath.exceptions
 import dpath.util
+from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.transformations import RecordTransformation
 from airbyte_cdk.sources.declarative.types import Config, FieldPointer, StreamSlice, StreamState
 
@@ -40,6 +41,10 @@ class RemoveFields(RecordTransformation):
 
     field_pointers: List[FieldPointer]
     parameters: InitVar[Mapping[str, Any]]
+    condition: str = ""
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._filter_interpolator = InterpolatedBoolean(condition=self.condition, parameters=parameters)
 
     def transform(
         self,
@@ -55,7 +60,11 @@ def transform(
         for pointer in self.field_pointers:
             # the dpath library by default doesn't delete fields from arrays
             try:
-                dpath.util.delete(record, pointer)
+                dpath.util.delete(
+                    record,
+                    pointer,
+                    afilter=(lambda x: self._filter_interpolator.eval(config or {}, property=x)) if self.condition else None,
+                )
             except dpath.exceptions.PathNotFound:
                 # if the (potentially nested) property does not exist, silently skip
                 pass