don't set fixed inputs under input bindings

Signed-off-by: Paul Dittamo <[email protected]>

flytekit/core/array_node.py

@@ -126,17 +126,19 @@ def interface(self) -> _interface_models.TypedInterface:
     @property
     def bindings(self) -> List[_literal_models.Binding]:
         # Required in get_serializable_node
-        return self._bindings
+        bindings = []
+        for binding in self._bindings:
+            if binding.var not in self._bound_inputs:
+                bindings.append(binding)
+        return bindings
 
     @property
     def fixed_inputs(self) -> List[_literal_models.Binding]:
-        # TODO - clean this up
+        # Required in get_serializable_node
         fixed_inputs = []
-        for binding in self.bindings:
+        for binding in self._bindings:
             if binding.var in self._bound_inputs:
                 fixed_inputs.append(binding)
-        if len(fixed_inputs) != len(self._bound_inputs):
-            raise ValueError("Error binding fixed inputs")
         return fixed_inputs
 
     @property

flytekit/core/array_node_map_task.py

@@ -371,10 +371,10 @@ def _raw_execute(self, **kwargs) -> Any:
 
 def map_task(
     target: Union[LaunchPlan, PythonFunctionTask, "FlyteLaunchPlan"],
+    fixed_inputs: Optional[Dict[str, Any]] = None,
     concurrency: Optional[int] = None,
     min_successes: Optional[int] = None,
     min_success_ratio: float = 1.0,
-    fixed_inputs: Optional[Dict[str, Any]] = None,
     **kwargs,
 ):
     """

flytekit/tools/translator.py

@@ -414,6 +414,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
         )
@@ -435,7 +436,7 @@ def get_serializable_node(
         node_model = workflow_model.Node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
-            inputs=entity.bindings,
+            inputs=entity.flyte_entity.bindings,
             fixed_inputs=entity.flyte_entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
@@ -446,6 +447,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             array_node=get_serializable_array_node_map_task(entity_mapping, settings, entity, options=options),
@@ -459,6 +461,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             task_node=workflow_model.TaskNode(
@@ -479,6 +482,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             workflow_node=workflow_model.WorkflowNode(sub_workflow_ref=wf_spec.template.id),
@@ -489,6 +493,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             branch_node=get_serializable(entity_mapping, settings, entity.flyte_entity, options=options),
@@ -502,11 +507,17 @@ def get_serializable_node(
         for b in entity.bindings:
             if b.var not in entity.flyte_entity.fixed_inputs.literals:
                 node_input.append(b)
+        # TODO clean up and explain
+        fixed_node_inputs = []
+        for b in entity.fixed_inputs:
+            if b.var not in entity.flyte_entity.fixed_inputs.literals:
+                fixed_node_inputs.append(b)
 
         node_model = workflow_model.Node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=node_input,
+            fixed_inputs=fixed_node_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             workflow_node=workflow_model.WorkflowNode(launchplan_ref=lp_spec.id),
@@ -528,6 +539,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             gate_node=gn,
@@ -540,6 +552,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             task_node=workflow_model.TaskNode(
@@ -559,6 +572,7 @@ def get_serializable_node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=entity.bindings,
+            fixed_inputs=entity.fixed_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             workflow_node=workflow_model.WorkflowNode(sub_workflow_ref=wf_spec.id),
@@ -571,11 +585,17 @@ def get_serializable_node(
         for b in entity.bindings:
             if b.var not in entity.flyte_entity.fixed_inputs.literals:
                 node_input.append(b)
+        # TODO clean up and explain
+        fixed_node_inputs = []
+        for b in entity.fixed_inputs:
+            if b.var not in entity.flyte_entity.fixed_inputs.literals:
+                fixed_node_inputs.append(b)
 
         node_model = workflow_model.Node(
             id=_dnsify(entity.id),
             metadata=entity.metadata,
             inputs=node_input,
+            fixed_inputs=fixed_node_inputs,
             upstream_node_ids=[n.id for n in upstream_nodes],
             output_aliases=[],
             workflow_node=workflow_model.WorkflowNode(launchplan_ref=entity.flyte_entity.id),
@@ -627,6 +647,7 @@ def get_serializable_array_node_map_task(
         id=entity.name,
         metadata=entity.sub_node_metadata,
         inputs=node.bindings,
+        fixed_inputs=entity.fixed_inputs,
         upstream_node_ids=[],
         output_aliases=[],
         task_node=task_node,

tests/flytekit/unit/core/test_array_node.py

@@ -154,33 +154,41 @@ def test_lp_serialization(target, overrides_metadata, upstream_nodes, fixed_inpu
     assert set(parent_node.upstream_node_ids) == set(upstream_nodes)
     assert len(parent_node.fixed_inputs) == len(fixed_inputs)
 
-    assert parent_node.inputs[0].var == "a"
+    inputs_map = {x.var: x for x in parent_node.inputs}
+    fixed_inputs_map = {x.var: x for x in parent_node.fixed_inputs}
+
     if "a" in fixed_inputs:
-        for fixed_input in parent_node.fixed_inputs:
-            if fixed_input.var == "a":
-                assert fixed_input == parent_node.inputs[0]
+        fixed_input = fixed_inputs_map["a"]
+        assert fixed_input
     else:
-        assert len(parent_node.inputs[0].binding.collection.bindings) == 3
-        for i, binding in enumerate(parent_node.inputs[0].binding.collection.bindings):
+        node_input = inputs_map["a"]
+        assert node_input
+        assert len(node_input.binding.collection.bindings) == 3
+        for i, binding in enumerate(node_input.binding.collection.bindings):
             if upstream_nodes and i == 0:
                 assert binding.promise.node_id == upstream_nodes[0]
             else:
                 assert (binding.scalar.primitive.integer is not None)
-    assert parent_node.inputs[1].var == "b"
     if "b" in fixed_inputs:
-        for fixed_input in parent_node.fixed_inputs:
-            if fixed_input.var == "b":
-                assert fixed_input == parent_node.inputs[1]
+        fixed_input = fixed_inputs_map["b"]
+        assert fixed_input
     else:
-        for binding in parent_node.inputs[1].binding.collection.bindings:
+        node_input = inputs_map["b"]
+        assert node_input
+        for binding in node_input.binding.collection.bindings:
             assert (binding.scalar.union is not None or
                     binding.scalar.primitive.integer is not None or
                     binding.scalar.primitive.string_value is not None)
-        assert len(parent_node.inputs[1].binding.collection.bindings) == 3
-    assert parent_node.inputs[2].var == "c"
-    assert len(parent_node.inputs[2].binding.collection.bindings) == 3
-    for binding in parent_node.inputs[2].binding.collection.bindings:
-        assert binding.scalar.primitive.integer is not None
+        assert len(node_input.binding.collection.bindings) == 3
+    if "c" in fixed_inputs:
+        fixed_input = fixed_inputs_map["c"]
+        assert fixed_input
+    else:
+        node_input = inputs_map["c"]
+        assert node_input
+        assert len(node_input.binding.collection.bindings) == 3
+        for binding in node_input.binding.collection.bindings:
+            assert binding.scalar.primitive.integer is not None
 
     serialized_array_node = parent_node.array_node
     assert (