feat(storage): per-warehouse Iceberg catalog and warehouse-scoped URIs - #6944
Draft
mengw15 wants to merge 11 commits into
Draft
feat(storage): per-warehouse Iceberg catalog and warehouse-scoped URIs#6944mengw15 wants to merge 11 commits into
mengw15 wants to merge 11 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) <noreply@anthropic.com>
Contributor
Automated Reviewer SuggestionsBased on the
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #6944 +/- ##
============================================
+ Coverage 79.20% 83.20% +3.99%
+ Complexity 3786 225 -3561
============================================
Files 1160 533 -627
Lines 46118 30124 -15994
Branches 5114 3222 -1892
============================================
- Hits 36529 25064 -11465
+ Misses 7969 4251 -3718
+ Partials 1620 809 -811
*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 | 395 | 0.241 | 24,687/32,327/32,327 us | 🔴 +105.4% / 🔴 +104.9% |
| 🔴 | bs=100 sw=10 sl=64 | 815 | 0.497 | 122,232/142,536/142,536 us | 🔴 +25.4% / 🔴 +32.8% |
| ⚪ | bs=1000 sw=10 sl=64 | 927 | 0.566 | 1,078,589/1,118,568/1,118,568 us | ⚪ within ±5% / 🔴 -10.1% |
Baseline details
Latest main a61702f from 2026-07-29T14:37:20.598Z
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 395 tuples/sec | 750.01 tuples/sec | 786.12 tuples/sec | -47.3% | -49.8% |
| bs=10 sw=10 sl=64 | MB/s | 0.241 MB/s | 0.458 MB/s | 0.48 MB/s | -47.4% | -49.8% |
| bs=10 sw=10 sl=64 | p50 | 24,687 us | 12,994 us | 12,305 us | +90.0% | +100.6% |
| bs=10 sw=10 sl=64 | p95 | 32,327 us | 15,738 us | 15,774 us | +105.4% | +104.9% |
| bs=10 sw=10 sl=64 | p99 | 32,327 us | 18,799 us | 18,978 us | +72.0% | +70.3% |
| bs=100 sw=10 sl=64 | throughput | 815 tuples/sec | 938.38 tuples/sec | 999.71 tuples/sec | -13.1% | -18.5% |
| bs=100 sw=10 sl=64 | MB/s | 0.497 MB/s | 0.573 MB/s | 0.61 MB/s | -13.2% | -18.5% |
| bs=100 sw=10 sl=64 | p50 | 122,232 us | 106,490 us | 100,616 us | +14.8% | +21.5% |
| bs=100 sw=10 sl=64 | p95 | 142,536 us | 113,622 us | 107,356 us | +25.4% | +32.8% |
| bs=100 sw=10 sl=64 | p99 | 142,536 us | 123,806 us | 113,255 us | +15.1% | +25.9% |
| bs=1000 sw=10 sl=64 | throughput | 927 tuples/sec | 966.12 tuples/sec | 1,031 tuples/sec | -4.0% | -10.1% |
| bs=1000 sw=10 sl=64 | MB/s | 0.566 MB/s | 0.59 MB/s | 0.63 MB/s | -4.0% | -10.1% |
| bs=1000 sw=10 sl=64 | p50 | 1,078,589 us | 1,035,855 us | 980,328 us | +4.1% | +10.0% |
| bs=1000 sw=10 sl=64 | p95 | 1,118,568 us | 1,083,196 us | 1,027,528 us | +3.3% | +8.9% |
| bs=1000 sw=10 sl=64 | p99 | 1,118,568 us | 1,097,913 us | 1,054,298 us | +1.9% | +6.1% |
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,505.81,200,128000,395,0.241,24687.33,32327.48,32327.48
1,100,10,64,20,2454.06,2000,1280000,815,0.497,122232.47,142536.27,142536.27
2,1000,10,64,20,21585.72,20000,12800000,927,0.566,1078588.68,1118568.40,1118568.40… 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) <noreply@anthropic.com>
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.
The URI had two parsers: decodeURI, which located wid/eid/port/resource by searching the segments, and a separate warehouseFromURI. Keeping them apart has already cost two bugs -- they disagreed first about whether a `wh` segment had to be leading, then about whether to split the raw or the decoded path -- and the reason to keep them apart is gone now that decodeURI returns a named structure (apache#7020): adding a field costs no call-site churn. Fold the warehouse into VFSUriComponents as a trailing field, so one parse yields every component and there is a single place where the URI's shape is interpreted. Scala and Python both derive it from the segments the decoder already split. This also lets DocumentFactory resolve the warehouse inside the VFS branch rather than before the scheme dispatch, where it also ran -- harmlessly, but pointlessly -- for dataset URIs that have no warehouse.
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