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

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/TestHadoopCatalogOperations.java

@@ -77,8 +77,11 @@
 import org.apache.gravitino.storage.IdGenerator;
 import org.apache.gravitino.storage.RandomIdGenerator;
 import org.apache.gravitino.storage.relational.RelationalEntityStore;
+import org.apache.gravitino.storage.relational.helper.CatalogIds;
+import org.apache.gravitino.storage.relational.helper.SchemaIds;
 import org.apache.gravitino.storage.relational.service.CatalogMetaService;
 import org.apache.gravitino.storage.relational.service.MetalakeMetaService;
+import org.apache.gravitino.storage.relational.service.SchemaMetaService;
 import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -204,17 +207,68 @@ public static void setUp() {
         .when(spyCatalogMetaService)
         .getCatalogIdByMetalakeIdAndName(Mockito.anyLong(), Mockito.anyString());
 
+    SchemaMetaService serviceMetaService = SchemaMetaService.getInstance();
+    SchemaMetaService spySchemaMetaService = Mockito.spy(serviceMetaService);
+
+    doReturn(new CatalogIds(1L, 1L))
+        .when(spyCatalogMetaService)
+        .getCatalogIdByMetalakeAndCatalogName(Mockito.anyString(), Mockito.anyString());
+
+    doReturn(new SchemaIds(1L, 1L, 1L))
+        .when(spySchemaMetaService)
+        .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+            Mockito.anyString(), Mockito.anyString(), Mockito.eq("schema11"));
+
+    for (int i = 10; i < 30; i++) {
+      doReturn(new SchemaIds(1L, 1L, (long) i))
+          .when(spySchemaMetaService)
+          .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+              Mockito.anyString(), Mockito.anyString(), Mockito.eq("schema" + i));
+    }
+
+    Stream<Arguments> argumentsStream = testRenameArguments();
+    argumentsStream.forEach(
+        arguments -> {
+          String oldName = (String) arguments.get()[0];
+          String newName = (String) arguments.get()[1];
+          long schemaId = idGenerator.nextId();
+          doReturn(new SchemaIds(1L, 1L, schemaId))
+              .when(spySchemaMetaService)
+              .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                  Mockito.anyString(), Mockito.anyString(), Mockito.eq("s24_" + oldName));
+          doReturn(new SchemaIds(1L, 1L, schemaId))
+              .when(spySchemaMetaService)
+              .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                  Mockito.anyString(), Mockito.anyString(), Mockito.eq("s24_" + newName));
+        });
+
+    locationArguments()
+        .forEach(
+            arguments -> {
+              String name = (String) arguments.get()[0];
+              long schemaId = idGenerator.nextId();
+              doReturn(new SchemaIds(1L, 1L, schemaId))
+                  .when(spySchemaMetaService)
+                  .getSchemaIdByMetalakeNameAndCatalogNameAndSchemaName(
+                      Mockito.anyString(), Mockito.anyString(), Mockito.eq("s1_" + name));
+            });
+
     MockedStatic<MetalakeMetaService> metalakeMetaServiceMockedStatic =
         Mockito.mockStatic(MetalakeMetaService.class);
     MockedStatic<CatalogMetaService> catalogMetaServiceMockedStatic =
         Mockito.mockStatic(CatalogMetaService.class);
+    MockedStatic<SchemaMetaService> schemaMetaServiceMockedStatic =
+        Mockito.mockStatic(SchemaMetaService.class);
 
     metalakeMetaServiceMockedStatic
         .when(MetalakeMetaService::getInstance)
         .thenReturn(spyMetaservice);
     catalogMetaServiceMockedStatic
         .when(CatalogMetaService::getInstance)
         .thenReturn(spyCatalogMetaService);
+    schemaMetaServiceMockedStatic
+        .when(SchemaMetaService::getInstance)
+        .thenReturn(spySchemaMetaService);
   }
 
   @AfterAll
@@ -856,8 +910,8 @@ void testTrailSlash() throws IOException {
 
   @Test
   public void testGetFileLocation() throws IOException {
-    String schemaName = "schema1024";
-    String comment = "comment1024";
+    String schemaName = "schema29";
+    String comment = "schema29";
     String schemaPath = TEST_ROOT_PATH + "/" + schemaName;
     createSchema(schemaName, comment, null, schemaPath);
 

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) {