feat(storage): warehouse REST API, Lakekeeper client, and per-execution injection - #7473
Open
mengw15 wants to merge 4 commits into
Open
feat(storage): warehouse REST API, Lakekeeper client, and per-execution injection#7473mengw15 wants to merge 4 commits into
mengw15 wants to merge 4 commits into
Conversation
github-actions
Bot
added
feature
engine
ddl-change
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7473 +/- ##
============================================
+ Coverage 86.75% 86.78% +0.03%
- Complexity 4226 4245 +19
============================================
Files 1170 1173 +3
Lines 46807 46910 +103
Branches 5211 5213 +2
============================================
+ Hits 40606 40712 +106
+ Misses 4496 4485 -11
- Partials 1705 1713 +8
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 382 | 0.233 | 24,517/36,477/36,477 us | 🔴 +32.1% / 🔴 +119.5% |
| ⚪ | bs=100 sw=10 sl=64 | 915 | 0.558 | 107,620/143,289/143,289 us | ⚪ within ±5% / 🔴 +26.6% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,092 | 0.666 | 923,934/950,496/950,496 us | ⚪ within ±5% / 🟢 -14.8% |
Baseline details
Latest main a7f4386 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 382 tuples/sec | 445 tuples/sec | 740.3 tuples/sec | -14.2% | -48.4% |
| bs=10 sw=10 sl=64 | MB/s | 0.233 MB/s | 0.271 MB/s | 0.452 MB/s | -14.0% | -48.4% |
| bs=10 sw=10 sl=64 | p50 | 24,517 us | 22,210 us | 13,173 us | +10.4% | +86.1% |
| bs=10 sw=10 sl=64 | p95 | 36,477 us | 27,603 us | 16,615 us | +32.1% | +119.5% |
| bs=10 sw=10 sl=64 | p99 | 36,477 us | 27,603 us | 19,272 us | +32.1% | +89.3% |
| bs=100 sw=10 sl=64 | throughput | 915 tuples/sec | 937 tuples/sec | 939.86 tuples/sec | -2.3% | -2.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.558 MB/s | 0.572 MB/s | 0.574 MB/s | -2.4% | -2.7% |
| bs=100 sw=10 sl=64 | p50 | 107,620 us | 104,646 us | 106,062 us | +2.8% | +1.5% |
| bs=100 sw=10 sl=64 | p95 | 143,289 us | 138,612 us | 113,212 us | +3.4% | +26.6% |
| bs=100 sw=10 sl=64 | p99 | 143,289 us | 138,612 us | 123,853 us | +3.4% | +15.7% |
| bs=1000 sw=10 sl=64 | throughput | 1,092 tuples/sec | 1,104 tuples/sec | 966.11 tuples/sec | -1.1% | +13.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.666 MB/s | 0.674 MB/s | 0.59 MB/s | -1.2% | +12.9% |
| bs=1000 sw=10 sl=64 | p50 | 923,934 us | 903,560 us | 1,038,733 us | +2.3% | -11.1% |
| bs=1000 sw=10 sl=64 | p95 | 950,496 us | 939,928 us | 1,085,289 us | +1.1% | -12.4% |
| bs=1000 sw=10 sl=64 | p99 | 950,496 us | 939,928 us | 1,115,555 us | +1.1% | -14.8% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,523.10,200,128000,382,0.233,24516.56,36476.72,36476.72
1,100,10,64,20,2186.89,2000,1280000,915,0.558,107620.48,143288.73,143288.73
2,1000,10,64,20,18318.72,20000,12800000,1092,0.666,923934.32,950495.74,950495.74
Contributor
There was a problem hiding this comment.
Pull request overview
Adds backend support for per-user Lakekeeper warehouses, including management APIs, execution routing, persistence, and feature gating.
Changes:
- Adds warehouse CRUD APIs and Lakekeeper integration.
- Propagates warehouse selection through execution storage and metadata.
- Adds schema migration and warehouse-focused tests.
Reviewed changes
Copilot reviewed 23 out of 23 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
sql/updates/33.sql |
Adds execution warehouse reference. |
sql/texera_ddl.sql |
Updates base execution schema. |
sql/changelog.xml |
Registers migration 33. |
common/config/.../StorageConfig.scala |
Exposes warehouse bucket configuration and testable gate. |
amber/src/test/.../WorkflowServiceWarehouseSpec.scala |
Tests warehouse resolution and ownership. |
amber/src/test/.../WorkflowExecutionServiceSpec.scala |
Updates execution request fixture. |
amber/src/test/.../WarehouseReadGuardSpec.scala |
Tests disabled read behavior. |
amber/src/test/.../LakekeeperClientSpec.scala |
Tests management payload and deletion order. |
amber/src/test/.../ExecutionsMetadataPersistServiceSpec.scala |
Tests warehouse persistence and deletion behavior. |
amber/src/test/.../WarehouseResourceSpec.scala |
Tests warehouse REST operations. |
amber/src/main/.../TexeraWebApplication.scala |
Registers the warehouse resource. |
amber/src/main/.../WorkflowService.scala |
Resolves and injects selected warehouses. |
amber/src/main/.../WarehouseReadGuard.scala |
Implements warehouse read gating. |
amber/src/main/.../LakekeeperClient.scala |
Implements Lakekeeper management operations. |
amber/src/main/.../ExecutionStatsService.scala |
Scopes runtime statistics by warehouse. |
amber/src/main/.../ExecutionsMetadataPersistService.scala |
Records execution warehouse IDs. |
amber/src/main/.../ExecutionResultService.scala |
Guards warehouse result pagination. |
amber/src/main/.../ExecutionConsoleService.scala |
Scopes console storage by warehouse. |
amber/src/main/.../SyncExecutionResource.scala |
Supplies the default warehouse selection. |
amber/src/main/.../WarehouseResource.scala |
Provides warehouse CRUD endpoints. |
amber/src/main/.../WorkflowExecuteRequest.scala |
Adds warehouse selection to execution requests. |
amber/src/main/.../ExpansionGreedyScheduleGenerator.scala |
Scopes generated result URIs. |
amber/src/main/.../CostBasedScheduleGenerator.scala |
Scopes generated result URIs. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
mengw15
marked this pull request as ready for review
August 10, 2026 18:53
kunwp1
reviewed
Aug 10, 2026
kunwp1
left a comment
kunwp1
left a comment
Contributor
There was a problem hiding this comment.
Left some comments. Do you have to also work on python side as well?
| storageUriOption match { | ||
| case Some(storageUri) => | ||
| // Refuse to read a per-user-warehouse result while the feature is off (#6930). | ||
| WarehouseReadGuard.assertReadable(storageUri) |
Contributor
There was a problem hiding this comment.
This assert error won't propagate because the catch below turns all the Exception to None. Is it intentional?
|
|
||
| uriOption.flatMap { uri => | ||
| // Refuse to read per-user-warehouse console messages while the feature is off (#6930). | ||
| WarehouseReadGuard.assertReadable(uri) |
| context | ||
| .deleteFrom(USER_WAREHOUSE) | ||
| .where(USER_WAREHOUSE.WHID.eq(whid)) | ||
| .execute() |
Contributor
There was a problem hiding this comment.
I think you need to handle the case where DB delete fails because LakeKeeper is deleted first before DB delete.
| val warehouseEnabled: Boolean = | ||
| // warehouseEnabled is a var because tests need to exercise the enabled path | ||
| // (mirroring s3Endpoint above); production code never writes it. | ||
| var warehouseEnabled: Boolean = |
Contributor
There was a problem hiding this comment.
Use val to make it consistent with others?
Contributor
Author
There was a problem hiding this comment.
I checked this, and I found some other are also using var, for example
var s3Endpoint: String = conf.getString("storage.s3.endpoint")
var lakefsEndpoint: String = conf.getString("storage.lakefs.endpoint")
Comment on lines
+45
to
+48
| 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() |
Contributor
There was a problem hiding this comment.
If this pattern is from VFSURIFactory codebase, can you deduplicate this pattern?
Comment on lines
+90
to
+96
| DashboardWarehouse( | ||
| row.getWhid, | ||
| row.getName, | ||
| row.getWarehouseName, | ||
| row.getFlavor.getLiteral, | ||
| row.getCreatedAt.toInstant.toEpochMilli | ||
| ) |
Contributor
There was a problem hiding this comment.
Maybe deduplicate this code with line 167
Comment on lines
+126
to
+136
| 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 | ||
| } |
Contributor
There was a problem hiding this comment.
Can you check if this API is paginated and this only returns the first few tables? The API may not return the entire pages and you might need to consider that.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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):WarehouseResource:GET /warehouse/status(always answers, so the frontend can hide the feature),POST /warehouseandDELETE /warehouse/{whid}(403 while the flag is off). Create validates the URI-safe name, mints the catalog nameuser-<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 withpurgeRequested=true, then the namespaces, then the warehouse entity.WorkflowExecuteRequestgainswarehouseId: Option[Int];WorkflowService.resolveWarehouseNamechecks ownership and refuses an explicit pick while the feature is off (never a silent fallback — [BYO-S3] Feature flag (storage.warehouse.enabled, default off) #6930); the resolved name ridesWorkflowContext.warehouseinto every storage URI (results, runtime statistics, console messages); the chosenwhidis recorded onworkflow_executions(ascuidis today) so the picker can preselect the workflow's last-used warehouse.whidisON DELETE SET NULL: deleting a warehouse purges its data, never the execution history.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" ([BYO-S3] Feature flag (storage.warehouse.enabled, default off) #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)