From 63480b329986dcadb5b065188320bfa9a9dd199f Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Thu, 30 Jul 2026 21:01:26 -0700 Subject: [PATCH] fix(workflow-operator): disallow multiple links into a loop operator's input port A Loop Start with two upstream operators feeding its single input port was accepted by the GUI but rejected at StartWorkflow with "expected exactly one reader URI, got 2" (discussion #6966). Nothing in the editor hinted that the plan was invalid until the run failed. InputPort already has a `disallowMultiLinks` flag that the frontend honors in two places -- the editor refuses to draw a second link into such a port (workflow-editor.component.ts) and workflow validation requires exactly one input (validation-workflow.service.ts, via WorkflowUtilService.inputPortToPortDescription which maps the backend flag onto the operator predicate). The loop operators simply never set it. Set it on the shared LoopOpDesc input port, so it applies to both Loop Start and Loop End: every reader on a materialized input port replays that port's states independently, so a second link would deliver the loop state twice per iteration (double `update`, double back-edge), and a Loop Start additionally needs a single reader for the scheduler to resolve its bookkeeping URIs from. No frontend change is needed. The scheduler's `require` stays as a defense-in-depth backstop for programmatically built plans. --- .../apache/texera/amber/operator/loop/LoopOpDesc.scala | 9 ++++++++- .../texera/amber/operator/loop/LoopEndOpDescSpec.scala | 8 ++++++++ .../texera/amber/operator/loop/LoopStartOpDescSpec.scala | 9 +++++++++ 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopOpDesc.scala index 5783cda99ec..e4990194572 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopOpDesc.scala @@ -94,7 +94,14 @@ abstract class LoopOpDesc extends LogicalOp { operatorName, operatorDescription, OperatorGroupConstants.CONTROL_GROUP, - inputPorts = List(InputPort()), + // A loop operator takes exactly one link on its input port. Every reader + // on a materialized input port replays that port's states independently, + // so a second link would deliver the loop state twice per iteration; a + // Loop Start additionally needs a single reader for the scheduler to + // resolve its bookkeeping URIs from. Declaring it here is what makes the + // GUI refuse to draw the second link, instead of the plan being rejected + // only at StartWorkflow (discussion #6966). + inputPorts = List(InputPort(disallowMultiLinks = true)), // Loop End reuses its output storage across region re-executions (it // accumulates across the iterations of its own loop); the flag is // declared on the output port and the region scheduler reads it there. diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala index 5f4b88b82ed..bbdab441202 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopEndOpDescSpec.scala @@ -49,6 +49,14 @@ class LoopEndOpDescSpec extends AnyFlatSpec with LoopOpDescSpecMixin { info.outputPorts should have length 1 } + it should "disallow more than one link into its input port" in { + // Each reader on an input port replays the loop state independently, so a + // second link would make this Loop End consume the same iteration twice + // (double `update`, double back-edge). Declaring it on the port stops the + // GUI from drawing the second link (discussion #6966). + desc().operatorInfo.inputPorts.head.disallowMultiLinks shouldBe true + } + "LoopEndOpDesc.generatePythonCode" should "wrap user inputs in the base64 decode template" in { // Distinct sentinels so we know the codegen wires the right user // field into the right `decode_python_template` site. If `condition` diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala index cc86c5d9d2f..0e6e59f8012 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopStartOpDescSpec.scala @@ -46,6 +46,15 @@ class LoopStartOpDescSpec extends AnyFlatSpec with LoopOpDescSpecMixin { info.outputPorts should have length 1 } + it should "disallow more than one link into its input port" in { + // The runtime resolves the loop's bookkeeping URIs from this port's single + // reader (WorkflowExecutionManager requires exactly one storage pair), and + // every extra reader would replay the loop state again. Declaring it on + // the port is what stops the GUI from drawing a second link at all, rather + // than failing at StartWorkflow (discussion #6966). + desc().operatorInfo.inputPorts.head.disallowMultiLinks shouldBe true + } + "LoopStartOpDesc.generatePythonCode" should "wrap user inputs in the base64 decode template" in { // Distinct sentinels prove the codegen routes the right user field // through the encode pipeline (not accidentally swapped) and that