feat(listSecurableObjects): add support for TOPIC batch retrieval

unknowntpo · unknowntpo · commit 8fea43ade040 · 2025-03-06T20:48:12.000+08:00
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaMapper.java b/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")
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java b/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);
diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TopicMetaBaseSQLProvider.java b/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,"
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
@@ -31,7 +31,9 @@
 import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableColumnMapper;
 import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.gravitino.storage.relational.po.ColumnPO;
 import org.apache.gravitino.storage.relational.po.FilesetPO;
@@ -386,6 +388,103 @@ public static Map<Long, String> getTableObjectFullNames(List<Long> ids) {
     return tableIdAndNameMap;
   }
 
+  public static Map<Long, String> getTopicObjectFullNames(List<Long> ids) {
+    List<TopicPO> topicPOs =
+        SessionUtils.getWithoutCommit(
+            TopicMetaMapper.class, mapper -> mapper.listTopicPOsByTopicIds(ids));
+
+    if (topicPOs == null || topicPOs.isEmpty()) {
+      return new HashMap<>();
+    }
+
+    List<Long> catalogIds =
+        topicPOs.stream().map(TopicPO::getCatalogId).collect(Collectors.toList());
+    List<Long> schemaIds = topicPOs.stream().map(TopicPO::getSchemaId).collect(Collectors.toList());
+
+    Map<Long, String> catalogIdAndNameMap = getCatalogIdAndNameMap(catalogIds);
+    Map<Long, String> schemaIdAndNameMap = getSchemaIdAndNameMap(schemaIds);
+
+    HashMap<Long, String> topicIdAndNameMap = new HashMap<>();
+
+    topicPOs.forEach(
+        tablePO -> {
+          // 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 fullName of tablePO to null
+          String catalogName = catalogIdAndNameMap.getOrDefault(tablePO.getCatalogId(), null);
+          if (catalogName == null) {
+            LOG.warn("The catalog of topic {} may be deleted", tablePO.getTopicId());
+            topicIdAndNameMap.put(tablePO.getTopicId(), null);
+            return;
+          }
+
+          String schemaName = schemaIdAndNameMap.getOrDefault(tablePO.getSchemaId(), null);
+          if (schemaName == null) {
+            LOG.warn("The schema of topic {} may be deleted", tablePO.getTopicId());
+            topicIdAndNameMap.put(tablePO.getTopicId(), null);
+            return;
+          }
+
+          String fullName = DOT_JOINER.join(catalogName, schemaName, tablePO.getTopicName());
+          topicIdAndNameMap.put(tablePO.getTopicId(), fullName);
+        });
+
+    return topicIdAndNameMap;
+  }
+
+  public static Map<Long, String> getColumnObjectFullNames(List<Long> ids) {
+    List<ColumnPO> columnPOs =
+        SessionUtils.getWithoutCommit(
+            TableColumnMapper.class, mapper -> mapper.listColumnPOsByColumnIds(ids));
+
+    if (columnPOs == null || columnPOs.isEmpty()) {
+      return new HashMap<>();
+    }
+
+    List<Long> catalogIds =
+        columnPOs.stream().map(ColumnPO::getCatalogId).collect(Collectors.toList());
+    List<Long> schemaIds =
+        columnPOs.stream().map(ColumnPO::getSchemaId).collect(Collectors.toList());
+    List<Long> tableIds = columnPOs.stream().map(ColumnPO::getTableId).collect(Collectors.toList());
+
+    Map<Long, String> catalogIdAndNameMap = getCatalogIdAndNameMap(catalogIds);
+    Map<Long, String> schemaIdAndNameMap = getSchemaIdAndNameMap(schemaIds);
+    Map<Long, String> tableIdAndNameMap = getTableIdAndNameMap(tableIds);
+
+    HashMap<Long, String> columnIdAndNameMap = new HashMap<>();
+
+    columnPOs.forEach(
+        columnPO -> {
+          // 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 fullName of filesetPO to null
+          String catalogName = catalogIdAndNameMap.getOrDefault(columnPO.getCatalogId(), null);
+          if (catalogName == null) {
+            LOG.warn("The catalog of column {} may be deleted", columnPO.getColumnId());
+            columnIdAndNameMap.put(columnPO.getColumnId(), null);
+            return;
+          }
+
+          String schemaName = schemaIdAndNameMap.getOrDefault(columnPO.getSchemaId(), null);
+          if (schemaName == null) {
+            LOG.warn("The schema of column {} may be deleted", columnPO.getColumnId());
+            columnIdAndNameMap.put(columnPO.getColumnId(), null);
+            return;
+          }
+
+          String tableName = tableIdAndNameMap.getOrDefault(columnPO.getTableId(), null);
+          if (tableName == null) {
+            LOG.warn("The table of column {} may be deleted", columnPO.getColumnId());
+            columnIdAndNameMap.put(columnPO.getColumnId(), null);
+            return;
+          }
+
+          String fullName =
+              DOT_JOINER.join(catalogName, schemaName, tableName, columnPO.getColumnName());
+          columnIdAndNameMap.put(columnPO.getColumnId(), fullName);
+        });
+
+    return columnIdAndNameMap;
+  }
+
   public static Map<Long, String> getCatalogObjectFullNames(List<Long> ids) {
     List<CatalogPO> catalogPOs =
         SessionUtils.getWithoutCommit(
@@ -409,6 +508,14 @@ public static Map<Long, String> getCatalogObjectFullNames(List<Long> ids) {
     return catalogIdAndNameMap;
   }
 
+  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));
+  }
+
   public static Map<Long, String> getSchemaIdAndNameMap(List<Long> schemaIds) {
     List<SchemaPO> schemaPOS =
         SessionUtils.getWithoutCommit(
@@ -417,11 +524,10 @@ public static Map<Long, String> getSchemaIdAndNameMap(List<Long> schemaIds) {
         .collect(Collectors.toMap(SchemaPO::getSchemaId, SchemaPO::getSchemaName));
   }
 
-  public static Map<Long, String> getCatalogIdAndNameMap(List<Long> catalogIds) {
-    List<CatalogPO> catalogPOs =
+  public static Map<Long, String> getTableIdAndNameMap(List<Long> tableIds) {
+    List<TablePO> tablePOS =
         SessionUtils.getWithoutCommit(
-            CatalogMetaMapper.class, mapper -> mapper.listCatalogPOsByCatalogIds(catalogIds));
-    return catalogPOs.stream()
-        .collect(Collectors.toMap(CatalogPO::getCatalogId, CatalogPO::getCatalogName));
+            TableMetaMapper.class, mapper -> mapper.listTablePOsByTableIds(tableIds));
+    return tablePOS.stream().collect(Collectors.toMap(TablePO::getTableId, TablePO::getTableName));
   }
 }
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
@@ -368,6 +368,7 @@ private static List<SecurableObject> listSecurableObjects(RolePO po) {
                 case FILESET:
                 case TABLE:
                 case MODEL:
+                case TOPIC:
                   List<Long> objectIds =
                       objects.stream()
                           .map(SecurableObjectPO::getMetadataObjectId)
@@ -385,7 +386,9 @@ private static List<SecurableObject> listSecurableObjects(RolePO po) {
                               MetadataObject.Type.TABLE,
                               MetadataObjectService::getTableObjectFullNames,
                               MetadataObject.Type.MODEL,
-                              MetadataObjectService::getModelObjectFullNames);
+                              MetadataObjectService::getModelObjectFullNames,
+                              MetadataObject.Type.TOPIC,
+                              MetadataObjectService::getTopicObjectFullNames);
 
                   // dynamically calling getter function based on type
                   Map<Long, String> objectIdAndNameMap =
diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestSecurableObjects.java
@@ -80,13 +80,15 @@ public void testAllTypeSecurableObjects() throws IOException {
             "fileset",
             auditInfo);
     backend.insert(fileset, false);
+
     TableEntity table =
         createTableEntity(
             RandomIdGenerator.INSTANCE.nextId(),
             Namespace.of("metalake", "catalog", "schema"),
             "table",
             auditInfo);
     backend.insert(table, false);
+
     TopicEntity topic =
         createTopicEntity(
             RandomIdGenerator.INSTANCE.nextId(),
@@ -122,15 +124,19 @@ public void testAllTypeSecurableObjects() throws IOException {
     SecurableObject schemaObject =
         SecurableObjects.ofSchema(
             catalogObject, "schema", Lists.newArrayList(Privileges.UseSchema.allow()));
+
     SecurableObject tableObject =
         SecurableObjects.ofTable(
             schemaObject, "table", Lists.newArrayList(Privileges.SelectTable.allow()));
+
     SecurableObject filesetObject =
         SecurableObjects.ofFileset(
             schemaObject, "fileset", Lists.newArrayList(Privileges.ReadFileset.allow()));
+
     SecurableObject topicObject =
         SecurableObjects.ofTopic(
             schemaObject, "topic", Lists.newArrayList(Privileges.ConsumeTopic.deny()));
+
     SecurableObject modelObject =
         SecurableObjects.ofModel(
             schemaObject, "model", Lists.newArrayList(Privileges.ConsumeTopic.deny()));
