Hide entity resolution from service implementations

Currently all service implementations, both the Polaris admin service and the (Iceberg REST) catalog service need knowledge about the implementation details about _how exactly_ entities and their instances are managed. This is rather a persistence implementation concern, not a service implementation concern. Services should tell, if necessary, tell a properly abstracted builder which entities it will need.

This change abstracts the entity resolution via interfaces and concrete implementation classes, eliminating the hard dependency of services to concrete persistence specific implementations.

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

@@ -27,11 +27,11 @@ public enum PolarisEntitySubType {
   // ANY_SUBTYPE is not stored but is used to indicate that any subtype entities should be
   // returned, for example when doing a list operation or checking if a table like object of
   // name X exists
-  ANY_SUBTYPE(-1, null),
+  ANY_SUBTYPE(-1, null, "Table of view"),
   // the NULL value is used when an entity has no subtype, i.e. NOT_APPLICABLE really
-  NULL_SUBTYPE(0, null),
-  TABLE(2, PolarisEntityType.TABLE_LIKE),
-  VIEW(3, PolarisEntityType.TABLE_LIKE);
+  NULL_SUBTYPE(0, null, "(null)"),
+  TABLE(2, PolarisEntityType.TABLE_LIKE, "Table"),
+  VIEW(3, PolarisEntityType.TABLE_LIKE, "View");
 
   // to efficiently map the code of a subtype to its corresponding subtype enum, use a reverse
   // array which is initialized below
@@ -63,10 +63,13 @@ public enum PolarisEntitySubType {
   // parent type for this entity
   private final PolarisEntityType parentType;
 
-  PolarisEntitySubType(int code, PolarisEntityType parentType) {
+  private final String readableName;
+
+  PolarisEntitySubType(int code, PolarisEntityType parentType, String readableName) {
     // remember the id of this entity
     this.code = code;
     this.parentType = parentType;
+    this.readableName = readableName;
   }
 
   /**
@@ -111,4 +114,8 @@ public PolarisEntityType getParentType() {
 
     return null;
   }
+
+  public String readableName() {
+    return readableName;
+  }
 }

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

@@ -24,17 +24,17 @@
 
 /** Types of entities with their id */
 public enum PolarisEntityType {
-  NULL_TYPE(0, null, false, false),
-  ROOT(1, null, false, false),
-  PRINCIPAL(2, ROOT, true, false),
-  PRINCIPAL_ROLE(3, ROOT, true, false),
-  CATALOG(4, ROOT, false, false),
-  CATALOG_ROLE(5, CATALOG, true, false),
-  NAMESPACE(6, CATALOG, false, true),
+  NULL_TYPE(0, null, false, false, "(null)"),
+  ROOT(1, null, false, false, "Root"),
+  PRINCIPAL(2, ROOT, true, false, "Principal"),
+  PRINCIPAL_ROLE(3, ROOT, true, false, "Principal role"),
+  CATALOG(4, ROOT, false, false, "Catalog"),
+  CATALOG_ROLE(5, CATALOG, true, false, "Catalog role"),
+  NAMESPACE(6, CATALOG, false, true, "Namespace"),
   // generic table is either a view or a real table
-  TABLE_LIKE(7, NAMESPACE, false, false),
-  TASK(8, ROOT, false, false),
-  FILE(9, TABLE_LIKE, false, false);
+  TABLE_LIKE(7, NAMESPACE, false, false, "Table/view"),
+  TASK(8, ROOT, false, false, "Task"),
+  FILE(9, TABLE_LIKE, false, false, "File");
 
   // to efficiently map a code to its corresponding entity type, use a reverse array which
   // is initialized below
@@ -70,13 +70,20 @@ public enum PolarisEntityType {
 
   // parent entity type, null for an ACCOUNT
   private final PolarisEntityType parentType;
-
-  PolarisEntityType(int id, PolarisEntityType parentType, boolean isGrantee, boolean sefRef) {
+  private final String readableName;
+
+  PolarisEntityType(
+      int id,
+      PolarisEntityType parentType,
+      boolean isGrantee,
+      boolean sefRef,
+      String readableName) {
     // remember the id of this entity
     this.code = id;
     this.isGrantee = isGrantee;
     this.parentType = parentType;
     this.parentSelfReference = sefRef;
+    this.readableName = readableName;
   }
 
   /**
@@ -132,4 +139,8 @@ public boolean isTopLevel() {
   public PolarisEntityType getParentType() {
     return this.parentType;
   }
+
+  public String readableName() {
+    return readableName;
+  }
 }

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

@@ -30,7 +30,6 @@
 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.storage.cache.StorageCredentialCache;
 
@@ -75,16 +74,17 @@ public Resolver prepareResolver(
         referenceCatalogName);
   }
 
-  public PolarisResolutionManifest prepareResolutionManifest(
+  public ResolutionManifestBuilder prepareResolutionManifest(
       @Nonnull CallContext callContext,
       @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
       @Nullable String referenceCatalogName) {
-    PolarisResolutionManifest manifest =
-        new PolarisResolutionManifest(
-            callContext, this, authenticatedPrincipal, referenceCatalogName);
-    manifest.setSimulatedResolvedRootContainerEntity(
-        getSimulatedResolvedRootContainerEntity(callContext));
-    return manifest;
+    return this.metaStoreManager
+        .newResolutionManifestBuilder(
+            callContext,
+            authenticatedPrincipal,
+            () -> prepareResolver(callContext, authenticatedPrincipal, referenceCatalogName),
+            referenceCatalogName)
+        .withRootContainerEntity(getSimulatedResolvedRootContainerEntity(callContext));
   }
 
   /**

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

@@ -25,9 +25,12 @@
 import jakarta.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
 import org.apache.polaris.core.auth.PolarisGrantManager;
 import org.apache.polaris.core.auth.PolarisSecretsManager;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityActiveRecord;
@@ -36,6 +39,7 @@
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
 import org.apache.polaris.core.persistence.cache.PolarisRemoteCache;
+import org.apache.polaris.core.persistence.resolver.Resolver;
 import org.apache.polaris.core.storage.PolarisCredentialVendor;
 
 /**
@@ -72,6 +76,13 @@ public interface PolarisMetaStoreManager
   @Nonnull
   BaseResult purge(@Nonnull PolarisCallContext callCtx);
 
+  @Nonnull
+  ResolutionManifestBuilder newResolutionManifestBuilder(
+      @Nonnull CallContext callContext,
+      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull Supplier<Resolver> resolverSupplier,
+      @Nullable String referenceCatalogName);
+
   /** the return for an entity lookup call */
   class EntityResult extends BaseResult {
 

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

@@ -34,8 +34,14 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.AsyncTaskType;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
@@ -51,6 +57,8 @@
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
 import org.apache.polaris.core.entity.PolarisPrivilege;
 import org.apache.polaris.core.entity.PolarisTaskConstants;
+import org.apache.polaris.core.persistence.resolver.Resolver;
+import org.apache.polaris.core.persistence.resolver.ResolverPath;
 import org.apache.polaris.core.storage.PolarisCredentialProperty;
 import org.apache.polaris.core.storage.PolarisStorageActions;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -1006,6 +1014,107 @@ public Map<String, String> deserializeProperties(PolarisCallContext callCtx, Str
         : new PrincipalSecretsResult(secrets);
   }
 
+  @Override
+  @Nonnull
+  public ResolutionManifestBuilder newResolutionManifestBuilder(
+      @Nonnull CallContext callContext,
+      @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
+      @Nonnull Supplier<Resolver> resolverSupplier,
+      @Nullable String referenceCatalogName) {
+    return new ResolutionManifestBuilder() {
+      private final PolarisResolutionManifest manifest =
+          new PolarisResolutionManifest(
+              authenticatedPrincipal,
+              referenceCatalogName,
+              resolverSupplier,
+              callContext.getPolarisCallContext().getDiagServices());
+
+      @Override
+      public ResolutionManifestBuilder withRootContainerEntity(
+          ResolvedPolarisEntity simulatedResolvedRootContainerEntity) {
+        manifest.setSimulatedResolvedRootContainerEntity(simulatedResolvedRootContainerEntity);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addTopLevelName(
+          String topLevelEntityName, PolarisEntityType entityType, boolean optional) {
+        manifest.addTopLevelName(topLevelEntityName, entityType, optional);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addPath(ResolverPath resolverPath, String catalogRoleName) {
+        manifest.addPath(resolverPath, catalogRoleName);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addPath(ResolverPath resolverPath, Namespace namespace) {
+        manifest.addPath(resolverPath, namespace);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addPath(
+          ResolverPath resolverPath, TableIdentifier tableIdentifier) {
+        manifest.addPath(resolverPath, tableIdentifier);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addPassthroughPath(
+          ResolverPath resolverPath, TableIdentifier tableIdentifier) {
+        manifest.addPassthroughPath(resolverPath, tableIdentifier);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifestBuilder addPassthroughPath(
+          ResolverPath resolverPath, Namespace namespace) {
+        manifest.addPassthroughPath(resolverPath, namespace);
+        return this;
+      }
+
+      @Override
+      public ResolutionManifest buildResolved() {
+        return buildResolved(null);
+      }
+
+      @Override
+      public ResolutionManifest buildResolved(PolarisEntitySubType subType) {
+        var status = manifest.resolveAll();
+
+        switch (status.getStatus()) {
+          case SUCCESS:
+            return manifest;
+          case CALLER_PRINCIPAL_DOES_NOT_EXIST:
+            throw new NotFoundException("Caller principal does not exist");
+          case ENTITY_COULD_NOT_BE_RESOLVED:
+            throw new NotFoundException(
+                "%s does not exist: %s",
+                subType != null
+                        && subType.getParentType()
+                            == status.getFailedToResolvePath().getLastEntityType()
+                    ? subType.readableName()
+                    : status.getFailedToResolvedEntityType().readableName(),
+                status.getFailedToResolvedEntityName());
+          case PATH_COULD_NOT_BE_FULLY_RESOLVED:
+            throw new NotFoundException(
+                "%s %s not found",
+                subType != null
+                        && subType.getParentType()
+                            == status.getFailedToResolvePath().getLastEntityType()
+                    ? subType.readableName()
+                    : status.getFailedToResolvePath().getLastEntityType().readableName(),
+                status.getFailedToResolvePath());
+          default:
+            throw new IllegalStateException("Unexpected status " + status);
+        }
+      }
+    };
+  }
+
   /** See {@link #} */
   private @Nullable PolarisPrincipalSecrets rotatePrincipalSecrets(
       @Nonnull PolarisCallContext callCtx,