Skip to content

refactor(workflow-compiler): unify duplicate compilers into common/workflow-compiler #6143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Yicong-Huang wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(workflow-compiler): unify duplicate compilers into common/workflow-compiler #6143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/build.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,7 @@ jobs:
"PyBuilder/jacoco" \
"WorkflowCore/jacoco" \
"WorkflowOperator/jacoco" \
"WorkflowCompiler/jacoco" \
"WorkflowExecutionService/jacoco"
- name: Upload amber and common coverage to Codecov
if: always()
Expand Down
2 changes: 1 addition & 1 deletion amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Workflow.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
package org.apache.texera.amber.engine.architecture.coordinator

import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext}
import org.apache.texera.workflow.LogicalPlan
import org.apache.texera.common.compiler.model.LogicalPlan

case class Workflow(
context: WorkflowContext,
Expand Down
2 changes: 1 addition & 1 deletion ...n/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
package org.apache.texera.web.model.websocket.request

import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.workflow.LogicalLink
import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo}

case class EditingTimeCompilationRequest(
operators: List[LogicalOp],
Expand Down
10 changes: 1 addition & 9 deletions ...src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,7 @@ package org.apache.texera.web.model.websocket.request

import com.fasterxml.jackson.databind.annotation.JsonDeserialize
import org.apache.texera.amber.core.workflow.WorkflowSettings
import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.workflow.LogicalLink
import org.apache.texera.common.compiler.model.LogicalPlanPojo

case class ReplayExecutionInfo(
@JsonDeserialize(contentAs = classOf[java.lang.Long])
Expand All @@ -39,10 +38,3 @@ case class WorkflowExecuteRequest(
emailNotificationEnabled: Boolean,
computingUnitId: Int
) extends TexeraWebSocketRequest

case class LogicalPlanPojo(
operators: List[LogicalOp],
links: List[LogicalLink],
opsToViewResult: List[String],
opsToReuseResult: List[String]
)
8 changes: 5 additions & 3 deletions amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,10 @@ import io.reactivex.rxjava3.core.Observable
import org.apache.texera.auth.SessionUser
import org.apache.texera.dao.SqlServer
import org.apache.texera.dao.jooq.generated.Tables.OPERATOR_EXECUTIONS
import org.apache.texera.web.model.websocket.request.{LogicalPlanPojo, WorkflowExecuteRequest}
import org.apache.texera.workflow.{LogicalLink, WorkflowCompiler}
import org.apache.texera.common.compiler.model.LogicalPlanPojo
import org.apache.texera.web.model.websocket.request.WorkflowExecuteRequest
import org.apache.texera.common.compiler.model.LogicalLink
import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler}
import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource
import org.apache.texera.web.service.{ExecutionResultService, WorkflowService}
import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState
Expand Down Expand Up @@ -895,7 +897,7 @@ class SyncExecutionResource extends LazyLogging {
try {
val tempContext = new WorkflowContext(WorkflowIdentity(workflowId))
val compiler = new WorkflowCompiler(tempContext)
compiler.compile(logicalPlan)
compiler.compile(logicalPlan, CompilationErrorHandling.Strict)
Map.empty
} catch {
case e: Exception =>
Expand Down
14 changes: 11 additions & 3 deletions amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutions
import org.apache.texera.web.storage.ExecutionStateStore
import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState
import org.apache.texera.web.{ComputingUnitMaster, SubscriptionManager, WebsocketInput}
import org.apache.texera.workflow.WorkflowCompiler
import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler}

import java.net.URI
import scala.collection.mutable
Expand Down Expand Up @@ -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)

@bobbai00 bobbai00 Jul 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we directly return to end this function here ?

@Xiao-zhen-Liu Xiao-zhen-Liu Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+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.

Expand Down
2 changes: 1 addition & 1 deletion amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ import org.apache.texera.web.service.WorkflowService.mkWorkflowStateId
import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState
import org.apache.texera.web.storage.{ExecutionStateStore, WorkflowStateStore}
import org.apache.texera.web.{SubscriptionManager, WorkflowLifecycleManager}
import org.apache.texera.workflow.LogicalPlan
import org.apache.texera.common.compiler.model.LogicalPlan
import play.api.libs.json.Json

import java.net.URI
Expand Down
124 changes: 0 additions & 124 deletions amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala
View file
Open in desktop

This file was deleted.

160 changes: 0 additions & 160 deletions amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala
View file
Open in desktop

This file was deleted.

Loading
Loading