Persist the spec of the computing_unit entity and its relationship with workflow_executions into DB

core/amber/src/main/scala/edu/uci/ics/amber/engine/common/Utils.scala

@@ -25,8 +25,6 @@ import com.fasterxml.jackson.module.noctordeser.NoCtorDeserModule
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
 import com.typesafe.scalalogging.LazyLogging
 import edu.uci.ics.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
-import org.jooq.DSLContext
-import org.jooq.impl.DSL
 
 import java.nio.file.{Files, Path, Paths}
 import java.text.SimpleDateFormat
@@ -121,6 +119,23 @@ object Utils extends LazyLogging {
     }
   }
 
+  def stringToAggregatedState(str: String): WorkflowAggregatedState = {
+    str.trim.toLowerCase match {
+      case "uninitialized" => WorkflowAggregatedState.UNINITIALIZED
+      case "ready"         => WorkflowAggregatedState.READY
+      case "initializing"  => WorkflowAggregatedState.READY // accept alias
+      case "running"       => WorkflowAggregatedState.RUNNING
+      case "pausing"       => WorkflowAggregatedState.PAUSING
+      case "paused"        => WorkflowAggregatedState.PAUSED
+      case "resuming"      => WorkflowAggregatedState.RESUMING
+      case "completed"     => WorkflowAggregatedState.COMPLETED
+      case "failed"        => WorkflowAggregatedState.FAILED
+      case "killed"        => WorkflowAggregatedState.KILLED
+      case "unknown"       => WorkflowAggregatedState.UNKNOWN
+      case other           => throw new IllegalArgumentException(s"Unrecognized state: $other")
+    }
+  }
+
   /**
     * @param state indicates the workflow state
     * @return code indicates the status of the execution in the DB it is 0 by default for any unused states.
@@ -151,15 +166,4 @@ object Utils extends LazyLogging {
       lock.unlock()
     }
   }
-
-  def withTransaction[T](dsl: DSLContext)(block: DSLContext => T): T = {
-    var result: Option[T] = None
-
-    dsl.transaction(configuration => {
-      val ctx = DSL.using(configuration)
-      result = Some(block(ctx))
-    })
-
-    result.getOrElse(throw new RuntimeException("Transaction failed without result!"))
-  }
 }

core/amber/src/main/scala/edu/uci/ics/texera/web/model/http/request/result/ResultExportRequest.scala

@@ -33,5 +33,7 @@ case class ResultExportRequest(
     rowIndex: Int, // used by "data" export
     columnIndex: Int, // used by "data" export
     filename: String, // optional filename override
-    destination: String // "dataset" or "local"
+    destination: String, // "dataset" or "local"
+    // TODO: remove it once the lifecycle of result and compute are unbundled
+    computingUnitId: Int // the id of the computing unit
 )

core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/request/WorkflowExecuteRequest.scala

@@ -36,7 +36,8 @@ case class WorkflowExecuteRequest(
     logicalPlan: LogicalPlanPojo,
     replayFromExecution: Option[ReplayExecutionInfo], // contains execution Id, interaction Id.
     workflowSettings: WorkflowSettings,
-    emailNotificationEnabled: Boolean
+    emailNotificationEnabled: Boolean,
+    computingUnitId: Int
 ) extends TexeraWebSocketRequest
 
 case class LogicalPlanPojo(

core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala

@@ -50,10 +50,11 @@ class WorkflowWebsocketResource extends LazyLogging {
     val sessionState = new SessionState(session)
     SessionState.setState(session.getId, sessionState)
     val wid = session.getRequestParameterMap.get("wid").get(0).toLong
+    val cuid = session.getRequestParameterMap.get("cuid").get(0).toInt
     // hack to refresh frontend run button state
     sessionState.send(WorkflowStateEvent("Uninitialized"))
     val workflowState =
-      WorkflowService.getOrCreate(WorkflowIdentity(wid))
+      WorkflowService.getOrCreate(WorkflowIdentity(wid), cuid)
     sessionState.subscribe(workflowState)
     sessionState.send(ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster))
     logger.info("connection open")

core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala

@@ -26,13 +26,15 @@ import edu.uci.ics.amber.core.virtualidentity._
 import edu.uci.ics.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
 import edu.uci.ics.amber.engine.architecture.logreplay.{ReplayDestination, ReplayLogRecord}
 import edu.uci.ics.amber.engine.common.AmberConfig
+import edu.uci.ics.amber.engine.common.Utils.{maptoStatusCode, stringToAggregatedState}
 import edu.uci.ics.amber.engine.common.storage.SequentialRecordStorage
 import edu.uci.ics.amber.util.serde.GlobalPortIdentitySerde.SerdeOps
 import edu.uci.ics.texera.dao.SqlServer
 import edu.uci.ics.texera.dao.jooq.generated.Tables._
 import edu.uci.ics.texera.dao.jooq.generated.tables.daos.WorkflowExecutionsDao
 import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.WorkflowExecutions
 import edu.uci.ics.texera.auth.SessionUser
+import edu.uci.ics.texera.dao.SqlServer.withTransaction
 import edu.uci.ics.texera.web.model.http.request.result.ResultExportRequest
 import edu.uci.ics.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource._
 import edu.uci.ics.texera.web.service.{ExecutionsMetadataPersistService, ResultExportService}
@@ -83,13 +85,13 @@ object WorkflowExecutionsResource {
     * @param wid workflow id
     * @return Integer
     */
