From 79c7a254ee4878a5ed6acc9583857a3f48064899 Mon Sep 17 00:00:00 2001 From: eugenegujing Date: Thu, 13 Aug 2026 16:59:54 -0700 Subject: [PATCH] test(workflow-compiler): Extend WorkflowCompilerSpec to cover codegen error path Extend WorkflowCompilerSpec from 15 to 22 tests to cover the #EXCEPTION DURING CODE GENERATION marker check. The lenient path now asserts that the error is attributed to the failing logical operator, that two failing operators keep their own messages, that the marker's message is trimmed and reported as a plain RuntimeException, and that a well-formed Python operator and a non-Python operator produce no such error. The strict path asserts that the same marker throws immediately and that a well-formed Python operator does not. Five of the new tests drive the real SortOpDesc, whose code generation rejects an unconfigured operator; only the trim test needs a test-only fixture, since no shipped operator raises a whitespace-padded message. No production code is changed and no existing test is modified. --- .../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 94187373d79..da7ac594829 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: ` + // 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) + } }