This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7473-133da7bbd6b27c7fba17a6ab894793745bf0c873 in repository https://gitbox.apache.org/repos/asf/texera.git
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;
