[#6238] improvement(storage): Improve get role performance when roles is bound to many metadata. (#6455)

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

fix issue #6238
improve performance when a single role is bound to many metadata.

### Why are the changes needed?

Use batch queries when getting role securable object full names instead
of loop queries to get each securable object full name.

Fix: #6238

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

### How was this patch tested?
Unit tests and integration tests have all passed, this feature has been
running internally at Xiaomi for two weeks.

Co-authored-by: luoxin5 <luoxin5@xiaomi.com>

Author: FourFriends

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

@@ -41,6 +41,9 @@ public interface CatalogMetaMapper {
   @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByMetalakeId")
   List<CatalogPO> listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId);
 
+  @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByCatalogIds")
+  List<CatalogPO> listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds);
+
   @SelectProvider(
       type = CatalogMetaSQLProviderFactory.class,
       method = "selectCatalogIdByMetalakeIdAndName")

core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.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.CatalogMetaBaseSQLProvider;
@@ -56,6 +57,10 @@ public static String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metala
     return getProvider().listCatalogPOsByMetalakeId(metalakeId);
   }
 
+  public static String listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds) {
+    return getProvider().listCatalogPOsByCatalogIds(catalogIds);
+  }
+
   public static String selectCatalogIdByMetalakeIdAndName(
       @Param("metalakeId") Long metalakeId, @Param("catalogName") String name) {
     return getProvider().selectCatalogIdByMetalakeIdAndName(metalakeId, name);

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

@@ -67,6 +67,31 @@ public interface FilesetMetaMapper {
   @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "listFilesetPOsBySchemaId")
   List<FilesetPO> listFilesetPOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @Results({
+    @Result(property = "filesetId", column = "fileset_id"),
+    @Result(property = "filesetName", column = "fileset_name"),
+    @Result(property = "metalakeId", column = "metalake_id"),
+    @Result(property = "catalogId", column = "catalog_id"),
+    @Result(property = "schemaId", column = "schema_id"),
+    @Result(property = "type", column = "type"),
+    @Result(property = "auditInfo", column = "audit_info"),
+    @Result(property = "currentVersion", column = "current_version"),
+    @Result(property = "lastVersion", column = "last_version"),
+    @Result(property = "deletedAt", column = "deleted_at"),
+    @Result(property = "filesetVersionPO.id", column = "id"),
+    @Result(property = "filesetVersionPO.metalakeId", column = "version_metalake_id"),
+    @Result(property = "filesetVersionPO.catalogId", column = "version_catalog_id"),
+    @Result(property = "filesetVersionPO.schemaId", column = "version_schema_id"),
+    @Result(property = "filesetVersionPO.filesetId", column = "version_fileset_id"),
+    @Result(property = "filesetVersionPO.version", column = "version"),
+    @Result(property = "filesetVersionPO.filesetComment", column = "fileset_comment"),
+    @Result(property = "filesetVersionPO.properties", column = "properties"),
+    @Result(property = "filesetVersionPO.storageLocation", column = "storage_location"),
+    @Result(property = "filesetVersionPO.deletedAt", column = "version_deleted_at")
+  })
+  @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "listFilesetPOsByFilesetIds")
+  List<FilesetPO> listFilesetPOsByFilesetIds(@Param("filesetIds") List<Long> filesetIds);
+
   @SelectProvider(
       type = FilesetMetaSQLProviderFactory.class,
       method = "selectFilesetIdBySchemaIdAndName")

core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.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.FilesetMetaBaseSQLProvider;
@@ -55,6 +56,10 @@ public static String listFilesetPOsBySchemaId(@Param("schemaId") Long schemaId)
     return getProvider().listFilesetPOsBySchemaId(schemaId);
   }
 
