feat(amber): decouple the Lakekeeper catalog name from the user-facing name

amber/src/main/scala/org/apache/texera/web/model/websocket/request/WorkflowExecuteRequest.scala

@@ -37,6 +37,8 @@ case class WorkflowExecuteRequest(
     workflowSettings: WorkflowSettings,
     emailNotificationEnabled: Boolean,
     computingUnitId: Int,
-    // The user_warehouse this run writes into; absent = the shared default warehouse.
+    // The user_warehouse row this run writes into; absent = the shared default
+    // warehouse. (Lakekeeper's own warehouse id is a different, UUID-typed
+    // identifier, always named `lakekeeperWarehouseId`.)
     warehouseId: Option[Int]
 ) extends TexeraWebSocketRequest

amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala

@@ -30,6 +30,7 @@ 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 org.jooq.impl.DSL
 
 import javax.annotation.security.RolesAllowed
 import javax.ws.rs._
@@ -41,17 +42,24 @@ object WarehouseResource {
       .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.
+  // The display name no longer reaches Lakekeeper or any URI — the catalog name is
+  // minted from uid and whid — but it stays under VFSURIFactory's character rule
+  // rather than growing a second charset; the length cap is this registration
+  // layer's own constraint.
   private[warehouse] def isValidWarehouseName(name: String): Boolean =
     name.length <= 64 && VFSURIFactory.isValidWarehouseName(name)
 
+  // The Lakekeeper catalog name minted for a warehouse row: stable, short, and
+  // mapping straight back to the row.
+  private[warehouse] def lakekeeperWarehouseName(uid: Integer, whid: Integer): String =
+    s"user-$uid-$whid"
+
   case class DashboardWarehouse(
       whid: Integer,
+      // The display name, unique per user and free to change.
       name: String,
-      warehouseName: String,
+      // The Lakekeeper catalog name, `user-<uid>-<whid>`.
+      lakekeeperWarehouseName: String,
       flavor: String,
       createdAtMillis: Long
   )
@@ -60,7 +68,7 @@ object WarehouseResource {
     DashboardWarehouse(
       row.getWhid,
       row.getName,
-      row.getWarehouseName,
+      row.getLakekeeperWarehouseName,
       row.getFlavor.getLiteral,
       row.getCreatedAt.toInstant.toEpochMilli
     )
@@ -131,21 +139,35 @@ class WarehouseResource(client: LakekeeperClient, enabled: Boolean) extends Lazy
       throw new WebApplicationException(s"a warehouse named '$name' already exists", 409)
     }
 
-    val warehouseName = s"user-$uid-$name"
+    // Never derive the catalog name from `name`: it is also the S3 key prefix and a
+    // component of every result URI an execution wrote, so a name-derived one could
+    // never change again. Drawing the id up front keeps the creation order below
+    // intact. The sequence is looked up rather than named literally -- its generated
+    // name is not a stable contract.
+    val whid: Integer = context.fetchValue(
+      DSL.field(
+        "nextval(pg_get_serial_sequence({0}, {1}))",
+        classOf[Integer],
+        DSL.inline(s"${USER_WAREHOUSE.getSchema.getName}.${USER_WAREHOUSE.getName}"),
+        DSL.inline(USER_WAREHOUSE.WHID.getName)
+      )
+    )
+    val mintedName = lakekeeperWarehouseName(uid, whid)
     // Create in Lakekeeper first, record after: a failed creation leaves no orphaned row.
-    val warehouseId =
+    val lakekeeperWarehouseId =
       try {
-        client.createWarehouse(warehouseName)
+        client.createWarehouse(mintedName)
       } catch {
         case e: Exception =>
           throw new WebApplicationException(e.getMessage, 502)
       }
 
     val row = context.newRecord(USER_WAREHOUSE)
+    row.setWhid(whid)
     row.setUid(uid)
     row.setName(name)
-    row.setWarehouseName(warehouseName)
-    row.setLakekeeperWarehouseId(warehouseId)
+    row.setLakekeeperWarehouseName(mintedName)
+    row.setLakekeeperWarehouseId(lakekeeperWarehouseId)
     row.setFlavor(UserWarehouseFlavorEnum.local)
     row.setS3Bucket(StorageConfig.icebergRESTCatalogS3Bucket)
     row.setS3Endpoint(StorageConfig.s3Endpoint)
@@ -159,11 +181,11 @@ class WarehouseResource(client: LakekeeperClient, enabled: Boolean) extends Lazy
         // 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)
+          client.deleteWarehouseEmptyFirst(lakekeeperWarehouseId)
         } catch {
           case cleanup: Exception =>
             logger.error(
-              s"failed to clean up Lakekeeper warehouse $warehouseId after a failed create",
+              s"failed to clean up Lakekeeper warehouse $lakekeeperWarehouseId after a failed create",
               cleanup
             )
         }

amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala

