Rework PolarisMetaStoreManager.listEntities

the central `BasePersistence.listEntities` method supports pagination,
filtering loaded entities via a Predicate and transforming the loaded
entities to a more specific type.

`PolarisMetaStoreManager.listEntities` only exposed a limited subset of
that functionality i.e. the retrieval of a `ListEntitiesResult`.

almost all callers then had to post-process the lookup records of
`ListEntitiesResult` and called `PolarisMetaStoreManager.loadEntity`
on the individual items to transform and filter them.

this is bad for the following reasons:
- suboptimal performance as we run N+1 queries to basically load every
  entity twice from the persistence backend
- suffering from race-conditions when entities get dropped between the
  `listEntities` and `loadEntity` call
- a lot of repeated code in all the callers (of which only some are
  dealing with the race-condition by filtering out `null` values)

the fix is to simply let `PolarisMetaStoreManager.listEntities` offer
all capabilities of the underlying `BasePersistence.listEntities`
method.
as it turns out `ListEntitiesResult` and `PolarisEntity.NameAndId`
become completely obsolete.

note that we remove `testCatalogNotReturnedWhenDeletedAfterListBeforeGet`
from `ManagementServiceTest` because the simulated race-condition
scenario can not happen any longer.

Author: XN137

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

@@ -451,14 +451,7 @@ public List<EntityNameLookupRecord> lookupEntityActiveBatchInCurrentTxn(
         parentId,
         entityType,
         entityFilter,
-        entity ->
-            new EntityNameLookupRecord(
-                entity.getCatalogId(),
-                entity.getId(),
-                entity.getParentId(),
-                entity.getName(),
-                entity.getTypeCode(),
-                entity.getSubTypeCode()),
+        EntityNameLookupRecord::new,
         pageToken);
   }
 

polaris-core/src/main/java/org/apache/polaris/core/catalog/PolarisCatalogHelpers.java

@@ -24,6 +24,7 @@
 import java.util.stream.Collectors;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
 
 /**
@@ -61,8 +62,8 @@ public static Namespace getParentNamespace(Namespace namespace) {
     return Namespace.of(parentLevels);
   }
 
-  public static Namespace nameAndIdToNamespace(
-      List<PolarisEntity> catalogPath, PolarisEntity.NameAndId entity) {
+  public static Namespace entityToNamespace(
+      List<PolarisEntity> catalogPath, PolarisBaseEntity entity) {
     // Skip element 0 which is the catalog entity
     String[] fullName = new String[catalogPath.size()];
     for (int i = 0; i < fullName.length - 1; ++i) {
@@ -86,11 +87,11 @@ public static Namespace parentNamespace(List<PolarisEntity> catalogPath) {
   }
 
   /**
-   * Given the shortnames/ids of entities that all live under the given catalogPath, reconstructs
-   * TableIdentifier objects for each that all hold the catalogPath excluding the catalog entity.
+   * Given the entities that all live under the given catalogPath, reconstructs TableIdentifier
+   * objects for each that all hold the catalogPath excluding the catalog entity.
    */
