This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8750-05d48eb94ca39912a38100696bdc2a1478402ece in repository https://gitbox.apache.org/repos/asf/texera.git
commit 3fa608d04c124ec6c646e194f26f2cd391f9c6a9 Author: Xinyuan Lin <[email protected]> AuthorDate: Sat Oct 10 08:50:14 2026 +0000 fix(amber): reject $i in settings the plan is built from (#8750) ### What changes were proposed in this PR? Root cause: inside a loop block, a loop-variable reference (#8639) is written into the operator's setting when the iteration's state arrives. The compiler, however, builds the operator's output schema and its partitioning earlier, when the workflow compiles, from the reference's placeholder. A reference in a property the plan is built from therefore never reaches the schema or the partitioning. ``` Before: Loop Start (a = "x") -> Projection (alias $a) -> Loop End output column named "$a", null in every row After: compile error: Projection cannot refer to loop variables (/attributes/0/alias -> $a): its output schema or partitioning is built from that property when the workflow is compiled, before the loop runs ``` | Piece | What it does | |---|---| | `@NoLoopVariable` | Marks a descriptor property that can't hold a loop variable because the plan is built from it: an output column's name or type, or what an input is partitioned on. `LogicalOp`'s documentation states the rule for operator authors. | | `StateReferencing.noLoopVariableProperties` / `noLoopVariableReferences` | Name the marked properties by their JSON names, and find the references under them at any depth. Both go through Jackson's bean introspection, so renamed JSON names such as `"result attribute"` resolve. | | `OperatorMetadataGenerator` | Puts `"noLoopVariable": true` on each marked property in the operator's JSON schema, from the same helper, so the panel (#8640) can use the same list. | | `WorkflowCompiler` | Inside a loop block, a reference under a marked property is a compile error, reported before the physical plan is built so it is the operator's first error. Outside every block nothing changes: a `$a` alias is still the literal column name. | | `ExecutionReconfigurationService` | Modifying an operator that refers to loop variables while the workflow runs is refused: `Limit refers to loop variables, so it can't be changed while the workflow runs.` Its rebuilt executor would never have the variables written into it. | Marked properties: | Operator | Properties | |---|---| | Projection | drop option, attributes | | Dictionary Matcher, Unnest String | result attribute | | Aggregate | aggregations, group-by keys | | File Scan From Input | attribute name, file-name column option (the attribute name is declared by the text-source trait, so File Scan and Text Input carry the mark too; as sources they are never inside a loop block) | | Type Casting | cast units | | Hash Join, Interval Join | the join attributes | | Sort Partitions | sort attribute, domain min and max | ### Any related issues, documentation, discussions? Closes #8749 Related to #8635 and #8639. ### How was this PR tested? New tests, each run red against the pre-change code first: | Spec | Covers | |---|---| | `StateReferenceModuleSpec` | `noLoopVariableProperties` names each marked property by its JSON name, a trait's included; `noLoopVariableReferences` picks the references under a marked property whether it is renamed, escaped, nested or declared by a trait, and nothing from a descriptor with no marked property or no reference. | | `OperatorMetadataGeneratorSpec` | The schema marks exactly the annotated properties (renamed ones under their JSON names) and no others, and for every operator the marked set equals `noLoopVariableProperties`. | | `WorkflowCompilerSpec` | Inside a loop block, a reference in each marked property is a compile error, and it comes before the operator's other errors. A reference in a property the executor reads, on an operator whose plan is built from other properties, still binds. Outside every block, a `$a` Projection alias stays the literal column name. | | `ExecutionReconfigurationServiceSpec` | An edit that makes an operator inside a loop block refer to a loop variable is refused, with the message above. An operator inside a block that refers to no loop variable, and a `$x` literal outside every block, can still be modified. | ``` sbt "WorkflowCore/testOnly org.apache.texera.amber.core.state.* org.apache.texera.amber.core.executor.*" # 75 passed sbt "WorkflowCompiler/test" # 83 passed sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.metadata.* ...projection.* ...aggregate.* \ ...sortPartitions.* ...dictionary.* ...unneststring.* ...typecasting.* ...hashJoin.* ...intervalJoin.*" # 467 passed sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionReconfigurationServiceSpec" # 18 passed sbt "scalafmtCheckAll" "<WorkflowCore|WorkflowOperator|WorkflowCompiler|WorkflowExecutionService>/scalafixAll --check" # clean ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- .../service/ExecutionReconfigurationService.scala | 51 ++++++-- .../ExecutionReconfigurationServiceSpec.scala | 102 +++++++++++++++- .../texera/common/compiler/WorkflowCompiler.scala | 35 +++++- .../common/compiler/WorkflowCompilerSpec.scala | 133 +++++++++++++++++++++ .../texera/amber/core/state/NoLoopVariable.java | 39 ++++++ .../texera/amber/core/state/StateReferencing.scala | 39 +++++- .../core/state/StateReferenceModuleSpec.scala | 55 ++++++++- .../apache/texera/amber/operator/LogicalOp.scala | 9 ++ .../amber/operator/aggregate/AggregateOpDesc.scala | 3 + .../dictionary/DictionaryMatcherOpDesc.scala | 5 +- .../amber/operator/hashJoin/HashJoinOpDesc.scala | 3 + .../operator/intervalJoin/IntervalJoinOpDesc.scala | 3 + .../metadata/OperatorMetadataGenerator.scala | 12 ++ .../operator/projection/ProjectionOpDesc.scala | 3 + .../sortPartitions/SortPartitionsOpDesc.scala | 4 + .../operator/source/scan/file/FileScanOpDesc.scala | 2 + .../source/scan/text/TextSourceOpDesc.scala | 2 + .../operator/typecasting/TypeCastingOpDesc.scala | 2 + .../operator/unneststring/UnnestStringOpDesc.scala | 2 + .../metadata/OperatorMetadataGeneratorSpec.scala | 56 +++++++++ 20 files changed, 537 insertions(+), 23 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala index d1da4b4935..e4093ce4a7 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala @@ -19,6 +19,8 @@ package org.apache.texera.web.service +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.core.state.StateReferencing.literalReferences import org.apache.texera.amber.core.storage.RepositoryMountManager import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity import org.apache.texera.amber.engine.architecture.coordinator.{UpdateExecutorCompleted, Workflow} @@ -27,6 +29,8 @@ import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ WorkflowReconfigureRequest } import org.apache.texera.amber.engine.common.client.AmberClient +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.util.JSONUtils.objectMapper import org.apache.texera.web.SubscriptionManager import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent import org.apache.texera.web.model.websocket.request.ModifyLogicRequest @@ -60,24 +64,47 @@ class ExecutionReconfigurationService( val newOp = modifyLogicRequest.operator val opId = newOp.operatorIdentifier val currentOp = workflow.logicalPlan.getOperator(opId) - val reconfiguredPhysicalOp = - currentOp.runtimeReconfiguration( - workflow.context.workflowId, - workflow.context.executionId, - currentOp, - newOp + if (refersToLoopVariables(currentOp) || refersToLoopVariables(newOp)) { + ModifyLogicResponse( + opId.id, + isValid = false, + s"${currentOp.operatorInfo.userFriendlyName} refers to loop variables, so it can't be " + + "changed while the workflow runs." ) - reconfiguredPhysicalOp match { - case Failure(exception) => ModifyLogicResponse(opId.id, isValid = false, exception.getMessage) - case Success(op) => { - stateStore.reconfigurationStore.updateState(old => - old.copy(unscheduledReconfigurations = old.unscheduledReconfigurations :+ op) + } else { + val reconfiguredPhysicalOp = + currentOp.runtimeReconfiguration( + workflow.context.workflowId, + workflow.context.executionId, + currentOp, + newOp ) - ModifyLogicResponse(opId.id, isValid = true, "") + reconfiguredPhysicalOp match { + case Failure(exception) => + ModifyLogicResponse(opId.id, isValid = false, exception.getMessage) + case Success(op) => { + stateStore.reconfigurationStore.updateState(old => + old.copy(unscheduledReconfigurations = old.unscheduledReconfigurations :+ op) + ) + ModifyLogicResponse(opId.id, isValid = true, "") + } } } } + /** + * Whether `op`, in the deployed operator's place, refers to loop variables: its sidecar names + * one (the compiler's final one on the deployed operator, the parse's typed ones on an edit), + * or it sits inside a loop block and holds a whole-string `$name`. The rebuilt executor would + * parse the placeholders after the loop state that writes them into its setting has arrived + * (`OperatorExecutor.registerState`), so nothing would ever set them. + */ + private def refersToLoopVariables(op: LogicalOp): Boolean = + op.stateReferences.nonEmpty || ( + workflow.logicalPlan.operatorsInsideLoopBlocks.contains(op.operatorIdentifier) && + literalReferences(objectMapper.valueToTree[ObjectNode](op)).nonEmpty + ) + // actually performs all reconfiguration requests the user made during pause // sends ModifyLogic messages to operators and workers, // see the Fries reconfiguration paper for the algorithm. diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala index fc60af0c35..654168186d 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala @@ -37,6 +37,7 @@ import org.apache.texera.amber.core.workflow.{ OutputPort, PhysicalOp, PhysicalPlan, + PortIdentity, WorkflowContext } import org.apache.texera.amber.engine.architecture.coordinator.{ @@ -48,11 +49,17 @@ import org.apache.texera.amber.engine.architecture.rpc.controlcommands.WorkflowR import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn import org.apache.texera.amber.engine.architecture.rpc.coordinatorservice.CoordinatorServiceFs2Grpc import org.apache.texera.amber.engine.common.client.AmberClient +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.loop.{LoopEndOpDesc, LoopStartOpDesc} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.operator.{LogicalOp, StateTransferFunc} import org.apache.texera.amber.util.JSONUtils.objectMapper -import org.apache.texera.common.compiler.model.LogicalPlan +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan} import org.apache.texera.web.model.websocket.event.TexeraWebSocketEvent import org.apache.texera.web.model.websocket.request.ModifyLogicRequest import org.apache.texera.web.model.websocket.response.{ @@ -470,6 +477,99 @@ class ExecutionReconfigurationServiceSpec queued.map(_._1.id) shouldBe List(alreadyQueued.id) } + // --------------------------------------------------------------------------- + // An operator that refers to loop variables + // --------------------------------------------------------------------------- + + private def loopStartOp(): LoopStartOpDesc = { + val op = new LoopStartOpDesc + op.initialization = "i = 0" + op.output = "table.iloc[i]" + op + } + + private def loopEndOp(): LoopEndOpDesc = { + val op = new LoopEndOpDesc + op.update = "i += 1" + op.condition = "i < len(table)" + op + } + + /** A service over a workflow whose plan is LoopStart -> `deployedOp` -> LoopEnd. */ + private def insideLoopBlock(deployedOp: LogicalOp): (ExecutionStateStore, WorkflowOnlyService) = { + val start = loopStartOp() + val end = loopEndOp() + def link(from: LogicalOp, to: LogicalOp): LogicalLink = + LogicalLink(from.operatorIdentifier, PortIdentity(), to.operatorIdentifier, PortIdentity()) + val workflow = Workflow( + new WorkflowContext(workflowId = testWorkflowId, executionId = testExecutionId), + LogicalPlan( + List(start, deployedOp, end), + List(link(start, deployedOp), link(deployedOp, end)) + ), + PhysicalPlan(Set.empty, Set.empty) + ) + val stateStore = new ExecutionStateStore + (stateStore, new WorkflowOnlyService(stateStore, workflow)) + } + + private def filterOp(opId: String, value: String): SpecializedFilterOpDesc = { + val op = new SpecializedFilterOpDesc + op.setOperatorId(opId) + op.predicates = List(new FilterPredicate("line", ComparisonType.EQUAL_TO, value)) + op + } + + private def refused(name: String): String = + s"$name refers to loop variables, so it can't be changed while the workflow runs." + + "modifyOperatorLogic" should + "refuse to modify an operator whose deployed setting refers to a loop variable" in { + // The references the compiler left on the deployed operator, as it does inside a loop block. + val deployedOp = mkLimitOp("limit-a", 0) + deployedOp.stateReferences = Map("/limit" -> "n") + val (stateStore, service) = insideLoopBlock(deployedOp) + + service.modifyOperatorLogic(ModifyLogicRequest(mkLimitOp("limit-a", 7))) shouldBe + ModifyLogicResponse("limit-a", isValid = false, refused("Limit")) + stateStore.reconfigurationStore.getState.unscheduledReconfigurations shouldBe empty + } + + it should "refuse an edit that makes an operator inside a loop block refer to a loop variable" in { + val (typedStore, typed) = insideLoopBlock(mkLimitOp("limit-a", 1)) + // As the frontend sends it: "$n" in the Int property. + val typedEdit = objectMapper.readValue( + """{"operatorID":"limit-a","limit":"$n","operatorType":"Limit"}""", + classOf[LogicalOp] + ) + typed.modifyOperatorLogic(ModifyLogicRequest(typedEdit)) shouldBe + ModifyLogicResponse("limit-a", isValid = false, refused("Limit")) + typedStore.reconfigurationStore.getState.unscheduledReconfigurations shouldBe empty + + // A whole-string "$x" refers to a loop variable inside a loop block; the compiler never saw it. + val (literalStore, literal) = insideLoopBlock(filterOp("filter-a", "Asia")) + literal.modifyOperatorLogic(ModifyLogicRequest(filterOp("filter-a", "$x"))) shouldBe + ModifyLogicResponse("filter-a", isValid = false, refused("Filter")) + literalStore.reconfigurationStore.getState.unscheduledReconfigurations shouldBe empty + } + + it should "still modify an operator inside a loop block that refers to no loop variable, and a '$x' literal outside every block" in { + val (insideStore, inside) = insideLoopBlock(mkLimitOp("limit-a", 1)) + inside.modifyOperatorLogic(ModifyLogicRequest(mkLimitOp("limit-a", 7))) shouldBe + ModifyLogicResponse("limit-a", isValid = true, "") + insideStore.reconfigurationStore.getState.unscheduledReconfigurations.map { + case (op, _) => limitOf(op) + } shouldBe List(7) + + // Outside every block "$x" is the literal it is on main. + val outsideStore = new ExecutionStateStore + val outside = + new WorkflowOnlyService(outsideStore, mkWorkflow(List(filterOp("filter-a", "Asia")))) + outside.modifyOperatorLogic(ModifyLogicRequest(filterOp("filter-a", "$x"))) shouldBe + ModifyLogicResponse("filter-a", isValid = true, "") + outsideStore.reconfigurationStore.getState.unscheduledReconfigurations should have size 1 + } + "performReconfigurationOnResume" should "return without dispatching when no reconfigurations are pending" in { val stateStore = new ExecutionStateStore() diff --git a/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala index b3b8144d25..372b0e67dd 100644 --- a/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala @@ -30,7 +30,10 @@ import org.apache.texera.common.compiler.WorkflowCompiler.{ } import org.apache.texera.common.compiler.model.{LogicalPlan, LogicalPlanPojo} import org.apache.texera.amber.core.executor.OpExecWithCode -import org.apache.texera.amber.core.state.StateReferencing.literalReferences +import org.apache.texera.amber.core.state.StateReferencing.{ + literalReferences, + noLoopVariableReferences +} import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.OperatorIdentity import org.apache.texera.amber.core.workflow.{ @@ -82,16 +85,19 @@ object WorkflowCompiler { /** * 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. + * of the operator's JSON (`StateReferencing.literalReferences`), and an entry in a property that + * cannot hold a loop variable is an error (`referencesInNoLoopVariableProperties`). 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, plan: LogicalPlan): Option[Throwable] = if (plan.operatorsInsideLoopBlocks.contains(logicalOp.operatorIdentifier)) { logicalOp.stateReferences = literalReferences(objectMapper.valueToTree[ObjectNode](logicalOp)) ++ logicalOp.stateReferences - inputsAheadOfLoopState(logicalOp, plan) + referencesInNoLoopVariableProperties(logicalOp) + .orElse(inputsAheadOfLoopState(logicalOp, plan)) } else { Option.when(logicalOp.stateReferences.nonEmpty) { new IllegalArgumentException( @@ -101,6 +107,25 @@ object WorkflowCompiler { } } + /** + * The loop state writes a loop variable into the operator's setting only once the workflow runs, + * but a property marked `NoLoopVariable` -- the name or type of an output column, or what an + * input is partitioned on -- has been built into the plan by then: from the placeholder. A + * Projection alias "$a" would output a column named "$a", null in every row; a SortPartitions + * domain would stay 0 to 0. A reference there is an error instead. + */ + private def referencesInNoLoopVariableProperties(logicalOp: LogicalOp): Option[Throwable] = { + val references = noLoopVariableReferences(logicalOp) + Option.when(references.nonEmpty) { + val those = if (references.size == 1) "that property" else "those properties" + new IllegalArgumentException( + s"${logicalOp.operatorInfo.userFriendlyName} cannot refer to loop variables " + + s"(${formatReferences(references)}): its output schema or partitioning is built from " + + s"$those when the workflow is compiled, before the loop runs" + ) + } + } + /** * The first tuple fails on a loop variable its state has not written yet, and the states of * the blocks around the operator come ahead of the tuples only on links from inside all of diff --git a/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala index 2dcefc4942..53927271e6 100644 --- a/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala +++ b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala @@ -1021,6 +1021,139 @@ class WorkflowCompilerSpec extends AnyFlatSpec { ) } + /** TextInput -> LoopStart -> `op` -> LoopEnd, every input of `op` fed by the LoopStart. */ + private def insideLoopBlock(op: LogicalOp): LogicalPlanPojo = { + val src = textInputOp("0\n1") + val start = loopStartOp() + val end = loopEndOp() + val inputs = op.operatorInfo.inputPorts.map(input => linked(start, op, input.id.id)) + pojo(List(src, start, op, end), linked(src, start) :: inputs ::: List(linked(op, end))) + } + + /** + * A reference in each property the compiled plan is built from -- an output column's name or + * type, or what the input is partitioned on: the operator as the frontend sends it, its name, + * and the references the error names. + */ + private val referencesInNoLoopVariableProperties: List[(String, String, String)] = List( + ( + """{"isDrop":false,"attributes":[{"originalAttribute":"line","alias":"$a"}], + |"operatorType":"Projection"}""".stripMargin, + "Projection", + "/attributes/0/alias -> $a" + ), + ( + """{"isDrop":"$d","attributes":[{"originalAttribute":"line","alias":""}], + |"operatorType":"Projection"}""".stripMargin, + "Projection", + "/isDrop -> $d" + ), + ( + """{"Dictionary":"x","Attribute":"line","result attribute":"$r","Matching type":"Scan", + |"operatorType":"DictionaryMatcher"}""".stripMargin, + "Dictionary matcher", + "/result attribute -> $r" + ), + ( + """{"Delimiter":",","Attribute":"line","Result attribute":"$r", + |"operatorType":"UnnestString"}""".stripMargin, + "Unnest String", + "/Result attribute -> $r" + ), + ( + """{"aggregations":[{"aggFunction":"count","attribute":"line","result attribute":"$r"}], + |"groupByKeys":["$g"],"operatorType":"Aggregate"}""".stripMargin, + "Aggregate", + "/aggregations/0/result attribute -> $r, /groupByKeys/0 -> $g" + ), + ( + """{"attributeName":"$c","outputFileName":"$f","operatorType":"FileScanOp"}""", + "File Scan From Input", + "/attributeName -> $c, /outputFileName -> $f" + ), + ( + """{"typeCastingUnits":[{"attribute":"$c","resultType":"integer"}], + |"operatorType":"TypeCasting"}""".stripMargin, + "Type Casting", + "/typeCastingUnits/0/attribute -> $c" + ), + ( + """{"buildAttributeName":"$b","probeAttributeName":"$p","joinType":"inner", + |"operatorType":"HashJoin"}""".stripMargin, + "Hash Join", + "/buildAttributeName -> $b, /probeAttributeName -> $p" + ), + ( + """{"leftAttributeName":"$l","rightAttributeName":"$r","constant":10,"includeLeftBound":true, + |"includeRightBound":true,"operatorType":"IntervalJoin"}""".stripMargin, + "Interval Join", + "/leftAttributeName -> $l, /rightAttributeName -> $r" + ), + ( + """{"sortAttributeName":"$s","domainMin":"$lo","domainMax":"$hi", + |"operatorType":"SortPartitions"}""".stripMargin, + "Sort Partitions", + "/domainMax -> $hi, /domainMin -> $lo, /sortAttributeName -> $s" + ) + ) + + it should "report a reference inside a loop block in a property the compiled plan is built from" in { + // Built from the placeholder, the plan would never see the loop's value: a Projection alias + // "$a" would output a column named "$a", every row null; a SortPartitions domain would stay + // 0 to 0. Reported before the physical plan, ahead of a schema error the placeholder causes. + referencesInNoLoopVariableProperties.foreach { + case (json, name, references) => + val those = if (references.contains(", ")) "those properties" else "that property" + val message = s"$name cannot refer to loop variables ($references): its output schema " + + s"or partitioning is built from $those when the workflow is compiled, before the loop " + + "runs" + val op = parsed(json) + + val result = new WorkflowCompiler(newContext()).compile(insideLoopBlock(op)) + + assert(result.physicalPlan.isEmpty, name) + assert( + result.operatorIdToError.get(op.operatorIdentifier).exists(_.message.contains(message)), + s"$name: ${result.operatorIdToError}" + ) + val ex = intercept[IllegalArgumentException] { + new WorkflowCompiler(newContext()) + .compile(insideLoopBlock(parsed(json)), CompilationErrorHandling.Strict) + } + assert(ex.getMessage == message, name) + } + } + + it should "accept a reference inside a loop block in a property the executor reads, on an operator the plan is built from in part" in { + // The dictionary and the delimiter are read by the executor, after the loop state wrote them. + Seq( + """{"Dictionary":"$w","Attribute":"line","result attribute":"matched","Matching type":"Scan", + |"operatorType":"DictionaryMatcher"}""".stripMargin -> Map("/Dictionary" -> "w"), + """{"Delimiter":"$d","Attribute":"line","Result attribute":"unnested", + |"operatorType":"UnnestString"}""".stripMargin -> Map("/Delimiter" -> "d") + ).foreach { + case (json, sidecar) => + val op = parsed(json) + val result = new WorkflowCompiler(newContext()).compile(insideLoopBlock(op)) + assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}") + val (_, descString) = executorInit(result, op) + assert(sidecarOf(descString) == sidecar) + } + } + + it should "keep a '$a' Projection alias outside every loop block the literal column name it is on main" in { + val src = textInputOp("0") + val projection = parsed(referencesInNoLoopVariableProperties.head._1) + + val result = new WorkflowCompiler(newContext()).compile( + pojo(List(src, projection), List(linked(src, projection))) + ) + + assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}") + val schema = result.operatorIdToOutputSchemas(projection.operatorIdentifier)(PortIdentity()) + assert(schema.map(_.getAttributeNames).contains(List("$a"))) + } + private def intervalJoinOp(): LogicalOp = parsed( """{"leftAttributeName":"line","rightAttributeName":"line","constant":"$i", diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/NoLoopVariable.java b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/NoLoopVariable.java new file mode 100644 index 0000000000..79b05840e2 --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/NoLoopVariable.java @@ -0,0 +1,39 @@ +/* + * 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.state; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Marks a descriptor property that cannot hold a loop variable, nor can anything inside it. The + * plan -- the name or type of an output column, or what an input is partitioned on -- is built + * from the property when the workflow is compiled, before any loop runs; the loop state writes a + * loop variable into the operator's setting only later, so the plan would be built from the + * {@code $name} placeholder instead. Inside a loop block such a reference is an error + * ({@code WorkflowCompiler.normalizeStateReferences}); outside every block a {@code $name} string + * stays the literal it is. + */ +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.FIELD}) +public @interface NoLoopVariable { +} diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/StateReferencing.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/StateReferencing.scala index 37c9ca3b3e..dcd98949ea 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/StateReferencing.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/state/StateReferencing.scala @@ -23,8 +23,10 @@ import com.fasterxml.jackson.annotation.JsonProperty import com.fasterxml.jackson.core.JsonPointer import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.util.JSONUtils.objectMapper -import scala.jdk.CollectionConverters.IteratorHasAsScala +import scala.jdk.CollectionConverters.{IteratorHasAsScala, ListHasAsScala} +import scala.util.Try /** * A descriptor whose properties may refer to loop variables. @@ -32,8 +34,9 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala * Inside a loop block (LoopStart ... LoopEnd) a property whose WHOLE value is `$K` refers to the * loop variable `K`, which reaches the operator in the iteration's state message. The * `stateReferences` sidecar maps each such property's JSON pointer to the variable name: the - * parse records the typed ones (`StateReferenceModule`), the compiler adds the string ones - * (`WorkflowCompiler.normalizeStateReferences`), and the worker writes the variables there + * parse records the typed ones (`StateReferenceModule`), the compiler adds the string ones and + * rejects one in a property that cannot hold a loop variable (`NoLoopVariable`, + * `WorkflowCompiler.normalizeStateReferences`), and the worker writes the variables there * (`OperatorExecutor.registerState`). The property panel never shows it. */ trait StateReferencing { @@ -57,6 +60,36 @@ object StateReferencing { case _ => None } + /** + * The JSON names of `descriptorClass`'s properties marked `NoLoopVariable`, a trait's or a + * superclass's included. The one list behind both the compiler's check + * (`noLoopVariableReferences`) and the `noLoopVariable` keyword of the operator's schema + * (`OperatorMetadataGenerator.generateOperatorJsonSchema`), so the two cannot drift apart. + */ + def noLoopVariableProperties(descriptorClass: Class[_]): Set[String] = + objectMapper.getSerializationConfig + .introspect(objectMapper.constructType(descriptorClass)) + .findProperties() + .asScala + .filter(property => + Option(property.getField).exists(_.hasAnnotation(classOf[NoLoopVariable])) + ) + .map(_.getName) + .toSet + + /** + * The entries of `descriptor`'s sidecar that fall under a property marked `NoLoopVariable`, + * however deep: the plan is built from that property before the loop runs, so nothing binds + * them. + */ + def noLoopVariableReferences(descriptor: StateReferencing): Map[String, String] = { + val marked = noLoopVariableProperties(descriptor.getClass) + descriptor.stateReferences.filter { + case (pointer, _) => + Try(JsonPointer.compile(pointer).getMatchingProperty).toOption.exists(marked.contains) + } + } + /** Every whole-string `$name` value of `tree` outside its sidecar: its JSON pointer -> name. */ def literalReferences(tree: ObjectNode): Map[String, String] = { def scan(node: JsonNode, at: JsonPointer): Iterator[(String, String)] = diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/state/StateReferenceModuleSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/state/StateReferenceModuleSpec.scala index e5ef3fd69e..7f86f248c9 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/state/StateReferenceModuleSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/state/StateReferenceModuleSpec.scala @@ -26,7 +26,12 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize import com.fasterxml.jackson.databind.exc.InvalidFormatException import com.fasterxml.jackson.databind.node.ObjectNode import com.fasterxml.jackson.module.scala.DefaultScalaModule -import org.apache.texera.amber.core.state.StateReferencing.{literalReferences, referencedVariable} +import org.apache.texera.amber.core.state.StateReferencing.{ + literalReferences, + noLoopVariableProperties, + noLoopVariableReferences, + referencedVariable +} import org.apache.texera.amber.core.tuple.AttributeType import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.flatspec.AnyFlatSpec @@ -202,6 +207,40 @@ class StateReferenceModuleSpec extends AnyFlatSpec { Map("/name" -> "i", "/tags/1" -> "t", "/a~1b~0c/deep/0/v" -> "z") ) } + + "StateReferencing.noLoopVariableProperties" should "name each property marked @NoLoopVariable by its JSON name, a trait's included" in { + assert( + noLoopVariableProperties(classOf[PlanBuilding]) == + Set("result attribute", "a/b", "keys", "column") + ) + } + + it should "name nothing for a descriptor with no marked property" in { + assert(noLoopVariableProperties(classOf[Bean]).isEmpty) + } + + "StateReferencing.noLoopVariableReferences" should "pick the references under a property marked @NoLoopVariable, renamed, escaped, nested or declared by a trait" in { + val descriptor = new PlanBuilding + descriptor.stateReferences = Map( + "/result attribute" -> "r", + "/a~1b" -> "s", + "/keys/0/label" -> "k", + "/column" -> "c", + "/value" -> "v", + "/others/0/count" -> "o" + ) + assert( + noLoopVariableReferences(descriptor) == + Map("/result attribute" -> "r", "/a~1b" -> "s", "/keys/0/label" -> "k", "/column" -> "c") + ) + } + + it should "pick nothing from a descriptor with no marked property, or with no reference" in { + val bean = new Bean + bean.stateReferences = Map("/name" -> "i", "/nested/0/count" -> "c") + assert(noLoopVariableReferences(bean).isEmpty) + assert(noLoopVariableReferences(new PlanBuilding).isEmpty) + } } object StateReferenceModuleSpec { @@ -230,6 +269,20 @@ object StateReferenceModuleSpec { @JsonProperty var color: AttributeType = _ // a Java enum } + /** As `TextSourceOpDesc`: a property a trait declares and marks. */ + trait NamedColumn { + @JsonProperty @NoLoopVariable var column: String = "line" + } + + /** Marked properties -- renamed, escaped, a list of beans, a trait's -- and unmarked ones. */ + class PlanBuilding extends StateReferencing with NamedColumn { + @JsonProperty("result attribute") @NoLoopVariable var resultAttribute: String = _ + @JsonProperty("a/b") @NoLoopVariable var slashed: String = _ + @JsonProperty @NoLoopVariable var keys: List[Nested] = List.empty + @JsonProperty var value: String = _ + @JsonProperty var others: List[Nested] = List.empty + } + /** A StateReferencing object holding another one. */ class Outer extends StateReferencing { @JsonProperty var inner: Bean = _ diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala index 7143e59dc0..7f4e5e6c52 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala @@ -159,6 +159,15 @@ trait StateTransferFunc extends ((OperatorExecutor, OperatorExecutor) => Unit) with java.io.Serializable +/** + * The descriptor of an operator: its settings, and the physical plan built from them. + * + * A property that the operator's output schema (column names or types) or its input partitioning + * is built from must be marked `@NoLoopVariable`. The plan is built when the workflow is + * compiled, before any loop runs: without the mark, a `$i` there inside a loop block is not + * rejected, and the run silently uses the placeholder (e.g. a column named "$a", null in every + * row). + */ @JsonTypeInfo( use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala index 1bace652c8..191f2348ea 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.aggregate import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.{ ExecutionIdentity, @@ -41,12 +42,14 @@ class AggregateOpDesc extends LogicalOp { @JsonPropertyDescription("multiple aggregation functions") @NotNull(message = "aggregation cannot be null") @Size(min = 1, message = "aggregations cannot be empty") + @NoLoopVariable var aggregations: List[AggregationOperation] = List() @JsonProperty("groupByKeys") @JsonSchemaTitle("Group By Keys") @JsonPropertyDescription("group by columns") @AutofillAttributeNameList + @NoLoopVariable var groupByKeys: List[String] = List() override def getPhysicalPlan( diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala index 8660fcdc69..e617502657 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/dictionary/DictionaryMatcherOpDesc.scala @@ -21,6 +21,7 @@ package org.apache.texera.amber.operator.dictionary import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{Attribute, AttributeType} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{ @@ -56,7 +57,9 @@ class DictionaryMatcherOpDesc extends MapOpDesc with StandaloneCodeGenerator { var attribute: String = _ @JsonProperty(value = "result attribute", required = true, defaultValue = "matched") - @JsonPropertyDescription("column name of the matching result") var resultAttribute: String = _ + @JsonPropertyDescription("column name of the matching result") + @NoLoopVariable + var resultAttribute: String = _ @JsonProperty(value = "Matching type", required = true) var matchingType: MatchingType = _ diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala index 708546a84a..e8ef7c8bbc 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.hashJoin import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{Attribute, Schema} import org.apache.texera.amber.core.virtualidentity.{ ExecutionIdentity, @@ -59,12 +60,14 @@ class HashJoinOpDesc[K] extends LogicalOp { @JsonSchemaTitle("Left Input Attribute") @JsonPropertyDescription("attribute to be joined on the Left Input") @AutofillAttributeName + @NoLoopVariable var buildAttributeName: String = _ @JsonProperty(required = true) @JsonSchemaTitle("Right Input Attribute") @JsonPropertyDescription("attribute to be joined on the Right Input") @AutofillAttributeNameOnPort1 + @NoLoopVariable var probeAttributeName: String = _ @JsonProperty(required = true, defaultValue = "inner") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala index 2218fe2cce..64c0ac435b 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/intervalJoin/IntervalJoinOpDesc.scala @@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.fasterxml.jackson.databind.annotation.JsonDeserialize import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{Attribute, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow._ @@ -58,12 +59,14 @@ class IntervalJoinOpDesc extends LogicalOp { @JsonSchemaTitle("Left Input attr") @JsonPropertyDescription("Choose one attribute in the left table") @AutofillAttributeName + @NoLoopVariable var leftAttributeName: String = _ @JsonProperty(required = true) @JsonSchemaTitle("Right Input attr") @JsonPropertyDescription("Choose one attribute in the right table") @AutofillAttributeNameOnPort1 + @NoLoopVariable var rightAttributeName: String = _ @JsonProperty(required = true, defaultValue = "10") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala index fb649cbf0b..fd958f660a 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGenerator.scala @@ -85,6 +85,9 @@ case class AllOperatorMetadata( */ object OperatorMetadataGenerator { + /** The schema keyword, `true` on each property that cannot hold a loop variable. */ + final val NO_LOOP_VARIABLE_KEYWORD = "noLoopVariable" + val texeraSchemaGeneratorConfig: JsonSchemaConfig = html5EnabledSchema.copy( useOneOfForOption = false, useOneOfForNullables = false, @@ -143,6 +146,15 @@ object OperatorMetadataGenerator { jsonSchema.get("required").asInstanceOf[ArrayNode].remove(operatorTypeIndex) // remove "title" for the operator - frontend uses userFriendlyName to show operator title jsonSchema.remove("title") + // mark each property that cannot hold a loop variable (`NoLoopVariable`): the compiler rejects + // a `$name` there inside a loop block, so the property panel need not offer one + val properties = jsonSchema.get("properties").asInstanceOf[ObjectNode] + StateReferencing.noLoopVariableProperties(opDescClass).foreach { name => + properties.get(name) match { + case property: ObjectNode => property.put(NO_LOOP_VARIABLE_KEYWORD, true) + case _ => + } + } // let an operator add the part of its schema that its annotations cannot state opDescClass.getConstructor().newInstance() match { case customizer: JsonSchemaCustomizer => customizer.customizeJsonSchema(jsonSchema) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala index 2004fb594c..375f7e61e0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.projection import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.PhysicalOp.oneToOnePhysicalOp @@ -37,12 +38,14 @@ class ProjectionOpDesc extends MapOpDesc with StandaloneCodeGenerator { @JsonProperty(required = true, defaultValue = "false") @JsonSchemaTitle("Drop Option") @JsonPropertyDescription("check to drop the selected attributes") + @NoLoopVariable var isDrop: Boolean = false // Named explicitly, without `required`: the form already asks for these and must go // on accepting an empty list, but a field carrying no annotation is invisible to // anything reading the operator's config by reflection. @JsonProperty + @NoLoopVariable var attributes: List[AttributeUnit] = List() override def getPhysicalOp( diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sortPartitions/SortPartitionsOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sortPartitions/SortPartitionsOpDesc.scala index 9290d549f2..e9badac486 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sortPartitions/SortPartitionsOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sortPartitions/SortPartitionsOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.sortPartitions import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, RangePartition} import org.apache.texera.amber.operator.LogicalOp @@ -44,16 +45,19 @@ class SortPartitionsOpDesc extends LogicalOp { @JsonSchemaTitle("Attribute") @JsonPropertyDescription("Attribute to sort (must be numerical).") @AutofillAttributeName + @NoLoopVariable var sortAttributeName: String = _ @JsonProperty(required = true) @JsonSchemaTitle("Attribute Domain Min") @JsonPropertyDescription("Minimum value of the domain of the attribute.") + @NoLoopVariable var domainMin: Long = _ @JsonProperty(required = true) @JsonSchemaTitle("Attribute Domain Max") @JsonPropertyDescription("Maximum value of the domain of the attribute.") + @NoLoopVariable var domainMax: Long = _ override def getPhysicalOp( diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala index f83d1d7630..35d260c9b0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/file/FileScanOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.source.scan.file import com.fasterxml.jackson.annotation.JsonProperty import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{ @@ -47,6 +48,7 @@ class FileScanOpDesc extends SourceOperatorDescriptor with TextSourceOpDesc { @JsonProperty(defaultValue = "false") @JsonSchemaTitle("Include Filename") + @NoLoopVariable var outputFileName: Boolean = false override def getPhysicalOp( diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala index 07a773580f..c3e5b509fa 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/text/TextSourceOpDesc.scala @@ -26,6 +26,7 @@ import com.kjetland.jackson.jsonSchema.annotations.{ JsonSchemaString, JsonSchemaTitle } +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.operator.metadata.annotations.HideAnnotation import org.apache.texera.amber.operator.source.scan.FileAttributeType @@ -46,6 +47,7 @@ trait TextSourceOpDesc { @JsonProperty(defaultValue = "line", required = true) @JsonSchemaTitle("Attribute Name") @JsonDeserialize(contentAs = classOf[java.lang.String]) + @NoLoopVariable var attributeName: String = "line" @JsonSchemaTitle("Limit (lines)") diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala index c23c3dd66f..145b175077 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/typecasting/TypeCastingOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.typecasting import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{AttributeType, AttributeTypeUtils, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow._ @@ -36,6 +37,7 @@ class TypeCastingOpDesc extends MapOpDesc with StandaloneCodeGenerator { @JsonProperty(required = true) @JsonSchemaTitle("TypeCasting Units") @JsonPropertyDescription("Multiple type castings") + @NoLoopVariable var typeCastingUnits: List[TypeCastingUnit] = List.empty override def getPhysicalOp( diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala index cb7f9f21b2..3a90db3009 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/unneststring/UnnestStringOpDesc.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.operator.unneststring import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.state.NoLoopVariable import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{ @@ -58,6 +59,7 @@ class UnnestStringOpDesc extends FlatMapOpDesc with StandaloneCodeGenerator { @JsonProperty(value = "Result attribute", required = true, defaultValue = "unnestResult") @JsonPropertyDescription("column name of the unnest result") + @NoLoopVariable var resultAttribute: String = _ override def operatorInfo: OperatorInfo = diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGeneratorSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGeneratorSpec.scala index 2a370c5fc2..3d90363059 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGeneratorSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/metadata/OperatorMetadataGeneratorSpec.scala @@ -19,9 +19,16 @@ package org.apache.texera.amber.operator.metadata +import com.fasterxml.jackson.databind.node.BooleanNode import org.apache.texera.amber.core.state.StateReferencing import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.dictionary.DictionaryMatcherOpDesc import org.apache.texera.amber.operator.filter.SpecializedFilterOpDesc +import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.projection.ProjectionOpDesc +import org.apache.texera.amber.operator.sortPartitions.SortPartitionsOpDesc +import org.apache.texera.amber.operator.source.scan.file.FileScanOpDesc +import org.apache.texera.amber.operator.unneststring.UnnestStringOpDesc import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -52,4 +59,53 @@ class OperatorMetadataGeneratorSpec extends AnyFlatSpec with Matchers { val required = schema.get("required").elements().asScala.map(_.asText()).toList required should not contain StateReferencing.SIDECAR_PROPERTY } + + /** The properties of `opDescClass`'s schema whose `noLoopVariable` keyword is `true`. */ + private def markedNoLoopVariable(opDescClass: Class[_ <: LogicalOp]): Set[String] = + OperatorMetadataGenerator + .generateOperatorJsonSchema(opDescClass) + .get("properties") + .fields() + .asScala + .filter(_.getValue.get("noLoopVariable") == BooleanNode.TRUE) + .map(_.getKey) + .toSet + + it should "mark each property that cannot hold a loop variable with 'noLoopVariable', by its JSON name" in { + markedNoLoopVariable(classOf[ProjectionOpDesc]) shouldBe Set("isDrop", "attributes") + // Renamed in JSON: the panel looks the property up by the name the schema gives it. + markedNoLoopVariable(classOf[DictionaryMatcherOpDesc]) shouldBe Set("result attribute") + markedNoLoopVariable(classOf[UnnestStringOpDesc]) shouldBe Set("Result attribute") + // `attributeName` is declared and marked by the TextSourceOpDesc trait. + markedNoLoopVariable(classOf[FileScanOpDesc]) shouldBe Set("attributeName", "outputFileName") + markedNoLoopVariable(classOf[SortPartitionsOpDesc]) shouldBe + Set("sortAttributeName", "domainMin", "domainMax") + } + + it should "leave 'noLoopVariable' off a property the executor reads after the loop state wrote it" in { + val properties = Seq( + classOf[SpecializedFilterOpDesc] -> "predicates", + classOf[LimitOpDesc] -> "limit", + classOf[DictionaryMatcherOpDesc] -> "Dictionary", + classOf[UnnestStringOpDesc] -> "Delimiter" + ) + properties.foreach { + case (opDescClass, name) => + val schema = OperatorMetadataGenerator.generateOperatorJsonSchema(opDescClass) + withClue(s"${opDescClass.getSimpleName}.$name: ") { + schema.get("properties").has(name) shouldBe true + schema.get("properties").get(name).has("noLoopVariable") shouldBe false + } + } + } + + it should "generate every operator's schema, marking exactly the properties the compiler rejects a loop variable in" in { + // One list behind both: what the schema marks is exactly what the compiler checks. + OperatorMetadataGenerator.operatorTypeMap.keys.foreach { opDescClass => + withClue(s"${opDescClass.getSimpleName}: ") { + markedNoLoopVariable(opDescClass) shouldBe + StateReferencing.noLoopVariableProperties(opDescClass) + } + } + } }
