add refresh credentials property to loadTableResult

CHANGELOG.md

@@ -81,6 +81,11 @@ at locations that better optimize for object storage.
 
 - Introduced bootstrap command options to specify custom schema files for database initialization.
 
+- Added refresh credentials endpoint configuration to LoadTableResponse for AWS, Azure, and GCP. Enabling
+automatic storage credential refresh per table on the client side. Java client version >= 1.8.0 is required.
+The endpoint path is always returned when using vended credentials, but clients must enable the 
+refresh-credentials flag for the desired storage provider.
+
 ### Changes
 
 - Polaris Management API clients must be prepared to deal with new attributes in `AwsStorageConfigInfo` objects.

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

@@ -1600,7 +1600,8 @@ private void revokeGrantRecord(
       PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
     BasePersistence ms = callCtx.getMetaStore();
@@ -1642,7 +1643,8 @@ private void revokeGrantRecord(
               callCtx.getRealmConfig(),
               allowListOperation,
               allowedReadLocations,
-              allowedWriteLocations);
+              allowedWriteLocations,
+              refreshCredentialsEndpoint);
       return new ScopedCredentialsResult(accessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

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

@@ -327,15 +327,17 @@ public ScopedCredentialsResult getSubscopedCredsForEntity(
       PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     return delegate.getSubscopedCredsForEntity(
         callCtx,
         catalogId,
         entityId,
         entityType,
         allowListOperation,
         allowedReadLocations,
-        allowedWriteLocations);
+        allowedWriteLocations,
+        refreshCredentialsEndpoint);
   }
 
   @Override

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

@@ -2022,7 +2022,8 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
       PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
 
     // get meta store session we should be using
     TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
@@ -2055,7 +2056,8 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
               callCtx.getRealmConfig(),
               allowListOperation,
               allowedReadLocations,
