Extract ResolverFactory from PolarisEntityManager

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

@@ -29,9 +29,8 @@
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisGrantRecord;
 import org.apache.polaris.core.entity.PolarisPrivilege;
-import org.apache.polaris.core.persistence.cache.EntityCache;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
-import org.apache.polaris.core.persistence.resolver.Resolver;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 
 /**
  * Wraps logic of handling name-caching and entity-caching against a concrete underlying entity
@@ -40,39 +39,28 @@
  */
 public class PolarisEntityManager {
   private final PolarisMetaStoreManager metaStoreManager;
-  private final EntityCache entityCache;
+  private final ResolverFactory resolverFactory;
 
   // Lazily instantiated only a single time per entity manager.
   private ResolvedPolarisEntity implicitResolvedRootContainerEntity = null;
 
   /**
    * @param metaStoreManager the metastore manager for the current realm
-   * @param entityCache the entity cache to use (it may be {@code null}).
+   * @param resolverFactory the resolver factory to use
    */
   public PolarisEntityManager(
-      @Nonnull PolarisMetaStoreManager metaStoreManager, @Nullable EntityCache entityCache) {
+      @Nonnull PolarisMetaStoreManager metaStoreManager, @Nonnull ResolverFactory resolverFactory) {
     this.metaStoreManager = metaStoreManager;
-    this.entityCache = entityCache;
-  }
-
-  public Resolver prepareResolver(
-      @Nonnull CallContext callContext,
-      @Nonnull SecurityContext securityContext,
-      @Nullable String referenceCatalogName) {
-    return new Resolver(
-        callContext.getPolarisCallContext(),
-        metaStoreManager,
-        securityContext,
-        entityCache,
-        referenceCatalogName);
+    this.resolverFactory = resolverFactory;
   }
 
   public PolarisResolutionManifest prepareResolutionManifest(
       @Nonnull CallContext callContext,
       @Nonnull SecurityContext securityContext,
       @Nullable String referenceCatalogName) {
     PolarisResolutionManifest manifest =
-        new PolarisResolutionManifest(callContext, this, securityContext, referenceCatalogName);
+        new PolarisResolutionManifest(
+            callContext, resolverFactory, securityContext, referenceCatalogName);
     manifest.setSimulatedResolvedRootContainerEntity(
         getSimulatedResolvedRootContainerEntity(callContext));
     return manifest;

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/PolarisResolutionManifest.java

@@ -34,7 +34,6 @@
 import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
-import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
 import org.slf4j.Logger;
@@ -51,7 +50,7 @@
 public class PolarisResolutionManifest implements PolarisResolutionManifestCatalogView {
   private static final Logger LOGGER = LoggerFactory.getLogger(PolarisResolutionManifest.class);
 
-  private final PolarisEntityManager entityManager;
+  private final ResolverFactory resolverFactory;
   private final CallContext callContext;
   private final SecurityContext securityContext;
   private final String catalogName;
@@ -78,13 +77,14 @@ public class PolarisResolutionManifest implements PolarisResolutionManifestCatal
 
   public PolarisResolutionManifest(
       CallContext callContext,
-      PolarisEntityManager entityManager,
+      ResolverFactory resolverFactory,
       SecurityContext securityContext,
       String catalogName) {
-    this.entityManager = entityManager;
     this.callContext = callContext;
+    this.resolverFactory = resolverFactory;
     this.catalogName = catalogName;
-    this.primaryResolver = entityManager.prepareResolver(callContext, securityContext, catalogName);
+    this.primaryResolver =
+        resolverFactory.createResolver(callContext, securityContext, catalogName);
     this.diagnostics = callContext.getPolarisCallContext().getDiagServices();
     this.diagnostics.checkNotNull(securityContext, "null_security_context_for_resolution_manifest");
     this.securityContext = securityContext;
@@ -193,7 +193,7 @@ public PolarisResolvedPathWrapper getPassthroughResolvedPath(Object key) {
 
     // Run a single-use Resolver for this path.
     Resolver passthroughResolver =
-        entityManager.prepareResolver(callContext, securityContext, catalogName);
+        resolverFactory.createResolver(callContext, securityContext, catalogName);
     passthroughResolver.addPath(requestedPath);
     ResolverStatus status = passthroughResolver.resolveAll();
 

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/ResolverFactory.java

@@ -0,0 +1,32 @@
+/*
+ * 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.persistence.resolver;
+
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import jakarta.ws.rs.core.SecurityContext;
+import org.apache.polaris.core.context.CallContext;
+
+public interface ResolverFactory {
+  Resolver createResolver(
+      @Nonnull CallContext callContext,
+      @Nonnull SecurityContext securityContext,
+      @Nullable String referenceCatalogName);
+}

runtime/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusProducers.java

@@ -49,6 +49,9 @@
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet;
+import org.apache.polaris.core.persistence.cache.EntityCache;
+import org.apache.polaris.core.persistence.resolver.Resolver;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
@@ -106,6 +109,24 @@ public StorageCredentialCache storageCredentialCache(
     return new StorageCredentialCache(storageCredentialCacheConfig);
   }
 
+  @Produces
+  @ApplicationScoped
+  public ResolverFactory resolverFactory(
+      MetaStoreManagerFactory metaStoreManagerFactory,
+      PolarisMetaStoreManager polarisMetaStoreManager) {
+    return (callContext, securityContext, referenceCatalogName) -> {
+      EntityCache entityCache =
+          metaStoreManagerFactory.getOrCreateEntityCache(
+              callContext.getRealmContext(), callContext.getRealmConfig());
+      return new Resolver(
+          callContext.getPolarisCallContext(),
+          polarisMetaStoreManager,
+          securityContext,
+          entityCache,
+          referenceCatalogName);
+    };
+  }
+
   @Produces
   @ApplicationScoped
   public PolarisAuthorizer polarisAuthorizer() {

runtime/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java

@@ -72,6 +72,7 @@
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.policy.PredefinedPolicyTypes;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
@@ -193,6 +194,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject protected CatalogHandlerUtils catalogHandlerUtils;
   @Inject protected PolarisConfigurationStore configurationStore;
   @Inject protected StorageCredentialCache storageCredentialCache;
+  @Inject protected ResolverFactory resolverFactory;
 
   protected IcebergCatalog baseCatalog;
   protected PolarisGenericTableCatalog genericTableCatalog;
@@ -468,7 +470,7 @@ private void initBaseCatalog() {
     this.baseCatalog =
         new IcebergCatalog(
             storageCredentialCache,
-            entityManager,
+            resolverFactory,
             metaStoreManager,
             callContext,
             passthroughView,
@@ -498,14 +500,14 @@ public TestPolarisCallContextCatalogFactory() {
     @Inject
     public TestPolarisCallContextCatalogFactory(
         StorageCredentialCache storageCredentialCache,
-        RealmEntityManagerFactory entityManagerFactory,
+        ResolverFactory resolverFactory,
         MetaStoreManagerFactory metaStoreManagerFactory,
         TaskExecutor taskExecutor,
         FileIOFactory fileIOFactory,
         PolarisEventListener polarisEventListener) {
       super(
           storageCredentialCache,
-          entityManagerFactory,
+          resolverFactory,
           metaStoreManagerFactory,
           taskExecutor,
           fileIOFactory,

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogTest.java

@@ -116,6 +116,8 @@
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.pagination.Page;
 import org.apache.polaris.core.persistence.pagination.PageToken;
+import org.apache.polaris.core.persistence.resolver.Resolver;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.PolarisStorageActions;
@@ -235,6 +237,7 @@ public Map<String, String> getConfigOverrides() {
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
   private PolarisAdminService adminService;
+  private ResolverFactory resolverFactory;
   private PolarisEntityManager entityManager;
   private FileIOFactory fileIOFactory;
   private InMemoryFileIO fileIO;
@@ -280,7 +283,17 @@ public void before(TestInfo testInfo) {
             Clock.systemDefaultZone());
 
     EntityCache entityCache = createEntityCache(polarisContext.getRealmConfig(), metaStoreManager);
-    entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
+    resolverFactory =
+        (callContext, securityContext, referenceCatalogName) ->
+            new Resolver(
+                callContext.getPolarisCallContext(),
+                metaStoreManager,
+                securityContext,
+                entityCache,
+                referenceCatalogName);
+    QuarkusMock.installMockForType(resolverFactory, ResolverFactory.class);
+
+    entityManager = new PolarisEntityManager(metaStoreManager, resolverFactory);
 
     // LocalPolarisMetaStoreManagerFactory.bootstrapServiceAndCreatePolarisPrincipalForRealm sets
     // the CallContext.setCurrentContext() but never clears it, whereas the NoSQL one resets it.
@@ -437,7 +450,7 @@ protected IcebergCatalog newIcebergCatalog(
     TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
     return new IcebergCatalog(
         storageCredentialCache,
-        entityManager,
+        resolverFactory,
         metaStoreManager,
         polarisContext,
         passthroughView,

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogViewTest.java

@@ -55,7 +55,7 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
-import org.apache.polaris.core.persistence.cache.EntityCache;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
@@ -117,6 +117,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisDiagnostics diagServices;
   @Inject PolarisEventListener polarisEventListener;
+  @Inject ResolverFactory resolverFactory;
 
   private IcebergCatalog catalog;
 
@@ -166,10 +167,8 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
-    EntityCache entityCache =
-        metaStoreManagerFactory.getOrCreateEntityCache(
-            polarisContext.getRealmContext(), polarisContext.getRealmConfig());
-    PolarisEntityManager entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
+    PolarisEntityManager entityManager =
+        new PolarisEntityManager(metaStoreManager, resolverFactory);
 
     CallContext.setCurrentContext(polarisContext);
 
@@ -230,7 +229,7 @@ public void before(TestInfo testInfo) {
     this.catalog =
         new IcebergCatalog(
             storageCredentialCache,
-            entityManager,
+            resolverFactory,
             metaStoreManager,
             polarisContext,
             passthroughView,

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractPolarisGenericTableCatalogTest.java

@@ -56,7 +56,7 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
-import org.apache.polaris.core.persistence.cache.EntityCache;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
@@ -104,6 +104,7 @@ public abstract class AbstractPolarisGenericTableCatalogTest {
   @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject PolarisDiagnostics diagServices;
+  @Inject ResolverFactory resolverFactory;
 
   private PolarisGenericTableCatalog genericTableCatalog;
   private IcebergCatalog icebergCatalog;
@@ -157,10 +158,7 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
-    EntityCache entityCache =
-        metaStoreManagerFactory.getOrCreateEntityCache(
-            realmContext, polarisContext.getRealmConfig());
-    entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
+    entityManager = new PolarisEntityManager(metaStoreManager, resolverFactory);
 
     PrincipalEntity rootEntity =
         new PrincipalEntity(
@@ -249,7 +247,7 @@ public void before(TestInfo testInfo) {
     this.icebergCatalog =
         new IcebergCatalog(
             storageCredentialCache,
-            entityManager,
+            resolverFactory,
             metaStoreManager,
             polarisContext,
             passthroughView,

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractPolicyCatalogTest.java

@@ -64,7 +64,7 @@
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.PolicyMappingAlreadyExistsException;
-import org.apache.polaris.core.persistence.cache.EntityCache;
+import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.core.policy.PredefinedPolicyTypes;
 import org.apache.polaris.core.policy.exceptions.NoSuchPolicyException;
 import org.apache.polaris.core.policy.exceptions.PolicyInUseException;
@@ -131,6 +131,7 @@ public abstract class AbstractPolicyCatalogTest {
   @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject PolarisDiagnostics diagServices;
+  @Inject ResolverFactory resolverFactory;
 
   private PolicyCatalog policyCatalog;
   private IcebergCatalog icebergCatalog;
@@ -180,10 +181,7 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
-    EntityCache entityCache =
-        metaStoreManagerFactory.getOrCreateEntityCache(
-            realmContext, polarisContext.getRealmConfig());
-    entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
+    entityManager = new PolarisEntityManager(metaStoreManager, resolverFactory);
 
     callContext = polarisContext;
 
@@ -270,7 +268,7 @@ public void before(TestInfo testInfo) {
     this.icebergCatalog =
         new IcebergCatalog(
             storageCredentialCache,
-            entityManager,
+            resolverFactory,
             metaStoreManager,
             callContext,
             passthroughView,

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogHandlerAuthzTest.java

@@ -1789,7 +1789,7 @@ public void testSendNotificationSufficientPrivileges() {
     PolarisCallContextCatalogFactory factory =
         new PolarisCallContextCatalogFactory(
             storageCredentialCache,
-            realmEntityManagerFactory,
+            resolverFactory,
             managerFactory,
             Mockito.mock(),
             new DefaultFileIOFactory(storageCredentialCache, managerFactory),