Core: Prevent stale CachingCatalog entries after commit

[ysung]

Summary

• reconcile an exact BaseTable cache entry after a successful commit: retain the committing table, but remove a different table loaded while the write was in flight
• rely on Caffeine's atomic same-key load/remapping ordering instead of maintaining a second in-flight-load registry
• retain metadata tables when their exact origin table is preserved, but evict metadata backed by a stale replacement
• prevent metadata-table post-processing from publishing after either its raw entry or origin table was invalidated
• conservatively invalidate local and wrapped-catalog state after CommitStateUnknownException

Change size

• production/business logic: 238 changed lines (+224/-14, net +210) in CachingCatalog
• tests: 358 changed lines (+351/-7, net +344) in TestCachingCatalog
• 60% of the changed lines are deterministic regression coverage

Root cause and fix

The stale-read sequence in #17338 is:

1. a writer holds table metadata at snapshot N
2. the cache entry expires
3. another caller reloads and caches snapshot N while the write is still in progress
4. the writer commits snapshot N+1
5. the pre-commit table remains cached until its new TTL expires

After a successful commit, this revision performs a same-key compute:

• if the current entry is the exact table that performed the commit, retain it
• if a different table is present, remove it because it may have loaded snapshot N during the write
• if no entry is present, leave the cache empty

Caffeine orders this remapping with loads of the same key. If a pre-commit load is already in flight, the remapping waits for it to publish and then removes its result. If the remapping runs first, a later load begins after the delegate commit has completed and reads committed state.

The callback only compares object references and returns the same value or null; it does not access or mutate another cache entry. Metadata-table invalidation runs after the callback. This avoids the nested cache-update pattern involved in #3791 and #3801.

If reconciliation retains the exact committing table, its cached metadata tables also remain valid because they share that table's operations. If reconciliation removes an absent or different origin entry, metadata tables are evicted because they may be backed by stale replacement operations. Caffeine dispatches any removal-listener cascade after the underlying map remapping returns, so it is not nested in the mapping callback.

Retaining the committing table preserves identity for clients such as Spark's relation cache. Unconditional invalidation caused TestCherrypickSnapshotProcedure.testCherrypickSnapshotRefreshesRelationCache to keep serving the pre-cherry-pick cached result.

Metadata-table publication

CachingCatalog first loads a raw metadata table, then separately loads the origin table and reconstructs the metadata table with the origin's TableOperations. That reconstruction occurs after the original cache load. This revision publishes the reconstructed metadata table through a same-key compute only if:

• the raw metadata entry is still the exact current entry
• the origin table is still the exact current origin entry

The callback's only lookup is getIfPresentQuietly on the origin key. It may return null to remove its own metadata-table entry; that removal has cause EXPLICIT, while the removal listener only cascades on EXPIRED. It deliberately does not record a cache access, load, invalidate, or otherwise mutate another cache key.

Scope

This focused PR installs commit tracking only when the loaded table's concrete class is exactly BaseTable. Other table implementations keep their existing caching behavior and are outside this PR's commit-aware guarantee.

Draft follow-up #17401 adds validated operations replacement for specialized table implementations, including RESTTable, and defines the unsupported-subtype policy. Keeping that API decision separate lets maintainers review this correctness mechanism without simultaneously accepting a new public SPI.

If a commit throws CommitStateUnknownException, the commit may have landed. The wrapper therefore invalidates local state and best-effort invalidates the wrapped catalog before rethrowing the original exception. Failures from either invalidation are attached as suppressed rather than masking the commit-state exception. A local reconciliation failure after a successful commit is logged rather than changing a durable success into an apparent commit failure.

Runtime behavior and tradeoffs

• the exact BaseTable that successfully commits remains cached, preserving its identity
• retaining that entry refreshes its cache-expiration age
• a table loaded during the write is evicted; a load that starts after the durable commit but before reconciliation may also be conservatively evicted and reloaded
• cached metadata tables retain their identity when the exact committing origin remains cached
• metadata tables backed by an absent or replaced origin are evicted and reload on their next access
• a commit can briefly wait for an in-flight load of the same cache key to complete
• operations() on tracked BaseTable instances returns the commit-tracking wrapper rather than the delegate operations' concrete runtime class
• specialized and custom table implementations are unchanged in this PR; Core: Support CachingCatalog commit tracking for specialized tables #17401 covers them