-              allowedWriteLocations);
+              allowedWriteLocations,
+              refreshCredentialsEndpoint);
       return new ScopedCredentialsResult(accessConfig);
     } catch (Exception ex) {
       return new ScopedCredentialsResult(

polaris-core/src/main/java/org/apache/polaris/core/rest/PolarisResourcePaths.java

@@ -57,6 +57,17 @@ public String genericTables(Namespace ns) {
         "polaris", "v1", prefix, "namespaces", RESTUtil.encodeNamespace(ns), "generic-tables");
   }
 
+  public String credentialsPath(TableIdentifier ident) {
+    return SLASH.join(
+        "v1",
+        prefix,
+        "namespaces",
+        RESTUtil.encodeNamespace(ident.namespace()),
+        "tables",
+        RESTUtil.encodeString(ident.name()),
+        "credentials");
+  }
+
   public String genericTable(TableIdentifier ident) {
     return SLASH.join(
         "polaris",

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.storage;
 
 import jakarta.annotation.Nonnull;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.PolarisEntityType;
@@ -37,6 +38,10 @@ public interface PolarisCredentialVendor {
    *     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
@@ -47,5 +52,6 @@ ScopedCredentialsResult getSubscopedCredsForEntity(
       PolarisEntityType entityType,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations);
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint);
 }

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

@@ -20,6 +20,7 @@
 
 import jakarta.annotation.Nonnull;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.config.RealmConfig;
 
@@ -55,13 +56,18 @@ public String getStorageIdentifierOrId() {
    *     locations
    * @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 including the scoped credentials
    */
   public abstract AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations);
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint);
 
   /**
    * Validate access for the provided operation actions and locations.

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

@@ -18,6 +18,9 @@
  */
 package org.apache.polaris.core.storage;
 
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.gcp.GCPProperties;
+
 /**
  * A subset of Iceberg catalog properties recognized by Polaris.
  *
@@ -39,6 +42,12 @@ public enum StorageAccessProperty {
       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"),
+  AWS_REFRESH_CREDENTIALS_ENDPOINT(
+      String.class,
+      AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT,
+      "the endpoint to load vended credentials for a table from the catalog",
+      false,
+      false),
 
   GCS_ACCESS_TOKEN(String.class, "gcs.oauth2.token", "the gcs scoped access token"),
   GCS_ACCESS_TOKEN_EXPIRES_AT(
@@ -47,11 +56,23 @@ public enum StorageAccessProperty {
       "the time the gcs access token expires, in milliseconds",
       true,
       true),
+  GCS_REFRESH_CREDENTIALS_ENDPOINT(
+      String.class,
+      GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT,
+      "the endpoint to load vended credentials for a table from the catalog",
+      false,
+      false),
 
   // Currently not using ACCESS TOKEN as the ResolvingFileIO is using ADLSFileIO for azure case and
   // it expects for SAS
   AZURE_ACCESS_TOKEN(String.class, "", "the azure scoped access token"),
   AZURE_SAS_TOKEN(String.class, "adls.sas-token.", "an azure shared access signature token"),
+  AZURE_REFRESH_CREDENTIALS_ENDPOINT(
+      String.class,
+      "adls.refresh-credentials-endpoint",
+      "the endpoint to load vended credentials for a table from the catalog",
+      false,
+      false),
   EXPIRATION_TIME(
       Long.class,
       "expiration-time",

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

@@ -74,7 +74,8 @@ public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     int storageCredentialDurationSeconds =
         realmConfig.getConfig(STORAGE_CREDENTIAL_DURATION_SECONDS);
     AwsStorageConfigurationInfo storageConfig = config();
@@ -120,6 +121,11 @@ public AccessConfig getSubscopedCreds(
       accessConfig.put(StorageAccessProperty.CLIENT_REGION, region);
     }
 
+    refreshCredentialsEndpoint.ifPresent(
+        endpoint -> {
+          accessConfig.put(StorageAccessProperty.AWS_REFRESH_CREDENTIALS_ENDPOINT, endpoint);
+        });
+
     URI endpointUri = storageConfig.getEndpointUri();
     if (endpointUri != null) {
       accessConfig.put(StorageAccessProperty.AWS_ENDPOINT, endpointUri.toString());

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

@@ -46,6 +46,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.HashSet;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.storage.AccessConfig;
@@ -76,7 +77,8 @@ public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     String loc =
         !allowedWriteLocations.isEmpty()
             ? allowedWriteLocations.stream().findAny().orElse(null)
@@ -169,15 +171,24 @@ public AccessConfig getSubscopedCreds(
           String.format("Endpoint %s not supported", location.getEndpoint()));
     }
 
-    return toAccessConfig(sasToken, storageDnsName, sanitizedEndTime.toInstant());
+    return toAccessConfig(
+        sasToken, storageDnsName, sanitizedEndTime.toInstant(), refreshCredentialsEndpoint);
   }
 
   @VisibleForTesting
-  static AccessConfig toAccessConfig(String sasToken, String storageDnsName, Instant expiresAt) {
+  static AccessConfig toAccessConfig(
+      String sasToken,
+      String storageDnsName,
+      Instant expiresAt,
+      Optional<String> refreshCredentialsEndpoint) {
     AccessConfig.Builder accessConfig = AccessConfig.builder();
     handleAzureCredential(accessConfig, sasToken, storageDnsName);
     accessConfig.put(
         StorageAccessProperty.EXPIRATION_TIME, String.valueOf(expiresAt.toEpochMilli()));
+    refreshCredentialsEndpoint.ifPresent(
+        endpoint -> {
+          accessConfig.put(StorageAccessProperty.AZURE_REFRESH_CREDENTIALS_ENDPOINT, endpoint);
+        });
     return accessConfig.build();
   }
 

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

@@ -105,7 +105,8 @@ public AccessConfig getOrGenerateSubScopeCreds(
       @Nonnull PolarisEntity polarisEntity,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     if (!isTypeSupported(polarisEntity.getType())) {
       callCtx
           .getDiagServices()
@@ -117,7 +118,8 @@ public AccessConfig getOrGenerateSubScopeCreds(
             polarisEntity,
             allowListOperation,
             allowedReadLocations,
-            allowedWriteLocations);
+            allowedWriteLocations,
+            refreshCredentialsEndpoint);
     LOGGER.atDebug().addKeyValue("key", key).log("subscopedCredsCache");
     Function<StorageCredentialCacheKey, StorageCredentialCacheEntry> loader =
         k -> {
@@ -130,7 +132,8 @@ public AccessConfig getOrGenerateSubScopeCreds(
                   polarisEntity.getType(),
                   k.allowedListAction(),
                   k.allowedReadLocations(),
-                  k.allowedWriteLocations());
+                  k.allowedWriteLocations(),
+                  k.refreshCredentialsEndpoint());
           if (scopedCredentialsResult.isSuccess()) {
             long maxCacheDurationMs = maxCacheDurationMs(callCtx.getRealmConfig());
             return new StorageCredentialCacheEntry(

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.storage.cache;
 
 import jakarta.annotation.Nullable;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -47,12 +48,16 @@ public interface StorageCredentialCacheKey {
   @Value.Parameter(order = 6)
   Set<String> allowedWriteLocations();
 
+  @Value.Parameter(order = 7)
+  Optional<String> refreshCredentialsEndpoint();
+
   static StorageCredentialCacheKey of(
       String realmId,
       PolarisEntity entity,
       boolean allowedListAction,
       Set<String> allowedReadLocations,
-      Set<String> allowedWriteLocations) {
+      Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     String storageConfigSerializedStr =
         entity
             .getInternalPropertiesAsMap()
@@ -63,6 +68,7 @@ static StorageCredentialCacheKey of(
         storageConfigSerializedStr,
         allowedListAction,
         allowedReadLocations,
-        allowedWriteLocations);
+        allowedWriteLocations,
+        refreshCredentialsEndpoint);
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java

@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
 import org.apache.polaris.core.config.RealmConfig;
@@ -75,7 +76,8 @@ public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
+      @Nonnull Set<String> allowedWriteLocations,
+      Optional<String> refreshCredentialsEndpoint) {
     try {
       sourceCredentials.refresh();
     } catch (IOException e) {
@@ -112,6 +114,12 @@ public AccessConfig getSubscopedCreds(
     accessConfig.put(
         StorageAccessProperty.GCS_ACCESS_TOKEN_EXPIRES_AT,
         String.valueOf(token.getExpirationTime().getTime()));
+
+    refreshCredentialsEndpoint.ifPresent(
+        endpoint -> {
+          accessConfig.put(StorageAccessProperty.GCS_REFRESH_CREDENTIALS_ENDPOINT, endpoint);
+        });
+
     return accessConfig.build();
   }
 

polaris-core/src/test/java/org/apache/polaris/core/storage/InMemoryStorageIntegrationTest.java

@@ -21,6 +21,7 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.config.RealmConfig;
@@ -197,7 +198,8 @@ public AccessConfig getSubscopedCreds(
         @Nonnull RealmConfig realmConfig,
         boolean allowListOperation,
         @Nonnull Set<String> allowedReadLocations,
-        @Nonnull Set<String> allowedWriteLocations) {
+        @Nonnull Set<String> allowedWriteLocations,
+        Optional<String> refreshCredentialsEndpoint) {
       return null;
     }
   }