bobbai00 commented on code in PR #6143:
URL: https://github.com/apache/texera/pull/6143#discussion_r3525373950
##########
common/workflow-compiler/DESIGN.md:
##########
@@ -0,0 +1,61 @@
+# Design: Unified `WorkflowCompiler`
Review Comment:
Should this one be removed ?
##########
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:
can we save the upstream links as a variable outside of this loop ?
##########
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:
can we directly `return` to end this function here ?
--
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]
