Use FileIOFactory to get the creds for cleanup task

Author: XJDKC

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/BasePolarisCatalogTest.java

@@ -1391,10 +1391,7 @@ public void testDropTableWithPurge() {
         .containsEntry(PolarisCredentialProperty.AWS_KEY_ID, TEST_ACCESS_KEY)
         .containsEntry(PolarisCredentialProperty.AWS_SECRET_KEY, SECRET_ACCESS_KEY)
         .containsEntry(PolarisCredentialProperty.AWS_TOKEN, SESSION_TOKEN);
-    FileIO fileIO =
-        new TaskFileIOSupplier(
-                createMockMetaStoreManagerFactory(), fileIOFactory, configurationStore)
-            .apply(taskEntity, realmId);
+    FileIO fileIO = new TaskFileIOSupplier(fileIOFactory).apply(taskEntity, realmId);
     Assertions.assertThat(fileIO).isNotNull().isInstanceOf(InMemoryFileIO.class);
   }
 

service/common/src/main/java/org/apache/polaris/service/task/TaskFileIOSupplier.java

@@ -21,68 +21,51 @@
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.BiFunction;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.io.FileIO;
-import org.apache.polaris.core.PolarisConfiguration;
-import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.context.RealmId;
 import org.apache.polaris.core.entity.PolarisTaskConstants;
+import org.apache.polaris.core.entity.TableLikeEntity;
 import org.apache.polaris.core.entity.TaskEntity;
-import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
-import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
-import org.apache.polaris.core.persistence.PolarisMetaStoreSession;
+import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
+import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
+import org.apache.polaris.core.storage.PolarisStorageActions;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 
 @ApplicationScoped
 public class TaskFileIOSupplier implements BiFunction<TaskEntity, RealmId, FileIO> {
-  private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final FileIOFactory fileIOFactory;
-  private final PolarisConfigurationStore configurationStore;
 
   @Inject
-  public TaskFileIOSupplier(
-      MetaStoreManagerFactory metaStoreManagerFactory,
-      FileIOFactory fileIOFactory,
-      PolarisConfigurationStore configurationStore) {
-    this.metaStoreManagerFactory = metaStoreManagerFactory;
+  public TaskFileIOSupplier(FileIOFactory fileIOFactory) {
     this.fileIOFactory = fileIOFactory;
-    this.configurationStore = configurationStore;
   }
 
   @Override
   public FileIO apply(TaskEntity task, RealmId realmId) {
     Map<String, String> internalProperties = task.getInternalPropertiesAsMap();
-    String location = internalProperties.get(PolarisTaskConstants.STORAGE_LOCATION);
-    PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerFactory.getOrCreateMetaStoreManager(realmId);
-    PolarisMetaStoreSession metaStoreSession =
-        metaStoreManagerFactory.getOrCreateSessionSupplier(realmId).get();
     Map<String, String> properties = new HashMap<>(internalProperties);
 
-    Boolean skipCredentialSubscopingIndirection =
-        configurationStore.getConfiguration(
-            realmId,
-            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.key,
-            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.defaultValue);
+    TableLikeEntity tableEntity = TableLikeEntity.of(task);
+    TableIdentifier identifier = tableEntity.getTableIdentifier();
+    String location = properties.get(PolarisTaskConstants.STORAGE_LOCATION);
+    Set<String> locations = Set.of(location);
+    Set<PolarisStorageActions> storageActions = Set.of(PolarisStorageActions.ALL);
+    ResolvedPolarisEntity resolvedTaskEntity =
+        new ResolvedPolarisEntity(task, List.of(), List.of());
+    PolarisResolvedPathWrapper resolvedPath =
+        new PolarisResolvedPathWrapper(List.of(resolvedTaskEntity));
 
-    if (!skipCredentialSubscopingIndirection) {
-      properties.putAll(
-          metaStoreManagerFactory
-              .getOrCreateStorageCredentialCache(realmId)
-              .getOrGenerateSubScopeCreds(
-                  metaStoreManager,
-                  metaStoreSession,
-                  task,
-                  true,
-                  Set.of(location),
-                  Set.of(location)));
-    }
     String ioImpl =
         properties.getOrDefault(
             CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.io.ResolvingFileIO");
-    return fileIOFactory.loadFileIO(realmId, ioImpl, properties);
+
+    return fileIOFactory.loadFileIO(
+        realmId, ioImpl, properties, identifier, locations, storageActions, resolvedPath);
   }
 }