[#6570] improvement(core): Optimize fetching entity parent id logic

core/src/main/java/org/apache/gravitino/storage/relational/helper/CatalogIds.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.helper;
+
+public class CatalogIds {
+  private Long metalakeId;
+  private Long catalogId;
+
+  public CatalogIds(Long metalakeId, Long catalogId) {
+    this.metalakeId = metalakeId;
+    this.catalogId = catalogId;
+  }
+
+  public Long getMetalakeId() {
+    return metalakeId;
+  }
+
+  public Long getCatalogId() {
+    return catalogId;
+  }
+}

core/src/main/java/org/apache/gravitino/storage/relational/helper/SchemaIds.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.helper;
+
+public class SchemaIds {
+  private Long metalakeId;
+  private Long catalogId;
+  private Long schemaId;
+
+  public SchemaIds(Long metalakeId, Long catalogId, Long schemaId) {
+    this.metalakeId = metalakeId;
+    this.catalogId = catalogId;
+    this.schemaId = schemaId;
+  }
+
+  public Long getMetalakeId() {
+    return metalakeId;
+  }
+
+  public Long getCatalogId() {
+    return catalogId;
+  }
+
+  public Long getSchemaId() {
+    return schemaId;
+  }
+}

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.helper.CatalogIds;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.ibatis.annotations.DeleteProvider;
 import org.apache.ibatis.annotations.InsertProvider;
@@ -87,4 +88,10 @@ Integer updateCatalogMeta(
       method = "deleteCatalogMetasByLegacyTimeline")
   Integer deleteCatalogMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+
+  @SelectProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "selectCatalogIdByMetalakeNameAndCatalogName")
+  CatalogIds selectCatalogIdByMetalakeNameAndCatalogName(
+      @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName);
 }

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

@@ -71,6 +71,11 @@ public static String selectCatalogMetaByMetalakeIdAndName(
     return getProvider().selectCatalogMetaByMetalakeIdAndName(metalakeId, name);
   }
 
+  public static String selectCatalogIdByMetalakeNameAndCatalogName(
+      @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) {
+    return getProvider().selectCatalogIdByMetalakeNameAndCatalogName(metalakeName, catalogName);
+  }
+
   public static String selectCatalogMetaById(@Param("catalogId") Long catalogId) {
     return getProvider().selectCatalogMetaById(catalogId);
   }

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.helper.SchemaIds;
 import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.ibatis.annotations.DeleteProvider;
 import org.apache.ibatis.annotations.InsertProvider;
@@ -91,4 +92,12 @@ Integer updateSchemaMeta(
       method = "deleteSchemaMetasByLegacyTimeline")
   Integer deleteSchemaMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
+
+  @SelectProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName")
+  SchemaIds selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName);
 }

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

@@ -103,4 +103,13 @@ public static String deleteSchemaMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return getProvider().deleteSchemaMetasByLegacyTimeline(legacyTimeline, limit);
   }
+
+  public static String selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName) {
+    return getProvider()
+        .selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+            metalakeName, catalogName, schemaName);
+  }
 }

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

@@ -74,6 +74,15 @@ public String selectCatalogMetaByMetalakeIdAndName(
         + " WHERE metalake_id = #{metalakeId} AND catalog_name = #{catalogName} AND deleted_at = 0";
   }
 
+  public String selectCatalogIdByMetalakeNameAndCatalogName(
+      @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) {
+    return "SELECT me.metalake_id as metalakeId, ca.catalog_id as catalogId FROM "
+        + TABLE_NAME
+        + " ca INNER JOIN metalake_meta me ON ca.metalake_id = me.metalake_id"
+        + " WHERE me.metalake_name = #{metalakeName} AND ca.catalog_name = #{catalogName} "
+        + " AND ca.deleted_at = 0 AND me.deleted_at = 0";
+  }
+
   public String selectCatalogMetaById(@Param("catalogId") Long catalogId) {
     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/mapper/provider/base/SchemaMetaBaseSQLProvider.java

@@ -190,4 +190,21 @@ public String deleteSchemaMetasByLegacyTimeline(
         + TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
+
+  public String selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName) {
+    return "SELECT metalake_meta.metalake_id as metalakeId, catalog_meta.catalog_id as catalogId, "
+        + " schema_id as schemaId"
+        + " FROM metalake_meta"
+        + " JOIN catalog_meta ON metalake_meta.metalake_id = catalog_meta.metalake_id"
+        + " JOIN schema_meta ON catalog_meta.catalog_id = schema_meta.catalog_id"
+        + " WHERE metalake_name = #{metalakeName}"
+        + " AND catalog_name = #{catalogName}"
+        + " AND schema_name = #{schemaName}"
+        + " AND schema_meta.deleted_at = 0"
+        + " AND catalog_meta.deleted_at = 0"
+        + " AND metalake_meta.deleted_at = 0";
+  }
 }

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

@@ -33,6 +33,7 @@
 import org.apache.gravitino.exceptions.NonEmptyEntityException;
 import org.apache.gravitino.meta.CatalogEntity;
 import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.storage.relational.helper.CatalogIds;
 import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
@@ -80,6 +81,12 @@ public CatalogPO getCatalogPOByMetalakeIdAndName(Long metalakeId, String catalog
     return catalogPO;
   }
 