-  public static List<TableIdentifier> nameAndIdToTableIdentifiers(
-      List<PolarisEntity> catalogPath, List<PolarisEntity.NameAndId> entities) {
+  public static List<TableIdentifier> entitiesToTableIdentifiers(
+      List<PolarisEntity> catalogPath, List<PolarisEntity> entities) {
     // Skip element 0 which is the catalog entity
     String[] parentNamespaces = new String[catalogPath.size() - 1];
     for (int i = 0; i < parentNamespaces.length; ++i) {

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

@@ -44,48 +44,6 @@
  */
 public class PolarisEntity extends PolarisBaseEntity {
 
-  public static class NameAndId {
-    private final String name;
-    private final long id;
-
-    public NameAndId(String name, long id) {
-      this.name = name;
-      this.id = id;
-    }
-
-    public String getName() {
-      return name;
-    }
-
-    public long getId() {
-      return id;
-    }
-  }
-
-  public static class TypeSubTypeAndName {
-    private final PolarisEntityType type;
-    private final PolarisEntitySubType subType;
-    private final String name;
-
-    public TypeSubTypeAndName(PolarisEntityType type, PolarisEntitySubType subType, String name) {
-      this.type = type;
-      this.subType = subType;
-      this.name = name;
-    }
-
-    public PolarisEntityType getType() {
-      return type;
-    }
-
-    public PolarisEntitySubType getSubType() {
-      return subType;
-    }
-
-    public String getName() {
-      return name;
-    }
-  }
-
   @JsonCreator
   private PolarisEntity(
       @JsonProperty("catalogId") long catalogId,
@@ -183,16 +141,6 @@ public static List<PolarisEntityCore> toCoreList(List<PolarisEntity> path) {
         .orElse(null);
   }
 
-  public static List<NameAndId> toNameAndIdList(List<EntityNameLookupRecord> entities) {
-    return Optional.ofNullable(entities)
-        .map(
-            list ->
-                list.stream()
-                    .map(record -> new NameAndId(record.getName(), record.getId()))
-                    .collect(Collectors.toList()))
-        .orElse(null);
-  }
-
   public PolarisEntity(@Nonnull PolarisBaseEntity sourceEntity) {
     super(
         sourceEntity.getCatalogId(),
@@ -223,11 +171,6 @@ public PolarisEntitySubType getSubType() {
     return PolarisEntitySubType.fromCode(getSubTypeCode());
   }
 
-  @JsonIgnore
-  public NameAndId nameAndId() {
-    return new NameAndId(name, id);
-  }
-
   @Override
   public String toString() {
     return "name="

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

@@ -57,7 +57,6 @@
 import org.apache.polaris.core.persistence.dao.entity.EntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityWithPath;
-import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.LoadGrantsResult;
 import org.apache.polaris.core.persistence.dao.entity.LoadPolicyMappingsResult;
 import org.apache.polaris.core.persistence.dao.entity.PolicyAttachmentResult;
@@ -692,11 +691,13 @@ private void revokeGrantRecord(
 
   /** {@inheritDoc} */
   @Override
-  public @Nonnull ListEntitiesResult listEntities(
+  public @Nonnull <T> Page<T> listEntities(
       @Nonnull PolarisCallContext callCtx,
       @Nullable List<PolarisEntityCore> catalogPath,
       @Nonnull PolarisEntityType entityType,
       @Nonnull PolarisEntitySubType entitySubType,
+      @Nonnull Predicate<PolarisBaseEntity> entityFilter,
+      @Nonnull Function<PolarisBaseEntity, T> transformer,
       @Nonnull PageToken pageToken) {
     // get meta store we should be using
     BasePersistence ms = callCtx.getMetaStore();
@@ -709,14 +710,12 @@ private void revokeGrantRecord(
             ? 0L
             : catalogPath.get(catalogPath.size() - 1).getId();
 
-    // prune the returned list with only entities matching the entity subtype
-    Predicate<PolarisBaseEntity> filter =
-        entitySubType != PolarisEntitySubType.ANY_SUBTYPE
-            ? e -> e.getSubTypeCode() == entitySubType.getCode()
-            : entity -> true;
-
-    Page<EntityNameLookupRecord> resultPage =
-        ms.listEntities(callCtx, catalogId, parentId, entityType, filter, pageToken);
+    Predicate<PolarisBaseEntity> filter;
+    if (entitySubType == PolarisEntitySubType.ANY_SUBTYPE) {
+      filter = entityFilter;
+    } else {
+      filter = e -> e.getSubTypeCode() == entitySubType.getCode() && entityFilter.test(e);
+    }
 
     // TODO: Use post-validation to enforce consistent view against catalogPath. In the
     // meantime, happens-before ordering semantics aren't guaranteed during high-concurrency
@@ -725,7 +724,8 @@ private void revokeGrantRecord(
     // in-flight request (the cache-based resolution follows a different path entirely).
 
     // done
-    return ListEntitiesResult.fromPage(resultPage);
+    return ms.listEntities(
+        callCtx, catalogId, parentId, entityType, filter, transformer, pageToken);
   }
 
   /** {@inheritDoc} */

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

@@ -23,6 +23,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.Predicate;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.auth.PolarisGrantManager;
 import org.apache.polaris.core.auth.PolarisSecretsManager;
@@ -45,8 +47,8 @@
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityWithPath;
 import org.apache.polaris.core.persistence.dao.entity.GenerateEntityIdResult;
-import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
+import org.apache.polaris.core.persistence.pagination.Page;
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.policy.PolarisPolicyMappingManager;
 import org.apache.polaris.core.storage.PolarisCredentialVendor;
@@ -110,25 +112,57 @@ EntityResult readEntityByName(
       @Nonnull String name);
 
   /**
-   * List all entities of the specified type under the specified catalogPath. If the catalogPath is
-   * null, listed entities will be top-level entities like catalogs.
+   * List entities where some predicate returns true and transform the entities with a function
    *
    * @param callCtx call context
    * @param catalogPath path inside a catalog. If null or empty, the entities to list are top-level,
    *     like catalogs
-   * @param entityType entity type
-   * @param entitySubType entity subtype. Can be the special value ANY_SUBTYPE to match any subtype.
-   *     Else exact match will be performed.
-   * @return all entities name, ids and subtype under the specified namespace.
+   * @param entityType type of entities to list
+   * @param entityFilter the filter to be applied to each entity. Only entities where the predicate
+   *     returns true are returned in the list
+   * @param transformer the transformation function applied to the {@link PolarisBaseEntity} before
+   *     returning
+   * @return the paged list of entities for which the predicate returns true
    */
   @Nonnull
-  ListEntitiesResult listEntities(
+  <T> Page<T> listEntities(
       @Nonnull PolarisCallContext callCtx,
       @Nullable List<PolarisEntityCore> catalogPath,
       @Nonnull PolarisEntityType entityType,
       @Nonnull PolarisEntitySubType entitySubType,
+      @Nonnull Predicate<PolarisBaseEntity> entityFilter,
+      @Nonnull Function<PolarisBaseEntity, T> transformer,
       @Nonnull PageToken pageToken);
 
+  /**
+   * List all entities and transform the entities with a function
+   *
+   * @param callCtx call context
+   * @param catalogPath path inside a catalog. If null or empty, the entities to list are top-level,
+   *     like catalogs
+   * @param entityType type of entities to list
+   * @param transformer the transformation function applied to the {@link PolarisBaseEntity} before
+   *     returning
+   * @return the full list of entities
+   */
+  @Nonnull
+  default <T> List<T> listAllEntities(
+      @Nonnull PolarisCallContext callCtx,
+      @Nullable List<PolarisEntityCore> catalogPath,
+      @Nonnull PolarisEntityType entityType,
+      @Nonnull PolarisEntitySubType entitySubType,
+      @Nonnull Function<PolarisBaseEntity, T> transformer) {
+    return listEntities(
+            callCtx,
+            catalogPath,
+            entityType,
+            entitySubType,
+            e -> true,
+            transformer,
+            PageToken.readEverything())
+        .items();
+  }
+
   /**
    * Generate a new unique id that can be used by the Polaris client when it needs to create a new
    * entity

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

@@ -26,6 +26,8 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.function.Predicate;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.LocationBasedEntity;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
@@ -44,14 +46,14 @@
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.dao.entity.EntityWithPath;
 import org.apache.polaris.core.persistence.dao.entity.GenerateEntityIdResult;
-import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.LoadGrantsResult;
 import org.apache.polaris.core.persistence.dao.entity.LoadPolicyMappingsResult;
 import org.apache.polaris.core.persistence.dao.entity.PolicyAttachmentResult;
 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.PolicyEntity;
 import org.apache.polaris.core.policy.PolicyType;
@@ -117,11 +119,13 @@ public EntityResult readEntityByName(
   }
 
   @Override
-  public ListEntitiesResult listEntities(
+  public @Nonnull <T> Page<T> listEntities(
       @Nonnull PolarisCallContext callCtx,
       @Nullable List<PolarisEntityCore> catalogPath,
       @Nonnull PolarisEntityType entityType,
       @Nonnull PolarisEntitySubType entitySubType,
+      @Nonnull Predicate<PolarisBaseEntity> entityFilter,
+      @Nonnull Function<PolarisBaseEntity, T> transformer,
       @Nonnull PageToken pageToken) {
     callCtx.getDiagServices().fail("illegal_method_in_transaction_workspace", "listEntities");
     return null;