This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 b44f7db7cb feat(storage): warehouse REST API, Lakekeeper client, and
per-execution injection (#7473)
b44f7db7cb is described below
commit b44f7db7cb0962f06fab1bb4353e6bfa353d50e6
Author: Meng Wang <[email protected]>
AuthorDate: Mon Aug 10 21:50:57 2026 -0700
feat(storage): warehouse REST API, Lakekeeper client, and per-execution
injection (#7473)
### What changes were proposed in this PR?
The backend of the per-user warehouse feature (umbrella #6870), all
gated by `warehouseEnabled` (default off — nothing changes for existing
deployments):
- **Warehouse management API** — `WarehouseResource`: `GET
/warehouse/status` (always answers, so the frontend can hide the
feature), `POST /warehouse` and `DELETE /warehouse/{whid}` (403 while
the flag is off). Create validates the URI-safe name, mints the catalog
name `user-<uid>-<name>`, creates in Lakekeeper first and records the
row after — a failed creation leaves no orphaned state.
- **`LakekeeperClient`** — management-API create (Local flavor: storage
profile on the deployment's own object store, per-warehouse key prefix,
STS off) and **empty-first delete**: drop every table with
`purgeRequested=true`, then the namespaces, then the warehouse entity.
- **Per-execution injection** — `WorkflowExecuteRequest` gains
`warehouseId: Option[Int]`; `WorkflowService.resolveWarehouseName`
checks ownership and refuses an explicit pick while the feature is off
(never a silent fallback — #6930); the resolved name rides
`WorkflowContext.warehouse` into every storage URI (results, runtime
statistics, console messages); the chosen `whid` is recorded on
`workflow_executions` (as `cuid` is today) so the picker can preselect
the workflow's last-used warehouse. `whid` is `ON DELETE SET NULL`:
deleting a warehouse purges its data, never the execution history.
- **Explicit read failure** — `WarehouseReadGuard`: paginating a
`/wh/<name>/…` result while the feature is off fails naming the
warehouse, instead of resolving against the shared warehouse and
surfacing "table not found" (#6930).
### Any related issues, documentation, discussions?
Closes #6932. Part of #6870 (design discussions #5293 and #6040). Builds
on #6944, #7359 and #7386.
### How was this PR tested?
Five specs, 30 cases green locally (`sbt
"WorkflowExecutionService/testOnly *LakekeeperClientSpec
*WarehouseResourceSpec *WarehouseReadGuardSpec
*WorkflowServiceWarehouseSpec *ExecutionsMetadataPersistServiceSpec"`):
the Lakekeeper client runs against an in-process HTTP stub
(create-payload shape; the purge → namespace → warehouse delete order);
the resource spec covers the disabled gate and the create/list/delete
flow on MockTexeraDB with a stubbed client; resolution pins ownership
and the no-silent-fallback rule; the read guard pins the explicit
failure message; the executions spec gains whid recording and the
SET-NULL-on-delete case. A delete-order assertion was deliberately
broken once to confirm it fails red.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
---
.../scheduling/CostBasedScheduleGenerator.scala | 3 +-
.../ExpansionGreedyScheduleGenerator.scala | 3 +-
.../apache/texera/web/TexeraWebApplication.scala | 2 +
.../websocket/request/WorkflowExecuteRequest.scala | 4 +-
.../web/resource/SyncExecutionResource.scala | 16 +-
.../user/warehouse/WarehouseResource.scala | 203 ++++++++++++++++++
.../user/workflow/WorkflowExecutionsResource.scala | 11 +-
.../dashboard/user/workflow/WorkflowResource.scala | 6 +-
.../web/service/ExecutionConsoleService.scala | 7 +-
.../web/service/ExecutionResultService.scala | 3 +
.../texera/web/service/ExecutionStatsService.scala | 3 +-
.../service/ExecutionsMetadataPersistService.scala | 5 +-
.../texera/web/service/LakekeeperClient.scala | 178 ++++++++++++++++
.../texera/web/service/ResultExportService.scala | 6 +-
.../texera/web/service/WarehouseReadGuard.scala | 75 +++++++
.../texera/web/service/WorkflowService.scala | 84 ++++++--
.../user/warehouse/WarehouseResourceSpec.scala | 233 +++++++++++++++++++++
.../workflow/WorkflowExecutionsResourceSpec.scala | 24 ++-
.../ExecutionsMetadataPersistServiceSpec.scala | 55 +++++
.../texera/web/service/LakekeeperClientSpec.scala | 171 +++++++++++++++
.../web/service/WarehouseReadGuardSpec.scala | 106 ++++++++++
.../web/service/WorkflowExecutionServiceSpec.scala | 3 +-
.../web/service/WorkflowServiceWarehouseSpec.scala | 99 +++++++++
.../texera/common/config/StorageConfig.scala | 2 +
.../texera/amber/core/storage/VFSURIFactory.scala | 7 +-
sql/changelog.xml | 5 +
sql/texera_ddl.sql | 4 +-
.../updates/34.sql | 28 +--
28 files changed, 1291 insertions(+), 55 deletions(-)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGenerator.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGenerator.scala
index 57564ef73f..dd43eb99aa 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGenerator.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/CostBasedScheduleGenerator.scala
@@ -194,7 +194,8 @@ class CostBasedScheduleGenerator(
val portBaseURI = createPortBaseURI(
workflowId = workflowContext.workflowId,
executionId = workflowContext.executionId,
- globalPortId = gpid
+ globalPortId = gpid,
+ warehouse = workflowContext.warehouse
)
gpid -> OutputPortConfig(portBaseURI)
}.toMap
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGenerator.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGenerator.scala
index 304e1496f8..4df5ab80fa 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGenerator.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/ExpansionGreedyScheduleGenerator.scala
@@ -334,7 +334,8 @@ class ExpansionGreedyScheduleGenerator(
createPortBaseURI(
workflowId = workflowContext.workflowId,
executionId = workflowContext.executionId,
- globalPortId = outputPortId
+ globalPortId = outputPortId,
+ warehouse = workflowContext.warehouse
)
}
diff --git
a/amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala
b/amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala
index 73e473ba7a..6dd624de60 100644
--- a/amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala
+++ b/amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala
@@ -39,6 +39,7 @@ import
org.apache.texera.web.resource.dashboard.admin.execution.AdminExecutionRe
import org.apache.texera.web.resource.dashboard.admin.user.AdminUserResource
import org.apache.texera.web.resource.dashboard.hub.HubResource
import org.apache.texera.web.resource.dashboard.user.UserResource
+import
org.apache.texera.web.resource.dashboard.user.warehouse.WarehouseResource
import org.apache.texera.web.resource.dashboard.user.project.{
ProjectAccessResource,
ProjectResource,
@@ -160,6 +161,7 @@ class TexeraWebApplication
environment.jersey.register(classOf[UserQuotaResource])
environment.jersey.register(classOf[AIAssistantResource])
environment.jersey.register(classOf[HuggingFaceModelResource])
+ environment.jersey.register(classOf[WarehouseResource])
AuthResource.createAdminUser()
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 0059af1c1a..fac3f67b32 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
@@ -36,5 +36,7 @@ case class WorkflowExecuteRequest(
replayFromExecution: Option[ReplayExecutionInfo], // contains execution
Id, interaction Id.
workflowSettings: WorkflowSettings,
emailNotificationEnabled: Boolean,
- computingUnitId: Int
+ computingUnitId: Int,
+ // The user_warehouse this run writes into; absent = the shared default
warehouse.
+ warehouseId: Option[Int]
) extends TexeraWebSocketRequest
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 afcd7f63a1..cd528a6cf5 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
@@ -50,6 +50,7 @@ import org.apache.texera.dao.SqlServer
import org.apache.texera.dao.jooq.generated.Tables.OPERATOR_EXECUTIONS
import org.apache.texera.common.compiler.model.LogicalPlanPojo
import org.apache.texera.web.model.websocket.request.WorkflowExecuteRequest
+import org.apache.texera.web.service.{WarehouseReadGuard,
WarehouseUnavailableException}
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
@@ -167,7 +168,8 @@ class SyncExecutionResource extends LazyLogging {
WorkflowSettings(dataTransferBatchSize =
ApplicationConfig.defaultDataTransferBatchSize)
),
emailNotificationEnabled = false,
- computingUnitId = computingUnitId
+ computingUnitId = computingUnitId,
+ warehouseId = None
)
workflowService.initExecutionService(
@@ -535,6 +537,8 @@ class SyncExecutionResource extends LazyLogging {
storageUriOption match {
case Some(storageUri) =>
+ // Refuse to read a per-user-warehouse result while the feature is
off (#6930).
+ WarehouseReadGuard.assertReadable(storageUri)
val document = DocumentFactory
.openDocument(storageUri)
._1
@@ -694,6 +698,9 @@ class SyncExecutionResource extends LazyLogging {
("table", None, None, None, None)
}
} catch {
+ // A kill-switch refusal must reach the caller instead of degrading into
an
+ // empty result (#6930); every other failure keeps the existing behavior.
+ case e: WarehouseUnavailableException => throw e
case e: Exception =>
logger.warn(s"Error collecting result for operator $opId:
${e.getMessage}", e)
("table", None, None, None, None)
@@ -769,6 +776,8 @@ class SyncExecutionResource extends LazyLogging {
val uriOption = getConsoleMessageUri(executionId, OperatorIdentity(opId))
uriOption.flatMap { uri =>
+ // Refuse to read per-user-warehouse console messages while the
feature is off (#6930).
+ WarehouseReadGuard.assertReadable(uri)
val document = DocumentFactory
.openDocument(uri)
._1
@@ -793,7 +802,10 @@ class SyncExecutionResource extends LazyLogging {
if (messages.nonEmpty) Some(messages) else None
}
} catch {
- case _: Exception => None
+ // A kill-switch refusal must reach the caller instead of degrading into
an
+ // empty result (#6930); every other failure keeps the existing behavior.
+ case e: WarehouseUnavailableException => throw e
+ case _: Exception => None
}
}
diff --git
a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala
new file mode 100644
index 0000000000..d4aed3ccdc
--- /dev/null
+++
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala
@@ -0,0 +1,203 @@
+/*
+ * 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.web.resource.dashboard.user.warehouse
+
+import com.typesafe.scalalogging.LazyLogging
+import io.dropwizard.auth.Auth
+import org.apache.texera.amber.core.storage.VFSURIFactory
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE
+import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum
+import org.apache.texera.dao.jooq.generated.tables.records.UserWarehouseRecord
+import
org.apache.texera.web.resource.dashboard.user.warehouse.WarehouseResource._
+import org.apache.texera.web.service.LakekeeperClient
+
+import javax.annotation.security.RolesAllowed
+import javax.ws.rs._
+import javax.ws.rs.core.MediaType
+
+object WarehouseResource {
+ private def context =
+ SqlServer
+ .getInstance()
+ .createDSLContext()
+
+ // A warehouse's user-facing name becomes part of the Lakekeeper catalog name
+ // `user-<uid>-<name>`, which in turn becomes a VFS URI path segment — so the
+ // character rule is delegated to VFSURIFactory (the layer that parses it);
the
+ // length cap is this registration layer's own constraint.
+ private[warehouse] def isValidWarehouseName(name: String): Boolean =
+ name.length <= 64 && VFSURIFactory.isValidWarehouseName(name)
+
+ case class DashboardWarehouse(
+ whid: Integer,
+ name: String,
+ warehouseName: String,
+ flavor: String,
+ createdAtMillis: Long
+ )
+
+ private def toDashboardWarehouse(row: UserWarehouseRecord):
DashboardWarehouse =
+ DashboardWarehouse(
+ row.getWhid,
+ row.getName,
+ row.getWarehouseName,
+ row.getFlavor.getLiteral,
+ row.getCreatedAt.toInstant.toEpochMilli
+ )
+
+ case class WarehouseStatus(enabled: Boolean, warehouses:
List[DashboardWarehouse])
+
+ case class CreateWarehouseRequest(name: String)
+}
+
+/**
+ * Per-user warehouse management (#6870): list the feature state and the
caller's
+ * warehouses, create a Local-flavor warehouse on the deployment's own object
store,
+ * and delete one (empty-first in Lakekeeper, purging its data files).
+ *
+ * Everything except `/status` is gated by the warehouse feature flag; the
mutating
+ * endpoints return 403 while it is off. `/status` always answers so the
frontend can
+ * decide whether to show the feature at all.
+ */
+@Path("/warehouse")
+@Produces(Array(MediaType.APPLICATION_JSON))
+class WarehouseResource(client: LakekeeperClient, enabled: Boolean) extends
LazyLogging {
+
+ // Jersey builds the resource through this constructor. The flag is captured
once,
+ // which is equivalent to reading it per call: storage.conf is resolved at
class load
+ // and never changes at runtime.
+ def this() = this(new LakekeeperClient(), StorageConfig.warehouseEnabled)
+
+ @GET
+ @Path("/status")
+ def status(@Auth current_user: SessionUser): WarehouseStatus = {
+ if (!enabled) {
+ return WarehouseStatus(enabled = false, warehouses = List())
+ }
+ val warehouses = context
+ .selectFrom(USER_WAREHOUSE)
+ .where(USER_WAREHOUSE.UID.eq(current_user.getUid))
+ .orderBy(USER_WAREHOUSE.CREATED_AT.asc())
+ .fetch()
+ .map(row => toDashboardWarehouse(row))
+ WarehouseStatus(
+ enabled = true,
+ warehouses = warehouses.toArray(Array[DashboardWarehouse]()).toList
+ )
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ def create(
+ request: CreateWarehouseRequest,
+ @Auth current_user: SessionUser
+ ): DashboardWarehouse = {
+ requireEnabled()
+ val name = Option(request.name).map(_.trim).getOrElse("")
+ if (!isValidWarehouseName(name)) {
+ throw new BadRequestException(
+ "warehouse name must start with a letter or digit and contain only
letters, " +
+ "digits, '-' and '_' (at most 64 characters)"
+ )
+ }
+ val uid = current_user.getUid
+ if (
+ context.fetchExists(
+ context
+ .selectFrom(USER_WAREHOUSE)
+ .where(USER_WAREHOUSE.UID.eq(uid).and(USER_WAREHOUSE.NAME.eq(name)))
+ )
+ ) {
+ throw new WebApplicationException(s"a warehouse named '$name' already
exists", 409)
+ }
+
+ val warehouseName = s"user-$uid-$name"
+ // Create in Lakekeeper first, record after: a failed creation leaves no
orphaned row.
+ val warehouseId =
+ try {
+ client.createWarehouse(warehouseName)
+ } catch {
+ case e: Exception =>
+ throw new WebApplicationException(e.getMessage, 502)
+ }
+
+ val row = context.newRecord(USER_WAREHOUSE)
+ row.setUid(uid)
+ row.setName(name)
+ row.setWarehouseName(warehouseName)
+ row.setLakekeeperWarehouseId(warehouseId)
+ row.setFlavor(UserWarehouseFlavorEnum.local)
+ row.setS3Bucket(StorageConfig.icebergRESTCatalogS3Bucket)
+ row.setS3Endpoint(StorageConfig.s3Endpoint)
+ row.setS3Region(StorageConfig.s3Region)
+ try {
+ row.store()
+ // created_at is filled by the DB default; fetch it back before
serializing.
+ row.refresh()
+ } catch {
+ case e: Exception =>
+ // Compensate: without the row the user could neither list nor delete
the
+ // just-created warehouse, so remove it (it is empty at this point).
+ try {
+ client.deleteWarehouseEmptyFirst(warehouseId)
+ } catch {
+ case cleanup: Exception =>
+ logger.error(
+ s"failed to clean up Lakekeeper warehouse $warehouseId after a
failed create",
+ cleanup
+ )
+ }
+ throw new WebApplicationException(e.getMessage, 500)
+ }
+ toDashboardWarehouse(row)
+ }
+
+ @DELETE
+ @Path("/{whid}")
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ def delete(@PathParam("whid") whid: Integer, @Auth current_user:
SessionUser): Unit = {
+ requireEnabled()
+ val row = context
+ .selectFrom(USER_WAREHOUSE)
+
.where(USER_WAREHOUSE.WHID.eq(whid).and(USER_WAREHOUSE.UID.eq(current_user.getUid)))
+ .fetchOne()
+ if (row == null) {
+ throw new NotFoundException(s"no warehouse with id $whid")
+ }
+ try {
+ client.deleteWarehouseEmptyFirst(row.getLakekeeperWarehouseId)
+ } catch {
+ case e: Exception =>
+ throw new WebApplicationException(e.getMessage, 502)
+ }
+ context
+ .deleteFrom(USER_WAREHOUSE)
+ .where(USER_WAREHOUSE.WHID.eq(whid))
+ .execute()
+ }
+
+ private def requireEnabled(): Unit =
+ if (!enabled) {
+ throw new ForbiddenException("per-user warehouses are disabled in this
deployment")
+ }
+}
diff --git
a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
index cca18443b7..7e847c43c0 100644
---
a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
+++
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
@@ -42,6 +42,7 @@ import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowExecutionsDao
import org.apache.texera.dao.jooq.generated.tables.pojos.{WorkflowExecutions,
User => UserPojo}
import org.apache.texera.web.model.http.request.result.ResultExportRequest
+import org.apache.texera.web.service.WarehouseReadGuard
import
org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource._
import org.apache.texera.web.service.{ExecutionsMetadataPersistService,
ResultExportService}
import org.jooq.DSLContext
@@ -320,6 +321,7 @@ object WorkflowExecutionsResource {
WORKFLOW_EXECUTIONS.EID,
WORKFLOW_EXECUTIONS.VID,
WORKFLOW_EXECUTIONS.CUID,
+ WORKFLOW_EXECUTIONS.WHID,
USER.NAME,
USER.AVATAR,
WORKFLOW_EXECUTIONS.STATUS,
@@ -379,8 +381,9 @@ object WorkflowExecutionsResource {
.where(WORKFLOW_EXECUTIONS.EID.in(eIdsList))
.execute()
- // Clear corresponding Iceberg documents
- uris.foreach { uri =>
+ // Clear corresponding Iceberg documents. While per-user warehouses are
disabled,
+ // cleanup must not reach into them (#6930) — those URIs are skipped.
+ uris.filterNot(WarehouseReadGuard.skipWhileDisabled(_)).foreach { uri =>
try {
DocumentFactory.openDocument(uri)._1.clear()
} catch {
@@ -523,6 +526,7 @@ object WorkflowExecutionsResource {
eId: Integer,
vId: Integer,
cuId: Integer,
+ whId: Integer,
userName: String,
googleAvatar: String,
status: Byte,
@@ -581,6 +585,7 @@ class WorkflowExecutionsResource {
WORKFLOW_EXECUTIONS.EID,
WORKFLOW_EXECUTIONS.VID,
WORKFLOW_EXECUTIONS.CUID,
+ WORKFLOW_EXECUTIONS.WHID,
USER.NAME,
USER.AVATAR,
WORKFLOW_EXECUTIONS.STATUS,
@@ -721,6 +726,8 @@ class WorkflowExecutionsResource {
}
val uri: URI = new URI(uriString)
+ // Refuse to read per-user-warehouse statistics while the feature is off
(#6930).
+ WarehouseReadGuard.assertReadable(uri)
val document = DocumentFactory.openDocument(uri)._1
// Read all records from Iceberg and convert to WorkflowRuntimeStatistics
diff --git
a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
index 72d70d5cf7..eaf4460b29 100644
---
a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
+++
b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
@@ -38,6 +38,7 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
import org.apache.texera.dao.jooq.generated.tables.pojos._
import org.apache.texera.service.util.LargeBinaryManager
import org.apache.texera.web.resource.dashboard.hub.EntityType
+import org.apache.texera.web.service.WarehouseReadGuard
import
org.apache.texera.web.resource.dashboard.hub.HubResource.recordCloneAction
import
org.apache.texera.web.resource.dashboard.user.workflow.WorkflowAccessResource.hasReadAccess
import
org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource._
@@ -654,9 +655,10 @@ class WorkflowResource extends LazyLogging {
// removed. Done after the transaction (like the document cleanup below).
eids.foreach(eid =>
LargeBinaryManager.deleteByExecution(eid.longValue()))
- // Clean up document storage
+ // Clean up document storage. While per-user warehouses are disabled,
cleanup must
+ // not reach into them (#6930) — those URIs are skipped.
try {
- uris.foreach { uri =>
+ uris.filterNot(WarehouseReadGuard.skipWhileDisabled(_)).foreach { uri
=>
try {
val (document, _) = DocumentFactory.openDocument(uri)
document.clear()
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionConsoleService.scala
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionConsoleService.scala
index 3aae5f2b89..55f72c35d8 100644
---
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionConsoleService.scala
+++
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionConsoleService.scala
@@ -144,7 +144,12 @@ class ExecutionConsoleService(
consoleMessageOpIdToWriterMap.getOrElseUpdate(
opId.id, {
val uri = VFSURIFactory
- .createConsoleMessagesURI(workflowContext.workflowId,
workflowContext.executionId, opId)
+ .createConsoleMessagesURI(
+ workflowContext.workflowId,
+ workflowContext.executionId,
+ opId,
+ warehouse = workflowContext.warehouse
+ )
val writer = DocumentFactory
.createDocument(uri, ResultSchema.consoleMessagesSchema)
.writer("console_messages")
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
index c2ffadb9f1..2fa63f6b86 100644
---
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
+++
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
@@ -481,6 +481,9 @@ class ExecutionResultService(
PortIdentity()
)
+ // Refuse to read a per-user-warehouse result while the feature is off
(#6930).
+ storageUriOption.foreach(WarehouseReadGuard.assertReadable(_))
+
storageUriOption match {
case Some(storageUri) =>
val (document, schemaOption) = DocumentFactory.openDocument(storageUri)
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
index f112f4f65d..ac33478451 100644
---
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
+++
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
@@ -78,7 +78,8 @@ class ExecutionStatsService(
val thread = Executors.newSingleThreadExecutor()
val uri = VFSURIFactory.createRuntimeStatisticsURI(
workflowContext.workflowId,
- workflowContext.executionId
+ workflowContext.executionId,
+ warehouse = workflowContext.warehouse
)
val writer = DocumentFactory
.createDocument(uri, ResultSchema.runtimeStatisticsSchema)
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala
index bafe7a178a..b3e4edf985 100644
---
a/amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala
+++
b/amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala
@@ -56,7 +56,8 @@ object ExecutionsMetadataPersistService extends LazyLogging {
uid: Integer,
executionName: String,
environmentVersion: String,
- computingUnitId: Integer
+ computingUnitId: Integer,
+ warehouseId: Option[Int] = None
): ExecutionIdentity = {
// first retrieve the latest version of this workflow
val vid = getLatestVersion(workflowId.id.toInt)
@@ -71,6 +72,8 @@ object ExecutionsMetadataPersistService extends LazyLogging {
// Set computing unit ID if provided
newExecution.setCuid(computingUnitId)
+ // The warehouse this run writes into (#6870); null = the shared default
warehouse.
+ warehouseId.foreach(whid => newExecution.setWhid(whid))
try {
workflowExecutionsDao.insert(newExecution)
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala
b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala
new file mode 100644
index 0000000000..7fdafc0df7
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala
@@ -0,0 +1,178 @@
+/*
+ * 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.web.service
+
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import kong.unirest.Unirest
+import org.apache.texera.common.config.StorageConfig
+
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets
+import java.util.UUID
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters.IteratorHasAsScala
+
+/**
+ * Client for the Lakekeeper APIs used to manage per-user warehouses (#6870).
+ *
+ * Two API families are involved: the **management** API
(`/management/v1/...`) creates and
+ * deletes warehouse entities, and the **catalog** API
(`/catalog/v1/{warehouseId}/...`) lists
+ * and drops the namespaces/tables inside one. The channel is unauthenticated
today;
+ * catalog-side authentication is Phase 2 (#6040).
+ *
+ * @param catalogUri the Iceberg REST catalog uri (ends with `/catalog`),
from which the
+ * management base is derived. Overridable for tests.
+ */
+class LakekeeperClient(catalogUri: String =
StorageConfig.icebergRESTCatalogUri) {
+
+ // Lakekeeper's default project; single-project deployments (ours) use the
nil UUID.
+ private val DefaultProjectId = "00000000-0000-0000-0000-000000000000"
+
+ private val managementBase: String = catalogUri.stripSuffix("/catalog") +
"/management/v1"
+ private val catalogBase: String = catalogUri + "/v1"
+
+ private val mapper = new ObjectMapper()
+
+ private def urlEncode(segment: String): String =
+ URLEncoder.encode(segment, StandardCharsets.UTF_8)
+
+ private def failOn(status: Int, body: String, action: String): Unit =
+ if (status < 200 || status >= 300) {
+ throw new RuntimeException(s"Lakekeeper $action failed (HTTP $status):
$body")
+ }
+
+ /**
+ * Creates a warehouse backed by this deployment's own object store (the
Local flavor):
+ * the storage profile points at the configured MinIO/S3 endpoint and
bucket, with the
+ * platform's static credentials and STS off.
+ *
+ * @return the Lakekeeper-assigned warehouse id.
+ */
+ def createWarehouse(warehouseName: String): UUID = {
+ val payload = mapper.createObjectNode()
+ payload.put("warehouse-name", warehouseName)
+ payload.put("project-id", DefaultProjectId)
+
+ val profile = payload.putObject("storage-profile")
+ profile.put("type", "s3")
+ profile.put("bucket", StorageConfig.icebergRESTCatalogS3Bucket)
+ profile.put("region", StorageConfig.s3Region)
+ profile.put("endpoint", StorageConfig.s3Endpoint)
+ profile.put("path-style-access", true)
+ // The warehouse name doubles as the key prefix, so each warehouse owns a
distinct
+ // subtree of the shared bucket.
+ profile.put("key-prefix", warehouseName)
+ profile.put("flavor", "s3-compat")
+ profile.put("sts-enabled", false)
+
+ val credential = payload.putObject("storage-credential")
+ credential.put("type", "s3")
+ credential.put("credential-type", "access-key")
+ credential.put("aws-access-key-id", StorageConfig.s3Username)
+ credential.put("aws-secret-access-key", StorageConfig.s3Password)
+
+ val response = Unirest
+ .post(s"$managementBase/warehouse")
+ .header("Content-Type", "application/json")
+ .body(payload.toString)
+ .asString()
+ failOn(response.getStatus, response.getBody, s"create warehouse
'$warehouseName'")
+
UUID.fromString(mapper.readTree(response.getBody).get("warehouse-id").asText())
+ }
+
+ /**
+ * Deletes a warehouse **empty-first** (Lakekeeper refuses to drop a
non-empty one):
+ * every table is dropped with `purgeRequested=true` — so the underlying
data files are
+ * purged along with it, matching how execution results are deleted today —
then the
+ * namespaces, then the warehouse entity itself.
+ */
+ def deleteWarehouseEmptyFirst(warehouseId: UUID): Unit = {
+ // 404 anywhere below means the entity is already gone — the goal state.
Tolerating
+ // it makes this method idempotent, so a retry after a partial failure
(e.g. the DB
+ // delete failing after the Lakekeeper delete succeeded) heals instead of
wedging.
+ listNamespaces(warehouseId).foreach { namespace =>
+ listTables(warehouseId, namespace).foreach { table =>
+ val response = Unirest
+ .delete(
+
s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}/tables/${urlEncode(table)}"
+ )
+ .queryString("purgeRequested", "true")
+ .asString()
+ if (response.getStatus != 404) {
+ failOn(response.getStatus, response.getBody, s"drop table
'$namespace.$table'")
+ }
+ }
+ val response = Unirest
+
.delete(s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}")
+ .asString()
+ if (response.getStatus != 404) {
+ failOn(response.getStatus, response.getBody, s"drop namespace
'$namespace'")
+ }
+ }
+ val response =
Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString()
+ if (response.getStatus != 404) {
+ failOn(response.getStatus, response.getBody, "delete warehouse")
+ }
+ }
+
+ /** Top-level namespaces in the warehouse. Texera's execution namespaces are
single-level. */
+ private def listNamespaces(warehouseId: UUID): List[String] =
+ fetchAllPages(s"$catalogBase/$warehouseId/namespaces", "namespaces", "list
namespaces")(parts =>
+ parts.get(0).asText()
+ )
+
+ private def listTables(warehouseId: UUID, namespace: String): List[String] =
+ fetchAllPages(
+ s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}/tables",
+ "identifiers",
+ s"list tables of '$namespace'"
+ )(identifier => identifier.get("name").asText())
+
+ /**
+ * Follows `next-page-token` until exhausted: the Iceberg REST list
endpoints may
+ * return partial pages, and the empty-first delete depends on seeing
everything.
+ * A 404 ends the listing with what was gathered — the entity is already
gone,
+ * which the idempotent delete treats as its goal state.
+ */
+ private def fetchAllPages(url: String, field: String, action: String)(
+ extract: JsonNode => String
+ ): List[String] = {
+ val results = ListBuffer[String]()
+ var pageToken: Option[String] = None
+ var more = true
+ while (more) {
+ val request = Unirest.get(url)
+ pageToken.foreach(request.queryString("pageToken", _))
+ val response = request.asString()
+ if (response.getStatus == 404) {
+ return results.toList
+ }
+ failOn(response.getStatus, response.getBody, action)
+ val tree = mapper.readTree(response.getBody)
+ tree.get(field).iterator().asScala.foreach(node => results +=
extract(node))
+ pageToken = Option(tree.get("next-page-token"))
+ .filterNot(_.isNull)
+ .map(_.asText())
+ .filter(_.nonEmpty)
+ more = pageToken.isDefined
+ }
+ results.toList
+ }
+}
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
b/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
index 4e73a0e655..cf4fb56983 100644
---
a/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
+++
b/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
@@ -490,7 +490,11 @@ class ResultExportService(workflowIdentity:
WorkflowIdentity, computingUnitId: I
)
storageUri
- .map(uri =>
DocumentFactory.openDocument(uri)._1.asInstanceOf[VirtualDocument[Tuple]])
+ .map(uri => {
+ // Refuse to export a per-user-warehouse result while the feature is
off (#6930).
+ WarehouseReadGuard.assertReadable(uri)
+
DocumentFactory.openDocument(uri)._1.asInstanceOf[VirtualDocument[Tuple]]
+ })
.orNull
}
diff --git
a/amber/src/main/scala/org/apache/texera/web/service/WarehouseReadGuard.scala
b/amber/src/main/scala/org/apache/texera/web/service/WarehouseReadGuard.scala
new file mode 100644
index 0000000000..1e98f86749
--- /dev/null
+++
b/amber/src/main/scala/org/apache/texera/web/service/WarehouseReadGuard.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.web.service
+
+import org.apache.texera.amber.core.storage.VFSURIFactory
+import org.apache.texera.common.config.StorageConfig
+
+import java.net.URI
+
+/**
+ * Signals that a read was refused because it targets a per-user warehouse
the deployment
+ * cannot serve. A distinct type so callers with a catch-all (e.g.
SyncExecutionResource,
+ * which degrades other failures into empty results) can let this one
propagate: a kill
+ * switch that silently returns "no data" is exactly the failure mode #6930
forbids.
+ */
+class WarehouseUnavailableException(message: String) extends
IllegalStateException(message)
+
+/**
+ * Guards reads of results that live in a per-user warehouse while the
feature is off (#6930).
+ *
+ * The warehouse switch is a kill switch: turning it off must disable reads
too, and it must
+ * fail *explicitly*. Without this guard a `/wh/<name>/…` URI would resolve
to the shared
+ * default warehouse and surface "table not found" — indistinguishable from
data loss. No data
+ * is lost; re-enabling the switch restores access.
+ */
+object WarehouseReadGuard {
+
+ def assertReadable(
+ uri: URI,
+ enabled: Boolean = StorageConfig.warehouseEnabled
+ ): Unit = {
+ val warehouse = VFSURIFactory.decodeURI(uri).warehouse
+ // A `/wh/` prefix that decodes to no warehouse is unresolvable (illegal
name); opening
+ // it would silently fall back to the shared warehouse — refuse instead
(#6930).
+ if (warehouse.isEmpty && hasWarehousePrefix(uri)) {
+ throw new WarehouseUnavailableException(s"unresolvable warehouse URI:
$uri")
+ }
+ warehouse.filterNot(_ => enabled).foreach { name =>
+ throw new WarehouseUnavailableException(
+ s"this result is stored in warehouse '$name'; per-user warehouses are
disabled in this deployment"
+ )
+ }
+ }
+
+ /**
+ * Whether a cleanup path should skip this URI: while the feature is off,
best-effort
+ * cleanup must not reach into (and delete from) a disabled per-user
warehouse. Loud
+ * failure is wrong here — cleanup runs on unrelated actions — so callers
skip instead.
+ */
+ def skipWhileDisabled(
+ uri: URI,
+ enabled: Boolean = StorageConfig.warehouseEnabled
+ ): Boolean =
+ !enabled && (VFSURIFactory.decodeURI(uri).warehouse.isDefined ||
hasWarehousePrefix(uri))
+
+ private def hasWarehousePrefix(uri: URI): Boolean =
+ Option(uri.getRawPath).exists(_.startsWith("/wh/"))
+}
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 51d5b5677d..a1cc08727b 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
@@ -23,7 +23,9 @@ import com.google.protobuf.timestamp.Timestamp
import com.typesafe.scalalogging.LazyLogging
import io.reactivex.rxjava3.disposables.{CompositeDisposable, Disposable}
import io.reactivex.rxjava3.subjects.BehaviorSubject
-import org.apache.texera.common.config.ApplicationConfig
+import org.apache.texera.common.config.{ApplicationConfig, StorageConfig}
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE
import org.apache.texera.amber.core.WorkflowRuntimeException
import org.apache.texera.amber.core.storage.DocumentFactory
import org.apache.texera.amber.core.storage.result.iceberg.OnIceberg
@@ -67,6 +69,39 @@ import scala.jdk.CollectionConverters.IterableHasAsScala
object WorkflowService {
private val workflowServiceMapping = new ConcurrentHashMap[String,
WorkflowService]()
+
+ /**
+ * Maps an execution's chosen warehouse (`whid`) to its Lakekeeper
warehouse name,
+ * checking that the requesting user owns it. `None` (no explicit pick)
keeps the
+ * shared default warehouse. With warehouses disabled, an explicit pick is
refused
+ * loudly rather than silently routed into the shared warehouse (#6930).
+ */
+ def resolveWarehouseName(
+ warehouseId: Option[Int],
+ uid: Integer,
+ enabled: Boolean = StorageConfig.warehouseEnabled
+ ): Option[String] = {
+ if (!enabled) {
+ warehouseId.foreach(_ =>
+ throw new IllegalArgumentException(
+ "per-user warehouses are disabled in this deployment"
+ )
+ )
+ return None
+ }
+ warehouseId.map(whid => {
+ val row = SqlServer
+ .getInstance()
+ .createDSLContext()
+ .selectFrom(USER_WAREHOUSE)
+ .where(USER_WAREHOUSE.WHID.eq(whid).and(USER_WAREHOUSE.UID.eq(uid)))
+ .fetchOne()
+ if (row == null) {
+ throw new IllegalArgumentException(s"no warehouse with id $whid owned
by this user")
+ }
+ row.getWarehouseName
+ })
+ }
val cleanUpDeadlineInSeconds: Int =
ApplicationConfig.executionStateCleanUpInSecs
def getAllWorkflowServices: Iterable[WorkflowService] =
workflowServiceMapping.values().asScala
@@ -198,6 +233,7 @@ class WorkflowService(
)
val workflowContext: WorkflowContext = createWorkflowContext()
+ workflowContext.warehouse =
WorkflowService.resolveWarehouseName(req.warehouseId, uid)
var coordinatorConf = CoordinatorConfig.default
// clean up results from previous run
@@ -212,7 +248,8 @@ class WorkflowService(
uid,
req.executionName,
convertToJson(req.engineVersion),
- req.computingUnitId
+ req.computingUnitId,
+ req.warehouseId
)
if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
@@ -333,30 +370,37 @@ class WorkflowService(
// Remove references from registry first
WorkflowExecutionsResource.deleteConsoleMessageAndExecutionResultUris(eid)
- // Clean up all result and console message documents
+ // Clean up all result and console message documents. While per-user
warehouses are
+ // disabled, cleanup must not reach into them (#6930) — those URIs are
skipped.
(resultUris ++ consoleMessagesUris).foreach { uri =>
- try DocumentFactory.openDocument(uri)._1.clear()
- catch {
- case error: Throwable =>
- logger.debug(s"Error processing document at $uri:
${error.getMessage}")
- }
+ if (WarehouseReadGuard.skipWhileDisabled(uri)) {
+ logger.info(s"skipping cleanup of $uri: per-user warehouses are
disabled")
+ } else
+ try DocumentFactory.openDocument(uri)._1.clear()
+ catch {
+ case error: Throwable =>
+ logger.debug(s"Error processing document at $uri:
${error.getMessage}")
+ }
}
// Expire any Iceberg snapshots for runtime statistics
WorkflowExecutionsResource.getRuntimeStatsUriByExecutionId(eid).foreach {
uri =>
- try {
- DocumentFactory.openDocument(uri)._1 match {
- case iceberg: OnIceberg => iceberg.expireSnapshots()
- case other =>
- logger.error(
- s"Cannot expire snapshots: document from URI [$uri] is of type
${other.getClass.getName}. " +
- s"Expected an instance of ${classOf[OnIceberg].getName}."
- )
+ if (WarehouseReadGuard.skipWhileDisabled(uri)) {
+ logger.info(s"skipping snapshot expiry of $uri: per-user warehouses
are disabled")
+ } else
+ try {
+ DocumentFactory.openDocument(uri)._1 match {
+ case iceberg: OnIceberg => iceberg.expireSnapshots()
+ case other =>
+ logger.error(
+ s"Cannot expire snapshots: document from URI [$uri] is of type
${other.getClass.getName}. " +
+ s"Expected an instance of ${classOf[OnIceberg].getName}."
+ )
+ }
+ } catch {
+ case error: Throwable =>
+ logger.debug(s"Error processing document at $uri:
${error.getMessage}")
}
- } catch {
- case error: Throwable =>
- logger.debug(s"Error processing document at $uri:
${error.getMessage}")
- }
}
// Delete this execution's large binaries
LargeBinaryManager.deleteByExecution(eid.id)
diff --git
a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResourceSpec.scala
b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResourceSpec.scala
new file mode 100644
index 0000000000..1d3aeb9eed
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResourceSpec.scala
@@ -0,0 +1,233 @@
+/*
+ * 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.web.resource.dashboard.user.warehouse
+
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE
+import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import
org.apache.texera.web.resource.dashboard.user.warehouse.WarehouseResource.CreateWarehouseRequest
+import org.apache.texera.web.service.LakekeeperClient
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import java.util.UUID
+import javax.ws.rs.{
+ BadRequestException,
+ ForbiddenException,
+ NotFoundException,
+ WebApplicationException
+}
+import scala.collection.mutable
+
+/**
+ * Spec for [[WarehouseResource]] (#6932): the disabled-gate behavior, and the
+ * create/list/delete flow against MockTexeraDB with a stubbed
[[LakekeeperClient]].
+ *
+ * The feature flag is a constructor dependency, so the two gate states are
two resource
+ * instances — no global state is touched, and suites cannot interfere with
each other.
+ */
+class WarehouseResourceSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach
+ with MockTexeraDB {
+
+ private val createdNames = mutable.Buffer[String]()
+ private val deletedIds = mutable.Buffer[UUID]()
+ private val stubWarehouseId = UUID.randomUUID()
+
+ @volatile private var createFailure: Option[Exception] = None
+ @volatile private var deleteFailure: Option[Exception] = None
+
+ private val stubClient: LakekeeperClient = new LakekeeperClient() {
+ override def createWarehouse(warehouseName: String): UUID = {
+ createFailure.foreach(throw _)
+ createdNames += warehouseName
+ stubWarehouseId
+ }
+ override def deleteWarehouseEmptyFirst(warehouseId: UUID): Unit = {
+ deleteFailure.foreach(throw _)
+ deletedIds += warehouseId
+ }
+ }
+
+ private val resource = new WarehouseResource(stubClient, enabled = true)
+ private val disabledResource = new WarehouseResource(stubClient, enabled =
false)
+ private var sessionUser: SessionUser = _
+ private var otherUser: SessionUser = _
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+
+ val userDao = new UserDao(getDSLContext.configuration())
+ val user = new User
+ user.setName("warehouse_spec_user")
+ user.setEmail(s"user_${UUID.randomUUID()}@example.com")
+ userDao.insert(user)
+ sessionUser = new SessionUser(user)
+
+ val other = new User
+ other.setName("warehouse_spec_other")
+ other.setEmail(s"user_${UUID.randomUUID()}@example.com")
+ userDao.insert(other)
+ otherUser = new SessionUser(other)
+ }
+
+ override protected def afterAll(): Unit = closeConnectionPool()
+
+ override protected def beforeEach(): Unit = {
+ createFailure = None
+ deleteFailure = None
+ createdNames.clear()
+ deletedIds.clear()
+ getDSLContext.deleteFrom(USER_WAREHOUSE).execute()
+ }
+
+ //
---------------------------------------------------------------------------
+ // Disabled gate
+ //
---------------------------------------------------------------------------
+
+ "status" should "report disabled with no warehouses while the flag is off"
in {
+ val status = disabledResource.status(sessionUser)
+ status.enabled shouldBe false
+ status.warehouses shouldBe empty
+ }
+
+ "create and delete" should "be refused while the flag is off" in {
+ a[ForbiddenException] should be thrownBy
+ disabledResource.create(CreateWarehouseRequest("mybucket"), sessionUser)
+ a[ForbiddenException] should be thrownBy disabledResource.delete(1,
sessionUser)
+ }
+
+ //
---------------------------------------------------------------------------
+ // Create / list / delete
+ //
---------------------------------------------------------------------------
+
+ "create" should "create in Lakekeeper, record the row, and mint
user-<uid>-<name>" in {
+ val created = resource.create(CreateWarehouseRequest("mybucket"),
sessionUser)
+
+ created.name shouldBe "mybucket"
+ created.warehouseName shouldBe s"user-${sessionUser.getUid}-mybucket"
+ created.flavor shouldBe "local"
+ createdNames.toList shouldBe List(s"user-${sessionUser.getUid}-mybucket")
+
+ val status = resource.status(sessionUser)
+ status.enabled shouldBe true
+ status.warehouses.map(_.whid) shouldBe List(created.whid)
+ }
+
+ it should "reject an unsafe or duplicate name" in {
+ a[BadRequestException] should be thrownBy
+ resource.create(CreateWarehouseRequest("a/b"), sessionUser)
+
+ resource.create(CreateWarehouseRequest("dup"), sessionUser)
+ val conflict = intercept[WebApplicationException] {
+ resource.create(CreateWarehouseRequest("dup"), sessionUser)
+ }
+ conflict.getResponse.getStatus shouldBe 409
+ }
+
+ "delete" should "empty the warehouse in Lakekeeper and remove the row" in {
+ val created = resource.create(CreateWarehouseRequest("doomed"),
sessionUser)
+
+ resource.delete(created.whid, sessionUser)
+
+ deletedIds.toList shouldBe List(stubWarehouseId)
+ resource.status(sessionUser).warehouses shouldBe empty
+ }
+
+ "a failed record write after Lakekeeper creation" should "compensate by
deleting the warehouse" in {
+ // Pre-claim the catalog name under the other user so our store() trips
the global
+ // UNIQUE(warehouse_name) after the (stubbed) Lakekeeper creation
succeeded.
+ val squatter = getDSLContext.newRecord(USER_WAREHOUSE)
+ squatter.setUid(otherUser.getUid)
+ squatter.setName("unrelated")
+ squatter.setWarehouseName(s"user-${sessionUser.getUid}-boom")
+ squatter.setLakekeeperWarehouseId(UUID.randomUUID())
+ squatter.setFlavor(
+ org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum.local
+ )
+ squatter.store()
+
+ val error = intercept[WebApplicationException] {
+ resource.create(CreateWarehouseRequest("boom"), sessionUser)
+ }
+ error.getResponse.getStatus shouldBe 500
+ // The just-created (empty) Lakekeeper warehouse must not be orphaned.
+ deletedIds.toList shouldBe List(stubWarehouseId)
+ resource.status(sessionUser).warehouses shouldBe empty
+ }
+
+ "a Lakekeeper failure" should "surface as 502 on create and delete" in {
+ createFailure = Some(new RuntimeException("Lakekeeper create failed (HTTP
500): boom"))
+ val createError = intercept[WebApplicationException] {
+ resource.create(CreateWarehouseRequest("unlucky"), sessionUser)
+ }
+ createError.getResponse.getStatus shouldBe 502
+ resource.status(sessionUser).warehouses shouldBe empty
+
+ createFailure = None
+ val created = resource.create(CreateWarehouseRequest("undeletable"),
sessionUser)
+ deleteFailure = Some(new RuntimeException("Lakekeeper delete failed (HTTP
500): boom"))
+ val deleteError = intercept[WebApplicationException] {
+ resource.delete(created.whid, sessionUser)
+ }
+ deleteError.getResponse.getStatus shouldBe 502
+ // The row must survive a failed Lakekeeper delete, so the user can retry.
+ resource.status(sessionUser).warehouses.map(_.whid) shouldBe
List(created.whid)
+ }
+
+ "a failed compensation" should "be logged and still surface the original
failure" in {
+ val squatter = getDSLContext.newRecord(USER_WAREHOUSE)
+ squatter.setUid(otherUser.getUid)
+ squatter.setName("unrelated-2")
+ squatter.setWarehouseName(s"user-${sessionUser.getUid}-doublefault")
+ squatter.setLakekeeperWarehouseId(UUID.randomUUID())
+ squatter.setFlavor(
+ org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum.local
+ )
+ squatter.store()
+
+ deleteFailure = Some(new RuntimeException("cleanup also failed"))
+ val error = intercept[WebApplicationException] {
+ resource.create(CreateWarehouseRequest("doublefault"), sessionUser)
+ }
+ error.getResponse.getStatus shouldBe 500
+ resource.status(sessionUser).warehouses shouldBe empty
+ }
+
+ "the no-arg constructor" should "read the configured flag and client" in {
+ // Jersey instantiates the resource reflectively via this constructor;
storage.conf
+ // keeps the feature off by default, so this exercises it without a
Lakekeeper call.
+ new WarehouseResource().status(sessionUser).enabled shouldBe
StorageConfig.warehouseEnabled
+ }
+
+ it should "not let a user delete someone else's warehouse" in {
+ val created = resource.create(CreateWarehouseRequest("mine"), sessionUser)
+
+ a[NotFoundException] should be thrownBy resource.delete(created.whid,
otherUser)
+ resource.status(sessionUser).warehouses.map(_.whid) shouldBe
List(created.whid)
+ }
+}
diff --git
a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
index a0a3622e06..48973699cf 100644
---
a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
@@ -30,6 +30,7 @@ import
org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps
import org.apache.texera.auth.SessionUser
import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum
import org.apache.texera.dao.jooq.generated.Tables._
import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum,
WorkflowComputingUnitTypeEnum}
import org.apache.texera.dao.jooq.generated.tables.daos.{
@@ -217,7 +218,8 @@ class WorkflowExecutionsResourceSpec
startOffsetMillis: Long = 0L,
lastUpdateOffsetMillis: Option[Long] = None,
cuid: Integer = null,
- runtimeStatsUri: String = null
+ runtimeStatsUri: String = null,
+ whid: Integer = null
): WorkflowExecutions = {
val execution = new WorkflowExecutions
execution.setVid(testVersion.getVid)
@@ -232,6 +234,7 @@ class WorkflowExecutionsResourceSpec
execution.setName(name)
execution.setEnvironmentVersion("test-env-1.0")
execution.setCuid(cuid)
+ execution.setWhid(whid)
execution.setRuntimeStatsUri(runtimeStatsUri)
workflowExecutionsDao.insert(execution)
execution
@@ -988,6 +991,25 @@ class WorkflowExecutionsResourceSpec
assert(entry.name == "second")
}
+ it should "expose the execution's warehouse (whId) for last-used
preselection" in {
+ grantReadAccess()
+ val warehouse = getDSLContext.newRecord(USER_WAREHOUSE)
+ warehouse.setUid(testUser.getUid)
+ warehouse.setName("latest-entry-warehouse")
+
warehouse.setWarehouseName(s"user-${testUser.getUid}-latest-entry-warehouse")
+ warehouse.setLakekeeperWarehouseId(UUID.randomUUID())
+ warehouse.setFlavor(UserWarehouseFlavorEnum.local)
+ warehouse.store()
+
+ insertExecution(name = "warehouse-run", whid = warehouse.getWhid)
+ val entry = resource.retrieveLatestExecutionEntry(testWorkflowWid,
session(testUser))
+ assert(entry.whId == warehouse.getWhid)
+
+ insertExecution(name = "default-run")
+ val defaultEntry = resource.retrieveLatestExecutionEntry(testWorkflowWid,
session(testUser))
+ assert(defaultEntry.whId == null)
+ }
+
"retrieveInteractionHistory" should "return an empty list when the user
lacks read access" in {
val result =
resource.retrieveInteractionHistory(
diff --git
a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
index ca704861fe..bc49d67e56 100644
---
a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
+++
b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
@@ -25,6 +25,7 @@ import
org.apache.texera.amber.engine.common.Utils.maptoStatusCode
import
org.apache.texera.amber.engine.common.executionruntimestate.ExecutionMetadataStore
import org.apache.texera.dao.MockTexeraDB
import org.apache.texera.dao.jooq.generated.Tables._
+import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum
import org.apache.texera.dao.jooq.generated.tables.daos.{
UserDao,
WorkflowComputingUnitDao,
@@ -222,6 +223,60 @@ class ExecutionsMetadataPersistServiceSpec
workflowExecutionsDao.fetchByVid(seededVid).size() shouldBe before
}
+ it should "record the chosen warehouse and leave it null when none is
picked" in {
+ val row = getDSLContext.newRecord(USER_WAREHOUSE)
+ row.setUid(testUid)
+ row.setName("exec-spec-warehouse")
+ row.setWarehouseName(s"user-$testUid-exec-spec-warehouse")
+ row.setLakekeeperWarehouseId(UUID.randomUUID())
+ row.setFlavor(UserWarehouseFlavorEnum.local)
+ row.store()
+
+ val withWarehouse = ExecutionsMetadataPersistService.insertNewExecution(
+ WorkflowIdentity(testWid.toLong),
+ testUid,
+ executionName = "warehouse-run",
+ environmentVersion = "env-4",
+ computingUnitId = seededCuid,
+ warehouseId = Some(row.getWhid)
+ )
+ workflowExecutionsDao.fetchOneByEid(withWarehouse.id.toInt).getWhid
shouldBe row.getWhid
+
+ val withoutWarehouse = ExecutionsMetadataPersistService.insertNewExecution(
+ WorkflowIdentity(testWid.toLong),
+ testUid,
+ executionName = "default-run",
+ environmentVersion = "env-4",
+ computingUnitId = seededCuid
+ )
+ workflowExecutionsDao.fetchOneByEid(withoutWarehouse.id.toInt).getWhid
shouldBe null
+ }
+
+ it should "keep execution history when its warehouse is deleted (whid SET
NULL)" in {
+ val row = getDSLContext.newRecord(USER_WAREHOUSE)
+ row.setUid(testUid)
+ row.setName("doomed-warehouse")
+ row.setWarehouseName(s"user-$testUid-doomed-warehouse")
+ row.setLakekeeperWarehouseId(UUID.randomUUID())
+ row.setFlavor(UserWarehouseFlavorEnum.local)
+ row.store()
+
+ val id = ExecutionsMetadataPersistService.insertNewExecution(
+ WorkflowIdentity(testWid.toLong),
+ testUid,
+ executionName = "history-run",
+ environmentVersion = "env-5",
+ computingUnitId = seededCuid,
+ warehouseId = Some(row.getWhid)
+ )
+
+
getDSLContext.deleteFrom(USER_WAREHOUSE).where(USER_WAREHOUSE.WHID.eq(row.getWhid)).execute()
+
+ val stored = workflowExecutionsDao.fetchOneByEid(id.id.toInt)
+ stored should not be null
+ stored.getWhid shouldBe null
+ }
+
// -- tryGetExistingExecution
------------------------------------------------
"tryGetExistingExecution" should "return Some(row) for a known eid" in {
diff --git
a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala
b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala
new file mode 100644
index 0000000000..ac047c590c
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala
@@ -0,0 +1,171 @@
+/*
+ * 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.web.service
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.sun.net.httpserver.{HttpExchange, HttpServer}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import java.net.InetSocketAddress
+import java.nio.charset.StandardCharsets
+import java.util.UUID
+import scala.collection.mutable
+
+/**
+ * Exercises [[LakekeeperClient]] end-to-end against an in-process HTTP stub
standing in
+ * for Lakekeeper (same approach as AsterixDBConnUtilSpec). No network
dependency; the
+ * stub binds port 0 to pick a free ephemeral port.
+ */
+class LakekeeperClientSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with BeforeAndAfterEach {
+
+ private val mapper = new ObjectMapper()
+ private val warehouseId = UUID.randomUUID()
+
+ // Every request in arrival order, as "METHOD path?query" plus the body for
POSTs.
+ private val requests = mutable.Buffer[String]()
+ @volatile private var lastCreateBody: String = ""
+
+ private val server: HttpServer = HttpServer.create(new InetSocketAddress(0),
0)
+
+ private def record(exchange: HttpExchange): String = {
+ val query = Option(exchange.getRequestURI.getQuery).map("?" +
_).getOrElse("")
+ val line = s"${exchange.getRequestMethod}
${exchange.getRequestURI.getPath}$query"
+ requests.synchronized { requests += line }
+ line
+ }
+
+ private def respond(exchange: HttpExchange, status: Int, body: String): Unit
= {
+ val bytes = body.getBytes(StandardCharsets.UTF_8)
+ exchange.getResponseHeaders.add("Content-Type", "application/json")
+ exchange.sendResponseHeaders(status, bytes.length)
+ exchange.getResponseBody.write(bytes)
+ exchange.close()
+ }
+
+ server.createContext(
+ "/management/v1/warehouse",
+ (exchange: HttpExchange) => {
+ record(exchange)
+ if (exchange.getRequestMethod == "POST") {
+ lastCreateBody = new String(exchange.getRequestBody.readAllBytes(),
StandardCharsets.UTF_8)
+ respond(exchange, 201, s"""{"warehouse-id": "$warehouseId"}""")
+ } else {
+ respond(exchange, 200, "{}")
+ }
+ }
+ )
+ server.createContext(
+ s"/catalog/v1/$warehouseId/namespaces",
+ (exchange: HttpExchange) => {
+ val line = record(exchange)
+ val query = Option(exchange.getRequestURI.getQuery).getOrElse("")
+ (exchange.getRequestMethod, exchange.getRequestURI.getPath) match {
+ case ("GET", path) if path.endsWith("/namespaces") =>
+ respond(exchange, 200, """{"namespaces":
[["operator-port-result"]]}""")
+ // The tables listing is served in TWO pages so the client's
next-page-token
+ // loop is pinned: missing the second page would leave the warehouse
non-empty.
+ case ("GET", path) if path.endsWith("/tables") &&
!query.contains("pageToken") =>
+ respond(
+ exchange,
+ 200,
+ """{"identifiers": [{"namespace": ["operator-port-result"],
"name": "wid_1_eid_2_result"}],
+ |"next-page-token": "page-2"}""".stripMargin
+ )
+ case ("GET", path) if path.endsWith("/tables") =>
+ respond(
+ exchange,
+ 200,
+ """{"identifiers": [{"namespace": ["operator-port-result"],
"name": "wid_1_eid_3_result"}]}"""
+ )
+ case ("DELETE", _) =>
+ respond(exchange, 204, "")
+ case _ =>
+ respond(exchange, 500, s"""{"error": "unexpected $line"}""")
+ }
+ }
+ )
+ private val erroringWarehouseId = UUID.randomUUID()
+ server.createContext(
+ s"/catalog/v1/$erroringWarehouseId/namespaces",
+ (exchange: HttpExchange) => {
+ record(exchange)
+ respond(exchange, 500, """{"error": "internal"}""")
+ }
+ )
+ server.start()
+
+ private val client = new LakekeeperClient(
+ s"http://localhost:${server.getAddress.getPort}/catalog";
+ )
+
+ override protected def beforeEach(): Unit = {
+ requests.synchronized { requests.clear() }
+ lastCreateBody = ""
+ }
+
+ override protected def afterAll(): Unit = server.stop(0)
+
+ "createWarehouse" should "post the Local storage profile and return the
assigned id" in {
+ client.createWarehouse("user-7-mybucket") shouldBe warehouseId
+
+ val payload = mapper.readTree(lastCreateBody)
+ payload.get("warehouse-name").asText() shouldBe "user-7-mybucket"
+ val profile = payload.get("storage-profile")
+ profile.get("type").asText() shouldBe "s3"
+ profile.get("sts-enabled").asBoolean() shouldBe false
+ profile.get("path-style-access").asBoolean() shouldBe true
+ // Each warehouse owns its own key prefix inside the shared bucket.
+ profile.get("key-prefix").asText() shouldBe "user-7-mybucket"
+ payload.get("storage-credential").get("credential-type").asText() shouldBe
"access-key"
+ }
+
+ "deleteWarehouseEmptyFirst" should "purge every page of tables, then
namespaces, then the warehouse" in {
+ client.deleteWarehouseEmptyFirst(warehouseId)
+
+ val deletes = requests.synchronized {
requests.filter(_.startsWith("DELETE")).toList }
+ deletes shouldBe List(
+ s"DELETE
/catalog/v1/$warehouseId/namespaces/operator-port-result/tables/wid_1_eid_2_result?purgeRequested=true",
+ s"DELETE
/catalog/v1/$warehouseId/namespaces/operator-port-result/tables/wid_1_eid_3_result?purgeRequested=true",
+ s"DELETE /catalog/v1/$warehouseId/namespaces/operator-port-result",
+ s"DELETE /management/v1/warehouse/$warehouseId"
+ )
+ }
+
+ it should "treat an already-gone warehouse as deleted (404s tolerated,
idempotent)" in {
+ val goneId = UUID.randomUUID()
+ // No stub context matches this warehouse → every request 404s. A retry
after a
+ // partial failure must heal rather than wedge on \"not found\".
+ noException should be thrownBy client.deleteWarehouseEmptyFirst(goneId)
+ }
+
+ it should "surface a Lakekeeper failure with its status and body" in {
+ val error = intercept[RuntimeException] {
+ client.deleteWarehouseEmptyFirst(erroringWarehouseId)
+ }
+ error.getMessage should include("Lakekeeper")
+ error.getMessage should include("500")
+ }
+}
diff --git
a/amber/src/test/scala/org/apache/texera/web/service/WarehouseReadGuardSpec.scala
b/amber/src/test/scala/org/apache/texera/web/service/WarehouseReadGuardSpec.scala
new file mode 100644
index 0000000000..0046507797
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/service/WarehouseReadGuardSpec.scala
@@ -0,0 +1,106 @@
+/*
+ * 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.web.service
+
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.net.URI
+import org.scalatest.matchers.should.Matchers
+
+class WarehouseReadGuardSpec extends AnyFlatSpec with Matchers {
+
+ private def uri(path: String) = new URI(s"vfs://$path")
+
+ "assertReadable" should "pass results that live in the shared default
warehouse" in {
+ noException should be thrownBy
+ WarehouseReadGuard.assertReadable(uri("/wid/1/eid/2/result"), enabled =
false)
+ noException should be thrownBy
+ WarehouseReadGuard.assertReadable(uri("/wid/1/eid/2/result"), enabled =
true)
+ }
+
+ it should "pass warehouse results while the feature is enabled" in {
+ noException should be thrownBy
+ WarehouseReadGuard.assertReadable(
+ uri("/wh/user-7-mybucket/wid/1/eid/2/result"),
+ enabled = true
+ )
+ }
+
+ it should "refuse a warehouse result explicitly while the feature is off" in
{
+ // Naming the situation matters: resolving the URI against the shared
warehouse
+ // would surface "table not found" — indistinguishable from data loss
(#6930).
+ val error = intercept[IllegalStateException] {
+ WarehouseReadGuard.assertReadable(
+ uri("/wh/user-7-mybucket/wid/1/eid/2/result"),
+ enabled = false
+ )
+ }
+ error.getMessage should include("user-7-mybucket")
+ error.getMessage should include("disabled")
+ }
+
+ it should "refuse an unresolvable /wh/ prefix instead of falling back
silently" in {
+ // decodeURI reports None for an illegal warehouse name; opening such a
URI would
+ // silently resolve against the shared warehouse — in either flag state.
+ an[IllegalStateException] should be thrownBy
+ WarehouseReadGuard.assertReadable(uri("/wh/a%2Fb/wid/1/eid/2/result"),
enabled = true)
+ an[IllegalStateException] should be thrownBy
+ WarehouseReadGuard.assertReadable(uri("/wh/a%2Fb/wid/1/eid/2/result"),
enabled = false)
+ }
+
+ it should "refuse with a typed exception callers can let through their
catch-alls" in {
+ // SyncExecutionResource degrades other failures into empty results; the
kill-switch
+ // refusal is typed so it can be rethrown there instead of vanishing
(#6930).
+ a[WarehouseUnavailableException] should be thrownBy
+ WarehouseReadGuard.assertReadable(
+ uri("/wh/user-7-mybucket/wid/1/eid/2/result"),
+ enabled = false
+ )
+ a[WarehouseUnavailableException] should be thrownBy
+ WarehouseReadGuard.assertReadable(uri("/wh/a%2Fb/wid/1/eid/2/result"),
enabled = true)
+ }
+
+ "the default enabled argument" should "read the configured flag" in {
+ // Covers the default-argument methods; a default-warehouse URI passes and
is
+ // never skipped in either flag state, so this is deterministic regardless
of
+ // the configured value.
+ noException should be thrownBy
WarehouseReadGuard.assertReadable(uri("/wid/1/eid/2/result"))
+ WarehouseReadGuard.skipWhileDisabled(uri("/wid/1/eid/2/result")) shouldBe
false
+ }
+
+ "skipWhileDisabled" should "skip exactly the warehouse-scoped URIs while the
feature is off" in {
+ WarehouseReadGuard.skipWhileDisabled(
+ uri("/wh/user-7-mybucket/wid/1/eid/2/result"),
+ enabled = false
+ ) shouldBe true
+ WarehouseReadGuard.skipWhileDisabled(
+ uri("/wh/a%2Fb/wid/1/eid/2/result"),
+ enabled = false
+ ) shouldBe true
+ WarehouseReadGuard.skipWhileDisabled(
+ uri("/wid/1/eid/2/result"),
+ enabled = false
+ ) shouldBe false
+ WarehouseReadGuard.skipWhileDisabled(
+ uri("/wh/user-7-mybucket/wid/1/eid/2/result"),
+ enabled = true
+ ) shouldBe false
+ }
+}
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 2edc102234..d8af24de4b 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
@@ -68,7 +68,8 @@ class WorkflowExecutionServiceSpec extends AnyFlatSpec with
Matchers {
replayFromExecution = None,
workflowSettings = WorkflowSettings(),
emailNotificationEnabled = false,
- computingUnitId = 0
+ computingUnitId = 0,
+ warehouseId = None
)
new WorkflowExecutionService(
null,
diff --git
a/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceWarehouseSpec.scala
b/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceWarehouseSpec.scala
new file mode 100644
index 0000000000..ef7d8cfb7c
--- /dev/null
+++
b/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceWarehouseSpec.scala
@@ -0,0 +1,99 @@
+/*
+ * 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.web.service
+
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE
+import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum
+import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.UUID
+
+/**
+ * Spec for [[WorkflowService.resolveWarehouseName]] (#6932): the ownership
check and the
+ * never-silently-fall-back rule for an execution's chosen warehouse.
+ */
+class WorkflowServiceWarehouseSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with MockTexeraDB {
+
+ private var ownerUid: Integer = _
+ private var intruderUid: Integer = _
+ private var whid: Integer = _
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+
+ val userDao = new UserDao(getDSLContext.configuration())
+ val owner = new User
+ owner.setName("resolve_spec_owner")
+ owner.setEmail(s"user_${UUID.randomUUID()}@example.com")
+ userDao.insert(owner)
+ ownerUid = owner.getUid
+
+ val intruder = new User
+ intruder.setName("resolve_spec_intruder")
+ intruder.setEmail(s"user_${UUID.randomUUID()}@example.com")
+ userDao.insert(intruder)
+ intruderUid = intruder.getUid
+
+ val row = getDSLContext.newRecord(USER_WAREHOUSE)
+ row.setUid(ownerUid)
+ row.setName("mybucket")
+ row.setWarehouseName(s"user-$ownerUid-mybucket")
+ row.setLakekeeperWarehouseId(UUID.randomUUID())
+ row.setFlavor(UserWarehouseFlavorEnum.local)
+ row.store()
+ whid = row.getWhid
+ }
+
+ override protected def afterAll(): Unit = closeConnectionPool()
+
+ "resolveWarehouseName" should "keep the shared default warehouse when
nothing is picked" in {
+ WorkflowService.resolveWarehouseName(None, ownerUid, enabled = true)
shouldBe None
+ WorkflowService.resolveWarehouseName(None, ownerUid, enabled = false)
shouldBe None
+ }
+
+ it should "resolve an owned warehouse to its Lakekeeper name" in {
+ WorkflowService.resolveWarehouseName(Some(whid), ownerUid, enabled = true)
shouldBe
+ Some(s"user-$ownerUid-mybucket")
+ }
+
+ it should "refuse another user's warehouse" in {
+ val error = intercept[IllegalArgumentException] {
+ WorkflowService.resolveWarehouseName(Some(whid), intruderUid, enabled =
true)
+ }
+ error.getMessage should include(whid.toString)
+ }
+
+ it should "refuse an explicit pick while warehouses are disabled" in {
+ // Never route the run into the shared warehouse silently (#6930).
+ val error = intercept[IllegalArgumentException] {
+ WorkflowService.resolveWarehouseName(Some(whid), ownerUid, enabled =
false)
+ }
+ error.getMessage should include("disabled")
+ }
+}
diff --git
a/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
b/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
index 2f9a33c291..9a98108e19 100644
---
a/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
+++
b/common/config/src/main/scala/org/apache/texera/common/config/StorageConfig.scala
@@ -39,6 +39,8 @@ object StorageConfig {
val icebergRESTCatalogUri: String =
conf.getString("storage.iceberg.catalog.rest.uri")
val icebergRESTCatalogWarehouseName: String =
conf.getString("storage.iceberg.catalog.rest.warehouse-name")
+ val icebergRESTCatalogS3Bucket: String =
+ conf.getString("storage.iceberg.catalog.rest.s3-bucket")
// Iceberg Postgres specifics
val icebergPostgresCatalogUriWithoutScheme: String =
diff --git
a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala
b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala
index 79e36e77f4..d0e6145983 100644
---
a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala
+++
b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala
@@ -61,7 +61,12 @@ object VFSURIFactory {
// invariant the URI layer itself depends on.
private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r
- private def isValidWarehouseName(name: String): Boolean =
+ /**
+ * Whether `name` is safe as a URI path segment. Public so registration-time
+ * validation (WarehouseResource) enforces the exact rule this layer parses
by,
+ * instead of keeping a drifting copy.
+ */
+ def isValidWarehouseName(name: String): Boolean =
warehouseNamePattern.pattern.matcher(name).matches()
// Warehouse is carried as a leading `/wh/<name>` path segment so a storage
URI
diff --git a/sql/changelog.xml b/sql/changelog.xml
index 586618f39b..05d1bb8ded 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -79,6 +79,11 @@
<sqlFile path="sql/updates/33.sql"/>
</changeSet>
+ <!-- Record the per-execution warehouse on workflow_executions (#6870) -->
+ <changeSet id="34" author="mengw15">
+ <sqlFile path="sql/updates/34.sql"/>
+ </changeSet>
+
<!-- example changeSet
<changeSet id="1" author="author">
<sqlFile path="sql/updates/1.sql"/>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index 45ade5b910..f991bf9b62 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -298,9 +298,11 @@ CREATE TABLE IF NOT EXISTS workflow_executions
log_location TEXT,
runtime_stats_uri TEXT,
runtime_stats_size BIGINT DEFAULT 0,
+ whid INT,
FOREIGN KEY (vid) REFERENCES workflow_version(vid) ON DELETE CASCADE,
FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE,
- FOREIGN KEY (cuid) REFERENCES workflow_computing_unit(cuid) ON DELETE
CASCADE
+ FOREIGN KEY (cuid) REFERENCES workflow_computing_unit(cuid) ON DELETE
CASCADE,
+ FOREIGN KEY (whid) REFERENCES user_warehouse(whid) ON DELETE SET NULL
);
-- public_project
diff --git
a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala
b/sql/updates/34.sql
similarity index 52%
copy from
amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala
copy to sql/updates/34.sql
index 0059af1c1a..38286c75af 100644
---
a/amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala
+++ b/sql/updates/34.sql
@@ -17,24 +17,16 @@
* under the License.
*/
-package org.apache.texera.web.model.websocket.request
+\c texera_db
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize
-import org.apache.texera.amber.core.workflow.WorkflowSettings
-import org.apache.texera.common.compiler.model.LogicalPlanPojo
+SET search_path TO texera_db;
-case class ReplayExecutionInfo(
- @JsonDeserialize(contentAs = classOf[java.lang.Long])
- eid: Long,
- interaction: String
-)
+BEGIN;
-case class WorkflowExecuteRequest(
- executionName: String,
- engineVersion: String,
- logicalPlan: LogicalPlanPojo,
- replayFromExecution: Option[ReplayExecutionInfo], // contains execution
Id, interaction Id.
- workflowSettings: WorkflowSettings,
- emailNotificationEnabled: Boolean,
- computingUnitId: Int
-) extends TexeraWebSocketRequest
+-- Record the per-execution warehouse (#6870): which user_warehouse an
execution wrote
+-- into, mirroring cuid. SET NULL on warehouse deletion — dropping a warehouse
purges
+-- its data but must not erase execution history.
+ALTER TABLE workflow_executions
+ ADD COLUMN whid INT REFERENCES user_warehouse (whid) ON DELETE SET NULL;
+
+COMMIT;
