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-7754-875aa72add3b7cb7ce23f024ce09ae1e983d39b7 in repository https://gitbox.apache.org/repos/asf/texera.git
commit d663833a7b5ac73010645fd1f254177c98d07891 Author: Meng Wang <[email protected]> AuthorDate: Thu Aug 20 18:51:05 2026 +0000 feat(amber): decouple the Lakekeeper catalog name from the user-facing name (#7754) ### What changes were proposed in this PR? `WarehouseResource` derived the Lakekeeper catalog name from the user-facing name: `user-<uid>-<name>`. That one string is simultaneously the Lakekeeper warehouse identifier, the REST catalog URL prefix, the S3 key prefix, and a component of every result URI stored by executions that wrote into the warehouse — so the display name was frozen at creation and a warehouse could never be renamed. Computing units rename freely precisely because their name is pure display metadata; `cuid` is the identity everywhere else. - **Derive the catalog name from the row id**: `user-<uid>-<whid>`. The id is drawn from the table's own sequence **before** the Lakekeeper call, so the creation order is unchanged — Lakekeeper first, DB row after, with the existing compensating delete — and no schema or nullability change is needed. The sequence is resolved through `pg_get_serial_sequence` rather than named literally, because the generated sequence name is not a stable contract (the jOOQ output already carries both `user_warehouse_whid_seq` and `..._seq1` from a re-created table). - **`name` stays the per-user-unique display name**, now free to change; a rename endpoint mirroring computing-unit rename becomes a straightforward follow-up. - **Rename the column**: `user_warehouse.warehouse_name` → `lakekeeper_warehouse_name`, matching its sibling `lakekeeper_warehouse_id`. - **Carry that naming into the code**: `resolveWarehouseName` → `resolveLakekeeperWarehouseName` and the DTO field `warehouseName` → `lakekeeperWarehouseName`, so the column, the method and the field all say which name they hold. `WorkflowContext.warehouse` keeps its name — it feeds a chain of parameters all called `warehouse` (`VFSURIFactory` here and in Python), none of which sit beside a display name. - **Tell the two warehouse ids apart**: `warehouseId` named both the `user_warehouse` row id we pick (an `Int`) and the id Lakekeeper assigns its own entity (a `UUID`), so a call site had to be read type-first. They are now `whid` and `lakekeeperWarehouseId`, matching their columns; anything prefixed `lakekeeper` is an identifier owned by that system. `WorkflowExecuteRequest.warehouseId` becomes `whid` as part of this — it is a wire field, but nothing sends it yet (the warehouse frontend is an unmerged draft), so renaming now costs nothing while doing it later would change a live contract. **Deviation from the issue, and why.** The issue proposed `user-<uid>-<8 random hex>`, on the reasoning that deriving from `whid` "would need the DB row before the Lakekeeper create — an order flip plus a nullable column". Taking the id from the sequence up front avoids both, so that cost does not apply. Doing so also removes the collision-retry path the random suffix required: a 32-bit suffix collides often enough to need one, and that retry would have to recognise Lakekeeper's name-conflict error — the same brittle response-parsing #7742 just had to harden. A sequence-derived name cannot collide, and `user-7-42` points straight at `whid = 42` when tracing storage back to a row. **The migration is guarded.** `texera_ddl.sql` now creates the column under its new name, so on a fresh database the column to rename was never there — and the local-dev replay of unrecorded change sets tolerates only "already exists", so an unguarded `RENAME COLUMN` aborts `up` with "column warehouse_name does not exist". `38.sql` guards the rename with the same existence check `33.sql` uses, verified against a live database both ways: a schema already holding `lakekeeper_warehouse_name` is a clean no-op, one still holding `warehouse_name` is renamed. **On "zero migration".** No *data* migration is needed — the table is empty in every deployment while the flag is off — but the column rename still needs a schema migration (`sql/updates/38.sql`): `texera_ddl.sql` is `CREATE TABLE IF NOT EXISTS`, so an existing database keeps the old column, and jOOQ generates its code from the live database. Without the migration, existing databases would generate `WAREHOUSE_NAME` and fail to compile against this change. ### Any related issues, documentation, discussions? Closes #7753. Part of #6870, follow-up to #6932. Worth settling while the flag is off everywhere: once real data exists under name-derived prefixes, this becomes a migration project. ### How was this PR tested? - `WarehouseResourceSpec` now asserts the catalog name equals `user-<uid>-<whid>` and **does not contain the display name**, pinning the decoupling itself. - New case: deleting a warehouse and recreating it with the same display name mints a **different** catalog name — a reused name would let a new warehouse inherit an old one's storage path. - The two compensation cases previously created their conflict by pre-claiming `user-<uid>-<name>`, which this change makes unreachable. They now draw an id from the sequence, set it explicitly on the squatter row (so storing it consumes nothing further), and squat on the next one — so they still exercise the `UNIQUE` conflict, and additionally pin the "catalog name = uid + sequence id" rule. - `WarehouseResourceSpec`, `WorkflowServiceWarehouseSpec`, `ExecutionsMetadataPersistServiceSpec`, `LakekeeperClientSpec` and `WorkflowServiceSpec` run locally: **38/38 passed**, plus `UserWarehouseSpec` **4/4**, against a database migrated with `sql/updates/38.sql` and jOOQ regenerated from it; `WorkflowExecutionService/scalafmtCheck` (main + Test) passes. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-8) --- .../user/warehouse/WarehouseResource.scala | 53 ++++++++++++++----- .../service/ExecutionsMetadataPersistService.scala | 2 +- .../texera/web/service/LakekeeperClient.scala | 28 +++++----- .../texera/web/service/WorkflowService.scala | 16 +++--- .../user/warehouse/WarehouseResourceSpec.scala | 59 ++++++++++++++-------- .../workflow/WorkflowExecutionsResourceSpec.scala | 2 +- .../ExecutionsMetadataPersistServiceSpec.scala | 4 +- .../texera/web/service/LakekeeperClientSpec.scala | 18 +++---- .../texera/web/service/WorkflowServiceSpec.scala | 2 +- .../web/service/WorkflowServiceWarehouseSpec.scala | 16 +++--- .../org/apache/texera/dao/UserWarehouseSpec.scala | 22 ++++++-- .../amber/core/workflow/WorkflowContext.scala | 2 + sql/changelog.xml | 5 ++ sql/texera_ddl.sql | 20 ++++---- sql/updates/38.sql | 52 +++++++++++++++++++ 15 files changed, 210 insertions(+), 91 deletions(-) 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 index cb196ceb2f..dafc212fd0 100644 --- 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 @@ -32,6 +32,8 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.User 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.DSLContext +import org.jooq.impl.DSL import javax.annotation.security.RolesAllowed import javax.ws.rs._ @@ -44,17 +46,37 @@ 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" + + // Draws the id the next insert will use from the table's sequence, without consuming + // an insert. The sequence is looked up rather than named literally -- its generated + // name is not a stable contract. + private[warehouse] def drawNextWhid(context: DSLContext): 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) + ) + ) + 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, // Owner display info, mirroring DashboardWorkflowComputingUnit: today every @@ -79,7 +101,7 @@ object WarehouseResource { DashboardWarehouse( row.getWhid, row.getName, - row.getWarehouseName, + row.getLakekeeperWarehouseName, row.getFlavor.getLiteral, row.getCreatedAt.toInstant.toEpochMilli, ownerName = owner._1, @@ -163,21 +185,26 @@ 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. + val whid: Integer = drawNextWhid(context) + 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) @@ -191,11 +218,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 ) } 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 b3e4edf985..f7e5b86ac8 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 @@ -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) 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 index ed95d8e4f4..cd2d48669b 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala @@ -55,7 +55,7 @@ object LakekeeperClient { * 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). * @@ -130,15 +130,15 @@ class LakekeeperClient( * 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() @@ -147,7 +147,7 @@ class LakekeeperClient( } } 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'") @@ -168,7 +168,7 @@ class LakekeeperClient( maxDelayMillis = purgeWait.maxDelayMillis, shouldRetry = _.isInstanceOf[UnfinishedTasksConflictException] ) { - val response = Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString() + val response = Unirest.delete(s"$managementBase/warehouse/$lakekeeperWarehouseId").asString() response.getStatus match { case 404 => // already gone — the idempotent goal state case 409 if isUnfinishedTasksBody(response.getBody) => @@ -193,14 +193,16 @@ class LakekeeperClient( } /** 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()) 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 a1cc08727b..9ea3e48c12 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 @@ -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 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 index 58d4f2e9c8..b7510980ad 100644 --- 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 @@ -25,7 +25,10 @@ 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.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 +70,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 } } @@ -126,13 +129,17 @@ 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)) created.ownerName shouldBe "warehouse_spec_user" // The fixture user has no avatar set; the DTO carries null, not "" (#7743). created.ownerAvatar shouldBe null @@ -142,6 +149,17 @@ class WarehouseResourceSpec 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 "resolve each entry's owner name and avatar for the dashboard" in { // Mirrors DashboardWorkflowComputingUnit: the UI binds to the entry's owner, // not the session user, so shared warehouses will render the right person. @@ -173,18 +191,27 @@ 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 = WarehouseResource.drawNextWhid(getDSLContext) + 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) @@ -215,15 +242,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] { 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 a0e14bcc8d..7dc5742a18 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 @@ -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() 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 bc49d67e56..620d8312b3 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 @@ -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() 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 index 1451720eb0..2f69ce2eb6 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala @@ -43,7 +43,7 @@ class LakekeeperClientSpec with BeforeAndAfterEach { private val mapper = new ObjectMapper() - private val warehouseId = UUID.randomUUID() + private val lakekeeperWarehouseId = UUID.randomUUID() // Every request in arrival order, as "METHOD path?query" plus the body for POSTs. private val requests = mutable.Buffer[String]() @@ -91,7 +91,7 @@ class LakekeeperClientSpec case ("POST", _) => lastCreateBody = new String(exchange.getRequestBody.readAllBytes(), StandardCharsets.UTF_8) - respond(exchange, 201, s"""{"warehouse-id": "$warehouseId"}""") + respond(exchange, 201, s"""{"warehouse-id": "$lakekeeperWarehouseId"}""") case ("DELETE", path) if path.endsWith(racingWarehouseId.toString) => if (deleteAttempts(racingWarehouseId) <= 2) respond(exchange, 409, unfinishedTasksBody) else respond(exchange, 204, "") @@ -112,7 +112,7 @@ class LakekeeperClientSpec } ) server.createContext( - s"/catalog/v1/$warehouseId/namespaces", + s"/catalog/v1/$lakekeeperWarehouseId/namespaces", (exchange: HttpExchange) => { val line = record(exchange) val query = Option(exchange.getRequestURI.getQuery).getOrElse("") @@ -170,7 +170,7 @@ class LakekeeperClientSpec 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 + client.createWarehouse("user-7-mybucket") shouldBe lakekeeperWarehouseId val payload = mapper.readTree(lastCreateBody) payload.get("warehouse-name").asText() shouldBe "user-7-mybucket" @@ -184,14 +184,14 @@ class LakekeeperClientSpec } "deleteWarehouseEmptyFirst" should "purge every page of tables, then namespaces, then the warehouse" in { - client.deleteWarehouseEmptyFirst(warehouseId) + client.deleteWarehouseEmptyFirst(lakekeeperWarehouseId) 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" + s"DELETE /catalog/v1/$lakekeeperWarehouseId/namespaces/operator-port-result/tables/wid_1_eid_2_result?purgeRequested=true", + s"DELETE /catalog/v1/$lakekeeperWarehouseId/namespaces/operator-port-result/tables/wid_1_eid_3_result?purgeRequested=true", + s"DELETE /catalog/v1/$lakekeeperWarehouseId/namespaces/operator-port-result", + s"DELETE /management/v1/warehouse/$lakekeeperWarehouseId" ) } diff --git a/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala index 719c8ead75..210d821b87 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceSpec.scala @@ -150,7 +150,7 @@ import scala.collection.mutable.ArrayBuffer * knowing if it ever acquires a reader: the completion diff handler runs only while something * consumes the metadata store's websocket-event stream, so the snapshot silently does not * happen when no client is connected. - * - `resolveWarehouseName`, owned by `WorkflowServiceWarehouseSpec`. + * - `resolveLakekeeperWarehouseName`, owned by `WorkflowServiceWarehouseSpec`. */ class WorkflowServiceSpec extends AnyFlatSpec 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 index ef7d8cfb7c..07c7d45176 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceWarehouseSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/WorkflowServiceWarehouseSpec.scala @@ -31,7 +31,7 @@ import org.scalatest.matchers.should.Matchers import java.util.UUID /** - * Spec for [[WorkflowService.resolveWarehouseName]] (#6932): the ownership check and the + * Spec for [[WorkflowService.resolveLakekeeperWarehouseName]] (#6932): the ownership check and the * never-silently-fall-back rule for an execution's chosen warehouse. */ class WorkflowServiceWarehouseSpec @@ -63,7 +63,7 @@ class WorkflowServiceWarehouseSpec val row = getDSLContext.newRecord(USER_WAREHOUSE) row.setUid(ownerUid) row.setName("mybucket") - row.setWarehouseName(s"user-$ownerUid-mybucket") + row.setLakekeeperWarehouseName(s"user-$ownerUid-mybucket") row.setLakekeeperWarehouseId(UUID.randomUUID()) row.setFlavor(UserWarehouseFlavorEnum.local) row.store() @@ -72,19 +72,19 @@ class WorkflowServiceWarehouseSpec 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 + "resolveLakekeeperWarehouseName" should "keep the shared default warehouse when nothing is picked" in { + WorkflowService.resolveLakekeeperWarehouseName(None, ownerUid, enabled = true) shouldBe None + WorkflowService.resolveLakekeeperWarehouseName(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 + WorkflowService.resolveLakekeeperWarehouseName(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) + WorkflowService.resolveLakekeeperWarehouseName(Some(whid), intruderUid, enabled = true) } error.getMessage should include(whid.toString) } @@ -92,7 +92,7 @@ class WorkflowServiceWarehouseSpec 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) + WorkflowService.resolveLakekeeperWarehouseName(Some(whid), ownerUid, enabled = false) } error.getMessage should include("disabled") } diff --git a/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala b/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala index 30c09b152f..602434aecf 100644 --- a/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala +++ b/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala @@ -54,17 +54,29 @@ class UserWarehouseSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll .fetchOne() .getUid - private def insertWarehouse(uid: Integer, name: String, warehouseName: String): Integer = + // Any globally-unique catalog name will do: this spec covers storage and constraints, + // not how WarehouseResource mints the name. + private def insertWarehouse( + uid: Integer, + name: String, + lakekeeperWarehouseName: String + ): Integer = getDSLContext .insertInto( USER_WAREHOUSE, USER_WAREHOUSE.UID, USER_WAREHOUSE.NAME, - USER_WAREHOUSE.WAREHOUSE_NAME, + USER_WAREHOUSE.LAKEKEEPER_WAREHOUSE_NAME, USER_WAREHOUSE.LAKEKEEPER_WAREHOUSE_ID, USER_WAREHOUSE.FLAVOR ) - .values(uid, name, warehouseName, UUID.randomUUID(), UserWarehouseFlavorEnum.local) + .values( + uid, + name, + lakekeeperWarehouseName, + UUID.randomUUID(), + UserWarehouseFlavorEnum.local + ) .returning(USER_WAREHOUSE.WHID) .fetchOne() .getWhid @@ -78,7 +90,7 @@ class UserWarehouseSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll .where(USER_WAREHOUSE.UID.eq(uid)) .fetchOne() row.getName shouldBe "mybucket" - row.getWarehouseName shouldBe s"user-$uid-mybucket" + row.getLakekeeperWarehouseName shouldBe s"user-$uid-mybucket" row.getFlavor shouldBe UserWarehouseFlavorEnum.local row.getLakekeeperWarehouseId should not be null row.getCreatedAt should not be null @@ -92,7 +104,7 @@ class UserWarehouseSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll insertWarehouse(uid, "dup", s"user-$uid-dup-2") } - it should "reject a duplicate warehouse_name across users" in { + it should "reject a duplicate lakekeeper_warehouse_name across users" in { val first = insertUser("catalog-name-owner") val second = insertUser("catalog-name-intruder") insertWarehouse(first, "shared", "user-shared-catalog-name") diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala index 5d12daf18a..71b325cfdd 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala @@ -36,5 +36,7 @@ class WorkflowContext( var executionId: ExecutionIdentity = DEFAULT_EXECUTION_ID, var workflowSettings: WorkflowSettings = DEFAULT_WORKFLOW_SETTINGS, var cuid: Option[Int] = None, + // The Lakekeeper catalog name this execution writes into; becomes the `/wh/<name>` + // URI segment. None routes to the shared default warehouse. var warehouse: Option[String] = None ) diff --git a/sql/changelog.xml b/sql/changelog.xml index d3a7d55228..c0f735a09c 100644 --- a/sql/changelog.xml +++ b/sql/changelog.xml @@ -99,6 +99,11 @@ <sqlFile path="sql/updates/37.sql"/> </changeSet> + <!-- Rename user_warehouse.warehouse_name to lakekeeper_warehouse_name (#7753) --> + <changeSet id="38" author="mengw15"> + <sqlFile path="sql/updates/38.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 b8ebe3caf6..e8137f9932 100644 --- a/sql/texera_ddl.sql +++ b/sql/texera_ddl.sql @@ -257,16 +257,16 @@ CREATE TABLE IF NOT EXISTS workflow_computing_unit -- Base columns only; the assume-role (BYO-S3) columns come in a later change. CREATE TABLE IF NOT EXISTS user_warehouse ( - whid SERIAL PRIMARY KEY, - uid INT NOT NULL, - name VARCHAR(128) NOT NULL, - warehouse_name VARCHAR(255) NOT NULL UNIQUE, - lakekeeper_warehouse_id UUID NOT NULL, - flavor user_warehouse_flavor_enum NOT NULL, - s3_bucket VARCHAR(255), - s3_endpoint VARCHAR(255), - s3_region VARCHAR(64), - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + whid SERIAL PRIMARY KEY, + uid INT NOT NULL, + name VARCHAR(128) NOT NULL, + lakekeeper_warehouse_name VARCHAR(255) NOT NULL UNIQUE, + lakekeeper_warehouse_id UUID NOT NULL, + flavor user_warehouse_flavor_enum NOT NULL, + s3_bucket VARCHAR(255), + s3_endpoint VARCHAR(255), + s3_region VARCHAR(64), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE (uid, name), FOREIGN KEY (uid) REFERENCES "user" (uid) ON DELETE CASCADE ); diff --git a/sql/updates/38.sql b/sql/updates/38.sql new file mode 100644 index 0000000000..40cebb7dd4 --- /dev/null +++ b/sql/updates/38.sql @@ -0,0 +1,52 @@ +/* + * 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. + */ + + +\c texera_db + +SET search_path TO texera_db; + +BEGIN; + +-- The Lakekeeper catalog name is no longer derived from the user-facing name +-- (#7753), and the table already has its own `name` column -- so the old +-- `warehouse_name` was ambiguous. Rename it to sit beside its sibling +-- `lakekeeper_warehouse_id`. The table is empty in every deployment (the +-- warehouse feature flag is off everywhere), so this carries no data. +-- Guarded so it is a no-op on a fresh DB, where texera_ddl.sql already created the +-- column under its new name: local-dev replays unrecorded change sets and only +-- tolerates "already exists" errors, so an unguarded rename would abort `up` with +-- "column warehouse_name does not exist" (same shape as 33.sql's rename). +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'texera_db' AND table_name = 'user_warehouse' + AND column_name = 'warehouse_name' + ) AND NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'texera_db' AND table_name = 'user_warehouse' + AND column_name = 'lakekeeper_warehouse_name' + ) THEN + ALTER TABLE user_warehouse RENAME COLUMN warehouse_name TO lakekeeper_warehouse_name; + END IF; +END +$$; + +COMMIT;