+  public static String listFilesetPOsByFilesetIds(@Param("filesetIds") List<Long> filesetIds) {
+    return getProvider().listFilesetPOsByFilesetIds(filesetIds);
+  }
+
   public static String selectFilesetIdBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("filesetName") String name) {
     return getProvider().selectFilesetIdBySchemaIdAndName(schemaId, name);

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

@@ -41,6 +41,9 @@ public interface SchemaMetaMapper {
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByCatalogId")
   List<SchemaPO> listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId);
 
+  @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsBySchemaIds")
+  List<SchemaPO> listSchemaPOsBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
+
   @SelectProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "selectSchemaIdByCatalogIdAndName")

core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.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.SchemaMetaBaseSQLProvider;
@@ -50,6 +51,10 @@ static class SchemaMetaMySQLProvider extends SchemaMetaBaseSQLProvider {}
 
   static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider {}
 
+  public static String listSchemaPOsBySchemaIds(@Param("schemaIds") List<Long> schemaIds) {
+    return getProvider().listSchemaPOsBySchemaIds(schemaIds);
+  }
+
   public static String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId) {
     return getProvider().listSchemaPOsByCatalogId(catalogId);
   }

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

@@ -21,6 +21,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -36,6 +37,24 @@ public String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
         + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
   }
 
