Yicong-Huang commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3607068805
##########
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:
Update: fixed in-PR after all (in 192f02d72) rather than deferred — Strict
now throws when schema-propagation errors are present (`compile` throws the
first collected schema error), and the pinning test flipped to assert the
throw. The docs and PR description now describe this as part of Strict's
contract.
##########
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:
Update: no longer an exception — Strict now fails fast on schema-propagation
errors too (fixed in 192f02d72), and the scaladoc says so.
##########
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:
Update in 192f02d72: the schema-error test now asserts the Strict throw
(fail-fast was fixed in-PR instead of deferred).
--
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]
