diff --git a/CHANGES.md b/CHANGES.md index cd87a9609674..b94ab739775f 100644 --- a/CHANGES.md +++ b/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 diff --git a/sdks/python/apache_beam/yaml/yaml_mapping.py b/sdks/python/apache_beam/yaml/yaml_mapping.py index 6bf2ad3fa13f..55871b3d3040 100644 --- a/sdks/python/apache_beam/yaml/yaml_mapping.py +++ b/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])) diff --git a/sdks/python/apache_beam/yaml/yaml_mapping_test.py b/sdks/python/apache_beam/yaml/yaml_mapping_test.py index 169c86d7b87b..2add3e38e7bc 100644 --- a/sdks/python/apache_beam/yaml/yaml_mapping_test.py +++ b/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( diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py index 989f49e8d967..6409c295f182 100755 --- a/sdks/python/apache_beam/yaml/yaml_provider.py +++ b/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 diff --git a/sdks/python/apache_beam/yaml/yaml_transform_test.py b/sdks/python/apache_beam/yaml/yaml_transform_test.py index ec393aefbe69..b45f68aae942 100644 --- a/sdks/python/apache_beam/yaml/yaml_transform_test.py +++ b/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: