You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "mengw15 (via GitHub)" <gi...@apache.org> on 2026/08/11 05:34:35 UTC
[PR] fix(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
mengw15 opened a new pull request, #7539:
URL: https://github.com/apache/texera/pull/7539
### 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) whose removal listener closes evicted catalogs; the next access simply rebuilds one.
For eviction to be safe, holders stop pinning a catalog: `IcebergDocument`'s `lazy val` becomes a per-use `def`, and `IcebergTableWriter` now takes the warehouse (not a `Catalog`) and resolves per use — a replaced or rebuilt catalog is picked up immediately. 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. `replaceInstance` skips a same-instance re-put: Guava reports one as a replacement, which would close a catalog that is still installed (parallel test suites re-register one shared catalog).
Also dedupes `DocumentFactory`'s three copies of the URI→(warehouse, namespace, storage key) decode block into one resolver, as promised in #6944 review.
### Any related issues, documentation, discussions?
Closes #7290.
### How was this PR tested?
New `IcebergCatalogInstanceSpec`: a genuine replacement closes the displaced catalog, a same-instance re-put does not, and `IcebergDocument` sees a replacement immediately (pinning the per-use `def` against a pinned `lazy val`). Existing iceberg suites (`IcebergDocumentSpec`, `IcebergTableWriterSpec`, `OnIcebergSpec`, `DocumentFactorySpec`) pass locally — 51 tests. Size-based eviction is deliberately not exercised: it would flood the JVM-wide cache parallel suites share; it flows through the same removal listener the replacement cases pin.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3777394761
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,11 +136,35 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ getOrLoad(catalogs, cacheKey(name), () => createCatalog(name))
}
+ /**
+ * `Cache.get` wraps loader failures (`UncheckedExecutionException`, `ExecutionException`,
+ * `ExecutionError`); unwrap them so `createCatalog` failures keep the types they had
+ * before the cache existed. Package-private so the spec can pin the unwrapping against
+ * an isolated cache with a throwing loader.
+ */
+ private[storage] def getOrLoad(
+ cache: Cache[String, Catalog],
+ key: String,
+ loader: () => Catalog
+ ): Catalog =
+ try {
+ // get(key, loader) locks per key, not globally: a cache miss's REST config
+ // round trip no longer blocks lookups of other warehouses.
+ cache.get(
+ key,
+ new Callable[Catalog] {
+ override def call(): Catalog = loader()
+ }
+ )
+ } catch {
+ case e: UncheckedExecutionException => throw e.getCause
+ case e: ExecutionException => throw e.getCause
+ case e: ExecutionError => throw e.getCause
Review Comment:
Done in 8a71bc4da — one clause via a type alternation.
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -82,6 +84,29 @@ class IcebergDocumentSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
new IcebergDocument[Tuple](tableNamespace, tableName, icebergSchema, serde, deserde)
}
+ /**
+ * Delegates to the shared local catalog while counting `loadTable` calls -- the
+ * discriminator for per-seek table re-resolution (#7290): a reader that only
+ * refreshed a pinned Table would touch the catalog exactly once, at iterator
+ * construction, no matter how long it polls.
+ */
+ private class CountingCatalog(delegate: Catalog) extends Catalog {
+ val loadTableCalls = new AtomicInteger()
+ override def name(): String = "counting"
+ override def loadTable(identifier: TableIdentifier): Table = {
+ loadTableCalls.incrementAndGet()
+ delegate.loadTable(identifier)
+ }
+ override def tableExists(identifier: TableIdentifier): Boolean =
+ delegate.tableExists(identifier)
+ override def listTables(namespace: Namespace): java.util.List[TableIdentifier] =
+ delegate.listTables(namespace)
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean =
+ delegate.dropTable(identifier, purge)
+ override def renameTable(from: TableIdentifier, to: TableIdentifier): Unit =
+ delegate.renameTable(from, to)
+ }
+
Review Comment:
Done in 8a71bc4da — extracted to a shared test helper (`CountingCatalog.scala`, next to `LocalHadoopIcebergCatalog`); both specs import it.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
Review Comment:
Good catch — done in 8a71bc4da. The constructor-time load was immediately overwritten by the construction-time seek, i.e. one wasted REST round trip per reader. Initialized to `None`, and the counting spec now pins **exactly two** loads per full drain (construction seek + final exhausted-files seek), so an eager load would fail it again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5283795941
Added a TL;DR paragraph at the top of the description — it now leads with the plain-language problem (a process-wide registry that never removes entries, unbounded once warehouses are per-user) and the two moves this PR makes, before the mechanism details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762552992
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
@@ -72,7 +72,10 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
private val lock = new ReentrantReadWriteLock()
- @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
+ // Resolved per use, never held: the catalog cache is bounded and closes evicted
+ // entries (#7290), so a pinned reference could outlive its catalog. A public def
+ // (not a lazy val) also means a replaced/rebuilt catalog is picked up immediately.
+ def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
Done in 92e9fe2f6. `seekToUsableFile` now re-resolves the table from the current catalog at every seek instead of refreshing a pinned one — snapshot continuity already lived in `lastSnapshotId`, not in the `Table` instance — so an actively polling reader bumps its cache entry's access time on every seek and can no longer look idle. Pinned by a workflow-core case that counts catalog resolutions through a delegating catalog (the pinned-refresh implementation resolved exactly once, at construction); the integration lazy-advancement spec now counts `loadTable` calls, which is where metadata loads surface after this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762485380
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
@@ -72,7 +72,10 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
private val lock = new ReentrantReadWriteLock()
- @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
+ // Resolved per use, never held: the catalog cache is bounded and closes evicted
+ // entries (#7290), so a pinned reference could outlive its catalog. A public def
+ // (not a lazy val) also means a replaced/rebuilt catalog is picked up immediately.
+ def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
This still leaves long-lived document readers holding a catalog-derived `Table`. `getUsingFileSequenceOrder` initializes its `table` once (lines 173–174), and later `seekToUsableFile` calls `table.refresh()` without resolving this accessor again (lines 198–216). An actively polled reader therefore looks idle to this cache; after 60 minutes another cache operation can expire and close its REST catalog, and the next refresh uses the closed `RESTTableOperations`. Resolve a fresh table from the current catalog at each polling/refresh boundary (while retaining the snapshot ID), or keep a lease for the iterator's lifetime.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5249395311
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/7539?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 87.41%. Comparing base ([`561cd0e`](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a6ff5b8`](https://app.codecov.io/gh/apache/texera/commit/a6ff5b880def8b7427e1d0ba888aef6a7b779553?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #7539 +/- ##
============================================
- Coverage 87.42% 87.41% -0.01%
+ Complexity 4253 4251 -2
============================================
Files 1176 1176
Lines 46986 46954 -32
Branches 5245 5245
============================================
- Hits 41077 41046 -31
Misses 4178 4178
+ Partials 1731 1730 -1
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `98.62% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `82.18% <ø> (-0.03%)` | :arrow_down: | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `60.38% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `65.97% <ø> (ø)` | | |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.05% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.47% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.89% <ø> (ø)` | | |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `97.52% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/7539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.31% <ø> (ø)` | | Carriedforward from [561cd0e](https://app.codecov.io/gh/apache/texera/commit/561cd0ef1fb85da3f2f59fe846947d93f5c14262?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/7539?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5249451402
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 2 better · 🔴 7 worse · ⚪ 6 noise (<±5%) · 0 without baseline
> Compared against main `561cd0e` 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](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/31462429442)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 364 | 0.222 | 25,353/43,699/43,699 us | 🔴 +13.4% / 🔴 +163.0% |
| 🔴 | bs=100 sw=10 sl=64 | 781 | 0.477 | 125,282/175,050/175,050 us | 🔴 +11.7% / 🔴 +54.6% |
| 🟢 | bs=1000 sw=10 sl=64 | 918 | 0.56 | 1,091,229/1,150,118/1,150,118 us | 🟢 -7.8% / 🔴 +6.0% |
<details><summary>Baseline details</summary>
Latest main `561cd0e` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 364 tuples/sec | 417 tuples/sec | 740.3 tuples/sec | -12.7% | -50.8% |
| bs=10 sw=10 sl=64 | MB/s | 0.222 MB/s | 0.254 MB/s | 0.452 MB/s | -12.6% | -50.9% |
| bs=10 sw=10 sl=64 | p50 | 25,353 us | 22,364 us | 13,173 us | +13.4% | +92.5% |
| bs=10 sw=10 sl=64 | p95 | 43,699 us | 40,509 us | 16,615 us | +7.9% | +163.0% |
| bs=10 sw=10 sl=64 | p99 | 43,699 us | 40,509 us | 19,272 us | +7.9% | +126.7% |
| bs=100 sw=10 sl=64 | throughput | 781 tuples/sec | 819 tuples/sec | 939.86 tuples/sec | -4.6% | -16.9% |
| bs=100 sw=10 sl=64 | MB/s | 0.477 MB/s | 0.5 MB/s | 0.574 MB/s | -4.6% | -16.8% |
| bs=100 sw=10 sl=64 | p50 | 125,282 us | 120,696 us | 106,062 us | +3.8% | +18.1% |
| bs=100 sw=10 sl=64 | p95 | 175,050 us | 156,722 us | 113,212 us | +11.7% | +54.6% |
| bs=100 sw=10 sl=64 | p99 | 175,050 us | 156,722 us | 123,853 us | +11.7% | +41.3% |
| bs=1000 sw=10 sl=64 | throughput | 918 tuples/sec | 914 tuples/sec | 966.11 tuples/sec | +0.4% | -5.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.56 MB/s | 0.558 MB/s | 0.59 MB/s | +0.4% | -5.0% |
| bs=1000 sw=10 sl=64 | p50 | 1,091,229 us | 1,096,477 us | 1,038,733 us | -0.5% | +5.1% |
| bs=1000 sw=10 sl=64 | p95 | 1,150,118 us | 1,247,197 us | 1,085,289 us | -7.8% | +6.0% |
| bs=1000 sw=10 sl=64 | p99 | 1,150,118 us | 1,247,197 us | 1,115,555 us | -7.8% | +3.1% |
</details>
<details><summary>Raw CSV</summary>
```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,549.58,200,128000,364,0.222,25353.32,43699.05,43699.05
1,100,10,64,20,2559.35,2000,1280000,781,0.477,125281.85,175050.30,175050.30
2,1000,10,64,20,21797.02,20000,12800000,918,0.560,1091228.79,1150117.94,1150117.94
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3755525391
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -36,11 +39,34 @@ import scala.collection.mutable
* Only the REST catalog varies by warehouse; the hadoop and postgres catalogs are warehouse-agnostic
* and ignore the warehouse argument.
*
- * Access is synchronized because the same JVM serves multiple warehouses concurrently.
+ * The cache is bounded (#7290): per-user warehouses (#6870) make the set of catalogs a
+ * long-lived JVM touches unbounded, and each REST catalog holds an HTTP client. Entries
+ * fall out by size or idleness and are closed by the removal listener; the next access
+ * simply rebuilds one. Callers must therefore resolve their catalog per use instead of
+ * holding one across an execution (see IcebergDocument / IcebergTableWriter).
*/
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
- private val catalogs = mutable.Map.empty[String, Catalog]
+ // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent: generous enough
+ // that eviction never hits a warehouse in active use, small enough to bound the JVM.
+ private val CatalogCacheMaxSize = 64L
+ private val CatalogCacheExpireAfterAccessMinutes = 60L
+
+ private val catalogs: Cache[String, Catalog] = CacheBuilder
+ .newBuilder()
+ .maximumSize(CatalogCacheMaxSize)
+ .expireAfterAccess(CatalogCacheExpireAfterAccessMinutes, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener[String, Catalog] {
+ override def onRemoval(notification: RemovalNotification[String, Catalog]): Unit =
+ notification.getValue match {
+ case closeable: AutoCloseable =>
+ Try(closeable.close()).failed.foreach(error =>
+ logger.warn(s"failed to close evicted catalog '${notification.getKey}'", error)
Review Comment:
The long-lived pin this describes was real, but it was the writer's eager `table` val — fixed in f0e5da334 to resolve per flush, which also keeps the entry's access time fresh for the whole execution. What remains is the within-one-operation window: a just-resolved entry is the cache's most recently used, so size eviction cannot pick it, and idle expiry needs 60 idle minutes. Closing it mid-operation would take ~64 new warehouses materializing inside a single operation; a lease/refcount layer seems heavier than that risk warrants at this scale.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -102,8 +133,14 @@ object IcebergCatalogInstance {
* @param catalog the catalog to cache.
* @param warehouse the warehouse to cache it under; `None` uses the configured default.
*/
- def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit =
- synchronized {
- catalogs(cacheKey(warehouse.getOrElse(defaultWarehouse))) = catalog
+ def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit = {
+ val key = cacheKey(warehouse.getOrElse(defaultWarehouse))
+ // Guava reports a same-value put as a replacement, which would fire the removal
+ // listener and close a catalog that is still installed: the shared test catalog
+ // is ensure()d repeatedly (and under several names) by parallel suites. Skip the
+ // no-op re-put so only a genuine replacement closes the previous catalog.
+ if (catalogs.getIfPresent(key) ne catalog) {
+ catalogs.put(key, catalog)
}
Review Comment:
Done in f0e5da334: `replaceInstance` now goes through `asMap`'s `putIfAbsent` + `replace(key, previous, catalog)`, so a lost race leaves the winner installed and only a genuine replacement fires the listener.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergTableWriter.scala:
##########
@@ -50,13 +52,17 @@ import scala.collection.mutable.ArrayBuffer
*/
private[storage] class IcebergTableWriter[T](
val writerIdentifier: String,
- val catalog: Catalog,
+ val warehouse: Option[String],
val tableNamespace: String,
val tableName: String,
val tableSchema: Schema,
val serde: (org.apache.iceberg.Schema, T) => Record
) extends BufferedItemWriter[T] {
+ // Resolved per use (#7290): the catalog cache is bounded and closes evicted entries,
+ // so the writer must not pin one across its lifetime.
+ private def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
Good catch — fixed in f0e5da334. The eager val pinned a `Table` across the writer's lifetime, exactly the reference the cache could close. `flushBuffer` now resolves the table per flush; commits already pay a REST round trip, and the per-flush lookup keeps the warehouse's entry live during long executions.
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.core.storage
+
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.Table
+import org.apache.iceberg.catalog.{Catalog, Namespace, TableIdentifier}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Spec for the bounded catalog cache (#7290): a genuine replacement closes the
+ * catalog it displaces, a same-instance re-registration does not (that is what
+ * [[LocalHadoopIcebergCatalog.ensure]] relies on), and holders resolve their
+ * catalog per use so a replacement is visible immediately.
+ *
+ * Size-based eviction is deliberately not exercised: forcing it would flood the
+ * JVM-wide cache that parallel suites share and could evict their live catalog.
+ * The close-on-removal wiring it would exercise is pinned by the replacement
+ * cases below, which Guava routes through the same removal listener.
Review Comment:
Done in f0e5da334: cache construction is behind a package-private `buildCatalogCache(maximumSize, expireAfterAccess, ticker)`; the spec pins size eviction and idle expiry on isolated instances with a manual ticker, without touching the JVM-wide cache.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,9 +96,14 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ // get(key, loader) locks per key, not globally: a cache miss's REST config
+ // round trip no longer blocks lookups of other warehouses.
+ catalogs.get(
+ cacheKey(name),
+ new Callable[Catalog] {
+ override def call(): Catalog = createCatalog(name)
+ }
+ )
Review Comment:
Done in f0e5da334: `getInstance` unwraps `UncheckedExecutionException` / `ExecutionException` / `ExecutionError` back to the loader's original exception, pinned by spec cases for both runtime and checked failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "kunwp1 (via GitHub)" <gi...@apache.org>.
kunwp1 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3776641942
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
Review Comment:
Can you check if this line needs to be changed now to initialize it to None? Your change adds a loadTableMetadata() in seekToUsableFile() so this line might cause inefficiency.
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -82,6 +84,29 @@ class IcebergDocumentSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
new IcebergDocument[Tuple](tableNamespace, tableName, icebergSchema, serde, deserde)
}
+ /**
+ * Delegates to the shared local catalog while counting `loadTable` calls -- the
+ * discriminator for per-seek table re-resolution (#7290): a reader that only
+ * refreshed a pinned Table would touch the catalog exactly once, at iterator
+ * construction, no matter how long it polls.
+ */
+ private class CountingCatalog(delegate: Catalog) extends Catalog {
+ val loadTableCalls = new AtomicInteger()
+ override def name(): String = "counting"
+ override def loadTable(identifier: TableIdentifier): Table = {
+ loadTableCalls.incrementAndGet()
+ delegate.loadTable(identifier)
+ }
+ override def tableExists(identifier: TableIdentifier): Boolean =
+ delegate.tableExists(identifier)
+ override def listTables(namespace: Namespace): java.util.List[TableIdentifier] =
+ delegate.listTables(namespace)
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean =
+ delegate.dropTable(identifier, purge)
+ override def renameTable(from: TableIdentifier, to: TableIdentifier): Unit =
+ delegate.renameTable(from, to)
+ }
+
Review Comment:
This same code also appears in the other test file. Try to de-duplicate this.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,11 +136,35 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ getOrLoad(catalogs, cacheKey(name), () => createCatalog(name))
}
+ /**
+ * `Cache.get` wraps loader failures (`UncheckedExecutionException`, `ExecutionException`,
+ * `ExecutionError`); unwrap them so `createCatalog` failures keep the types they had
+ * before the cache existed. Package-private so the spec can pin the unwrapping against
+ * an isolated cache with a throwing loader.
+ */
+ private[storage] def getOrLoad(
+ cache: Cache[String, Catalog],
+ key: String,
+ loader: () => Catalog
+ ): Catalog =
+ try {
+ // get(key, loader) locks per key, not globally: a cache miss's REST config
+ // round trip no longer blocks lookups of other warehouses.
+ cache.get(
+ key,
+ new Callable[Catalog] {
+ override def call(): Catalog = loader()
+ }
+ )
+ } catch {
+ case e: UncheckedExecutionException => throw e.getCause
+ case e: ExecutionException => throw e.getCause
+ case e: ExecutionError => throw e.getCause
Review Comment:
Collapse these three identical clauses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "kunwp1 (via GitHub)" <gi...@apache.org>.
kunwp1 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3778123967
##########
amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py:
##########
@@ -35,6 +35,11 @@ class IcebergCatalogInstance:
testing or reconfiguration.
"""
+ # Deliberately unbounded, unlike the Scala side's bounded cache (#7290): a Python
+ # worker process is spawned per worker and destroyed when the execution ends
+ # (PythonWorkflowWorker.postStop), so this dict holds the single warehouse that
+ # execution used and dies with the process. If PVMs ever become long-lived (e.g.
+ # pooled across executions), this needs the same bounding the JVM singleton has.
Review Comment:
I understand that you wanted to say python side fix is not needed but I don't think we need to add this comment in the codebase. The PR description is good enough to explain this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3778034439
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.core.storage
+
+import com.google.common.base.Ticker
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.Table
+import org.apache.iceberg.catalog.{Catalog, Namespace, TableIdentifier}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Duration
+
+/**
+ * Spec for the bounded catalog cache (#7290): only *idle-expired* entries are closed.
+ * A size-evicted catalog is dropped un-closed (it may be mid-operation under load),
+ * a catalog displaced by replaceInstance stays open (the replacing caller may still
+ * hold and later restore it -- wrap-and-restore, as amber's integration
+ * IcebergDocumentSpec does), and holders resolve their catalog per operation so a
+ * replacement is visible immediately.
+ *
+ * Size and idle eviction are exercised on isolated caches built through the
+ * package-private factory (with a manual ticker), never on the JVM-wide cache
+ * that parallel suites share. Tests that do touch the shared cache use their
+ * own spec-unique warehouse keys.
+ */
+class IcebergCatalogInstanceSpec extends AnyFlatSpec with Matchers {
+
+ /** A closable catalog stub; the cache only ever needs `close()` on eviction. */
+ private class FakeCatalog(catalogName: String) extends Catalog with AutoCloseable {
+ @volatile var closed = false
+ override def close(): Unit = closed = true
+ override def name(): String = catalogName
+ override def listTables(namespace: Namespace): java.util.List[TableIdentifier] =
+ throw new UnsupportedOperationException
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean =
+ throw new UnsupportedOperationException
+ override def renameTable(from: TableIdentifier, to: TableIdentifier): Unit =
+ throw new UnsupportedOperationException
+ override def loadTable(identifier: TableIdentifier): Table =
+ throw new UnsupportedOperationException
+ }
+
+ /** A ticker the tests advance by hand, making idle expiry deterministic. */
+ private class ManualTicker extends Ticker {
+ @volatile private var nanos = 0L
+ def advance(duration: Duration): Unit = nanos += duration.toNanos
+ override def read(): Long = nanos
+ }
+
+ "the catalog cache" should "drop entries beyond the size bound without closing them" in {
+ // Size pressure means more simultaneously hot warehouses than the bound; the
+ // evicted catalog may be mid-operation, so it must decay via GC, never be closed.
+ val cache =
+ IcebergCatalogInstance.buildCatalogCache(2, Duration.ofMinutes(60), new ManualTicker)
+ val fakes = (1 to 3).map(i => new FakeCatalog(s"size-$i"))
+
+ fakes.zipWithIndex.foreach { case (fake, i) => cache.put(s"warehouse-$i", fake) }
+
+ cache.size() should be <= 2L
+ fakes.count(_.closed) shouldBe 0
+ }
+
+ it should "close an entry left idle beyond the expiry window" in {
+ val ticker = new ManualTicker
+ val cache = IcebergCatalogInstance.buildCatalogCache(64, Duration.ofMinutes(60), ticker)
+ val idle = new FakeCatalog("idle")
+ cache.put("idle", idle)
+
+ ticker.advance(Duration.ofMinutes(61))
+ // Reads alone may defer removal processing; cleanUp() drains it deterministically.
+ cache.cleanUp()
+
+ cache.getIfPresent("idle") shouldBe null
+ idle.closed shouldBe true
+ }
+
+ it should "surface loader failures with their original exception type" in {
+ val cache =
+ IcebergCatalogInstance.buildCatalogCache(64, Duration.ofMinutes(60), new ManualTicker)
+
+ // Guava wraps a runtime failure in UncheckedExecutionException and a checked one
+ // in ExecutionException; getOrLoad must rethrow the original in both cases.
+ val runtimeFailure = intercept[IllegalArgumentException] {
+ IcebergCatalogInstance.getOrLoad(
+ cache,
+ "unsupported",
+ () => throw new IllegalArgumentException("Unsupported catalog type")
+ )
+ }
+ runtimeFailure.getMessage should include("Unsupported catalog type")
+
+ an[java.io.IOException] should be thrownBy
+ IcebergCatalogInstance.getOrLoad(
+ cache,
+ "unreachable",
+ () => throw new java.io.IOException("connection refused")
+ )
+ }
+
+ "getInstance" should "return the catalog installed for its warehouse" in {
+ val installed = new FakeCatalog("installed")
+ IcebergCatalogInstance.replaceInstance(installed, Some("catalog-cache-spec-get"))
+
+ IcebergCatalogInstance.getInstance(Some("catalog-cache-spec-get")) should be theSameInstanceAs
+ installed
+ }
+
+ "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")
+ IcebergCatalogInstance.replaceInstance(original, Some("catalog-cache-spec-replace"))
+
+ IcebergCatalogInstance.replaceInstance(wrapper, Some("catalog-cache-spec-replace"))
+ original.closed shouldBe false
+
+ IcebergCatalogInstance.replaceInstance(original, Some("catalog-cache-spec-replace"))
+ wrapper.closed shouldBe false
+ IcebergCatalogInstance.getInstance(
+ Some("catalog-cache-spec-replace")
+ ) should be theSameInstanceAs
+ original
+ }
+
+ it should "keep a re-registered shared instance open" in {
+ // LocalHadoopIcebergCatalog.ensure re-puts one shared instance from every suite
+ // (and under several warehouse names); none of that may close it.
+ val shared = new FakeCatalog("shared")
+ IcebergCatalogInstance.replaceInstance(shared, Some("catalog-cache-spec-idempotent"))
+
+ IcebergCatalogInstance.replaceInstance(shared, Some("catalog-cache-spec-idempotent"))
+
+ shared.closed shouldBe false
+ IcebergCatalogInstance.getInstance(Some("catalog-cache-spec-idempotent")) should
+ be theSameInstanceAs shared
+ }
+
+ "IcebergDocument.clear" should "address one catalog for the whole check-then-drop" in {
+ // Per-use resolution means per logical operation, not per call: the fake below
+ // swaps the cache entry from INSIDE the existence check, and the drop must still
+ // land on the catalog the operation started with (#7290 review, round 2).
+ class ImpostorCatalog extends FakeCatalog("impostor") {
+ @volatile var dropCalls = 0
+ override def tableExists(identifier: TableIdentifier): Boolean = true
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean = {
+ dropCalls += 1; true
+ }
+ }
+ class SwappingCatalog(impostor: ImpostorCatalog) extends FakeCatalog("swapping") {
+ @volatile var dropCalls = 0
+ override def tableExists(identifier: TableIdentifier): Boolean = {
+ IcebergCatalogInstance.replaceInstance(impostor, Some("catalog-cache-spec-clear"))
+ true
+ }
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean = {
+ dropCalls += 1; true
+ }
+ }
+ val impostor = new ImpostorCatalog
+ val swapping = new SwappingCatalog(impostor)
+ IcebergCatalogInstance.replaceInstance(swapping, Some("catalog-cache-spec-clear"))
+ val amberSchema = Schema().add("id", AttributeType.INTEGER)
+ val document = new IcebergDocument[Tuple](
+ "catalog_cache_spec",
+ "clear_probe",
+ IcebergUtil.toIcebergSchema(amberSchema),
+ IcebergUtil.toGenericRecord,
+ (schema, record) => IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema)),
+ Some("catalog-cache-spec-clear")
+ )
+
+ document.clear()
+
+ swapping.dropCalls shouldBe 1
+ impostor.dropCalls shouldBe 0
+ }
+
+ "IcebergDocument" should "resolve its catalog per use, seeing a replacement immediately" in {
+ // Pins the per-use `def` (#7290): a `lazy val` would keep returning the catalog
+ // that was current at first access, i.e. a reference the cache may have closed.
+ val amberSchema = Schema().add("id", AttributeType.INTEGER)
+ val document = new IcebergDocument[Tuple](
+ "catalog_cache_spec",
+ "swap_probe",
+ IcebergUtil.toIcebergSchema(amberSchema),
+ IcebergUtil.toGenericRecord,
+ (schema, record) => IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema)),
Review Comment:
Done in 22afc3e6b — a local helper installs the catalog and reads it back, so each case is one line.
##########
amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py:
##########
@@ -35,6 +35,11 @@ class IcebergCatalogInstance:
testing or reconfiguration.
"""
+ # Deliberately unbounded, unlike the Scala side's bounded cache (#7290): a Python
+ # worker process is spawned per worker and destroyed when the execution ends
+ # (PythonWorkflowWorker.postStop), so this dict holds the single warehouse that
+ # execution used and dies with the process. If PVMs ever become long-lived (e.g.
+ # pooled across executions), this needs the same bounding the JVM singleton has.
Review Comment:
It documents a deliberate asymmetry this PR creates: the Scala registry becomes bounded while the Python one stays a plain dict, and the natural reading is that Python was missed. It wasn't — a PVM is spawned per worker and destroyed when the execution ends, so its dict holds one warehouse and dies with the process. The last sentence is the load-bearing one: it names the assumption ("PVMs are short-lived") that makes "unbounded" safe, right above the code someone would change if PVMs were ever pooled across executions.
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
Review Comment:
The wrapping was scalafmt's, not a style choice — the renamed helper pushed the line past `maxColumn = 100`. Shortened the names in 22afc3e6b so it fits on one line again and the diff stays minimal. The rename itself is load-bearing though: the spy counted `table.refresh()`, and after this PR the reader never calls `refresh()` — it re-resolves the table instead — so the assertion would have been vacuously true (0 <= 4) and could no longer catch a lazy-advancement regression. It now counts `loadTable`, where those metadata loads surface.
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
val readerDoc = getDocument
try {
val retrieved = readerDoc.get().toList
assert(
retrieved.toSet == items.toSet,
"All records from both files should be read correctly"
)
- // With lazy file advancement seekToUsableFile() (and therefore table.refresh()) is called:
- // once on iterator creation, once when the last file is exhausted → 2 total.
- // Without the fix it would be called once per hasNext() on the last file → O(batchSize).
+ // With lazy file advancement the table is (re-)resolved once at iterator
+ // construction plus once per seekToUsableFile — construction seek and the
+ // final exhausted-files seek → 3 total. Without lazy advancement it would be
+ // once per hasNext() on the last file → O(batchSize).
assert(
- refreshCount.get() <= 4,
- s"table.refresh() should be called at most 4 times (lazy advancement), but was ${refreshCount.get()}"
+ metadataLoadCount.get() <= 4,
+ s"the table should be loaded at most 4 times (lazy advancement), but was ${metadataLoadCount.get()}"
Review Comment:
Same rename as above (22afc3e6b). The comment did need correcting on its own account: it said 3 total, which went stale when the iterator's eager constructor-time load was dropped — it is 2 now (construction seek + final exhausted-files seek).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5249375164
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- **Contributors with relevant context:** `@Yicong-Huang`, `@aglinxinyuan`
You can notify them by mentioning `@Yicong-Huang`, `@aglinxinyuan` in a comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5260719052
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 8ab409c84 — 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 merged PR #7539:
URL: https://github.com/apache/texera/pull/7539
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "kunwp1 (via GitHub)" <gi...@apache.org>.
kunwp1 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3777587715
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:
##########
@@ -0,0 +1,220 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.core.storage
+
+import com.google.common.base.Ticker
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.Table
+import org.apache.iceberg.catalog.{Catalog, Namespace, TableIdentifier}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Duration
+
+/**
+ * Spec for the bounded catalog cache (#7290): only *idle-expired* entries are closed.
+ * A size-evicted catalog is dropped un-closed (it may be mid-operation under load),
+ * a catalog displaced by replaceInstance stays open (the replacing caller may still
+ * hold and later restore it -- wrap-and-restore, as amber's integration
+ * IcebergDocumentSpec does), and holders resolve their catalog per operation so a
+ * replacement is visible immediately.
+ *
+ * Size and idle eviction are exercised on isolated caches built through the
+ * package-private factory (with a manual ticker), never on the JVM-wide cache
+ * that parallel suites share. Tests that do touch the shared cache use their
+ * own spec-unique warehouse keys.
+ */
+class IcebergCatalogInstanceSpec extends AnyFlatSpec with Matchers {
+
+ /** A closable catalog stub; the cache only ever needs `close()` on eviction. */
+ private class FakeCatalog(catalogName: String) extends Catalog with AutoCloseable {
+ @volatile var closed = false
+ override def close(): Unit = closed = true
+ override def name(): String = catalogName
+ override def listTables(namespace: Namespace): java.util.List[TableIdentifier] =
+ throw new UnsupportedOperationException
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean =
+ throw new UnsupportedOperationException
+ override def renameTable(from: TableIdentifier, to: TableIdentifier): Unit =
+ throw new UnsupportedOperationException
+ override def loadTable(identifier: TableIdentifier): Table =
+ throw new UnsupportedOperationException
+ }
+
+ /** A ticker the tests advance by hand, making idle expiry deterministic. */
+ private class ManualTicker extends Ticker {
+ @volatile private var nanos = 0L
+ def advance(duration: Duration): Unit = nanos += duration.toNanos
+ override def read(): Long = nanos
+ }
+
+ "the catalog cache" should "drop entries beyond the size bound without closing them" in {
+ // Size pressure means more simultaneously hot warehouses than the bound; the
+ // evicted catalog may be mid-operation, so it must decay via GC, never be closed.
+ val cache =
+ IcebergCatalogInstance.buildCatalogCache(2, Duration.ofMinutes(60), new ManualTicker)
+ val fakes = (1 to 3).map(i => new FakeCatalog(s"size-$i"))
+
+ fakes.zipWithIndex.foreach { case (fake, i) => cache.put(s"warehouse-$i", fake) }
+
+ cache.size() should be <= 2L
+ fakes.count(_.closed) shouldBe 0
+ }
+
+ it should "close an entry left idle beyond the expiry window" in {
+ val ticker = new ManualTicker
+ val cache = IcebergCatalogInstance.buildCatalogCache(64, Duration.ofMinutes(60), ticker)
+ val idle = new FakeCatalog("idle")
+ cache.put("idle", idle)
+
+ ticker.advance(Duration.ofMinutes(61))
+ // Reads alone may defer removal processing; cleanUp() drains it deterministically.
+ cache.cleanUp()
+
+ cache.getIfPresent("idle") shouldBe null
+ idle.closed shouldBe true
+ }
+
+ it should "surface loader failures with their original exception type" in {
+ val cache =
+ IcebergCatalogInstance.buildCatalogCache(64, Duration.ofMinutes(60), new ManualTicker)
+
+ // Guava wraps a runtime failure in UncheckedExecutionException and a checked one
+ // in ExecutionException; getOrLoad must rethrow the original in both cases.
+ val runtimeFailure = intercept[IllegalArgumentException] {
+ IcebergCatalogInstance.getOrLoad(
+ cache,
+ "unsupported",
+ () => throw new IllegalArgumentException("Unsupported catalog type")
+ )
+ }
+ runtimeFailure.getMessage should include("Unsupported catalog type")
+
+ an[java.io.IOException] should be thrownBy
+ IcebergCatalogInstance.getOrLoad(
+ cache,
+ "unreachable",
+ () => throw new java.io.IOException("connection refused")
+ )
+ }
+
+ "getInstance" should "return the catalog installed for its warehouse" in {
+ val installed = new FakeCatalog("installed")
+ IcebergCatalogInstance.replaceInstance(installed, Some("catalog-cache-spec-get"))
+
+ IcebergCatalogInstance.getInstance(Some("catalog-cache-spec-get")) should be theSameInstanceAs
+ installed
+ }
+
+ "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")
+ IcebergCatalogInstance.replaceInstance(original, Some("catalog-cache-spec-replace"))
+
+ IcebergCatalogInstance.replaceInstance(wrapper, Some("catalog-cache-spec-replace"))
+ original.closed shouldBe false
+
+ IcebergCatalogInstance.replaceInstance(original, Some("catalog-cache-spec-replace"))
+ wrapper.closed shouldBe false
+ IcebergCatalogInstance.getInstance(
+ Some("catalog-cache-spec-replace")
+ ) should be theSameInstanceAs
+ original
+ }
+
+ it should "keep a re-registered shared instance open" in {
+ // LocalHadoopIcebergCatalog.ensure re-puts one shared instance from every suite
+ // (and under several warehouse names); none of that may close it.
+ val shared = new FakeCatalog("shared")
+ IcebergCatalogInstance.replaceInstance(shared, Some("catalog-cache-spec-idempotent"))
+
+ IcebergCatalogInstance.replaceInstance(shared, Some("catalog-cache-spec-idempotent"))
+
+ shared.closed shouldBe false
+ IcebergCatalogInstance.getInstance(Some("catalog-cache-spec-idempotent")) should
+ be theSameInstanceAs shared
+ }
+
+ "IcebergDocument.clear" should "address one catalog for the whole check-then-drop" in {
+ // Per-use resolution means per logical operation, not per call: the fake below
+ // swaps the cache entry from INSIDE the existence check, and the drop must still
+ // land on the catalog the operation started with (#7290 review, round 2).
+ class ImpostorCatalog extends FakeCatalog("impostor") {
+ @volatile var dropCalls = 0
+ override def tableExists(identifier: TableIdentifier): Boolean = true
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean = {
+ dropCalls += 1; true
+ }
+ }
+ class SwappingCatalog(impostor: ImpostorCatalog) extends FakeCatalog("swapping") {
+ @volatile var dropCalls = 0
+ override def tableExists(identifier: TableIdentifier): Boolean = {
+ IcebergCatalogInstance.replaceInstance(impostor, Some("catalog-cache-spec-clear"))
+ true
+ }
+ override def dropTable(identifier: TableIdentifier, purge: Boolean): Boolean = {
+ dropCalls += 1; true
+ }
+ }
+ val impostor = new ImpostorCatalog
+ val swapping = new SwappingCatalog(impostor)
+ IcebergCatalogInstance.replaceInstance(swapping, Some("catalog-cache-spec-clear"))
+ val amberSchema = Schema().add("id", AttributeType.INTEGER)
+ val document = new IcebergDocument[Tuple](
+ "catalog_cache_spec",
+ "clear_probe",
+ IcebergUtil.toIcebergSchema(amberSchema),
+ IcebergUtil.toGenericRecord,
+ (schema, record) => IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema)),
+ Some("catalog-cache-spec-clear")
+ )
+
+ document.clear()
+
+ swapping.dropCalls shouldBe 1
+ impostor.dropCalls shouldBe 0
+ }
+
+ "IcebergDocument" should "resolve its catalog per use, seeing a replacement immediately" in {
+ // Pins the per-use `def` (#7290): a `lazy val` would keep returning the catalog
+ // that was current at first access, i.e. a reference the cache may have closed.
+ val amberSchema = Schema().add("id", AttributeType.INTEGER)
+ val document = new IcebergDocument[Tuple](
+ "catalog_cache_spec",
+ "swap_probe",
+ IcebergUtil.toIcebergSchema(amberSchema),
+ IcebergUtil.toGenericRecord,
+ (schema, record) => IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema)),
Review Comment:
These two lines also appear to be duplicates. Can you refer to`IcebergDocumentSpec.scala` to see how to remove duplicates?
##########
amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py:
##########
@@ -35,6 +35,11 @@ class IcebergCatalogInstance:
testing or reconfiguration.
"""
+ # Deliberately unbounded, unlike the Scala side's bounded cache (#7290): a Python
+ # worker process is spawned per worker and destroyed when the execution ends
+ # (PythonWorkflowWorker.postStop), so this dict holds the single warehouse that
+ # execution used and dies with the process. If PVMs ever become long-lived (e.g.
+ # pooled across executions), this needs the same bounding the JVM singleton has.
Review Comment:
What's the purpose of adding these comments?
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
Review Comment:
Aren't these unnecessary changes?
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
val readerDoc = getDocument
try {
val retrieved = readerDoc.get().toList
assert(
retrieved.toSet == items.toSet,
"All records from both files should be read correctly"
)
- // With lazy file advancement seekToUsableFile() (and therefore table.refresh()) is called:
- // once on iterator creation, once when the last file is exhausted → 2 total.
- // Without the fix it would be called once per hasNext() on the last file → O(batchSize).
+ // With lazy file advancement the table is (re-)resolved once at iterator
+ // construction plus once per seekToUsableFile — construction seek and the
+ // final exhausted-files seek → 3 total. Without lazy advancement it would be
+ // once per hasNext() on the last file → O(batchSize).
assert(
- refreshCount.get() <= 4,
- s"table.refresh() should be called at most 4 times (lazy advancement), but was ${refreshCount.get()}"
+ metadataLoadCount.get() <= 4,
+ s"the table should be loaded at most 4 times (lazy advancement), but was ${metadataLoadCount.get()}"
Review Comment:
Unnecessary changes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762263910
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
@@ -72,7 +72,10 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
private val lock = new ReentrantReadWriteLock()
- @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
+ // Resolved per use, never held: the catalog cache is bounded and closes evicted
+ // entries (#7290), so a pinned reference could outlive its catalog. A public def
+ // (not a lazy val) also means a replaced/rebuilt catalog is picked up immediately.
+ def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
Resolving this accessor on every call makes `clear()` non-atomic: its `tableExists` and `dropTable` calls each invoke `catalog`, so a concurrent `replaceInstance` can make the existence check run against catalog A and the drop run against catalog B. This can silently leave the table in A or delete the same identifier from the replacement catalog. Resolve once at the start of each logical operation (especially `clear`) and reuse that local catalog; replacement should become visible on the next operation, not midway through one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5249374495
<!-- backport-auto-label-report -->
### 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](https://github.com/apache/texera/actions/runs/31462096060)._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762242279
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -36,11 +39,34 @@ import scala.collection.mutable
* Only the REST catalog varies by warehouse; the hadoop and postgres catalogs are warehouse-agnostic
* and ignore the warehouse argument.
*
- * Access is synchronized because the same JVM serves multiple warehouses concurrently.
+ * The cache is bounded (#7290): per-user warehouses (#6870) make the set of catalogs a
+ * long-lived JVM touches unbounded, and each REST catalog holds an HTTP client. Entries
+ * fall out by size or idleness and are closed by the removal listener; the next access
+ * simply rebuilds one. Callers must therefore resolve their catalog per use instead of
+ * holding one across an execution (see IcebergDocument / IcebergTableWriter).
*/
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
- private val catalogs = mutable.Map.empty[String, Catalog]
+ // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent: generous enough
+ // that eviction never hits a warehouse in active use, small enough to bound the JVM.
+ private val CatalogCacheMaxSize = 64L
+ private val CatalogCacheExpireAfterAccessMinutes = 60L
+
+ private val catalogs: Cache[String, Catalog] = CacheBuilder
+ .newBuilder()
+ .maximumSize(CatalogCacheMaxSize)
+ .expireAfterAccess(CatalogCacheExpireAfterAccessMinutes, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener[String, Catalog] {
+ override def onRemoval(notification: RemovalNotification[String, Catalog]): Unit =
+ notification.getValue match {
+ case closeable: AutoCloseable =>
+ Try(closeable.close()).failed.foreach(error =>
+ logger.warn(s"failed to close evicted catalog '${notification.getKey}'", error)
Review Comment:
CI proved the replacement half of this right: amber's integration `IcebergDocumentSpec` wrap-and-restores the shared catalog around a spy, and closing the displaced instance handed every later suite a dead catalog (10 failures). Fixed in 98f88e787: the listener now closes only entries Guava actually evicted (`wasEvicted()`), never an explicitly replaced one — the replacing caller owns what it displaced. For the evicted path the earlier argument stands: a just-resolved entry is the most recently used, so mid-operation closure stays out of reach at this scale.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -102,8 +133,14 @@ object IcebergCatalogInstance {
* @param catalog the catalog to cache.
* @param warehouse the warehouse to cache it under; `None` uses the configured default.
*/
- def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit =
- synchronized {
- catalogs(cacheKey(warehouse.getOrElse(defaultWarehouse))) = catalog
+ def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit = {
+ val key = cacheKey(warehouse.getOrElse(defaultWarehouse))
+ // Guava reports a same-value put as a replacement, which would fire the removal
+ // listener and close a catalog that is still installed: the shared test catalog
+ // is ensure()d repeatedly (and under several names) by parallel suites. Skip the
+ // no-op re-put so only a genuine replacement closes the previous catalog.
+ if (catalogs.getIfPresent(key) ne catalog) {
+ catalogs.put(key, catalog)
}
Review Comment:
Superseded in 98f88e787: with the listener no longer closing on replacement, the compare-and-swap pair became unnecessary — `replaceInstance` is a single atomic `put` again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3778152626
##########
amber/src/main/python/core/storage/iceberg/iceberg_catalog_instance.py:
##########
@@ -35,6 +35,11 @@ class IcebergCatalogInstance:
testing or reconfiguration.
"""
+ # Deliberately unbounded, unlike the Scala side's bounded cache (#7290): a Python
+ # worker process is spawned per worker and destroyed when the execution ends
+ # (PythonWorkflowWorker.postStop), so this dict holds the single warehouse that
+ # execution used and dies with the process. If PVMs ever become long-lived (e.g.
+ # pooled across executions), this needs the same bounding the JVM singleton has.
Review Comment:
Fair enough — removed in 10783f13d, and the reasoning now lives in the PR description instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762467051
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
@@ -72,7 +72,10 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
private val lock = new ReentrantReadWriteLock()
- @transient lazy val catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
+ // Resolved per use, never held: the catalog cache is bounded and closes evicted
+ // entries (#7290), so a pinned reference could outlive its catalog. A public def
+ // (not a lazy val) also means a replaced/rebuilt catalog is picked up immediately.
+ def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
Done in babda0e8e — and the framing is right: per-use has to mean per logical operation, not per accessor call (the writer's `flushBuffer` already resolved once; `clear()` was the one multi-touch operation left). `clear()` now resolves a single catalog for its check-then-drop, pinned by a spec case whose fake swaps the cache entry from inside `tableExists` — the drop must still land on the catalog the operation started with. Every other method in the class resolves once per call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3755483996
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.core.storage
+
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.Table
+import org.apache.iceberg.catalog.{Catalog, Namespace, TableIdentifier}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Spec for the bounded catalog cache (#7290): a genuine replacement closes the
+ * catalog it displaces, a same-instance re-registration does not (that is what
+ * [[LocalHadoopIcebergCatalog.ensure]] relies on), and holders resolve their
+ * catalog per use so a replacement is visible immediately.
+ *
+ * Size-based eviction is deliberately not exercised: forcing it would flood the
+ * JVM-wide cache that parallel suites share and could evict their live catalog.
+ * The close-on-removal wiring it would exercise is pinned by the replacement
+ * cases below, which Guava routes through the same removal listener.
Review Comment:
The PR's core size and idle-expiration behavior remains untested; replacement only exercises the listener, so removing `maximumSize`, changing the limit/TTL, or breaking time-based eviction would still pass. Extract cache construction behind a package-private factory with configurable size/ticker, then test size and expiration on an isolated cache rather than mutating the JVM-wide singleton.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -36,11 +39,34 @@ import scala.collection.mutable
* Only the REST catalog varies by warehouse; the hadoop and postgres catalogs are warehouse-agnostic
* and ignore the warehouse argument.
*
- * Access is synchronized because the same JVM serves multiple warehouses concurrently.
+ * The cache is bounded (#7290): per-user warehouses (#6870) make the set of catalogs a
+ * long-lived JVM touches unbounded, and each REST catalog holds an HTTP client. Entries
+ * fall out by size or idleness and are closed by the removal listener; the next access
+ * simply rebuilds one. Callers must therefore resolve their catalog per use instead of
+ * holding one across an execution (see IcebergDocument / IcebergTableWriter).
*/
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
- private val catalogs = mutable.Map.empty[String, Catalog]
+ // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent: generous enough
+ // that eviction never hits a warehouse in active use, small enough to bound the JVM.
+ private val CatalogCacheMaxSize = 64L
+ private val CatalogCacheExpireAfterAccessMinutes = 60L
+
+ private val catalogs: Cache[String, Catalog] = CacheBuilder
+ .newBuilder()
+ .maximumSize(CatalogCacheMaxSize)
+ .expireAfterAccess(CatalogCacheExpireAfterAccessMinutes, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener[String, Catalog] {
+ override def onRemoval(notification: RemovalNotification[String, Catalog]): Unit =
+ notification.getValue match {
+ case closeable: AutoCloseable =>
+ Try(closeable.close()).failed.foreach(error =>
+ logger.warn(s"failed to close evicted catalog '${notification.getKey}'", error)
Review Comment:
Closing a removed entry immediately is unsafe because `Cache.get` returns an unpinned reference. A caller can obtain catalog A, then another warehouse insertion can evict and close A before `tableExists`/`loadTable` runs; loaded `Table` objects and iterators can also continue using A's REST client after eviction. Resolving on each access does not close this lifetime gap. Use a borrow/lease (or equivalent quiescent-retirement mechanism) and close an evicted catalog only after no in-flight catalog or derived table operation still uses it.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergTableWriter.scala:
##########
@@ -50,13 +52,17 @@ import scala.collection.mutable.ArrayBuffer
*/
private[storage] class IcebergTableWriter[T](
val writerIdentifier: String,
- val catalog: Catalog,
+ val warehouse: Option[String],
val tableNamespace: String,
val tableName: String,
val tableSchema: Schema,
val serde: (org.apache.iceberg.Schema, T) => Record
) extends BufferedItemWriter[T] {
+ // Resolved per use (#7290): the catalog cache is bounded and closes evicted entries,
+ // so the writer must not pin one across its lifetime.
+ private def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
This accessor is not actually resolved for each writer use: it is evaluated once by the eager `private val table` at line 76. That `Table` retains REST table operations backed by the catalog client, so eviction or replacement after writer construction can close the client before `flushBuffer` commits. Resolve the table for each flush, or retain a cache lease until the writer closes.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,9 +96,14 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ // get(key, loader) locks per key, not globally: a cache miss's REST config
+ // round trip no longer blocks lookups of other warehouses.
+ catalogs.get(
+ cacheKey(name),
+ new Callable[Catalog] {
+ override def call(): Catalog = createCatalog(name)
+ }
+ )
Review Comment:
`Cache.get` wraps loader failures (`UncheckedExecutionException` for runtime exceptions, `ExecutionException` for checked exceptions, and `ExecutionError` for errors). Previously `createCatalog` failures such as `RESTException` or the unsupported-type `IllegalArgumentException` propagated directly, so this silently changes `getInstance`'s error contract and can break existing error handling. Unwrap these Guava wrappers before returning.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -102,8 +133,14 @@ object IcebergCatalogInstance {
* @param catalog the catalog to cache.
* @param warehouse the warehouse to cache it under; `None` uses the configured default.
*/
- def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit =
- synchronized {
- catalogs(cacheKey(warehouse.getOrElse(defaultWarehouse))) = catalog
+ def replaceInstance(catalog: Catalog, warehouse: Option[String] = None): Unit = {
+ val key = cacheKey(warehouse.getOrElse(defaultWarehouse))
+ // Guava reports a same-value put as a replacement, which would fire the removal
+ // listener and close a catalog that is still installed: the shared test catalog
+ // is ensure()d repeatedly (and under several names) by parallel suites. Skip the
+ // no-op re-put so only a genuine replacement closes the previous catalog.
+ if (catalogs.getIfPresent(key) ne catalog) {
+ catalogs.put(key, catalog)
}
Review Comment:
This check-then-put is not atomic. If two threads re-register the same catalog while another value is installed, both can observe the old value; the first installs the shared catalog and the second `put` reports that same shared instance as replaced, causing the listener to close the value that remains cached. Use an atomic compare-and-replace loop and add a concurrent regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3778034854
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
Review Comment:
The rename is load-bearing: the spy counted `table.refresh()`, and after this PR the reader re-resolves the table instead of refreshing it — the assertion would have been vacuously true (0 <= 4) and could no longer catch a lazy-advancement regression. It now counts `loadTable`, where those metadata loads surface.
##########
amber/src/test/integration/org/apache/texera/amber/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -114,32 +113,33 @@ class IcebergDocumentSpec extends VirtualDocumentSpec[Tuple] with BeforeAndAfter
val (batch1, batch2) = items.splitAt(batchSize)
// Write two separate batches to produce two committed data files.
- // This also initialises `document.catalog` (lazy val) with the real catalog, which
- // is why we open a fresh reader document below after injecting the spy.
val writer1 = document.writer(UUID.randomUUID().toString)
writer1.open(); batch1.foreach(writer1.putOne); writer1.close()
val writer2 = document.writer(UUID.randomUUID().toString)
writer2.open(); batch2.foreach(writer2.putOne); writer2.close()
- val refreshCount = new AtomicInteger(0)
+ val metadataLoadCount = new AtomicInteger(0)
val realCatalog = IcebergCatalogInstance.getInstance()
- IcebergCatalogInstance.replaceInstance(catalogWithRefreshSpy(realCatalog, refreshCount))
- // Open a fresh reader: its `catalog` lazy val hasn't been initialised yet, so it
- // will pick up the spy catalog on first access inside seekToUsableFile.
+ IcebergCatalogInstance.replaceInstance(
+ catalogWithMetadataLoadSpy(realCatalog, metadataLoadCount)
+ )
+ // Open a fresh reader; it resolves its catalog per use (#7290), so every metadata
+ // load inside seekToUsableFile goes through the spy installed above.
val readerDoc = getDocument
try {
val retrieved = readerDoc.get().toList
assert(
retrieved.toSet == items.toSet,
"All records from both files should be read correctly"
)
- // With lazy file advancement seekToUsableFile() (and therefore table.refresh()) is called:
- // once on iterator creation, once when the last file is exhausted → 2 total.
- // Without the fix it would be called once per hasNext() on the last file → O(batchSize).
+ // With lazy file advancement the table is (re-)resolved once at iterator
+ // construction plus once per seekToUsableFile — construction seek and the
+ // final exhausted-files seek → 3 total. Without lazy advancement it would be
+ // once per hasNext() on the last file → O(batchSize).
assert(
- refreshCount.get() <= 4,
- s"table.refresh() should be called at most 4 times (lazy advancement), but was ${refreshCount.get()}"
+ metadataLoadCount.get() <= 4,
+ s"the table should be loaded at most 4 times (lazy advancement), but was ${metadataLoadCount.get()}"
Review Comment:
Same rename as above. The comment also needed correcting on its own: it said 3 total, stale since the iterator's eager constructor-time load was dropped — it is 2 now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on PR #7539:
URL: https://github.com/apache/texera/pull/7539#issuecomment-5284755864
On coverage: raised in 22afc3e6b with cases for the close-failure path (a catalog whose `close()` throws is still removed and nothing escapes), a catalog that is not `AutoCloseable` (hadoop/postgres need not be), and the `ExecutionError` unwrap. The one path that stays uncovered is the cache-miss loader actually building a catalog — that would need a live REST server, so covering it would make the unit suite depend on external infrastructure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(storage): bound the per-warehouse catalog cache and release evicted catalogs [texera]
Posted by "mengw15 (via GitHub)" <gi...@apache.org>.
mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3773840829
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -36,11 +39,34 @@ import scala.collection.mutable
* Only the REST catalog varies by warehouse; the hadoop and postgres catalogs are warehouse-agnostic
* and ignore the warehouse argument.
*
- * Access is synchronized because the same JVM serves multiple warehouses concurrently.
+ * The cache is bounded (#7290): per-user warehouses (#6870) make the set of catalogs a
+ * long-lived JVM touches unbounded, and each REST catalog holds an HTTP client. Entries
+ * fall out by size or idleness and are closed by the removal listener; the next access
+ * simply rebuilds one. Callers must therefore resolve their catalog per use instead of
+ * holding one across an execution (see IcebergDocument / IcebergTableWriter).
*/
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
- private val catalogs = mutable.Map.empty[String, Catalog]
+ // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent: generous enough
+ // that eviction never hits a warehouse in active use, small enough to bound the JVM.
+ private val CatalogCacheMaxSize = 64L
+ private val CatalogCacheExpireAfterAccessMinutes = 60L
+
+ private val catalogs: Cache[String, Catalog] = CacheBuilder
+ .newBuilder()
+ .maximumSize(CatalogCacheMaxSize)
+ .expireAfterAccess(CatalogCacheExpireAfterAccessMinutes, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener[String, Catalog] {
+ override def onRemoval(notification: RemovalNotification[String, Catalog]): Unit =
+ notification.getValue match {
+ case closeable: AutoCloseable =>
+ Try(closeable.close()).failed.foreach(error =>
+ logger.warn(s"failed to close evicted catalog '${notification.getKey}'", error)
Review Comment:
Follow-up in 239c4971e: the size-eviction half of this concern is now retired entirely — only idle-expired entries are closed (`RemovalCause.EXPIRED`). A size-evicted catalog is dropped un-closed: it lives only as long as its in-flight operations (per-operation resolution bounds every borrow), then GC reclaims it while the server's keepalive timeout severs its idle connections. Overload therefore degrades into rebuild churn instead of closing a catalog mid-operation; the only remaining close targets entries untouched for 60 minutes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org