Fix to manage fixed shard dependencies on a single non scatter/gather/sharded step

CHANGELOG.rst

@@ -3,6 +3,11 @@
 Change Log
 ==========
 
+3.15.0
+=====
+* fix to manage fixed shard dependencies on a single non scatter/gather/sharded step
+
+
 3.14.0
 =====
 * Add `purge_meta_workflow_run` function to `wrangler_utils` to delete a MetaWorkflowRun and its associated files.

magma/metawfl.py

@@ -425,8 +425,20 @@ def write_run(self, input_structure, end_steps=[]):
                                 run_step_['dependencies'].append('{0}:{1}'.format(dependency, ':'.join(s_g)))
                             #end if
                         #end for
-                    else:
-                        run_step_['dependencies'].append('{0}:{1}'.format(dependency, ':'.join(s)))
+                    else: # No gather, normal dependency
+                        # Choose dependency shard:
+                        # - if producer has fixed_shards and it's exactly one, always use that one
+                        # - else if producer is scattered, align to current shard `s`
+                        # - else (producer not scattered), use ['0']
+                        if dependency in fixed_shards:
+                            dep_fixed = fixed_shards[dependency]   # e.g. [['0']] or [['0'], ['1'], ...]
+                            dep_shard = dep_fixed[0] if len(dep_fixed) == 1 else s
+                        elif dependency in scatter:
+                            dep_shard = s
+                        else:
+                            dep_shard = ['0']
+                        # Add dependency with shard
+                        run_step_['dependencies'].append('{0}:{1}'.format(dependency, ':'.join(dep_shard)))
                     #end if
                 #end for
                 run_json['workflow_runs'].append(run_step_)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "magma-suite"
-version = "3.14.0"
+version = "3.15.0"
 description = "Collection of tools to manage meta-workflows automation."
 authors = ["Michele Berselli <berselli.michele@gmail.com>", "Doug Rioux", "Soo Lee", "CGAP team"]
 license = "MIT"

test/files/test_METAWFL_shards_fixed.json

@@ -0,0 +1,52 @@
+{
+  "accession": "ACCID",
+  "uuid": "test-uuid",
+  "input": [],
+  "workflows": [
+    {
+      "name": "A",
+      "workflow": "A-UUID",
+      "config": {},
+      "input": [
+        {
+          "argument_name": "INPUT_A",
+          "argument_type": "file"
+        }
+      ],
+      "outputs": ["OUTPUT_A"]
+    },
+    {
+      "name": "B",
+      "workflow": "B-UUID",
+      "config": {},
+      "shards": [["0"], ["1"], ["2"], ["3"]],
+      "input": [
+        {
+          "argument_name": "INPUT_B",
+          "argument_type": "file",
+          "source": "A"
+        }
+      ],
+      "outputs": ["OUTPUT_B"]
+    },
+    {
+      "name": "C",
+      "workflow": "C-UUID",
+      "config": {},
+      "input": [
+        {
+          "argument_name": "INPUT_C",
+          "argument_type": "file",
+          "source": "A"
+        },
+        {
+          "argument_name": "INPUT_CC",
+          "argument_type": "file",
+          "source": "B",
+          "gather_input": 1
+        }
+      ],
+      "outputs": ["OUTPUT_C"]
+    }
+  ]
+}

test/test_metawfl.py

@@ -363,3 +363,43 @@ def test_wfl_fixed_shards_gather_input_error():
     with pytest.raises(ValueError) as e:
         wfl_obj = wfl.MetaWorkflow(data)
     assert str(e.value) == 'JSON validation error, gather and gather_input can\'t be used together in the same step\n'
+
+def test_wfl_fixed_shards_gather_input_no_gather():
+    # Results expected
+    results = {
+     'meta_workflow': 'test-uuid',
+     'workflow_runs': [
+      # A
+      {'name': 'A', 'status': 'pending', 'shard': '0'},
+      # B ("shards": [["0"], ["1"], ["2"], ["3"]])
+      {'name': 'B',
+       'status': 'pending',
+       'shard': '0',
+       'dependencies': ['A:0']},
+      {'name': 'B',
+       'status': 'pending',
+       'shard': '1',
+       'dependencies': ['A:0']},
+      {'name': 'B',
+       'status': 'pending',
+       'shard': '2',
+       'dependencies': ['A:0']},
+      {'name': 'B',
+       'status': 'pending',
+       'shard': '3',
+       'dependencies': ['A:0']},
+       # C (source [A], "gather_input": 1 [B])
+      {'name': 'C',
+       'status': 'pending',
+       'shard': '0',
+       'dependencies': ['A:0', 'B:0', 'B:1', 'B:2', 'B:3']}
+       ],
+     'input': [],
+     'final_status': 'pending'}
+    # Run test
+    with open('test/files/test_METAWFL_shards_fixed.json') as json_file:
+        data = json.load(json_file)
+    wfl_obj = wfl.MetaWorkflow(data)
+    x = wfl_obj.write_run(['f1', 'f2'])
+    assert x == results
+#end def