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,19 @@
 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 = "listCatalogPOsByCatalogName")
+  List<CatalogPO> listCatalogPOsByCatalogName(@Param("catalogName") String catalogName);
+
+  @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsBySchemaName")
+  List<CatalogPO> listCatalogPOsBySchemaName(@Param("schemaName") String schemaName);
+
   @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,18 @@ 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 listCatalogPOsByCatalogName(@Param("catalogName") String catalogName) {
+    return getProvider().listCatalogPOsByCatalogName(catalogName);
+  }
+
+  public static String listCatalogPOsBySchemaName(@Param("schemaName") String schemaName) {
+    return getProvider().listCatalogPOsBySchemaName(schemaName);
+  }
+
   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,57 @@
 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.mapper.SchemaMetaMapper;
 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 listCatalogPOsByCatalogName(@Param("catalogName") String catalogName) {
+    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 cm.catalog_name = #{catalogName}"
+        + " AND mm.deleted_at = 0 AND cm.deleted_at = 0";
+  }
+
+  public String listCatalogPOsBySchemaName(@Param("schemaName") String schemaName) {
+    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 "
+        + SchemaMetaMapper.TABLE_NAME
+        + " mm ON cm.metalake_id = sm.metalake_id"
+        + " WHERE sm.schema_name = #{schemaName}"
+        + " AND cm.deleted_at = 0 AND sm.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

@@ -20,6 +20,7 @@
 
 import com.google.common.base.Preconditions;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
@@ -119,12 +120,26 @@ public CatalogEntity getCatalogByIdentifier(NameIdentifier identifier) {
 
   public List<CatalogEntity> listCatalogsByNamespace(Namespace namespace) {
     NamespaceUtil.checkCatalog(namespace);
+    List<CatalogPO> catalogPOS = Collections.emptyList();
+    if (namespace.levels().length >= 1) {
+      catalogPOS =
+          SessionUtils.getWithoutCommit(
+              CatalogMetaMapper.class,
+              mapper -> mapper.listCatalogPOsByMetalakeName(namespace.level(0)));
+    }
+    if (namespace.levels().length >= 2) {
+      catalogPOS =
+          SessionUtils.getWithoutCommit(
+              CatalogMetaMapper.class,
+              mapper -> mapper.listCatalogPOsByCatalogName(namespace.level(1)));
+    }
 
-    Long metalakeId = CommonMetaService.getInstance().getParentEntityIdByNamespace(namespace);
-
-    List<CatalogPO> catalogPOS =
-        SessionUtils.getWithoutCommit(
-            CatalogMetaMapper.class, mapper -> mapper.listCatalogPOsByMetalakeId(metalakeId));
+    if (namespace.levels().length >= 3) {
+      catalogPOS =
+          SessionUtils.getWithoutCommit(
+              CatalogMetaMapper.class,
+              mapper -> mapper.listCatalogPOsBySchemaName(namespace.level(2)));
+    }
 
     return POConverters.fromCatalogPOs(catalogPOS, namespace);
   }