Related work

• Core: Invalidate cached table on commit to prevent stale reads #17341 uses unconditional post-commit invalidation. That closes the stale-table race, but it also changes the cached table identity. This revision uses identity-aware reconciliation to retain the committing entry while removing a concurrent replacement.
• Core: Fix race condition in CachingCatalog loadTable (#17338) #17375 coordinates metadata loading through a nested mutation from inside an outer cache remapping callback. A reviewer requested a redesign because that can recreate the recursive-update/deadlock class removed by Core: fix deadlock in CachingCatalog #3801 after Deadlock when using CachingCatalog with Iceberg 0.13.0-SNAPSHOT #3791. This PR does not load or mutate another cache key from a cache mapping callback.

Validation

• ./gradlew :iceberg-core:test --tests org.apache.iceberg.hadoop.TestCachingCatalog
• ./gradlew :iceberg-core:spotlessApply
• deterministic regression where an in-flight stale data-table load publishes before post-commit reconciliation
• regression coverage that preserves metadata-table identity with the committing origin but evicts metadata backed by a concurrent stale replacement
• deterministic regression where invalidation occurs after metadata reconstruction has loaded its origin but before it publishes
• CommitStateUnknownException, including failure of wrapped-catalog invalidation
• full combined implementation passed all 37 GitHub checks before the split; this rewritten PR will receive a fresh CI run

Part of #17338

---

AI Disclosure

• Model: GPT-5
• Platform/Tool: Codex
• Human Oversight: partially reviewed
• Prompt Summary: Reconcile exact BaseTable cache entries after commit without recursive cache mutation and add deterministic concurrent regression coverage.

[RussellSpitzer]

This PR Description is not human friendly, I would strongly recommend summarizing it down to a paragraph or two to explain the problem and what you are doing to solve it.

[RussellSpitzer]

So the underlying issue here is that the CachingCatalog isn't actually racing, we are intentionally serving bounded-stale results in order to avoid a serializability problem.

The basic problem is this:

• I have a query that refers to table "X" several times.
• Spark can load "X" separately for different references to it (self-joins, CTEs reused across executions, a lazily-resolved plan re-touched at execution time).
• If "X" is at a different state when a subsequent load comes in, you essentially get two different "X" tables in the same query plan.
• This leads to a broken plan that produces very wrong results.

So the CachingCatalog exists (partly) to avoid this scenario. By guaranteeing that for a certain amount of time (the cache TTL) we ignore changes made by other threads, we avoid ending up with the same table read at multiple points in time within a single query. The guarantee here is bounded staleness, not freshness, and that's a deliberate trade, not a bug. This is a structural problem in Spark that folks are still trying to fix even now.

Because of this, I don't think we can "fix" this the way the PR proposes. Evicting or reconciling the entry on commit reintroduces exactly the two-snapshots-in-one-plan failure the cache exists to suppress; it isn't a narrower or safer variant of invalidate-on-commit. (Retaining the committing instance doesn't help here either, since callers re-read currentSnapshot() on every load.)

For users who understand the risk and want always-fresh reads, they can disable the cache (cache-enabled=false) and have the catalog reload on every access. Notably, the Iceberg source itself and the incremental-query tests already set cache-enabled=false for this reason.

[xndai]

If "X" is at a different state when a subsequent load comes in, you essentially get two different "X" tables in the same query plan. This leads to a broken plan that produces very wrong results. 

So the CachingCatalog exists (partly) to avoid this scenario.

I think this is totally a client's responsibility and client's choice. When to refresh "X" completely depends on the transaction semantics provided by the client. Say if client does "Read Committed", then they can refresh any time they want as long as they read the committed snapshots. And if the semantics is "Snapshot Read", then the snapshot "X" needs to remain unchanged for the duration of the transaction. To force a refresh on server side using a TTL seems wrong to me.