aglinxinyuan commented on code in PR #7154:
URL: https://github.com/apache/texera/pull/7154#discussion_r3697120790
##########
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)),
Review Comment:
Confirmed — the `require` is under `filter(_.isLoopStart)`, so Loop End has
no runtime guard at all and a programmatic plan with two inbound links would
double-consume (the state is replayed per reader, so `update` and the back-edge
each run twice) with no error. Filed as #7246 with the two options: an
`isLoopEnd` marker plus a matching `require`, or a general single-reader
assertion driven by `disallowMultiLinks` that would cover future single-input
operators too. Keeping it out of this PR as you suggested.
##########
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)),
Review Comment:
You're right, and I've corrected the description. Verified
`updateOperatorVersion`: it only rebuilds ports from the schema when
`op.inputPorts.length === 0`, otherwise it keeps `op.inputPorts` as saved — so
a loop operator saved before this change keeps `disallowMultiInputs: false` and
validation stays green, including GUI-opened external JSON.
The description now spells out which guard covers what: the editor guard
reads the dynamic schema so it refuses new second links on already-saved
operators, while validation reads the saved predicate and does not. I've left
the re-application of `inputPortToPortDescription` on load (or a schema
fallback in validation) out of scope, as you suggested.
##########
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
+ }
+
Review Comment:
Done — moved to `assertSingleInputLink` on `LoopOpDescSpecMixin` alongside
the other shared PhysicalOp assertions, with the rationale in the helper; both
specs now just call it. 29/29 pass.
##########
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
+ }
+
Review Comment:
Agreed — added a `validation-workflow.service.spec.ts` case that drives the
guard instead of the constant: an operator whose input port declares
`disallowMultiInputs` is valid with one inbound link and invalid with two,
asserting the `requires 1 input, has 2` message. 7/7 in that spec.
It also makes the gap you found on the other thread explicit: the test has
to set `disallowMultiInputs` on the *predicate* for validation to see it, which
is exactly why already-saved operators aren't covered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
