feat(storage): bound the per-warehouse catalog cache and release evicted catalogs

[mengw15]

What changes were proposed in this PR?

IcebergCatalogInstance kept one catalog client per warehouse name for the life of the process; with per-user warehouses (#6870) that set is unbounded, and each REST catalog holds an HTTP client. The map is now a Guava cache (maximumSize 64 + expireAfterAccess 60 min, mirroring HuggingFaceModelResource's bounded-cache precedent). An entry idle for the expiry window is closed — nothing can be using it, and idle entries are exactly what a long-lived JVM accumulates. An entry evicted by size is only dropped, never closed: size pressure means more simultaneously hot warehouses than the bound, and closing a hot catalog would fail the operations still using it. Load degrades into rebuild churn, not errors — a dropped catalog lives only as long as its in-flight operations (per-operation resolution bounds every borrow), after which GC reclaims it while the server's keepalive timeout severs its idle connections.

For eviction to be safe, holders stop pinning a catalog — or anything derived from one — across a logical operation: IcebergDocument's lazy val becomes a per-use def, its clear() resolves one catalog for the whole check-then-drop, the reader re-resolves its table on every seek instead of refreshing a pinned one (which also keeps a polling reader's cache entry live), and IcebergTableWriter takes the warehouse rather than a Catalog and loads its table per flush. The lookup rides Guava's per-key locking, dropping the previous JVM-wide synchronized that held one lock across a cache miss's REST config round trip, and unwraps Guava's ExecutionException family so createCatalog failures keep the types they had before.

Only idle-expired entries are closed. replaceInstance stays a plain put: a caller that replaces an entry may still hold and later restore the old reference — amber's integration spec wrap-and-restores the shared catalog, and endpoint reconfiguration (#7358) will swap catalogs the same way.

Also dedupes DocumentFactory's three copies of the URI→(warehouse, namespace, storage key) decode block into one resolver, as promised in #6944 review, and records on the Python side why its per-process catalog dict stays unbounded.

Any related issues, documentation, discussions?

Closes #7290.

How was this PR tested?

New IcebergCatalogInstanceSpec covers the cache contract: idle expiry closes the catalog while size eviction drops it un-closed — on isolated caches built through a package-private factory with a manual ticker, so the JVM-wide cache that parallel suites share is never touched — a catalog displaced by replaceInstance stays open for its owner, and loader failures keep their original exception type.

Per-use resolution is pinned at every holder: IcebergDocument sees a replacement immediately, clear() addresses one catalog across its check-then-drop, the reader re-resolves per seek, and the writer loads through the catalog installed at flush time. Existing iceberg suites (IcebergDocumentSpec, IcebergTableWriterSpec, OnIcebergSpec, DocumentFactorySpec) pass locally — 769 tests in workflow-core — and amber's integration IcebergDocumentSpec is green in CI.

Was this PR authored or co-authored using generative AI tooling?

Generated-by: Claude Code (claude-fable-5)

[github-actions]

Backport auto-label report

