[YAML] Fix WindowInto day duration parsing and ExtractWindowingInfo duplicate field check (#40902)

CHANGES.md

@@ -82,6 +82,7 @@
 ## Bugfixes
 
 * (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)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/python/apache_beam/yaml/yaml_mapping.py

@@ -950,13 +950,14 @@ def _ExtractWindowingInfo(
           'Fields must be a mapping or iterable of strings, got {fields}')
 
   existing_fields = named_fields_from_element_type(pcoll.element_type)
+  existing_field_names = {name for name, _ in existing_fields}
   new_fields = []
   for field, value in fields.items():
     if value not in _WINDOWING_INFO_TYPES:
       raise ValueError(
           f'{value} is not a valid windowing parameter; '
           f'must be one of {list(_WINDOWING_INFO_TYPES.keys())}')
-    elif field in existing_fields:
+    elif field in existing_field_names:
       raise ValueError(f'Input schema already has a field named {field}.')
     else:
       new_fields.append((field, _WINDOWING_INFO_TYPES[value]))

sdks/python/apache_beam/yaml/yaml_mapping_test.py

@@ -528,6 +528,19 @@ def test_extract_windowing_info(self):
                       True, True, 'UNKNOWN', 0, 0)),
           ]))
 
+  def test_extract_windowing_info_duplicate_field(self):
+    with self.assertRaisesRegex(
+        ValueError, 'Input schema already has a field named timestamp'):
+      with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+          pickle_library='cloudpickle')) as p:
+        elements = p | beam.Create([beam.Row(timestamp=1, value='a')])
+        _ = elements | YamlTransform(
+            '''
+            type: ExtractWindowingInfo
+            config:
+                fields: [timestamp]
+            ''')
+
   def test_extract_windowing_info_iterable(self):
     T = typing.TypeVar('T')
     with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(

sdks/python/apache_beam/yaml/yaml_provider.py

@@ -1136,7 +1136,7 @@ def expand(self, pcoll):
     @staticmethod
     def _parse_duration(value, name):
       time_units = {
-          'ms': 0.001, 's': 1, 'm': 60, 'h': 60 * 60, 'd': 60 * 60 * 12
+          'ms': 0.001, 's': 1, 'm': 60, 'h': 60 * 60, 'd': 60 * 60 * 24
       }
       value, suffix = re.match(r'^(.*?)([^\d]*)$', str(value)).groups()
       # Default to seconds if time unit suffix is not defined

sdks/python/apache_beam/yaml/yaml_transform_test.py

@@ -1247,6 +1247,35 @@ def test_windowing_on_outer(self):
           providers=TEST_PROVIDERS)
       assert_that(result, equal_to([6, 9]))
 
+  def test_parse_duration(self):
+    parse = yaml_provider.YamlProviders.WindowInto._parse_duration
+    self.assertEqual(parse('500ms', 'size'), 0.5)
+    self.assertEqual(parse('10', 'size'), 10)
+    self.assertEqual(parse('10s', 'size'), 10)
+    self.assertEqual(parse('2m', 'size'), 120)
+    self.assertEqual(parse('1h', 'size'), 3600)
+    self.assertEqual(parse('1d', 'size'), 86400)
+    self.assertEqual(parse('2d', 'size'), 172800)
+
+  def test_windowing_days(self):
+    with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+        pickle_library='cloudpickle')) as p:
+      result = p | YamlTransform(
+          '''
+          type: chain
+          transforms:
+            - type: CreateTimestamped
+              config:
+                  elements: [0, 50000, 90000]
+            - type: WindowInto
+              windowing:
+                type: fixed
+                size: 1d
+            - type: SumGlobally
+          ''',
+          providers=TEST_PROVIDERS)
+      assert_that(result, equal_to([50000, 90000]))
+
   def test_assign_timestamps(self):
     with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
         pickle_library='cloudpickle')) as p: