diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
index b0eec7d3d..21908ac09 100644
--- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
+++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml
@@ -2983,6 +2983,10 @@ definitions:
           - "$ref": "#/definitions/SchemaNormalization"
           - "$ref": "#/definitions/CustomSchemaNormalization"
         default: None
+      transform_before_filtering:
+        description: If true, transformation will be applied before record filtering.
+        type: boolean
+        default: false
       $parameters:
         type: object
         additionalProperties: true
diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
index a49b66c03..2c6d386c5 100644
--- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
+++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py
@@ -1771,6 +1771,10 @@ class RecordSelector(BaseModel):
         description="Responsible for normalization according to the schema.",
         title="Schema Normalization",
     )
+    transform_before_filtering: Optional[bool] = Field(
+        False,
+        description="If true, transformation will be applied before record filtering.",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
index c46c76cef..8f91719c1 100644
--- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
+++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
@@ -2564,7 +2564,9 @@ def create_record_selector(
             else None
         )
 
-        transform_before_filtering = False
+        assert model.transform_before_filtering is not None  # for mypy
+
+        transform_before_filtering = model.transform_before_filtering
         if client_side_incremental_sync:
             record_filter = ClientSideIncrementalRecordFilterDecorator(
                 config=config,