add multiline windowing config support back

Author: derrickaw

sdks/python/apache_beam/yaml/yaml_transform.py

@@ -1030,9 +1030,25 @@ def push_windowing_to_roots(spec):
 def preprocess_windowing(spec):
   if spec['type'] == 'WindowInto':
     # This is the transform where it is actually applied.
+
     if 'windowing' in spec:
       spec['config'] = spec.get('config', {})
       spec['config']['windowing'] = spec.pop('windowing')
+
+    if spec.get('config', {}).get('windowing'):
+      windowing_config = spec['config']['windowing']
+      if isinstance(windowing_config, str):
+        try:
+          # PyYAML can load a JSON string - one-line and multi-line.
+          # Without this code, multi-line is not supported.
+          parsed_config = yaml.safe_load(windowing_config)
+          if not isinstance(parsed_config, dict):
+            raise TypeError('Windowing config string must be a YAML/JSON map.')
+          spec['config']['windowing'] = parsed_config
+        except Exception as e:
+          raise ValueError(
+              f'Error parsing windowing config string at \
+                {identify_object(spec)}: {e}') from e
     return spec
   elif 'windowing' not in spec:
     # Nothing to do.

sdks/python/apache_beam/yaml/yaml_transform_test.py

@@ -993,7 +993,7 @@ def test_explicit_window_into(self):
           providers=TEST_PROVIDERS)
       assert_that(result, equal_to([6, 9]))
 
-  def test_explicit_window_into_with_json_string_config(self):
+  def test_explicit_window_into_with_json_string_config_one_line(self):
     with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
         pickle_library='cloudpickle')) as p:
       result = p | YamlTransform(
@@ -1011,6 +1011,25 @@ def test_explicit_window_into_with_json_string_config(self):
           providers=TEST_PROVIDERS)
       assert_that(result, equal_to([6, 9]))
 
+  def test_explicit_window_into_with_json_string_config_multi_line(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, 1, 2, 3, 4, 5]
+            - type: WindowInto
+              config:
+                windowing: |
+                  {"type": "fixed", "size": "4s"}
+            - type: SumGlobally
+          ''',
+          providers=TEST_PROVIDERS)
+      assert_that(result, equal_to([6, 9]))
+
   def test_explicit_window_into_with_string_config_fails(self):
     with self.assertRaisesRegex(ValueError, 'Error parsing windowing config'):
       with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(