fix(sdk): compile ParallelFor in a deterministic manner (#4926)

* fix(sdk): compile ParallelFor in a deterministic manner

During compilataion ParallelFor components end up with randomized names,
which makes it very inconvenient to compare two versions of a pipeline.
This commit fixes this issue.

* fix(sdk): fix new parallel-for test cases

Author: vit-

sdk/python/kfp/dsl/_ops_group.py

@@ -53,7 +53,7 @@ def __init__(self, group_type: str, name: str=None, parallelism: int=None):
   def _get_matching_opsgroup_already_in_pipeline(group_type, name):
     """Retrieves the opsgroup when the pipeline already contains it.
     the opsgroup might be already in the pipeline in case of recursive calls.
-  
+
     Args:
       group_type (str): one of 'pipeline', 'exit_handler', 'condition', and 'graph'.
       name (str): the name before conversion.
@@ -193,27 +193,23 @@ class ParallelFor(OpsGroup):
   """
   TYPE_NAME = 'for_loop'
 
-  @staticmethod
-  def _get_unique_id_code():
-    return uuid.uuid4().hex[:_for_loop.LoopArguments.NUM_CODE_CHARS]
-
   def __init__(self,  loop_args: Union[_for_loop.ItemList, _pipeline_param.PipelineParam],
                parallelism: int=None):
     if parallelism and parallelism < 1:
         raise ValueError('ParallelFor parallism set to < 1, allowed values are > 0')
 
     self.items_is_pipeline_param = isinstance(loop_args, _pipeline_param.PipelineParam)
 
-    # use a random code to uniquely identify this loop
-    code = self._get_unique_id_code()
-    group_name = 'for-loop-{}'.format(code)
-    super().__init__(self.TYPE_NAME, name=group_name, parallelism=parallelism)
+    super().__init__(self.TYPE_NAME, parallelism=parallelism)
 
     if self.items_is_pipeline_param:
       loop_args = _for_loop.LoopArguments.from_pipeline_param(loop_args)
     elif not self.items_is_pipeline_param and not isinstance(loop_args, _for_loop.LoopArguments):
       # we were passed a raw list, wrap it in loop args
-      loop_args = _for_loop.LoopArguments(loop_args, code)
+      loop_args = _for_loop.LoopArguments(
+        loop_args,
+        code=str(_pipeline.Pipeline.get_default_pipeline().get_next_group_id()),
+      )
 
     self.loop_args = loop_args
 

sdk/python/tests/compiler/testdata/loop_over_lightweight_output.py

@@ -17,17 +17,6 @@
 from kfp import dsl
 from kfp.dsl import _for_loop
 
-class Coder:
-  def __init__(self, ):
-    self._code_id = 0
-
-  def get_code(self, ):
-    self._code_id += 1
-    return '{code:0{num_chars:}d}'.format(code=self._code_id, num_chars=_for_loop.LoopArguments.NUM_CODE_CHARS)
-
-
-dsl.ParallelFor._get_unique_id_code = Coder().get_code
-
 produce_op = kfp.components.load_component_from_text('''\
 name: Produce list
 outputs:
@@ -61,4 +50,4 @@ def get_code(self, ):
 def pipeline():
   source_task = produce_op()
   with dsl.ParallelFor(source_task.output) as item:
-    consume_op(item)
+    consume_op(item)

sdk/python/tests/compiler/testdata/loop_over_lightweight_output.yaml

@@ -66,7 +66,7 @@
           - "name": |-
               produce-list-data_list-loop-item
       "name": |-
-        for-loop-for-loop-00000001-1
+        for-loop-1
     - "dag":
         "tasks":
           - "arguments":
@@ -79,9 +79,9 @@
               - |-
                 produce-list
             "name": |-
-              for-loop-for-loop-00000001-1
+              for-loop-1
             "template": |-
-              for-loop-for-loop-00000001-1
+              for-loop-1
             "withParam": |-
               {{tasks.produce-list.outputs.parameters.produce-list-data_list}}
           - "name": |-

sdk/python/tests/compiler/testdata/parallelfor_item_argument_resolving.py

@@ -13,24 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import NamedTuple
-
 import kfp
 from kfp.components import func_to_container_op
 
 
-# Stabilizing the test output
-class StableIDGenerator:
-    def __init__(self, ):
-        self._index = 0
-
-    def get_next_id(self, ):
-        self._index += 1
-        return '{code:0{num_chars:}d}'.format(code=self._index, num_chars=kfp.dsl._for_loop.LoopArguments.NUM_CODE_CHARS)
-
-kfp.dsl.ParallelFor._get_unique_id_code = StableIDGenerator().get_next_id
-
-
 @func_to_container_op
 def produce_str() -> str:
     return "Hello"
@@ -77,6 +63,11 @@ def parallelfor_item_argument_resolving():
         #consume(loop_item) # Cannot use the full loop item when it's a dict
         consume(loop_item.aaa)
 
+    loop_args = [{'a': 1, 'b': 2}, {'a': 10, 'b': 20}]
+    with kfp.dsl.ParallelFor(loop_args) as loop_item:
+        consume(loop_args)
+        consume(loop_item)
+
 
 if __name__ == '__main__':
     import kfp.compiler as compiler