Yicong-Huang commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3606735822
##########
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:
Good catch. Went with the wording fix in
bbbf9f5841677f4708c4c17c3b77f98889fcae80: the comment now states explicitly
that schema-propagation failures are NOT thrown in Strict and surface as `None`
schemas (legacy execution-path behavior), and a new spec pins exactly that
(`currently not throw on schema-propagation errors`). Making Strict throw there
is a behavior change I'd rather do as a follow-up.
##########
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:
Right — `Try` catches only non-fatal errors, so a fatal error during
expansion now fails the `/compile` request instead of being recorded
per-operator. Keeping it: swallowing fatal JVM errors into a per-operator error
map was arguably the worse behavior. Called this out in the updated PR
description as a known minor behavior change.
##########
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:
Fixed in bbbf9f5841677f4708c4c17c3b77f98889fcae80 — the scaladoc now says
Strict throws during file resolution and logical-to-physical expansion, but
schema-propagation failures surface as `None` schemas (follow-up tracks failing
fast there too).
--
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]
