pass skip_credential_subscoping_indirection param to TaskFileIOSupplier

dropwizard/service/src/main/java/org/apache/polaris/service/dropwizard/PolarisApplication.java

@@ -82,6 +82,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.iceberg.rest.RESTSerializers;
+import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
 import org.apache.polaris.core.auth.PolarisAuthorizer;
 import org.apache.polaris.core.auth.PolarisAuthorizerImpl;
@@ -329,7 +330,10 @@ protected void configure() {
                 TaskExecutorImpl taskExecutor =
                     new TaskExecutorImpl(taskConfig.executorService(), metaStoreManagerFactory);
                 TaskFileIOSupplier fileIOSupplier =
-                    new TaskFileIOSupplier(metaStoreManagerFactory, fileIOFactory);
+                    new TaskFileIOSupplier(
+                        metaStoreManagerFactory,
+                        fileIOFactory,
+                        configuration.findService(PolarisConfigurationStore.class));
                 taskExecutor.addTaskHandler(
                     new TableCleanupTaskHandler(
                         taskExecutor, metaStoreManagerFactory, fileIOSupplier));

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

@@ -1357,7 +1357,10 @@ public void testDropTableWithPurge() {
         .containsEntry(PolarisCredentialProperty.AWS_SECRET_KEY, SECRET_ACCESS_KEY)
         .containsEntry(PolarisCredentialProperty.AWS_TOKEN, SESSION_TOKEN);
     FileIO fileIO =
-        new TaskFileIOSupplier(createMockMetaStoreManagerFactory(), new DefaultFileIOFactory())
+        new TaskFileIOSupplier(
+                createMockMetaStoreManagerFactory(),
+                new DefaultFileIOFactory(),
+                polarisContext.getConfigurationStore())
             .apply(taskEntity);
     Assertions.assertThat(fileIO).isNotNull().isInstanceOf(InMemoryFileIO.class);
   }

polaris-core/src/main/java/org/apache/polaris/core/PolarisConfiguration.java

@@ -110,6 +110,21 @@ public static <T> Builder<T> builder() {
               .defaultValue(false)
               .build();
 
+  public static final PolarisConfiguration<Boolean> SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION")
+          .description(
+              "If set to true, skip credential-subscoping indirection entirely whenever trying\n"
+                  + "   to obtain storage credentials for instantiating a FileIO. If 'true', no attempt is made\n"
+                  + "   to use StorageConfigs to generate table-specific storage credentials, but instead the default\n"
+                  + "   fallthrough of table-level credential properties or else provider-specific APPLICATION_DEFAULT\n"
+                  + "   credential-loading will be used for the FileIO.\n"
+                  + "   Typically this setting is used in single-tenant server deployments that don't rely on\n"
+                  + "   \"credential-vending\" and can use server-default environment variables or credential config\n"
+                  + "   files for all storage access, or in test/dev scenarios.")
+          .defaultValue(false)
+          .build();
+
   public static final PolarisConfiguration<Boolean> ALLOW_TABLE_LOCATION_OVERLAP =
       PolarisConfiguration.<Boolean>builder()
           .key("ALLOW_TABLE_LOCATION_OVERLAP")

service/common/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java

@@ -121,18 +121,6 @@ public class BasePolarisCatalog extends BaseMetastoreViewCatalog
   static final String ALLOW_SPECIFYING_FILE_IO_IMPL = "ALLOW_SPECIFYING_FILE_IO_IMPL";
   static final boolean ALLOW_SPECIFYING_FILE_IO_IMPL_DEFAULT = false;
 
-  // Config key for whether to skip credential-subscoping indirection entirely whenever trying
-  // to obtain storage credentials for instantiating a FileIO. If 'true', no attempt is made
-  // to use StorageConfigs to generate table-specific storage credentials, but instead the default
-  // fallthrough of table-level credential properties or else provider-specific APPLICATION_DEFAULT
-  // credential-loading will be used for the FileIO.
-  // Typically this setting is used in single-tenant server deployments that don't rely on
-  // "credential-vending" and can use server-default environment variables or credential config
-  // files for all storage access, or in test/dev scenarios.
-  static final String SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION =
-      "SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION";
-  static final boolean SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION_DEFAULT = false;
-
   // Config key for initializing a default "catalogFileIO" that is available either via getIo()
   // or for any TableOperations/ViewOperations instantiated, via ops.io() before entity-specific
   // FileIO initialization is triggered for any such operations.
@@ -874,7 +862,8 @@ private Map<String, String> refreshCredentials(
       PolarisEntity entity) {
     Boolean skipCredentialSubscopingIndirection =
         getBooleanContextConfiguration(
-            SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION, SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION_DEFAULT);
+            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.key,
+            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.defaultValue);
     if (Boolean.TRUE.equals(skipCredentialSubscopingIndirection)) {
       LOGGER
           .atInfo()

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

@@ -24,6 +24,8 @@
 import java.util.function.Function;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.io.FileIO;
+import org.apache.polaris.core.PolarisConfiguration;
+import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.PolarisTaskConstants;
 import org.apache.polaris.core.entity.TaskEntity;
@@ -34,11 +36,15 @@
 public class TaskFileIOSupplier implements Function<TaskEntity, FileIO> {
   private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final FileIOFactory fileIOFactory;
+  private final PolarisConfigurationStore configurationStore;
 
   public TaskFileIOSupplier(
-      MetaStoreManagerFactory metaStoreManagerFactory, FileIOFactory fileIOFactory) {
+      MetaStoreManagerFactory metaStoreManagerFactory,
+      FileIOFactory fileIOFactory,
+      PolarisConfigurationStore configurationStore) {
     this.metaStoreManagerFactory = metaStoreManagerFactory;
     this.fileIOFactory = fileIOFactory;
+    this.configurationStore = configurationStore;
   }
 
   @Override
@@ -49,16 +55,25 @@ public FileIO apply(TaskEntity task) {
         metaStoreManagerFactory.getOrCreateMetaStoreManager(
             CallContext.getCurrentContext().getRealmContext());
     Map<String, String> properties = new HashMap<>(internalProperties);
-    properties.putAll(
-        metaStoreManagerFactory
-            .getOrCreateStorageCredentialCache(CallContext.getCurrentContext().getRealmContext())
-            .getOrGenerateSubScopeCreds(
-                metaStoreManager,
-                CallContext.getCurrentContext().getPolarisCallContext(),
-                task,
-                true,
-                Set.of(location),
-                Set.of(location)));
+
+    Boolean skipCredentialSubscopingIndirection =
+        configurationStore.getConfiguration(
+            null,
+            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.key,
+            PolarisConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.defaultValue);
+
+    if (!skipCredentialSubscopingIndirection) {
+      properties.putAll(
+          metaStoreManagerFactory
+              .getOrCreateStorageCredentialCache(CallContext.getCurrentContext().getRealmContext())
+              .getOrGenerateSubScopeCreds(
+                  metaStoreManager,
+                  CallContext.getCurrentContext().getPolarisCallContext(),
+                  task,
+                  true,
+                  Set.of(location),
+                  Set.of(location)));
+    }
     String ioImpl =
         properties.getOrDefault(
             CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.io.ResolvingFileIO");