feat(storage): per-warehouse Iceberg catalog and warehouse-scoped URIs - #6944
Draft
mengw15 wants to merge 8 commits into
Draft
feat(storage): per-warehouse Iceberg catalog and warehouse-scoped URIs#6944mengw15 wants to merge 8 commits into
mengw15 wants to merge 8 commits into
Conversation
Replace the single-warehouse catalog singleton with a per-warehouse cache on both the JVM and Python sides, and encode the warehouse as a leading /wh/<name> segment in VFS URIs so a URI fully identifies where its tables live. The read path derives the warehouse from the URI; non-BYO URIs omit the segment and fall back to the configured warehouse, leaving existing behavior unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #6944 +/- ##
============================================
- Coverage 78.97% 78.88% -0.10%
+ Complexity 3784 3783 -1
============================================
Files 1160 1160
Lines 46105 45910 -195
Branches 5115 5096 -19
============================================
- Hits 36413 36216 -197
+ Misses 8068 8065 -3
- Partials 1624 1629 +5
*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 | 374 | 0.229 | 24,296/37,596/37,596 us | 🔴 -9.7% / 🔴 +137.9% |
| ⚪ | bs=100 sw=10 sl=64 | 783 | 0.478 | 123,768/176,255/176,255 us | ⚪ within ±5% / 🔴 +64.2% |
| ⚪ | bs=1000 sw=10 sl=64 | 929 | 0.567 | 1,073,640/1,106,271/1,106,271 us | ⚪ within ±5% / 🔴 -9.8% |
Baseline details
Latest main 589f601 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 374 tuples/sec | 414 tuples/sec | 787.55 tuples/sec | -9.7% | -52.5% |
| bs=10 sw=10 sl=64 | MB/s | 0.229 MB/s | 0.253 MB/s | 0.481 MB/s | -9.5% | -52.4% |
| bs=10 sw=10 sl=64 | p50 | 24,296 us | 22,818 us | 12,255 us | +6.5% | +98.3% |
| bs=10 sw=10 sl=64 | p95 | 37,596 us | 34,421 us | 15,802 us | +9.2% | +137.9% |
| bs=10 sw=10 sl=64 | p99 | 37,596 us | 34,421 us | 19,008 us | +9.2% | +97.8% |
| bs=100 sw=10 sl=64 | throughput | 783 tuples/sec | 780 tuples/sec | 997.81 tuples/sec | +0.4% | -21.5% |
| bs=100 sw=10 sl=64 | MB/s | 0.478 MB/s | 0.476 MB/s | 0.609 MB/s | +0.4% | -21.5% |
| bs=100 sw=10 sl=64 | p50 | 123,768 us | 120,685 us | 100,690 us | +2.6% | +22.9% |
| bs=100 sw=10 sl=64 | p95 | 176,255 us | 178,762 us | 107,316 us | -1.4% | +64.2% |
| bs=100 sw=10 sl=64 | p99 | 176,255 us | 178,762 us | 113,823 us | -1.4% | +54.8% |
| bs=1000 sw=10 sl=64 | throughput | 929 tuples/sec | 920 tuples/sec | 1,030 tuples/sec | +1.0% | -9.8% |
| bs=1000 sw=10 sl=64 | MB/s | 0.567 MB/s | 0.561 MB/s | 0.629 MB/s | +1.1% | -9.8% |
| bs=1000 sw=10 sl=64 | p50 | 1,073,640 us | 1,094,113 us | 981,213 us | -1.9% | +9.4% |
| bs=1000 sw=10 sl=64 | p95 | 1,106,271 us | 1,142,780 us | 1,027,605 us | -3.2% | +7.7% |
| bs=1000 sw=10 sl=64 | p99 | 1,106,271 us | 1,142,780 us | 1,055,466 us | -3.2% | +4.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,534.12,200,128000,374,0.229,24296.37,37596.22,37596.22
1,100,10,64,20,2554.17,2000,1280000,783,0.478,123768.03,176254.85,176254.85
2,1000,10,64,20,21523.82,20000,12800000,929,0.567,1073640.36,1106271.38,1106271.38… URIs Add unit tests for the new warehouse-aware storage behavior: VFSURIFactory warehouseFromURI extraction and the /wh/<name> URI segment (present with a warehouse, absent without), that a warehouse-scoped URI still round-trips through decodeURI, and that DocumentFactory creates/finds a document for a warehouse-scoped URI (the /wh/ prefix is stripped from the storage key). Python VFSURIFactory.warehouse_from_uri gets matching coverage. Also update the existing state-materialization test to save/restore the IcebergCatalogInstance per-warehouse cache instead of the removed _instance singleton field. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
mengw15
force-pushed
the
feat/6929-byo-s3-storage-foundation
branch
from
July 27, 2026 22:32
03303dd to
acad885
Compare
…nostic catalogs Three defects in the per-warehouse storage foundation, plus the test gap that let them through. warehouseFromURI scanned the whole path for a `wh` segment while the writer emits it as a leading prefix and DocumentFactory.sanitizeURIPath strips only a leading one. Any later `wh` segment therefore selected a warehouse the URI was never built for, and the storage key disagreed with it. Operator ids reach that path verbatim (they come straight off the workflow JSON), so an id such as `a/wh/<name>/b` in a console-messages URI redirected the write to another warehouse -- and the read path applies no ownership check, since the warehouse arrives in the path rather than through the API. Anchor the parse to the first segment, and reject '/' in operator ids and in the GlobalPortIdentity components alongside the existing '_' guard, which was added for this same reason. The catalog cache keyed every entry by warehouse name before dispatching on catalog type, but createCatalog ignores the warehouse for hadoop and postgres. Two warehouse names in a postgres deployment therefore built two equivalent JdbcCatalogs, each with its own connection pool against the same database. Key the warehouse-agnostic types under a constant instead, matching the Python side, which already did this. On the Python side warehouse_from_uri read urlparse().path without unquoting while Scala reads java.net.URI.getPath, which decodes -- so the two languages resolved a percent-encoded name to different warehouses, splitting one execution's data across two of them. create_port_base_uri also had no warehouse parameter, so Python could read the segment but never write it. The warehouse case in DocumentFactorySpec looked up an unregistered warehouse name, which under the default `rest` catalog type missed the cache and tried to build a live REST catalog; it passed only where the environment happened to configure a postgres catalog. Register the name with the local test catalog so the suite exercises the code under test rather than the machine's config.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR is a foundation step for the “per-user BYO-S3 warehouse” work: it makes the storage layer warehouse-aware by (1) caching Iceberg REST catalog clients per warehouse and (2) embedding a warehouse selector into VFS URIs via a leading /wh/<name> path segment, while keeping the default (no-warehouse) behavior unchanged.
Changes:
- Introduces per-warehouse catalog caching on both Scala (
IcebergCatalogInstance) and Python (iceberg_catalog_instance.py) sides. - Extends VFS URI construction/parsing to optionally include a leading
/wh/<name>segment, and wiresDocumentFactoryto derive the warehouse from the URI. - Adds unit tests in Scala and Python to cover warehouse-scoped URI handling and document creation/existence checks.
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/VFSURIFactorySpec.scala | Adds Scala tests for warehouseFromURI, /wh/<name> prefix behavior, and decode round-trips. |
| common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala | Enhances the shared local test catalog helper to register the same catalog under multiple warehouse keys for tests. |
| common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala | Adds Scala test coverage for warehouse-scoped VFS URIs when creating/finding documents. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/util/serde/GlobalPortIdentitySerde.scala | Adds /-rejection to prevent URI path segment forging in serialized global port identity components. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/WorkflowContext.scala | Adds an optional warehouse field to carry per-execution selection through runtime context. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala | Implements /wh/<name> prefix support and warehouse extraction from VFS URIs; extends URI builders with optional warehouse. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala | Threads a warehouse value into IcebergDocument and uses it to select the correct catalog instance. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala | Replaces singleton catalog with a per-warehouse cache (REST) while keeping non-REST catalogs shared. |
| common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala | Derives warehouse from VFS URI and strips /wh/<name>/ from storage key generation. |
| amber/src/test/python/core/storage/test_vfs_uri_factory.py | Adds Python unit tests for warehouse_from_uri, /wh/ prefix writing/reading, and decode round-trips. |
| amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py | Updates test scaffolding to save/restore the new catalog cache instead of a singleton. |
| amber/src/main/python/core/storage/vfs_uri_factory.py | Adds Python warehouse_from_uri and writes /wh/<name> prefix in create_port_base_uri. |
| amber/src/main/python/core/storage/iceberg/iceberg_document.py | Passes warehouse through to catalog selection in Python IcebergDocument. |
| amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py | Implements per-warehouse catalog caching on Python side (REST keyed by warehouse). |
| amber/src/main/python/core/storage/document_factory.py | Parses warehouse from URI and strips /wh/<name>/ when generating storage keys; selects per-warehouse catalog. |
Comments suppressed due to low confidence (3)
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala:61
warehouseFromURIusesuri.getPath, which is already percent-decoded byjava.net.URI. That means a warehouse name containing%2Fwill be decoded into/before splitting, letting a crafted URI forge extra path segments and potentially select the wrong warehouse. Consider parsinggetRawPathand decoding the extracted segment yourself, rejecting decoded names that contain/.
def warehouseFromURI(uri: URI): Option[String] =
uri.getPath.stripPrefix("/").split("/").toList match {
case "wh" :: name :: _ if name.nonEmpty => Some(name)
case _ => None
}
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/VFSURIFactory.scala:208
- The new
operatorIdvalidation only rejects a literal/, butjava.net.URI.getPathalso decodes percent-encoded slashes. AnoperatorIdcontaining%2Fcan still become extra path segments at parse time and confusedecodeURI(which usessegments.indexOf).
require(
!opId.id.contains('/'),
s"operatorId must not contain '/' (VFS URI parsing relies on this): ${opId.id}"
)
amber/src/main/python/core/storage/vfs_uri_factory.py:125
create_port_base_uriinterpolateswarehousedirectly into the URI path. Ifwarehousecontains/or a percent-encoded slash (%2F), the URI can gain extra path segments when parsed/decoded, which can confuse downstream parsing (decode_uri,warehouse_from_uri) and potentially route data to the wrong warehouse.
wh_segment = f"/wh/{warehouse}" if warehouse else ""
return (
f"{VFSURIFactory.VFS_FILE_URI_SCHEME}://{wh_segment}/wid/{workflow_id.id}"
f"/eid/{execution_id.id}/globalportid/"
f"{serialize_global_port_identity(global_port_id)}"
)
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
… segments The `/` guards added for operator ids only reject a literal slash, and the warehouse name was interpolated into the URI path with no validation at all. Since `java.net.URI.getPath` percent-decodes, a `%2F` slipped past both: a warehouse named `a%2Fwid%2F999` produced a path that reads back as `/wh/a/wid/999/wid/1/...`, so the warehouse resolved to `a` and `decodeURI`'s key search found the injected `wid` rather than the real one -- one execution's data landing under another's storage key. Python was not affected there, because its parser splits the undecoded path, so the two languages also disagreed. Fix it structurally rather than by blacklisting more characters: split the RAW path in both languages, so a percent-encoded slash stays inside the segment that contains it and cannot become a separator, and require the warehouse segment to be a legal name, so a URI that `warehousePathSegment` could never have written resolves to no warehouse instead of to a wrong one. Validate the name on the way in as well -- registration applies a stricter rule, but the URI layer should not depend on that. Reported by Copilot on apache#6944.
The Scala serde was given a '/' guard alongside its existing '_' guard, because both characters would otherwise let a user-chosen operator id forge structure in the VFS URI the id is interpolated into. Python's serialize_global_port_identity mirrors that serializer but only ever checked '_', so the contract held in one language and not the other. No production Python code builds these URIs today, so this closes the gap rather than fixing a reachable bug -- but the guard is the contract both sides parse against, and it should not depend on which language wrote the string.
warehouseFromURI was changed to split the raw path so a percent-encoded slash
cannot forge segments, but decodeURI -- which finds `wid`/`eid` by searching the
same segments -- was left splitting the decoded path. A URI whose warehouse name
contained `%2F` therefore read as `/wh/a/wid/999/wid/1/...` there, and the key
search returned the injected `wid` rather than the real one, putting one
execution's data under another's storage key. Python's decode_uri already split
the raw path, so the two languages also disagreed about which execution a URI
identified.
Split the raw path here too. Legitimate URIs are unaffected: the characters the
serialized port identity contributes -- '(', ')', ',', '=' -- are all legal in a
path segment and are never percent-encoded, so raw and decoded are identical for
anything the factory writes.
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?
First slice of the per-user bring-your-own-S3 warehouse feature (umbrella #6870). This is a
foundation-only, backward-compatible change that makes the storage layer warehouse-aware
without changing any behavior yet:
(
IcebergCatalogInstance) and Python (iceberg_catalog_instance.py) sides, so one workerprocess can hold several REST catalogs — one per warehouse it touches. Only the REST
(Lakekeeper) catalog varies by warehouse; the hadoop/postgres catalogs stay warehouse-agnostic.
/wh/<name>segment in VFS URIs (VFSURIFactory), so a URIfully identifies where its tables live; the read path (
DocumentFactory) derives the warehouseback out of the URI.
warehousefield toWorkflowContextto carry the selection through an execution.With no warehouse in play (the default), URIs and catalog behavior are byte-for-byte identical to
today, so nothing changes for existing deployments.
Any related issues, documentation, discussions?
Closes #6929. Part of #6870 (design discussions #5293 and #6040).
How was this PR tested?
Adds unit tests for the new warehouse-aware behavior:
VFSURIFactorySpec—warehouseFromURIextraction, the/wh/<name>segment (present with awarehouse, absent without), and that a warehouse-scoped URI still round-trips through
decodeURI.DocumentFactorySpec— creating and finding a document for a warehouse-scoped URI against a localIceberg catalog (the
/wh/prefix is stripped from the storage key).test_vfs_uri_factory.py— matching coverage for the Pythonwarehouse_from_uri.The existing storage/Iceberg tests continue to cover the default (no-warehouse) path unchanged.
The full build and test suite run in CI.
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8