diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 449556bab4a..9e83e258c61 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -289,6 +289,7 @@ jobs: "PyBuilder/jacoco" \ "WorkflowCore/jacoco" \ "WorkflowOperator/jacoco" \ + "WorkflowCompiler/jacoco" \ "WorkflowExecutionService/jacoco" - name: Upload amber and common coverage to Codecov if: always() 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 6ec92deab83..b6f4a0d31d4 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,7 +20,7 @@ package org.apache.texera.amber.engine.architecture.coordinator import org.apache.texera.amber.core.workflow.{PhysicalPlan, WorkflowContext} -import org.apache.texera.workflow.LogicalPlan +import org.apache.texera.common.compiler.model.LogicalPlan case class Workflow( context: WorkflowContext, 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 e15b441fcae..f7a585bb155 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 a346a1ec0a4..0059af1c1a6 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 3c9da603c86..afcd7f63a1a 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 43b2288b39f..40c1ae5a326 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,8 +110,16 @@ class WorkflowExecutionService( def executeWorkflow(): Unit = { try { - workflow = new WorkflowCompiler(workflowContext) - .compile(request.logicalPlan) + val compilationResult = new WorkflowCompiler(workflowContext) + .compile(request.logicalPlan, CompilationErrorHandling.Strict) + workflowContext.workflowSettings = workflowContext.workflowSettings.copy( + outputPortsNeedingStorage = compilationResult.outputPortsNeedingStorage + ) + workflow = Workflow( + workflowContext, + compilationResult.logicalPlan, + compilationResult.physicalPlan.get + ) } catch { case err: Throwable => errorHandler(err) 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 46e738f35bf..51d5b5677de 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/LogicalPlan.scala b/amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala deleted file mode 100644 index 974d17f40a4..00000000000 --- a/amber/src/main/scala/org/apache/texera/workflow/LogicalPlan.scala +++ /dev/null @@ -1,124 +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.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 - -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: OperatorIdentity): LogicalOp = operatorMap(opId) - - def getTerminalOperatorIds: List[OperatorIdentity] = - operatorMap.keys - .filter(op => jgraphtDag.outDegreeOf(op) == 0) - .toList - - 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 input file name")) - 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 match { - case Some(errList) => - errList.append((operator.operatorIdentifier, err)) - case None => - // Throw the error if no errorList is provided - throw err - } - } - - case _ => // Skip non-ScanSourceOpDesc operators - } - } -} 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 ae576c4053f..00000000000 --- 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/MultiRegionWorkflowIntegrationSpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/MultiRegionWorkflowIntegrationSpec.scala index 23a7e5c4eea..83e942a3002 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 3be29cebe57..6ffd8b356be 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 ae4ce991d98..8cf601e2d62 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/scheduling/CostBasedScheduleGeneratorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGeneratorSpec.scala index c9c2843acdf..f2f187d7d8d 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 31158ba2d61..b8f5617e132 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 e720b9c6cb5..9fb912cddf4 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 9500ed47303..70958ec059d 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 3c819319934..12275df8789 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 6e0eb6e2305..b6b815a120c 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 81438525510..ebd910dcea8 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 6dfb23f6ac5..17a7ecb6715 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 b0ebccbabdb..295a5707594 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,20 @@ 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 ) + // Write the storage ports back onto the context so the schedule generators + // materialize terminal results — the unified compiler returns them instead + // of mutating the context, so the execution path must copy them over (as + // WorkflowExecutionService does in production). + context.workflowSettings = context.workflowSettings.copy( + outputPortsNeedingStorage = compilationResult.outputPortsNeedingStorage + ) + Workflow(context, compilationResult.logicalPlan, compilationResult.physicalPlan.get) } /** 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 145b1b0bc8b..8eb58fc580a 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 c460cfdb63d..2edc102234c 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 94a8ffce552..00000000000 --- 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 `scanOp.fileName.getOrElse(throw new RuntimeException("no input file name"))` - // 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 == "no input file name") - } -} diff --git a/build.sbt b/build.sbt index 6d7674fbd24..b8c48edfa98 100644 --- a/build.sbt +++ b/build.sbt @@ -143,8 +143,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( @@ -156,7 +160,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( @@ -208,6 +212,7 @@ lazy val TexeraProject = (project in file(".")) PyBuilder, WorkflowCore, WorkflowOperator, + WorkflowCompiler, // services AccessControlService, ComputingUnitManagingService, diff --git a/common/workflow-compiler/DESIGN.md b/common/workflow-compiler/DESIGN.md new file mode 100644 index 00000000000..38636a7b4b3 --- /dev/null +++ b/common/workflow-compiler/DESIGN.md @@ -0,0 +1,61 @@ +# Design: Unified `WorkflowCompiler` + +## Why + +Two independent `WorkflowCompiler` copies existed — one in amber (execution path) and one in +workflow-compiling-service (editing path) — with near-identical logical→physical expansion that +drifted and had to be hand-synced. `LogicalPlan`, `LogicalPlanPojo`, and `LogicalLink` were +duplicated too. Their only real differences were the error mode (collect vs. throw), amber's +extra `outputPortsNeedingStorage`, and amber returning the engine `Workflow`. None justified two +codebases. + +## Approach + +One compiler source, reused **in-process** by both callers — no service-to-service RPC, no +serializing physical plans across a process boundary. The duplication is solved at the code +level, not by extracting a runtime compilation service (considered and rejected). + +``` +common/workflow-compiler .dependsOn(WorkflowOperator) + │ WorkflowCompiler + WorkflowCompilationResult + │ LogicalPlan / LogicalPlanPojo / LogicalLink (package org.apache.texera.common.compiler) + │ + ┌────┴─────┐ + ▼ ▼ +compiling- amber (execution) +service thin adapter → engine Workflow +(Lenient) (Strict) +``` + +The module depends only on `WorkflowOperator` (Dropwizard/HTTP never leak into it), so there is +no service-to-service dependency. Source lives under +[`src/main/scala/org/apache/texera/common/compiler/`](src/main/scala/org/apache/texera/common/compiler/): +[`WorkflowCompiler.scala`](src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala), +[`CompilationErrorHandling.scala`](src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala), +and [`model/`](src/main/scala/org/apache/texera/common/compiler/model/). + +## Key decisions + +- **One strict/lenient API, not two entry points.** `compile(pojo, errorHandling)` takes + `Lenient` (editing — accumulate per-operator errors, `physicalPlan = None` when any exist) or + `Strict` (execution — fail-fast). The internal `expandLogicalPlan` already parameterized this + via `Option[errorList]`; the mode just names it. See `compile` in `WorkflowCompiler.scala`. + +- **`outputPortsNeedingStorage` is always computed and returned**, never gated behind the mode. + Cost is one extra topological pass; the compiling-service caller simply ignores the field. + This keeps both paths on a single code path rather than branching the compiler. + +- **The engine `Workflow` stays in amber.** + `…engine.architecture.coordinator.Workflow` cannot descend into a `common/*` module, so amber + keeps a thin adapter that wraps the result into a `Workflow` (Strict guarantees a defined + `physicalPlan`). Call sites: `SyncExecutionResource`, `WorkflowExecutionService`, `TestUtils`. + +- **Package unified** to `org.apache.texera.common.compiler`, dropping the misleading + `org.apache.texera.amber.compiler`. + +## Tests + +Compiler and model unit specs live with the code under +[`src/test/scala/org/apache/texera/common/compiler/`](src/test/scala/org/apache/texera/common/compiler/): +`WorkflowCompilerSpec` (both modes, physical-plan shape, storage-port collection, schema +propagation), `LogicalPlanSpec`, and `model/LogicalLinkSpec`. diff --git a/common/workflow-compiler/build.sbt b/common/workflow-compiler/build.sbt new file mode 100644 index 00000000000..fb802f5d3a0 --- /dev/null +++ b/common/workflow-compiler/build.sbt @@ -0,0 +1,42 @@ +/* + * 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 +) diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala similarity index 53% rename from workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala index 5c7662f9668..45a17db90cb 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalLink.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/CompilationErrorHandling.scala @@ -17,25 +17,18 @@ * under the License. */ -package org.apache.texera.amber.compiler.model +package org.apache.texera.common.compiler -import com.fasterxml.jackson.annotation.{JsonCreator, JsonProperty} -import org.apache.texera.amber.core.virtualidentity.OperatorIdentity -import org.apache.texera.amber.core.workflow.PortIdentity +/** + * Controls how the compiler reacts to a per-operator compilation error. + * + * - [[CompilationErrorHandling.Lenient]] collects every error and returns them in the result, + * with `physicalPlan = None` when any error occurred. Used for editing-time validation. + * - [[CompilationErrorHandling.Strict]] throws on the first error. Used before execution. + */ +sealed trait CompilationErrorHandling -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) - } +object CompilationErrorHandling { + case object Lenient extends CompilationErrorHandling + case object Strict extends CompilationErrorHandling } diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala similarity index 51% rename from workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala index 25166e7ac52..8914fa79695 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/WorkflowCompiler.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/WorkflowCompiler.scala @@ -17,18 +17,19 @@ * under the License. */ -package org.apache.texera.amber.compiler +package org.apache.texera.common.compiler import com.google.protobuf.timestamp.Timestamp import com.typesafe.scalalogging.{LazyLogging, Logger} -import org.apache.texera.amber.compiler.WorkflowCompiler.{ +import org.apache.texera.common.compiler.WorkflowCompiler.{ collectOutputSchemaFromPhysicalPlan, convertErrorListToWorkflowFatalErrorMap } -import org.apache.texera.amber.compiler.model.{LogicalPlan, LogicalPlanPojo} +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, @@ -41,6 +42,7 @@ 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 for extracting the error causes @@ -116,111 +118,159 @@ object WorkflowCompiler { } case class WorkflowCompilationResult( + logicalPlan: LogicalPlan, physicalPlan: Option[PhysicalPlan], // if physical plan is none, the compilation is failed operatorIdToOutputSchemas: Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]], - operatorIdToError: Map[OperatorIdentity, WorkflowFatalError] + operatorIdToError: Map[OperatorIdentity, WorkflowFatalError], + outputPortsNeedingStorage: Set[GlobalPortIdentity] ) class WorkflowCompiler( context: WorkflowContext ) extends LazyLogging { - // function to expand logical plan to physical plan + /** + * Function to expand logical plan to physical plan + * @return the expanded physical plan and a set of output ports that need storage + */ private def expandLogicalPlan( logicalPlan: LogicalPlan, + logicalOpsToViewResult: List[String], errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] - ): PhysicalPlan = { + ): (PhysicalPlan, Set[GlobalPortIdentity]) = { + val terminalLogicalOps = logicalPlan.getTerminalOperatorIds + val logicalOpsNeedingStorage = + (terminalLogicalOps ++ logicalOpsToViewResult.map(OperatorIdentity(_))).toSet var physicalPlan = PhysicalPlan(operators = Set.empty, links = Set.empty) + val outputPortsNeedingStorage: mutable.HashSet[GlobalPortIdentity] = mutable.HashSet() - logicalPlan.getTopologicalOpIds.asScala.foreach { logicalOpId => - val logicalOp = logicalPlan.getOperator(logicalOpId) - val allUpstreamLinks = logicalPlan.getUpstreamLinks(logicalOp.operatorIdentifier) + logicalPlan.getTopologicalOpIds.asScala.foreach(logicalOpId => + Try { + val logicalOp = logicalPlan.getOperator(logicalOpId) - try { val subPlan = logicalOp.getPhysicalPlan(context.workflowId, context.executionId) - subPlan .topologicalIterator() .map(subPlan.getOperator) - .foreach { physicalOp => - val externalLinks = allUpstreamLinks - .filter(link => physicalOp.inputPorts.contains(link.toPortId)) - .flatMap { link => - physicalPlan - .getPhysicalOpsOfLogicalOp(link.fromOpId) - .find(_.outputPorts.contains(link.fromPortId)) - .map(fromOp => - PhysicalLink(fromOp.id, link.fromPortId, physicalOp.id, link.toPortId) - ) - } + .foreach({ physicalOp => + { + val externalLinks = logicalPlan + .getUpstreamLinks(logicalOp.operatorIdentifier) + .filter(link => physicalOp.inputPorts.contains(link.toPortId)) + .flatMap { link => + physicalPlan + .getPhysicalOpsOfLogicalOp(link.fromOpId) + .find(_.outputPorts.contains(link.fromPortId)) + .map(fromOp => + PhysicalLink(fromOp.id, link.fromPortId, physicalOp.id, link.toPortId) + ) + } - val internalLinks = subPlan.getUpstreamPhysicalLinks(physicalOp.id) + val internalLinks = subPlan.getUpstreamPhysicalLinks(physicalOp.id) - // Add the operator to the physical plan - physicalPlan = physicalPlan.addOperator(physicalOp.propagateSchema()) + // Add the operator to the physical plan + physicalPlan = physicalPlan.addOperator(physicalOp.propagateSchema()) - // Add all the links to the physical plan - physicalPlan = (externalLinks ++ internalLinks).foldLeft(physicalPlan) { (plan, link) => - plan.addLink(link) - } + // Add all the links to the physical plan + physicalPlan = (externalLinks ++ internalLinks) + .foldLeft(physicalPlan) { (plan, link) => plan.addLink(link) } - // **Check for Python-based operator errors during code generation** - if (physicalOp.isPythonBased) { - val code = physicalOp.getCode - val exceptionPattern = """#EXCEPTION DURING CODE GENERATION:\s*(.*)""".r + // **Check for Python-based operator errors during code generation** + if (physicalOp.isPythonBased) { + val code = physicalOp.getCode + val exceptionPattern = """#EXCEPTION DURING CODE GENERATION:\s*(.*)""".r - exceptionPattern.findFirstMatchIn(code).foreach { matchResult => - val errorMessage = matchResult.group(1).trim - val error = - new RuntimeException(s"Operator is not configured properly: $errorMessage") + exceptionPattern.findFirstMatchIn(code).foreach { matchResult => + val errorMessage = matchResult.group(1).trim + val error = + new RuntimeException(s"Operator is not configured properly: $errorMessage") - errorList match { - case Some(list) => list.append((logicalOpId, error)) // Store error and continue - case None => throw error // Throw immediately if no error list is provided + errorList match { + case Some(list) => list.append((logicalOpId, error)) // Store error and continue + case None => throw error // Throw immediately if no error list is provided + } } } } + }) + + // convert logical operators needing storage to output ports needing storage + subPlan + .topologicalIterator() + .filter(opId => logicalOpsNeedingStorage.contains(opId.logicalOpId)) + .map(physicalPlan.getOperator) + .foreach { physicalOp => + physicalOp.outputPorts + .filterNot(_._1.internal) + .foreach { + case (outputPortId, _) => + outputPortsNeedingStorage += GlobalPortIdentity( + opId = physicalOp.id, + portId = outputPortId + ) + } } - } catch { - case e: Throwable => + } match { + case Success(_) => + + case Failure(err) => errorList match { - case Some(list) => list.append((logicalOpId, e)) // Store error - case None => throw e // Throw if no list is provided + case Some(list) => list.append((logicalOpId, err)) + case None => throw err } } - } - - physicalPlan + ) + (physicalPlan, outputPortsNeedingStorage.toSet) } /** - * Compile a workflow to physical plan, along with the schema propagation result and error(if any) + * Compile a workflow to physical plan, along with the schema propagation result and error(if any). * * @param logicalPlanPojo the pojo parsed from workflow str provided by user - * @return WorkflowCompilationResult, containing the physical plan, input schemas per op and error per op + * @param errorHandling Lenient (editing-time, collect all errors) or Strict (pre-execution, throw) + * @return WorkflowCompilationResult, containing the logical plan, physical plan, output schemas per + * op, errors per op, and the output ports that need storage */ def compile( - logicalPlanPojo: LogicalPlanPojo + logicalPlanPojo: LogicalPlanPojo, + errorHandling: CompilationErrorHandling = CompilationErrorHandling.Lenient ): WorkflowCompilationResult = { - val errorList = new ArrayBuffer[(OperatorIdentity, Throwable)]() - var opIdToOutputSchema: Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]] = Map() + // Lenient collects into a buffer; Strict passes None so the first error is thrown. + val errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] = + errorHandling match { + case CompilationErrorHandling.Lenient => + Some(new ArrayBuffer[(OperatorIdentity, Throwable)]()) + case CompilationErrorHandling.Strict => None + } + // 1. convert the pojo to logical plan val logicalPlan: LogicalPlan = LogicalPlan(logicalPlanPojo) // 2. resolve the file name in each scan source operator - logicalPlan.resolveScanSourceOpFileName(Some(errorList)) + logicalPlan.resolveScanSourceOpFileName(errorList) - // 3. expand the logical plan to the physical plan - val physicalPlan = expandLogicalPlan(logicalPlan, Some(errorList)) + // 3. expand the logical plan to the physical plan, and get the output ports that need storage + val (physicalPlan, outputPortsNeedingStorage) = + expandLogicalPlan(logicalPlan, logicalPlanPojo.opsToViewResult, errorList) // 4. collect the output schema for each logical op - // even if error is encountered when logical => physical, we still want to get the input schemas for rest no-error operators - opIdToOutputSchema = collectOutputSchemaFromPhysicalPlan(physicalPlan, errorList) + // even if error is encountered when logical => physical, we still want to get the input schemas + // for the rest of the no-error operators. In Strict mode there is no buffer (errors already thrown), + // so schema errors go to a throwaway buffer and do not affect the result. + val schemaErrorList = errorList.getOrElse(new ArrayBuffer[(OperatorIdentity, Throwable)]()) + val opIdToOutputSchema = collectOutputSchemaFromPhysicalPlan(physicalPlan, schemaErrorList) + + val hasErrors = errorList.exists(_.nonEmpty) WorkflowCompilationResult( - physicalPlan = if (errorList.nonEmpty) None else Some(physicalPlan), + 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.toList) + 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 e6553e3cdf1..30d5d3f3ab2 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/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala similarity index 89% rename from workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala rename to common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala index eecb435cc87..97f39fddc63 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/compiler/model/LogicalPlan.scala +++ b/common/workflow-compiler/src/main/scala/org/apache/texera/common/compiler/model/LogicalPlan.scala @@ -17,7 +17,7 @@ * under the License. */ -package org.apache.texera.amber.compiler.model +package org.apache.texera.common.compiler.model import com.typesafe.scalalogging.LazyLogging import org.apache.texera.amber.core.storage.FileResolver @@ -61,7 +61,6 @@ object LogicalPlan { ): LogicalPlan = { LogicalPlan(pojo.operators, pojo.links) } - } case class LogicalPlan( @@ -81,6 +80,11 @@ case class LogicalPlan( def getOperator(opId: OperatorIdentity): LogicalOp = operatorMap(opId) + def getTerminalOperatorIds: List[OperatorIdentity] = + operatorMap.keys + .filter(op => jgraphtDag.outDegreeOf(op) == 0) + .toList + def addOperator(op: LogicalOp): LogicalPlan = { // TODO: fix schema for the new operator this.copy(operators :+ op, links) @@ -108,7 +112,8 @@ case class LogicalPlan( /** * 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 + * + * @param errorList if given, put errors during resolving to it; otherwise the error is thrown */ def resolveScanSourceOpFileName( errorList: Option[ArrayBuffer[(OperatorIdentity, Throwable)]] @@ -124,10 +129,18 @@ case class LogicalPlan( 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))) + errorList match { + case Some(errList) => + errList.append((operator.operatorIdentifier, err)) + case None => + // Throw the error if no errorList is provided + throw err + } } + case _ => // Skip non-ScanSourceOpDesc operators } } 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 151ee863fa7..efda9ceda79 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 cbc28388dd1..6e926d47158 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 63% 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 ee221728d31..a095a089fd2 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,168 @@ class WorkflowCompilerSpec extends AnyFlatSpec { s"expected both csvs in errors, got ${result.operatorIdToError.keySet}" ) } + + // -------------------- physical-plan shape -------------------- + + private def pojo( + operators: List[org.apache.texera.amber.operator.LogicalOp], + links: List[LogicalLink], + opsToViewResult: List[String] = List.empty + ): LogicalPlanPojo = + LogicalPlanPojo(operators, links, opsToViewResult, List.empty) + + // Re-anchor the subject after the sub-section. + "WorkflowCompiler" should "produce a physical plan that contains at least one physical op per logical op" in { + val csv = TestOperators.smallCsvScanOpDesc() + val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, keyword), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(), + keyword.operatorIdentifier, + PortIdentity() + ) + ) + ) + ) + + assert(result.logicalPlan.operators.size == 2) + val physicalPlan = result.physicalPlan.get + assert(physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).nonEmpty) + assert(physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).nonEmpty) + } + + it should "translate a logical link into a physical link between the two logical ops' physical ops" in { + val csv = TestOperators.smallCsvScanOpDesc() + val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, keyword), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(), + keyword.operatorIdentifier, + PortIdentity() + ) + ) + ) + ) + + val physicalPlan = result.physicalPlan.get + val csvPhysIds = + physicalPlan.getPhysicalOpsOfLogicalOp(csv.operatorIdentifier).map(_.id).toSet + val keywordPhysIds = + physicalPlan.getPhysicalOpsOfLogicalOp(keyword.operatorIdentifier).map(_.id).toSet + + val bridging = physicalPlan.links.filter(l => + csvPhysIds.contains(l.fromOpId) && keywordPhysIds.contains(l.toOpId) + ) + assert(bridging.nonEmpty, "expected at least one physical link from csv to keyword") + } + + // -------------------- storage-port collection -------------------- + + // The compiler walks `logicalPlan.getTerminalOperatorIds` (logical ops with + // out-degree 0) plus `opsToViewResult`, and for every physical op of those + // logical ops collects every non-internal output port into the result's + // `outputPortsNeedingStorage`. These tests pin both the terminal-default and + // the opsToViewResult-additive paths, and that internal ports are filtered. + + "WorkflowCompiler" should "mark the terminal op's output port as needing storage" in { + val csv = TestOperators.smallCsvScanOpDesc() + val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, keyword), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(), + keyword.operatorIdentifier, + PortIdentity() + ) + ) + ) + ) + + val storage = result.outputPortsNeedingStorage + assert( + storage.exists(_.opId.logicalOpId == keyword.operatorIdentifier), + s"expected keyword to be marked for storage, got ${storage.map(_.opId.logicalOpId)}" + ) + assert( + !storage.exists(_.opId.logicalOpId == csv.operatorIdentifier), + "csv is not terminal and was not requested via opsToViewResult; it should not be in storage" + ) + } + + it should "also mark a non-terminal op for storage when it is named in opsToViewResult" in { + val csv = TestOperators.smallCsvScanOpDesc() + val keyword = TestOperators.keywordSearchOpDesc("Region", "Asia") + + val result = new WorkflowCompiler(newContext()).compile( + pojo( + List(csv, keyword), + List( + LogicalLink( + csv.operatorIdentifier, + PortIdentity(), + keyword.operatorIdentifier, + PortIdentity() + ) + ), + opsToViewResult = List(csv.operatorIdentifier.id) + ) + ) + + val logicalOpsInStorage = result.outputPortsNeedingStorage.map(_.opId.logicalOpId) + assert( + logicalOpsInStorage.contains(csv.operatorIdentifier), + s"opsToViewResult should add csv to storage, got $logicalOpsInStorage" + ) + assert( + logicalOpsInStorage.contains(keyword.operatorIdentifier), + s"terminal keyword should remain in storage, got $logicalOpsInStorage" + ) + } + + it should "treat a single source op as terminal and mark its output port for storage" in { + val csv = TestOperators.smallCsvScanOpDesc() + + val result = new WorkflowCompiler(newContext()).compile(pojo(List(csv), List.empty)) + + val storage = result.outputPortsNeedingStorage + assert( + storage.exists(_.opId.logicalOpId == csv.operatorIdentifier), + "single op has out-degree 0, so its output port should land in storage" + ) + assert( + storage.forall(!_.portId.internal), + "compiler must filter out internal ports; storage should expose only user-visible outputs" + ) + } + + // -------------------- strict-mode error semantics -------------------- + + // Re-anchor the subject after the sub-section. + "WorkflowCompiler in strict mode" should "throw when a scan source has no fileName set" in { + // 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 == "no input file name") + } } 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 bd56aa7d5f6..6ffb9d497dc 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/service/resource/WorkflowCompilationResource.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowCompilationResource.scala index 501498b1d5c..526910dc2c3 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.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} 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 fd99c3d27dc..89a36bdc68a 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 {