refactor(core): Optimize Catalog storage queries using JOIN for JDBC …

…backend

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

@@ -38,6 +38,11 @@
 public interface CatalogMetaMapper {
   String TABLE_NAME = "catalog_meta";
 
+  @SelectProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "listCatalogPOsByMetalakeName")
+  List<CatalogPO> listCatalogPOsByMetalakeName(@Param("metalakeName") String metalakeName);
+
   @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByMetalakeId")
   List<CatalogPO> listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId);
 

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

@@ -53,6 +53,10 @@ static class CatalogMetaMySQLProvider extends CatalogMetaBaseSQLProvider {}
 
   static class CatalogMetaH2Provider extends CatalogMetaBaseSQLProvider {}
 
+  public static String listCatalogPOsByMetalakeName(@Param("metalakeName") String metalakeName) {
+    return getProvider().listCatalogPOsByMetalakeName(metalakeName);
+  }
+
   public static String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return getProvider().listCatalogPOsByMetalakeId(metalakeId);
   }

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

@@ -22,10 +22,26 @@
 import static org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper.TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.ibatis.annotations.Param;
 
 public class CatalogMetaBaseSQLProvider {
+  public String listCatalogPOsByMetalakeName(@Param("metalakeName") String metalakeName) {
+    return "SELECT cm.catalog_id as catalogId, cm.catalog_name as catalogName,"
+        + " cm.metalake_id as metalakeId, cm.type, cm.provider,"
+        + " cm.catalog_comment as catalogComment, cm.properties, cm.audit_info as auditInfo,"
+        + " cm.current_version as currentVersion, cm.last_version as lastVersion,"
+        + " cm.deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " cm JOIN "
+        + MetalakeMetaMapper.TABLE_NAME
+        + " mm ON cm.metalake_id = mm.metalake_id"
+        + " WHERE mm.metalake_name = #{metalakeName}"
+        + " AND mm.deleted_at = 0 AND cm.deleted_at = 0";
+  }
+
   public String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return "SELECT catalog_id as catalogId, catalog_name as catalogName,"
         + " metalake_id as metalakeId, type, provider,"

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

@@ -119,12 +119,10 @@ public CatalogEntity getCatalogByIdentifier(NameIdentifier identifier) {
 
   public List<CatalogEntity> listCatalogsByNamespace(Namespace namespace) {
     NamespaceUtil.checkCatalog(namespace);
-
-    Long metalakeId = CommonMetaService.getInstance().getParentEntityIdByNamespace(namespace);
-
     List<CatalogPO> catalogPOS =
         SessionUtils.getWithoutCommit(
-            CatalogMetaMapper.class, mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId));
+            CatalogMetaMapper.class,
+            mapper -> mapper.listCatalogPOsByMetalakeName(namespace.level(0)));
 
     return POConverters.fromCatalogPOs(catalogPOS, namespace);
   }