From 06f9a4a672d5de0c6672df0a7f93aa58cf173a32 Mon Sep 17 00:00:00 2001 From: Yun-Chieh Sung Date: Tue, 28 Jul 2026 10:13:12 -0700 Subject: [PATCH 1/2] Core: Reconcile caching catalog entries after commit Generated-by: Codex Co-authored-by: Humzah Kiani Co-authored-by: Codex (GPT-5) --- .../org/apache/iceberg/CachingCatalog.java | 238 +++++++++++- .../iceberg/hadoop/TestCachingCatalog.java | 358 +++++++++++++++++- 2 files changed, 575 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/CachingCatalog.java b/core/src/main/java/org/apache/iceberg/CachingCatalog.java index 5c92fe7db602..d71a1c95c90b 100644 --- a/core/src/main/java/org/apache/iceberg/CachingCatalog.java +++ b/core/src/main/java/org/apache/iceberg/CachingCatalog.java @@ -31,7 +31,11 @@ import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.LocationProvider; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.slf4j.Logger; @@ -144,13 +148,13 @@ public Table loadTable(TableIdentifier ident) { return cached; } - Table table = tableCache.get(canonicalized, catalog::loadTable); + Table table = tableCache.get(canonicalized, this::loadTableForCache); if (table instanceof BaseMetadataTable) { // Cache underlying table TableIdentifier originTableIdentifier = TableIdentifier.of(canonicalized.namespace().levels()); - Table originTable = tableCache.get(originTableIdentifier, catalog::loadTable); + Table originTable = tableCache.get(originTableIdentifier, this::loadTableForCache); // Share TableOperations instance of origin table for all metadata tables, so that metadata // table instances are refreshed as well when origin table instance is refreshed. @@ -161,14 +165,188 @@ public Table loadTable(TableIdentifier ident) { Table metadataTable = MetadataTableUtils.createMetadataTableInstance( ops, catalog.name(), originTableIdentifier, canonicalized, type); - tableCache.put(canonicalized, metadataTable); - return metadataTable; + return publishMetadataTable( + canonicalized, table, originTableIdentifier, originTable, metadataTable); } } return table; } + private Table loadTableForCache(TableIdentifier identifier) { + Table loaded = catalog.loadTable(identifier); + if (loaded instanceof BaseMetadataTable) { + return loaded; + } + + return prepareTable(identifier, loaded); + } + + private Table publishMetadataTable( + TableIdentifier identifier, + Table initiallyLoaded, + TableIdentifier originIdentifier, + Table originTable, + Table metadataTable) { + Table published = + tableCache + .asMap() + .compute( + identifier, + (ignored, current) -> { + // If this entry was invalidated or replaced after its initial load, do not + // resurrect or overwrite it. + if (current != initiallyLoaded) { + return current; + } + + // Do not replace this quiet lookup with getIfPresent. Recording an access can + // schedule cache maintenance while this mapping callback is running. + Table currentOrigin = tableCache.policy().getIfPresentQuietly(originIdentifier); + if (currentOrigin != originTable) { + // Invalidation removes the origin entry before its metadata entries. If the + // origin changed while this metadata table was being prepared, remove the + // initially loaded entry. + return null; + } + + return metadataTable; + }); + return published != null ? published : metadataTable; + } + + /** + * Installs commit-tracking operations on a table handed out by this cache, so that a commit + * performed through the table removes a different table cached during the write (and invalidates + * its metadata tables). Otherwise a table that was reloaded during an in-flight write keeps + * serving the pre-commit snapshot until the entry expires. See + * https://github.com/apache/iceberg/issues/17338. + * + *

A plain {@link BaseTable} is re-created directly with the tracking operations. Other table + * implementations are returned unchanged. Metadata tables share the safely prepared origin + * table's operations and are invalidated together with it. + */ + private Table prepareTable(TableIdentifier identifier, Table table) { + if (table.getClass() == BaseTable.class) { + BaseTable baseTable = (BaseTable) table; + TableOperations operations = baseTable.operations(); + CacheInvalidatingTableOperations replacementOperations = + new CacheInvalidatingTableOperations(operations, identifier); + Table wrapped = new BaseTable(replacementOperations, baseTable.name(), baseTable.reporter()); + replacementOperations.track(wrapped); + return wrapped; + } + + return table; + } + + /** + * A {@link TableOperations} wrapper that reconciles the cached table entry after a commit. It + * preserves the committing table when it is still cached, but removes an entry that was loaded + * concurrently during the write. This ensures that a subsequent {@link + * #loadTable(TableIdentifier)} observes the commit without unnecessarily changing the table + * identity used by clients such as Spark's relation cache. Reconciliation can wait for an + * in-progress load of the same table, and retaining the committing entry refreshes its cache + * expiration age. See #17338. + */ + private class CacheInvalidatingTableOperations implements TableOperations { + private final TableOperations delegate; + private final TableIdentifier identifier; + private volatile Table trackedTable; + + private CacheInvalidatingTableOperations(TableOperations delegate, TableIdentifier identifier) { + this.delegate = delegate; + this.identifier = identifier; + } + + private void track(Table table) { + Preconditions.checkState(trackedTable == null, "Cannot replace the tracked table"); + this.trackedTable = Preconditions.checkNotNull(table, "Invalid table: null"); + } + + @Override + public TableMetadata current() { + return delegate.current(); + } + + @Override + public TableMetadata refresh() { + return delegate.refresh(); + } + + @Override + public void commit(TableMetadata base, TableMetadata metadata) { + try { + delegate.commit(base, metadata); + } catch (CommitStateUnknownException e) { + // The commit may have succeeded, so the cached table may be stale. Invalidate before + // rethrowing so that a subsequent load re-reads the table metadata. + try { + invalidateLocalCache(identifier); + } catch (RuntimeException invalidationFailure) { + e.addSuppressed(invalidationFailure); + } + + try { + catalog.invalidateTable(identifier); + } catch (RuntimeException invalidationFailure) { + e.addSuppressed(invalidationFailure); + } + + throw e; + } + + try { + reconcileLocalCacheAfterCommit(identifier, trackedTable); + } catch (RuntimeException e) { + // The commit is durable. Do not turn an invalidation failure into an apparent commit + // failure, which could cause callers to clean up files that are now referenced by the + // table. + LOG.warn( + "Failed to reconcile CachingCatalog entry {} after a successful commit; " + + "the entry may remain stale until it is evicted or explicitly invalidated", + identifier, + e); + } + } + + @Override + public FileIO io() { + return delegate.io(); + } + + @Override + public EncryptionManager encryption() { + return delegate.encryption(); + } + + @Override + public String metadataFileLocation(String fileName) { + return delegate.metadataFileLocation(fileName); + } + + @Override + public LocationProvider locationProvider() { + return delegate.locationProvider(); + } + + @Override + public TableOperations temp(TableMetadata uncommittedMetadata) { + // Temporary operations are not used to commit, so they do not need commit tracking. + return delegate.temp(uncommittedMetadata); + } + + @Override + public long newSnapshotId() { + return delegate.newSnapshotId(); + } + + @Override + public boolean requireStrictCleanup() { + return delegate.requireStrictCleanup(); + } + } + @Override public boolean dropTable(TableIdentifier ident, boolean purge) { boolean dropped = catalog.dropTable(ident, purge); @@ -185,14 +363,40 @@ public void renameTable(TableIdentifier from, TableIdentifier to) { @Override public void invalidateTable(TableIdentifier ident) { catalog.invalidateTable(ident); - TableIdentifier canonicalized = canonicalizeIdentifier(ident); + invalidateLocalCache(canonicalizeIdentifier(ident)); + } + + private void invalidateLocalCache(TableIdentifier canonicalized) { tableCache.invalidate(canonicalized); tableCache.invalidateAll(metadataTableIdentifiers(canonicalized)); } + private void reconcileLocalCacheAfterCommit( + TableIdentifier canonicalized, Table committingTable) { + // This same-key remapping is ordered with loads of the table. Keep the committing instance if + // it is still cached; otherwise remove the table loaded while the commit was in flight. The + // callback intentionally performs no other cache operation (see #3791). + Table reconciled = + tableCache + .asMap() + .compute( + canonicalized, + (ignored, cachedTable) -> cachedTable == committingTable ? cachedTable : null); + + if (reconciled != committingTable) { + // Keep metadata invalidation outside the remapping callback to avoid recursive cache updates. + // If compute removes an expired origin entry, Caffeine also dispatches its removal listener + // after the underlying map remapping returns, so that listener's cascade is not nested here. + tableCache.invalidateAll(metadataTableIdentifiers(canonicalized)); + } + } + @Override public Table registerTable(TableIdentifier identifier, String metadataFileLocation) { - Table table = catalog.registerTable(identifier, metadataFileLocation); + Table table = + prepareTable( + canonicalizeIdentifier(identifier), + catalog.registerTable(identifier, metadataFileLocation)); invalidateTable(identifier); return table; } @@ -200,7 +404,10 @@ public Table registerTable(TableIdentifier identifier, String metadataFileLocati @Override public Table registerTable( TableIdentifier identifier, String metadataFileLocation, boolean overwrite) { - Table table = catalog.registerTable(identifier, metadataFileLocation, overwrite); + Table table = + prepareTable( + canonicalizeIdentifier(identifier), + catalog.registerTable(identifier, metadataFileLocation, overwrite)); invalidateTable(identifier); return table; } @@ -264,13 +471,7 @@ public TableBuilder withProperty(String key, String value) { @Override public Table create() { AtomicBoolean created = new AtomicBoolean(false); - Table table = - tableCache.get( - canonicalizeIdentifier(ident), - identifier -> { - created.set(true); - return innerBuilder.create(); - }); + Table table = createThroughCache(canonicalizeIdentifier(ident), created); if (!created.get()) { throw new AlreadyExistsException("Table already exists: %s", ident); @@ -279,6 +480,15 @@ public Table create() { return table; } + private Table createThroughCache(TableIdentifier identifier, AtomicBoolean created) { + return tableCache.get( + identifier, + ignored -> { + created.set(true); + return prepareTable(identifier, innerBuilder.create()); + }); + } + @Override public Transaction createTransaction() { // create a new transaction without altering the cache. the table doesn't exist until the diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java index 1c36c1acea2e..c74a89ffea3f 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java @@ -20,6 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; import com.github.benmanes.caffeine.cache.Cache; import java.io.IOException; @@ -27,21 +30,32 @@ import java.util.Arrays; import java.util.List; import java.util.Locale; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.AppendFiles; import org.apache.iceberg.BaseMetadataTable; +import org.apache.iceberg.BaseTable; import org.apache.iceberg.CachingCatalog; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.MetadataTableType; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; import org.apache.iceberg.TestableCachingCatalog; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -68,7 +82,7 @@ public void afterEach() { } @Test - public void testInvalidateMetadataTablesIfBaseTableIsModified() throws Exception { + public void metadataTablesShareOperationsWithModifiedBaseTable() throws Exception { Catalog catalog = CachingCatalog.wrap(hadoopCatalog()); TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl"); Table table = catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key2", "value2")); @@ -89,14 +103,14 @@ public void testInvalidateMetadataTablesIfBaseTableIsModified() throws Exception Table filesMetaTable2 = catalog.loadTable(filesMetaTableIdent); Table manifestsMetaTable2 = catalog.loadTable(manifestsMetaTableIdent); - // metadata tables are cached - assertThat(filesMetaTable2).isEqualTo(filesMetaTable); - assertThat(manifestsMetaTable2).isEqualTo(manifestsMetaTable); + // metadata tables remain cached when their origin table is preserved + assertThat(filesMetaTable2).isSameAs(filesMetaTable); + assertThat(manifestsMetaTable2).isSameAs(manifestsMetaTable); // the current snapshot of origin table is updated after committing assertThat(table.currentSnapshot()).isNotEqualTo(oldSnapshot); - // underlying table operation in metadata tables are shared with the origin table + // shared operations let the cached metadata tables observe the origin table's new snapshot assertThat(filesMetaTable2.currentSnapshot()).isEqualTo(table.currentSnapshot()); assertThat(manifestsMetaTable2.currentSnapshot()).isEqualTo(table.currentSnapshot()); } @@ -269,10 +283,338 @@ public void testCatalogExpirationTtlRefreshesAfterAccessViaCatalog() throws IOEx table.refresh(); assertThat(catalog.ageOf(tableIdent)).get().isEqualTo(HALF_OF_EXPIRATION); assertThat(catalog.remainingAgeFor(tableIdent)).get().isEqualTo(HALF_OF_EXPIRATION); + } + + @Test + public void testCommitPreservesCachedTableIdentity() throws IOException { + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(hadoopCatalog(), EXPIRATION_TTL, ticker); + TableIdentifier tableIdent = TableIdentifier.of("db", "tbl"); + Table table = catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of()); table.newAppend().appendFile(FILE_A).commit(); - assertThat(catalog.ageOf(tableIdent)).get().isEqualTo(HALF_OF_EXPIRATION); - assertThat(catalog.remainingAgeFor(tableIdent)).get().isEqualTo(HALF_OF_EXPIRATION); + + assertThat(catalog.cache().asMap()).containsEntry(tableIdent, table); + assertThat(catalog.loadTable(tableIdent).currentSnapshot()).isEqualTo(table.currentSnapshot()); + } + + @Test + public void testLoadTableObservesCommitAfterConcurrentReloadDuringWrite() throws IOException { + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(hadoopCatalog(), EXPIRATION_TTL, ticker); + Namespace namespace = Namespace.of("db", "ns1", "ns2"); + TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl"); + catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key", "value")); + + // Start from a clean cache so the writer is loaded fresh from the underlying catalog. + catalog.cache().invalidateAll(); + + // A writer loads the table and establishes an initial snapshot. + Table writer = catalog.loadTable(tableIdent); + writer.newAppend().appendFile(FILE_A).commit(); + Snapshot committedBeforeWrite = writer.currentSnapshot(); + assertThat(committedBeforeWrite).isNotNull(); + + // The writer begins a long-running write that will outlive the cache TTL. + AppendFiles pendingWrite = writer.newAppend().appendFile(FILE_B); + + // The cache entry expires while the write is still in progress. + ticker.advance(EXPIRATION_TTL.plus(Duration.ofSeconds(1))); + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + + // A concurrent loader (e.g. an OpenLineage listener thread) reloads the table before the + // write commits, re-caching the pre-commit table with a fresh TTL. + Table concurrentlyReloaded = catalog.loadTable(tableIdent); + assertThat(catalog.cache().asMap()).containsKey(tableIdent); + assertThat(concurrentlyReloaded.currentSnapshot()).isEqualTo(committedBeforeWrite); + TableIdentifier snapshotsIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl", "snapshots"); + Table staleMetadataTable = catalog.loadTable(snapshotsIdent); + assertThat(staleMetadataTable.currentSnapshot()).isEqualTo(committedBeforeWrite); + + // The writer commits the in-flight write through its own table reference. + pendingWrite.commit(); + Snapshot committedAfterWrite = writer.currentSnapshot(); + assertThat(committedAfterWrite).isNotEqualTo(committedBeforeWrite); + + // A subsequent load must observe the committed snapshot, not the stale table that was + // re-cached during the write (https://github.com/apache/iceberg/issues/17338). + assertThat(catalog.loadTable(tableIdent).currentSnapshot()) + .as("loadTable after commit must not return a snapshot re-cached during the write") + .isEqualTo(committedAfterWrite); + Table reloadedMetadataTable = catalog.loadTable(snapshotsIdent); + assertThat(reloadedMetadataTable).isNotSameAs(staleMetadataTable); + assertThat(reloadedMetadataTable.currentSnapshot()) + .as("metadata tables backed by the stale replacement must also be evicted") + .isEqualTo(committedAfterWrite); + } + + @Test + public void testCreatedTableObservesCommitAfterConcurrentReloadDuringWrite() throws IOException { + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(hadoopCatalog(), EXPIRATION_TTL, ticker); + Namespace namespace = Namespace.of("db", "ns1", "ns2"); + TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl"); + + // The writer keeps the table reference returned by createTable. + Table writer = catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key", "value")); + writer.newAppend().appendFile(FILE_A).commit(); + Snapshot committedBeforeWrite = writer.currentSnapshot(); + assertThat(committedBeforeWrite).isNotNull(); + + // The writer begins a long-running write that will outlive the cache TTL. + AppendFiles pendingWrite = writer.newAppend().appendFile(FILE_B); + + // The cache entry is not present while the write is in progress. + ticker.advance(EXPIRATION_TTL.plus(Duration.ofSeconds(1))); + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + + // A concurrent loader reloads the table before the write commits, re-caching the pre-commit + // table with a fresh TTL. + Table concurrentlyReloaded = catalog.loadTable(tableIdent); + assertThat(catalog.cache().asMap()).containsKey(tableIdent); + assertThat(concurrentlyReloaded.currentSnapshot()).isEqualTo(committedBeforeWrite); + + // The writer commits the in-flight write through the created-table reference. + pendingWrite.commit(); + Snapshot committedAfterWrite = writer.currentSnapshot(); + assertThat(committedAfterWrite).isNotEqualTo(committedBeforeWrite); + + // A subsequent load must observe the committed snapshot, not the stale re-cached table. + assertThat(catalog.loadTable(tableIdent).currentSnapshot()) + .as("loadTable after a commit through a created table must not return a stale snapshot") + .isEqualTo(committedAfterWrite); + } + + @Test + public void testCommitReconciliationWaitsForInFlightLoadBeforeRemovingIt() throws Exception { + TableIdentifier tableIdent = TableIdentifier.of("db", "tbl"); + CountDownLatch staleTableLoaded = new CountDownLatch(1); + CountDownLatch releaseStaleLoad = new CountDownLatch(1); + CountDownLatch commitApplied = new CountDownLatch(1); + AtomicBoolean instrumentWriter = new AtomicBoolean(true); + AtomicBoolean signalNextCommit = new AtomicBoolean(false); + AtomicBoolean blockNextLoad = new AtomicBoolean(false); + + HadoopCatalog underlyingCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + Table loaded = super.loadTable(ident); + if (instrumentWriter.compareAndSet(true, false)) { + TableOperations operations = spy(((HasTableOperations) loaded).operations()); + doAnswer( + invocation -> { + invocation.callRealMethod(); + if (signalNextCommit.compareAndSet(true, false)) { + commitApplied.countDown(); + } + + return null; + }) + .when(operations) + .commit(any(TableMetadata.class), any(TableMetadata.class)); + return new BaseTable(operations, loaded.name()); + } + + if (blockNextLoad.compareAndSet(true, false)) { + staleTableLoaded.countDown(); + try { + if (!releaseStaleLoad.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("Timed out waiting to release stale table load"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + + return loaded; + } + }; + underlyingCatalog.setConf(new Configuration()); + underlyingCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + underlyingCatalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of()); + + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker); + Table writer = catalog.loadTable(tableIdent); + writer.newAppend().appendFile(FILE_A).commit(); + Snapshot staleSnapshot = writer.currentSnapshot(); + AppendFiles pendingWrite = writer.newAppend().appendFile(FILE_B); + + ticker.advance(EXPIRATION_TTL.plus(Duration.ofSeconds(1))); + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + + blockNextLoad.set(true); + signalNextCommit.set(true); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future staleLoad = executor.submit(() -> catalog.loadTable(tableIdent)); + assertThat(staleTableLoaded.await(10, TimeUnit.SECONDS)).isTrue(); + + Future commit = executor.submit(pendingWrite::commit); + assertThat(commitApplied.await(10, TimeUnit.SECONDS)).isTrue(); + + // Caffeine orders loads and remapping of the same key through atomic map operations. The + // commit is durable, but reconciliation waits until this in-flight load publishes. + assertThatThrownBy(() -> commit.get(100, TimeUnit.MILLISECONDS)) + .isInstanceOf(TimeoutException.class) + .hasMessage((String) null); + + releaseStaleLoad.countDown(); + assertThat(staleLoad.get(10, TimeUnit.SECONDS).currentSnapshot()).isEqualTo(staleSnapshot); + commit.get(10, TimeUnit.SECONDS); + + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + assertThat(catalog.loadTable(tableIdent).currentSnapshot()) + .isEqualTo(writer.currentSnapshot()); + } finally { + releaseStaleLoad.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + public void testCommitDuringMetadataLoadCannotPublishStaleTable() throws Exception { + TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl"); + TableIdentifier snapshotsIdent = TableIdentifier.of("db", "ns1", "ns2", "tbl", "snapshots"); + CountDownLatch metadataReadyToPublish = new CountDownLatch(1); + CountDownLatch releaseMetadataPublish = new CountDownLatch(1); + AtomicBoolean armMetadataPublish = new AtomicBoolean(false); + AtomicBoolean blockNextName = new AtomicBoolean(false); + HadoopCatalog underlyingCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + Table loaded = super.loadTable(ident); + if (ident.equals(tableIdent) && armMetadataPublish.compareAndSet(true, false)) { + // CachingCatalog asks for the catalog name immediately before constructing and + // publishing the metadata table. + blockNextName.set(true); + } + + return loaded; + } + + @Override + public String name() { + if (blockNextName.compareAndSet(true, false)) { + metadataReadyToPublish.countDown(); + try { + if (!releaseMetadataPublish.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException( + "Timed out waiting to release metadata publication"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + + return super.name(); + } + }; + underlyingCatalog.setConf(new Configuration()); + underlyingCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + + Table writer = underlyingCatalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of()); + writer.newAppend().appendFile(FILE_A).commit(); + Snapshot staleSnapshot = writer.currentSnapshot(); + + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker); + armMetadataPublish.set(true); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future
staleLoad = executor.submit(() -> catalog.loadTable(snapshotsIdent)); + assertThat(metadataReadyToPublish.await(10, TimeUnit.SECONDS)).isTrue(); + + writer.newAppend().appendFile(FILE_B).commit(); + Snapshot committedSnapshot = writer.currentSnapshot(); + catalog.invalidateTable(tableIdent); + assertThat(catalog.cache().asMap()).doesNotContainKeys(tableIdent, snapshotsIdent); + + releaseMetadataPublish.countDown(); + assertThat(staleLoad.get(10, TimeUnit.SECONDS).currentSnapshot()).isEqualTo(staleSnapshot); + assertThat(catalog.cache().asMap()) + .as("The late metadata result must not be published after invalidation") + .doesNotContainKey(snapshotsIdent); + assertThat(catalog.loadTable(snapshotsIdent).currentSnapshot()) + .as("A metadata load overlapping invalidation must not remain cached") + .isEqualTo(committedSnapshot); + } finally { + releaseMetadataPublish.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + public void testCommitStateUnknownInvalidatesCachedTable() { + AtomicReference
nextLoad = new AtomicReference<>(); + AtomicInteger invalidationCount = new AtomicInteger(); + AtomicBoolean failInvalidation = new AtomicBoolean(true); + HadoopCatalog underlyingCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + Table replacement = nextLoad.getAndSet(null); + return replacement != null ? replacement : super.loadTable(ident); + } + + @Override + public void invalidateTable(TableIdentifier ident) { + invalidationCount.incrementAndGet(); + if (failInvalidation.get()) { + throw new IllegalStateException("Wrapped catalog invalidation failed"); + } + + super.invalidateTable(ident); + } + }; + underlyingCatalog.setConf(new Configuration()); + underlyingCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + + TableIdentifier tableIdent = TableIdentifier.of("db", "tbl"); + underlyingCatalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of()); + Table loaded = underlyingCatalog.loadTable(tableIdent); + TableOperations operations = spy(((HasTableOperations) loaded).operations()); + doAnswer( + invocation -> { + invocation.callRealMethod(); + throw new CommitStateUnknownException(new RuntimeException("Unknown commit state")); + }) + .when(operations) + .commit(any(TableMetadata.class), any(TableMetadata.class)); + nextLoad.set(new BaseTable(operations, loaded.name())); + + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker); + Table writer = catalog.loadTable(tableIdent); + assertThat(catalog.cache().asMap()).containsKey(tableIdent); + + assertThatThrownBy(() -> writer.newAppend().appendFile(FILE_A).commit()) + .isInstanceOf(CommitStateUnknownException.class) + .hasMessageContaining("Unknown commit state") + .satisfies( + failure -> + assertThat(failure.getSuppressed()) + .singleElement() + .satisfies( + suppressed -> { + assertThat(suppressed).isInstanceOf(IllegalStateException.class); + assertThat(suppressed.getMessage()) + .isEqualTo("Wrapped catalog invalidation failed"); + })); + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + assertThat(invalidationCount).hasValue(1); + failInvalidation.set(false); + assertThat(catalog.loadTable(tableIdent).currentSnapshot()).isNotNull(); } @Test @@ -285,6 +627,8 @@ public void testCacheExpirationEagerlyRemovesMetadataTables() throws IOException assertThat(catalog.cache().asMap()).containsKey(tableIdent); table.newAppend().appendFile(FILE_A).commit(); + + // The commit preserves the cached table and resets its cache age. assertThat(catalog.cache().asMap()).containsKey(tableIdent); assertThat(catalog.ageOf(tableIdent)).get().isEqualTo(Duration.ZERO); From 2591c20d8616c10e3b582fe6d1584e2ed7c6ae7b Mon Sep 17 00:00:00 2001 From: Yun-Chieh Sung Date: Tue, 28 Jul 2026 15:35:40 -0700 Subject: [PATCH 2/2] Core: Support commit tracking for specialized tables Generated-by: Codex Co-authored-by: Humzah Kiani Co-authored-by: Codex (GPT-5) --- .../org/apache/iceberg/CachingCatalog.java | 166 ++++++++++-- .../SupportsOperationsReplacement.java | 44 +++ .../org/apache/iceberg/rest/RESTTable.java | 20 +- .../iceberg/hadoop/TestCachingCatalog.java | 255 ++++++++++++++++++ .../iceberg/rest/TestRESTScanPlanning.java | 56 ++++ 5 files changed, 518 insertions(+), 23 deletions(-) create mode 100644 core/src/main/java/org/apache/iceberg/SupportsOperationsReplacement.java diff --git a/core/src/main/java/org/apache/iceberg/CachingCatalog.java b/core/src/main/java/org/apache/iceberg/CachingCatalog.java index d71a1c95c90b..670e8094128b 100644 --- a/core/src/main/java/org/apache/iceberg/CachingCatalog.java +++ b/core/src/main/java/org/apache/iceberg/CachingCatalog.java @@ -27,6 +27,8 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; @@ -73,6 +75,8 @@ public static Catalog wrap( @SuppressWarnings("checkstyle:VisibilityModifier") protected final Cache tableCache; + private final Set invalidReplacementTableTypeNames = ConcurrentHashMap.newKeySet(); + private CachingCatalog(Catalog catalog, boolean caseSensitive, long expirationIntervalMillis) { this(catalog, caseSensitive, expirationIntervalMillis, Ticker.systemTicker()); } @@ -148,13 +152,24 @@ public Table loadTable(TableIdentifier ident) { return cached; } - Table table = tableCache.get(canonicalized, this::loadTableForCache); + Table table; + try { + table = tableCache.get(canonicalized, this::loadTableForCache); + } catch (UncacheableTableException e) { + return e.table(); + } if (table instanceof BaseMetadataTable) { // Cache underlying table TableIdentifier originTableIdentifier = TableIdentifier.of(canonicalized.namespace().levels()); - Table originTable = tableCache.get(originTableIdentifier, this::loadTableForCache); + Table originTable; + try { + originTable = tableCache.get(originTableIdentifier, this::loadTableForCache); + } catch (UncacheableTableException e) { + tableCache.invalidate(canonicalized); + return table; + } // Share TableOperations instance of origin table for all metadata tables, so that metadata // table instances are refreshed as well when origin table instance is refreshed. @@ -179,7 +194,7 @@ private Table loadTableForCache(TableIdentifier identifier) { return loaded; } - return prepareTable(identifier, loaded); + return requireCacheable(prepareTable(identifier, loaded)); } private Table publishMetadataTable( @@ -222,22 +237,84 @@ private Table publishMetadataTable( * serving the pre-commit snapshot until the entry expires. See * https://github.com/apache/iceberg/issues/17338. * - *

