aglinxinyuan commented on code in PR #8639:
URL: https://github.com/apache/texera/pull/8639#discussion_r4140046129
##########
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 compiler fixes the output columns from the placeholder, and
nothing after that can change them. #8750 marks the settings the compiled plan
is built from with `@FixedAtCompileTime` (Projection's columns and Drop Option,
Unnest String's and Dictionary Matcher's result column, Aggregate, the join
keys, Type Casting, Sort Partitions, ...) and makes `$i` there a compile error
that names the property: `... cannot refer to loop variables (/isDrop ->
$drop): its output schema or partitioning is built from that property when the
workflow is compiled, before the loop runs`. That also replaces Projection's
missing-column message for `$col`. I'm keeping it in #8750 so this PR stays
about the binding itself.
##########
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:
I cut it to shrink the PR in 338db7842a, and that was the wrong cut.
da8d62f24a brings it back exactly as it was in 716334088b
(`inputsAheadOfLoopState`, called from `normalizeStateReferences`), with both
Interval Join cases. Your example is now a compile error: `Interval Join refers
to loop variables (/constant -> $k), but its input 'left table' is fed from
outside the loop block, so tuples from there can arrive before the loop state`.
The mixed case is rejected too, since one outside link on an input is enough.
7894d1d962 pins it with a Limit that has a link from the Loop Start and one
from outside; changing `forall` to `exists` fails that test.
Restoring it also showed that the check was flat: an operator inside *any*
block counted as carrying the state. So a join in an inner block whose first
input came from the outer loop's body passed, and would fail the same way at
run time. 7894d1d962 checks each link against every block around the operator
(`LogicalPlan.enclosingLoopStarts`): the link's source must be inside all of
them, or be the innermost block's Loop Start. The outer-body and sibling-block
cases are now compile errors as well. An input the operator reads only after
one that carries the state (a join's probe side) can still come from anywhere.
##########
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:
Good catch: it only had `DefaultScalaModule`. f0950a6e47 moves the
registration into one helper, `JSONUtils.registerWorkflowModules`
(`DefaultScalaModule` + `StateReferenceModule`). `JSONUtils.objectMapper`,
`WorkflowCompilingService` and `ComputingUnitMaster` all call it. Each
application has a test that its bootstrap mapper parses a typed `"$n"`:
`WorkflowCompilingServiceRunSpec` for an operator, and
`ComputingUnitMasterSpec` for a whole `SyncExecutionRequest`.
`TexeraWebApplication` keeps plain `DefaultScalaModule`, because its resources
keep a workflow's content as a string and never parse an operator. These
services have no common bootstrap or Jackson module service loader to hook
into, so this helper is the one call a new mapper that parses workflows needs.
##########
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 in da8d62f24a, as it was in 716334088b.
`EndChannelHandler` calls `dp.bindStateReferences()` before
`produceStateOnFinish` / `onFinishMultiPort`, so an iteration with no rows
fails at the operator's finish, before either callback sees the placeholder
(two `EndChannelHandlerSpec` cases).
The backend can't check the names before the run. A Loop Start's variables
are whatever its initialization assigns, and that is Python code the engine
only evaluates at run time, so there is no list of names to check against. The
panel can only approximate it by reading the code statically. Without the
panel, for example through `SyncExecutionResource`, a misspelled name fails at
the operator's first row or at its finish, and the error names the property and
the variable: `property /limit refers to loop variable n, but no state message
carried 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:
An operator now fails when the copies of one loop's state that reach it
disagree (03e82e3033). The worker passes each message's `loopCounter` (from
`StateFrame`) into `registerState`, and `StateReferenceBinding` keeps each
referenced variable's value with the depth it came from:
- a message from a more deeply nested loop replaces an outer loop's value
(shadowing);
- an outer loop's message that arrives later does not replace it;
- a message at the same depth with a different value for a referenced
variable fails with `loop variable i got two different values in one iteration,
1 and 2: a loop's variables must not change inside its body`;
- an equal value is fine, since several upstream workers or branches each
send a copy.
The check runs on every message, also after the first row, so the result
doesn't depend on arrival order. In your example, the Limit after the branches
meet now fails instead of picking one value. On a single path there is only one
copy, though. A UDF that changes `i` with no other copy reaching the Limit just
hands the Limit the changed value.
##########
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 lists both rules. An
executor keeps exactly one settings object in a field: the descriptor parsed
from its descString (with none or several, the worker fails to start). And it
reads its settings only while processing data, never in its constructor, `open`
or `produceStateOnStart`. Those run before the state arrives, so a value worked
out there silently keeps the placeholder.
##########
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:
0f6bbb39d9 fixes the doc: it now says the panel does its own walk, and
points at the test. `LogicalPlanSpec` has a test-only Scala port of #8640's
`getEnclosingLoopStarts`. Every membership case is checked against it, plus 400
random DAGs of Loop Start / Loop End / plain operators (fixed seed, shuffled so
the topological order differs from the ids). Since 7894d1d962 it also compares
each operator's enclosing Loop Starts, which the outside-input check now uses.
The two agree on all of them.
What the test can't catch is the two seeing different graphs. The panel
walks the whole canvas, while the compiler gets the plan without disabled
operators and, for run-to-operator, cut down to the target's upstream. I'll
note that in the description.
##########
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: a replacement executor would start from the placeholders after the
iteration's state has already passed it. #8750 refuses the edit instead. When
the running or the new operator refers to loop variables,
`ExecutionReconfigurationService.modifyOperatorLogic` answers `... refers to
loop variables, so it cannot be modified while the workflow runs: its executor
would be rebuilt from placeholders, after the loop state that sets them has
arrived`. No change here.
--
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]
