[yaml]: fix filter sql issue (#35490)

* fix keep sql expression

* add filter sql test pipeline

Author: derrickaw

sdks/python/apache_beam/yaml/tests/filter.yaml

@@ -116,6 +116,29 @@ pipelines:
               - {transaction_id: "T0302", product_name: "Monitor", category: "Electronics", price: 249.99}
 
 
+  # Simple Filter using SQL
+  - pipeline:
+      type: chain
+      transforms:
+        - type: Create
+          config:
+            elements:
+              - {transaction_id: "T0012", product_name: "Headphones", category: "Electronics", price: 59.99}
+              - {transaction_id: "T5034", product_name: "Leather Jacket", category: "Apparel", price: 109.99}
+              - {transaction_id: "T0024", product_name: "Aluminum Mug", category: "Kitchen", price: 29.9}
+              - {transaction_id: "T0104", product_name: "Headphones", category: "Electronics", price: 59.99}
+              - {transaction_id: "T0302", product_name: "Monitor", category: "Electronics", price: 249.99}
+        - type: Filter
+          config:
+            language: sql
+            keep: category = 'Electronics' and price < 100
+        - type: AssertEqual
+          config:
+            elements:
+              - {transaction_id: "T0012", product_name: "Headphones", category: "Electronics", price: 59.99}
+              - {transaction_id: "T0104", product_name: "Headphones", category: "Electronics", price: 59.99}
+
+
   # Simple Filter with error handling
   - pipeline:
       type: composite

sdks/python/apache_beam/yaml/yaml_mapping.py

@@ -724,7 +724,7 @@ def _PyJsMapToFields(
 @beam.ptransform.ptransform_fn
 def _SqlFilterTransform(pcoll, sql_transform_constructor, keep, language):
   return pcoll | sql_transform_constructor(
-      f'SELECT * FROM PCOLLECTION WHERE {keep}')
+      f"SELECT * FROM PCOLLECTION WHERE {keep.get('expression')}")
 
 
 @beam.ptransform.ptransform_fn