A plain {@link BaseTable} is re-created directly with the tracking operations. Other table - * implementations are returned unchanged. Metadata tables share the safely prepared origin - * table's operations and are invalidated together with it. + *

The concrete table type is preserved. A plain {@link BaseTable} is re-created directly. A + * subclass opts in by implementing {@link SupportsOperationsReplacement} (for example {@code + * RESTTable}, which keeps its server-side scan planning). Other table implementations retain the + * existing caching behavior without commit tracking. Metadata tables share the safely prepared + * origin table's operations and are invalidated together with it. */ - private Table prepareTable(TableIdentifier identifier, Table table) { - if (table.getClass() == BaseTable.class) { + private PreparedTable prepareTable(TableIdentifier identifier, Table table) { + TableOperations operations; + Table wrapped; + if (table instanceof SupportsOperationsReplacement replaceable) { + operations = replaceable.operations(); + CacheInvalidatingTableOperations replacementOperations = + new CacheInvalidatingTableOperations(operations, identifier); + try { + wrapped = replaceable.withOperations(replacementOperations); + } catch (RuntimeException e) { + warnInvalidReplacement(identifier, table, "withOperations threw an exception", e); + return PreparedTable.uncacheable(table); + } + + if (wrapped != null + && wrapped != table + && wrapped.getClass() == table.getClass() + && ((SupportsOperationsReplacement) wrapped).operations() == replacementOperations + && replaceable.operations() == operations) { + replacementOperations.track(wrapped); + return PreparedTable.cacheable(wrapped); + } + + warnInvalidReplacement( + identifier, + table, + "withOperations must return an independent copy of the same concrete type without " + + "modifying the original table", + null); + return PreparedTable.uncacheable(table); + } else if (table.getClass() == BaseTable.class) { BaseTable baseTable = (BaseTable) table; - TableOperations operations = baseTable.operations(); + operations = baseTable.operations(); CacheInvalidatingTableOperations replacementOperations = new CacheInvalidatingTableOperations(operations, identifier); - Table wrapped = new BaseTable(replacementOperations, baseTable.name(), baseTable.reporter()); + wrapped = new BaseTable(replacementOperations, baseTable.name(), baseTable.reporter()); replacementOperations.track(wrapped); - return wrapped; + return PreparedTable.cacheable(wrapped); } - return table; + return PreparedTable.cacheable(table); + } + + private Table requireCacheable(PreparedTable prepared) { + if (!prepared.cacheable()) { + throw new UncacheableTableException(prepared.table()); + } + + return prepared.table(); + } + + private void warnInvalidReplacement( + TableIdentifier identifier, Table table, String reason, RuntimeException failure) { + String tableTypeName = table.getClass().getName(); + if (invalidReplacementTableTypeNames.add(tableTypeName)) { + if (failure != null) { + LOG.warn( + "Table {} of type {} cannot use CachingCatalog commit tracking because {}. " + + "This instance will not be cached", + identifier, + tableTypeName, + reason, + failure); + } else { + LOG.warn( + "Table {} of type {} cannot use CachingCatalog commit tracking because {}. " + + "This instance will not be cached", + identifier, + tableTypeName, + reason); + } + } } /** @@ -395,8 +472,9 @@ private void reconcileLocalCacheAfterCommit( public Table registerTable(TableIdentifier identifier, String metadataFileLocation) { Table table = prepareTable( - canonicalizeIdentifier(identifier), - catalog.registerTable(identifier, metadataFileLocation)); + canonicalizeIdentifier(identifier), + catalog.registerTable(identifier, metadataFileLocation)) + .table(); invalidateTable(identifier); return table; } @@ -406,8 +484,9 @@ public Table registerTable( TableIdentifier identifier, String metadataFileLocation, boolean overwrite) { Table table = prepareTable( - canonicalizeIdentifier(identifier), - catalog.registerTable(identifier, metadataFileLocation, overwrite)); + canonicalizeIdentifier(identifier), + catalog.registerTable(identifier, metadataFileLocation, overwrite)) + .table(); invalidateTable(identifier); return table; } @@ -481,12 +560,16 @@ public Table create() { } private Table createThroughCache(TableIdentifier identifier, AtomicBoolean created) { - return tableCache.get( - identifier, - ignored -> { - created.set(true); - return prepareTable(identifier, innerBuilder.create()); - }); + try { + return tableCache.get( + identifier, + ignored -> { + created.set(true); + return requireCacheable(prepareTable(identifier, innerBuilder.create())); + }); + } catch (UncacheableTableException e) { + return e.table(); + } } @Override @@ -520,4 +603,43 @@ public Transaction createOrReplaceTransaction() { innerBuilder.createOrReplaceTransaction(), () -> invalidateTable(ident)); } } + + private static class PreparedTable { + private final Table table; + private final boolean cacheable; + + private static PreparedTable cacheable(Table table) { + return new PreparedTable(table, true); + } + + private static PreparedTable uncacheable(Table table) { + return new PreparedTable(table, false); + } + + private PreparedTable(Table table, boolean cacheable) { + this.table = Preconditions.checkNotNull(table, "Prepared table cannot be null"); + this.cacheable = cacheable; + } + + private Table table() { + return table; + } + + private boolean cacheable() { + return cacheable; + } + } + + private static class UncacheableTableException extends RuntimeException { + private final Table table; + + private UncacheableTableException(Table table) { + super(null, null, false, false); + this.table = table; + } + + private Table table() { + return table; + } + } } diff --git a/core/src/main/java/org/apache/iceberg/SupportsOperationsReplacement.java b/core/src/main/java/org/apache/iceberg/SupportsOperationsReplacement.java new file mode 100644 index 000000000000..7bbf72d3631d --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/SupportsOperationsReplacement.java @@ -0,0 +1,44 @@ +/* + * 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.iceberg; + +/** + * Internal SPI for a table that can copy itself with replacement operations while preserving its + * concrete type. + * + *

