aglinxinyuan commented on code in PR #8639:
URL: https://github.com/apache/texera/pull/8639#discussion_r4140046311
##########
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)) {
+ logicalOp.stateReferences =
+ literalReferences(objectMapper.valueToTree[ObjectNode](logicalOp)) ++
+ logicalOp.stateReferences
+ None
Review Comment:
Dropping it was a mistake — it's back in da8d62f24a with both Interval Join
cases, so your example (and the mixed case) is a compile error again.
7894d1d962 also checks each input against every enclosing block, so an input
from an outer loop's body or a sibling block is rejected too.
##########
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:
Right — the output columns are fixed from the placeholder at compile time.
#8750 marks those settings `@FixedAtCompileTime` and makes `$i` there a compile
error naming the property; I kept it separate so this PR stays about the
binding.
##########
workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala:
##########
@@ -47,6 +48,8 @@ class WorkflowCompilingService extends
Application[WorkflowCompilingServiceConfi
)
// register scala module to dropwizard default object mapper
bootstrap.getObjectMapper.registerModule(DefaultScalaModule)
+ // parse `$K` loop-variable references in typed operator properties (the
editor compiles here)
+ bootstrap.getObjectMapper.registerModule(new StateReferenceModule())
Review Comment:
Fixed in f0950a6e47: one helper, `JSONUtils.registerWorkflowModules`, used
by `JSONUtils.objectMapper`, `WorkflowCompilingService` and
`ComputingUnitMaster`, each with a test that its mapper parses a typed `"$n"`.
`TexeraWebApplication` never parses an operator, so it doesn't need it.
##########
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:
The end-of-input check is back (da8d62f24a), so an iteration with no rows
fails too. Names can't be checked before the run: a Loop Start's variables come
from Python that only runs at run time.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/StateReferenceBinding.scala:
##########
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.core.executor
+
+import com.fasterxml.jackson.core.{JsonPointer, JsonProcessingException}
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.node.{ArrayNode, JsonNodeFactory,
ObjectNode}
+import org.apache.texera.amber.core.state.StateReferencing.SIDECAR_PROPERTY
+import org.apache.texera.amber.core.state.{State, StateReferencing}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+
+import scala.collection.mutable
+import scala.jdk.CollectionConverters.IteratorHasAsScala
+import scala.util.Try
+
+/**
+ * Writes the loop variables each state message carries into one executor's
setting, the
+ * descriptor it holds, in place, at the placeholders its `stateReferences`
sidecar names. Each
+ * value is coerced to the placeholder's JSON type. A later message's value
replaces an earlier
+ * one's, so an inner loop's variable shadows an outer one of the same name.
A value that cannot be
+ * written leaves the property as it was and is reported by `unbound`, unless
a later one can be.
+ *
+ * @param className the executor's class, named in the errors.
+ */
+private[executor] final class StateReferenceBinding(
+ className: String,
+ setting: StateReferencing,
+ references: Map[String, String]
+) {
+
+ /** Why each pointer is not written yet; a pointer is removed once a value
is written to it. */
+ private val problems: mutable.SortedMap[String, String] =
mutable.SortedMap.from(references.map {
+ case (pointer, name) =>
+ pointer -> s"property $pointer refers to loop variable $name, but no
state message carried it"
+ })
+
+ /** Writes each referenced variable `state` carries into the setting. */
+ def write(state: State): Unit =
Review Comment:
Now an error (03e82e3033): copies of one loop's state that disagree fail
with `loop variable i got two different values in one iteration, 1 and 2`. A
deeper loop's value still shadows an outer one's, using the message's
`loopCounter`.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/ExecFactory.scala:
##########
@@ -29,11 +33,62 @@ object ExecFactory {
.asInstanceOf[OperatorExecutor]
}
+ /**
+ * When the descString's `stateReferences` sidecar names loop variables,
the descriptor the
+ * executor holds is its setting, which each state message writes them into
+ * (`OperatorExecutor.registerState`). The compiler fills the sidecar in
only inside a block.
+ */
def newExecFromJavaClassName[K](
className: String,
descString: String = "",
idx: Int = 0,
workerCount: Int = 1
+ ): OperatorExecutor = {
+ val executor = instantiate[K](className, descString, idx, workerCount)
+ val references = StateReferenceBinding.sidecarOf(descString)
+ if (references.nonEmpty) {
+ def refuse(reason: String): Nothing = {
+ val named = references.toSeq.sorted.map { case (pointer, name) =>
s"$pointer -> $$$name" }
+ throw new IllegalStateException(
+ s"$className refers to loop variables (${named.mkString(", ")}), but
it holds $reason"
+ )
+ }
+ settingsHeldBy(executor) match {
Review Comment:
Done in 2c7287e133 — `OperatorExecutor`'s scaladoc now states both rules.
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala:
##########
@@ -86,6 +88,42 @@ case class LogicalPlan(
links.filter(l => l.toOpId == opId)
}
+ /**
+ * The operators inside some loop block: on a path LoopStart -> ... ->
operator -> ... -> LoopEnd
+ * whose two ends match. A control operator is not inside its own block,
but an inner block's
+ * are inside the outer one. The frontend's `getEnclosingLoopStarts`
(loop-block.util.ts), which
Review Comment:
Fixed the doc (0f6bbb39d9). `LogicalPlanSpec` now checks membership and
enclosing Loop Starts against a port of #8640's walk, on every case plus 400
random DAGs — they agree. One caveat: the panel sees the whole canvas, the
compiler only the plan without disabled operators.
##########
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:
Right — #8750 refuses modifying an operator that refers to loop variables
while the workflow runs.
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala:
##########
@@ -317,6 +329,26 @@ class LoopIntegrationSpec
)
}
+ it should "bind a $i loop-variable reference in a JVM operator's property on
every iteration" in {
Review Comment:
c823b4b9b7 adds the three runs: a Limit with `$n`, nested loops that both
name `i`, and two branches meeting before a `$i` Filter. The compile and mapper
cases are unit tests (`WorkflowCompilerSpec`, `ComputingUnitMasterSpec`,
`WorkflowCompilingServiceRunSpec`).
--
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]
