WIP Cred vending isolation

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/EclipseLinkPolarisMetaStoreManagerFactory.java

@@ -31,7 +31,6 @@
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet;
 import org.apache.polaris.core.persistence.transactional.TransactionalPersistence;
-import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 
 /**
  * The implementation of Configuration interface for configuring the {@link PolarisMetaStoreManager}
@@ -43,17 +42,20 @@
 public class EclipseLinkPolarisMetaStoreManagerFactory
     extends LocalPolarisMetaStoreManagerFactory<PolarisEclipseLinkStore> {
 
-  @Inject EclipseLinkConfiguration eclipseLinkConfiguration;
-  @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
+  private final EclipseLinkConfiguration eclipseLinkConfiguration;
 
   @SuppressWarnings("unused") // Required by CDI
   protected EclipseLinkPolarisMetaStoreManagerFactory() {
-    this(null, null);
+    this(null, null, null);
   }
 
   @Inject
-  protected EclipseLinkPolarisMetaStoreManagerFactory(Clock clock, PolarisDiagnostics diagnostics) {
+  protected EclipseLinkPolarisMetaStoreManagerFactory(
+      Clock clock,
+      PolarisDiagnostics diagnostics,
+      EclipseLinkConfiguration eclipseLinkConfiguration) {
     super(clock, diagnostics);
+    this.eclipseLinkConfiguration = eclipseLinkConfiguration;
   }
 
   @Override
@@ -69,7 +71,6 @@ protected TransactionalPersistence createMetaStoreSession(
       @Nonnull PolarisDiagnostics diagnostics) {
     return new PolarisEclipseLinkMetaStoreSessionImpl(
         store,
-        storageIntegrationProvider,
         realmContext,
         configurationFile(),
         persistenceUnitName(),

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java

@@ -53,17 +53,13 @@
 import org.apache.polaris.core.entity.PolarisGrantRecord;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
 import org.apache.polaris.core.exceptions.AlreadyExistsException;
-import org.apache.polaris.core.persistence.BaseMetaStoreManager;
 import org.apache.polaris.core.persistence.PrincipalSecretsGenerator;
 import org.apache.polaris.core.persistence.RetryOnConcurrencyException;
 import org.apache.polaris.core.persistence.pagination.EntityIdToken;
 import org.apache.polaris.core.persistence.pagination.Page;
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.persistence.transactional.AbstractTransactionalPersistence;
 import org.apache.polaris.core.policy.PolarisPolicyMappingRecord;
-import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
-import org.apache.polaris.core.storage.PolarisStorageIntegration;
-import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 import org.apache.polaris.extension.persistence.impl.eclipselink.models.ModelEntity;
 import org.apache.polaris.extension.persistence.impl.eclipselink.models.ModelEntityActive;
 import org.apache.polaris.extension.persistence.impl.eclipselink.models.ModelEntityChangeTracking;
@@ -93,21 +89,18 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
   private final ThreadLocal<EntityManager> localSession = new ThreadLocal<>();
 
   private final PolarisEclipseLinkStore store;
-  private final PolarisStorageIntegrationProvider storageIntegrationProvider;
   private final PrincipalSecretsGenerator secretsGenerator;
 
   /**
    * Create a meta store session against provided realm. Each realm has its own database.
    *
    * @param store Backing store of EclipseLink implementation
-   * @param storageIntegrationProvider Storage integration provider
    * @param realmContext Realm context used to communicate with different database.
    * @param confFile Optional EclipseLink configuration file. Default to 'META-INF/persistence.xml'.
    * @param persistenceUnitName Optional persistence-unit name in confFile. Default to 'polaris'.
    */
   public PolarisEclipseLinkMetaStoreSessionImpl(
       @Nonnull PolarisEclipseLinkStore store,
-      @Nonnull PolarisStorageIntegrationProvider storageIntegrationProvider,
       @Nonnull RealmContext realmContext,
       @Nullable String confFile,
       @Nullable String persistenceUnitName,
@@ -121,7 +114,6 @@ public PolarisEclipseLinkMetaStoreSessionImpl(
     try (EntityManager session = emf.createEntityManager()) {
       this.store.initialize(session);
     }
-    this.storageIntegrationProvider = storageIntegrationProvider;
     this.secretsGenerator = secretsGenerator;
   }
 
@@ -276,16 +268,6 @@ public void writeToEntitiesInCurrentTxn(
     this.store.writeToEntities(localSession.get(), entity);
   }
 
-  /** {@inheritDoc} */
-  @Override
-  public <T extends PolarisStorageConfigurationInfo>
-      void persistStorageIntegrationIfNeededInCurrentTxn(
-          @Nonnull PolarisCallContext callContext,
-          @Nonnull PolarisBaseEntity entity,
-          @Nullable PolarisStorageIntegration<T> storageIntegration) {
-    // not implemented for eclipselink store
-  }
-
   /** {@inheritDoc} */
   @Override
   public void writeToEntitiesActiveInCurrentTxn(
@@ -661,28 +643,6 @@ public void deletePrincipalSecretsInCurrentTxn(
     this.store.deletePrincipalSecrets(localSession.get(), clientId);
   }
 
-  /** {@inheritDoc} */
-  @Override
-  public @Nullable <T extends PolarisStorageConfigurationInfo>
-      PolarisStorageIntegration<T> createStorageIntegrationInCurrentTxn(
-          @Nonnull PolarisCallContext callCtx,
-          long catalogId,
-          long entityId,
-          PolarisStorageConfigurationInfo polarisStorageConfigurationInfo) {
-    return storageIntegrationProvider.getStorageIntegrationForConfig(
-        polarisStorageConfigurationInfo);
-  }
-
-  /** {@inheritDoc} */
-  @Override
-  public @Nullable <T extends PolarisStorageConfigurationInfo>
-      PolarisStorageIntegration<T> loadPolarisStorageIntegrationInCurrentTxn(
-          @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
-    PolarisStorageConfigurationInfo storageConfig =
-        BaseMetaStoreManager.extractStorageConfiguration(callCtx.getDiagServices(), entity);
-    return storageIntegrationProvider.getStorageIntegrationForConfig(storageConfig);
-  }
-
   /** {@inheritDoc} */
   @Override
   public void writeToPolicyMappingRecordsInCurrentTxn(

persistence/eclipselink/src/test/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreManagerTest.java

@@ -48,7 +48,6 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
-import org.mockito.Mockito;
 
 /**
  * Integration test for EclipseLink based metastore implementation
@@ -86,7 +85,7 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
     RealmContext realmContext = () -> "realm";
     PolarisEclipseLinkMetaStoreSessionImpl session =
         new PolarisEclipseLinkMetaStoreSessionImpl(
-            store, Mockito.mock(), realmContext, null, "polaris", RANDOM_SECRETS);
+            store, realmContext, null, "polaris", RANDOM_SECRETS);
     TransactionalMetaStoreManagerImpl metaStoreManager =
         new TransactionalMetaStoreManagerImpl(clock);
     PolarisCallContext callCtx = new PolarisCallContext(realmContext, session, diagServices);
@@ -104,7 +103,7 @@ void testCreateStoreSession(String confFile, boolean success) {
     try {
       var session =
           new PolarisEclipseLinkMetaStoreSessionImpl(
-              store, Mockito.mock(), () -> "realm", confFile, "polaris", RANDOM_SECRETS);
+              store, () -> "realm", confFile, "polaris", RANDOM_SECRETS);
       assertNotNull(session);
       assertTrue(success);
     } catch (Exception e) {

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -47,7 +47,6 @@
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisGrantRecord;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
-import org.apache.polaris.core.persistence.BaseMetaStoreManager;
 import org.apache.polaris.core.persistence.BasePersistence;
 import org.apache.polaris.core.persistence.EntityAlreadyExistsException;
 import org.apache.polaris.core.persistence.IntegrationPersistence;
@@ -60,9 +59,6 @@
 import org.apache.polaris.core.policy.PolarisPolicyMappingRecord;
 import org.apache.polaris.core.policy.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyType;
-import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
-import org.apache.polaris.core.storage.PolarisStorageIntegration;
-import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 import org.apache.polaris.core.storage.StorageLocation;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelEntity;
 import org.apache.polaris.persistence.relational.jdbc.models.ModelGrantRecord;
@@ -78,7 +74,6 @@ public class JdbcBasePersistenceImpl implements BasePersistence, IntegrationPers
 
   private final DatasourceOperations datasourceOperations;
   private final PrincipalSecretsGenerator secretsGenerator;
-  private final PolarisStorageIntegrationProvider storageIntegrationProvider;
   private final String realmId;
   private final int schemaVersion;
 
@@ -88,12 +83,10 @@ public class JdbcBasePersistenceImpl implements BasePersistence, IntegrationPers
   public JdbcBasePersistenceImpl(
       DatasourceOperations databaseOperations,
       PrincipalSecretsGenerator secretsGenerator,
-      PolarisStorageIntegrationProvider storageIntegrationProvider,
       String realmId,
       int schemaVersion) {
     this.datasourceOperations = databaseOperations;
     this.secretsGenerator = secretsGenerator;
-    this.storageIntegrationProvider = storageIntegrationProvider;
     this.realmId = realmId;
     this.schemaVersion = schemaVersion;
   }
@@ -1103,34 +1096,6 @@ private List<PolarisPolicyMappingRecord> fetchPolicyMappingRecords(
     }
   }
 
-  @Nullable
-  @Override
-  public <T extends PolarisStorageConfigurationInfo>
-      PolarisStorageIntegration<T> createStorageIntegration(
-          @Nonnull PolarisCallContext callCtx,
-          long catalogId,
-          long entityId,
-          PolarisStorageConfigurationInfo polarisStorageConfigurationInfo) {
-    return storageIntegrationProvider.getStorageIntegrationForConfig(
-        polarisStorageConfigurationInfo);
-  }
-
-  @Override
-  public <T extends PolarisStorageConfigurationInfo> void persistStorageIntegrationIfNeeded(
-      @Nonnull PolarisCallContext callContext,
-      @Nonnull PolarisBaseEntity entity,
-      @Nullable PolarisStorageIntegration<T> storageIntegration) {}
-
-  @Nullable
-  @Override
-  public <T extends PolarisStorageConfigurationInfo>
-      PolarisStorageIntegration<T> loadPolarisStorageIntegration(
-          @Nonnull PolarisCallContext callContext, @Nonnull PolarisBaseEntity entity) {
-    PolarisStorageConfigurationInfo storageConfig =
-        BaseMetaStoreManager.extractStorageConfiguration(callContext.getDiagServices(), entity);
-    return storageIntegrationProvider.getStorageIntegrationForConfig(storageConfig);
-  }
-
   @FunctionalInterface
   private interface QueryAction {
     Integer apply(Connection connection, QueryGenerator.PreparedQuery query) throws SQLException;

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -105,7 +105,6 @@ private void initializeForRealm(
             new JdbcBasePersistenceImpl(
                 datasourceOperations,
                 secretsGenerator(realmId, rootCredentialsSet),
-                storageIntegrationProvider,
                 realmId,
                 schemaVersion));
 

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/AtomicMetastoreManagerWithJdbcBasePersistenceImplTest.java

@@ -32,7 +32,6 @@
 import org.apache.polaris.core.persistence.BasePolarisMetaStoreManagerTest;
 import org.apache.polaris.core.persistence.PolarisTestMetaStoreManager;
 import org.h2.jdbcx.JdbcConnectionPool;
-import org.mockito.Mockito;
 
 public class AtomicMetastoreManagerWithJdbcBasePersistenceImplTest
     extends BasePolarisMetaStoreManagerTest {
@@ -64,11 +63,7 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
     RealmContext realmContext = () -> "REALM";
     JdbcBasePersistenceImpl basePersistence =
         new JdbcBasePersistenceImpl(
-            datasourceOperations,
-            RANDOM_SECRETS,
-            Mockito.mock(),
-            realmContext.getRealmIdentifier(),
-            schemaVersion);
+            datasourceOperations, RANDOM_SECRETS, realmContext.getRealmIdentifier(), schemaVersion);
     AtomicOperationMetaStoreManager metaStoreManager = new AtomicOperationMetaStoreManager(clock);
     PolarisCallContext callCtx =
         new PolarisCallContext(realmContext, basePersistence, diagServices);

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -141,6 +141,7 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setEndpoint(awsConfig.getEndpoint())
             .setStsEndpoint(awsConfig.getStsEndpoint())
             .setPathStyleAccess(awsConfig.getPathStyleAccess())
+            .setEndpointInternal(awsConfig.getEndpointInternal())
             .build();
       }
       if (configInfo instanceof AzureStorageConfigurationInfo) {

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

@@ -64,16 +64,12 @@
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
 import org.apache.polaris.core.persistence.dao.entity.PrivilegeResult;
 import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
-import org.apache.polaris.core.persistence.dao.entity.ScopedCredentialsResult;
 import org.apache.polaris.core.persistence.pagination.Page;
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.policy.PolarisPolicyMappingRecord;
 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.AccessConfig;
-import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
-import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -433,26 +429,6 @@ private void revokeGrantRecord(
     // validate input
     callCtx.getDiagServices().checkNotNull(catalog, "unexpected_null_catalog");
 
-    Map<String, String> internalProp = getInternalPropertyMap(catalog);
-    String integrationIdentifierOrId =
-        internalProp.get(PolarisEntityConstants.getStorageIntegrationIdentifierPropertyName());
-    String storageConfigInfoStr =
-        internalProp.get(PolarisEntityConstants.getStorageConfigInfoPropertyName());
-    PolarisStorageIntegration<?> integration;
-    // storageConfigInfo's presence is needed to create a storage integration
-    // and the catalog should not have an internal property of storage identifier or id yet
-    if (storageConfigInfoStr != null && integrationIdentifierOrId == null) {
-      integration =
-          ((IntegrationPersistence) ms)
-              .createStorageIntegration(
-                  callCtx,
-                  catalog.getCatalogId(),
-                  catalog.getId(),
-                  PolarisStorageConfigurationInfo.deserialize(storageConfigInfoStr));
-    } else {
-      integration = null;
-    }
-
     // check if that catalog has already been created
     // This can be done safely as a separate atomic operation before trying to create the catalog
     // because same-id idempotent-retry collisions of this sort are necessarily sequential, so
@@ -489,7 +465,6 @@ private void revokeGrantRecord(
       // done, return the existing catalog
       return new CreateCatalogResult(refreshCatalog, catalogAdminRole);
     }
-    ((IntegrationPersistence) ms).persistStorageIntegrationIfNeeded(callCtx, catalog, integration);
 
     // now create and persist new catalog entity
     EntityResult lowLevelResult = this.persistNewEntity(callCtx, ms, catalog);
@@ -1568,65 +1543,6 @@ private void revokeGrantRecord(
     return EntitiesResult.fromPage(Page.fromItems(loadedTasks));
   }
 
-  /** {@inheritDoc} */
-  @Override
-  public @Nonnull ScopedCredentialsResult getSubscopedCredsForEntity(
-      @Nonnull PolarisCallContext callCtx,
-      long catalogId,
-      long entityId,
-      PolarisEntityType entityType,
-      boolean allowListOperation,
-      @Nonnull Set<String> allowedReadLocations,
-      @Nonnull Set<String> allowedWriteLocations) {
-
-    // get meta store session we should be using
-    BasePersistence ms = callCtx.getMetaStore();
-    callCtx
-        .getDiagServices()
-        .check(
-            !allowedReadLocations.isEmpty() || !allowedWriteLocations.isEmpty(),
-            "allowed_locations_to_subscope_is_required");
-
-    // reload the entity, error out if not found
-    EntityResult reloadedEntity = loadEntity(callCtx, catalogId, entityId, entityType);
-    if (reloadedEntity.getReturnStatus() != BaseResult.ReturnStatus.SUCCESS) {
-      return new ScopedCredentialsResult(
-          reloadedEntity.getReturnStatus(), reloadedEntity.getExtraInformation());
-    }
-
-    // TODO: Consider whether this independent lookup is safe for the model already or whether
-    // we need better atomicity semantics between the base entity and the embedded storage
-    // integration.
-
-    // get storage integration
-    PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
-        ((IntegrationPersistence) ms)
-            .loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
-
-    // cannot be null
-    callCtx
-        .getDiagServices()
-        .checkNotNull(
-            storageIntegration,
-            "storage_integration_not_exists",
-            "catalogId={}, entityId={}",
-            catalogId,
-            entityId);
-
-    try {
-      AccessConfig accessConfig =
-          storageIntegration.getSubscopedCreds(
-              callCtx.getRealmConfig(),
-              allowListOperation,
-              allowedReadLocations,
-              allowedWriteLocations);
-      return new ScopedCredentialsResult(accessConfig);
-    } catch (Exception ex) {
-      return new ScopedCredentialsResult(
-          BaseResult.ReturnStatus.SUBSCOPE_CREDS_ERROR, ex.getMessage());
-    }
-  }
-
   /**
    * Get the internal property map for an entity
    *