[#6474] improvement(storage): batch listing securable objects in RoleMetaService (#6601)

<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
   Examples:
     - "[#123] feat(operator): support xxx"
     - "[#233] fix: check null before access result in xxx"
     - "[MINOR] refactor: fix typo in variable name"
     - "[MINOR] docs: fix typo in README"
     - "[#255] test: fix flaky test NameOfTheTest"
   Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->

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

Batch retrieving securable objects for `metalake`, `catalog`, `schema`,
`table`, `topic`, `fileset` (fileset has been implemented at #6455).

### Why are the changes needed?

Improve performance when role is bounded to lots of `SecurableObject`s.

Fix: #6474
### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

In `TestSecurableObjects.testAllTypeSecurableObjects`.

Related Issue and PR: #6238 #6455

Author: unknowntpo

api/src/main/java/org/apache/gravitino/authorization/SecurableObjects.java

@@ -105,7 +105,7 @@ public static SecurableObject ofTopic(
   }
 
   /**
-   * Create the table {@link SecurableObject} with the given securable schema object, fileset name
+   * Create the fileset {@link SecurableObject} with the given securable schema object, fileset name
    * and privileges.
    *
    * @param schema The schema securable object

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

@@ -47,6 +47,11 @@ public interface MetalakeMetaMapper {
   @SelectProvider(type = MetalakeMetaSQLProviderFactory.class, method = "selectMetalakeMetaById")
   MetalakePO selectMetalakeMetaById(@Param("metalakeId") Long metalakeId);
 
+  @SelectProvider(
+      type = MetalakeMetaSQLProviderFactory.class,
+      method = "listMetalakePOsByMetalakeIds")
+  List<MetalakePO> listMetalakePOsByMetalakeIds(@Param("metalakeIds") List<Long> metalakeIds);
+
   @SelectProvider(
       type = MetalakeMetaSQLProviderFactory.class,
       method = "selectMetalakeIdMetaByName")

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
 import org.apache.gravitino.storage.relational.mapper.provider.base.MetalakeMetaBaseSQLProvider;
@@ -68,6 +69,10 @@ public static String selectMetalakeIdMetaByName(@Param("metalakeName") String me
     return getProvider().selectMetalakeIdMetaByName(metalakeName);
   }
 
+  public static String listMetalakePOsByMetalakeIds(@Param("metalakeIds") List<Long> metalakeIds) {
+    return getProvider().listMetalakePOsByMetalakeIds(metalakeIds);
+  }
+
   public static String insertMetalakeMeta(@Param("metalakeMeta") MetalakePO metalakePO) {
     return getProvider().insertMetalakeMeta(metalakePO);
   }

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

@@ -40,6 +40,9 @@ public interface ModelMetaMapper {
   @SelectProvider(type = ModelMetaSQLProviderFactory.class, method = "listModelPOsBySchemaId")
   List<ModelPO> listModelPOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @SelectProvider(type = ModelMetaSQLProviderFactory.class, method = "listModelPOsByModelIds")
+  List<ModelPO> listModelPOsByModelIds(@Param("modelIds") List<Long> modelIds);
+
   @SelectProvider(
       type = ModelMetaSQLProviderFactory.class,
       method = "selectModelMetaBySchemaIdAndModelName")

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

@@ -19,6 +19,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
 import org.apache.gravitino.storage.relational.mapper.provider.base.ModelMetaBaseSQLProvider;
@@ -62,6 +63,10 @@ public static String listModelPOsBySchemaId(@Param("schemaId") Long schemaId) {
     return getProvider().listModelPOsBySchemaId(schemaId);
   }
 
+  public static String listModelPOsByModelIds(@Param("modelIds") List<Long> modelIds) {
+    return getProvider().listModelPOsByModelIds(modelIds);
+  }
+
   public static String selectModelMetaBySchemaIdAndModelName(
       @Param("schemaId") Long schemaId, @Param("modelName") String modelName) {
     return getProvider().selectModelMetaBySchemaIdAndModelName(schemaId, modelName);

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

@@ -41,6 +41,9 @@ public interface TableMetaMapper {
   @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "listTablePOsBySchemaId")
   List<TablePO> listTablePOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "listTablePOsByTableIds")
+  List<TablePO> listTablePOsByTableIds(@Param("tableIds") List<Long> tableIds);
+
   @SelectProvider(
       type = TableMetaSQLProviderFactory.class,
       method = "selectTableIdBySchemaIdAndName")

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

@@ -19,6 +19,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
 import org.apache.gravitino.storage.relational.mapper.provider.base.TableMetaBaseSQLProvider;
@@ -54,6 +55,10 @@ public static String listTablePOsBySchemaId(@Param("schemaId") Long schemaId) {
     return getProvider().listTablePOsBySchemaId(schemaId);
   }
 
+  public static String listTablePOsByTableIds(@Param("tableIds") List<Long> tableIds) {
+    return getProvider().listTablePOsByTableIds(tableIds);
+  }
+
   public static String selectTableIdBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("tableName") String name) {
     return getProvider().selectTableIdBySchemaIdAndName(schemaId, name);

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

@@ -40,6 +40,9 @@ public interface TopicMetaMapper {
   @SelectProvider(type = TopicMetaSQLProviderFactory.class, method = "listTopicPOsBySchemaId")
   List<TopicPO> listTopicPOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @SelectProvider(type = TopicMetaSQLProviderFactory.class, method = "listTopicPOsByTopicIds")
+  List<TopicPO> listTopicPOsByTopicIds(@Param("topicIds") List<Long> topicIds);
+
   @SelectProvider(
       type = TopicMetaSQLProviderFactory.class,
       method = "selectTopicMetaBySchemaIdAndName")

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
 import org.apache.gravitino.storage.relational.mapper.provider.base.TopicMetaBaseSQLProvider;
@@ -63,6 +64,10 @@ public static String listTopicPOsBySchemaId(@Param("schemaId") Long schemaId) {
     return getProvider().listTopicPOsBySchemaId(schemaId);
   }
 
+  public static String listTopicPOsByTopicIds(@Param("topicIds") List<Long> topicIds) {
+    return getProvider().listTopicPOsByTopicIds(topicIds);
+  }
+
   public static String selectTopicMetaBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("topicName") String topicName) {
     return getProvider().selectTopicMetaBySchemaIdAndName(schemaId, topicName);

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

@@ -20,6 +20,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.MetalakePO;
 import org.apache.ibatis.annotations.Param;
 
@@ -65,6 +66,24 @@ public String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeN
         + " WHERE metalake_name = #{metalakeName} and deleted_at = 0";
   }
 
+  public String listMetalakePOsByMetalakeIds(@Param("metalakeIds") List<Long> metalakeIds) {
+    return "<script>"
+        + " SELECT metalake_id as metalakeId, metalake_name as metalakeName,"
+        + " metalake_comment as metalakeComment, properties,"
+        + " audit_info as auditInfo, schema_version as schemaVersion,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE deleted_at = 0"
+        + " AND metalake_id in ("
+        + "<foreach collection='metalakeIds' item='metalakeId' separator=','>"
+        + "#{metalakeId}"
+        + "</foreach>"
+        + ") "
+        + "</script>";
+  }
+
   public String insertMetalakeMeta(@Param("metalakeMeta") MetalakePO metalakePO) {
     return "INSERT INTO "
         + TABLE_NAME

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

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.storage.relational.mapper.provider.base;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
 import org.apache.gravitino.storage.relational.po.ModelPO;
 import org.apache.ibatis.annotations.Param;
@@ -66,6 +67,23 @@ public String listModelPOsBySchemaId(@Param("schemaId") Long schemaId) {
         + " WHERE schema_id = #{schemaId} AND deleted_at = 0";
   }
 
+  public String listModelPOsByModelIds(List<Long> modelIds) {
+    return "<script>"
+        + " SELECT model_id AS modelId, model_name AS modelName, metalake_id AS metalakeId,"
+        + " catalog_id AS catalogId, schema_id AS schemaId, model_comment AS modelComment,"
+        + " model_properties AS modelProperties, model_latest_version AS"
+        + " modelLatestVersion, audit_info AS auditInfo, deleted_at AS deletedAt"
+        + " FROM "
+        + ModelMetaMapper.TABLE_NAME
+        + " WHERE deleted_at = 0"
+        + " AND model_id in ("
+        + "<foreach collection='modelIds' item='modelId' separator=','>"
+        + "#{modelId}"
+        + "</foreach>"
+        + ") "
+        + "</script>";
+  }
+
   public String selectModelMetaBySchemaIdAndModelName(
       @Param("schemaId") Long schemaId, @Param("modelName") String modelName) {
     return "SELECT model_id AS modelId, model_name AS modelName, metalake_id AS metalakeId,"

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

@@ -20,6 +20,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.TableMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.TablePO;
 import org.apache.ibatis.annotations.Param;
 
@@ -36,6 +37,24 @@ public String listTablePOsBySchemaId(@Param("schemaId") Long schemaId) {
         + " WHERE schema_id = #{schemaId} AND deleted_at = 0";
   }
 
+  public String listTablePOsByTableIds(List<Long> tableIds) {
+    return "<script>"
+        + " SELECT table_id as tableId, table_name as tableName,"
+        + " metalake_id as metalakeId, catalog_id as catalogId,"
+        + " schema_id as schemaId, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE deleted_at = 0"
+        + " AND table_id in ("
+        + "<foreach collection='tableIds' item='tableId' separator=','>"
+        + "#{tableId}"
+        + "</foreach>"
+        + ") "
+        + "</script>";
+  }
+
   public String selectTableIdBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("tableName") String name) {
     return "SELECT table_id as tableId FROM "

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

@@ -21,6 +21,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.TopicMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.TopicPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -90,6 +91,24 @@ public String listTopicPOsBySchemaId(@Param("schemaId") Long schemaId) {
         + " WHERE schema_id = #{schemaId} AND deleted_at = 0";
   }
 
+  public String listTopicPOsByTopicIds(@Param("topicIds") List<Long> topicIds) {
+    return "<script>"
+        + " SELECT topic_id as topicId, topic_name as topicName, metalake_id as metalakeId,"
+        + " catalog_id as catalogId, schema_id as schemaId,"
+        + " comment as comment, properties as properties, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE deleted_at = 0"
+        + " AND topic_id in ("
+        + "<foreach collection='topicIds' item='topicId' separator=','>"
+        + "#{topicId}"
+        + "</foreach>"
+        + ") "
+        + "</script>";
+  }
+
   public String selectTopicMetaBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("topicName") String topicName) {
     return "SELECT topic_id as topicId, topic_name as topicName,"