Make AccessConfigProvider request-scoped

- add `StorageCredentialsVendor` as request-scoped wrapper around `PolarisCredentialVendor`
- make `FileIOFactory` request-scoped
- make `TaskFileIOSupplier` request-scoped

Author: XN137

polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java

@@ -1594,7 +1594,7 @@ private void revokeGrantRecord(
       @Nonnull PolarisCallContext callCtx,
       long catalogId,
       long entityId,
-      PolarisEntityType entityType,
+      @Nonnull PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,

polaris-core/src/main/java/org/apache/polaris/core/persistence/TransactionWorkspaceMetaStoreManager.java

@@ -338,11 +338,11 @@ public EntitiesResult loadTasks(
   }
 
   @Override
-  public ScopedCredentialsResult getSubscopedCredsForEntity(
+  public @Nonnull ScopedCredentialsResult getSubscopedCredsForEntity(
       @Nonnull PolarisCallContext callCtx,
       long catalogId,
       long entityId,
-      PolarisEntityType entityType,
+      @Nonnull PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java

@@ -2091,7 +2091,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
       @Nonnull PolarisCallContext callCtx,
       long catalogId,
       long entityId,
-      PolarisEntityType entityType,
+      @Nonnull PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisCredentialVendor.java

@@ -49,7 +49,7 @@ ScopedCredentialsResult getSubscopedCredsForEntity(
       @Nonnull PolarisCallContext callCtx,
       long catalogId,
       long entityId,
-      PolarisEntityType entityType,
+      @Nonnull PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,

polaris-core/src/main/java/org/apache/polaris/core/storage/StorageCredentialsVendor.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.core.storage;
+
+import jakarta.annotation.Nonnull;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.PolarisEntity;
+import org.apache.polaris.core.persistence.dao.entity.ScopedCredentialsResult;
+
+public class StorageCredentialsVendor {
+
+  private final PolarisCredentialVendor polarisCredentialVendor;
+  private final CallContext callContext;
+
+  public StorageCredentialsVendor(
+      PolarisCredentialVendor polarisCredentialVendor, CallContext callContext) {
+    this.polarisCredentialVendor = polarisCredentialVendor;
+    this.callContext = callContext;
+  }
+
+  public RealmContext getRealmContext() {
+    return callContext.getRealmContext();
+  }
+
+  public RealmConfig getRealmConfig() {
+    return callContext.getRealmConfig();
+  }
+
+  /**
+   * Get sub-scoped credentials for an entity against the provided allowed read and write locations.
+   *
+   * @param entity the entity
+   * @param allowListOperation whether to allow LIST operation on the allowedReadLocations and
+   *     allowedWriteLocations
+   * @param allowedReadLocations a set of allowed to read locations
+   * @param allowedWriteLocations a set of allowed to write locations
+   * @param refreshCredentialsEndpoint an optional endpoint to use for refreshing credentials. If
+   *     supported by the storage type it will be returned to the client in the appropriate
+   *     properties. The endpoint may be relative to the base URI and the client is responsible for
+   *     handling the relative path
+   * @return an enum map containing the scoped credentials
+   */
+  @Nonnull
+  public ScopedCredentialsResult getSubscopedCredsForEntity(
+      @Nonnull PolarisEntity entity,
+      boolean allowListOperation,
+      @Nonnull Set<String> allowedReadLocations,
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
+    return polarisCredentialVendor.getSubscopedCredsForEntity(
+        callContext.getPolarisCallContext(),
+        entity.getCatalogId(),
+        entity.getId(),
+        entity.getType(),
+        allowListOperation,
+        allowedReadLocations,
+        allowedWriteLocations,
+        refreshCredentialsEndpoint);
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCache.java

@@ -30,15 +30,15 @@
 import java.util.Set;
 import java.util.function.Function;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
-import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.persistence.dao.entity.ScopedCredentialsResult;
 import org.apache.polaris.core.storage.AccessConfig;
-import org.apache.polaris.core.storage.PolarisCredentialVendor;
+import org.apache.polaris.core.storage.StorageCredentialsVendor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -95,29 +95,30 @@ private long maxCacheDurationMs(RealmConfig realmConfig) {
   /**
    * Either get from the cache or generate a new entry for a scoped creds
    *
-   * @param credentialVendor the credential vendor used to generate a new scoped creds if needed
-   * @param callCtx the call context
+   * @param storageCredentialsVendor the credential vendor used to generate a new scoped creds if
+   *     needed
    * @param polarisEntity the polaris entity that is going to scoped creds
    * @param allowListOperation whether allow list action on the provided read and write locations
    * @param allowedReadLocations a set of allowed to read locations
    * @param allowedWriteLocations a set of allowed to write locations.
    * @return the a map of string containing the scoped creds information
    */
   public AccessConfig getOrGenerateSubScopeCreds(
-      @Nonnull PolarisCredentialVendor credentialVendor,
-      @Nonnull PolarisCallContext callCtx,
+      @Nonnull StorageCredentialsVendor storageCredentialsVendor,
       @Nonnull PolarisEntity polarisEntity,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations,
       Optional<String> refreshCredentialsEndpoint) {
+    RealmContext realmContext = storageCredentialsVendor.getRealmContext();
+    RealmConfig realmConfig = storageCredentialsVendor.getRealmConfig();
     if (!isTypeSupported(polarisEntity.getType())) {
       diagnostics.fail(
           "entity_type_not_suppported_to_scope_creds", "type={}", polarisEntity.getType());
     }
     StorageCredentialCacheKey key =
         StorageCredentialCacheKey.of(
-            callCtx.getRealmContext().getRealmIdentifier(),
+            realmContext.getRealmIdentifier(),
             polarisEntity,
             allowListOperation,
             allowedReadLocations,
@@ -128,17 +129,14 @@ public AccessConfig getOrGenerateSubScopeCreds(
         k -> {
           LOGGER.atDebug().log("StorageCredentialCache::load");
           ScopedCredentialsResult scopedCredentialsResult =
-              credentialVendor.getSubscopedCredsForEntity(
-                  callCtx,
-                  k.catalogId(),
-                  polarisEntity.getId(),
-                  polarisEntity.getType(),
-                  k.allowedListAction(),
-                  k.allowedReadLocations(),
-                  k.allowedWriteLocations(),
-                  k.refreshCredentialsEndpoint());
+              storageCredentialsVendor.getSubscopedCredsForEntity(
+                  polarisEntity,
+                  allowListOperation,
+                  allowedReadLocations,
+                  allowedWriteLocations,
+                  refreshCredentialsEndpoint);
           if (scopedCredentialsResult.isSuccess()) {
-            long maxCacheDurationMs = maxCacheDurationMs(callCtx.getRealmConfig());
+            long maxCacheDurationMs = maxCacheDurationMs(realmConfig);
             return new StorageCredentialCacheEntry(
                 scopedCredentialsResult.getAccessConfig(), maxCacheDurationMs);
           }