[#9802] feat(authz): support preload table metadata in batch metadata authorization (#9831)

### What changes were proposed in this pull request?

support preload table metadata in batch metadata authorization

### Why are the changes needed?

Fix: #9802

### Does this PR introduce _any_ user-facing change?

None

### How was this patch tested?

Existing IT test in
org.apache.gravitino.client.integration.test.authorization.TableAuthorizationIT

---------

Co-authored-by: yangyang zhong <35210666+hdygxsj@users.noreply.github.com>
Co-authored-by: Rory <he@datastrato.com>

Author: 3 people

core/src/main/java/org/apache/gravitino/EntityStore.java

@@ -20,6 +20,8 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.lang.reflect.Array;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Function;
 import org.apache.commons.lang3.tuple.Pair;
@@ -165,6 +167,35 @@ <E extends Entity & HasIdentifier> E update(
   <E extends Entity & HasIdentifier> E get(NameIdentifier ident, EntityType entityType, Class<E> e)
       throws NoSuchEntityException, IOException;
 
+  /**
+   * Batch get the entity from the underlying storage.
+   *
+   * @param idents the unique identifier of the entity
+   * @param entityType the general type of the entity
+   * @param clazz the entity class instance
+   * @param <E> the class of entity
+   * @return the entity retrieved from the underlying storage
+   * @throws NoSuchEntityException if the entity does not exist
+   */
+  <E extends Entity & HasIdentifier> List<E> batchGet(
+      List<NameIdentifier> idents, EntityType entityType, Class<E> clazz);
+
+  /**
+   * Batch get the entity from the underlying storage.
+   *
+   * @param idents the unique identifier of the entity
+   * @param entityType the general type of the entity
+   * @param clazz the entity class instance
+   * @param <E> the class of entity
+   * @return the entity retrieved from the underlying storage
+   * @throws NoSuchEntityException if the entity does not exist
+   */
+  default <E extends Entity & HasIdentifier> E[] batchGet(
+      NameIdentifier[] idents, EntityType entityType, Class<E> clazz) {
+    return batchGet(Arrays.asList(idents), entityType, clazz)
+        .toArray(size -> (E[]) Array.newInstance(clazz, size));
+  }
+
   /**
    * Delete the entity from the underlying storage by the specified {@link
    * org.apache.gravitino.NameIdentifier}.

core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java

@@ -286,6 +286,18 @@ public <E extends Entity & HasIdentifier> E get(
     }
   }
 
+  @Override
+  public <E extends Entity & HasIdentifier> List<E> batchGet(
+      List<NameIdentifier> identifiers, Entity.EntityType entityType) {
+    switch (entityType) {
+      case TABLE:
+        return (List<E>) TableMetaService.getInstance().batchGetTableByIdentifier(identifiers);
+      default:
+        throw new UnsupportedEntityTypeException(
+            "Unsupported entity type: %s for get operation", entityType);
+    }
+  }
+
   @Override
   public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean cascade)
       throws IOException {

core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java

@@ -113,6 +113,18 @@ <E extends Entity & HasIdentifier> E update(
   <E extends Entity & HasIdentifier> E get(NameIdentifier ident, Entity.EntityType entityType)
       throws IOException;
 
+  /**
+   * Batch retrieves the entities associated with the identifiers and the entity type.
+   *
+   * @param <E> The type of the entity returned.
+   * @param identifiers The identifiers of the entities.
+   * @param entityType The type of the entity.
+   * @return The entities associated with the identifiers and the entity type, or null if the key
+   *     does not exist.
+   */
+  <E extends Entity & HasIdentifier> List<E> batchGet(
+      List<NameIdentifier> identifiers, Entity.EntityType entityType);
+
   /**
    * Soft deletes the entity associated with the identifier and the entity type.
    *

core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java

@@ -23,6 +23,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
@@ -153,6 +154,27 @@ public <E extends Entity & HasIdentifier> E get(
         });
   }
 
+  @Override
+  public <E extends Entity & HasIdentifier> List<E> batchGet(
+      List<NameIdentifier> idents, Entity.EntityType entityType, Class<E> clazz) {
+    List<E> allEntities = new ArrayList<>();
+    List<NameIdentifier> noCacheIdents =
+        idents.stream()
+            .filter(
+                ident -> {
+                  Optional<E> entity = cache.getIfPresent(ident, entityType);
+                  entity.ifPresent(allEntities::add);
+                  return entity.isEmpty();
+                })
+            .toList();
+    List<E> fetchEntities = backend.batchGet(noCacheIdents, entityType);
+    for (E entity : fetchEntities) {
+      cache.put(entity);
+      allEntities.add(entity);
+    }
+    return allEntities;
+  }
+
   @Override
   public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean cascade)
       throws IOException {

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaMapper.java

@@ -95,4 +95,8 @@ Integer updateTableMeta(
       method = "deleteTableMetasByLegacyTimeline")
   Integer deleteTableMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+
+  @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "batchSelectTableByIdentifier")
+  List<TablePO> batchSelectTableByIdentifier(
+      @Param("schemaId") Long schemaId, @Param("tableNames") List<String> tableNames);
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java

@@ -108,4 +108,9 @@ public static String deleteTableMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return getProvider().deleteTableMetasByLegacyTimeline(legacyTimeline, limit);
   }
+
+  public static String batchSelectTableByIdentifier(
+      @Param("schemaId") Long schemaId, @Param("tableNames") List<String> tableNames) {
+    return getProvider().batchSelectTableByIdentifier(schemaId, tableNames);
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TableMetaBaseSQLProvider.java

@@ -21,6 +21,9 @@
 import static org.apache.gravitino.storage.relational.mapper.TableMetaMapper.TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TableVersionMapper;
 import org.apache.gravitino.storage.relational.po.TablePO;
 import org.apache.ibatis.annotations.Param;
@@ -230,4 +233,81 @@ public String deleteTableMetasByLegacyTimeline(
         + TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
+
+  public String selectTableByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("tableName") String tableName) {
+    return """
+        SELECT
+            sm.schema_id AS schemaId,
+            cm.catalog_id AS catalogId,
+            tm.table_id AS tableId,
+            tm.table_name AS tableName,
+            tm.metalake_id AS metalakeId,
+            tm.audit_info AS auditInfo,
+            tm.current_version AS currentVersion,
+            tm.last_version AS lastVersion,
+            tm.deleted_at AS deletedAt,
+            tvi.format AS format,
+            tvi.properties AS properties,
+            tvi.partitioning AS partitions,
+            tvi.sort_orders AS sortOrders,
+            tvi.distribution AS distribution,
+            tvi.indexes AS indexes,
+            tvi.comment AS comment
+        FROM
+            %s mm
+        INNER JOIN
+            %s cm ON mm.metalake_id = cm.metalake_id
+            AND cm.catalog_name = #{catalogName}
+            AND cm.deleted_at = 0
+        LEFT JOIN
+            %s sm ON cm.catalog_id = sm.catalog_id
+            AND sm.schema_name = #{schemaName}
+            AND sm.deleted_at = 0
+        LEFT JOIN
+            %s tm ON sm.schema_id = tm.schema_id
+            AND tm.table_name = #{tableName}
+            AND tm.deleted_at = 0
+        LEFT JOIN
+            %s tvi ON tm.table_id = tvi.table_id
+            AND tm.current_version = tvi.version
+            AND tvi.deleted_at = 0
+        WHERE
+            mm.metalake_name = #{metalakeName}
+            AND mm.deleted_at = 0;
+            """
+        .formatted(
+            MetalakeMetaMapper.TABLE_NAME,
+            CatalogMetaMapper.TABLE_NAME,
+            SchemaMetaMapper.TABLE_NAME,
+            TABLE_NAME,
+            TableVersionMapper.TABLE_NAME);
+  }
+
+  public String batchSelectTableByIdentifier(
+      @Param("schemaId") Long schemaId, @Param("tableNames") List<String> tableNames) {
+    return """
+            <script>
+            SELECT
+            tm.table_id AS tableId,
+            tm.table_name AS tableName,
+            tm.metalake_id AS metalakeId,
+            tm.audit_info AS auditInfo,
+            tm.current_version AS currentVersion,
+            tm.last_version AS lastVersion,
+            tm.deleted_at AS deletedAt
+            FROM %s tm
+            WHERE schema_id = #{schemaId}
+            AND table_name IN
+            <foreach collection="tableNames" item="tableName" open="(" separator="," close=")">
+            #{tableName}
+            </foreach>
+             AND deleted_at = 0
+             </script>
+            """
+        .formatted(TABLE_NAME);
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/service/TableMetaService.java

@@ -22,11 +22,14 @@
 
 import com.google.common.base.Preconditions;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.HasIdentifier;
 import org.apache.gravitino.MetadataObject;
@@ -319,6 +322,32 @@ public int deleteTableVersionByLegacyTimeline(Long legacyTimeline, int limit) {
         mapper -> mapper.deleteTableVersionByLegacyTimeline(legacyTimeline, limit));
   }
 
+  @Monitored(
+      metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+      baseMetricName = "batchGetTableByIdentifier")
+  public List<TableEntity> batchGetTableByIdentifier(List<NameIdentifier> identifiers) {
+    if (CollectionUtils.isEmpty(identifiers)) {
+      return Collections.emptyList();
+    }
+    NameIdentifier firstIdent = identifiers.get(0);
+    NameIdentifier schemaIdent = NameIdentifierUtil.getSchemaIdentifier(firstIdent);
+    List<String> tableNames = new ArrayList<>(identifiers.size());
+    tableNames.add(identifiers.get(0).name());
+    for (int i = 1; i < identifiers.size(); i++) {
+      NameIdentifier ident = identifiers.get(i);
+      Preconditions.checkArgument(
+          Objects.equals(schemaIdent, NameIdentifierUtil.getSchemaIdentifier(ident)));
+      tableNames.add(ident.name());
+    }
+    Long schemaId = EntityIdService.getEntityId(schemaIdent, Entity.EntityType.SCHEMA);
+    return SessionUtils.doWithCommitAndFetchResult(
+        TableMetaMapper.class,
+        mapper -> {
+          List<TablePO> tableList = mapper.batchSelectTableByIdentifier(schemaId, tableNames);
+          return POConverters.fromTablePOs(tableList, firstIdent.namespace());
+        });
+  }
+
   private void fillTablePOBuilderParentEntityId(TablePO.Builder builder, Namespace namespace) {
     NamespaceUtil.checkTable(namespace);
     NamespacedEntityId namespacedEntityId =

core/src/test/java/org/apache/gravitino/storage/memory/TestMemoryEntityStore.java

@@ -145,6 +145,12 @@ public <E extends Entity & HasIdentifier> E get(
       return e;
     }
 
+    @Override
+    public <E extends Entity & HasIdentifier> List<E> batchGet(
+        List<NameIdentifier> idents, EntityType entityType, Class<E> e) {
+      return idents.stream().map(ident -> (E) entityMap.get(ident)).toList();
+    }
+
     @Override
     public boolean delete(NameIdentifier ident, EntityType entityType, boolean cascade)
         throws IOException {

server-common/src/main/java/org/apache/gravitino/server/authorization/MetadataAuthzHelper.java

@@ -38,6 +38,7 @@
 import org.apache.gravitino.authorization.AuthorizationRequestContext;
 import org.apache.gravitino.authorization.GravitinoAuthorizer;
 import org.apache.gravitino.dto.tag.MetadataObjectDTO;
+import org.apache.gravitino.meta.TableEntity;
 import org.apache.gravitino.server.authorization.expression.AuthorizationExpressionConstants;
 import org.apache.gravitino.server.authorization.expression.AuthorizationExpressionEvaluator;
 import org.apache.gravitino.utils.MetadataObjectUtil;
@@ -136,6 +137,7 @@ public static NameIdentifier[] filterByExpression(
       String expression,
       Entity.EntityType entityType,
       NameIdentifier[] nameIdentifiers) {
+    preloadToCache(entityType, nameIdentifiers);
     return filterByExpression(metalake, expression, entityType, nameIdentifiers, e -> e);
   }
 
@@ -291,7 +293,12 @@ private static <E> E[] doFilter(
     return futures.stream()
         .map(CompletableFuture::join)
         .filter(Objects::nonNull)
-        .toArray(size -> (E[]) Array.newInstance(entities.getClass().getComponentType(), size));
+        .toArray(size -> createArray(entities.getClass().getComponentType(), size));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <E> E[] createArray(Class<?> componentType, int size) {
+    return (E[]) Array.newInstance(componentType, size);
   }
 
   private static boolean enableAuthorization() {
@@ -318,4 +325,16 @@ private static void checkExecutor() {
       }
     }
   }
+
+  private static void preloadToCache(
+      Entity.EntityType entityType, NameIdentifier[] nameIdentifiers) {
+    Config config = GravitinoEnv.getInstance().config();
+    if (config != null && !config.get(Configs.CACHE_ENABLED)) {
+      if (entityType == Entity.EntityType.TABLE) {
+        GravitinoEnv.getInstance()
+            .entityStore()
+            .batchGet(nameIdentifiers, entityType, TableEntity.class);
+      }
+    }
+  }
 }