+  public CatalogIds getCatalogIdByMetalakeAndCatalogName(String metalakeName, String catalogName) {
+    return SessionUtils.getWithoutCommit(
+        CatalogMetaMapper.class,
+        mapper -> mapper.selectCatalogIdByMetalakeNameAndCatalogName(metalakeName, catalogName));
+  }
+
   // Catalog may be deleted, so the CatalogPO may be null.
   @Nullable
   public CatalogPO getCatalogPOById(Long catalogId) {

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

@@ -20,7 +20,11 @@
 package org.apache.gravitino.storage.relational.service;
 
 import com.google.common.base.Preconditions;
+import org.apache.gravitino.Entity;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.storage.relational.helper.CatalogIds;
+import org.apache.gravitino.storage.relational.helper.SchemaIds;
 
 /** The service class for common metadata operations. */
 public class CommonMetaService {
@@ -36,25 +40,47 @@ public Long getParentEntityIdByNamespace(Namespace namespace) {
     Preconditions.checkArgument(
         !namespace.isEmpty() && namespace.levels().length <= 3,
         "Namespace should not be empty and length should be less than or equal to 3.");
+
     Long parentEntityId = null;
-    if (namespace.levels().length >= 1) {
+    if (namespace.levels().length == 1) {
       parentEntityId = MetalakeMetaService.getInstance().getMetalakeIdByName(namespace.level(0));
+      if (parentEntityId == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.METALAKE.name().toLowerCase(),
+            namespace);
+      }
     }
 
-    if (namespace.levels().length >= 2) {
-      parentEntityId =
+    if (namespace.levels().length == 2) {
+      CatalogIds catalogIds =
           CatalogMetaService.getInstance()
-              .getCatalogIdByMetalakeIdAndName(parentEntityId, namespace.level(1));
+              .getCatalogIdByMetalakeAndCatalogName(namespace.level(0), namespace.level(1));
+      parentEntityId = catalogIds == null ? null : catalogIds.getCatalogId();
+
+      if (parentEntityId == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.CATALOG.name().toLowerCase(),
+            namespace);
+      }
     }
 
-    if (namespace.levels().length >= 3) {
-      parentEntityId =
+    if (namespace.levels().length == 3) {
+      SchemaIds schemaIds =
           SchemaMetaService.getInstance()
-              .getSchemaIdByCatalogIdAndName(parentEntityId, namespace.level(2));
+              .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                  namespace.level(0), namespace.level(1), namespace.level(2));
+      parentEntityId = schemaIds == null ? null : schemaIds.getSchemaId();
+
+      if (parentEntityId == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.SCHEMA.name().toLowerCase(),
+            namespace);
+      }
     }
-    Preconditions.checkState(
-        parentEntityId != null && parentEntityId > 0,
-        "Parent entity id should not be null and should be greater than 0.");
+
     return parentEntityId;
   }
 
@@ -63,21 +89,46 @@ public Long[] getParentEntityIdsByNamespace(Namespace namespace) {
         !namespace.isEmpty() && namespace.levels().length <= 3,
         "Namespace should not be empty and length should be less than or equal to 3.");
     Long[] parentEntityIds = new Long[namespace.levels().length];
-    if (namespace.levels().length >= 1) {
+
+    if (namespace.levels().length == 1) {
       parentEntityIds[0] =
           MetalakeMetaService.getInstance().getMetalakeIdByName(namespace.level(0));
-    }
+      if (parentEntityIds[0] == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.METALAKE.name().toLowerCase(),
+            namespace);
+      }
 
-    if (namespace.levels().length >= 2) {
-      parentEntityIds[1] =
+    } else if (namespace.levels().length == 2) {
+      CatalogIds catalogIds =
           CatalogMetaService.getInstance()
-              .getCatalogIdByMetalakeIdAndName(parentEntityIds[0], namespace.level(1));
-    }
+              .getCatalogIdByMetalakeAndCatalogName(namespace.level(0), namespace.level(1));
+      parentEntityIds[0] = catalogIds == null ? null : catalogIds.getMetalakeId();
+      parentEntityIds[1] = catalogIds == null ? null : catalogIds.getCatalogId();
 
-    if (namespace.levels().length >= 3) {
-      parentEntityIds[2] =
+      if (parentEntityIds[1] == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.CATALOG.name().toLowerCase(),
+            namespace);
+      }
+
+    } else if (namespace.levels().length == 3) {
+      SchemaIds schemaIds =
           SchemaMetaService.getInstance()
-              .getSchemaIdByCatalogIdAndName(parentEntityIds[1], namespace.level(2));
+              .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                  namespace.level(0), namespace.level(1), namespace.level(2));
+      parentEntityIds[0] = schemaIds == null ? null : schemaIds.getMetalakeId();
+      parentEntityIds[1] = schemaIds == null ? null : schemaIds.getCatalogId();
+      parentEntityIds[2] = schemaIds == null ? null : schemaIds.getSchemaId();
+
+      if (parentEntityIds[2] == null) {
+        throw new NoSuchEntityException(
+            NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+            Entity.EntityType.SCHEMA.name().toLowerCase(),
+            namespace);
+      }
     }
 
     return parentEntityIds;

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

@@ -34,6 +34,7 @@
 import org.apache.gravitino.meta.ModelEntity;
 import org.apache.gravitino.meta.SchemaEntity;
 import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.storage.relational.helper.SchemaIds;
 import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.FilesetVersionMapper;
 import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
@@ -78,6 +79,15 @@ public SchemaPO getSchemaPOByCatalogIdAndName(Long catalogId, String schemaName)
     return schemaPO;
   }
 
+  public SchemaIds getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+      String metalakeName, String catalogName, String schemaName) {
+    return SessionUtils.getWithoutCommit(
+        SchemaMetaMapper.class,
+        mapper ->
+            mapper.selectSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                metalakeName, catalogName, schemaName));
+  }
+
   // Schema may be deleted, so the SchemaPO may be null.
   public SchemaPO getSchemaPOById(Long schemaId) {
     return SessionUtils.getWithoutCommit(