(beam) branch master updated: [yaml] Fix yaml provider schema validation and merging (#31974)

[damccorm]

This is an automated email from the ASF dual-hosted git repository.

damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 6a1d917a5ee [yaml] Fix yaml provider schema validation and merging 
(#31974)
6a1d917a5ee is described below

commit 6a1d917a5ee8d8306f7b21165732dc54466c3ef9
Author: Jeff Kinard <j...@thekinards.com>
AuthorDate: Thu Jul 25 08:03:18 2024 -0400

    [yaml] Fix yaml provider schema validation and merging (#31974)
    
    Signed-off-by: Jeffrey Kinard <j...@thekinards.com>
---
 sdks/python/apache_beam/yaml/pipeline.schema.yaml | 6 ++++--
 sdks/python/apache_beam/yaml/yaml_transform.py    | 4 ++--
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/sdks/python/apache_beam/yaml/pipeline.schema.yaml 
b/sdks/python/apache_beam/yaml/pipeline.schema.yaml
index f68a7306d94..c3937e61131 100644
--- a/sdks/python/apache_beam/yaml/pipeline.schema.yaml
+++ b/sdks/python/apache_beam/yaml/pipeline.schema.yaml
@@ -168,8 +168,10 @@ $defs:
 
   providerOrProviderInclude:
     if:
-      properties:
-        include {}
+      allOf: [
+        { properties: { include: { type: string }}},
+        { required: [ "include" ] }
+      ]
     then:
       $ref: '#/$defs/providerInclude'
     else:
diff --git a/sdks/python/apache_beam/yaml/yaml_transform.py 
b/sdks/python/apache_beam/yaml/yaml_transform.py
index c1c509ebde2..ffef9bbcd8f 100644
--- a/sdks/python/apache_beam/yaml/yaml_transform.py
+++ b/sdks/python/apache_beam/yaml/yaml_transform.py
@@ -1064,5 +1064,5 @@ def expand_pipeline(
   return YamlTransform(
       pipeline_as_composite(pipeline_spec['pipeline']),
       yaml_provider.merge_providers(
-          pipeline_spec.get('providers', []), providers or
-          {})).expand(beam.pvalue.PBegin(pipeline))
+          yaml_provider.parse_providers(pipeline_spec.get('providers', [])),
+          providers or {})).expand(beam.pvalue.PBegin(pipeline))