@@ -73,7 +73,7 @@ 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))
+    warehouseId.foreach(id => newExecution.setWhid(id))
 
     try {
       workflowExecutionsDao.insert(newExecution)

amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala

@@ -33,7 +33,7 @@ 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
+  * deletes warehouse entities, and the **catalog** API (`/catalog/v1/{lakekeeperWarehouseId}/...`) lists
   * and drops the namespaces/tables inside one. The channel is unauthenticated today;
   * catalog-side authentication is Phase 2 (#6040).
   *
@@ -103,15 +103,15 @@ class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri)
     * purged along with it, matching how execution results are deleted today — then the
     * namespaces, then the warehouse entity itself.
     */
-  def deleteWarehouseEmptyFirst(warehouseId: UUID): Unit = {
+  def deleteWarehouseEmptyFirst(lakekeeperWarehouseId: 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 =>
+    listNamespaces(lakekeeperWarehouseId).foreach { namespace =>
+      listTables(lakekeeperWarehouseId, namespace).foreach { table =>
         val response = Unirest
           .delete(
-            s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}/tables/${urlEncode(table)}"
+            s"$catalogBase/$lakekeeperWarehouseId/namespaces/${urlEncode(namespace)}/tables/${urlEncode(table)}"
           )
           .queryString("purgeRequested", "true")
           .asString()
@@ -120,27 +120,29 @@ class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri)
         }
       }
       val response = Unirest
-        .delete(s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}")
+        .delete(s"$catalogBase/$lakekeeperWarehouseId/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()
+    val response = Unirest.delete(s"$managementBase/warehouse/$lakekeeperWarehouseId").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 listNamespaces(lakekeeperWarehouseId: UUID): List[String] =
+    fetchAllPages(
+      s"$catalogBase/$lakekeeperWarehouseId/namespaces",
+      "namespaces",
+      "list namespaces"
+    )(parts => parts.get(0).asText())
 
-  private def listTables(warehouseId: UUID, namespace: String): List[String] =
+  private def listTables(lakekeeperWarehouseId: UUID, namespace: String): List[String] =
     fetchAllPages(
-      s"$catalogBase/$warehouseId/namespaces/${urlEncode(namespace)}/tables",
+      s"$catalogBase/$lakekeeperWarehouseId/namespaces/${urlEncode(namespace)}/tables",
       "identifiers",
       s"list tables of '$namespace'"
     )(identifier => identifier.get("name").asText())

amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala

@@ -71,12 +71,12 @@ 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
+    * Maps an execution's chosen warehouse (its user_warehouse row id) 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(
+  def resolveLakekeeperWarehouseName(
       warehouseId: Option[Int],
       uid: Integer,
       enabled: Boolean = StorageConfig.warehouseEnabled
@@ -89,17 +89,17 @@ object WorkflowService {
       )
       return None
     }
-    warehouseId.map(whid => {
+    warehouseId.map(id => {
       val row = SqlServer
         .getInstance()
         .createDSLContext()
         .selectFrom(USER_WAREHOUSE)
-        .where(USER_WAREHOUSE.WHID.eq(whid).and(USER_WAREHOUSE.UID.eq(uid)))
+        .where(USER_WAREHOUSE.WHID.eq(id).and(USER_WAREHOUSE.UID.eq(uid)))
         .fetchOne()
       if (row == null) {
-        throw new IllegalArgumentException(s"no warehouse with id $whid owned by this user")
+        throw new IllegalArgumentException(s"no warehouse with id $id owned by this user")
       }
-      row.getWarehouseName
+      row.getLakekeeperWarehouseName
     })
   }
   val cleanUpDeadlineInSeconds: Int = ApplicationConfig.executionStateCleanUpInSecs
@@ -233,7 +233,7 @@ class WorkflowService(
     )
 
     val workflowContext: WorkflowContext = createWorkflowContext()
-    workflowContext.warehouse = WorkflowService.resolveWarehouseName(req.warehouseId, uid)
+    workflowContext.warehouse = WorkflowService.resolveLakekeeperWarehouseName(req.warehouseId, uid)
     var coordinatorConf = CoordinatorConfig.default
 
     // clean up results from previous run

amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResourceSpec.scala

@@ -22,10 +22,14 @@ 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.jooq.impl.DSL
 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.resource.dashboard.user.warehouse.WarehouseResource.{
+  lakekeeperWarehouseName,
+  CreateWarehouseRequest
+}
 import org.apache.texera.web.service.LakekeeperClient
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -67,9 +71,9 @@ class WarehouseResourceSpec
       createdNames += warehouseName
       stubWarehouseId
     }
-    override def deleteWarehouseEmptyFirst(warehouseId: UUID): Unit = {
+    override def deleteWarehouseEmptyFirst(lakekeeperWarehouseId: UUID): Unit = {
       deleteFailure.foreach(throw _)
-      deletedIds += warehouseId
+      deletedIds += lakekeeperWarehouseId
     }
   }
 
@@ -125,19 +129,34 @@ class WarehouseResourceSpec
   // Create / list / delete
   // ---------------------------------------------------------------------------
 
-  "create" should "create in Lakekeeper, record the row, and mint user-<uid>-<name>" in {
+  "create" should "create in Lakekeeper, record the row, and mint user-<uid>-<whid>" in {
     val created = resource.create(CreateWarehouseRequest("mybucket"), sessionUser)
 
     created.name shouldBe "mybucket"
-    created.warehouseName shouldBe s"user-${sessionUser.getUid}-mybucket"
+    created.lakekeeperWarehouseName shouldBe lakekeeperWarehouseName(
+      sessionUser.getUid,
+      created.whid
+    )
+    created.lakekeeperWarehouseName should not include "mybucket"
     created.flavor shouldBe "local"
-    createdNames.toList shouldBe List(s"user-${sessionUser.getUid}-mybucket")
+    createdNames.toList shouldBe List(lakekeeperWarehouseName(sessionUser.getUid, created.whid))
 
     val status = resource.status(sessionUser)
     status.enabled shouldBe true
     status.warehouses.map(_.whid) shouldBe List(created.whid)
   }
 
+  it should "mint a fresh catalog name when the same display name is reused" in {
+    // A reused catalog name would let the new warehouse inherit the old one's storage
+    // path, since stored result URIs embed it.
+    val first = resource.create(CreateWarehouseRequest("recycled"), sessionUser)
+    resource.delete(first.whid, sessionUser)
+    val second = resource.create(CreateWarehouseRequest("recycled"), sessionUser)
+
+    second.name shouldBe first.name
+    second.lakekeeperWarehouseName should not be first.lakekeeperWarehouseName
+  }
+
   it should "reject an unsafe or duplicate name" in {
     a[BadRequestException] should be thrownBy
       resource.create(CreateWarehouseRequest("a/b"), sessionUser)
@@ -158,18 +177,34 @@ class WarehouseResourceSpec
     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.
+  // Pre-claims the catalog name create() will mint next, under the other user, so the
+  // caller's store() trips the global UNIQUE(lakekeeper_warehouse_name) after the
+  // (stubbed) Lakekeeper creation succeeded: one sequence number is taken for the
+  // squatter itself (set explicitly, so storing it consumes nothing further), and the
+  // squatter registers the name belonging to the next.
+  private def squatOnNextMintedName(squatterDisplayName: String): Unit = {
     val squatter = getDSLContext.newRecord(USER_WAREHOUSE)
     squatter.setUid(otherUser.getUid)
-    squatter.setName("unrelated")
-    squatter.setWarehouseName(s"user-${sessionUser.getUid}-boom")
+    squatter.setName(squatterDisplayName)
+    val takenWhid = getDSLContext.fetchValue(
+      DSL.field(
+        "nextval(pg_get_serial_sequence({0}, {1}))",
+        classOf[Integer],
+        DSL.inline(s"${USER_WAREHOUSE.getSchema.getName}.${USER_WAREHOUSE.getName}"),
+        DSL.inline(USER_WAREHOUSE.WHID.getName)
+      )
+    )
+    squatter.setWhid(takenWhid)
+    squatter.setLakekeeperWarehouseName(lakekeeperWarehouseName(sessionUser.getUid, takenWhid + 1))
     squatter.setLakekeeperWarehouseId(UUID.randomUUID())
     squatter.setFlavor(
       org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum.local
     )
     squatter.store()
+  }
+
+  "a failed record write after Lakekeeper creation" should "compensate by deleting the warehouse" in {
+    squatOnNextMintedName("unrelated")
 
     val error = intercept[WebApplicationException] {
       resource.create(CreateWarehouseRequest("boom"), sessionUser)
@@ -200,15 +235,7 @@ class WarehouseResourceSpec
   }
 
   "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()
+    squatOnNextMintedName("unrelated-2")
 
     deleteFailure = Some(new RuntimeException("cleanup also failed"))
     val error = intercept[WebApplicationException] {

amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala

@@ -1136,7 +1136,7 @@ class WorkflowExecutionsResourceSpec
     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.setLakekeeperWarehouseName(s"user-${testUser.getUid}-latest-entry-warehouse")
     warehouse.setLakekeeperWarehouseId(UUID.randomUUID())
     warehouse.setFlavor(UserWarehouseFlavorEnum.local)
     warehouse.store()

amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala

@@ -227,7 +227,7 @@ class ExecutionsMetadataPersistServiceSpec
     val row = getDSLContext.newRecord(USER_WAREHOUSE)
     row.setUid(testUid)
     row.setName("exec-spec-warehouse")
-    row.setWarehouseName(s"user-$testUid-exec-spec-warehouse")
+    row.setLakekeeperWarehouseName(s"user-$testUid-exec-spec-warehouse")
     row.setLakekeeperWarehouseId(UUID.randomUUID())
     row.setFlavor(UserWarehouseFlavorEnum.local)
     row.store()
@@ -256,7 +256,7 @@ class ExecutionsMetadataPersistServiceSpec
     val row = getDSLContext.newRecord(USER_WAREHOUSE)
     row.setUid(testUid)
     row.setName("doomed-warehouse")
-    row.setWarehouseName(s"user-$testUid-doomed-warehouse")
+    row.setLakekeeperWarehouseName(s"user-$testUid-doomed-warehouse")
     row.setLakekeeperWarehouseId(UUID.randomUUID())
     row.setFlavor(UserWarehouseFlavorEnum.local)
     row.store()