[GitHub] [beam] lukecwik commented on a diff in pull request #25394: Validate that GBK coders are always set correctly.

["lukecwik (via GitHub)" <gi...@apache.org>]

lukecwik commented on code in PR #25394:
URL: https://github.com/apache/beam/pull/25394#discussion_r1101779443


##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1852,3 +1853,46 @@ def visit_transform(self, transform_node):
               self.deterministic_key_coders and transform_node.full_label)
 
   return GroupByKeyInputVisitor(deterministic_key_coders)
+
+
+def validate_pipeline_graph(pipeline_proto):
+  """Ensures this is a correctly constructed Beam pipeline.
+  """
+  def get_coder(pcoll_id):
+    return pipeline_proto.components.coders[
+        pipeline_proto.components.pcollections[pcoll_id].coder_id]
+
+  def validate_transform(transform_id):
+    transform_proto = pipeline_proto.components.transforms[transform_id]
+
+    # Currently the only validation we perform is that GBK operations have
+    # their coders set properly.
+    if transform_proto.spec.urn == common_urns.primitives.GROUP_BY_KEY.urn:
+      if len(transform_proto.inputs) != 1:
+        raise ValueError("Unexpected number of inputs: %s" % transform_proto)
+      if len(transform_proto.outputs) != 1:
+        raise ValueError("Unexpected number of outputs: %s" % transform_proto)
+      input_coder = get_coder(next(iter(transform_proto.inputs.values())))
+      output_coder = get_coder(next(iter(transform_proto.outputs.values())))
+      if input_coder.spec.urn != common_urns.coders.KV.urn:
+        raise ValueError(
+            "Bad coder for input of %s: %s" % (transform_id, input_coder))
+      if output_coder.spec.urn != common_urns.coders.KV.urn:
+        raise ValueError(
+            "Bad coder for output of %s: %s" % (transform_id, input_coder))
+      output_values_coder = pipeline_proto.components.coders[
+          output_coder.component_coder_ids[1]]
+      if (input_coder.component_coder_ids[0] !=
+          output_coder.component_coder_ids[0] or
+          output_values_coder.spec.urn != common_urns.coders.ITERABLE.urn or
+          output_values_coder.component_coder_ids[0] !=
+          input_coder.component_coder_ids[1]):
+        raise ValueError(
+            "Incompatable input and output coders for %s: %s" %

Review Comment:
   Fix typo in string substitution
   ```suggestion
               "Incompatible input coder %s and output coder %s for transform %s" %
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org