aglinxinyuan commented on code in PR #8639:
URL: https://github.com/apache/texera/pull/8639#discussion_r4170970168
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala:
##########
@@ -72,6 +79,54 @@ object WorkflowCompiler {
opIdToError.toMap
}
+ /**
+ * Makes `logicalOp`'s `stateReferences` sidecar final, before its physical
plan serializes it
+ * into the executor's descString. Inside a loop block it adds every
whole-string `$name` value
+ * of the operator's JSON (`StateReferencing.literalReferences`). Outside
every block a string
+ * such as "$AAPL" stays a literal, and a typed placeholder (the user wrote
"$n" for an Int),
+ * which nothing would bind, is an error; it stays in the sidecar, so a
recompile reports it too.
+ */
+ def normalizeStateReferences(
+ logicalOp: LogicalOp,
+ insideLoopBlocks: Set[OperatorIdentity]
+ ): Option[Throwable] =
+ if (insideLoopBlocks.contains(logicalOp.operatorIdentifier)) {
Review Comment:
Yes — they'll go in together: #8750 is stacked on this branch, and I'll
merge it right after this one, not this one alone. The description now says so.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala:
##########
@@ -50,4 +53,61 @@ trait OperatorExecutor {
def close(): Unit = {}
+ /**
+ * The state message most recently registered on this executor, `None`
until one arrives; inside
+ * a loop block it carries the iteration's loop variables.
+ */
+ final def state: Option[State] =
OperatorExecutor.registrationOf(this).flatMap(_.state)
+
+ /**
+ * The worker calls it for every state message, right before
`processState`. Until
+ * `bindStateReferences`, it also writes each loop variable the message
carries that the
+ * executor's setting (the descriptor it holds) refers to into that
setting, in place.
+ */
+ final def registerState(state: State): Unit = {
+ val registration = OperatorExecutor.registrationFor(this)
+ registration.state = Some(state)
+ registration.references.foreach(_.write(state))
+ }
+
+ /**
+ * The worker calls it right before this executor first sees data. It
fails, naming
+ * each reference no state message could write, and otherwise ends the
writing: later messages
+ * no longer change the setting.
+ */
+ final def bindStateReferences(): Unit =
+ OperatorExecutor.registrationOf(this).foreach { registration =>
+ registration.references.foreach { references =>
+ val unbound = references.unbound
+ if (unbound.nonEmpty) throw new
IllegalStateException(unbound.mkString("; "))
Review Comment:
Agreed — reading the names from the Loop Start's Python the way #8640 does
can be a follow-up.
--
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]