+  public String listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds) {
+    return "<script>"
+        + "SELECT catalog_id as catalogId, catalog_name as catalogName,"
+        + " metalake_id as metalakeId, type, provider,"
+        + " catalog_comment as catalogComment, properties, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE catalog_id in ("
+        + "<foreach collection='catalogIds' item='catalogId' separator=','>"
+        + "#{catalogId}"
+        + "</foreach>"
+        + ") "
+        + " AND deleted_at = 0"
+        + "</script>";
+  }
+
   public String selectCatalogIdByMetalakeIdAndName(
       @Param("metalakeId") Long metalakeId, @Param("catalogName") String name) {
     return "SELECT catalog_id as catalogId FROM "

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

@@ -22,6 +22,7 @@
 import static org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper.META_TABLE_NAME;
 import static org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper.VERSION_TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.FilesetPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -50,6 +51,28 @@ public String selectFilesetIdBySchemaIdAndName(
         + " AND deleted_at = 0";
   }
 
+  public String listFilesetPOsByFilesetIds(@Param("filesetIds") List<Long> filesetIds) {
+    return "<script>"
+        + "SELECT fm.fileset_id, fm.fileset_name, fm.metalake_id, fm.catalog_id, fm.schema_id,"
+        + " fm.type, fm.audit_info, fm.current_version, fm.last_version, fm.deleted_at,"
+        + " vi.id, vi.metalake_id as version_metalake_id, vi.catalog_id as version_catalog_id,"
+        + " vi.schema_id as version_schema_id, vi.fileset_id as version_fileset_id,"
+        + " vi.version, vi.fileset_comment, vi.properties, vi.storage_location,"
+        + " vi.deleted_at as version_deleted_at"
+        + " FROM "
+        + META_TABLE_NAME
+        + " fm INNER JOIN "
+        + VERSION_TABLE_NAME
+        + " vi ON fm.fileset_id = vi.fileset_id AND fm.current_version = vi.version"
+        + " WHERE fm.fileset_id in ("
+        + "<foreach collection='filesetIds' item='filesetId' separator=','>"
+        + "#{filesetId}"
+        + "</foreach>"
+        + ") "
+        + " AND fm.deleted_at = 0 AND vi.deleted_at = 0"
+        + "</script>";
+  }
+
   public String selectFilesetMetaBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("filesetName") String name) {
     return "SELECT fm.fileset_id, fm.fileset_name, fm.metalake_id, fm.catalog_id, fm.schema_id,"

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

@@ -20,6 +20,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -35,6 +36,24 @@ public String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId) {
         + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
+  public String listSchemaPOsBySchemaIds(@Param("schemaIds") List<Long> schemaIds) {
+    return "<script>"
+        + "SELECT schema_id as schemaId, schema_name as schemaName,"
+        + " metalake_id as metalakeId, catalog_id as catalogId,"
+        + " schema_comment as schemaComment, properties, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE schema_id in ("
+        + "<foreach collection='schemaIds' item='schemaId' separator=','>"
+        + "#{schemaId}"
+        + "</foreach>"
+        + ") "
+        + " AND deleted_at = 0"
+        + "</script>";
+  }
+
   public String selectSchemaIdByCatalogIdAndName(
       @Param("catalogId") Long catalogId, @Param("schemaName") String name) {
     return "SELECT schema_id as schemaId FROM "

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

@@ -18,13 +18,17 @@
  */
 package org.apache.gravitino.storage.relational.service;
 
+import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
@@ -38,12 +42,18 @@
 import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.GroupRoleRelMapper;
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper;
 import org.apache.gravitino.storage.relational.mapper.UserRoleRelMapper;
+import org.apache.gravitino.storage.relational.po.CatalogPO;
+import org.apache.gravitino.storage.relational.po.FilesetPO;
 import org.apache.gravitino.storage.relational.po.RolePO;
+import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
 import org.apache.gravitino.storage.relational.utils.POConverters;
@@ -54,6 +64,8 @@
 
 /** The service class for role metadata. It provides the basic database operations for role. */
 public class RoleMetaService {
+  private static final String DOT = ".";
+  private static final Joiner DOT_JOINER = Joiner.on(DOT);
 
   private static final Logger LOG = LoggerFactory.getLogger(RoleMetaService.class);
   private static final RoleMetaService INSTANCE = new RoleMetaService();
@@ -353,21 +365,58 @@ private static List<SecurableObject> listSecurableObjects(RolePO po) {
     List<SecurableObjectPO> securableObjectPOs = listSecurableObjectsByRoleId(po.getRoleId());
     List<SecurableObject> securableObjects = Lists.newArrayList();
 
-    for (SecurableObjectPO securableObjectPO : securableObjectPOs) {
-      String fullName =
-          MetadataObjectService.getMetadataObjectFullName(
-              securableObjectPO.getType(), securableObjectPO.getMetadataObjectId());
-      if (fullName != null) {
-        securableObjects.add(
-            POConverters.fromSecurableObjectPO(
-                fullName, securableObjectPO, getType(securableObjectPO.getType())));
-      } else {
-        LOG.warn(
-            "The securable object {} {} may be deleted",
-            securableObjectPO.getMetadataObjectId(),
-            securableObjectPO.getType());
-      }
-    }
+    securableObjectPOs.stream()
+        .collect(Collectors.groupingBy(SecurableObjectPO::getType))
+        .forEach(
+            (type, objects) -> {
+              // If the type is Fileset, use the batch retrieval interface;
+              // otherwise, use the single retrieval interface
+              if (type.equals(MetadataObject.Type.FILESET.name())) {
+                List<Long> filesetIds =
+                    objects.stream()
+                        .map(SecurableObjectPO::getMetadataObjectId)
+                        .collect(Collectors.toList());
+
+                Map<Long, String> filesetIdAndNameMap = getFilesetObjectFullNames(filesetIds);
+
+                for (SecurableObjectPO securableObjectPO : objects) {
+                  String fullName =
+                      filesetIdAndNameMap.get(securableObjectPO.getMetadataObjectId());
+                  if (fullName != null) {
+                    securableObjects.add(
+                        POConverters.fromSecurableObjectPO(
+                            fullName, securableObjectPO, getType(securableObjectPO.getType())));
+                  } else {
+                    LOG.warn(
+                        "The securable object {} {} may be deleted",
+                        securableObjectPO.getMetadataObjectId(),
+                        securableObjectPO.getType());
+                  }
+                }
+              } else {
+                // todo：to get other securable object fullNames using batch retrieving
+                for (SecurableObjectPO securableObjectPO : objects) {
+                  String fullName =
+                      MetadataObjectService.getMetadataObjectFullName(
+                          securableObjectPO.getType(), securableObjectPO.getMetadataObjectId());
+                  if (fullName != null) {
+                    securableObjects.add(
+                        POConverters.fromSecurableObjectPO(
+                            fullName, securableObjectPO, getType(securableObjectPO.getType())));
+                  } else {
+                    LOG.warn(
+                        "The securable object {} {} may be deleted",
+                        securableObjectPO.getMetadataObjectId(),
+                        securableObjectPO.getType());
+                  }
+                }
+              }
+            });
+
+    // To ensure that the order of the returned securable objects remains consistent,
+    // the securable objects are sorted by fullName here,
+    // since the order of securable objects after grouping by is different each time.
+    securableObjects.sort(Comparator.comparing(MetadataObject::fullName));
 
     return securableObjects;
   }
@@ -394,4 +443,64 @@ private static MetadataObject.Type getType(String type) {
   private static String getEntityType(SecurableObject securableObject) {
     return securableObject.type().name();
   }
+
+  public static Map<Long, String> getFilesetObjectFullNames(List<Long> ids) {
+    List<FilesetPO> filesetPOs =
+        SessionUtils.getWithoutCommit(
+            FilesetMetaMapper.class, mapper -> mapper.listFilesetPOsByFilesetIds(ids));
+
+    if (filesetPOs == null || filesetPOs.isEmpty()) {
+      return new HashMap<>();
+    }
+
+    List<Long> catalogIds =
+        filesetPOs.stream().map(FilesetPO::getCatalogId).collect(Collectors.toList());
+    List<Long> schemaIds =
+        filesetPOs.stream().map(FilesetPO::getSchemaId).collect(Collectors.toList());
+
+    Map<Long, String> catalogIdAndNameMap = getCatalogIdAndNameMap(catalogIds);
+    Map<Long, String> schemaIdAndNameMap = getSchemaIdAndNameMap(schemaIds);
+
+    HashMap<Long, String> filesetIdAndNameMap = new HashMap<>();
+
+    filesetPOs.forEach(
+        filesetPO -> {
+          // since the catalog or schema can be deleted, we need to check the null value,
+          // and when catalog or schema is deleted, we will set catalogName or schemaName to null
+          String catalogName = catalogIdAndNameMap.getOrDefault(filesetPO.getCatalogId(), null);
+          if (catalogName == null) {
+            LOG.warn("The catalog of fileset {} may be deleted", filesetPO.getFilesetId());
+            filesetIdAndNameMap.put(filesetPO.getFilesetId(), null);
+            return;
+          }
+
+          String schemaName = schemaIdAndNameMap.getOrDefault(filesetPO.getSchemaId(), null);
+          if (schemaName == null) {
+            LOG.warn("The schema of fileset {} may be deleted", filesetPO.getFilesetId());
+            filesetIdAndNameMap.put(filesetPO.getFilesetId(), null);
+            return;
+          }
+
+          String fullName = DOT_JOINER.join(catalogName, schemaName, filesetPO.getFilesetName());
+          filesetIdAndNameMap.put(filesetPO.getFilesetId(), fullName);
+        });
+
+    return filesetIdAndNameMap;
+  }
+
+  public static Map<Long, String> getSchemaIdAndNameMap(List<Long> schemaIds) {
+    List<SchemaPO> schemaPOS =
+        SessionUtils.getWithoutCommit(
+            SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsBySchemaIds(schemaIds));
+    return schemaPOS.stream()
+        .collect(Collectors.toMap(SchemaPO::getSchemaId, SchemaPO::getSchemaName));
+  }
+
+  public static Map<Long, String> getCatalogIdAndNameMap(List<Long> catalogIds) {
+    List<CatalogPO> catalogPOs =
+        SessionUtils.getWithoutCommit(
+            CatalogMetaMapper.class, mapper -> mapper.listCatalogPOsByCatalogIds(catalogIds));
+    return catalogPOs.stream()
+        .collect(Collectors.toMap(CatalogPO::getCatalogId, CatalogPO::getCatalogName));
+  }
 }