Xiao-zhen-Liu commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3553405004
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala:
##########
@@ -116,111 +118,159 @@ object WorkflowCompiler {
}
case class WorkflowCompilationResult(
+ logicalPlan: LogicalPlan,
physicalPlan: Option[PhysicalPlan], // if physical plan is none, the
compilation is failed
operatorIdToOutputSchemas: Map[OperatorIdentity, Map[PortIdentity,
Option[Schema]]],
- operatorIdToError: Map[OperatorIdentity, WorkflowFatalError]
+ operatorIdToError: Map[OperatorIdentity, WorkflowFatalError],
+ outputPortsNeedingStorage: Set[GlobalPortIdentity]
)
class WorkflowCompiler(
context: WorkflowContext
) extends LazyLogging {
- // function to expand logical plan to physical plan
+ /**
+ * Function to expand logical plan to physical plan
+ * @return the expanded physical plan and a set of output ports that need
storage
+ */
private def expandLogicalPlan(
logicalPlan: LogicalPlan,
+ logicalOpsToViewResult: List[String],
errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]]
- ): PhysicalPlan = {
+ ): (PhysicalPlan, Set[GlobalPortIdentity]) = {
+ val terminalLogicalOps = logicalPlan.getTerminalOperatorIds
+ val logicalOpsNeedingStorage =
+ (terminalLogicalOps ++
logicalOpsToViewResult.map(OperatorIdentity(_))).toSet
var physicalPlan = PhysicalPlan(operators = Set.empty, links = Set.empty)
+ val outputPortsNeedingStorage: mutable.HashSet[GlobalPortIdentity] =
mutable.HashSet()
- logicalPlan.getTopologicalOpIds.asScala.foreach { logicalOpId =>
- val logicalOp = logicalPlan.getOperator(logicalOpId)
- val allUpstreamLinks =
logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier)
+ logicalPlan.getTopologicalOpIds.asScala.foreach(logicalOpId =>
+ Try {
+ val logicalOp = logicalPlan.getOperator(logicalOpId)
- try {
val subPlan = logicalOp.getPhysicalPlan(context.workflowId,
context.executionId)
-
subPlan
.topologicalIterator()
.map(subPlan.getOperator)
- .foreach { physicalOp =>
- val externalLinks = allUpstreamLinks
- .filter(link => physicalOp.inputPorts.contains(link.toPortId))
- .flatMap { link =>
- physicalPlan
- .getPhysicalOpsOfLogicalOp(link.fromOpId)
- .find(_.outputPorts.contains(link.fromPortId))
- .map(fromOp =>
- PhysicalLink(fromOp.id, link.fromPortId, physicalOp.id,
link.toPortId)
- )
- }
+ .foreach({ physicalOp =>
+ {
+ val externalLinks = logicalPlan
+ .getUpstreamLinks(logicalOp.operatorIdentifier)
+ .filter(link => physicalOp.inputPorts.contains(link.toPortId))
+ .flatMap { link =>
+ physicalPlan
+ .getPhysicalOpsOfLogicalOp(link.fromOpId)
+ .find(_.outputPorts.contains(link.fromPortId))
+ .map(fromOp =>
+ PhysicalLink(fromOp.id, link.fromPortId, physicalOp.id,
link.toPortId)
+ )
+ }
- val internalLinks = subPlan.getUpstreamPhysicalLinks(physicalOp.id)
+ val internalLinks =
subPlan.getUpstreamPhysicalLinks(physicalOp.id)
- // Add the operator to the physical plan
- physicalPlan =
physicalPlan.addOperator(physicalOp.propagateSchema())
+ // Add the operator to the physical plan
+ physicalPlan =
physicalPlan.addOperator(physicalOp.propagateSchema())
- // Add all the links to the physical plan
- physicalPlan = (externalLinks ++
internalLinks).foldLeft(physicalPlan) { (plan, link) =>
- plan.addLink(link)
- }
+ // Add all the links to the physical plan
+ physicalPlan = (externalLinks ++ internalLinks)
+ .foldLeft(physicalPlan) { (plan, link) => plan.addLink(link) }
- // **Check for Python-based operator errors during code
generation**
- if (physicalOp.isPythonBased) {
- val code = physicalOp.getCode
- val exceptionPattern = """#EXCEPTION DURING CODE
GENERATION:\s*(.*)""".r
+ // **Check for Python-based operator errors during code
generation**
+ if (physicalOp.isPythonBased) {
+ val code = physicalOp.getCode
+ val exceptionPattern = """#EXCEPTION DURING CODE
GENERATION:\s*(.*)""".r
- exceptionPattern.findFirstMatchIn(code).foreach { matchResult =>
- val errorMessage = matchResult.group(1).trim
- val error =
- new RuntimeException(s"Operator is not configured properly:
$errorMessage")
+ exceptionPattern.findFirstMatchIn(code).foreach { matchResult
=>
+ val errorMessage = matchResult.group(1).trim
+ val error =
+ new RuntimeException(s"Operator is not configured
properly: $errorMessage")
- errorList match {
- case Some(list) => list.append((logicalOpId, error)) //
Store error and continue
- case None => throw error // Throw immediately if no
error list is provided
+ errorList match {
+ case Some(list) => list.append((logicalOpId, error)) //
Store error and continue
+ case None => throw error // Throw immediately if no
error list is provided
+ }
}
}
}
+ })
+
+ // convert logical operators needing storage to output ports needing
storage
+ subPlan
+ .topologicalIterator()
+ .filter(opId => logicalOpsNeedingStorage.contains(opId.logicalOpId))
+ .map(physicalPlan.getOperator)
+ .foreach { physicalOp =>
+ physicalOp.outputPorts
+ .filterNot(_._1.internal)
+ .foreach {
+ case (outputPortId, _) =>
+ outputPortsNeedingStorage += GlobalPortIdentity(
+ opId = physicalOp.id,
+ portId = outputPortId
+ )
+ }
}
- } catch {
- case e: Throwable =>
+ } match {
+ case Success(_) =>
+
+ case Failure(err) =>
errorList match {
- case Some(list) => list.append((logicalOpId, e)) // Store error
- case None => throw e // Throw if no list is provided
+ case Some(list) => list.append((logicalOpId, err))
+ case None => throw err
}
}
- }
-
- physicalPlan
+ )
+ (physicalPlan, outputPortsNeedingStorage.toSet)
}
/**
- * Compile a workflow to physical plan, along with the schema propagation
result and error(if any)
+ * Compile a workflow to physical plan, along with the schema propagation
result and error(if any).
*
* @param logicalPlanPojo the pojo parsed from workflow str provided by user
- * @return WorkflowCompilationResult, containing the physical plan, input
schemas per op and error per op
+ * @param errorHandling Lenient (editing-time, collect all errors) or
Strict (pre-execution, throw)
+ * @return WorkflowCompilationResult, containing the logical plan, physical
plan, output schemas per
+ * op, errors per op, and the output ports that need storage
*/
def compile(
- logicalPlanPojo: LogicalPlanPojo
+ logicalPlanPojo: LogicalPlanPojo,
+ errorHandling: CompilationErrorHandling =
CompilationErrorHandling.Lenient
): WorkflowCompilationResult = {
- val errorList = new ArrayBuffer[(OperatorIdentity, Throwable)]()
- var opIdToOutputSchema: Map[OperatorIdentity, Map[PortIdentity,
Option[Schema]]] = Map()
+ // Lenient collects into a buffer; Strict passes None so the first error
is thrown.
+ val errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] =
+ errorHandling match {
+ case CompilationErrorHandling.Lenient =>
+ Some(new ArrayBuffer[(OperatorIdentity, Throwable)]())
+ case CompilationErrorHandling.Strict => None
+ }
+
// 1. convert the pojo to logical plan
val logicalPlan: LogicalPlan = LogicalPlan(logicalPlanPojo)
// 2. resolve the file name in each scan source operator
- logicalPlan.resolveScanSourceOpFileName(Some(errorList))
+ logicalPlan.resolveScanSourceOpFileName(errorList)
- // 3. expand the logical plan to the physical plan
- val physicalPlan = expandLogicalPlan(logicalPlan, Some(errorList))
+ // 3. expand the logical plan to the physical plan, and get the output
ports that need storage
+ val (physicalPlan, outputPortsNeedingStorage) =
+ expandLogicalPlan(logicalPlan, logicalPlanPojo.opsToViewResult,
errorList)
// 4. collect the output schema for each logical op
- // even if error is encountered when logical => physical, we still want to
get the input schemas for rest no-error operators
- opIdToOutputSchema = collectOutputSchemaFromPhysicalPlan(physicalPlan,
errorList)
+ // even if error is encountered when logical => physical, we still want to
get the input schemas
+ // for the rest of the no-error operators. In Strict mode there is no
buffer (errors already thrown),
+ // so schema errors go to a throwaway buffer and do not affect the result.
+ val schemaErrorList = errorList.getOrElse(new
ArrayBuffer[(OperatorIdentity, Throwable)]())
Review Comment:
Strict sets `errorList = None`, so this buffer is never read and `hasErrors`
(line 263) is always false. Schema-propagation errors (e.g. a Projection on a
missing column) get dropped, `physicalPlan` stays `Some`, and the run proceeds.
Matches old amber (not a regression) but contradicts the fail-fast claim above
— and such a plan can still be launched from the UI. Fix the wording, or throw
when the buffer is non-empty (a behavior change → probably a follow-up).
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala:
##########
@@ -17,25 +17,18 @@
* under the License.
*/
-package org.apache.texera.amber.compiler.model
+package org.apache.texera.common.compiler
-import com.fasterxml.jackson.annotation.{JsonCreator, JsonProperty}
-import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
-import org.apache.texera.amber.core.workflow.PortIdentity
+/**
+ * Controls how the compiler reacts to a per-operator compilation error.
+ *
+ * - [[CompilationErrorHandling.Lenient]] collects every error and returns
them in the result,
+ * with `physicalPlan = None` when any error occurred. Used for
editing-time validation.
+ * - [[CompilationErrorHandling.Strict]] throws on the first error. Used
before execution.
Review Comment:
Small accuracy fix: Strict doesn't throw on schema-propagation errors — they
become `Left` and are never thrown (see `WorkflowCompiler.scala:260`).
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala:
##########
@@ -116,111 +118,159 @@ object WorkflowCompiler {
}
case class WorkflowCompilationResult(
+ logicalPlan: LogicalPlan,
physicalPlan: Option[PhysicalPlan], // if physical plan is none, the
compilation is failed
operatorIdToOutputSchemas: Map[OperatorIdentity, Map[PortIdentity,
Option[Schema]]],
- operatorIdToError: Map[OperatorIdentity, WorkflowFatalError]
+ operatorIdToError: Map[OperatorIdentity, WorkflowFatalError],
+ outputPortsNeedingStorage: Set[GlobalPortIdentity]
)
class WorkflowCompiler(
context: WorkflowContext
) extends LazyLogging {
- // function to expand logical plan to physical plan
+ /**
+ * Function to expand logical plan to physical plan
+ * @return the expanded physical plan and a set of output ports that need
storage
+ */
private def expandLogicalPlan(
logicalPlan: LogicalPlan,
+ logicalOpsToViewResult: List[String],
errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]]
- ): PhysicalPlan = {
+ ): (PhysicalPlan, Set[GlobalPortIdentity]) = {
+ val terminalLogicalOps = logicalPlan.getTerminalOperatorIds
+ val logicalOpsNeedingStorage =
+ (terminalLogicalOps ++
logicalOpsToViewResult.map(OperatorIdentity(_))).toSet
var physicalPlan = PhysicalPlan(operators = Set.empty, links = Set.empty)
+ val outputPortsNeedingStorage: mutable.HashSet[GlobalPortIdentity] =
mutable.HashSet()
- logicalPlan.getTopologicalOpIds.asScala.foreach { logicalOpId =>
- val logicalOp = logicalPlan.getOperator(logicalOpId)
- val allUpstreamLinks =
logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier)
+ logicalPlan.getTopologicalOpIds.asScala.foreach(logicalOpId =>
+ Try {
Review Comment:
Behavior change from the merge: the old compiling-service used `catch { case
e: Throwable }`; `scala.util.Try` catches only non-fatal errors. So a fatal
error during expansion now fails the `/compile` request instead of being
recorded per-operator. Probably fine — flagging since the PR says nothing
changes.
##########
common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala:
##########
@@ -81,6 +80,11 @@ case class LogicalPlan(
def getOperator(opId: OperatorIdentity): LogicalOp = operatorMap(opId)
+ def getTerminalOperatorIds: List[OperatorIdentity] =
+ operatorMap.keys
+ .filter(op => jgraphtDag.outDegreeOf(op) == 0)
+ .toList
+
def addOperator(op: LogicalOp): LogicalPlan = {
Review Comment:
Stale TODO, and `addOperator`/`addLink` came over from the old code with no
callers I could find — drop them or note why they stay.
##########
common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala:
##########
@@ -314,4 +316,168 @@ class WorkflowCompilerSpec extends AnyFlatSpec {
s"expected both csvs in errors, got ${result.operatorIdToError.keySet}"
)
}
+
+ // -------------------- physical-plan shape --------------------
+
+ private def pojo(
+ operators: List[org.apache.texera.amber.operator.LogicalOp],
+ links: List[LogicalLink],
+ opsToViewResult: List[String] = List.empty
+ ): LogicalPlanPojo =
+ LogicalPlanPojo(operators, links, opsToViewResult, List.empty)
+
+ // Re-anchor the subject after the sub-section.
+ "WorkflowCompiler" should "produce a physical plan that contains at least
one physical op per logical op" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ assert(result.logicalPlan.operators.size == 2)
+ val physicalPlan = result.physicalPlan.get
+
assert(physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).nonEmpty)
+
assert(physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).nonEmpty)
+ }
+
+ it should "translate a logical link into a physical link between the two
logical ops' physical ops" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ val physicalPlan = result.physicalPlan.get
+ val csvPhysIds =
+
physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).map(_.id).toSet
+ val keywordPhysIds =
+
physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).map(_.id).toSet
+
+ val bridging = physicalPlan.links.filter(l =>
+ csvPhysIds.contains(l.fromOpId) && keywordPhysIds.contains(l.toOpId)
+ )
+ assert(bridging.nonEmpty, "expected at least one physical link from csv to
keyword")
+ }
+
+ // -------------------- storage-port collection --------------------
+
+ // The compiler walks `logicalPlan.getTerminalOperatorIds` (logical ops with
+ // out-degree 0) plus `opsToViewResult`, and for every physical op of those
+ // logical ops collects every non-internal output port into the result's
+ // `outputPortsNeedingStorage`. These tests pin both the terminal-default and
+ // the opsToViewResult-additive paths, and that internal ports are filtered.
+
+ "WorkflowCompiler" should "mark the terminal op's output port as needing
storage" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ )
+ )
+ )
+
+ val storage = result.outputPortsNeedingStorage
+ assert(
+ storage.exists(_.opId.logicalOpId == keyword.operatorIdentifier),
+ s"expected keyword to be marked for storage, got
${storage.map(_.opId.logicalOpId)}"
+ )
+ assert(
+ !storage.exists(_.opId.logicalOpId == csv.operatorIdentifier),
+ "csv is not terminal and was not requested via opsToViewResult; it
should not be in storage"
+ )
+ }
+
+ it should "also mark a non-terminal op for storage when it is named in
opsToViewResult" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+ val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia")
+
+ val result = new WorkflowCompiler(newContext()).compile(
+ pojo(
+ List(csv, keyword),
+ List(
+ LogicalLink(
+ csv.operatorIdentifier,
+ PortIdentity(),
+ keyword.operatorIdentifier,
+ PortIdentity()
+ )
+ ),
+ opsToViewResult = List(csv.operatorIdentifier.id)
+ )
+ )
+
+ val logicalOpsInStorage =
result.outputPortsNeedingStorage.map(_.opId.logicalOpId)
+ assert(
+ logicalOpsInStorage.contains(csv.operatorIdentifier),
+ s"opsToViewResult should add csv to storage, got $logicalOpsInStorage"
+ )
+ assert(
+ logicalOpsInStorage.contains(keyword.operatorIdentifier),
+ s"terminal keyword should remain in storage, got $logicalOpsInStorage"
+ )
+ }
+
+ it should "treat a single source op as terminal and mark its output port for
storage" in {
+ val csv = TestOperators.smallCsvScanOpDesc()
+
+ val result = new WorkflowCompiler(newContext()).compile(pojo(List(csv),
List.empty))
+
+ val storage = result.outputPortsNeedingStorage
+ assert(
+ storage.exists(_.opId.logicalOpId == csv.operatorIdentifier),
+ "single op has out-degree 0, so its output port should land in storage"
+ )
+ assert(
+ storage.forall(!_.portId.internal),
+ "compiler must filter out internal ports; storage should expose only
user-visible outputs"
+ )
+ }
+
+ // -------------------- strict-mode error semantics --------------------
+
+ // Re-anchor the subject after the sub-section.
+ "WorkflowCompiler in strict mode" should "throw when a scan source has no
fileName set" in {
Review Comment:
Only one Strict test. Worth adding a Strict success case (the execution path
relies on `physicalPlan.get`) and a schema-error case to pin the behavior. The
Python code-gen error branch (~line 179) is untested in both modes too.
##########
amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala:
##########
@@ -110,8 +110,16 @@ class WorkflowExecutionService(
def executeWorkflow(): Unit = {
try {
- workflow = new WorkflowCompiler(workflowContext)
- .compile(request.logicalPlan)
+ val compilationResult = new WorkflowCompiler(workflowContext)
+ .compile(request.logicalPlan, CompilationErrorHandling.Strict)
+ workflowContext.workflowSettings = workflowContext.workflowSettings.copy(
+ outputPortsNeedingStorage = compilationResult.outputPortsNeedingStorage
+ )
+ workflow = Workflow(
+ workflowContext,
+ compilationResult.logicalPlan,
+ compilationResult.physicalPlan.get
+ )
} catch {
case err: Throwable =>
errorHandler(err)
Review Comment:
+1 to bobbai00's point here: if `compile` throws, this reports the error but
execution then falls through to `workflow.context` below with `workflow` still
null → a second NPE that masks the real message. A `return` after
`errorHandler(err)` fixes it. Pre-existing, but a good spot to fix now that
Strict is the fail-fast path.
--
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]
