From 311863576c263cccae3fc428b491d693662e9591 Mon Sep 17 00:00:00 2001
From: Jignesh Shah <jkshah@google.com>
Date: Fri, 6 Nov 2026 17:10:37 +0000
Subject: [PATCH] [Python/YAML] Accept >=, <=, == and != in generic Filter
 expressions (#41043)

validate_generic_expression split comparisons with a greedy regex, so a
two-character operator lost its first character to the left operand
('age >= 21' became 'age >', '=', ' 21') and the expression was rejected.
Only < and > worked. Make the left operand non-greedy and add tests for
every operator.

Fixes #41043.
---
 CHANGES.md                                    |  1 +
 sdks/python/apache_beam/yaml/yaml_mapping.py  |  4 +-
 .../apache_beam/yaml/yaml_mapping_test.py     | 53 +++++++++++++++++++
 3 files changed, 57 insertions(+), 1 deletion(-)

diff --git a/CHANGES.md b/CHANGES.md
index 1eb1229261ad..66755451809c 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -127,6 +127,7 @@
 
 ## Bugfixes
 
+* (Python/YAML) Beam YAML `generic` filters now accept `>=`, `<=`, `==` and `!=`; previously only `<` and `>` were recognized ([#41043](https://github.com/apache/beam/issues/41043)).
 * (Python) RunInference now measures model memory (the `model_byte_size` metric) when `torch` is not installed. It was always 0, because a failed `ModelManager` import also disabled the `resource` module ([#40854](https://github.com/apache/beam/issues/40854)).
 * (Python/YAML) Fixed `WindowInto` `'d'` (day) duration unit parsing (`24h` instead of `12h`) and fixed `ExtractWindowingInfo` duplicate field name validation ([#40902](https://github.com/apache/beam/issues/40902)).
 * (Python) Fixed `GeminiModelHandler` returning only one result per batch: each prompt is now sent as its own request (up to 32 at a time), so every input gets its own result. A custom `request_fn` must now return one response per input ([#40913](https://github.com/apache/beam/issues/40913)).
diff --git a/sdks/python/apache_beam/yaml/yaml_mapping.py b/sdks/python/apache_beam/yaml/yaml_mapping.py
index 55871b3d3040..d0ac375c5fb6 100644
--- a/sdks/python/apache_beam/yaml/yaml_mapping.py
+++ b/sdks/python/apache_beam/yaml/yaml_mapping.py
@@ -125,7 +125,9 @@ def is_atomic(expr: str):
     return
 
   if allow_cmp:
-    maybe_cmp = re.fullmatch('(.*)([<>=!]+)(.*)', expr)
+    # The left side is non-greedy so that two-character operators such as
+    # '>=' stay whole; a greedy match would split 'a >= 1' into 'a >', '='.
+    maybe_cmp = re.fullmatch('(.*?)([<>=!]+)(.*)', expr)
     if maybe_cmp:
       left, cmp, right = maybe_cmp.groups()
       if (is_atomic(left.strip()) and is_atomic(right.strip()) and
diff --git a/sdks/python/apache_beam/yaml/yaml_mapping_test.py b/sdks/python/apache_beam/yaml/yaml_mapping_test.py
index 2add3e38e7bc..a742c2f95109 100644
--- a/sdks/python/apache_beam/yaml/yaml_mapping_test.py
+++ b/sdks/python/apache_beam/yaml/yaml_mapping_test.py
@@ -571,6 +571,59 @@ def test_extract_windowing_info_iterable(self):
                   window_type='GlobalWindow'),
           ]))
 
+  def test_filter_generic(self):
+    with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+        pickle_library='cloudpickle')) as p:
+      elements = p | beam.Create(DATA)
+      result = elements | YamlTransform(
+          '''
+          type: Filter
+          config:
+              language: generic
+              keep: "rank >= 1"
+          ''')
+      assert_that(
+          result
+          | beam.Map(lambda named_tuple: beam.Row(**named_tuple._asdict())),
+          equal_to([
+              beam.Row(label='37a', conductor=37, rank=1),
+              beam.Row(label='389a', conductor=389, rank=2),
+          ]))
+
+
+class ValidateGenericExpressionTest(unittest.TestCase):
+  FIELDS = ['age', 'name']
+
+  def test_accepts_every_comparison_operator(self):
+    for op in ('==', '!=', '<', '<=', '>', '>='):
+      with self.subTest(op=op):
+        yaml_mapping.validate_generic_expression({'expression': f'age {op} 21'},
+                                                 self.FIELDS,
+                                                 True,
+                                                 'keep')
+
+  def test_accepts_comparisons_without_spaces(self):
+    yaml_mapping.validate_generic_expression({'expression': 'age>=21'},
+                                             self.FIELDS,
+                                             True,
+                                             'keep')
+
+  def test_rejects_non_atomic_sides(self):
+    for expr in ('age + 1 >= 21', 'unknown >= 21', 'age >= 21 >= 1'):
+      with self.subTest(expr=expr):
+        with self.assertRaises(ValueError):
+          yaml_mapping.validate_generic_expression({'expression': expr},
+                                                   self.FIELDS,
+                                                   True,
+                                                   'keep')
+
+  def test_rejects_comparisons_when_not_allowed(self):
+    with self.assertRaises(ValueError):
+      yaml_mapping.validate_generic_expression({'expression': 'age >= 21'},
+                                               self.FIELDS,
+                                               False,
+                                               'field')
+
 
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)