Xiao-zhen-Liu commented on code in PR #8639:
URL: https://github.com/apache/texera/pull/8639#discussion_r4168816428
##########
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:
Fine to keep it in #8750. But #8639 on its own already produces this from
today's editor: Projection with the alias `$a` inside a loop gives a column
`$a` that is null in every row, your own example in #8750. On main the same
workflow gives a column `$a` with the values. So I'd rather not merge #8639
without #8750: can they go in together?
##########
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:
OK to leave the name check out of this PR; the error at the first tuple
names the setting and the variable. It's not impossible, though: #8640 already
reads the names from the Loop Start's Python, so the compiler could do the same
for the usual case later.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala:
##########
@@ -97,6 +100,16 @@ class DataProcessor(
def collectStatistics(): WorkerStatistics =
statisticsManager.getStatistics(executor)
+ /**
+ * `OperatorExecutor.bindStateReferences`, before the executor first sees
data; after
+ * that a reference comparison per tuple. An executor installed in its
place binds its own.
Review Comment:
OK, #8750 is the right place for it.
--
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]
