Make credential cache TTLs configurable via IcebergConfig

- Add iceberg.credential-cache-ttl, iceberg.coordinator-credential-prefetch,
  and iceberg.worker-credential-cache-ttl config properties with defaults
  (25min, 2min, 5min respectively)
- Remove static TTL constants from IcebergUtil
- Propagate configurable TTLs through IcebergMetadataFactory, IcebergMetadata,
  IcebergPageSourceProviderFactory, IcebergPageSinkProvider
- Use io.airlift.units.Duration throughout plumbing, convert to millis at leaf sites
- Broaden vended credential detection in IcebergTableCredentials.forFileIO to
  cover S3, GCS, and Azure ADLS (not just S3)
- Handle TableNotFoundException in getOrLoadTableCredentials for CTAS queries
  where the table does not yet exist
- Fix checkstyle: import ordering and remove unused TableChangesFunctionHandle import
- Update all test files to use configurable TTL values

Author: kaveti

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

@@ -45,6 +45,7 @@
 import static java.util.Locale.ENGLISH;
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.HOURS;
+import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 @DefunctConfig({
@@ -104,6 +105,8 @@ public class IcebergConfig
     private int metadataParallelism = 8;
     private boolean bucketExecutionEnabled = true;
     private boolean fileBasedConflictDetectionEnabled = true;
+    private Duration credentialCacheTtl = new Duration(25, MINUTES);
+    private Duration workerCredentialCacheTtl = new Duration(5, MINUTES);
 
     public CatalogType getCatalogType()
     {
@@ -695,4 +698,32 @@ public IcebergConfig setFileBasedConflictDetectionEnabled(boolean fileBasedConfl
         this.fileBasedConflictDetectionEnabled = fileBasedConflictDetectionEnabled;
         return this;
     }
+
+    @NotNull
+    public Duration getCredentialCacheTtl()
+    {
+        return credentialCacheTtl;
+    }
+
+    @Config("iceberg.credential-cache-ttl")
+    @ConfigDescription("How long vended credentials are assumed to be valid. Must be shorter than the minimum STS token lifetime.")
+    public IcebergConfig setCredentialCacheTtl(Duration credentialCacheTtl)
+    {
+        this.credentialCacheTtl = credentialCacheTtl;
+        return this;
+    }
+
+    @NotNull
+    public Duration getWorkerCredentialCacheTtl()
+    {
+        return workerCredentialCacheTtl;
+    }
+
+    @Config("iceberg.worker-credential-cache-ttl")
+    @ConfigDescription("How long a worker caches refreshed credentials to coalesce concurrent refresh requests")
+    public IcebergConfig setWorkerCredentialCacheTtl(Duration workerCredentialCacheTtl)
+    {
+        this.workerCredentialCacheTtl = workerCredentialCacheTtl;
+        return this;
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -18,23 +18,20 @@
 import com.google.common.base.Splitter.MapSplitter;
 import com.google.common.base.Suppliers;
 import com.google.common.base.VerifyException;
-import com.google.common.cache.Cache;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Streams;
-import com.google.common.util.concurrent.UncheckedExecutionException;
 import io.airlift.concurrent.MoreFutures;
 import io.airlift.json.JsonCodec;
 import io.airlift.log.Logger;
 import io.airlift.slice.Slice;
 import io.airlift.slice.Slices;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
-import io.trino.cache.EvictableCacheBuilder;
 import io.trino.filesystem.FileEntry;
 import io.trino.filesystem.FileIterator;
 import io.trino.filesystem.Location;
@@ -55,7 +52,6 @@
 import io.trino.plugin.iceberg.delete.DeletionVectorWriter;
 import io.trino.plugin.iceberg.delete.DeletionVectorWriter.DeletionVectorInfo;
 import io.trino.plugin.iceberg.functions.IcebergFunctionProvider;
-import io.trino.plugin.iceberg.functions.tablechanges.TableChangesFunctionHandle;
 import io.trino.plugin.iceberg.procedure.IcebergAddFilesFromTableHandle;
 import io.trino.plugin.iceberg.procedure.IcebergAddFilesHandle;
 import io.trino.plugin.iceberg.procedure.IcebergDropExtendedStatsHandle;
@@ -268,7 +264,6 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
-import static com.google.common.base.Throwables.throwIfUnchecked;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.base.Verify.verifyNotNull;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -279,7 +274,6 @@
 import static com.google.common.collect.Maps.transformValues;
 import static com.google.common.collect.Sets.difference;
 import static io.airlift.units.Duration.ZERO;
-import static io.trino.cache.CacheUtils.uncheckedCacheGet;
 import static io.trino.filesystem.Locations.isS3Tables;
 import static io.trino.plugin.base.filter.UtcConstraintExtractor.extractTupleDomain;
 import static io.trino.plugin.base.projection.ApplyProjectionUtil.extractSupportedProjectedColumns;
@@ -359,8 +353,6 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.getPartitioning;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getTableLocation;
 import static io.trino.plugin.iceberg.IcebergTableProperties.validateCompression;
-import static io.trino.plugin.iceberg.IcebergUtil.COORDINATOR_CREDENTIAL_PREFETCH;
-import static io.trino.plugin.iceberg.IcebergUtil.CREDENTIAL_CACHE_TTL;
 import static io.trino.plugin.iceberg.IcebergUtil.buildPath;
 import static io.trino.plugin.iceberg.IcebergUtil.canEnforceColumnConstraintInSpecs;
 import static io.trino.plugin.iceberg.IcebergUtil.checkFormatForProperty;
@@ -538,7 +530,8 @@ public class IcebergMetadata
     private final int materializedViewRefreshMaxSnapshotsToExpire;
     private final Duration materializedViewRefreshSnapshotRetentionPeriod;
     private final Map<IcebergTableHandle, AtomicReference<TableStatistics>> tableStatisticsCache = new ConcurrentHashMap<>();
-    private final Cache<SchemaTableName, IcebergTableCredentials> tableCredentialsCache;
+    private final ConcurrentHashMap<SchemaTableName, IcebergTableCredentials> tableCredentialsCache = new ConcurrentHashMap<>();
+    private final Duration credentialCacheTtl;
     private final DeletionVectorWriter deletionVectorWriter;
 
     private Transaction transaction;
@@ -560,7 +553,8 @@ public IcebergMetadata(
             ExecutorService icebergPlanningExecutor,
             ExecutorService icebergFileDeleteExecutor,
             int materializedViewRefreshMaxSnapshotsToExpire,
-            Duration materializedViewRefreshSnapshotRetentionPeriod)
+            Duration materializedViewRefreshSnapshotRetentionPeriod,
+            Duration credentialCacheTtl)
     {
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.commitTaskCodec = requireNonNull(commitTaskCodec, "commitTaskCodec is null");
@@ -578,15 +572,7 @@ public IcebergMetadata(
         this.deletionVectorWriter = requireNonNull(deletionVectorWriter, "deletionVectorWriter is null");
         this.materializedViewRefreshMaxSnapshotsToExpire = materializedViewRefreshMaxSnapshotsToExpire;
         this.materializedViewRefreshSnapshotRetentionPeriod = materializedViewRefreshSnapshotRetentionPeriod;
-        // Credentials are cached with a TTL so that the coordinator will re-fetch fresh credentials
-        // when getTableCredentials() is called for new tasks in long-running queries.
-        // The cache evicts COORDINATOR_CREDENTIAL_PREFETCH (2 min) before the token expires,
-        // ensuring the next getTableCredentials() call re-fetches while the old token is still valid.
-        // EvictableCache guarantees that invalidate() is immediately visible to subsequent get() calls.
-        this.tableCredentialsCache = EvictableCacheBuilder.newBuilder()
-                .expireAfterWrite(CREDENTIAL_CACHE_TTL.minus(COORDINATOR_CREDENTIAL_PREFETCH))
-                .shareNothingWhenDisabled()
-                .build();
+        this.credentialCacheTtl = requireNonNull(credentialCacheTtl, "credentialCacheTtl is null");
     }
 
     @Override
@@ -604,20 +590,31 @@ public Optional<ConnectorTableCredentials> getTableCredentials(ConnectorSession
     private Optional<ConnectorTableCredentials> getOrLoadTableCredentials(ConnectorSession session, SchemaTableName schemaTableName)
     {
         try {
-            return Optional.of(uncheckedCacheGet(
-                tableCredentialsCache,
-                schemaTableName,
-                () -> {
-                    BaseTable baseTable = catalog.loadTable(session, schemaTableName);
-                    return IcebergTableCredentials.forFileIO(baseTable.io());
-                }));
+            // Check if existing cached entry is still fresh (per-entry expiry).
+            // Each IcebergTableCredentials carries its own expiresAt, so entries with
+            // different token lifetimes are handled correctly without a cache-wide static TTL.
+            IcebergTableCredentials cached = tableCredentialsCache.get(schemaTableName);
+            if (cached != null && !cached.shouldRefresh()) {
+                return Optional.of(cached);
+            }
+            // Entry is missing or approaching expiry — reload from catalog
+            IcebergTableCredentials fresh = loadTableCredentials(session, schemaTableName);
+            tableCredentialsCache.put(schemaTableName, fresh);
+            return Optional.of(fresh);
         }
-        catch (UncheckedExecutionException e) {
-            throwIfUnchecked(e.getCause());
-            throw e;
+        catch (TableNotFoundException e) {
+            // Table may not exist yet (e.g. during CREATE TABLE AS SELECT);
+            // no credentials to vend in that case.
+            return Optional.empty();
         }
     }
 
+    private IcebergTableCredentials loadTableCredentials(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        BaseTable baseTable = catalog.loadTable(session, schemaTableName);
+        return IcebergTableCredentials.forFileIO(baseTable.io(), credentialCacheTtl.toMillis());
+    }
+
     private static SchemaTableName getSchemaTableName(ConnectorTableHandle tableHandle)
     {
         if (tableHandle instanceof IcebergTableHandle handle) {
@@ -1621,10 +1618,9 @@ private List<String> getChildNamespaces(ConnectorSession session, String parentN
 
     private IcebergWritableTableHandle newWritableTableHandle(SchemaTableName name, Table table)
     {
-        // Invalidate so the next getOrLoadTableCredentials() call re-loads fresh credentials
-        // from the catalog. EvictableCache does not support put(); invalidate + lazy reload is
-        // equivalent and avoids races with concurrent get() calls.
-        tableCredentialsCache.invalidate(name);
+        // Remove so the next getOrLoadTableCredentials() call re-loads fresh credentials
+        // from the catalog for the newly created writable table handle.
+        tableCredentialsCache.remove(name);
         SortFieldInfo sortInfo = getSupportedSortFields(table.schema(), table.sortOrder());
         return new IcebergWritableTableHandle(
                 name,

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadataFactory.java

@@ -52,6 +52,7 @@ public class IcebergMetadataFactory
     private final DeletionVectorWriter deletionVectorWriter;
     private final int materializedViewRefreshMaxSnapshotsToExpire;
     private final Duration materializedViewRefreshSnapshotRetentionPeriod;
+    private final Duration credentialCacheTtl;
 
     @Inject
     public IcebergMetadataFactory(
@@ -96,6 +97,7 @@ public IcebergMetadataFactory(
         this.icebergFileDeleteExecutor = requireNonNull(icebergFileDeleteExecutor, "icebergFileDeleteExecutor is null");
         this.materializedViewRefreshMaxSnapshotsToExpire = config.getMaterializedViewRefreshMaxSnapshotsToExpire();
         this.materializedViewRefreshSnapshotRetentionPeriod = config.getMaterializedViewRefreshSnapshotRetentionPeriod();
+        this.credentialCacheTtl = config.getCredentialCacheTtl();
     }
 
     public IcebergMetadata create(ConnectorIdentity identity)
@@ -116,6 +118,7 @@ public IcebergMetadata create(ConnectorIdentity identity)
                 icebergPlanningExecutor,
                 icebergFileDeleteExecutor,
                 materializedViewRefreshMaxSnapshotsToExpire,
-                materializedViewRefreshSnapshotRetentionPeriod);
+                materializedViewRefreshSnapshotRetentionPeriod,
+                credentialCacheTtl);
     }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSinkProvider.java

@@ -54,11 +54,11 @@
 
 import static com.google.common.collect.Maps.transformValues;
 import static io.trino.plugin.iceberg.IcebergSessionProperties.maxPartitionsPerWriter;
-import static io.trino.plugin.iceberg.IcebergUtil.WORKER_CREDENTIAL_CACHE_TTL;
 import static io.trino.plugin.iceberg.IcebergUtil.getFileIoProperties;
 import static io.trino.plugin.iceberg.IcebergUtil.getLocationProvider;
 import static io.trino.plugin.iceberg.IcebergUtil.maybeRefreshVendedCredentials;
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 public class IcebergPageSinkProvider
         implements ConnectorPageSinkProvider
@@ -73,12 +73,8 @@ public class IcebergPageSinkProvider
     private final TypeManager typeManager;
     private final PageSorter pageSorter;
     private final TrinoCatalogFactory catalogFactory;
-    private final Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache =
-            EvictableCacheBuilder.newBuilder()
-                    .maximumSize(1_000)
-                    .expireAfterWrite(WORKER_CREDENTIAL_CACHE_TTL)
-                    .shareNothingWhenDisabled()
-                    .build();
+    private final long credentialCacheTtlMillis;
+    private final Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache;
 
     @Inject
     public IcebergPageSinkProvider(
@@ -102,6 +98,12 @@ public IcebergPageSinkProvider(
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.pageSorter = requireNonNull(pageSorter, "pageSorter is null");
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
+        this.credentialCacheTtlMillis = icebergConfig.getCredentialCacheTtl().toMillis();
+        this.refreshedCredentialCache = EvictableCacheBuilder.newBuilder()
+                .maximumSize(1_000)
+                .expireAfterWrite(icebergConfig.getWorkerCredentialCacheTtl().toMillis(), MILLISECONDS)
+                .shareNothingWhenDisabled()
+                .build();
     }
 
     @Override
@@ -246,7 +248,7 @@ private Optional<ConnectorTableCredentials> maybeRefreshCredentials(
                 () -> {
                     TrinoCatalog catalog = catalogFactory.create(session.getIdentity());
                     BaseTable freshTable = catalog.loadTable(session, schemaTableName);
-                    return IcebergTableCredentials.forFileIO(freshTable.io());
+                    return IcebergTableCredentials.forFileIO(freshTable.io(), credentialCacheTtlMillis);
                 });
     }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java

@@ -238,6 +238,7 @@ public class IcebergPageSourceProvider
     // Owned by the factory so it is shared across all providers created on this worker,
     // preventing many concurrent tasks from all calling the REST catalog simultaneously.
     private final Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache;
+    private final long credentialCacheTtlMillis;
     private final DeleteManager unpartitionedTableDeleteManager;
     private final Map<Integer, Function<PartitionData, PartitionKey>> partitionKeyFactories = new ConcurrentHashMap<>();
     private final Map<PartitionKey, DeleteManager> partitionedDeleteManagers = new ConcurrentHashMap<>();
@@ -250,7 +251,8 @@ public IcebergPageSourceProvider(
             ParquetReaderOptions parquetReaderOptions,
             TypeManager typeManager,
             TrinoCatalogFactory catalogFactory,
-            Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache)
+            Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache,
+            long credentialCacheTtlMillis)
     {
         this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
         this.fileIoFactory = requireNonNull(fileIoFactory, "fileIoFactory is null");
@@ -260,6 +262,7 @@ public IcebergPageSourceProvider(
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
         this.refreshedCredentialCache = requireNonNull(refreshedCredentialCache, "refreshedCredentialCache is null");
+        this.credentialCacheTtlMillis = credentialCacheTtlMillis;
         this.unpartitionedTableDeleteManager = new DeleteManager(typeManager);
     }
 
@@ -332,7 +335,7 @@ private Optional<ConnectorTableCredentials> maybeRefreshCredentials(
                 () -> {
                     TrinoCatalog catalog = catalogFactory.create(session.getIdentity());
                     BaseTable freshTable = catalog.loadTable(session, schemaTableName);
-                    return IcebergTableCredentials.forFileIO(freshTable.io());
+                    return IcebergTableCredentials.forFileIO(freshTable.io(), credentialCacheTtlMillis);
                 });
     }
 

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProviderFactory.java

@@ -26,8 +26,8 @@
 import io.trino.spi.connector.ConnectorPageSourceProviderFactory;
 import io.trino.spi.type.TypeManager;
 
-import static io.trino.plugin.iceberg.IcebergUtil.WORKER_CREDENTIAL_CACHE_TTL;
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 public class IcebergPageSourceProviderFactory
         implements ConnectorPageSourceProviderFactory
@@ -39,15 +39,11 @@ public class IcebergPageSourceProviderFactory
     private final ParquetReaderOptions parquetReaderOptions;
     private final TypeManager typeManager;
     private final TrinoCatalogFactory catalogFactory;
+    private final long credentialCacheTtlMillis;
     // Shared across all providers created by this factory so that concurrent tasks on the
     // same worker do not all call the REST catalog when credentials need refreshing.
     // Size-bounded to cap memory even when many tables are being scanned simultaneously.
-    private final Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache =
-            EvictableCacheBuilder.newBuilder()
-                    .maximumSize(10_000)
-                    .expireAfterWrite(WORKER_CREDENTIAL_CACHE_TTL)
-                    .shareNothingWhenDisabled()
-                    .build();
+    private final Cache<CredentialCacheKey, IcebergTableCredentials> refreshedCredentialCache;
 
     @Inject
     public IcebergPageSourceProviderFactory(
@@ -57,7 +53,8 @@ public IcebergPageSourceProviderFactory(
             OrcReaderConfig orcReaderConfig,
             ParquetReaderConfig parquetReaderConfig,
             TypeManager typeManager,
-            TrinoCatalogFactory catalogFactory)
+            TrinoCatalogFactory catalogFactory,
+            IcebergConfig config)
     {
         this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
         this.fileIoFactory = requireNonNull(fileIoFactory, "fileIoFactory is null");
@@ -66,11 +63,17 @@ public IcebergPageSourceProviderFactory(
         this.parquetReaderOptions = parquetReaderConfig.toParquetReaderOptions();
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
+        this.credentialCacheTtlMillis = config.getCredentialCacheTtl().toMillis();
+        this.refreshedCredentialCache = EvictableCacheBuilder.newBuilder()
+                .maximumSize(10_000)
+                .expireAfterWrite(config.getWorkerCredentialCacheTtl().toMillis(), MILLISECONDS)
+                .shareNothingWhenDisabled()
+                .build();
     }
 
     @Override
     public IcebergPageSourceProvider createPageSourceProvider()
     {
-        return new IcebergPageSourceProvider(fileSystemFactory, fileIoFactory, fileFormatDataSourceStats, orcReaderOptions, parquetReaderOptions, typeManager, catalogFactory, refreshedCredentialCache);
+        return new IcebergPageSourceProvider(fileSystemFactory, fileIoFactory, fileFormatDataSourceStats, orcReaderOptions, parquetReaderOptions, typeManager, catalogFactory, refreshedCredentialCache, credentialCacheTtlMillis);
     }
 }