This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 436b37e9b6 fix(workflow-operator): disallow multiple links into Loop
Start's input port (#7154)
436b37e9b6 is described below
commit 436b37e9b62e6be2ef8ea911ec30387dd7e80415
Author: Xinyuan Lin <[email protected]>
AuthorDate: Sat Aug 1 18:38:55 2026 -0700
fix(workflow-operator): disallow multiple links into Loop Start's input
port (#7154)
### What changes were proposed in this PR?
A Loop Start whose single input port is fed by **two** upstream
operators is accepted by the GUI, then rejected at `StartWorkflow`:
```
requirement failed: Loop Start input port ... expected exactly one reader
URI, got 2
```
Nothing in the editor hints the plan is invalid until the run fails
([discussion #6966](https://github.com/apache/texera/discussions/6966)).
The restriction itself is intended — fan-in belongs in a `Union` before
the loop — but it should be visible while building the workflow.
`InputPort` already has a `disallowMultiLinks` flag, and the frontend
honors it in two places:
| Guard | Where | Source of the flag |
|---|---|---|
| Editor refuses to draw a second link into the port |
`workflow-editor.component.ts` | operator metadata
(`additionalMetadata.inputPorts[i].disallowMultiLinks`) |
| Workflow validation requires exactly one input |
`validation-workflow.service.ts` | the operator predicate, via
`WorkflowUtilService.inputPortToPortDescription` mapping
`disallowMultiLinks` → `disallowMultiInputs` |
The loop operators simply never set it. This PR sets it on the shared
`LoopOpDesc` input port — **one line**, no frontend change.
It applies to **Loop Start only**. Loop End supports fan-in — a loop
body may branch and converge on it — so this PR also makes that work:
every reader on the input port replays its own branch's copy of the same
iteration's state, so `MainLoop` now consumes it once per iteration and
drops the duplicates (the copies are identical, being one emission from
the matching Loop Start, and a consume emits nothing downstream).
Without that, `update` would run once per branch and the loop would end
early with wrong results.
| | inbound links | why |
|---|---|---|
| Loop Start | exactly 1 | the scheduler resolves the loop's bookkeeping
URIs from that port's single reader — put a `Union` before the loop |
| Loop End | 1 or more | a branching loop body converges here; the loop
state is consumed once per iteration |
The scheduler's `require` (which is Loop-Start-only, under
`filter(_.isLoopStart)`) stays as a defense-in-depth backstop for
programmatically built plans, which bypass the GUI entirely.
**Scope of the guard (corrected).** The two frontend guards read the
flag from different places, so they cover different cases:
| Guard | Reads from | Covers already-saved loop operators? |
|---|---|---|
| Editor refuses to draw a 2nd link
(`workflow-editor.component.ts:1124`) | dynamic **schema** | **yes** |
| Validation requires exactly 1 input
(`validation-workflow.service.ts:328`) | the saved **operator
predicate** | **no** — `updateOperatorVersion` only rebuilds ports from
the schema when the saved port list is empty, so an operator saved
before this change keeps `disallowMultiInputs: false` |
So this stops new second links everywhere (the valuable half), but a
workflow that *already* has two links — including JSON assembled outside
the GUI and then opened — still validates clean and still fails at
`StartWorkflow`. Making validation cover existing content needs
`inputPortToPortDescription` re-applied on load, or a schema fallback in
validation; that is a separate change.
### Any related issues, documentation, discussions?
Closes #7155
Closes #7246
Addresses the GUI half of [discussion
#6966](https://github.com/apache/texera/discussions/6966).
### How was this PR tested?
`LoopStartOpDescSpec` and `LoopEndOpDescSpec` each gain a case pinning
`inputPorts.head.disallowMultiLinks shouldBe true` (both fail before the
change, 29/29 pass after). `scalafmtCheckAll` + `scalafixAll --check`
clean on Java 17.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <[email protected]>
---
amber/src/main/python/core/runnables/main_loop.py | 18 +++++++++++
.../test/python/core/runnables/test_main_loop.py | 37 ++++++++++++++++++++++
.../texera/amber/operator/loop/LoopOpDesc.scala | 17 +++++++++-
.../amber/operator/loop/LoopStartOpDesc.scala | 5 +++
.../amber/operator/loop/LoopEndOpDescSpec.scala | 7 ++++
.../amber/operator/loop/LoopOpDescSpecMixin.scala | 12 +++++++
.../amber/operator/loop/LoopStartOpDescSpec.scala | 4 +++
.../validation/validation-workflow.service.spec.ts | 37 ++++++++++++++++++++++
8 files changed, 136 insertions(+), 1 deletion(-)
diff --git a/amber/src/main/python/core/runnables/main_loop.py
b/amber/src/main/python/core/runnables/main_loop.py
index d252be934e..fe44b36044 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -87,6 +87,12 @@ class MainLoop(StoppableQueueBlockingRunnable):
# LoopEnd (loop_counter == 0) takes a state; used for the jump RPC
# and the setup-config URI lookup (context.loop_start_state_uris).
self._loop_start_id: str = ""
+ # Whether this LoopEnd already consumed its loop state this execution.
+ # A loop body may branch and converge on the Loop End, and every reader
+ # on its input port replays that branch's states independently, so the
+ # same iteration's state arrives once per branch. Workers are recreated
+ # on each region re-execution, so this instance flag is per iteration.
+ self._loop_state_consumed: bool = False
self.context = Context(worker_id, input_queue)
self._async_rpc_server = AsyncRPCServer(output_queue,
context=self.context)
@@ -377,6 +383,18 @@ class MainLoop(StoppableQueueBlockingRunnable):
# Matching LoopEnd (in_counter == 0): it will consume this state
# and jump back. Remember which LoopStart to jump to (it rides
# the envelope) for complete()/_jump_to_loop_start.
+ #
+ # With a branching loop body, each branch's reader replays the same
+ # iteration's state, so this fires once per inbound link. Consume
+ # only the first: running the user's `update` again would advance
+ # the loop variables once per branch. The copies are identical --
+ # they are the same state emitted by the one matching LoopStart --
+ # so dropping them loses nothing (a consume emits nothing
+ # downstream either way).
+ if self._loop_state_consumed:
+ self._check_and_process_control()
+ return
+ self._loop_state_consumed = True
self._loop_start_id = frame.loop_start_id
self.context.state_processing_manager.current_input_state = state
diff --git a/amber/src/test/python/core/runnables/test_main_loop.py
b/amber/src/test/python/core/runnables/test_main_loop.py
index 78ec517635..73f3957559 100644
--- a/amber/src/test/python/core/runnables/test_main_loop.py
+++ b/amber/src/test/python/core/runnables/test_main_loop.py
@@ -2495,6 +2495,43 @@ class TestMainLoop:
assert "loop_start_id" not in passed_to_operator
assert "loop_counter" not in passed_to_operator
+ def test_loopend_consumes_its_loop_state_once_per_iteration(
+ self, main_loop, monkeypatch
+ ):
+ # A loop body may branch and converge on the Loop End, so its input
+ # port takes fan-in. Every reader on that port replays its own
+ # branch's states, so the SAME iteration's state arrives once per
+ # branch. Only the first may be consumed: running the user's `update`
+ # again would advance the loop variables once per branch (e.g. `i += 1`
+ # twice), ending the loop early with wrong results.
+ main_loop.context.executor_manager.executor = _FalseLoopEnd()
+ emitted, switched, reset_calls = self._capture_state_emit(
+ main_loop, monkeypatch
+ )
+ monkeypatch.setattr(
+ main_loop.context.state_processing_manager,
+ "get_output_state",
+ lambda: None,
+ )
+
+ def deliver():
+ main_loop._process_state_frame(
+ StateFrame(
+ State({"i": 42}),
+ loop_counter=0,
+ loop_start_id="outer-loop",
+ )
+ )
+
+ deliver() # branch A
+ deliver() # branch B replays the same iteration's state
+
+ assert switched == [True], "the operator must consume exactly once"
+ assert emitted == [], "a consume emits nothing downstream, duplicate
or not"
+ assert reset_calls == []
+ assert main_loop._loop_start_id == "outer-loop"
+ assert main_loop._loop_state_consumed is True
+
# ------------------------------------------------------------------ #
# _jump_to_loop_start
#
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 5783cda99e..89e73261d7 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
@@ -67,6 +67,17 @@ abstract class LoopOpDesc extends LogicalOp {
@JsonIgnore
protected def isLoopStart: Boolean = false
+ /**
+ * Whether this operator's input port takes exactly one inbound link.
+ *
+ * Only Loop Start does: the scheduler resolves its loop bookkeeping URIs
+ * from that port's single reader. Loop End accepts fan-in -- a loop body
+ * may branch and converge on it -- and the runtime consumes the loop state
+ * once per iteration no matter how many branches replay it (see
+ * MainLoop._process_state_frame).
+ */
+ protected def disallowMultiInputLinks: Boolean = false
+
override def getPhysicalOp(
workflowId: WorkflowIdentity,
executionId: ExecutionIdentity
@@ -94,7 +105,11 @@ abstract class LoopOpDesc extends LogicalOp {
operatorName,
operatorDescription,
OperatorGroupConstants.CONTROL_GROUP,
- inputPorts = List(InputPort()),
+ // Declaring the single-link restriction here is what makes the GUI
+ // refuse to draw a second link, instead of the plan being rejected only
+ // at StartWorkflow (discussion #6966). It applies to Loop Start alone --
+ // see disallowMultiInputLinks.
+ inputPorts = List(InputPort(disallowMultiLinks =
disallowMultiInputLinks)),
// 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/main/scala/org/apache/texera/amber/operator/loop/LoopStartOpDesc.scala
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopStartOpDesc.scala
index 7817d7035f..aa9c1fcf5e 100644
---
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopStartOpDesc.scala
+++
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/loop/LoopStartOpDesc.scala
@@ -42,6 +42,11 @@ class LoopStartOpDesc extends LoopOpDesc {
// this operator's input-port state URI and ships it to workers at setup.
override protected def isLoopStart: Boolean = true
+ // The scheduler resolves this loop's bookkeeping URIs from this port's
+ // single reader (WorkflowExecutionManager requires exactly one storage
+ // pair), so fan-in here has no meaning -- put a Union before the loop.
+ override protected def disallowMultiInputLinks: Boolean = true
+
// `initialization` and `output` are base64-wrapped by `pyb`; see
// LoopOpDesc.generatePythonCode.
override def generatePythonCode(): String = {
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 5f4b88b82e..c8dc04688f 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,13 @@ class LoopEndOpDescSpec extends AnyFlatSpec with
LoopOpDescSpecMixin {
info.outputPorts should have length 1
}
+ it should "allow more than one link into its input port" in {
+ // A loop body may branch and converge on the Loop End, so fan-in is
+ // supported: the runtime consumes the loop state once per iteration
+ // however many branches replay it (MainLoop._process_state_frame).
+ assertDisallowsMultiInputLinks(desc(), expected = false)
+ }
+
"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/LoopOpDescSpecMixin.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopOpDescSpecMixin.scala
index 7b9eed231c..a2044eee21 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopOpDescSpecMixin.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/loop/LoopOpDescSpecMixin.scala
@@ -74,6 +74,18 @@ trait LoopOpDescSpecMixin extends Matchers {
physical.outputPorts.size shouldBe opDesc.operatorInfo.outputPorts.size
}
+ /** Pins whether the GUI may draw more than one link into this operator's
+ * input port. Only Loop Start restricts it -- the scheduler resolves its
+ * loop bookkeeping URIs from that port's single reader. Loop End accepts
+ * fan-in (a loop body may branch and converge on it); the runtime consumes
+ * the loop state once per iteration however many branches replay it.
+ */
+ protected def assertDisallowsMultiInputLinks(
+ opDesc: LogicalOp,
+ expected: Boolean
+ ): Assertion =
+ opDesc.operatorInfo.inputPorts.head.disallowMultiLinks shouldBe expected
+
protected def assertOpExecWithPythonCodeForClass(
physical: PhysicalOp,
expectedSubclassDecl: String
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 cc86c5d9d2..94451b6ed7 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,10 @@ class LoopStartOpDescSpec extends AnyFlatSpec with
LoopOpDescSpecMixin {
info.outputPorts should have length 1
}
+ it should "disallow more than one link into its input port" in {
+ assertDisallowsMultiInputLinks(desc(), expected = 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
diff --git
a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts
b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts
index 4039c441f2..3fe5944c14 100644
---
a/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts
+++
b/frontend/src/app/workspace/service/validation/validation-workflow.service.spec.ts
@@ -227,4 +227,41 @@ describe("ValidationWorkflowService", () => {
workflowActionservice.getTexeraGraph().disableOperator(mockResultPredicate2.operatorID);
expect(Object.entries(validationWorkflowService.getCurrentWorkflowValidationError().errors).length).toEqual(0);
});
+
+ it("should report an operator invalid when a disallowMultiInputs port has
two enabled links", () => {
+ // The guard behind `disallowMultiLinks` on a loop operator's input port
+ // (LoopOpDesc): a port declared single-input must have exactly one inbound
+ // link, so fanning two producers into it is invalid rather than silently
+ // accepted and failing at StartWorkflow.
+ const singleInputSink = {
+ ...mockResultPredicate,
+ operatorID: "single-input-sink",
+ inputPorts: [{ portID: "input-0", disallowMultiInputs: true }],
+ };
+ const secondSource = { ...mockScanPredicate, operatorID: "scan-2" };
+ const linkFromFirst = {
+ linkID: "link-single-input-1",
+ source: { operatorID: mockScanPredicate.operatorID, portID: "output-0" },
+ target: { operatorID: singleInputSink.operatorID, portID: "input-0" },
+ };
+ const linkFromSecond = {
+ linkID: "link-single-input-2",
+ source: { operatorID: secondSource.operatorID, portID: "output-0" },
+ target: { operatorID: singleInputSink.operatorID, portID: "input-0" },
+ };
+
+ workflowActionservice.addOperator(mockScanPredicate, mockPoint);
+ workflowActionservice.addOperator(secondSource, mockPoint);
+ workflowActionservice.addOperator(singleInputSink, mockPoint);
+
+ workflowActionservice.addLink(linkFromFirst);
+
expect(validationWorkflowService.validateOperator(singleInputSink.operatorID).isValid).toBeTruthy();
+
+ workflowActionservice.addLink(linkFromSecond);
+ const validation =
validationWorkflowService.validateOperator(singleInputSink.operatorID);
+ expect(validation.isValid).toBeFalsy();
+ if (!validation.isValid) {
+ expect(validation.messages["inputs"]).toContain("requires 1 input, has
2");
+ }
+ });
});
