Fix download operators result (#3241)

This PR is in continuous of [PR
3221](#3221) to implement download
logic in backend.

## Export request structure updates
In this PR, we update `ResultExportRequest` to include a new variable
named `destination`. If its set to `local`, means the user wants to
download operators result. Also, `operatorIds` is now plural so with one
request, multiple operators can be exported (locally or to dataset).

## Export endpoint updates
In this PR, we update `ResultResource` to support `destination=local`.
In this regard, we have two cases:
- If destination is local and one operator is selected, download one
operator result based on export type
- If destination is local and multiple operators are selected, download
a zip file containing all
The export to dataset follows the old logic.

## Export service updates
In this PR, we update `ResultExportService` to have two core more new
functions:
- `exportOperatorResultAsStream`: For local download of a single
operator. Streams the data directly.
- `exportOperatorsAsZip`: For local download of multiple operators. Zip
the output and stream back.

## Demo video


https://github.com/user-attachments/assets/b46b33a7-9503-4ed8-99f2-9dc15878e54d

---------

Co-authored-by: Jiadong Bai <[email protected]>

Author: aicam

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

@@ -1,13 +1,13 @@
 package edu.uci.ics.texera.web.model.websocket.request
 
 case class ResultExportRequest(
-    exportType: String,
+    exportType: String, // e.g. "csv", "google_sheet", "arrow", "data"
     workflowId: Int,
     workflowName: String,
-    operatorId: String,
-    operatorName: String,
+    operatorIds: List[String], // changed from single operatorId: String -> List of strings
     datasetIds: List[Int],
-    rowIndex: Int,
-    columnIndex: Int,
-    filename: String
+    rowIndex: Int, // used by "data" export
+    columnIndex: Int, // used by "data" export
+    filename: String, // optional filename override
+    destination: String // "dataset" or "local"
 )

core/amber/src/main/scala/edu/uci/ics/texera/web/model/websocket/response/ResultExportResponse.scala

@@ -1,5 +1,3 @@
 package edu.uci.ics.texera.web.model.websocket.response
 
-import edu.uci.ics.texera.web.model.websocket.event.TexeraWebSocketEvent
-
-case class ResultExportResponse(status: String, message: String) extends TexeraWebSocketEvent
+case class ResultExportResponse(status: String, message: String)

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

@@ -9,10 +9,12 @@ import edu.uci.ics.texera.web.service.ResultExportService
 import io.dropwizard.auth.Auth
 
 import javax.ws.rs._
-import javax.ws.rs.core.Response
+import javax.ws.rs.core.{MediaType, Response}
+import javax.ws.rs.core.Response.Status
 import scala.jdk.CollectionConverters._
 
 @Path("/result")
+@Produces(Array(MediaType.APPLICATION_JSON, MediaType.APPLICATION_OCTET_STREAM))
 class ResultResource extends LazyLogging {
 
   @POST
@@ -22,21 +24,73 @@ class ResultResource extends LazyLogging {
       @Auth user: SessionUser
   ): Response = {
 
+    if (request.operatorIds.size <= 0)
+      Response
+        .status(Response.Status.BAD_REQUEST)
+        .`type`(MediaType.APPLICATION_JSON)
+        .entity(Map("error" -> "No operator selected").asJava)
+        .build()
+
     try {
-      val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+      request.destination match {
+        case "local" =>
+          // CASE A: multiple operators => produce ZIP
+          if (request.operatorIds.size > 1) {
+            val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+            val (zipStream, zipFileNameOpt) =
+              resultExportService.exportOperatorsAsZip(user.user, request)
+
+            if (zipStream == null) {
+              throw new RuntimeException("Zip stream is null")
+            }
+
+            val finalFileName = zipFileNameOpt.getOrElse("operators.zip")
+            return Response
+              .ok(zipStream, "application/zip")
+              .header("Content-Disposition", "attachment; filename=\"" + finalFileName + "\"")
+              .build()
+          }
 
-      val exportResponse: ResultExportResponse =
-        resultExportService.exportResult(user.user, request)
+          // CASE B: exactly one operator => single file
+          if (request.operatorIds.size != 1) {
+            return Response
+              .status(Response.Status.BAD_REQUEST)
+              .`type`(MediaType.APPLICATION_JSON)
+              .entity(Map("error" -> "Local download supports no operator or many.").asJava)
+              .build()
+          }
+          val singleOpId = request.operatorIds.head
 
-      Response.ok(exportResponse).build()
+          val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+          val (streamingOutput, fileNameOpt) =
+            resultExportService.exportOperatorResultAsStream(request, singleOpId)
 
+          if (streamingOutput == null) {
+            return Response
+              .status(Response.Status.INTERNAL_SERVER_ERROR)
+              .`type`(MediaType.APPLICATION_JSON)
+              .entity(Map("error" -> "Failed to export operator").asJava)
+              .build()
+          }
+
+          val finalFileName = fileNameOpt.getOrElse("download.dat")
+          Response
+            .ok(streamingOutput, MediaType.APPLICATION_OCTET_STREAM)
+            .header("Content-Disposition", "attachment; filename=\"" + finalFileName + "\"")
+            .build()
+        case _ =>
+          // destination = "dataset" by default
+          val resultExportService = new ResultExportService(WorkflowIdentity(request.workflowId))
+          val exportResponse = resultExportService.exportResult(user.user, request)
+          Response.ok(exportResponse).build()
+      }
     } catch {
       case ex: Exception =>
         Response
           .status(Response.Status.INTERNAL_SERVER_ERROR)
+          .`type`(MediaType.APPLICATION_JSON)
           .entity(Map("error" -> ex.getMessage).asJava)
           .build()
     }
   }
-
 }