This is an automated email from the ASF dual-hosted git repository.
aicam pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 13171990db feat: switch workflow result downloads to use browser
native downloads (#3728)
13171990db is described below
commit 13171990db1b6959f99fead4cac26f443c52817b
Author: Madison Lin <[email protected]>
AuthorDate: Fri Oct 10 19:22:40 2025 -0700
feat: switch workflow result downloads to use browser native downloads
(#3728)
### Purpose ###
This pull request partially addresses issue #3404 by switching workflow
result downloads to use the browser's native download functionality.
Using the browser to handle downloads enables the browser's built-in
download UI, including a progress bar when downloading larger files.
### Changes ###
Previously, we used Angular's `HttpClient` to request the result file
from the backend as a `Blob` and download the `Blob` to the local
filesystem. Now, we use a standard HTML form submission to request the
result file, allowing the browser to handle the download. The backend
logic for generating the file stream remains the same.
Headers such as `Authorization` (used for Bearer tokens) cannot be added
to standard form submissions. To work around this, we pass the JWT token
directly as a form parameter and manually check for validity of the
token in the backend. We also manually verify that the user role is
valid.
#### Summary of Changes ####
- Changed frontend logic for local workflow result downloads to use a
standard form submission instead of `HttpClient` to request the
download, allowing the browser to handle the download
- Added endpoint to receive form request and authenticate the request
via the JWT token before triggering download
### Demonstration ###
https://github.com/user-attachments/assets/a2d5278e-5ebf-441f-99bf-ff93d1667271
---------
Signed-off-by: Xinyuan Lin <[email protected]>
Co-authored-by: Chris <[email protected]>
Co-authored-by: ali risheh <[email protected]>
Co-authored-by: ali <[email protected]>
Co-authored-by: Xinyuan Lin <[email protected]>
---
core/access-control-service/build.sbt | 3 +-
.../service/resource/AccessControlResource.scala | 99 ++++++++++++---
.../uci/ics/texera/AccessControlResourceSpec.scala | 2 +-
.../http/request/result/ResultExportRequest.scala | 11 +-
.../user/workflow/WorkflowExecutionsResource.scala | 139 +++++++--------------
.../texera/web/service/ResultExportService.scala | 84 +++++++++++--
.../service/user/download/download.service.ts | 113 +++++++++++------
.../workflow-result-export.service.ts | 47 ++++---
.../texera-helmchart/templates/envoy-config.yaml | 4 +
deployment/k8s/texera-helmchart/values.yaml | 3 +-
10 files changed, 324 insertions(+), 181 deletions(-)
diff --git a/core/access-control-service/build.sbt
b/core/access-control-service/build.sbt
index 052dad4c13..abdba34645 100644
--- a/core/access-control-service/build.sbt
+++ b/core/access-control-service/build.sbt
@@ -76,6 +76,5 @@ libraryDependencies ++= Seq(
libraryDependencies ++= Seq(
"io.dropwizard" % "dropwizard-core" % dropwizardVersion,
"io.dropwizard" % "dropwizard-auth" % dropwizardVersion, // Dropwizard
Authentication module
- "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.15.2",
- "org.playframework" %% "play-json" % "3.1.0-M1",
+ "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.15.2"
)
\ No newline at end of file
diff --git
a/core/access-control-service/src/main/scala/edu/uci/ics/texera/service/resource/AccessControlResource.scala
b/core/access-control-service/src/main/scala/edu/uci/ics/texera/service/resource/AccessControlResource.scala
index 2ff2355ca8..50492f2332 100644
---
a/core/access-control-service/src/main/scala/edu/uci/ics/texera/service/resource/AccessControlResource.scala
+++
b/core/access-control-service/src/main/scala/edu/uci/ics/texera/service/resource/AccessControlResource.scala
@@ -17,6 +17,7 @@
package edu.uci.ics.texera.service.resource
+import com.fasterxml.jackson.databind.ObjectMapper
import com.typesafe.scalalogging.LazyLogging
import edu.uci.ics.texera.auth.JwtParser.parseToken
import edu.uci.ics.texera.auth.SessionUser
@@ -24,13 +25,19 @@ import edu.uci.ics.texera.auth.util.{ComputingUnitAccess,
HeaderField}
import edu.uci.ics.texera.dao.jooq.generated.enums.PrivilegeEnum
import jakarta.ws.rs.core._
import jakarta.ws.rs.{GET, POST, Path, Produces}
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import java.io.StringReader
+import java.net.URLDecoder
+import java.nio.charset.StandardCharsets
import java.util.Optional
import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala}
import scala.util.matching.Regex
object AccessControlResource extends LazyLogging {
+ private val mapper: ObjectMapper = new
ObjectMapper().registerModule(DefaultScalaModule)
+
// Regex for the paths that require authorization
private val wsapiWorkflowWebsocket: Regex =
""".*/wsapi/workflow-websocket.*""".r
private val apiExecutionsStats: Regex =
""".*/api/executions/[0-9]+/stats/[0-9]+.*""".r
@@ -43,20 +50,28 @@ object AccessControlResource extends LazyLogging {
* headers sent by the client (browser)
* @return HTTP Response with appropriate status code and headers
*/
- def authorize(uriInfo: UriInfo, headers: HttpHeaders): Response = {
+ def authorize(
+ uriInfo: UriInfo,
+ headers: HttpHeaders,
+ bodyOpt: Option[String] = None
+ ): Response = {
val path = uriInfo.getPath
logger.info(s"Authorizing request for path: $path")
path match {
case wsapiWorkflowWebsocket() | apiExecutionsStats() |
apiExecutionsResultExport() =>
- checkComputingUnitAccess(uriInfo, headers)
+ checkComputingUnitAccess(uriInfo, headers, bodyOpt)
case _ =>
logger.warn(s"No authorization logic for path: $path. Denying access.")
Response.status(Response.Status.FORBIDDEN).build()
}
}
- private def checkComputingUnitAccess(uriInfo: UriInfo, headers:
HttpHeaders): Response = {
+ private def checkComputingUnitAccess(
+ uriInfo: UriInfo,
+ headers: HttpHeaders,
+ bodyOpt: Option[String]
+ ): Response = {
val queryParams: Map[String, String] = uriInfo
.getQueryParameters()
.asScala
@@ -68,15 +83,17 @@ object AccessControlResource extends LazyLogging {
s"Request URI: ${uriInfo.getRequestUri} and headers:
${headers.getRequestHeaders.asScala} and queryParams: $queryParams"
)
- val token = queryParams.getOrElse(
- "access-token",
- headers
- .getRequestHeader("Authorization")
- .asScala
- .headOption
- .getOrElse("")
- .replace("Bearer ", "")
- )
+ val token: String = {
+ val qToken = queryParams.get("access-token").filter(_.nonEmpty)
+ val hToken = Option(headers.getRequestHeader("Authorization"))
+ .flatMap(_.asScala.headOption)
+ .map(_.replaceFirst("(?i)^Bearer\\s+", "")) // case-insensitive
"Bearer "
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ val bToken = bodyOpt.flatMap(extractTokenFromBody)
+ qToken.orElse(hToken).orElse(bToken).getOrElse("")
+ }
+ logger.info(s"token extracted from request $token")
val cuid = queryParams.getOrElse("cuid", "")
val cuidInt =
try {
@@ -99,6 +116,7 @@ object AccessControlResource extends LazyLogging {
return Response.status(Response.Status.FORBIDDEN).build()
} catch {
case e: Exception =>
+ logger.error(s"Failed parsing token $e")
return Response.status(Response.Status.FORBIDDEN).build()
}
@@ -110,6 +128,57 @@ object AccessControlResource extends LazyLogging {
.header(HeaderField.UserEmail, userSession.get().getEmail)
.build()
}
+
+ // Extracts a top-level "token" field from a JSON body
+ private def extractTokenFromBody(body: String): Option[String] = {
+ // 1) Try JSON
+ val jsonToken: Option[String] =
+ try {
+ val node = mapper.readTree(body)
+ if (node != null && node.has("token"))
+ Option(node.get("token").asText()).map(_.trim).filter(_.nonEmpty)
+ else None
+ } catch {
+ case _: Exception => None
+ }
+
+ // 2) Try application/x-www-form-urlencoded
+ def extractTokenFromUrlEncoded(s: String): Option[String] = {
+ // fast path: must contain '=' or '&'
+ if (!s.contains("=")) return None
+ val pairs = s.split("&").iterator
+ var found: Option[String] = None
+ while (pairs.hasNext && found.isEmpty) {
+ val p = pairs.next()
+ val idx = p.indexOf('=')
+ val key = if (idx >= 0) p.substring(0, idx) else p
+ if (key == "token") {
+ val raw = if (idx >= 0) p.substring(idx + 1) else ""
+ val decoded = URLDecoder.decode(raw, StandardCharsets.UTF_8.name())
+ val v = decoded.trim
+ if (v.nonEmpty) found = Some(v)
+ }
+ }
+ found
+ }
+
+ // 3) Try multipart/form-data (best-effort; parses raw body text)
+ def extractTokenFromMultipart(s: String): Option[String] = {
+ // Look for the part with name="token" and capture its content until the
next boundary
+ val partWithBoundary =
"(?s)name\\s*=\\s*\"token\"[^\\r\\n]*\\r?\\n\\r?\\n(.*?)\\r?\\n--".r
+ val partToEnd =
"(?s)name\\s*=\\s*\"token\"[^\\r\\n]*\\r?\\n\\r?\\n(.*)".r
+
+ partWithBoundary
+ .findFirstMatchIn(s)
+ .map(_.group(1).trim)
+ .filter(_.nonEmpty)
+
.orElse(partToEnd.findFirstMatchIn(s).map(_.group(1).trim).filter(_.nonEmpty))
+ }
+
+ jsonToken
+ .orElse(extractTokenFromUrlEncoded(body))
+ .orElse(extractTokenFromMultipart(body))
+ }
}
@Produces(Array(MediaType.APPLICATION_JSON))
@Path("/auth")
@@ -128,8 +197,10 @@ class AccessControlResource extends LazyLogging {
@Path("/{path:.*}")
def authorizePost(
@Context uriInfo: UriInfo,
- @Context headers: HttpHeaders
+ @Context headers: HttpHeaders,
+ body: String
): Response = {
- AccessControlResource.authorize(uriInfo, headers)
+ logger.info("Request body: " + body)
+ AccessControlResource.authorize(uriInfo, headers,
Option(body).map(_.trim).filter(_.nonEmpty))
}
}
diff --git
a/core/access-control-service/src/test/scala/edu/uci/ics/texera/AccessControlResourceSpec.scala
b/core/access-control-service/src/test/scala/edu/uci/ics/texera/AccessControlResourceSpec.scala
index 3aa71b48aa..cc4d8f8652 100644
---
a/core/access-control-service/src/test/scala/edu/uci/ics/texera/AccessControlResourceSpec.scala
+++
b/core/access-control-service/src/test/scala/edu/uci/ics/texera/AccessControlResourceSpec.scala
@@ -165,7 +165,7 @@ class AccessControlResourceSpec
when(mockHttpHeaders.getRequestHeader("Authorization")).thenReturn(new
util.ArrayList[String]())
val accessControlResource = new AccessControlResource()
- val response = accessControlResource.authorizePost(mockUriInfo,
mockHttpHeaders)
+ val response = accessControlResource.authorizePost(mockUriInfo,
mockHttpHeaders, null)
response.getStatus shouldBe Response.Status.FORBIDDEN.getStatusCode
}
diff --git
a/core/amber/src/main/scala/edu/uci/ics/texera/web/model/http/request/result/ResultExportRequest.scala
b/core/amber/src/main/scala/edu/uci/ics/texera/web/model/http/request/result/ResultExportRequest.scala
index 8190cbe7e1..156135711e 100644
---
a/core/amber/src/main/scala/edu/uci/ics/texera/web/model/http/request/result/ResultExportRequest.scala
+++
b/core/amber/src/main/scala/edu/uci/ics/texera/web/model/http/request/result/ResultExportRequest.scala
@@ -19,11 +19,17 @@
package edu.uci.ics.texera.web.model.http.request.result
+import play.api.libs.json._
+
case class OperatorExportInfo(
id: String,
outputType: String
)
+object OperatorExportInfo {
+ implicit val fmt: OFormat[OperatorExportInfo] =
Json.format[OperatorExportInfo]
+}
+
case class ResultExportRequest(
exportType: String, // e.g. "csv", "google_sheet", "arrow", "data"
workflowId: Int,
@@ -33,7 +39,10 @@ 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"
// TODO: remove it once the lifecycle of result and compute are unbundled
computingUnitId: Int // the id of the computing unit
)
+
+object ResultExportRequest {
+ implicit val fmt: OFormat[ResultExportRequest] =
Json.format[ResultExportRequest]
+}
diff --git
a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
index 7192f7f3fd..8c71c46799 100644
---
a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
+++
b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
@@ -38,6 +38,13 @@ import edu.uci.ics.texera.dao.SqlServer
import edu.uci.ics.texera.dao.SqlServer.withTransaction
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.{JwtParser, SessionUser}
+
+import scala.jdk.CollectionConverters._
+import edu.uci.ics.texera.config.UserSystemConfig
+import edu.uci.ics.texera.dao.SqlServer.withTransaction
+import edu.uci.ics.texera.dao.jooq.generated.enums.UserRoleEnum
import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.{User => UserPojo,
WorkflowExecutions}
import edu.uci.ics.texera.web.model.http.request.result.ResultExportRequest
import
edu.uci.ics.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource._
@@ -52,7 +59,7 @@ import javax.annotation.security.RolesAllowed
import javax.ws.rs._
import javax.ws.rs.core.{MediaType, Response}
import scala.collection.mutable
-import scala.jdk.CollectionConverters._
+import play.api.libs.json.Json
object WorkflowExecutionsResource {
final private lazy val context = SqlServer
@@ -823,106 +830,50 @@ class WorkflowExecutionsResource {
}
@POST
- @Path("/result/export")
+ @Path("/result/export/dataset")
@RolesAllowed(Array("REGULAR", "ADMIN"))
- def exportResult(
- request: ResultExportRequest,
- @Auth user: SessionUser
- ): Response = {
-
- if (request.operators.size <= 0)
- Response
- .status(Response.Status.BAD_REQUEST)
- .`type`(MediaType.APPLICATION_JSON)
- .entity(Map("error" -> "No operator selected").asJava)
- .build()
-
- // Get ALL non-downloadable in workflow
- val datasetRestrictions =
getNonDownloadableOperatorMap(request.workflowId, user.user)
- // Filter to only user's selection
- val restrictedOperators = request.operators.filter(op =>
datasetRestrictions.contains(op.id))
- // Check if any selected operator is restricted
- if (restrictedOperators.nonEmpty) {
- val restrictedDatasets = restrictedOperators.flatMap { op =>
- datasetRestrictions(op.id).map {
- case (ownerEmail, datasetName) =>
- Map(
- "operatorId" -> op.id,
- "ownerEmail" -> ownerEmail,
- "datasetName" -> datasetName
- ).asJava
- }
- }
+ def exportResultToDataset(request: ResultExportRequest, @Auth user:
SessionUser): Response = {
+ try {
+ val resultExportService =
+ new ResultExportService(WorkflowIdentity(request.workflowId),
request.computingUnitId)
+ resultExportService.exportToDataset(user.user, request)
- return Response
- .status(Response.Status.FORBIDDEN)
- .`type`(MediaType.APPLICATION_JSON)
- .entity(
- Map(
- "error" -> "Export blocked due to dataset restrictions",
- "restrictedDatasets" -> restrictedDatasets.asJava
- ).asJava
- )
- .build()
+ } catch {
+ case ex: Exception =>
+ Response
+ .status(Response.Status.INTERNAL_SERVER_ERROR)
+ .`type`(MediaType.APPLICATION_JSON)
+ .entity(Map("error" -> ex.getMessage).asJava)
+ .build()
}
+ }
- try {
- request.destination match {
- case "local" =>
- // CASE A: multiple operators => produce ZIP
- if (request.operators.size > 1) {
- val resultExportService =
- new ResultExportService(WorkflowIdentity(request.workflowId),
request.computingUnitId)
- val (zipStream, zipFileNameOpt) =
- resultExportService.exportOperatorsAsZip(request)
-
- if (zipStream == null) {
- throw new RuntimeException("Zip stream is null")
- }
+ @POST
+ @Path("/result/export/local")
+ @Consumes(Array(MediaType.APPLICATION_FORM_URLENCODED))
+ def exportResultToLocal(
+ @FormParam("request") requestJson: String,
+ @FormParam("token") token: String
+ ): Response = {
- val finalFileName = zipFileNameOpt.getOrElse("operators.zip")
- return Response
- .ok(zipStream, "application/zip")
- .header("Content-Disposition", "attachment; filename=\"" +
finalFileName + "\"")
- .build()
- }
+ try {
+ val userOpt = JwtParser.parseToken(token)
+ if (userOpt.isPresent) {
+ val user = userOpt.get()
+ val role = user.getUser.getRole
+ val RolesAllowed = Set(UserRoleEnum.REGULAR, UserRoleEnum.ADMIN)
+ if (!RolesAllowed.contains(role)) {
+ throw new RuntimeException("User role is not allowed to perform this
download")
+ }
+ } else {
+ throw new RuntimeException("Invalid or expired token")
+ }
- // CASE B: exactly one operator => single file
- if (request.operators.size != 1) {
- return Response
- .status(Response.Status.BAD_REQUEST)
- .`type`(MediaType.APPLICATION_JSON)
- .entity(Map("error" -> "Local download does not support no
operator.").asJava)
- .build()
- }
- val singleOp = request.operators.head
-
- val resultExportService =
- new ResultExportService(WorkflowIdentity(request.workflowId),
request.computingUnitId)
- val (streamingOutput, fileNameOpt) =
- resultExportService.exportOperatorResultAsStream(request, singleOp)
-
- 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 request = Json.parse(requestJson).as[ResultExportRequest]
+ val resultExportService =
+ new ResultExportService(WorkflowIdentity(request.workflowId),
request.computingUnitId)
+ resultExportService.exportToLocal(request)
- 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),
request.computingUnitId)
- val exportResponse =
- resultExportService.exportAllOperatorsResultToDataset(user.user,
request)
- Response.ok(exportResponse).build()
- }
} catch {
case ex: Exception =>
Response
diff --git
a/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ResultExportService.scala
b/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ResultExportService.scala
index d4125883a9..1da0ce6e3c 100644
---
a/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ResultExportService.scala
+++
b/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ResultExportService.scala
@@ -19,6 +19,9 @@
package edu.uci.ics.texera.web.service
+import com.fasterxml.jackson.core.`type`.TypeReference
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.github.tototoshi.csv.CSVWriter
import edu.uci.ics.amber.config.EnvironmentalVariable
import edu.uci.ics.amber.core.storage.DocumentFactory
@@ -38,6 +41,7 @@ import
edu.uci.ics.texera.web.resource.dashboard.user.workflow.{
}
import
edu.uci.ics.texera.web.service.WorkflowExecutionService.getLatestExecutionId
+import scala.jdk.CollectionConverters._
import java.io.{FilterOutputStream, IOException, OutputStream}
import java.nio.channels.Channels
import java.nio.charset.StandardCharsets
@@ -53,10 +57,9 @@ import org.apache.arrow.vector.ipc.ArrowFileWriter
import org.apache.commons.lang3.StringUtils
import javax.ws.rs.WebApplicationException
-import javax.ws.rs.core.StreamingOutput
+import javax.ws.rs.core.{MediaType, Response, StreamingOutput}
import java.net.{HttpURLConnection, URL, URLEncoder}
import scala.collection.mutable.ArrayBuffer
-
import org.apache.commons.io.IOUtils
object Constants {
@@ -91,12 +94,12 @@ class ResultExportService(workflowIdentity:
WorkflowIdentity, computingUnitId: I
import ResultExportService._
/**
- * Export results for all specified operators in the request.
+ * Export operator results to a dataset and return the result.
*/
- def exportAllOperatorsResultToDataset(
+ def exportToDataset(
user: User,
request: ResultExportRequest
- ): ResultExportResponse = {
+ ): Response = {
val successMessages = new mutable.ListBuffer[String]()
val errorMessages = new mutable.ListBuffer[String]()
@@ -111,13 +114,49 @@ class ResultExportService(workflowIdentity:
WorkflowIdentity, computingUnitId: I
}
}
+ var exportResponse: ResultExportResponse = null
if (errorMessages.isEmpty) {
- ResultExportResponse("success", successMessages.mkString("\n"))
+ exportResponse = ResultExportResponse("success",
successMessages.mkString("\n"))
} else if (successMessages.isEmpty) {
- ResultExportResponse("error", errorMessages.mkString("\n"))
+ exportResponse = ResultExportResponse("error",
errorMessages.mkString("\n"))
} else {
// At least one success, so we consider overall success (with partial
possible).
- ResultExportResponse("success", successMessages.mkString("\n"))
+ exportResponse = ResultExportResponse("success",
successMessages.mkString("\n"))
+ }
+
+ Response.ok(exportResponse).build()
+ }
+
+ /**
+ * Export operator results as downloadable files.
+ * If multiple operators are selected, their results are streamed as a ZIP
file.
+ * If a single operator is selected, its result is streamed directly.
+ */
+ def exportToLocal(request: ResultExportRequest): Response = {
+ if (request.operators.size > 1) {
+ val (zipStream, zipFileNameOpt) = exportOperatorsAsZip(request)
+ if (zipStream == null) {
+ throw new RuntimeException("Zip stream is null")
+ }
+ val fileName = zipFileNameOpt.getOrElse("operators.zip")
+
+ Response
+ .ok(zipStream, "application/zip")
+ .header("Content-Disposition", s"""attachment; filename="$fileName"""")
+ .build()
+
+ } else {
+ val op = request.operators.head
+ val (streamingOutput, fileNameOpt) =
exportOperatorResultAsStream(request, op)
+ if (streamingOutput == null) {
+ throw new RuntimeException("Failed to export operator")
+ }
+ val fileName = fileNameOpt.getOrElse("download.dat")
+
+ Response
+ .ok(streamingOutput, MediaType.APPLICATION_OCTET_STREAM)
+ .header("Content-Disposition", s"""attachment; filename="$fileName"""")
+ .build()
}
}
@@ -205,10 +244,6 @@ class ResultExportService(workflowIdentity:
WorkflowIdentity, computingUnitId: I
def exportOperatorsAsZip(
request: ResultExportRequest
): (StreamingOutput, Option[String]) = {
- if (request.operators.isEmpty) {
- return (null, None)
- }
-
val timestamp = LocalDateTime
.now()
.truncatedTo(ChronoUnit.SECONDS)
@@ -562,4 +597,29 @@ class ResultExportService(workflowIdentity:
WorkflowIdentity, computingUnitId: I
// remove path separators
StringUtils.replaceEach(rawName, Array("/", "\\"), Array("", ""))
}
+
+ /**
+ * Parse a JSON string array of operators into a list of OperatorExportInfo
objects.
+ */
+ def parseOperators(operatorsJson: String): List[OperatorExportInfo] = {
+ new ObjectMapper()
+ .registerModule(DefaultScalaModule)
+ .readValue(operatorsJson, new TypeReference[List[OperatorExportInfo]] {})
+ }
+
+ /**
+ * Validate an export request by checking if any operators are selected.
+ * Return an error response if none are selected, otherwise None.
+ */
+ def validateExportRequest(request: ResultExportRequest): Option[Response] = {
+ if (request.operators.isEmpty) {
+ Some(
+ Response
+ .status(Response.Status.BAD_REQUEST)
+ .`type`(MediaType.APPLICATION_JSON)
+ .entity(Map("error" -> "No operator selected").asJava)
+ .build()
+ )
+ } else None
+ }
}
diff --git
a/core/gui/src/app/dashboard/service/user/download/download.service.ts
b/core/gui/src/app/dashboard/service/user/download/download.service.ts
index aafb14a44c..eff9fa71b6 100644
--- a/core/gui/src/app/dashboard/service/user/download/download.service.ts
+++ b/core/gui/src/app/dashboard/service/user/download/download.service.ts
@@ -30,9 +30,11 @@ import { AppSettings } from "../../../../common/app-setting";
import { HttpClient, HttpResponse } from "@angular/common/http";
import { WORKFLOW_EXECUTIONS_API_BASE_URL } from
"../workflow-executions/workflow-executions.service";
import { DashboardWorkflowComputingUnit } from
"../../../../workspace/types/workflow-computing-unit";
+import { TOKEN_KEY } from "../../../../common/service/user/auth.service";
var contentDisposition = require("content-disposition");
export const EXPORT_BASE_URL = "result/export";
+const IFRAME_TIMEOUT_MS = 10000;
export const DOWNLOADABILITY_BASE_URL = "result/downloadability";
interface DownloadableItem {
@@ -148,7 +150,6 @@ export class DownloadService {
rowIndex: number,
columnIndex: number,
filename: string,
- destination: "local" | "dataset" = "dataset", // "local" or "dataset" =>
default to "dataset"
unit: DashboardWorkflowComputingUnit // computing unit for cluster setting
): Observable<HttpResponse<Blob> | HttpResponse<ExportWorkflowJsonResponse>>
{
const computingUnitId = unit.computingUnit.cuid;
@@ -161,51 +162,91 @@ export class DownloadService {
rowIndex,
columnIndex,
filename,
- destination,
computingUnitId,
};
const urlPath =
unit && unit.computingUnit.type == "kubernetes" &&
unit.computingUnit?.cuid
- ?
`${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}?cuid=${unit.computingUnit.cuid}`
- : `${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}`;
- if (destination === "local") {
- return this.http.post(urlPath, requestBody, {
- responseType: "blob",
- observe: "response",
- headers: {
- "Content-Type": "application/json",
- Accept: "application/octet-stream",
- },
- });
- } else {
- // dataset => return JSON
- return this.http.post<ExportWorkflowJsonResponse>(urlPath, requestBody, {
- responseType: "json",
- observe: "response",
- headers: {
- "Content-Type": "application/json",
- Accept: "application/json",
- },
- });
- }
+ ?
`${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}/dataset?cuid=${unit.computingUnit.cuid}`
+ : `${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}/dataset`;
+
+ return this.http.post<ExportWorkflowJsonResponse>(urlPath, requestBody, {
+ responseType: "json",
+ observe: "response",
+ headers: {
+ "Content-Type": "application/json",
+ Accept: "application/json",
+ },
+ });
}
/**
- * Utility function to download a file from the server from blob object.
+ * Export the workflow result to local filesystem. The export is handled by
the browser.
*/
- public saveBlobFile(response: any, defaultFileName: string): void {
- // If the server sets "Content-Disposition: attachment;
filename="someName.csv"" header,
- // we can parse that out. Otherwise just use defaultFileName.
- const dispositionHeader = response.headers.get("Content-Disposition");
- let fileName = defaultFileName;
- if (dispositionHeader) {
- const parsed = contentDisposition.parse(dispositionHeader);
- fileName = parsed.parameters.filename || defaultFileName;
- }
+ public exportWorkflowResultToLocal(
+ exportType: string,
+ workflowId: number,
+ workflowName: string,
+ operators: {
+ id: string;
+ outputType: string;
+ }[],
+ rowIndex: number,
+ columnIndex: number,
+ filename: string,
+ unit: DashboardWorkflowComputingUnit // computing unit for cluster setting
+ ): void {
+ const computingUnitId = unit.computingUnit.cuid;
+ const datasetIds: number[] = [];
+ const requestBody = {
+ exportType,
+ workflowId,
+ workflowName,
+ operators,
+ datasetIds,
+ rowIndex,
+ columnIndex,
+ filename,
+ computingUnitId,
+ };
+ const token = localStorage.getItem(TOKEN_KEY) ?? "";
- const blob = response.body; // the actual file data
- this.fileSaverService.saveAs(blob, fileName);
+ const urlPath =
+ unit && unit.computingUnit.type == "kubernetes" &&
unit.computingUnit?.cuid
+ ?
`${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}/local?cuid=${unit.computingUnit.cuid}`
+ : `${WORKFLOW_EXECUTIONS_API_BASE_URL}/${EXPORT_BASE_URL}/local`;
+
+ const iframe = document.createElement("iframe");
+ iframe.name = "download-iframe";
+ iframe.style.display = "none";
+ document.body.appendChild(iframe);
+
+ const form = document.createElement("form");
+ form.method = "POST";
+ form.action = urlPath;
+ form.target = "download-iframe";
+ form.enctype = "application/x-www-form-urlencoded";
+ form.style.display = "none";
+
+ const requestInput = document.createElement("input");
+ requestInput.type = "hidden";
+ requestInput.name = "request";
+ requestInput.value = JSON.stringify(requestBody);
+ form.appendChild(requestInput);
+
+ const tokenInput = document.createElement("input");
+ tokenInput.type = "hidden";
+ tokenInput.name = "token";
+ tokenInput.value = token;
+ form.appendChild(tokenInput);
+
+ document.body.appendChild(form);
+ form.submit();
+
+ setTimeout(() => {
+ document.body.removeChild(form);
+ document.body.removeChild(iframe);
+ }, IFRAME_TIMEOUT_MS);
}
downloadOperatorsResult(
diff --git
a/core/gui/src/app/workspace/service/workflow-result-export/workflow-result-export.service.ts
b/core/gui/src/app/workspace/service/workflow-result-export/workflow-result-export.service.ts
index 1abad167ec..d57815485e 100644
---
a/core/gui/src/app/workspace/service/workflow-result-export/workflow-result-export.service.ts
+++
b/core/gui/src/app/workspace/service/workflow-result-export/workflow-result-export.service.ts
@@ -306,27 +306,34 @@ export class WorkflowResultExportService {
this.notificationService.loading("Exporting...");
// Make request
- this.downloadService
- .exportWorkflowResult(
+ if (destination === "local") {
+ // Dataset export to local filesystem (download handled by browser)
+ this.downloadService.exportWorkflowResultToLocal(
exportType,
workflowId,
workflowName,
operatorArray,
- [...datasetIds],
rowIndex,
columnIndex,
filename,
- destination,
unit
- )
- .subscribe({
- next: response => {
- if (destination === "local") {
- // "local" => response is a blob
- // We can parse the file name from header or use fallback
- this.downloadService.saveBlobFile(response, filename);
- this.notificationService.info("Files downloaded successfully");
- } else {
+ );
+ } else {
+ // Dataset export to dataset via API call
+ this.downloadService
+ .exportWorkflowResult(
+ exportType,
+ workflowId,
+ workflowName,
+ operatorArray,
+ [...datasetIds],
+ rowIndex,
+ columnIndex,
+ filename,
+ unit
+ )
+ .subscribe({
+ next: response => {
// "dataset" => response is JSON
// The server should return a JSON with {status, message}
const jsonResponse = response as
HttpResponse<ExportWorkflowJsonResponse>;
@@ -336,13 +343,13 @@ export class WorkflowResultExportService {
} else {
this.notificationService.error(responseBody?.message || "An
error occurred during export");
}
- }
- },
- error: (err: unknown) => {
- const errorMessage = (err as any)?.error?.message || (err as
any)?.error || err;
- this.notificationService.error(`An error happened in exporting
operator results: ${errorMessage}`);
- },
- });
+ },
+ error: (err: unknown) => {
+ const errorMessage = (err as any)?.error?.message || (err as
any)?.error || err;
+ this.notificationService.error(`An error happened in exporting
operator results: ${errorMessage}`);
+ },
+ });
+ }
}
/**
diff --git a/deployment/k8s/texera-helmchart/templates/envoy-config.yaml
b/deployment/k8s/texera-helmchart/templates/envoy-config.yaml
index cdce931176..2ba17e63f8 100644
--- a/deployment/k8s/texera-helmchart/templates/envoy-config.yaml
+++ b/deployment/k8s/texera-helmchart/templates/envoy-config.yaml
@@ -66,6 +66,10 @@ data:
- name: envoy.filters.http.ext_authz
typed_config:
"@type":
type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthz
+ with_request_body:
+ # Max allowed by Envoy (uint32): 4,294,967,295
bytes (~4 GiB)
+ max_request_bytes: 4294967295
+ allow_partial_message: true # forward partial
body if larger than limit
http_service:
server_uri:
uri: http://{{ .Release.Name }}-{{
.Values.accessControlService.name }}-svc.{{ .Release.Namespace
}}.svc.cluster.local
diff --git a/deployment/k8s/texera-helmchart/values.yaml
b/deployment/k8s/texera-helmchart/values.yaml
index 7a10c08642..a86c0e71b0 100644
--- a/deployment/k8s/texera-helmchart/values.yaml
+++ b/deployment/k8s/texera-helmchart/values.yaml
@@ -342,7 +342,8 @@ ingressPaths:
pathType: ImplementationSpecific
serviceName: envoy-svc
servicePort: 10000
- - path: /api/executions/result/export
+ - path: /api/executions/result/export/*
+ pathType: ImplementationSpecific
serviceName: envoy-svc
servicePort: 10000
- path: /api
