Use AccessConfigProvider.getAccessConfig in DefaultFileIOFactory

runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java

@@ -38,7 +38,6 @@
 import org.apache.polaris.core.storage.AccessConfig;
 import org.apache.polaris.core.storage.PolarisCredentialVendor;
 import org.apache.polaris.core.storage.PolarisStorageActions;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 
 /**
  * A default FileIO factory implementation for creating Iceberg {@link FileIO} instances with
@@ -52,15 +51,14 @@
 @Identifier("default")
 public class DefaultFileIOFactory implements FileIOFactory {
 
-  private final StorageCredentialCache storageCredentialCache;
   private final MetaStoreManagerFactory metaStoreManagerFactory;
+  private final AccessConfigProvider accessConfigProvider;
 
   @Inject
   public DefaultFileIOFactory(
-      StorageCredentialCache storageCredentialCache,
-      MetaStoreManagerFactory metaStoreManagerFactory) {
-    this.storageCredentialCache = storageCredentialCache;
+      MetaStoreManagerFactory metaStoreManagerFactory, AccessConfigProvider accessConfigProvider) {
     this.metaStoreManagerFactory = metaStoreManagerFactory;
+    this.accessConfigProvider = accessConfigProvider;
   }
 
   @Override
@@ -83,15 +81,13 @@ public FileIO loadFileIO(
     Optional<AccessConfig> accessConfig =
         storageInfoEntity.map(
             storageInfo ->
-                FileIOUtil.refreshAccessConfig(
+                accessConfigProvider.getAccessConfig(
                     callContext,
-                    storageCredentialCache,
-                    credentialVendor,
                     identifier,
                     tableLocations,
                     storageActions,
-                    storageInfo,
-                    Optional.empty()));
+                    Optional.empty(),
+                    resolvedEntityPath));
 
     // Update the FileIO with the subscoped credentials
     // Update with properties in case there are table-level overrides the credentials should

runtime/service/src/main/java/org/apache/polaris/service/catalog/io/WasbTranslatingFileIOFactory.java

@@ -42,9 +42,9 @@ public class WasbTranslatingFileIOFactory implements FileIOFactory {
   @Inject
   public WasbTranslatingFileIOFactory(
       StorageCredentialCache storageCredentialCache,
-      MetaStoreManagerFactory metaStoreManagerFactory) {
-    defaultFileIOFactory =
-        new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+      MetaStoreManagerFactory metaStoreManagerFactory,
+      AccessConfigProvider accessConfigProvider) {
+    defaultFileIOFactory = new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
   }
 
   @Override

runtime/service/src/test/java/org/apache/polaris/service/catalog/generic/AbstractPolarisGenericTableCatalogTest.java

@@ -67,6 +67,7 @@
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
+import org.apache.polaris.service.catalog.io.AccessConfigProvider;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.config.ReservedProperties;
@@ -120,6 +121,7 @@ public abstract class AbstractPolarisGenericTableCatalogTest {
   private PolarisPrincipal authenticatedRoot;
   private PolarisEntity catalogEntity;
   private SecurityContext securityContext;
+  private AccessConfigProvider accessConfigProvider;
 
   protected static final Schema SCHEMA =
       new Schema(
@@ -156,6 +158,8 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             configurationStore);
     realmConfig = polarisContext.getRealmConfig();
+    accessConfigProvider =
+        new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -211,7 +215,7 @@ public void before(TestInfo testInfo) {
         new PolarisPassthroughResolutionView(
             resolutionManifestFactory, securityContext, CATALOG_NAME);
     TaskExecutor taskExecutor = Mockito.mock();
-    this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+    this.fileIOFactory = new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
 
     StsClient stsClient = Mockito.mock(StsClient.class);
     when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java

@@ -135,6 +135,7 @@
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.Profiles;
+import org.apache.polaris.service.catalog.io.AccessConfigProvider;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.ExceptionMappingFileIO;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
@@ -249,6 +250,7 @@ public Map<String, String> getConfigOverrides() {
   private SecurityContext securityContext;
   private TestPolarisEventListener testPolarisEventListener;
   private ReservedProperties reservedProperties;
+  private AccessConfigProvider accessConfigProvider;
 
   @BeforeAll
   public static void setUpMocks() {
@@ -286,7 +288,8 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             configurationStore);
     realmConfig = polarisContext.getRealmConfig();
-
+    accessConfigProvider =
+        new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
     EntityCache entityCache = createEntityCache(diagServices, realmConfig, metaStoreManager);
     resolverFactory =
         (securityContext, referenceCatalogName) ->
@@ -352,7 +355,7 @@ public void before(TestInfo testInfo) {
                     .build()
                     .asCatalog(serviceIdentityProvider)));
 
-    this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+    this.fileIOFactory = new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
 
     StsClient stsClient = Mockito.mock(StsClient.class);
     when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))
@@ -997,7 +1000,7 @@ public void testValidateNotificationFailToCreateFileIO() {
     final String tableLocation = "s3://externally-owned-bucket/validate_table/";
     final String tableMetadataLocation = tableLocation + "metadata/";
     FileIOFactory fileIOFactory =
-        spy(new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory));
+        spy(new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider));
     IcebergCatalog catalog = newIcebergCatalog(catalog().name(), metaStoreManager, fileIOFactory);
     catalog.initialize(
         CATALOG_NAME,
@@ -1915,7 +1918,7 @@ public void testDropTableWithPurge() {
         .containsEntry(StorageAccessProperty.AWS_TOKEN.getPropertyName(), SESSION_TOKEN);
     FileIO fileIO =
         new TaskFileIOSupplier(
-                new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory))
+                new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider))
             .apply(taskEntity, TABLE, polarisContext);
     Assertions.assertThat(fileIO).isNotNull().isInstanceOf(ExceptionMappingFileIO.class);
     Assertions.assertThat(((ExceptionMappingFileIO) fileIO).getInnerIo())
@@ -2042,7 +2045,7 @@ static Stream<Arguments> testRetriableException() {
   @Test
   public void testFileIOWrapper() {
     MeasuredFileIOFactory measured =
-        new MeasuredFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+        new MeasuredFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
     IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, metaStoreManager, measured);
     catalog.initialize(
         CATALOG_NAME,

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogViewTest.java

@@ -60,6 +60,7 @@
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.Profiles;
+import org.apache.polaris.service.catalog.io.AccessConfigProvider;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.config.ReservedProperties;
@@ -123,6 +124,7 @@ public Map<String, String> getConfigOverrides() {
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
   private RealmConfig realmConfig;
+  private AccessConfigProvider accessConfigProvider;
 
   private TestPolarisEventListener testPolarisEventListener;
 
@@ -163,7 +165,8 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             configurationStore);
     realmConfig = polarisContext.getRealmConfig();
-
+    accessConfigProvider =
+        new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
     PolarisPrincipal authenticatedRoot = PolarisPrincipal.of(rootPrincipal, Set.of());
@@ -208,7 +211,7 @@ public void before(TestInfo testInfo) {
         new PolarisPassthroughResolutionView(
             resolutionManifestFactory, securityContext, CATALOG_NAME);
     FileIOFactory fileIOFactory =
-        new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+        new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
 
     testPolarisEventListener = (TestPolarisEventListener) polarisEventListener;
     testPolarisEventListener.clear();

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandlerAuthzTest.java

@@ -1899,7 +1899,7 @@ public void testSendNotificationSufficientPrivileges() {
             resolverFactory,
             managerFactory,
             Mockito.mock(),
-            new DefaultFileIOFactory(storageCredentialCache, managerFactory),
+            new DefaultFileIOFactory(managerFactory, accessConfigProvider),
             polarisEventListener) {
           @Override
           public Catalog createCallContextCatalog(

runtime/service/src/test/java/org/apache/polaris/service/catalog/io/FileIOFactoryTest.java

@@ -102,9 +102,9 @@ public void before(TestInfo testInfo) {
 
     // Spy FileIOFactory and check if the credentials are passed to the FileIO
     TestServices.FileIOFactorySupplier fileIOFactorySupplier =
-        (storageCredentialCache, metaStoreManagerFactory) ->
+        (metaStoreManagerFactory, accessConfigProvider) ->
             Mockito.spy(
-                new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory) {
+                new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider) {
                   @Override
                   FileIO loadFileIOInternal(
                       @Nonnull String ioImplClassName, @Nonnull Map<String, String> properties) {

runtime/service/src/test/java/org/apache/polaris/service/catalog/policy/AbstractPolicyCatalogTest.java

@@ -79,6 +79,7 @@
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
+import org.apache.polaris.service.catalog.io.AccessConfigProvider;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.config.ReservedProperties;
@@ -146,6 +147,7 @@ public abstract class AbstractPolicyCatalogTest {
   private PolarisPrincipal authenticatedRoot;
   private PolarisEntity catalogEntity;
   private SecurityContext securityContext;
+  private AccessConfigProvider accessConfigProvider;
 
   @BeforeAll
   public static void setUpMocks() {
@@ -177,6 +179,8 @@ public void before(TestInfo testInfo) {
             metaStoreManagerFactory.getOrCreateSession(realmContext),
             configurationStore);
     realmConfig = polarisContext.getRealmConfig();
+    accessConfigProvider =
+        new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
 
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
@@ -230,7 +234,7 @@ public void before(TestInfo testInfo) {
         new PolarisPassthroughResolutionView(
             resolutionManifestFactory, securityContext, CATALOG_NAME);
     TaskExecutor taskExecutor = Mockito.mock();
-    this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+    this.fileIOFactory = new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
 
     StsClient stsClient = Mockito.mock(StsClient.class);
     when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))

runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java

@@ -117,7 +117,7 @@ public record TestServices(
 
   @FunctionalInterface
   public interface FileIOFactorySupplier
-      extends BiFunction<StorageCredentialCache, MetaStoreManagerFactory, FileIOFactory> {}
+      extends BiFunction<MetaStoreManagerFactory, AccessConfigProvider, FileIOFactory> {}
 
   private static class MockedConfigurationStore implements PolarisConfigurationStore {
     private final Map<String, Object> defaults;
@@ -242,8 +242,10 @@ public TestServices build() {
       PolarisCredentialManager credentialManager =
           new DefaultPolarisCredentialManager(realmContext, mockCredentialVendors);
 
+      AccessConfigProvider accessConfigProvider =
+          new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
       FileIOFactory fileIOFactory =
-          fileIOFactorySupplier.apply(storageCredentialCache, metaStoreManagerFactory);
+          fileIOFactorySupplier.apply(metaStoreManagerFactory, accessConfigProvider);
 
       TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
 
@@ -257,9 +259,6 @@ public TestServices build() {
               fileIOFactory,
               polarisEventListener);
 
-      AccessConfigProvider accessConfigProvider =
-          new AccessConfigProvider(storageCredentialCache, metaStoreManagerFactory);
-
       ReservedProperties reservedProperties = ReservedProperties.NONE;
 
       CatalogHandlerUtils catalogHandlerUtils = new CatalogHandlerUtils(realmConfig);

runtime/service/src/testFixtures/java/org/apache/polaris/service/catalog/io/MeasuredFileIOFactory.java

@@ -33,7 +33,6 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.storage.PolarisStorageActions;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 
 /**
  * A FileIOFactory that measures the number of bytes read, files written, and files deleted. It can
@@ -53,10 +52,8 @@ public class MeasuredFileIOFactory implements FileIOFactory {
 
   @Inject
   public MeasuredFileIOFactory(
-      StorageCredentialCache storageCredentialCache,
-      MetaStoreManagerFactory metaStoreManagerFactory) {
-    defaultFileIOFactory =
-        new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
+      MetaStoreManagerFactory metaStoreManagerFactory, AccessConfigProvider accessConfigProvider) {
+    defaultFileIOFactory = new DefaultFileIOFactory(metaStoreManagerFactory, accessConfigProvider);
   }
 
   @Override