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