Xiao-zhen-Liu commented on code in PR #8639:
URL: https://github.com/apache/texera/pull/8639#discussion_r4136881516
##########
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:
A setting that changes an operator's output columns doesn't work with a loop
variable, and nothing reports an error.
The compiler decides each operator's output columns before the run, using
the placeholder (false for `$drop`), and doesn't decide them again. At run
time, the rows the operator produces are put into those columns. Example:
Projection selects `a` from a table with columns `a`, `b`, `c`, and its Drop
Option is `$drop`. Before the run, `$drop` counts as false, so Projection's
output columns are just `a`. In an iteration where `drop` is true, Projection
outputs rows with `b` and `c`, but they are put into the column `a`: `a` comes
out null in every row, and `b` and `c` are lost. The column name of Unnest
String's result has the same problem: set to `$name`, the new column is called
`$name` in every iteration.
A column name such as `$col` in Projection does fail before the run, but
with Projection's own missing-column message, which doesn't mention loop
variables.
##########
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:
This used to reject an operator that refers to a loop variable and takes an
input from outside the loop, and that check is gone now. Such an operator fails
at run time instead. Example: an Interval Join with its Interval Constant set
to `$k`, whose left table comes from outside the loop and whose right table
comes from inside it. The left table is read first, so its first row arrives
before any state, and the join fails with "no state message carried it" in
every run. With one link from inside the loop and one from outside on the same
input, whether it fails depends on which rows arrive first. Why was the check
dropped?
##########
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:
`ComputingUnitMaster` needs this too. It registers only `DefaultScalaModule`
on its Dropwizard mapper, and that mapper parses the requests of
`SyncExecutionResource`. So a workflow sent there with `"limit": "$n"` inside a
loop should fail to parse. Text settings still work, since the compiler finds
those itself.
More generally, nothing makes sure that every mapper that reads workflows
registers this module, so the next one added can miss it the same way.
##########
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:
This is the only check that every referenced variable got a value, and it
happens during the run, at the operator's first tuple. The panel (#8640) checks
names before the run, but workflows also reach the engine without the panel,
for example through `SyncExecutionResource`, and then a misspelled name only
fails once the loop is running. And since the end-of-input check was dropped,
an iteration in which the operator gets no rows is not checked at all.
##########
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:
Writing the value into the setting in place relies on two things about each
Java operator that nothing states: it keeps exactly one settings object in a
field (otherwise it fails when the worker starts), and it reads its settings
only while processing data, not when it is created or opened (otherwise it
silently keeps the placeholder). Today's operators follow both, after this PR
changes Reservoir Sampling, Split and Dictionary Matcher. A new operator that
works something out from a setting in its constructor would quietly use the
placeholder inside a loop. Could this be written down in `OperatorExecutor`'s
documentation, where operator authors will see it?
##########
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:
When two state messages carry different values for the same variable, the
later one wins, with no error. The copies agree only if nothing inside the loop
changes the loop's variables, and nothing stops that today: a Python UDF can
return a changed state from `process_state`. If a UDF on one branch changes
`i`, a Limit with `$i` after the branches meet gets one value or the other,
depending on which message arrives last.
##########
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:
This says the panel's `getEnclosingLoopStarts` in #8640 follows the same
rule, but the two are now different algorithms: this one takes two passes over
the topological order, and #8640 still walks the graph from each operator. If
they disagree on an operator, the panel offers `$i` there, and the compiler
either rejects it (a number setting) or keeps it as the literal text `$i` (a
text setting), with no error. Nothing tests that the two agree.
##########
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:
Minor: an executor installed in its place, for example by
`UpdateExecutorHandler` when the operator is reconfigured during an iteration,
starts with the placeholders and has missed the iteration's state message, so
its first tuple fails with "no state message carried it".
##########
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:
This is still the only test that runs a workflow, and it uses a text
setting. Please add runs for a number setting (a Limit with `$n`;
`WorkflowCompilerSpec` only builds the operator directly), for nested loops
(the inner variable winning), and for two branches that meet again. The cases
in my comments on `WorkflowCompiler` and `WorkflowCompilingService` need tests
too.
--
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]