{@link CachingCatalog} uses this capability to observe commits without mutating a table + * instance owned by the wrapped catalog. Tables that do not support operations replacement retain + * the existing caching behavior without commit tracking. Implementations that violate this contract + * are returned normally but are not cached. This interface is internal and may change without + * notice. + */ +public interface SupportsOperationsReplacement extends HasTableOperations { + /** + * Returns a copy of this table backed by the given operations. + * + *

This table must not be modified. The returned table must be a different instance with the + * same concrete type, and its {@link #operations()} method must return {@code newOperations}. + * Implementations must only construct the copy; this method is invoked while a cache entry is + * being computed and must not call back into the catalog or perform table operations. + * + * @param newOperations operations for the returned table + * @return a copy backed by {@code newOperations} + */ + Table withOperations(TableOperations newOperations); +} diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTTable.java b/core/src/main/java/org/apache/iceberg/rest/RESTTable.java index 21d84d847781..b61ecd0e523c 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTTable.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTTable.java @@ -26,12 +26,15 @@ import org.apache.iceberg.BatchScanAdapter; import org.apache.iceberg.ImmutableTableScanContext; import org.apache.iceberg.SupportsDistributedScanPlanning; +import org.apache.iceberg.SupportsOperationsReplacement; +import org.apache.iceberg.Table; import org.apache.iceberg.TableOperations; import org.apache.iceberg.TableScan; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.metrics.MetricsReporter; -class RESTTable extends BaseTable implements SupportsDistributedScanPlanning { +class RESTTable extends BaseTable + implements SupportsDistributedScanPlanning, SupportsOperationsReplacement { private final RESTClient client; private final Supplier> headers; private final MetricsReporter reporter; @@ -88,4 +91,19 @@ public BatchScan newBatchScan() { public boolean allowDistributedPlanning() { return false; } + + @Override + public Table withOperations(TableOperations newOperations) { + return new RESTTable( + newOperations, + name(), + reporter, + client, + headers, + tableIdentifier, + resourcePaths, + supportedEndpoints, + catalogProperties, + hadoopConf); + } } diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java index c74a89ffea3f..22ef2256aeae 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestCachingCatalog.java @@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import com.github.benmanes.caffeine.cache.Cache; @@ -39,6 +40,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.AppendFiles; import org.apache.iceberg.BaseMetadataTable; @@ -48,6 +50,7 @@ import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.MetadataTableType; import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SupportsOperationsReplacement; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableOperations; @@ -617,6 +620,155 @@ public void invalidateTable(TableIdentifier ident) { assertThat(catalog.loadTable(tableIdent).currentSnapshot()).isNotNull(); } + @Test + public void testCachingCatalogPreservesTableSubtypeAndIdentityOnCommit() throws IOException { + // A delegate catalog that returns a BaseTable subclass (like RESTTable) from loadTable. + HadoopCatalog subtypeReturningCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + Table loaded = super.loadTable(ident); + return new CustomBaseTable(((HasTableOperations) loaded).operations(), loaded.name()); + } + }; + subtypeReturningCatalog.setConf(new Configuration()); + subtypeReturningCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(subtypeReturningCatalog, EXPIRATION_TTL, ticker); + Namespace namespace = Namespace.of("db", "ns1", "ns2"); + TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl"); + catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key", "value")); + + // Load fresh from the delegate so the cache wraps the delegate's subtype. + catalog.cache().invalidateAll(); + Table loaded = catalog.loadTable(tableIdent); + + // The cache preserves the delegate's concrete table type via SupportsOperationsReplacement, + // rather than downgrading it to a plain BaseTable (which would discard specialized behavior). + assertThat(loaded).isInstanceOf(CustomBaseTable.class); + + // A commit through the subtype preserves the committing table's cache identity. + loaded.newAppend().appendFile(FILE_A).commit(); + assertThat(catalog.cache().asMap()).containsEntry(tableIdent, loaded); + } + + @Test + public void testCachingCatalogLeavesUnknownSubclassUnchanged() throws IOException { + AtomicInteger loadCount = new AtomicInteger(); + // A delegate that returns a BaseTable subclass which does NOT opt in to operations replacement. + HadoopCatalog subtypeReturningCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + loadCount.incrementAndGet(); + Table loaded = super.loadTable(ident); + return new NonReplaceableBaseTable( + ((HasTableOperations) loaded).operations(), loaded.name()); + } + }; + subtypeReturningCatalog.setConf(new Configuration()); + subtypeReturningCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(subtypeReturningCatalog, EXPIRATION_TTL, ticker); + Namespace namespace = Namespace.of("db", "ns1", "ns2"); + TableIdentifier tableIdent = TableIdentifier.of(namespace, "tbl"); + catalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of("key", "value")); + + // Load fresh from the delegate so the cache processes the subtype. + catalog.cache().invalidateAll(); + Table loaded = catalog.loadTable(tableIdent); + + // A subclass that does not opt in must be returned unchanged, not downgraded to a plain + // BaseTable (which would discard its overridden behavior). + assertThat(loaded).isInstanceOf(NonReplaceableBaseTable.class); + assertThat(catalog.loadTable(tableIdent)).isSameAs(loaded); + assertThat(loadCount).hasValue(1); + assertThat(catalog.cache().asMap()).containsEntry(tableIdent, loaded); + } + + @Test + public void testCachingCatalogPreservesExistingBehaviorForUnsupportedTableTypes() { + AtomicInteger loadCount = new AtomicInteger(); + Table unsupported = mock(Table.class); + HadoopCatalog underlyingCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + loadCount.incrementAndGet(); + return unsupported; + } + }; + TableIdentifier tableIdent = TableIdentifier.of("db", "tbl"); + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker); + + assertThat(catalog.loadTable(tableIdent)).isSameAs(unsupported); + assertThat(catalog.loadTable(tableIdent)).isSameAs(unsupported); + assertThat(loadCount).hasValue(1); + assertThat(catalog.cache().asMap()).containsEntry(tableIdent, unsupported); + } + + @Test + public void brokenOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(BrokenReplaceableBaseTable::new); + } + + @Test + public void nullOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(NullReplaceableBaseTable::new); + } + + @Test + public void differentTypeOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(DifferentTypeReplaceableBaseTable::new); + } + + @Test + public void wrongOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(WrongOperationsReplaceableBaseTable::new); + } + + @Test + public void throwingOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(ThrowingReplaceableBaseTable::new); + } + + @Test + public void mutatingOperationsReplacementIsNotCached() throws IOException { + assertInvalidOperationsReplacementIsNotCached(MutatingReplaceableBaseTable::new); + } + + private void assertInvalidOperationsReplacementIsNotCached( + BiFunction tableFactory) throws IOException { + AtomicInteger loadCount = new AtomicInteger(); + HadoopCatalog underlyingCatalog = + new HadoopCatalog() { + @Override + public Table loadTable(TableIdentifier ident) { + loadCount.incrementAndGet(); + Table loaded = super.loadTable(ident); + return tableFactory.apply(((HasTableOperations) loaded).operations(), loaded.name()); + } + }; + underlyingCatalog.setConf(new Configuration()); + underlyingCatalog.initialize( + "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, tempDir.getAbsolutePath())); + + TableIdentifier tableIdent = TableIdentifier.of("db", "tbl"); + underlyingCatalog.createTable(tableIdent, SCHEMA, SPEC, ImmutableMap.of()); + TestableCachingCatalog catalog = + TestableCachingCatalog.wrap(underlyingCatalog, EXPIRATION_TTL, ticker); + + Table loaded = catalog.loadTable(tableIdent); + assertThat(catalog.loadTable(tableIdent)).isInstanceOf(loaded.getClass()); + assertThat(loadCount).hasValue(2); + assertThat(catalog.cache().asMap()).doesNotContainKey(tableIdent); + } + @Test public void testCacheExpirationEagerlyRemovesMetadataTables() throws IOException { TestableCachingCatalog catalog = @@ -785,4 +937,107 @@ public static TableIdentifier[] metadataTables(TableIdentifier tableIdent) { .map(type -> TableIdentifier.parse(tableIdent + "." + type.name().toLowerCase(Locale.ROOT))) .toArray(TableIdentifier[]::new); } + + /** + * A {@link BaseTable} subclass that opts in to operations replacement, preserving its type when + * re-created with new operations. + */ + private static class CustomBaseTable extends BaseTable implements SupportsOperationsReplacement { + private CustomBaseTable(TableOperations ops, String name) { + super(ops, name); + } + + @Override + public Table withOperations(TableOperations newOps) { + return new CustomBaseTable(newOps, name()); + } + } + + /** A {@link BaseTable} subclass that does not opt in to operations replacement. */ + private static class NonReplaceableBaseTable extends BaseTable { + private NonReplaceableBaseTable(TableOperations ops, String name) { + super(ops, name); + } + } + + private static class BrokenReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private BrokenReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + } + + @Override + public Table withOperations(TableOperations newOperations) { + return this; + } + } + + private static class NullReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private NullReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + } + + @Override + public Table withOperations(TableOperations newOperations) { + return null; + } + } + + private static class DifferentTypeReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private DifferentTypeReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + } + + @Override + public Table withOperations(TableOperations newOperations) { + return new BaseTable(newOperations, name()); + } + } + + private static class WrongOperationsReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private WrongOperationsReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + } + + @Override + public Table withOperations(TableOperations newOperations) { + return new WrongOperationsReplaceableBaseTable(operations(), name()); + } + } + + private static class ThrowingReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private ThrowingReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + } + + @Override + public Table withOperations(TableOperations newOperations) { + throw new IllegalStateException("Cannot replace operations"); + } + } + + private static class MutatingReplaceableBaseTable extends BaseTable + implements SupportsOperationsReplacement { + private TableOperations currentOperations; + + private MutatingReplaceableBaseTable(TableOperations operations, String name) { + super(operations, name); + this.currentOperations = operations; + } + + @Override + public TableOperations operations() { + return currentOperations; + } + + @Override + public Table withOperations(TableOperations newOperations) { + this.currentOperations = newOperations; + return new MutatingReplaceableBaseTable(newOperations, name()); + } + } } diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java index 117bf7fac24b..4c34157b7873 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java @@ -36,10 +36,12 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectReader; import java.io.IOException; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.function.Consumer; import java.util.function.Function; +import org.apache.iceberg.AppendFiles; import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.DataFile; @@ -52,8 +54,10 @@ import org.apache.iceberg.Scan; import org.apache.iceberg.ScanTask; import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableScan; +import org.apache.iceberg.TestableCachingCatalog; import org.apache.iceberg.catalog.SessionCatalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.expressions.Expressions; @@ -71,6 +75,7 @@ import org.apache.iceberg.rest.responses.LoadTableResponse; import org.apache.iceberg.rest.responses.PlanTableScanResponse; import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.FakeTicker; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -1650,6 +1655,57 @@ public void planningModeWithDifferentCasesOnServer(String serverMode) { verifyTableTypeForPlanningMode(serverMode, table); } + @Test + public void testCachingCatalogObservesCommitAfterConcurrentReloadWithServerSidePlanning() + throws IOException { + configurePlanningBehavior(TestPlanningBehavior.Builder::synchronous); + Duration expiration = Duration.ofMinutes(5); + FakeTicker ticker = new FakeTicker(); + TestableCachingCatalog cachingCatalog = + TestableCachingCatalog.wrap(restCatalog, expiration, ticker); + + String tableName = "caching_stale_read"; + TableIdentifier ident = TableIdentifier.of(NS, tableName); + createTableWithScanPlanning(restCatalog, tableName); + + // The writer loads the table through the caching catalog. With server-side scan planning the + // catalog returns a RESTTable. Establish an initial snapshot. + Table writer = cachingCatalog.loadTable(ident); + assertThat(writer).isInstanceOf(RESTTable.class); + writer.newAppend().appendFile(FILE_A).commit(); + Snapshot committedBeforeWrite = writer.currentSnapshot(); + assertThat(committedBeforeWrite).isNotNull(); + + // The writer begins a long-running write that will outlive the cache TTL. + AppendFiles pendingWrite = writer.newAppend().appendFile(FILE_B); + + // The cache entry is not present while the write is in progress. + ticker.advance(expiration.plus(Duration.ofSeconds(1))); + assertThat(cachingCatalog.cache().asMap()).doesNotContainKey(ident); + + // A concurrent loader reloads the table before the write commits, re-caching the pre-commit + // RESTTable with a fresh TTL. + Table concurrentlyReloaded = cachingCatalog.loadTable(ident); + assertThat(concurrentlyReloaded).isInstanceOf(RESTTable.class); + assertThat(concurrentlyReloaded.currentSnapshot()).isEqualTo(committedBeforeWrite); + + // The writer commits the in-flight write. + pendingWrite.commit(); + Snapshot committedAfterWrite = writer.currentSnapshot(); + assertThat(committedAfterWrite).isNotEqualTo(committedBeforeWrite); + + // A subsequent load observes the committed snapshot, and the cached table is still a RESTTable + // whose newScan() plans server-side, so wrapping the operations preserved neither a stale + // snapshot nor downgraded the concrete type or scan planning. + Table reloaded = cachingCatalog.loadTable(ident); + assertThat(reloaded.currentSnapshot()).isEqualTo(committedAfterWrite); + assertThat(restTableScanFor(reloaded)).isInstanceOf(RESTTableScan.class); + setParserContext(reloaded); + try (CloseableIterable tasks = reloaded.newScan().planFiles()) { + assertThat(Lists.newArrayList(tasks)).hasSize(2); + } + } + private void verifyTableTypeForPlanningMode(String planingMode, Table table) { if (planingMode.equalsIgnoreCase("client")) { assertThat(table).isNotInstanceOf(RESTTable.class).isInstanceOf(BaseTable.class);