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-7649-363537e0251182472f540079d262599f6cbb5240 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 9c189970a35b6337991f3d6006f1038f85be1aca Author: Eugene Gu <[email protected]> AuthorDate: Mon Aug 17 02:24:03 2026 +0000 test(workflow-compiler): Extend WorkflowCompilerSpec to cover the Python code-generation error path (#7649) ### What changes were proposed in this PR? `WorkflowCompiler` scans every Python-based physical operator's generated code for the `#EXCEPTION DURING CODE GENERATION:` marker that `PythonOperatorDescriptor` embeds when an operator's `generatePythonCode` throws. On a hit it either appends a `RuntimeException("Operator is not configured properly: ...")` to the caller's error list (the editing-time, lenient path) or throws it immediately when no error list was given (the pre-execution, strict path). `WorkflowCompilerSpec`'s 15 tests never exercised either branch; the only existing test of the marker is the producer-side `PythonOperatorDescriptorSpec`, which asserts the marker is written but not that the compiler reacts to it. This adds 7 tests to `WorkflowCompilerSpec` (15 → 22). Test-only: no production file is touched, and none of the 15 existing tests is modified — the diff removes exactly three lines, all of them import lines being widened. Lenient path: 1. `should accumulate a per-operator error when a Python operator's code generation fails` — the error is attributed to the right logical operator, carries the expected message, and compilation **continues** (the rest of the plan still lands in the physical plan and the storage set). 2. `should attribute each Python code-generation failure to its own logical operator` — two failing Python operators with two distinct messages, each keyed to its own id. 3. `should trim the marker's message and report it as a plain RuntimeException` 4. `should report no code-generation error for a well-formed Python operator` — asserts the operator really is still Python-based, so it cannot pass by quietly ceasing to be one. 5. `should not subject non-Python operators to the code-generation check` Strict path: 6. `in strict mode should throw immediately when a Python operator's code generation failed` 7. `in strict mode should not throw for a well-formed Python operator` Five of the seven drive the **real shipped `SortOpDesc`**, whose `generatePythonCode` opens with `require(attributes.nonEmpty, ...)` and a per-key `require` (`SortOpDesc.scala:34-38`). A Sort dropped on the canvas and left unconfigured is therefore a genuine, user-reachable route into the marker state, and it conveniently yields two distinct messages — which is what makes test 2's per-operator attribution meaningful. A configured Sort gives the negative control in tests 4 and 7: same operator, same code path, codegen simply succeeds. Test 3 needs a small test-only fixture (`PaddedFailurePyOp`, ~15 lines, modelled on `PythonOperatorDescriptorSpec`'s `ThrowingPyOp`): no shipped operator raises a whitespace-padded message, and without padding the `.trim` is unobservable. It is also the spec's only source-operator Python case. Two notes: - Four assertions hard-code `SortOpDesc`'s exact `require` wording, so rewording those messages in `workflow-operator` will fail these `workflow-compiler` tests. That cross-module coupling is deliberate — it pins the end-to-end string a user actually sees — but it is worth knowing. - Test 3 also asserts the reported message starts with `java.lang.RuntimeException: `. That prefix comes from `err.toString` at `WorkflowCompiler.scala:66` and is part of what the UI renders today; the assertion carries a comment saying so, so if that pre-existing wart is ever fixed there is one self-explaining test to update. ### Any related issues, documentation, discussions? Closes #7647 Builds on the specs added by #5019 / #5022 and the module unification in #6143. ### How was this PR tested? `sbt "WorkflowCompiler/testOnly *WorkflowCompilerSpec"` — 22 tests, all passing (15 pre-existing + 7 new). Running the whole module (`WorkflowCompiler/test`) is green too: 3 suites / 57 tests / 0 failures. `WorkflowCompiler/scalafmtCheck` and `WorkflowCompiler/Test/scalafmtCheck` are clean. Every new test was mutation-checked, twice and independently: the production check was temporarily broken, the suite re-run, and the file reverted (verified byte-identical afterwards). Highlights, with the tests that are the *sole* killer of a mutant: | Mutant in `WorkflowCompiler.scala` | Caught by | |---|---| | lenient arm appends **and then** throws (abort instead of continue) | 1 only | | record only the first codegen failure per compile | 2 only | | drop `.trim` on the captured message | 3 only | | skip the check for source operators (`&& !isSourceOperator`) | 3 only | | strict arm swallows instead of throwing (`case None => ()`) | 6 only | | flip the `errorList` arms (`Some` → throw, `None` → skip) | 1, 6 | | change the message prefix | 1, 2, 3, 6 | | `group(1)` → `group(0)` | 1, 2, 3, 6 | | marker regex changed so it never matches | 1, 2, 3, 6 | | regex relaxed to `(.*)` (matches every operator) | 1, 2, 3, 4, 6, 7 | | attribute the error to the wrong logical operator id | 1, 2, 3 | | build the error but never record it | 1, 2, 3, 6 | | `RuntimeException` → `IllegalArgumentException` | 3 | Two mutants survive, both semantically equivalent rather than gaps: reporting the *last* marker match instead of the first (a codegen-failure body is exactly the one-line marker, so first == last), and anchoring the regex with `^` — the latter is in fact the fix for a bug found along the way (below), and all 22 tests stay green under it, so the suite does not over-fit to the current unanchored check. One more mutant is worth calling out honestly: removing the `isPythonBased` guard entirely does not fail test 5 — it aborts the whole suite, because `PhysicalOp.getCode` throws an `IllegalAccessError` (a `LinkageError`, which is not `NonFatal`) for non-code operators. Test 5 therefore documents the guard's intent rather than detecting its removal. It is kept for that reason, with a self-check that stops it silently becoming a no-op. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) Co-authored-by: Meng Wang <[email protected]> --- .../common/compiler/WorkflowCompilerSpec.scala | 308 ++++++++++++++++++++- 1 file changed, 305 insertions(+), 3 deletions(-) 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 94187373d7..da7ac59482 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 @@ -20,9 +20,9 @@ package org.apache.texera.common.compiler import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} -import org.apache.texera.amber.core.tuple.{Attribute, AttributeType} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity -import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.amber.core.workflow.{OutputPort, PortIdentity, WorkflowContext} import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.COMPILATION_ERROR import org.apache.texera.amber.operator.filter.{ ComparisonType, @@ -30,9 +30,11 @@ import org.apache.texera.amber.operator.filter.{ SpecializedFilterOpDesc } import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} +import org.apache.texera.amber.operator.sort.{SortCriteriaUnit, SortOpDesc, SortPreference} import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc -import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, TestOperators} import org.scalatest.flatspec.AnyFlatSpec /** @@ -88,6 +90,45 @@ class WorkflowCompilerSpec extends AnyFlatSpec { op } + // Sort is a real, shipped `PythonOperatorDescriptor` whose `generatePythonCode` + // rejects an unconfigured operator, so `sortOp()` (no sort keys) and + // `sortOp("" -> ASC)` (a key with no attribute) are genuine ways for a user to + // land in the `#EXCEPTION DURING CODE GENERATION:` state. + private def sortOp(criteria: (String, SortPreference)*): SortOpDesc = { + val op = new SortOpDesc + op.attributes = criteria.map { + case (attributeName, preference) => + val unit = new SortCriteriaUnit + unit.attributeName = attributeName + unit.sortPreference = preference + unit + }.toList + op + } + + /** + * A test-only Python operator whose code generation fails with a + * whitespace-padded message. No shipped operator raises a padded message, so + * this is the only way to pin the compiler's regex-group + `trim` extraction + * of the marker's payload. + */ + private class PaddedFailurePyOp extends PythonOperatorDescriptor { + override def asSource(): Boolean = true + override def generatePythonCode(): String = + throw new RuntimeException(" padded codegen failure ") + override def getOutputSchemas( + inputSchemas: Map[PortIdentity, Schema] + ): Map[PortIdentity, Schema] = Map(PortIdentity() -> Schema()) + override def operatorInfo: OperatorInfo = + OperatorInfo( + "padded", + "raises a padded message during code generation", + OperatorGroupConstants.PYTHON_GROUP, + List.empty, + List(OutputPort()) + ) + } + private val realCsvPath = "workflow-compiling-service/src/test/resources/country_sales_small.csv" @@ -317,6 +358,214 @@ class WorkflowCompilerSpec extends AnyFlatSpec { ) } + // -------------------- Python code-generation error path -------------------- + + // A `PythonOperatorDescriptor` whose `generatePythonCode` throws does not + // propagate the failure: it embeds `#EXCEPTION DURING CODE GENERATION: <msg>` + // in the generated code so schema propagation can still run. The compiler is + // the consumer that turns that marker back into a per-operator error, so these + // tests pin the marker -> error translation from the compiler's side. + + // Re-anchor the subject after the sub-section. + "WorkflowCompiler" should "accumulate a per-operator error when a Python operator's code generation fails" in { + val csv = csvOp(realCsvPath) + val unconfiguredSort = sortOp() // no sort keys -> generatePythonCode throws + + val result = new WorkflowCompiler(newContext()).compile( + LogicalPlanPojo( + operators = List(csv, unconfiguredSort), + links = List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + unconfiguredSort.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assert(result.physicalPlan.isEmpty, "any error must clear the physical plan") + val err = result.operatorIdToError(unconfiguredSort.operatorIdentifier) + assert(err.`type` == COMPILATION_ERROR) + assert(err.operatorId == unconfiguredSort.operatorIdentifier.id) + assert( + err.message.contains( + "Operator is not configured properly: " + + "requirement failed: Sort operator requires at least one sort key." + ), + s"unexpected message: ${err.message}" + ) + // The failure belongs to the Python operator alone; the upstream csv compiled. + assert( + !result.operatorIdToError.contains(csv.operatorIdentifier), + s"only the Python op should have errored, got ${result.operatorIdToError.keySet}" + ) + // Lenient mode records the error and keeps going *within* the same operator: + // the terminal sort's output port is still collected for storage, which only + // happens if the marker check did not abort the operator's expansion. + assert( + result.outputPortsNeedingStorage.exists( + _.opId.logicalOpId == unconfiguredSort.operatorIdentifier + ), + s"expected the sort's port to still be collected, got ${result.outputPortsNeedingStorage}" + ) + } + + it should "attribute each Python code-generation failure to its own logical operator" in { + val csv = csvOp(realCsvPath) + val noKeys = sortOp() + val blankKey = sortOp("" -> SortPreference.ASC) + + val result = new WorkflowCompiler(newContext()).compile( + LogicalPlanPojo( + operators = List(csv, noKeys, blankKey), + links = List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + noKeys.operatorIdentifier, + PortIdentity(0) + ), + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + blankKey.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assert( + result.operatorIdToError.keySet == + Set(noKeys.operatorIdentifier, blankKey.operatorIdentifier), + s"expected exactly the two Python ops in errors, got ${result.operatorIdToError.keySet}" + ) + // Each operator carries the message its *own* code generation raised — a + // mixed-up mapping would put the wrong diagnostic on the wrong UI node. + assert( + result + .operatorIdToError(noKeys.operatorIdentifier) + .message + .contains("Operator is not configured properly: requirement failed: Sort operator requires") + ) + assert( + result + .operatorIdToError(blankKey.operatorIdentifier) + .message + .contains( + "Operator is not configured properly: " + + "requirement failed: Each sort key must have an attribute selected." + ) + ) + // The rest of the plan still compiled: the csv's schemas survive. + assert( + result.operatorIdToOutputSchemas.contains(csv.operatorIdentifier), + "upstream csv's schemas should be retained even when downstream Python ops fail" + ) + } + + it should "trim the marker's message and report it as a plain RuntimeException" in { + val padded = new PaddedFailurePyOp + + val result = new WorkflowCompiler(newContext()).compile( + LogicalPlanPojo( + operators = List(padded), + links = List.empty, + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + // `message` is the RuntimeException's toString, so the extracted payload is + // the tail of it: exactly the raised message with its padding removed, and + // with the marker itself stripped off by the regex. + val message = result.operatorIdToError(padded.operatorIdentifier).message + assert( + message.endsWith("Operator is not configured properly: padded codegen failure"), + s"unexpected message: [$message]" + ) + // The head of it is the exception's class name: the compiler wraps the + // extracted payload in a plain `RuntimeException` and the error map stores + // `err.toString`, so the type is part of what the UI renders. Pinning it + // here keeps the wrapper type from silently drifting. + assert( + message.startsWith("java.lang.RuntimeException: "), + s"expected a plain RuntimeException to be reported, got: [$message]" + ) + assert( + !message.contains("#EXCEPTION DURING CODE GENERATION"), + s"the marker itself must not leak into the user-facing message: [$message]" + ) + } + + it should "report no code-generation error for a well-formed Python operator" in { + val csv = csvOp(realCsvPath) + val configuredSort = sortOp("Region" -> SortPreference.ASC) + + val result = new WorkflowCompiler(newContext()).compile( + LogicalPlanPojo( + operators = List(csv, configuredSort), + links = List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + configuredSort.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}") + assert(result.physicalPlan.isDefined) + // Same operator, same Python code path — the only difference is that code + // generation succeeded, so no marker is present to be turned into an error. + val sortPhysicalOps = + result.physicalPlan.get.getPhysicalOpsOfLogicalOp(configuredSort.operatorIdentifier) + assert(sortPhysicalOps.nonEmpty) + assert(sortPhysicalOps.forall(_.isPythonBased), "Sort must still be a Python-based operator") + assert(sortPhysicalOps.forall(!_.getCode.contains("#EXCEPTION DURING CODE GENERATION"))) + } + + it should "not subject non-Python operators to the code-generation check" in { + // Non-Python operators carry no code at all — `getCode` throws + // IllegalAccessError on them — so the check must stay behind the + // `isPythonBased` guard or every Scala operator would fail to compile. + val csv = csvOp(realCsvPath) + val filter = filterOp(new FilterPredicate("Region", ComparisonType.EQUAL_TO, "Asia")) + + val result = new WorkflowCompiler(newContext()).compile( + LogicalPlanPojo( + operators = List(csv, filter), + links = List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + filter.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assert(result.operatorIdToError.isEmpty, s"unexpected errors: ${result.operatorIdToError}") + val physicalOps = result.physicalPlan.get.operators + assert( + physicalOps.forall(!_.isPythonBased), + "this plan must contain no Python-based op, otherwise the test proves nothing" + ) + } + // -------------------- physical-plan shape -------------------- private def pojo( @@ -538,4 +787,57 @@ class WorkflowCompilerSpec extends AnyFlatSpec { s"the thrown schema error should name the missing attribute, got: $ex" ) } + + it should "throw immediately when a Python operator's code generation failed" in { + // The execution path passes no error buffer, so the marker found in the + // generated code must abort the compile instead of being collected. The + // lenient counterpart above turns the same marker into a per-operator error. + val csv = csvOp(realCsvPath) + val unconfiguredSort = sortOp() + + val ex = intercept[RuntimeException] { + new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, unconfiguredSort), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + unconfiguredSort.operatorIdentifier, + PortIdentity(0) + ) + ) + ), + CompilationErrorHandling.Strict + ) + } + assert( + ex.getMessage == "Operator is not configured properly: " + + "requirement failed: Sort operator requires at least one sort key.", + s"unexpected message: ${ex.getMessage}" + ) + } + + it should "not throw for a well-formed Python operator" in { + val csv = csvOp(realCsvPath) + val configuredSort = sortOp("Region" -> SortPreference.DESC) + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, configuredSort), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + configuredSort.operatorIdentifier, + PortIdentity(0) + ) + ) + ), + CompilationErrorHandling.Strict + ) + + assert(result.physicalPlan.isDefined) + assert(result.operatorIdToError.isEmpty) + } }
