This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6143-42f9d0e90e15e13acc420b70d4f635f30d3e832c in repository https://gitbox.apache.org/repos/asf/texera.git
commit a7989ee78e27ab7b909bdf8aaed586a5f030528c Author: Yicong Huang <[email protected]> AuthorDate: Mon Jul 27 13:17:42 2026 -0700 refactor(workflow-compiler): unify duplicate compilers into common/workflow-compiler (#6143) ### What changes were proposed in this PR? **This is a source-level de-duplication only — the runtime behavior of two compilations stays.** A workflow is still compiled twice: once at editing time (through workflow-compiling-service) and once at execution time (through amber). This PR only makes both paths share one compiler implementation instead of two hand-synced copies. amber and workflow-compiling-service each maintained a separate copy of the workflow compiler and its models (`WorkflowCompiler`, `LogicalPlan`, `LogicalLink`, `LogicalPlanPojo`), which had to be kept in sync by hand (e.g. #6288 had to patch both copies). This PR consolidates them into a single shared `common/workflow-compiler` module that both services depend on; the old per-service copies are removed. The unified `WorkflowCompiler.compile` returns a `WorkflowCompilationResult` (logical plan, optional physical plan, per-port output schemas, per-operator errors, and the output ports needing storage) and takes a `CompilationErrorHandling` mode: `Lenient` for the editing-time path (accumulate per-operator errors so the UI can render them) and `Strict` for the execution path (fail-fast before a run). amber's engine `Workflow` becomes a thin wrapper over the compilation result, built via `Workflow.fromCompilationResult`. Two known minor behavior changes (both flagged in review, both intentional): - Plan expansion now uses `scala.util.Try` (non-fatal only) where the old compiling-service caught `Throwable`, so a fatal JVM error during expansion now fails the `/compile` request instead of being recorded per-operator. - `Strict` now also fails fast on schema-propagation errors (e.g. a Projection on a missing column). The legacy execution path silently launched such plans and they only failed at runtime; per review discussion the fail-fast contract now covers this case too, pinned by a test. ### Any related issues, documentation, discussions? Closes #6142. Design rationale lives in the scaladoc of `WorkflowCompiler`. ### How was this PR tested? - Migrated the compiler and model unit specs into the new module and added strict-mode coverage (success path and the schema-propagation fail-fast path). - amber and workflow-compiling-service compile and test against the shared library. - Rebased over #6288 and verified its error-message changes carry over into the shared module (they now exist in one place instead of two). ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 (Claude Code); review follow-ups by Claude Fable 5 (Claude Code) --------- Co-authored-by: Claude Fable 5 <[email protected]> --- .github/workflows/build.yml | 1 + .../engine/architecture/coordinator/Workflow.scala | 29 ++- .../org/apache/texera/amber/error/ErrorUtils.scala | 17 +- .../request/EditingTimeCompilationRequest.scala | 2 +- .../websocket/request/WorkflowExecuteRequest.scala | 10 +- .../web/resource/SyncExecutionResource.scala | 8 +- .../web/service/WorkflowExecutionService.scala | 10 +- .../texera/web/service/WorkflowService.scala | 2 +- .../apache/texera/workflow/WorkflowCompiler.scala | 160 ------------ .../amber/engine/e2e/LoopIntegrationSpec.scala | 2 +- .../e2e/MultiRegionWorkflowIntegrationSpec.scala | 2 +- .../e2e/ReconfigurationIntegrationSpec.scala | 2 +- .../coordinator/WorkflowSchedulerSpec.scala | 2 +- .../architecture/coordinator/WorkflowSpec.scala | 83 ++++++ .../CostBasedScheduleGeneratorSpec.scala | 2 +- .../scheduling/DefaultCostEstimatorSpec.scala | 2 +- .../ExpansionGreedyScheduleGeneratorSpec.scala | 2 +- .../resourcePolicies/ResourcePoliciesSpec.scala | 2 +- .../engine/e2e/BatchSizePropagationSpec.scala | 2 +- .../amber/engine/e2e/DataProcessingSpec.scala | 2 +- .../apache/texera/amber/engine/e2e/PauseSpec.scala | 2 +- .../amber/engine/e2e/ReconfigurationSpec.scala | 2 +- .../apache/texera/amber/engine/e2e/TestUtils.scala | 13 +- .../engine/faulttolerance/CheckpointSpec.scala | 2 +- .../web/service/WorkflowExecutionServiceSpec.scala | 3 +- .../texera/workflow/WorkflowCompilerSpec.scala | 212 --------------- build.sbt | 9 +- common/workflow-compiler/build.sbt | 54 ++++ .../common/compiler/CompilationErrorHandling.scala | 26 +- .../texera/common/compiler/WorkflowCompiler.scala | 284 +++++++++++++++++++++ .../common/compiler/model}/LogicalLink.scala | 2 +- .../common/compiler/model}/LogicalPlan.scala | 9 +- .../common}/compiler/model/LogicalPlanPojo.scala | 2 +- .../texera/common/compiler}/LogicalPlanSpec.scala | 4 +- .../common}/compiler/WorkflowCompilerSpec.scala | 240 ++++++++++++++++- .../common/compiler/model}/LogicalLinkSpec.scala | 2 +- .../apache/texera/amber/util/StackTraceUtils.scala | 37 +-- .../texera/amber/compiler/WorkflowCompiler.scala | 231 ----------------- .../texera/amber/compiler/model/LogicalPlan.scala | 138 ---------- .../resource/WorkflowCompilationResource.scala | 10 +- .../resource/WorkflowCompilationResourceSpec.scala | 4 +- 41 files changed, 781 insertions(+), 847 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index b2feaad21d..becd8deb34 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -312,6 +312,7 @@ jobs: "PyBuilder/jacoco" \ "WorkflowCore/jacoco" \ "WorkflowOperator/jacoco" \ + "WorkflowCompiler/jacoco" \ "WorkflowExecutionService/jacoco" - name: Upload amber and common coverage to Codecov if: always() diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Workflow.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Workflow.scala index 6ec92deab8..b485b1e7a2 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Workflow.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/Workflow.scala @@ -20,10 +20,37 @@ 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.WorkflowCompilationResult +import org.apache.texera.common.compiler.model.LogicalPlan case class Workflow( context: WorkflowContext, logicalPlan: LogicalPlan, physicalPlan: PhysicalPlan ) + +object Workflow { + + /** + * Builds the engine `Workflow` from a Strict-mode compilation result. Also writes the + * result's `outputPortsNeedingStorage` back onto the context so the schedule generators + * materialize terminal results — every execution-path caller must do both, so they are + * kept together here. Strict mode guarantees `physicalPlan` is defined. + */ + def fromCompilationResult( + context: WorkflowContext, + compilationResult: WorkflowCompilationResult + ): Workflow = { + val physicalPlan = compilationResult.physicalPlan.getOrElse( + throw new IllegalStateException( + "fromCompilationResult requires a compilation result with a defined physicalPlan " + + "(compile with CompilationErrorHandling.Strict); this result carries " + + s"${compilationResult.operatorIdToError.size} compilation error(s) instead" + ) + ) + context.workflowSettings = context.workflowSettings.copy( + outputPortsNeedingStorage = compilationResult.outputPortsNeedingStorage + ) + Workflow(context, compilationResult.logicalPlan, physicalPlan) + } +} diff --git a/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala b/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala index d7c021524c..139f7b41a7 100644 --- a/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala +++ b/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala @@ -24,7 +24,7 @@ import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity import org.apache.texera.amber.engine.architecture.rpc.controlcommands.ConsoleMessage import org.apache.texera.amber.engine.architecture.rpc.controlcommands.ConsoleMessageType.ERROR import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{ControlError, ErrorLanguage} -import org.apache.texera.amber.util.VirtualIdentityUtils +import org.apache.texera.amber.util.{StackTraceUtils, VirtualIdentityUtils} import java.time.Instant import scala.util.control.ControlThrowable @@ -90,19 +90,8 @@ object ErrorUtils { } } - def getStackTraceWithAllCauses(err: Throwable, topLevel: Boolean = true): String = { - val header = if (topLevel) { - "Stack trace for developers: \n\n" - } else { - "\n\nCaused by:\n" - } - val message = header + err.toString + "\n" + err.getStackTrace.mkString("\n") - if (err.getCause != null) { - message + getStackTraceWithAllCauses(err.getCause, topLevel = false) - } else { - message - } - } + def getStackTraceWithAllCauses(err: Throwable, topLevel: Boolean = true): String = + StackTraceUtils.getStackTraceWithAllCauses(err, topLevel) def getOperatorFromActorIdOpt( actorIdOpt: Option[ActorVirtualIdentity] diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala index e15b441fca..f7a585bb15 100644 --- a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala +++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala @@ -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], diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala index a346a1ec0a..0059af1c1a 100644 --- a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala +++ b/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala @@ -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]) @@ -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] -) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala index 3c9da603c8..afcd7f63a1 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala @@ -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 @@ -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 => diff --git a/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala b/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala index 43b2288b39..4c8eb209b2 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala @@ -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 @@ -110,11 +110,15 @@ class WorkflowExecutionService( def executeWorkflow(): Unit = { try { - workflow = new WorkflowCompiler(workflowContext) - .compile(request.logicalPlan) + val compilationResult = new WorkflowCompiler(workflowContext) + .compile(request.logicalPlan, CompilationErrorHandling.Strict) + workflow = Workflow.fromCompilationResult(workflowContext, compilationResult) } catch { case err: Throwable => + // stop here: `workflow` is still null, so falling through would NPE + // below and mask the reported compilation error errorHandler(err) + return } client = ComputingUnitMaster.createAmberRuntime( diff --git a/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala b/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala index 46e738f35b..51d5b5677d 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala @@ -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 diff --git a/amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala b/amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala deleted file mode 100644 index ae576c4053..0000000000 --- a/amber/src/main/scala/org/apache/texera/workflow/WorkflowCompiler.scala +++ /dev/null @@ -1,160 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.workflow - -import com.typesafe.scalalogging.LazyLogging -import org.apache.texera.amber.core.virtualidentity.OperatorIdentity -import org.apache.texera.amber.core.workflow._ -import org.apache.texera.amber.engine.architecture.coordinator.Workflow -import org.apache.texera.web.model.websocket.request.LogicalPlanPojo - -import scala.collection.mutable -import scala.collection.mutable.ArrayBuffer -import scala.jdk.CollectionConverters.IteratorHasAsScala -import scala.util.{Failure, Success, Try} - -class WorkflowCompiler( - context: WorkflowContext -) extends LazyLogging { - - /** - * 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, 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 => - Try { - val logicalOp = logicalPlan.getOperator(logicalOpId) - - val subPlan = logicalOp.getPhysicalPlan(context.workflowId, context.executionId) - subPlan - .topologicalIterator() - .map(subPlan.getOperator) - .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) - - // 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) } - - // **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") - - 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 - ) - } - } - } match { - case Success(_) => - - case Failure(err) => - errorList match { - case Some(list) => list.append((logicalOpId, err)) - case None => throw err - } - } - ) - (physicalPlan, outputPortsNeedingStorage.toSet) - } - - /** - * Compile a workflow to physical plan, along with the schema propagation result and error(if any) - * - * Comparing to WorkflowCompilingService's compiler, which is used solely for workflow editing, - * This compile is used before executing the workflow. - * - * TODO: we should consider merge this compile with WorkflowCompilingService's compile - * @param logicalPlanPojo the pojo parsed from workflow str provided by user - * @return Workflow, containing the physical plan, logical plan and workflow context - */ - def compile( - logicalPlanPojo: LogicalPlanPojo - ): Workflow = { - // 1. convert the pojo to logical plan - val logicalPlan: LogicalPlan = LogicalPlan(logicalPlanPojo) - - // 2. resolve the file name in each scan source operator - logicalPlan.resolveScanSourceOpFileName(None) - - // 3. expand the logical plan to the physical plan, and get a set of output ports that need storage - val (physicalPlan, outputPortsNeedingStorage) = - expandLogicalPlan(logicalPlan, logicalPlanPojo.opsToViewResult, None) - - context.workflowSettings = context.workflowSettings.copy( - outputPortsNeedingStorage = outputPortsNeedingStorage - ) - - Workflow(context, logicalPlan, physicalPlan) - } -} diff --git a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala index 3dec510b90..9824d6d316 100644 --- a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala +++ b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala @@ -46,7 +46,7 @@ import org.apache.texera.amber.operator.loop.{LoopEndOpDesc, LoopStartOpDesc} import org.apache.texera.amber.operator.sleep.SleepOpDesc import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc import org.apache.texera.amber.tags.IntegrationTest -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpecLike import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} diff --git a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala index 23a7e5c4ee..83e942a300 100644 --- a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala +++ b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala @@ -41,7 +41,7 @@ import org.apache.texera.amber.operator.sort.{SortCriteriaUnit, SortOpDesc, Sort import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc import org.apache.texera.amber.operator.union.UnionOpDesc import org.apache.texera.amber.tags.IntegrationTest -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} import org.scalatest.flatspec.AnyFlatSpecLike diff --git a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala index 3be29cebe5..6ffd8b356b 100644 --- a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala +++ b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/ReconfigurationIntegrationSpec.scala @@ -46,7 +46,7 @@ import org.apache.texera.amber.engine.e2e.TestUtils.{ import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc import org.apache.texera.amber.operator.{LogicalOp, TestOperators} import org.apache.texera.amber.tags.IntegrationTest -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} import org.scalatest.flatspec.AnyFlatSpecLike diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSchedulerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSchedulerSpec.scala index ae4ce991d9..8cf601e2d6 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSchedulerSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSchedulerSpec.scala @@ -23,7 +23,7 @@ import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow import org.apache.texera.amber.operator.TestOperators -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpec class WorkflowSchedulerSpec extends AnyFlatSpec { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSpec.scala new file mode 100644 index 0000000000..e8987945de --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/WorkflowSpec.scala @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.engine.architecture.coordinator + +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} +import org.scalatest.flatspec.AnyFlatSpec + +/** + * Direct unit coverage for [[Workflow.fromCompilationResult]]. The adapter owns the + * execution path's two obligations after a Strict compile: writing the result's + * `outputPortsNeedingStorage` back onto the workflow context (the schedule generators + * read it from there to materialize terminal results) and unwrapping the physical + * plan. The context write used to live inside amber's compiler and was pinned by its + * spec; these cases keep that behavior directly anchored after the move. + */ +class WorkflowSpec extends AnyFlatSpec { + + "Workflow.fromCompilationResult" should + "write outputPortsNeedingStorage onto the context and unwrap the physical plan" in { + val csv = TestOperators.smallCsvScanOpDesc() + val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") + val context = new WorkflowContext() + + val result = new WorkflowCompiler(context).compile( + LogicalPlanPojo( + List(csv, keyword), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(), + keyword.operatorIdentifier, + PortIdentity() + ) + ), + List(), + List() + ), + CompilationErrorHandling.Strict + ) + val workflow = Workflow.fromCompilationResult(context, result) + + assert(context.workflowSettings.outputPortsNeedingStorage == result.outputPortsNeedingStorage) + assert(context.workflowSettings.outputPortsNeedingStorage.nonEmpty) + assert(workflow.physicalPlan eq result.physicalPlan.get) + assert(workflow.logicalPlan eq result.logicalPlan) + assert(workflow.context eq context) + } + + it should "reject a result without a physical plan with a clear error" in { + val context = new WorkflowContext() + // Lenient default: the unset fileName becomes a per-operator error and physicalPlan = None. + val failed = new WorkflowCompiler(context).compile( + LogicalPlanPojo(List(new CSVScanSourceOpDesc()), List(), List(), List()) + ) + assert(failed.physicalPlan.isEmpty) + + val ex = intercept[IllegalStateException] { + Workflow.fromCompilationResult(context, failed) + } + assert(ex.getMessage.contains("physicalPlan")) + } +} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala index 73f80c1b3c..0b4aabe3b1 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala @@ -29,7 +29,7 @@ import org.apache.texera.amber.core.workflow.{ import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow import org.apache.texera.amber.operator.TestOperators -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalamock.scalatest.MockFactory import org.scalatest.flatspec.AnyFlatSpec diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala index 31158ba2d6..b8f5617e13 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala @@ -39,7 +39,7 @@ import org.apache.texera.dao.MockTexeraDB import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum import org.apache.texera.dao.jooq.generated.tables.daos._ import org.apache.texera.dao.jooq.generated.tables.pojos._ -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGeneratorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGeneratorSpec.scala index e720b9c6cb..9fb912cddf 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGeneratorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGeneratorSpec.scala @@ -28,7 +28,7 @@ import org.apache.texera.amber.operator.udf.python.{ DualInputPortsPythonUDFOpDescV2, PythonUDFOpDescV2 } -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalamock.scalatest.MockFactory import org.scalatest.flatspec.AnyFlatSpec diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/resourcePolicies/ResourcePoliciesSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/resourcePolicies/ResourcePoliciesSpec.scala index 9500ed4730..70958ec059 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/resourcePolicies/ResourcePoliciesSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/resourcePolicies/ResourcePoliciesSpec.scala @@ -31,7 +31,7 @@ import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.{ } import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow import org.apache.texera.amber.operator.TestOperators -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpec class ResourcePoliciesSpec extends AnyFlatSpec { diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala index 3c81931993..12275df878 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/BatchSizePropagationSpec.scala @@ -30,7 +30,7 @@ import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow import org.apache.texera.amber.operator.TestOperators import org.apache.texera.amber.operator.aggregate.AggregationFunction -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpecLike import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala index 6e0eb6e230..b6b815a120 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/DataProcessingSpec.scala @@ -42,7 +42,7 @@ import org.apache.texera.amber.engine.e2e.TestUtils.{ } import org.apache.texera.amber.operator.TestOperators import org.apache.texera.amber.operator.aggregate.AggregationFunction -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpecLike import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala index 8143852551..ebd910dcea 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/PauseSpec.scala @@ -44,7 +44,7 @@ import org.apache.texera.amber.engine.e2e.TestUtils.{ stateReached } import org.apache.texera.amber.operator.{LogicalOp, TestOperators} -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.flatspec.AnyFlatSpecLike import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala index 6dfb23f6ac..17a7ecb671 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala @@ -35,7 +35,7 @@ import org.apache.texera.amber.engine.e2e.TestUtils.{ setUpWorkflowExecutionData } import org.apache.texera.amber.operator.{LogicalOp, TestOperators} -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Outcome, Retries} import org.scalatest.flatspec.AnyFlatSpecLike diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala index b0ebccbabd..399b351342 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala @@ -65,10 +65,11 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ WorkflowVersion, Workflow => WorkflowPojo } -import org.apache.texera.web.model.websocket.request.LogicalPlanPojo +import org.apache.texera.common.compiler.model.LogicalPlanPojo import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource.getResultUriByLogicalPortId import org.apache.texera.web.service.ExecutionResultService -import org.apache.texera.workflow.{LogicalLink, WorkflowCompiler} +import org.apache.texera.common.compiler.model.LogicalLink +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} object TestUtils { @@ -96,9 +97,13 @@ object TestUtils { val workflowCompiler = new WorkflowCompiler( context ) - workflowCompiler.compile( - LogicalPlanPojo(operators, links, List(), List()) + // Execution path: strict, fail-fast on compilation errors. Strict guarantees + // a defined physicalPlan (errors throw rather than clearing it). + val compilationResult = workflowCompiler.compile( + LogicalPlanPojo(operators, links, List(), List()), + CompilationErrorHandling.Strict ) + Workflow.fromCompilationResult(context, compilationResult) } /** diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/faulttolerance/CheckpointSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/faulttolerance/CheckpointSpec.scala index 145b1b0bc8..8eb58fc580 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/faulttolerance/CheckpointSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/faulttolerance/CheckpointSpec.scala @@ -33,7 +33,7 @@ import org.apache.texera.amber.engine.common.virtualidentity.util.{COORDINATOR, import org.apache.texera.amber.engine.common.{AmberRuntime, CheckpointState} import org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow import org.apache.texera.amber.operator.TestOperators -import org.apache.texera.workflow.LogicalLink +import org.apache.texera.common.compiler.model.LogicalLink import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpecLike diff --git a/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala index c460cfdb63..2edc102234 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala @@ -32,7 +32,8 @@ import org.apache.texera.web.model.websocket.event.{ WorkflowErrorEvent, WorkflowStateEvent } -import org.apache.texera.web.model.websocket.request.{LogicalPlanPojo, WorkflowExecuteRequest} +import org.apache.texera.common.compiler.model.LogicalPlanPojo +import org.apache.texera.web.model.websocket.request.WorkflowExecuteRequest import org.apache.texera.web.storage.ExecutionStateStore import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState import org.scalatest.flatspec.AnyFlatSpec diff --git a/amber/src/test/scala/org/apache/texera/workflow/WorkflowCompilerSpec.scala b/amber/src/test/scala/org/apache/texera/workflow/WorkflowCompilerSpec.scala deleted file mode 100644 index 3ee61d3f5c..0000000000 --- a/amber/src/test/scala/org/apache/texera/workflow/WorkflowCompilerSpec.scala +++ /dev/null @@ -1,212 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.workflow - -import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} -import org.apache.texera.amber.operator.TestOperators -import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc -import org.apache.texera.web.model.websocket.request.LogicalPlanPojo -import org.scalatest.flatspec.AnyFlatSpec - -/** - * Direct unit coverage for [[WorkflowCompiler]]. Today the compiler is only - * exercised transitively by e2e/scheduler specs through - * [[org.apache.texera.amber.engine.e2e.TestUtils.buildWorkflow]]. The cases - * below pin its contract — physical-plan shape, storage-port collection, and - * strict-mode error behavior — so future refactors (notably the planned - * merge with workflow-compiling-service's compiler) have a direct anchor. - * - * Not yet covered: the Python codegen `#EXCEPTION DURING CODE GENERATION:` - * regex branch. Triggering it requires a `PythonOperatorDescriptor` subclass - * whose `generatePythonCode()` throws; left for a follow-up so this initial - * spec stays focused on plumbing the compiler boundary itself. - */ -class WorkflowCompilerSpec extends AnyFlatSpec { - - 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) - - // -------------------- physical-plan shape -------------------- - - "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 ctx = new WorkflowContext() - - val workflow = new WorkflowCompiler(ctx).compile( - pojo( - List(csv, keyword), - List( - LogicalLink( - csv.operatorIdentifier, - PortIdentity(), - keyword.operatorIdentifier, - PortIdentity() - ) - ) - ) - ) - - assert(workflow.logicalPlan.operators.size == 2) - assert(workflow.physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).nonEmpty) - assert(workflow.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 ctx = new WorkflowContext() - - val workflow = new WorkflowCompiler(ctx).compile( - pojo( - List(csv, keyword), - List( - LogicalLink( - csv.operatorIdentifier, - PortIdentity(), - keyword.operatorIdentifier, - PortIdentity() - ) - ) - ) - ) - - val csvPhysIds = - workflow.physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).map(_.id).toSet - val keywordPhysIds = - workflow.physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).map(_.id).toSet - - val bridging = workflow.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 - // `outputPortsNeedingStorage`, which it then writes back onto the - // workflow context. These tests pin that the *mutation* lands on the - // context (not just a side value), and that both the terminal-default and - // the opsToViewResult-additive paths populate it. - - "WorkflowCompiler" should "mark the terminal op's output port as needing storage on the context" in { - val csv = TestOperators.smallCsvScanOpDesc() - val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") - val ctx = new WorkflowContext() - - new WorkflowCompiler(ctx).compile( - pojo( - List(csv, keyword), - List( - LogicalLink( - csv.operatorIdentifier, - PortIdentity(), - keyword.operatorIdentifier, - PortIdentity() - ) - ) - ) - ) - - val storage = ctx.workflowSettings.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 ctx = new WorkflowContext() - - new WorkflowCompiler(ctx).compile( - pojo( - List(csv, keyword), - List( - LogicalLink( - csv.operatorIdentifier, - PortIdentity(), - keyword.operatorIdentifier, - PortIdentity() - ) - ), - opsToViewResult = List(csv.operatorIdentifier.id) - ) - ) - - val storage = ctx.workflowSettings.outputPortsNeedingStorage - val logicalOpsInStorage = storage.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 ctx = new WorkflowContext() - - new WorkflowCompiler(ctx).compile(pojo(List(csv), List.empty)) - - val storage = ctx.workflowSettings.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 above. - "WorkflowCompiler in strict mode (no errorList)" should - "throw when a scan source has no fileName set" in { - // CSVScanSourceOpDesc defaults fileName to None; `resolveScanSourceOpFileName(None)` - // hits the "No file selected" RuntimeException thrown from `scanOp.fileName.getOrElse` - // and surfaces that exception out of `compile` because the compiler passes - // `None` for the errorList (i.e. fail-fast on the execution path). - val orphanCsv = new CSVScanSourceOpDesc() - val ctx = new WorkflowContext() - - val ex = intercept[RuntimeException] { - new WorkflowCompiler(ctx).compile(pojo(List(orphanCsv), List.empty)) - } - assert(ex.getMessage.contains("No file selected")) - } -} diff --git a/build.sbt b/build.sbt index b1ab4b6db3..720fc28b80 100644 --- a/build.sbt +++ b/build.sbt @@ -225,8 +225,12 @@ lazy val FileService = (project in file("file-service")) ) lazy val WorkflowOperator = (project in file("common/workflow-operator")).settings(commonModuleSettingsWithVendored).dependsOn(WorkflowCore) +lazy val WorkflowCompiler = (project in file("common/workflow-compiler")) + .settings(commonModuleSettings) + .configs(Test) + .dependsOn(WorkflowOperator) lazy val WorkflowCompilingService = (project in file("workflow-compiling-service")) - .dependsOn(WorkflowOperator, Auth, Config, Resource) + .dependsOn(WorkflowCompiler, Auth, Config, Resource) .settings(commonModuleSettings) .settings( dependencyOverrides ++= Seq( @@ -238,7 +242,7 @@ lazy val WorkflowCompilingService = (project in file("workflow-compiling-service ) lazy val WorkflowExecutionService = (project in file("amber")) - .dependsOn(WorkflowOperator, Auth, Config) + .dependsOn(WorkflowCompiler, Auth, Config) .settings(commonModuleSettings) .settings( dependencyOverrides ++= Seq( @@ -278,6 +282,7 @@ lazy val TexeraProject = (project in file(".")) PyBuilder, WorkflowCore, WorkflowOperator, + WorkflowCompiler, // services AccessControlService, ComputingUnitManagingService, diff --git a/common/workflow-compiler/build.sbt b/common/workflow-compiler/build.sbt new file mode 100644 index 0000000000..d8d7aea89c --- /dev/null +++ b/common/workflow-compiler/build.sbt @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +////////////////////////////////////////////////////////////////////////////// +// Compilation +////////////////////////////////////////////////////////////////////////////// + +scalacOptions += "-Ymacro-annotations" + +// Scala compiler options (mirrors the other common modules; `-Ywarn-unused:imports` +// is required by the scalafix RemoveUnused rule that CI runs via scalafixAll). +Compile / scalacOptions ++= Seq( + "-Xelide-below", + "WARNING", + "-feature", + "-deprecation", + "-Ywarn-unused:imports" +) + +////////////////////////////////////////////////////////////////////////////// +// Dependencies +////////////////////////////////////////////////////////////////////////////// + +libraryDependencies ++= Seq( + "org.scalatest" %% "scalatest" % "3.2.15" % Test +) + +// Arrow 19's transitive deps (via WorkflowOperator -> WorkflowCore) pull +// jackson-databind past the 2.18 line that jackson-module-scala is pinned to; +// force the Jackson core family back so the Scala module can initialize in +// tests (else Test aborts with "Scala module 2.18.x requires Jackson Databind +// version >= 2.18.0 and < 2.19.0"). Mirrors common/workflow-core/build.sbt. +val jacksonVersion = "2.18.8" +dependencyOverrides ++= Seq( + "com.fasterxml.jackson.core" % "jackson-core" % jacksonVersion, + "com.fasterxml.jackson.core" % "jackson-databind" % jacksonVersion, + "com.fasterxml.jackson.core" % "jackson-annotations" % jacksonVersion +) diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala similarity index 54% copy from amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala copy to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala index e15b441fca..0c9217ffc2 100644 --- a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/EditingTimeCompilationRequest.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala @@ -17,19 +17,19 @@ * under the License. */ -package org.apache.texera.web.model.websocket.request +package org.apache.texera.common.compiler -import org.apache.texera.amber.operator.LogicalOp -import org.apache.texera.workflow.LogicalLink +/** + * 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 — during file resolution, + * logical-to-physical expansion, and schema propagation. Used before execution. + */ +sealed trait CompilationErrorHandling -case class EditingTimeCompilationRequest( - operators: List[LogicalOp], - links: List[LogicalLink], - opsToViewResult: List[String], - opsToReuseResult: List[String] -) extends TexeraWebSocketRequest { - - def toLogicalPlanPojo: LogicalPlanPojo = { - LogicalPlanPojo(operators, links, opsToViewResult, opsToReuseResult) - } +object CompilationErrorHandling { + case object Lenient extends CompilationErrorHandling + case object Strict extends CompilationErrorHandling } diff --git a/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala new file mode 100644 index 0000000000..65d13e4179 --- /dev/null +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala @@ -0,0 +1,284 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.common.compiler + +import com.google.protobuf.timestamp.Timestamp +import com.typesafe.scalalogging.{LazyLogging, Logger} +import org.apache.texera.common.compiler.WorkflowCompiler.{ + collectOutputSchemaFromPhysicalPlan, + convertErrorListToWorkflowFatalErrorMap +} +import org.apache.texera.common.compiler.model.{LogicalPlan, LogicalPlanPojo} +import org.apache.texera.amber.core.tuple.Schema +import org.apache.texera.amber.core.virtualidentity.OperatorIdentity +import org.apache.texera.amber.core.workflow.{ + GlobalPortIdentity, + PhysicalLink, + PhysicalPlan, + PortIdentity, + WorkflowContext +} +import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.COMPILATION_ERROR +import org.apache.texera.amber.core.workflowruntimestate.WorkflowFatalError +import org.apache.texera.amber.util.StackTraceUtils.getStackTraceWithAllCauses + +import java.time.Instant +import scala.collection.mutable +import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters.IteratorHasAsScala +import scala.util.{Failure, Success, Try} + +object WorkflowCompiler { + // util function to convert the error list to an error map, and report the errors in the log + private def convertErrorListToWorkflowFatalErrorMap( + logger: Logger, + errorList: List[(OperatorIdentity, Throwable)] + ): Map[OperatorIdentity, WorkflowFatalError] = { + val opIdToError = mutable.Map[OperatorIdentity, WorkflowFatalError]() + errorList.foreach { + case (opId, err) => + // map each error to WorkflowFatalError, and report them in the log + logger.error(s"Error occurred in logical plan compilation for opId: $opId", err) + // keep only the first error per operator: it is the root cause (e.g. a file + // resolution failure), while later stages re-fail on the same operator with + // less specific messages (e.g. schema propagation seeing an unresolved file) + if (!opIdToError.contains(opId)) { + opIdToError += (opId -> WorkflowFatalError( + COMPILATION_ERROR, + Timestamp(Instant.now), + err.toString, + getStackTraceWithAllCauses(err), + opId.id + )) + } + } + opIdToError.toMap + } + + private def collectOutputSchemaFromPhysicalPlan( + physicalPlan: PhysicalPlan, + errorList: ArrayBuffer[(OperatorIdentity, Throwable)] + ): Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]] = { + + // Collect output schemas per physical operator + val physicalOutputSchemas = + physicalPlan.operators.map { physicalOp => + val portSchemas = physicalOp.outputPorts.values + .filterNot(_._1.id.internal) + .map { + case (port, _, schema) => + schema match { + case Left(err) => + errorList.append((physicalOp.id.logicalOpId, err)) + port.id -> None + case Right(validSchema) => + port.id -> Some(validSchema) + } + } + .toMap + physicalOp.id -> portSchemas + } + + // Group by logical operator ID and merge port schemas + physicalOutputSchemas + .groupBy(_._1.logicalOpId) + .view + .mapValues { list => + list.flatMap(_._2).toMap + } + .toMap + } + +} + +case class WorkflowCompilationResult( + logicalPlan: LogicalPlan, + physicalPlan: Option[PhysicalPlan], // if physicalPlan is None, compilation failed + operatorIdToOutputSchemas: Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]], + operatorIdToError: Map[OperatorIdentity, WorkflowFatalError], + outputPortsNeedingStorage: Set[GlobalPortIdentity] +) + +/** + * The single workflow compiler shared in-process by both compilation call sites: + * workflow-compiling-service (editing path, [[CompilationErrorHandling.Lenient]] — accumulate + * per-operator errors so the UI can render them, `physicalPlan = None` when any exist) and + * amber (execution path, [[CompilationErrorHandling.Strict]] — fail fast before a run). + * + * This module depends only on WorkflowOperator, so neither service leaks its HTTP stack into + * the other; the engine `Workflow` wrapper stays in amber as a thin adapter over + * [[WorkflowCompilationResult]]. `outputPortsNeedingStorage` is always computed — the + * editing-path caller simply ignores it — keeping both paths on one code path. + */ +class WorkflowCompiler( + context: WorkflowContext +) extends LazyLogging { + + /** + * Expands the logical plan to a 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, 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 => + Try { + val logicalOp = logicalPlan.getOperator(logicalOpId) + val upstreamLinks = logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier) + + val subPlan = logicalOp.getPhysicalPlan(context.workflowId, context.executionId) + subPlan + .topologicalIterator() + .map(subPlan.getOperator) + .foreach({ physicalOp => + { + val externalLinks = upstreamLinks + .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) + + // 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) } + + // **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") + + 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 + ) + } + } + } match { + case Success(_) => + + case Failure(err) => + errorList match { + case Some(list) => list.append((logicalOpId, err)) + case None => throw err + } + } + ) + (physicalPlan, outputPortsNeedingStorage.toSet) + } + + /** + * Compiles a workflow to a physical plan, along with the schema propagation result and + * errors (if any). + * + * @param logicalPlanPojo the POJO parsed from the workflow string provided by the user + * @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, + errorHandling: CompilationErrorHandling = CompilationErrorHandling.Lenient + ): WorkflowCompilationResult = { + // 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(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 an error is encountered during logical => physical expansion, we still want to + // collect the output schemas of the remaining no-error operators. In Lenient mode + // schema-propagation failures land in + // the shared buffer alongside the other errors; in Strict mode they must fail fast too (e.g. a + // Projection on a missing column would otherwise be launched and only fail at runtime). + val schemaErrorList = errorList.getOrElse(new ArrayBuffer[(OperatorIdentity, Throwable)]()) + val opIdToOutputSchema = collectOutputSchemaFromPhysicalPlan(physicalPlan, schemaErrorList) + if (errorHandling == CompilationErrorHandling.Strict && schemaErrorList.nonEmpty) { + throw schemaErrorList.head._2 + } + + val hasErrors = errorList.exists(_.nonEmpty) + WorkflowCompilationResult( + logicalPlan = logicalPlan, + physicalPlan = if (hasErrors) None else Some(physicalPlan), + operatorIdToOutputSchemas = opIdToOutputSchema, + // map each error from OpId to WorkflowFatalError, and report them via logger + operatorIdToError = convertErrorListToWorkflowFatalErrorMap( + logger, + errorList.map(_.toList).getOrElse(List.empty) + ), + outputPortsNeedingStorage = outputPortsNeedingStorage + ) + } +} diff --git a/amber/src/main/scala/org/apache/texera/workflow/LogicalLink.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalLink.scala similarity index 97% rename from amber/src/main/scala/org/apache/texera/workflow/LogicalLink.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalLink.scala index e6553e3cdf..30d5d3f3ab 100644 --- a/amber/src/main/scala/org/apache/texera/workflow/LogicalLink.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalLink.scala @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.texera.workflow +package org.apache.texera.common.compiler.model import com.fasterxml.jackson.annotation.{JsonCreator, JsonProperty} import org.apache.texera.amber.core.virtualidentity.OperatorIdentity diff --git a/amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala similarity index 92% rename from amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala index 6e759439a0..f4d694ff66 100644 --- a/amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala @@ -17,14 +17,13 @@ * under the License. */ -package org.apache.texera.workflow +package org.apache.texera.common.compiler.model import com.typesafe.scalalogging.LazyLogging import org.apache.texera.amber.core.storage.FileResolver import org.apache.texera.amber.core.virtualidentity.OperatorIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc -import org.apache.texera.web.model.websocket.request.LogicalPlanPojo import org.jgrapht.graph.DirectedAcyclicGraph import org.jgrapht.util.SupplierUtil @@ -88,9 +87,11 @@ case class LogicalPlan( } /** - * Resolve all user-given filename for the scan source operators to URIs, and call op.setFileUri to set the URi + * Resolves each scan source operator's user-given file name to a URI and sets it on the + * operator via `setResolvedFileName`. * - * @param errorList if given, put errors during resolving to it + * @param errorList if given, errors encountered during resolution are appended to it; + * otherwise the first error is thrown */ def resolveScanSourceOpFileName( errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlanPojo.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlanPojo.scala similarity index 95% rename from workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlanPojo.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlanPojo.scala index 151ee863fa..efda9ceda7 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlanPojo.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlanPojo.scala @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.texera.amber.compiler.model +package org.apache.texera.common.compiler.model import org.apache.texera.amber.operator.LogicalOp diff --git a/amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/LogicalPlanSpec.scala similarity index 99% rename from amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala rename to common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/LogicalPlanSpec.scala index cbc28388dd..6e926d4715 100644 --- a/amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala +++ b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/LogicalPlanSpec.scala @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.texera.workflow +package org.apache.texera.common.compiler import org.apache.commons.vfs2.FileNotFoundException import org.apache.texera.amber.core.virtualidentity.OperatorIdentity @@ -32,7 +32,7 @@ import org.apache.texera.amber.operator.udf.python.DualInputPortsPythonUDFOpDesc import org.apache.texera.amber.operator.udf.python.source.PythonUDFSourceOpDescV2 import org.apache.texera.amber.operator.union.UnionOpDesc import org.apache.texera.amber.operator.{LogicalOp, TestOperators} -import org.apache.texera.web.model.websocket.request.LogicalPlanPojo +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan, LogicalPlanPojo} import org.scalatest.flatspec.AnyFlatSpec import scala.collection.mutable.ArrayBuffer diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala similarity index 57% rename from workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala rename to common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala index 79df9e98ee..94187373d7 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/compiler/WorkflowCompilerSpec.scala +++ b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/WorkflowCompilerSpec.scala @@ -17,9 +17,9 @@ * under the License. */ -package org.apache.texera.amber.compiler +package org.apache.texera.common.compiler -import org.apache.texera.amber.compiler.model.{LogicalLink, LogicalPlanPojo} +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} import org.apache.texera.amber.core.tuple.{Attribute, AttributeType} import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} @@ -32,16 +32,18 @@ import org.apache.texera.amber.operator.filter.{ import org.apache.texera.amber.operator.limit.LimitOpDesc import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.amber.operator.TestOperators import org.scalatest.flatspec.AnyFlatSpec /** - * Direct unit coverage for the editing-time [[WorkflowCompiler]]. + * Direct unit coverage for the unified [[WorkflowCompiler]]. * - * Owns *compiler-behavior* tests — schema propagation through multi-op - * chains, lenient-mode error accumulation, terminal-storage selection. - * `WorkflowCompilationResourceSpec` owns *resource-layer* tests — HTTP - * status, response type discriminator, JSON envelope. Drawing the line - * here keeps each spec focused on a single failure axis. + * Owns *compiler-behavior* tests across both paths: lenient (editing-time — + * accumulate per-operator errors, schema propagation) and strict + * (pre-execution — fail-fast), plus physical-plan shape and the set of + * output ports needing storage. `WorkflowCompilationResourceSpec` owns + * *resource-layer* tests — HTTP status, response type discriminator, JSON + * envelope. Drawing the line here keeps each spec focused. * * Bypassing the resource layer also sidesteps a separate NPE in response * serialization (apache/texera#5021); these compiler-level tests stay @@ -314,4 +316,226 @@ 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 { + // Strict passes no error buffer, so `resolveScanSourceOpFileName` rethrows + // the first failure instead of accumulating it (the execution path's + // fail-fast contract). The lenient counterpart above accumulates the same + // failure without throwing. + val orphanCsv = new CSVScanSourceOpDesc() + + val ex = intercept[RuntimeException] { + new WorkflowCompiler(newContext()) + .compile(pojo(List(orphanCsv), List.empty), CompilationErrorHandling.Strict) + } + assert(ex.getMessage.contains("No file selected")) + } + + it should "return a defined physicalPlan for a well-formed plan" in { + // The execution path calls `physicalPlan.get` on the result, so a strict + // success must always carry a plan. + val csv = csvOp(realCsvPath) + val proj = projectOp(List("Region", "Total Profit")) + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, proj), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + proj.operatorIdentifier, + PortIdentity(0) + ) + ) + ), + CompilationErrorHandling.Strict + ) + + assert(result.physicalPlan.isDefined, "strict success must yield a physical plan") + assert(result.operatorIdToError.isEmpty) + assert(result.outputPortsNeedingStorage.nonEmpty, "terminal ports still collected in strict") + } + + it should "throw on schema-propagation errors" in { + // A projection on a missing column fails schema *propagation*, not plan + // expansion: `propagateSchema` stores a Left on the output port instead of + // throwing, so this error only becomes visible when output schemas are + // collected. Strict must fail fast on it too — otherwise the plan would be + // launched and only fail at runtime. The lenient counterpart above turns + // the same failure into a per-operator error instead. + val csv = csvOp(realCsvPath) + val badProjection = projectOp(List("DoesNotExist")) + + val ex = intercept[Throwable] { + new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, badProjection), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(0), + badProjection.operatorIdentifier, + PortIdentity(0) + ) + ) + ), + CompilationErrorHandling.Strict + ) + } + assert( + ex.getMessage != null && ex.getMessage.contains("DoesNotExist"), + s"the thrown schema error should name the missing attribute, got: $ex" + ) + } } diff --git a/amber/src/test/scala/org/apache/texera/workflow/LogicalLinkSpec.scala b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/model/LogicalLinkSpec.scala similarity index 99% rename from amber/src/test/scala/org/apache/texera/workflow/LogicalLinkSpec.scala rename to common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/model/LogicalLinkSpec.scala index bd56aa7d5f..6ffb9d497d 100644 --- a/amber/src/test/scala/org/apache/texera/workflow/LogicalLinkSpec.scala +++ b/common/workflow-compiler/src/test/scala/org/apache/texera/common/compiler/model/LogicalLinkSpec.scala @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.texera.workflow +package org.apache.texera.common.compiler.model import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.exc.{MismatchedInputException, ValueInstantiationException} diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/StackTraceUtils.scala similarity index 53% rename from workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala rename to common/workflow-core/src/main/scala/org/apache/texera/amber/util/StackTraceUtils.scala index 5c7662f966..adade6a2e5 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/StackTraceUtils.scala @@ -17,25 +17,26 @@ * under the License. */ -package org.apache.texera.amber.compiler.model +package org.apache.texera.amber.util -import com.fasterxml.jackson.annotation.{JsonCreator, JsonProperty} -import org.apache.texera.amber.core.virtualidentity.OperatorIdentity -import org.apache.texera.amber.core.workflow.PortIdentity +object StackTraceUtils { -case class LogicalLink( - @JsonProperty("fromOpId") fromOpId: OperatorIdentity, - fromPortId: PortIdentity, - @JsonProperty("toOpId") toOpId: OperatorIdentity, - toPortId: PortIdentity -) { - @JsonCreator - def this( - @JsonProperty("fromOpId") fromOpId: String, - fromPortId: PortIdentity, - @JsonProperty("toOpId") toOpId: String, - toPortId: PortIdentity - ) = { - this(OperatorIdentity(fromOpId), fromPortId, OperatorIdentity(toOpId), toPortId) + /** + * Renders a throwable and its full cause chain as a single string for + * user-facing error details. Shared by amber's ErrorUtils and the + * workflow compiler. + */ + def getStackTraceWithAllCauses(err: Throwable, topLevel: Boolean = true): String = { + val header = if (topLevel) { + "Stack trace for developers: \n\n" + } else { + "\n\nCaused by:\n" + } + val message = header + err.toString + "\n" + err.getStackTrace.mkString("\n") + if (err.getCause != null) { + message + getStackTraceWithAllCauses(err.getCause, topLevel = false) + } else { + message + } } } diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala deleted file mode 100644 index 666ddb70a8..0000000000 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala +++ /dev/null @@ -1,231 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.amber.compiler - -import com.google.protobuf.timestamp.Timestamp -import com.typesafe.scalalogging.{LazyLogging, Logger} -import org.apache.texera.amber.compiler.WorkflowCompiler.{ - collectOutputSchemaFromPhysicalPlan, - convertErrorListToWorkflowFatalErrorMap -} -import org.apache.texera.amber.compiler.model.{LogicalPlan, LogicalPlanPojo} -import org.apache.texera.amber.core.tuple.Schema -import org.apache.texera.amber.core.virtualidentity.OperatorIdentity -import org.apache.texera.amber.core.workflow.{ - PhysicalLink, - PhysicalPlan, - PortIdentity, - WorkflowContext -} -import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.COMPILATION_ERROR -import org.apache.texera.amber.core.workflowruntimestate.WorkflowFatalError - -import java.time.Instant -import scala.collection.mutable -import scala.collection.mutable.ArrayBuffer -import scala.jdk.CollectionConverters.IteratorHasAsScala - -object WorkflowCompiler { - // util function for extracting the error causes - private def getStackTraceWithAllCauses(err: Throwable, topLevel: Boolean = true): String = { - val header = if (topLevel) { - "Stack trace for developers: \n\n" - } else { - "\n\nCaused by:\n" - } - val message = header + err.toString + "\n" + err.getStackTrace.mkString("\n") - if (err.getCause != null) { - message + getStackTraceWithAllCauses(err.getCause, topLevel = false) - } else { - message - } - } - - // util function for convert the error list to error map, and report the error in log - private def convertErrorListToWorkflowFatalErrorMap( - logger: Logger, - errorList: List[(OperatorIdentity, Throwable)] - ): Map[OperatorIdentity, WorkflowFatalError] = { - val opIdToError = mutable.Map[OperatorIdentity, WorkflowFatalError]() - errorList.foreach { - case (opId, err) => - // map each error to WorkflowFatalError, and report them in the log - logger.error(s"Error occurred in logical plan compilation for opId: $opId", err) - // keep only the first error per operator: it is the root cause (e.g. a file - // resolution failure), while later stages re-fail on the same operator with - // less specific messages (e.g. schema propagation seeing an unresolved file) - if (!opIdToError.contains(opId)) { - opIdToError += (opId -> WorkflowFatalError( - COMPILATION_ERROR, - Timestamp(Instant.now), - err.toString, - getStackTraceWithAllCauses(err), - opId.id - )) - } - } - opIdToError.toMap - } - - private def collectOutputSchemaFromPhysicalPlan( - physicalPlan: PhysicalPlan, - errorList: ArrayBuffer[(OperatorIdentity, Throwable)] - ): Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]] = { - - // Collect output schemas per physical operator - val physicalOutputSchemas = - physicalPlan.operators.map { physicalOp => - val portSchemas = physicalOp.outputPorts.values - .filterNot(_._1.id.internal) - .map { - case (port, _, schema) => - schema match { - case Left(err) => - errorList.append((physicalOp.id.logicalOpId, err)) - port.id -> None - case Right(validSchema) => - port.id -> Some(validSchema) - } - } - .toMap - physicalOp.id -> portSchemas - } - - // Group by logical operator ID and merge port schemas - physicalOutputSchemas - .groupBy(_._1.logicalOpId) - .view - .mapValues { list => - list.flatMap(_._2).toMap - } - .toMap - } - -} - -case class WorkflowCompilationResult( - physicalPlan: Option[PhysicalPlan], // if physical plan is none, the compilation is failed - operatorIdToOutputSchemas: Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]], - operatorIdToError: Map[OperatorIdentity, WorkflowFatalError] -) - -class WorkflowCompiler( - context: WorkflowContext -) extends LazyLogging { - - // function to expand logical plan to physical plan - private def expandLogicalPlan( - logicalPlan: LogicalPlan, - errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] - ): PhysicalPlan = { - var physicalPlan = PhysicalPlan(operators = Set.empty, links = Set.empty) - - logicalPlan.getTopologicalOpIds.asScala.foreach { logicalOpId => - val logicalOp = logicalPlan.getOperator(logicalOpId) - val allUpstreamLinks = logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier) - - 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) - ) - } - - val internalLinks = subPlan.getUpstreamPhysicalLinks(physicalOp.id) - - // 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) - } - - // **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") - - 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 - } - } - } - } - } catch { - case e: Throwable => - errorList match { - case Some(list) => list.append((logicalOpId, e)) // Store error - case None => throw e // Throw if no list is provided - } - } - } - - physicalPlan - } - - /** - * 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 - */ - def compile( - logicalPlanPojo: LogicalPlanPojo - ): WorkflowCompilationResult = { - val errorList = new ArrayBuffer[(OperatorIdentity, Throwable)]() - var opIdToOutputSchema: Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]] = Map() - // 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)) - - // 3. expand the logical plan to the physical plan - val physicalPlan = expandLogicalPlan(logicalPlan, Some(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) - WorkflowCompilationResult( - physicalPlan = if (errorList.nonEmpty) None else Some(physicalPlan), - operatorIdToOutputSchemas = opIdToOutputSchema, - // map each error from OpId to WorkflowFatalError, and report them via logger - operatorIdToError = convertErrorListToWorkflowFatalErrorMap(logger, errorList.toList) - ) - } -} diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala deleted file mode 100644 index d7fb25679b..0000000000 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala +++ /dev/null @@ -1,138 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.amber.compiler.model - -import com.typesafe.scalalogging.LazyLogging -import org.apache.texera.amber.core.storage.FileResolver -import org.apache.texera.amber.core.virtualidentity.OperatorIdentity -import org.apache.texera.amber.core.workflow.PortIdentity -import org.apache.texera.amber.operator.LogicalOp -import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc -import org.jgrapht.graph.DirectedAcyclicGraph -import org.jgrapht.util.SupplierUtil - -import java.util -import scala.collection.mutable.ArrayBuffer -import scala.util.{Failure, Success, Try} - -object LogicalPlan { - - private def toJgraphtDAG( - operatorList: List[LogicalOp], - links: List[LogicalLink] - ): DirectedAcyclicGraph[OperatorIdentity, LogicalLink] = { - val workflowDag = - new DirectedAcyclicGraph[OperatorIdentity, LogicalLink]( - null, // vertexSupplier - SupplierUtil.createSupplier(classOf[LogicalLink]), // edgeSupplier - false, // weighted - true // allowMultipleEdges - ) - operatorList.foreach(op => workflowDag.addVertex(op.operatorIdentifier)) - links.foreach(l => - workflowDag.addEdge( - l.fromOpId, - l.toOpId, - l - ) - ) - workflowDag - } - - def apply( - pojo: LogicalPlanPojo - ): LogicalPlan = { - LogicalPlan(pojo.operators, pojo.links) - } - -} - -case class LogicalPlan( - operators: List[LogicalOp], - links: List[LogicalLink] -) extends LazyLogging { - - private lazy val operatorMap: Map[OperatorIdentity, LogicalOp] = - operators.map(op => (op.operatorIdentifier, op)).toMap - - private lazy val jgraphtDag: DirectedAcyclicGraph[OperatorIdentity, LogicalLink] = - LogicalPlan.toJgraphtDAG(operators, links) - - def getTopologicalOpIds: util.Iterator[OperatorIdentity] = jgraphtDag.iterator() - - def getOperator(opId: String): LogicalOp = operatorMap(OperatorIdentity(opId)) - - def getOperator(opId: OperatorIdentity): LogicalOp = operatorMap(opId) - - def addOperator(op: LogicalOp): LogicalPlan = { - // TODO: fix schema for the new operator - this.copy(operators :+ op, links) - } - - def addLink( - fromOpId: OperatorIdentity, - fromPortId: PortIdentity, - toOpId: OperatorIdentity, - toPortId: PortIdentity - ): LogicalPlan = { - val newLink = LogicalLink( - fromOpId, - fromPortId, - toOpId, - toPortId - ) - val newLinks = links :+ newLink - this.copy(operators, newLinks) - } - - def getUpstreamLinks(opId: OperatorIdentity): List[LogicalLink] = { - links.filter(l => l.toOpId == opId) - } - - /** - * Resolve all user-given filename for the scan source operators to URIs, and call op.setFileUri to set the URi - * @param errorList if given, put errors during resolving to it - */ - def resolveScanSourceOpFileName( - errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] - ): Unit = { - operators.foreach { - case operator @ (scanOp: ScanSourceOpDesc) => - Try { - // Resolve file path for ScanSourceOpDesc - val fileName = scanOp.fileName.getOrElse( - throw new RuntimeException( - "No file selected. Please select a file from the 'File' dropdown in the right panel." - ) - ) - val fileUri = FileResolver.resolve(fileName) // Convert to URI - - // Set the URI in the ScanSourceOpDesc - scanOp.setResolvedFileName(fileUri) - } match { - case Success(_) => // Successfully resolved and set the file URI - case Failure(err) => - logger.error("Error resolving file path for ScanSourceOpDesc", err) - errorList.foreach(_.append((operator.operatorIdentifier, err))) - } - case _ => // Skip non-ScanSourceOpDesc operators - } - } -} diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala index 501498b1d5..ae12cd085b 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala +++ b/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala @@ -24,8 +24,8 @@ import com.typesafe.scalalogging.LazyLogging import jakarta.annotation.security.RolesAllowed import jakarta.ws.rs.core.MediaType import jakarta.ws.rs.{Consumes, POST, Path, Produces} -import org.apache.texera.amber.compiler.WorkflowCompiler -import org.apache.texera.amber.compiler.model.LogicalPlanPojo +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} +import org.apache.texera.common.compiler.model.LogicalPlanPojo import org.apache.texera.amber.core.tuple.Attribute import org.apache.texera.amber.core.virtualidentity.WorkflowIdentity import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext} @@ -68,8 +68,10 @@ class WorkflowCompilationResource extends LazyLogging { // a placeholder workflow context, as compiling a workflow doesn't require a wid from the frontend val context = new WorkflowContext(workflowId = WorkflowIdentity(0)) - // Compile the pojo using WorkflowCompiler - val compilationResult = new WorkflowCompiler(context).compile(logicalPlanPojo) + // Compile the pojo using WorkflowCompiler; the editing path must never + // throw, so pass Lenient explicitly (mirrors amber passing Strict) + val compilationResult = + new WorkflowCompiler(context).compile(logicalPlanPojo, CompilationErrorHandling.Lenient) val operatorOutputSchemas = compilationResult.operatorIdToOutputSchemas.map { case (operatorIdentity, schemas) => diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala index fd99c3d27d..89a36bdc68 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowCompilationResourceSpec.scala @@ -23,7 +23,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode import io.dropwizard.testing.junit5.ResourceExtension import jakarta.ws.rs.client.Entity import jakarta.ws.rs.core.{MediaType, Response} -import org.apache.texera.amber.compiler.model.{LogicalLink, LogicalPlanPojo} +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc @@ -40,7 +40,7 @@ import org.scalatest.flatspec.AnyFlatSpec * * All compiler-behavior assertions (schema propagation, lenient-mode * error accumulation, multi-op chains) live in - * `org.apache.texera.amber.compiler.WorkflowCompilerSpec` so a regression + * `org.apache.texera.common.compiler.WorkflowCompilerSpec` so a regression * lands on the right spec. */ class WorkflowCompilationResourceSpec extends AnyFlatSpec with BeforeAndAfterAll {