This fix: PR was checked against each actively-supported release branch. release/* labels drive the post-merge backport, so add or remove one to change where this fix lands.

Release branch	Analysis
✅ release/v1.2	Change detected on this branch — label added; this fix is queued to backport here. Requested review from @xuang7.

Auto-label run.

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @aglinxinyuan, @Yicong-Huang
  You can notify them by mentioning @aglinxinyuan, @Yicong-Huang in a comment.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.67347% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 89.62%. Comparing base (eefd860) to head (239c497).

Files with missing lines	Patch %	Lines
...he/texera/amber/core/storage/DocumentFactory.scala	77.77%	1 Missing and 3 partials ⚠️
...ra/amber/core/storage/IcebergCatalogInstance.scala	84.00%	2 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #7539      +/-   ##
============================================
- Coverage     89.62%   89.62%   -0.01%     
- Complexity     4395     4398       +3     
============================================
  Files          1177     1177              
  Lines         46884    46899      +15     
  Branches       5239     5246       +7     
============================================
+ Hits          42022    42031       +9     
- Misses         3109     3112       +3     
- Partials       1753     1756       +3     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	98.62% <ø> (ø)		Carriedforward from eefd860
amber	86.35% <83.67%> (-0.03%)	⬇️
computing-unit-managing-service	72.46% <ø> (ø)
config-service	77.31% <ø> (ø)
file-service	68.90% <ø> (ø)
frontend	90.59% <ø> (ø)		Carriedforward from eefd860
notebook-migration-service	78.89% <ø> (ø)
pyamber	97.57% <ø> (ø)
workflow-compiling-service	57.89% <ø> (ø)

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Copilot]

Pull request overview

Bounds per-warehouse Iceberg catalog clients and updates consumers to resolve catalogs through the shared cache.

Changes:

• Adds size/idle eviction and catalog cleanup.
• Updates document and writer catalog resolution.
• Deduplicates VFS-to-Iceberg location resolution and adds tests.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
OnIcebergSpec.scala	Registers a test-specific warehouse catalog.
IcebergTableWriterSpec.scala	Adapts writer tests to warehouse lookup.
IcebergCatalogInstanceSpec.scala	Tests replacement and catalog resolution.
IcebergTableWriter.scala	Resolves catalogs using warehouse identity.
IcebergDocument.scala	Removes pinned catalog references.
IcebergCatalogInstance.scala	Implements the bounded catalog cache.
DocumentFactory.scala	Centralizes Iceberg URI resolution and serde functions.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 7 worse · ⚪ 8 noise (<±5%) · 0 without baseline

Compared against main eefd860 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	372	0.227	25,139/34,982/34,982 us	🔴 +13.7% / 🔴 +114.5%
🔴	bs=100 sw=10 sl=64	773	0.472	123,606/160,098/160,098 us	🔴 +12.1% / 🔴 +48.0%
⚪	bs=1000 sw=10 sl=64	923	0.563	1,085,078/1,121,830/1,121,830 us	⚪ within ±5% / 🔴 -10.1%

Baseline details

Latest main eefd860 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	372 tuples/sec	430 tuples/sec	775.33 tuples/sec	-13.5%	-52.0%
bs=10 sw=10 sl=64	MB/s	0.227 MB/s	0.262 MB/s	0.473 MB/s	-13.4%	-52.0%
bs=10 sw=10 sl=64	p50	25,139 us	22,104 us	12,743 us	+13.7%	+97.3%
bs=10 sw=10 sl=64	p95	34,982 us	34,673 us	16,310 us	+0.9%	+114.5%
bs=10 sw=10 sl=64	p99	34,982 us	34,673 us	18,926 us	+0.9%	+84.8%
bs=100 sw=10 sl=64	throughput	773 tuples/sec	836 tuples/sec	1,001 tuples/sec	-7.5%	-22.8%
bs=100 sw=10 sl=64	MB/s	0.472 MB/s	0.51 MB/s	0.611 MB/s	-7.5%	-22.7%
bs=100 sw=10 sl=64	p50	123,606 us	118,080 us	101,399 us	+4.7%	+21.9%
bs=100 sw=10 sl=64	p95	160,098 us	142,799 us	108,206 us	+12.1%	+48.0%
bs=100 sw=10 sl=64	p99	160,098 us	142,799 us	118,195 us	+12.1%	+35.5%
bs=1000 sw=10 sl=64	throughput	923 tuples/sec	928 tuples/sec	1,026 tuples/sec	-0.5%	-10.1%
bs=1000 sw=10 sl=64	MB/s	0.563 MB/s	0.567 MB/s	0.626 MB/s	-0.7%	-10.1%
bs=1000 sw=10 sl=64	p50	1,085,078 us	1,077,862 us	996,304 us	+0.7%	+8.9%
bs=1000 sw=10 sl=64	p95	1,121,830 us	1,180,321 us	1,042,531 us	-5.0%	+7.6%
bs=1000 sw=10 sl=64	p99	1,121,830 us	1,180,321 us	1,074,934 us	-5.0%	+4.4%

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,538.21,200,128000,372,0.227,25138.63,34981.65,34981.65
1,100,10,64,20,2586.75,2000,1280000,773,0.472,123606.28,160097.59,160097.59
2,1000,10,64,20,21672.50,20000,12800000,923,0.563,1085078.21,1121830.01,1121830.01

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:87

• Eviction closes by cache entry, but the same catalog object can be installed under multiple keys. LocalHadoopIcebergCatalog.ensure does exactly that (LocalHadoopIcebergCatalog.scala:73-76). If one idle/size-evicted alias is removed while another alias remains active, this listener closes the shared object and the surviving entry starts returning a closed catalog. Track ownership/reference count by catalog identity, or otherwise close only after no cache key still references the instance.

          if (notification.wasEvicted()) {
            notification.getValue match {
              case closeable: AutoCloseable =>
                Try(closeable.close()).failed.foreach(error =>
                  logger.warn(s"failed to close evicted catalog '${notification.getKey}'", error)

[Copilot]

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.

Suppressed comments (2)

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergTableWriter.scala:114

• This core safety behavior has no regression test: the writer specs install one catalog before construction and never replace it between construction and a flush. An eager/pinned Table could therefore return without those specs failing. Add a case that constructs the writer with catalog A cached, replaces the warehouse entry with catalog B before a flush (or between two flushes), and verifies the metadata load/commit goes through B.

      val table: Table = IcebergUtil
        .loadTableMetadata(catalog, tableNamespace, tableName)
        .get

common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:130

• The PR description is stale: its testing section says a genuine replacement closes the displaced catalog and that size eviction is not exercised, while this spec deliberately keeps displaced catalogs open and now tests size eviction on an isolated cache. Update the description so reviewers see the actual lifecycle contract and coverage.

  "replaceInstance" should "leave the displaced catalog open for its owner (wrap-and-restore)" in {
    // Integration tests wrap the shared catalog in a spy and restore it afterwards;
    // closing the displaced instance would hand back a dead catalog (#7290 review).
    val original = new FakeCatalog("original")
    val wrapper = new FakeCatalog("wrapper")

[mengw15]

Both suppressed comments addressed:

• Writer regression test — right that nothing guarded it: reverting flushBuffer to an eagerly-held Table left all existing writer specs green. Added in 8ab409c — the writer is constructed with catalog A cached, the warehouse entry is swapped for catalog B before the flush, and the test asserts the metadata load goes through B (verified red against the pinned implementation).
• Stale description — updated: the body described the replacement-closes-displaced semantics that CI disproved, and claimed size eviction was untested. It now states the actual contract (only evicted entries are closed; a displaced catalog stays open for its owner) and the coverage (size/idle eviction on isolated caches with a manual ticker, per-use resolution pinned at each of the four holders).