Yicong-Huang commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3606734335
##########
common/workflow-compiler/DESIGN.md:
##########
@@ -0,0 +1,61 @@
+# Design: Unified `WorkflowCompiler`
Review Comment:
Removed in bbbf9f5841677f4708c4c17c3b77f98889fcae80 — the design rationale
now lives in `WorkflowCompiler`'s class-level scaladoc instead (no other module
ships a DESIGN.md, agreed it didn't belong).
##########
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:
Done in bbbf9f5841677f4708c4c17c3b77f98889fcae80 — added an early `return`
after `errorHandler(err)` so a failed compile no longer falls through to the
null `workflow` below.
##########
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))
Review Comment:
Done in bbbf9f5841677f4708c4c17c3b77f98889fcae80 — `getUpstreamLinks` is now
computed once per logical op, outside the per-physical-op loop.
--
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]
