mengw15 commented on code in PR #7473:
URL: https://github.com/apache/texera/pull/7473#discussion_r3754377177
##########
amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala:
##########
@@ -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)
Review Comment:
Good catch — not intentional. Done in 1b06dc2bc: the URI resolution and the
guard now run before the try, so a kill-switch refusal propagates instead of
degrading into an empty result; other failures keep the existing catch-all.
##########
amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala:
##########
@@ -769,6 +773,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)
Review Comment:
Same fix in 1b06dc2bc — hoisted above the try.
##########
amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.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.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 it
is
+ // restricted to the same characters VFSURIFactory accepts for a warehouse
name.
+ private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r
+
+ private[warehouse] def isValidWarehouseName(name: String): Boolean =
+ name.length <= 64 && warehouseNamePattern.pattern.matcher(name).matches()
+
+ case class DashboardWarehouse(
+ whid: Integer,
+ name: String,
+ warehouseName: String,
+ flavor: String,
+ createdAtMillis: Long
+ )
+
+ 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
[[StorageConfig.warehouseEnabled]]; the
+ * mutating endpoints return 403 while the flag 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) extends LazyLogging {
+
+ def this() = this(new LakekeeperClient())
+
+ @GET
+ @Path("/status")
+ def status(@Auth current_user: SessionUser): WarehouseStatus = {
+ if (!StorageConfig.warehouseEnabled) {
+ 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 =>
+ DashboardWarehouse(
+ row.getWhid,
+ row.getName,
+ row.getWarehouseName,
+ row.getFlavor.getLiteral,
+ row.getCreatedAt.toInstant.toEpochMilli
+ )
+ )
+ 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)
+ }
+ DashboardWarehouse(
+ row.getWhid,
+ row.getName,
+ row.getWarehouseName,
+ row.getFlavor.getLiteral,
+ row.getCreatedAt.toInstant.toEpochMilli
+ )
+ }
+
+ @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()
Review Comment:
Done in 1b06dc2bc — `deleteWarehouseEmptyFirst` is now idempotent: 404s
(already-gone tables / namespaces / warehouse) are treated as the goal state,
so a retry after a failed DB delete heals instead of wedging on "not found".
##########
amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.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.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 it
is
+ // restricted to the same characters VFSURIFactory accepts for a warehouse
name.
+ private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r
+
+ private[warehouse] def isValidWarehouseName(name: String): Boolean =
+ name.length <= 64 && warehouseNamePattern.pattern.matcher(name).matches()
Review Comment:
Done in 1b06dc2bc — `VFSURIFactory.isValidWarehouseName` is public now and
the resource delegates to it, keeping only its own ≤64 length cap.
##########
amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.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.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 it
is
+ // restricted to the same characters VFSURIFactory accepts for a warehouse
name.
+ private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r
+
+ private[warehouse] def isValidWarehouseName(name: String): Boolean =
+ name.length <= 64 && warehouseNamePattern.pattern.matcher(name).matches()
+
+ case class DashboardWarehouse(
+ whid: Integer,
+ name: String,
+ warehouseName: String,
+ flavor: String,
+ createdAtMillis: Long
+ )
+
+ 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
[[StorageConfig.warehouseEnabled]]; the
+ * mutating endpoints return 403 while the flag 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) extends LazyLogging {
+
+ def this() = this(new LakekeeperClient())
+
+ @GET
+ @Path("/status")
+ def status(@Auth current_user: SessionUser): WarehouseStatus = {
+ if (!StorageConfig.warehouseEnabled) {
+ 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 =>
+ DashboardWarehouse(
+ row.getWhid,
+ row.getName,
+ row.getWarehouseName,
+ row.getFlavor.getLiteral,
+ row.getCreatedAt.toInstant.toEpochMilli
+ )
Review Comment:
Done in 1b06dc2bc — extracted `toDashboardWarehouse`; both sites use it.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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 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.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 = {
+ 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()
+ failOn(response.getStatus, response.getBody, s"drop table
'$namespace.$table'")
+ }
+ val response = Unirest
+
.delete(s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}")
+ .asString()
+ failOn(response.getStatus, response.getBody, s"drop namespace
'$namespace'")
+ }
+ val response =
Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString()
+ 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] = {
+ val response =
Unirest.get(s"$catalogBase/$warehouseId/namespaces").asString()
+ failOn(response.getStatus, response.getBody, "list namespaces")
+ mapper
+ .readTree(response.getBody)
+ .get("namespaces")
+ .iterator()
+ .asScala
+ .map(parts => parts.get(0).asText())
+ .toList
+ }
Review Comment:
Good catch — the Iceberg REST list endpoints do paginate. Done in 1b06dc2bc:
both listings follow `next-page-token` to exhaustion (full snapshot first, then
delete by name), pinned by a two-page stub case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
