(NOT_READY) Fixed IcebergCatalog stale FileIO after metadata refresh

CHANGELOG.md

@@ -84,6 +84,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 ### Fixes
 
+- Fixed IcebergCatalog stale FileIO after metadata refresh, resolving AWS assumeRole permission errors.
+
 ### Commits
 
 ## [1.2.0-incubating]

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -1409,6 +1409,22 @@ public void doRefresh() {
                       Set.of(PolarisStorageActions.READ, PolarisStorageActions.LIST));
               return TableMetadataParser.read(fileIO, metadataLocation);
             });
+
+        // After a refresh, re-load the FileIO with the new table metadata properties to
+        // ensure the right permissions are present for subsequent file system interactions.
+        if (currentMetadata != null) {
+          tableFileIO =
+              loadFileIOForTableLike(
+                  tableIdentifier,
+                  StorageUtil.getLocationsUsedByTable(currentMetadata),
+                  resolvedEntities,
+                  new HashMap<>(currentMetadata.properties()),
+                  Set.of(
+                      PolarisStorageActions.READ,
+                      PolarisStorageActions.WRITE,
+                      PolarisStorageActions.LIST));
+        }
+
         polarisEventListener.onEvent(
             new PolarisEvent(
                 PolarisEventType.AFTER_REFRESH_TABLE,

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

@@ -114,6 +114,7 @@
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.persistence.cache.EntityCache;
 import org.apache.polaris.core.persistence.dao.entity.BaseResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
@@ -124,6 +125,7 @@
 import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.storage.CredentialVendingContext;
+import org.apache.polaris.core.storage.PolarisStorageActions;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 import org.apache.polaris.core.storage.StorageAccessConfig;
@@ -2503,4 +2505,66 @@ public void testPaginatedListNamespaces() {
       }
     }
   }
+
+  @Test
+  public void testFileIOIsRefreshedOnTableRefresh() {
+    // Catalog use the spied provider to verify internal behavior
+    StorageAccessConfigProvider spiedProvider = spy(storageAccessConfigProvider);
+    PolarisPassthroughResolutionView passthroughView =
+        new PolarisPassthroughResolutionView(
+            resolutionManifestFactory, authenticatedRoot, CATALOG_NAME);
+    TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
+    IcebergCatalog catalog =
+        new IcebergCatalog(
+            diagServices,
+            resolverFactory,
+            metaStoreManager,
+            polarisContext,
+            passthroughView,
+            authenticatedRoot,
+            taskExecutor,
+            spiedProvider,
+            fileIOFactory,
+            polarisEventListener,
+            eventMetadataFactory);
+    catalog.setCatalogFileIo(new InMemoryFileIO());
+    catalog.initialize(
+        CATALOG_NAME,
+        Map.of(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+    // Create a table
+    TableIdentifier tableIdentifier = TableIdentifier.of(NS, "refresh_test_table");
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(NS);
+    }
+    catalog.buildTable(tableIdentifier, SCHEMA).create();
+
+    // Get the table operations
+    IcebergCatalog.BasePolarisTableOperations operations =
+        (IcebergCatalog.BasePolarisTableOperations)
+            ((BaseTable) catalog.loadTable(tableIdentifier)).operations();
+
+    // Verify initial state
+    assertThat(operations.io()).isNotNull();
+
+    // refresh the table
+    operations.refresh();
+
+    // Verify that getStorageAccessConfig was called with WRITE permissions at least twice:
+    // 1. during table creation
+    // 2. during table refresh, to update the FileIO with WRITE permissions
+    Mockito.verify(spiedProvider, Mockito.atLeast(2))
+        .getStorageAccessConfig(
+            Mockito.eq(tableIdentifier),
+            Mockito.anySet(),
+            Mockito.argThat(
+                actions ->
+                    actions.containsAll(
+                        Set.of(
+                            PolarisStorageActions.READ,
+                            PolarisStorageActions.WRITE,
+                            PolarisStorageActions.LIST))),
+            Mockito.any(),
+            Mockito.any(PolarisResolvedPathWrapper.class));
+  }
 }

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

@@ -177,9 +177,10 @@ public void testLoadFileIOForCleanupTask(String scheme) {
         .isInstanceOf(InMemoryFileIO.class);
 
     // 1. BasePolarisCatalog:doCommit: for writing the table during the creation
-    // 2. BasePolarisCatalog:doRefresh: for reading the table during the drop
+    // 2. BasePolarisCatalog:doRefresh: for reading the table during the drop (2 calls: 1 to read, 1
+    // to reload)
     // 3. TaskFileIOSupplier:apply: for clean up metadata files and merge files
-    Mockito.verify(testServices.fileIOFactory(), Mockito.times(3))
+    Mockito.verify(testServices.fileIOFactory(), Mockito.times(4))
         .loadFileIO(Mockito.any(), Mockito.any(), Mockito.any());
   }