From 6dfb680e3d5aa630973c808cb65a03a314965b9c Mon Sep 17 00:00:00 2001
From: Eric Chang <e850506@gmail.com>
Date: Thu, 6 Mar 2025 21:27:38 +0800
Subject: [PATCH] feat(listSecurableObjects): add batch retrieval for SCHEMA

---
 .../service/MetadataObjectService.java        | 38 +++++++++++++++++++
 .../relational/service/RoleMetaService.java   |  6 ++-
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetadataObjectService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetadataObjectService.java
index e0db76cd26b..496ae988165 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetadataObjectService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetadataObjectService.java
@@ -508,6 +508,44 @@ public static Map<Long, String> getCatalogObjectFullNames(List<Long> ids) {
     return catalogIdAndNameMap;
   }
 
+  public static Map<Long, String> getSchemaObjectFullNames(List<Long> ids) {
+    List<SchemaPO> schemaPOs =
+        SessionUtils.getWithoutCommit(
+            SchemaMetaMapper.class, mapper -> mapper.listSchemaPOsBySchemaIds(ids));
+
+    if (schemaPOs == null || schemaPOs.isEmpty()) {
+      return new HashMap<>();
+    }
+
+    List<Long> catalogIds =
+        schemaPOs.stream().map(SchemaPO::getCatalogId).collect(Collectors.toList());
+
+    Map<Long, String> catalogIdAndNameMap = getCatalogIdAndNameMap(catalogIds);
+
+    HashMap<Long, String> schemaIdAndNameMap = new HashMap<>();
+
+    schemaPOs.forEach(
+        schemaPO -> {
+          if (schemaPO.getSchemaId() == null) {
+            schemaIdAndNameMap.put(schemaPO.getSchemaId(), null);
+            return;
+          }
+
+          String catalogName = catalogIdAndNameMap.getOrDefault(schemaPO.getCatalogId(), null);
+          if (catalogName == null) {
+            LOG.warn("The catalog of schema {} may be deleted", schemaPO.getSchemaId());
+            schemaIdAndNameMap.put(schemaPO.getSchemaId(), null);
+            return;
+          }
+
+          String fullName = DOT_JOINER.join(catalogName, schemaPO.getSchemaName());
+
+          schemaIdAndNameMap.put(schemaPO.getSchemaId(), fullName);
+        });
+
+    return schemaIdAndNameMap;
+  }
+
   public static Map<Long, String> getCatalogIdAndNameMap(List<Long> catalogIds) {
     List<CatalogPO> catalogPOs =
         SessionUtils.getWithoutCommit(
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
index 8903cbdb3ae..e3a043e5dd6 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
@@ -372,10 +372,12 @@ private static List<SecurableObject> listSecurableObjects(RolePO po) {
                           MetadataObjectService::getMetalakeObjectFullNames,
                           MetadataObject.Type.CATALOG,
                           MetadataObjectService::getCatalogObjectFullNames,
-                          MetadataObject.Type.FILESET,
-                          MetadataObjectService::getFilesetObjectFullNames,
+                          MetadataObject.Type.SCHEMA,
+                          MetadataObjectService::getSchemaObjectFullNames,
                           MetadataObject.Type.TABLE,
                           MetadataObjectService::getTableObjectFullNames,
+                          MetadataObject.Type.FILESET,
+                          MetadataObjectService::getFilesetObjectFullNames,
                           MetadataObject.Type.MODEL,
                           MetadataObjectService::getModelObjectFullNames,
                           MetadataObject.Type.TOPIC,