Refactor storage access configuration handling

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

@@ -66,9 +66,9 @@
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyMappingUtil;
 import org.apache.polaris.core.policy.PolicyType;
-import org.apache.polaris.core.storage.PolarisCredentialProperty;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -1605,7 +1605,7 @@ private void revokeGrantRecord(
     PolarisStorageConfigurationInfo storageConfigurationInfo =
         BaseMetaStoreManager.extractStorageConfiguration(callCtx, reloadedEntity.getEntity());
     try {
-      EnumMap<PolarisCredentialProperty, String> creds =
+      EnumMap<StorageAccessProperty, String> creds =
           storageIntegration.getSubscopedCreds(
               callCtx.getDiagServices(),
               storageConfigurationInfo,

polaris-core/src/main/java/org/apache/polaris/core/persistence/dao/entity/ScopedCredentialsResult.java

@@ -24,13 +24,13 @@
 import jakarta.annotation.Nullable;
 import java.util.EnumMap;
 import java.util.Map;
-import org.apache.polaris.core.storage.PolarisCredentialProperty;
+import org.apache.polaris.core.storage.StorageAccessProperty;
 
 /** Result of a getSubscopedCredsForEntity() call */
 public class ScopedCredentialsResult extends BaseResult {
 
   // null if not success. Else, set of name/value pairs for the credentials
-  private final EnumMap<PolarisCredentialProperty, String> credentials;
+  private final EnumMap<StorageAccessProperty, String> credentials;
 
   /**
    * Constructor for an error
@@ -49,7 +49,7 @@ public ScopedCredentialsResult(
    *
    * @param credentials credentials
    */
-  public ScopedCredentialsResult(@Nonnull EnumMap<PolarisCredentialProperty, String> credentials) {
+  public ScopedCredentialsResult(@Nonnull EnumMap<StorageAccessProperty, String> credentials) {
     super(ReturnStatus.SUCCESS);
     this.credentials = credentials;
   }
@@ -60,13 +60,13 @@ private ScopedCredentialsResult(
       @JsonProperty("extraInformation") String extraInformation,
       @JsonProperty("credentials") Map<String, String> credentials) {
     super(returnStatus, extraInformation);
-    this.credentials = new EnumMap<>(PolarisCredentialProperty.class);
+    this.credentials = new EnumMap<>(StorageAccessProperty.class);
     if (credentials != null) {
-      credentials.forEach((k, v) -> this.credentials.put(PolarisCredentialProperty.valueOf(k), v));
+      credentials.forEach((k, v) -> this.credentials.put(StorageAccessProperty.valueOf(k), v));
     }
   }
 
-  public EnumMap<PolarisCredentialProperty, String> getCredentials() {
+  public EnumMap<StorageAccessProperty, String> getCredentials() {
     return credentials;
   }
 }

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

@@ -67,9 +67,9 @@
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyMappingUtil;
 import org.apache.polaris.core.policy.PolicyType;
-import org.apache.polaris.core.storage.PolarisCredentialProperty;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
+import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -2035,7 +2035,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
     PolarisStorageConfigurationInfo storageConfigurationInfo =
         BaseMetaStoreManager.extractStorageConfiguration(callCtx, reloadedEntity.getEntity());
     try {
-      EnumMap<PolarisCredentialProperty, String> creds =
+      EnumMap<StorageAccessProperty, String> creds =
           storageIntegration.getSubscopedCreds(
               callCtx.getDiagServices(),
               storageConfigurationInfo,

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

@@ -0,0 +1,33 @@
+/*
+ * 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 java.util.Map;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+@PolarisImmutable
+public interface AccessConfig {
+  Map<String, String> credentials();
+
+  Map<String, String> extraProperties();
+
+  static ImmutableAccessConfig.Builder builder() {
+    return ImmutableAccessConfig.builder();
+  }
+}

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

@@ -54,7 +54,7 @@ public String getStorageIdentifierOrId() {
    * @param allowedWriteLocations a set of allowed to write locations
    * @return An enum map including the scoped credentials
    */
-  public abstract EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
+  public abstract EnumMap<StorageAccessProperty, String> getSubscopedCreds(
       @Nonnull PolarisDiagnostics diagnostics,
       @Nonnull T storageConfig,
       boolean allowListOperation,

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisCredentialProperty.java → polaris-core/src/main/java/org/apache/polaris/core/storage/StorageAccessProperty.java

@@ -18,15 +18,23 @@
  */
 package org.apache.polaris.core.storage;
 
-/** Enum of polaris supported credential properties */
-public enum PolarisCredentialProperty {
+/**
+ * A subset of Iceberg catalog properties recognized by Polaris.
+ *
+ * <p>Most of these properties are meant to configure Iceberg FileIO objects for accessing data in
+ * storage.
+ */
+public enum StorageAccessProperty {
   AWS_KEY_ID(String.class, "s3.access-key-id", "the aws access key id"),
   AWS_SECRET_KEY(String.class, "s3.secret-access-key", "the aws access key secret"),
   AWS_TOKEN(String.class, "s3.session-token", "the aws scoped access token"),
   AWS_SESSION_TOKEN_EXPIRES_AT_MS(
       String.class,
       "s3.session-token-expires-at-ms",
       "the time the aws session token expires, in milliseconds"),
+  AWS_ENDPOINT(String.class, "s3.endpoint", "the S3 endpoint to use for requests", false),
+  AWS_PATH_STYLE_ACCESS(
+      Boolean.class, "s3.path-style-access", "whether to use S3 path style access", false),
   CLIENT_REGION(
       String.class, "client.region", "region to configure client for making requests to AWS"),
 
@@ -50,19 +58,30 @@ public enum PolarisCredentialProperty {
   private final Class valueType;
   private final String propertyName;
   private final String description;
+  private final boolean isCredential;
 
   /*
   s3.access-key-id`: id for for credentials that provide access to the data in S3
            - `s3.secret-access-key`: secret for credentials that provide access to data in S3
            - `s3.session-token
    */
-  PolarisCredentialProperty(Class valueType, String propertyName, String description) {
+  StorageAccessProperty(Class valueType, String propertyName, String description) {
+    this(valueType, propertyName, description, true);
+  }
+
+  StorageAccessProperty(
+      Class valueType, String propertyName, String description, boolean isCredential) {
     this.valueType = valueType;
     this.propertyName = propertyName;
     this.description = description;
+    this.isCredential = isCredential;
   }
 
   public String getPropertyName() {
     return propertyName;
   }
+
+  public boolean isCredential() {
+    return isCredential;
+  }
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java

@@ -31,7 +31,7 @@
 import java.util.stream.Stream;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
-import org.apache.polaris.core.storage.PolarisCredentialProperty;
+import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.apache.polaris.core.storage.StorageUtil;
 import software.amazon.awssdk.policybuilder.iam.IamConditionOperator;
 import software.amazon.awssdk.policybuilder.iam.IamEffect;
@@ -54,7 +54,7 @@ public AwsCredentialsStorageIntegration(StsClient stsClient) {
 
   /** {@inheritDoc} */
   @Override
-  public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
+  public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
       @Nonnull PolarisDiagnostics diagnostics,
       @Nonnull AwsStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
@@ -75,28 +75,28 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
                         .toJson())
                 .durationSeconds(loadConfig(STORAGE_CREDENTIAL_DURATION_SECONDS))
                 .build());
-    EnumMap<PolarisCredentialProperty, String> credentialMap =
-        new EnumMap<>(PolarisCredentialProperty.class);
-    credentialMap.put(PolarisCredentialProperty.AWS_KEY_ID, response.credentials().accessKeyId());
+    EnumMap<StorageAccessProperty, String> credentialMap =
+        new EnumMap<>(StorageAccessProperty.class);
+    credentialMap.put(StorageAccessProperty.AWS_KEY_ID, response.credentials().accessKeyId());
     credentialMap.put(
-        PolarisCredentialProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
-    credentialMap.put(PolarisCredentialProperty.AWS_TOKEN, response.credentials().sessionToken());
+        StorageAccessProperty.AWS_SECRET_KEY, response.credentials().secretAccessKey());
+    credentialMap.put(StorageAccessProperty.AWS_TOKEN, response.credentials().sessionToken());
     Optional.ofNullable(response.credentials().expiration())
         .ifPresent(
             i -> {
               credentialMap.put(
-                  PolarisCredentialProperty.EXPIRATION_TIME, String.valueOf(i.toEpochMilli()));
+                  StorageAccessProperty.EXPIRATION_TIME, String.valueOf(i.toEpochMilli()));
               credentialMap.put(
-                  PolarisCredentialProperty.AWS_SESSION_TOKEN_EXPIRES_AT_MS,
+                  StorageAccessProperty.AWS_SESSION_TOKEN_EXPIRES_AT_MS,
                   String.valueOf(i.toEpochMilli()));
             });
 
     if (storageConfig.getRegion() != null) {
-      credentialMap.put(PolarisCredentialProperty.CLIENT_REGION, storageConfig.getRegion());
+      credentialMap.put(StorageAccessProperty.CLIENT_REGION, storageConfig.getRegion());
     }
 
     if (storageConfig.getAwsPartition().equals("aws-us-gov")
-        && credentialMap.get(PolarisCredentialProperty.CLIENT_REGION) == null) {
+        && credentialMap.get(StorageAccessProperty.CLIENT_REGION) == null) {
       throw new IllegalArgumentException(
           String.format(
               "AWS region must be set when using partition %s", storageConfig.getAwsPartition()));

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureCredentialsStorageIntegration.java

@@ -48,7 +48,7 @@
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
-import org.apache.polaris.core.storage.PolarisCredentialProperty;
+import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -70,14 +70,14 @@ public AzureCredentialsStorageIntegration() {
   }
 
   @Override
-  public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
+  public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
       @Nonnull PolarisDiagnostics diagnostics,
       @Nonnull AzureStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {
-    EnumMap<PolarisCredentialProperty, String> credentialMap =
-        new EnumMap<>(PolarisCredentialProperty.class);
+    EnumMap<StorageAccessProperty, String> credentialMap =
+        new EnumMap<>(StorageAccessProperty.class);
     String loc =
         !allowedWriteLocations.isEmpty()
             ? allowedWriteLocations.stream().findAny().orElse(null)
@@ -170,10 +170,10 @@ public EnumMap<PolarisCredentialProperty, String> getSubscopedCreds(
       throw new RuntimeException(
           String.format("Endpoint %s not supported", location.getEndpoint()));
     }
-    credentialMap.put(PolarisCredentialProperty.AZURE_SAS_TOKEN, sasToken);
-    credentialMap.put(PolarisCredentialProperty.AZURE_ACCOUNT_HOST, storageDnsName);
+    credentialMap.put(StorageAccessProperty.AZURE_SAS_TOKEN, sasToken);
+    credentialMap.put(StorageAccessProperty.AZURE_ACCOUNT_HOST, storageDnsName);
     credentialMap.put(
-        PolarisCredentialProperty.EXPIRATION_TIME,
+        StorageAccessProperty.EXPIRATION_TIME,
         String.valueOf(sanitizedEndTime.toInstant().toEpochMilli()));
     return credentialMap;
   }

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

@@ -23,6 +23,7 @@
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.annotations.VisibleForTesting;
 import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
@@ -35,6 +36,7 @@
 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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -99,7 +101,7 @@ private static long maxCacheDurationMs() {
    * @param allowedWriteLocations a set of allowed to write locations.
    * @return the a map of string containing the scoped creds information
    */
-  public Map<String, String> getOrGenerateSubScopeCreds(
+  public AccessConfig getOrGenerateSubScopeCreds(
       @Nonnull PolarisCredentialVendor credentialVendor,
       @Nonnull PolarisCallContext callCtx,
       @Nonnull PolarisEntity polarisEntity,
@@ -142,13 +144,19 @@ public Map<String, String> getOrGenerateSubScopeCreds(
               "Failed to get subscoped credentials: %s",
               scopedCredentialsResult.getExtraInformation());
         };
-    return cache.get(key, loader).convertToMapOfString();
+    return cache.get(key, loader).toAccessConfig();
   }
 
-  public Map<String, String> getIfPresent(StorageCredentialCacheKey key) {
+  @VisibleForTesting
+  @Nullable
+  Map<String, String> getIfPresent(StorageCredentialCacheKey key) {
+    return getAccessConfig(key).map(AccessConfig::credentials).orElse(null);
+  }
+
+  @VisibleForTesting
+  Optional<AccessConfig> getAccessConfig(StorageCredentialCacheKey key) {
     return Optional.ofNullable(cache.getIfPresent(key))
-        .map(StorageCredentialCacheEntry::convertToMapOfString)
-        .orElse(null);
+        .map(StorageCredentialCacheEntry::toAccessConfig);
   }
 
   private boolean isTypeSupported(PolarisEntityType type) {