-  def getLatestExecutionID(wid: Integer): Option[Integer] = {
+  def getLatestExecutionID(wid: Integer, cuid: Integer): Option[Integer] = {
     val executions = context
       .select(WORKFLOW_EXECUTIONS.EID)
       .from(WORKFLOW_EXECUTIONS)
       .join(WORKFLOW_VERSION)
       .on(WORKFLOW_EXECUTIONS.VID.eq(WORKFLOW_VERSION.VID))
-      .where(WORKFLOW_VERSION.WID.eq(wid))
+      .where(WORKFLOW_VERSION.WID.eq(wid).and(WORKFLOW_EXECUTIONS.CUID.eq(cuid)))
       .fetchInto(classOf[Integer])
       .asScala
       .toList
@@ -196,11 +198,24 @@ object WorkflowExecutionsResource {
         .map(URI.create)
     else None
 
-  def getWorkflowExecutions(wid: Integer, context: DSLContext): List[WorkflowExecutionEntry] = {
+  def getWorkflowExecutions(
+      wid: Integer,
+      context: DSLContext,
+      statusCodes: Set[Byte] = Set.empty
+  ): List[WorkflowExecutionEntry] = {
+    var condition = WORKFLOW_VERSION.WID.eq(wid)
+
+    if (statusCodes.nonEmpty) {
+      condition = condition.and(
+        WORKFLOW_EXECUTIONS.STATUS.in(statusCodes.map(Byte.box).asJava)
+      )
+    }
+
     context
       .select(
         WORKFLOW_EXECUTIONS.EID,
         WORKFLOW_EXECUTIONS.VID,
+        WORKFLOW_EXECUTIONS.CUID,
         USER.NAME,
         USER.GOOGLE_AVATAR,
         WORKFLOW_EXECUTIONS.STATUS,
@@ -216,7 +231,7 @@ object WorkflowExecutionsResource {
       .on(WORKFLOW_VERSION.VID.eq(WORKFLOW_EXECUTIONS.VID))
       .join(USER)
       .on(WORKFLOW_EXECUTIONS.UID.eq(USER.UID))
-      .where(WORKFLOW_VERSION.WID.eq(wid))
+      .where(condition)
       .orderBy(WORKFLOW_EXECUTIONS.EID.desc())
       .fetchInto(classOf[WorkflowExecutionEntry])
       .asScala
@@ -426,6 +441,7 @@ object WorkflowExecutionsResource {
   case class WorkflowExecutionEntry(
       eId: Integer,
       vId: Integer,
+      cuId: Integer,
       userName: String,
       googleAvatar: String,
       status: Byte,
@@ -466,6 +482,56 @@ case class ExecutionRenameRequest(wid: Integer, eId: Integer, executionName: Str
 @Path("/executions")
 class WorkflowExecutionsResource {
 
+  @GET
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @Path("/{wid}/latest")
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  def retrieveLatestExecutionEntry(
+      @PathParam("wid") wid: Integer,
+      @Auth sessionUser: SessionUser
+  ): WorkflowExecutionEntry = {
+
+    validateUserCanAccessWorkflow(sessionUser.getUser.getUid, wid)
+
+    withTransaction(context) { ctx =>
+      val latestEntryOpt =
+        ctx
+          .select(
+            WORKFLOW_EXECUTIONS.EID,
+            WORKFLOW_EXECUTIONS.VID,
+            WORKFLOW_EXECUTIONS.CUID,
+            USER.NAME,
+            USER.GOOGLE_AVATAR,
+            WORKFLOW_EXECUTIONS.STATUS,
+            WORKFLOW_EXECUTIONS.RESULT,
+            WORKFLOW_EXECUTIONS.STARTING_TIME,
+            WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME,
+            WORKFLOW_EXECUTIONS.BOOKMARKED,
+            WORKFLOW_EXECUTIONS.NAME,
+            WORKFLOW_EXECUTIONS.LOG_LOCATION
+          )
+          .from(WORKFLOW_EXECUTIONS)
+          .join(WORKFLOW_VERSION)
+          .on(WORKFLOW_VERSION.VID.eq(WORKFLOW_EXECUTIONS.VID))
+          .join(USER)
+          .on(WORKFLOW_EXECUTIONS.UID.eq(USER.UID))
+          .where(WORKFLOW_VERSION.WID.eq(wid))
+          // sort by latest VID first, then latest start-time
+          .orderBy(
+            WORKFLOW_EXECUTIONS.VID.desc(),
+            WORKFLOW_EXECUTIONS.EID.desc()
+          )
+          .limit(1)
+          .fetchInto(classOf[WorkflowExecutionEntry])
+          .asScala
+          .headOption
+
+      latestEntryOpt.getOrElse {
+        throw new ForbiddenException("Executions doesn't exist")
+      }
+    }
+  }
+
   @GET
   @Produces(Array(MediaType.APPLICATION_JSON))
   @Path("/{wid}/interactions/{eid}")
@@ -513,19 +579,34 @@ class WorkflowExecutionsResource {
   @RolesAllowed(Array("REGULAR", "ADMIN"))
   def retrieveExecutionsOfWorkflow(
       @PathParam("wid") wid: Integer,
-      @Auth sessionUser: SessionUser
+      @Auth sessionUser: SessionUser,
+      @QueryParam("status") status: String
   ): List[WorkflowExecutionEntry] = {
     val user = sessionUser.getUser
     if (!WorkflowAccessResource.hasReadAccess(wid, user.getUid)) {
       List()
     } else {
-      getWorkflowExecutions(wid, context)
+      val statusCodes: Set[Byte] =
+        Option(status)
+          .map(_.trim)
+          .filter(_.nonEmpty)
+          .map { raw =>
+            val tokens = raw.split(',').map(_.trim.toLowerCase).filter(_.nonEmpty)
+            try {
+              tokens.map(stringToAggregatedState).map(maptoStatusCode).toSet
+            } catch {
+              case e: IllegalArgumentException =>
+                throw new BadRequestException(e.getMessage)
+            }
+          }
+          .getOrElse(Set.empty[Byte])
+      getWorkflowExecutions(wid, context, statusCodes)
     }
   }
 
   @GET
   @Produces(Array(MediaType.APPLICATION_JSON))
-  @Path("/{wid}/{eid}")
+  @Path("/{wid}/stats/{eid}")
   def retrieveWorkflowRuntimeStatistics(
       @PathParam("wid") wid: Integer,
       @PathParam("eid") eid: Integer
@@ -662,7 +743,8 @@ class WorkflowExecutionsResource {
         case "local" =>
           // CASE A: multiple operators => produce ZIP
           if (request.operators.size > 1) {
-            val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+            val resultExportService =
+              new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId)
             val (zipStream, zipFileNameOpt) =
               resultExportService.exportOperatorsAsZip(request)
 
@@ -687,7 +769,8 @@ class WorkflowExecutionsResource {
           }
           val singleOp = request.operators.head
 
-          val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+          val resultExportService =
+            new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId)
           val (streamingOutput, fileNameOpt) =
             resultExportService.exportOperatorResultAsStream(request, singleOp)
 
@@ -706,7 +789,8 @@ class WorkflowExecutionsResource {
             .build()
         case _ =>
           // destination = "dataset" by default
-          val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+          val resultExportService =
+            new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId)
           val exportResponse =
             resultExportService.exportAllOperatorsResultToDataset(user.user, request)
           Response.ok(exportResponse).build()

core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala

@@ -283,6 +283,7 @@ object ExecutionResultService {
   */
 class ExecutionResultService(
     workflowIdentity: WorkflowIdentity,
+    computingUnitId: Int,
     val workflowStateStore: WorkflowStateStore
 ) extends SubscriptionManager
     with LazyLogging {
@@ -425,7 +426,7 @@ class ExecutionResultService(
   def handleResultPagination(request: ResultPaginationRequest): TexeraWebSocketEvent = {
     // calculate from index (pageIndex starts from 1 instead of 0)
     val from = request.pageSize * (request.pageIndex - 1)
-    val latestExecutionId = getLatestExecutionId(workflowIdentity).getOrElse(
+    val latestExecutionId = getLatestExecutionId(workflowIdentity, computingUnitId).getOrElse(
       throw new IllegalStateException("No execution is recorded")
     )
 

core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionsMetadataPersistService.scala

@@ -54,7 +54,8 @@ object ExecutionsMetadataPersistService extends LazyLogging {
       workflowId: WorkflowIdentity,
       uid: Option[Integer],
       executionName: String,
-      environmentVersion: String
+      environmentVersion: String,
+      computingUnitId: Integer
   ): ExecutionIdentity = {
     if (!AmberConfig.isUserSystemEnabled) return DEFAULT_EXECUTION_ID
     // first retrieve the latest version of this workflow
@@ -67,6 +68,10 @@ object ExecutionsMetadataPersistService extends LazyLogging {
     newExecution.setUid(uid.orNull)
     newExecution.setStartingTime(new Timestamp(System.currentTimeMillis()))
     newExecution.setEnvironmentVersion(environmentVersion)
+
+    // Set computing unit ID if provided
+    newExecution.setCuid(computingUnitId)
+
     workflowExecutionsDao.insert(newExecution)
     ExecutionIdentity(newExecution.getEid.longValue())
   }

core/amber/src/main/scala/edu/uci/ics/texera/web/service/ResultExportService.scala

@@ -83,7 +83,7 @@ object ResultExportService {
       .trim
 }
 
-class ResultExportService(workflowIdentity: WorkflowIdentity) {
+class ResultExportService(workflowIdentity: WorkflowIdentity, computingUnitId: Int) {
 
   import ResultExportService._
 
@@ -127,11 +127,11 @@ class ResultExportService(workflowIdentity: WorkflowIdentity) {
       operatorRequest: OperatorExportInfo
   ): (Option[String], Option[String]) = {
 
-    val execIdOpt = getLatestExecutionId(workflowIdentity)
-    if (execIdOpt.isEmpty)
+    val execIdOpt = getLatestExecutionId(workflowIdentity, computingUnitId)
+    if (execIdOpt.isEmpty) {
       return (None, Some(s"Workflow ${request.workflowId} has no execution result"))
-
-    val operatorDocument = getOperatorDocument(operatorRequest.id)
+    }
+    val operatorDocument = getOperatorDocument(operatorRequest.id, computingUnitId)
     if (operatorDocument == null || operatorDocument.getCount == 0)
       return (None, Some(s"No results to export for operator $operatorRequest"))
 
@@ -162,12 +162,12 @@ class ResultExportService(workflowIdentity: WorkflowIdentity) {
       request: ResultExportRequest,
       operatorRequest: OperatorExportInfo
   ): (StreamingOutput, Option[String]) = {
-    val execIdOpt = getLatestExecutionId(workflowIdentity)
+    val execIdOpt = getLatestExecutionId(workflowIdentity, computingUnitId)
     if (execIdOpt.isEmpty) {
       return (null, None)
     }
 
-    val operatorDocument = getOperatorDocument(operatorRequest.id)
+    val operatorDocument = getOperatorDocument(operatorRequest.id, computingUnitId)
     if (operatorDocument == null || operatorDocument.getCount == 0) {
       return (null, None)
     }
@@ -210,7 +210,7 @@ class ResultExportService(workflowIdentity: WorkflowIdentity) {
       .format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH-mm-ss"))
     val zipFileName = s"${request.workflowName}-$timestamp.zip"
 
-    val execIdOpt = getLatestExecutionId(workflowIdentity)
+    val execIdOpt = getLatestExecutionId(workflowIdentity, computingUnitId)
     if (execIdOpt.isEmpty) {
       throw new WebApplicationException(
         s"No execution result for workflow ${request.workflowId}"
@@ -221,7 +221,7 @@ class ResultExportService(workflowIdentity: WorkflowIdentity) {
       override def write(outputStream: OutputStream): Unit = {
         Using.resource(new ZipOutputStream(outputStream)) { zipOut =>
           request.operators.foreach { op =>
-            val operatorDocument = getOperatorDocument(op.id)
+            val operatorDocument = getOperatorDocument(op.id, computingUnitId)
             if (operatorDocument == null || operatorDocument.getCount == 0) {
               // create an "empty" file for this operator
               zipOut.putNextEntry(new ZipEntry(s"${op.id}-empty.txt"))
@@ -425,11 +425,14 @@ class ResultExportService(workflowIdentity: WorkflowIdentity) {
     * Generate the VirtualDocument for one operator's result.
     * Incorporates the remote code's extra parameter `None` for sub-operator ID.
     */
-  private def getOperatorDocument(operatorId: String): VirtualDocument[Tuple] = {
+  private def getOperatorDocument(
+      operatorId: String,
+      computingUnitId: Int
+  ): VirtualDocument[Tuple] = {
     // By now the workflow should finish running
     // Only supports external port 0 for now. TODO: support multiple ports
     val storageUri = WorkflowExecutionsResource.getResultUriByLogicalPortId(
-      getLatestExecutionId(workflowIdentity).get,
+      getLatestExecutionId(workflowIdentity, computingUnitId).get,
       OperatorIdentity(operatorId),
       PortIdentity()
     )