[#3854] feat(core): Add a cache service to cache the mapping of name identifer to id

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

@@ -186,6 +186,9 @@ public static void setUp() {
     MetalakeMetaService metalakeMetaService = MetalakeMetaService.getInstance();
     MetalakeMetaService spyMetaservice = Mockito.spy(metalakeMetaService);
     doReturn(1L).when(spyMetaservice).getMetalakeIdByName(Mockito.anyString());
+    doReturn(1L)
+        .when(spyMetaservice)
+        .getMetalakeIdByNameIdentifier(Mockito.any(NameIdentifier.class));
 
     CatalogMetaService catalogMetaService = CatalogMetaService.getInstance();
     CatalogMetaService spyCatalogMetaService = Mockito.spy(catalogMetaService);

catalogs/catalog-jdbc-doris/src/test/java/org/apache/gravitino/catalog/doris/operation/TestDorisTablePartitionOperations.java

@@ -57,7 +57,7 @@
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
-@Tag("gravitino-docker-it")
+@Tag("gravitino-docker-test")
 public class TestDorisTablePartitionOperations extends TestDoris {
   private static final String databaseName = GravitinoITUtils.genRandomName("doris_test_db");
   private static final Integer DEFAULT_BUCKET_SIZE = 1;

catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java

@@ -157,6 +157,9 @@ public static void setUp() {
     MetalakeMetaService metalakeMetaService = MetalakeMetaService.getInstance();
     MetalakeMetaService spyMetaservice = Mockito.spy(metalakeMetaService);
     doReturn(1L).when(spyMetaservice).getMetalakeIdByName(Mockito.anyString());
+    doReturn(1L)
+        .when(spyMetaservice)
+        .getMetalakeIdByNameIdentifier(Mockito.any(NameIdentifier.class));
 
     CatalogMetaService catalogMetaService = CatalogMetaService.getInstance();
     CatalogMetaService spyCatalogMetaService = Mockito.spy(catalogMetaService);

core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java

@@ -30,6 +30,7 @@
 import org.apache.gravitino.Config;
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.Entity;
+import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.HasIdentifier;
 import org.apache.gravitino.MetadataObject;
@@ -54,6 +55,8 @@
 import org.apache.gravitino.storage.relational.service.FilesetMetaService;
 import org.apache.gravitino.storage.relational.service.GroupMetaService;
 import org.apache.gravitino.storage.relational.service.MetalakeMetaService;
+import org.apache.gravitino.storage.relational.service.NameIdMappingService;
+import org.apache.gravitino.storage.relational.service.NameIdMappingService.EntityIdentifier;
 import org.apache.gravitino.storage.relational.service.OwnerMetaService;
 import org.apache.gravitino.storage.relational.service.RoleMetaService;
 import org.apache.gravitino.storage.relational.service.SchemaMetaService;
@@ -123,36 +126,54 @@ public boolean exists(NameIdentifier ident, Entity.EntityType entityType) throws
   @Override
   public <E extends Entity & HasIdentifier> void insert(E e, boolean overwritten)
       throws EntityAlreadyExistsException, IOException {
+    EntityType entityType;
     if (e instanceof BaseMetalake) {
       MetalakeMetaService.getInstance().insertMetalake((BaseMetalake) e, overwritten);
+      entityType = EntityType.METALAKE;
     } else if (e instanceof CatalogEntity) {
       CatalogMetaService.getInstance().insertCatalog((CatalogEntity) e, overwritten);
+      entityType = EntityType.CATALOG;
     } else if (e instanceof SchemaEntity) {
       SchemaMetaService.getInstance().insertSchema((SchemaEntity) e, overwritten);
+      entityType = EntityType.SCHEMA;
     } else if (e instanceof TableEntity) {
       TableMetaService.getInstance().insertTable((TableEntity) e, overwritten);
+      entityType = EntityType.TABLE;
     } else if (e instanceof FilesetEntity) {
       FilesetMetaService.getInstance().insertFileset((FilesetEntity) e, overwritten);
+      entityType = EntityType.FILESET;
     } else if (e instanceof TopicEntity) {
       TopicMetaService.getInstance().insertTopic((TopicEntity) e, overwritten);
+      entityType = EntityType.TOPIC;
     } else if (e instanceof UserEntity) {
       UserMetaService.getInstance().insertUser((UserEntity) e, overwritten);
+      entityType = EntityType.USER;
     } else if (e instanceof RoleEntity) {
       RoleMetaService.getInstance().insertRole((RoleEntity) e, overwritten);
+      entityType = EntityType.ROLE;
     } else if (e instanceof GroupEntity) {
       GroupMetaService.getInstance().insertGroup((GroupEntity) e, overwritten);
+      entityType = EntityType.GROUP;
     } else if (e instanceof TagEntity) {
       TagMetaService.getInstance().insertTag((TagEntity) e, overwritten);
+      entityType = EntityType.TAG;
     } else {
       throw new UnsupportedEntityTypeException(
           "Unsupported entity type: %s for insert operation", e.getClass());
     }
+
+    NameIdMappingService.getInstance()
+        .put(EntityIdentifier.of(e.nameIdentifier(), entityType), e.id());
   }
 
   @Override
   public <E extends Entity & HasIdentifier> E update(
       NameIdentifier ident, Entity.EntityType entityType, Function<E, E> updater)
       throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
+    // Remove all the children entities in the cache as we can't guarantee the children entities
+    // are still valid after the parent entity is updated.
+    EntityIdentifier entityIdentifier = EntityIdentifier.of(ident, entityType);
+    NameIdMappingService.getInstance().invalidateWithPrefix(entityIdentifier);
     switch (entityType) {
       case METALAKE:
         return (E) MetalakeMetaService.getInstance().updateMetalake(ident, updater);
@@ -210,32 +231,45 @@ public <E extends Entity & HasIdentifier> E get(
   }
 
   @Override
-  public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean cascade)
-      throws IOException {
-    switch (entityType) {
-      case METALAKE:
-        return MetalakeMetaService.getInstance().deleteMetalake(ident, cascade);
-      case CATALOG:
-        return CatalogMetaService.getInstance().deleteCatalog(ident, cascade);
-      case SCHEMA:
-        return SchemaMetaService.getInstance().deleteSchema(ident, cascade);
-      case TABLE:
-        return TableMetaService.getInstance().deleteTable(ident);
-      case FILESET:
-        return FilesetMetaService.getInstance().deleteFileset(ident);
-      case TOPIC:
-        return TopicMetaService.getInstance().deleteTopic(ident);
-      case USER:
-        return UserMetaService.getInstance().deleteUser(ident);
-      case GROUP:
-        return GroupMetaService.getInstance().deleteGroup(ident);
-      case ROLE:
-        return RoleMetaService.getInstance().deleteRole(ident);
-      case TAG:
-        return TagMetaService.getInstance().deleteTag(ident);
-      default:
-        throw new UnsupportedEntityTypeException(
-            "Unsupported entity type: %s for delete operation", entityType);
+  public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolean cascade) {
+    // Invalidate the cache first
+    EntityIdentifier entityIdentifier = EntityIdentifier.of(ident, entityType);
+    NameIdMappingService.getInstance().invalidate(entityIdentifier);
+    if (cascade) {
+      // Remove all the children entities in the cache;
+      NameIdMappingService.getInstance().invalidateWithPrefix(entityIdentifier);
+    }
+
+    try {
+      switch (entityType) {
+        case METALAKE:
+          return MetalakeMetaService.getInstance().deleteMetalake(ident, cascade);
+        case CATALOG:
+          return CatalogMetaService.getInstance().deleteCatalog(ident, cascade);
+        case SCHEMA:
+          return SchemaMetaService.getInstance().deleteSchema(ident, cascade);
+        case TABLE:
+          return TableMetaService.getInstance().deleteTable(ident);
+        case FILESET:
+          return FilesetMetaService.getInstance().deleteFileset(ident);
+        case TOPIC:
+          return TopicMetaService.getInstance().deleteTopic(ident);
+        case USER:
+          return UserMetaService.getInstance().deleteUser(ident);
+        case GROUP:
+          return GroupMetaService.getInstance().deleteGroup(ident);
+        case ROLE:
+          return RoleMetaService.getInstance().deleteRole(ident);
+        case TAG:
+          return TagMetaService.getInstance().deleteTag(ident);
+        default:
+          throw new UnsupportedEntityTypeException(
+              "Unsupported entity type: %s for delete operation", entityType);
+      }
+    } finally {
+      // Remove the entity from the cache again because we may add the cache during the deletion
+      // process
+      NameIdMappingService.getInstance().invalidateWithPrefix(entityIdentifier);
     }
   }
 
@@ -327,6 +361,8 @@ public int deleteOldVersionData(Entity.EntityType entityType, long versionRetent
   public void close() throws IOException {
     SqlSessionFactoryHelper.getInstance().close();
 
+    NameIdMappingService.getInstance().close();
+
     if (jdbcDatabase != null) {
       jdbcDatabase.close();
     }

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

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.storage.relational.service;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.util.List;
@@ -41,6 +42,7 @@
 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.service.NameIdMappingService.EntityIdentifier;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
 import org.apache.gravitino.storage.relational.utils.POConverters;
 import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -84,6 +86,7 @@ public CatalogPO getCatalogPOById(Long catalogId) {
     return catalogPO;
   }
 
+  @VisibleForTesting
   public Long getCatalogIdByMetalakeIdAndName(Long metalakeId, String catalogName) {
     Long catalogId =
         SessionUtils.getWithoutCommit(
@@ -99,6 +102,22 @@ public Long getCatalogIdByMetalakeIdAndName(Long metalakeId, String catalogName)
     return catalogId;
   }
 
+  public Long getCatalogIdByNameIdentifier(NameIdentifier identifier) {
+    NameIdentifierUtil.checkCatalog(identifier);
+    EntityIdentifier catalogIdent = EntityIdentifier.of(identifier, Entity.EntityType.CATALOG);
+
+    return NameIdMappingService.getInstance()
+        .get(
+            catalogIdent,
+            ident -> {
+              String catalogName = ident.ident.name();
+              Long metalakeId =
+                  CommonMetaService.getInstance()
+                      .getParentEntityIdByNamespace(ident.ident.namespace());
+              return getCatalogIdByMetalakeIdAndName(metalakeId, catalogName);
+            });
+  }
+
   public CatalogEntity getCatalogByIdentifier(NameIdentifier identifier) {
     NameIdentifierUtil.checkCatalog(identifier);
     String catalogName = identifier.name();
@@ -192,10 +211,7 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) {
     NameIdentifierUtil.checkCatalog(identifier);
 
     String catalogName = identifier.name();
-    Long metalakeId =
-        CommonMetaService.getInstance().getParentEntityIdByNamespace(identifier.namespace());
-
-    Long catalogId = getCatalogIdByMetalakeIdAndName(metalakeId, catalogName);
+    Long catalogId = getCatalogIdByNameIdentifier(identifier);
 
     if (cascade) {
       SessionUtils.doMultipleWithCommit(
@@ -255,8 +271,6 @@ public boolean deleteCatalog(NameIdentifier identifier, boolean cascade) {
   public int deleteCatalogMetasByLegacyTimeline(Long legacyTimeline, int limit) {
     return SessionUtils.doWithCommitAndFetchResult(
         CatalogMetaMapper.class,
-        mapper -> {
-          return mapper.deleteCatalogMetasByLegacyTimeline(legacyTimeline, limit);
-        });
+        mapper -> mapper.deleteCatalogMetasByLegacyTimeline(legacyTimeline, limit));
   }
 }

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.storage.relational.service;
 
 import com.google.common.base.Preconditions;
+import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 
 /** The service class for common metadata operations. */
@@ -36,25 +37,25 @@ 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) {
-      parentEntityId = MetalakeMetaService.getInstance().getMetalakeIdByName(namespace.level(0));
+    String[] level = namespace.levels();
+    NameIdentifier ident = NameIdentifier.of(level);
+    Long entityId = null;
+
+    switch (level.length) {
+      case 1:
+        entityId = MetalakeMetaService.getInstance().getMetalakeIdByNameIdentifier(ident);
+        break;
+      case 2:
+        entityId = CatalogMetaService.getInstance().getCatalogIdByNameIdentifier(ident);
+        break;
+      case 3:
+        entityId = SchemaMetaService.getInstance().getSchemaIdByNameIdentifier(ident);
+        break;
     }
 
-    if (namespace.levels().length >= 2) {
-      parentEntityId =
-          CatalogMetaService.getInstance()
-              .getCatalogIdByMetalakeIdAndName(parentEntityId, namespace.level(1));
-    }
-
-    if (namespace.levels().length >= 3) {
-      parentEntityId =
-          SchemaMetaService.getInstance()
-              .getSchemaIdByCatalogIdAndName(parentEntityId, namespace.level(2));
-    }
     Preconditions.checkState(
-        parentEntityId != null && parentEntityId > 0,
+        entityId != null && entityId > 0,
         "Parent entity id should not be null and should be greater than 0.");
-    return parentEntityId;
+    return entityId;
   }
 }

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

@@ -23,7 +23,9 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
+import org.apache.commons.lang3.ArrayUtils;
 import org.apache.gravitino.Entity;
+import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.HasIdentifier;
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.NameIdentifier;
@@ -35,6 +37,7 @@
 import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper;
 import org.apache.gravitino.storage.relational.po.FilesetMaxVersionPO;
 import org.apache.gravitino.storage.relational.po.FilesetPO;
+import org.apache.gravitino.storage.relational.service.NameIdMappingService.EntityIdentifier;
 import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
 import org.apache.gravitino.storage.relational.utils.POConverters;
 import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -81,7 +84,22 @@ public FilesetPO getFilesetPOById(Long filesetId) {
     return filesetPO;
   }
 
-  public Long getFilesetIdBySchemaIdAndName(Long schemaId, String filesetName) {
+  public Long getFilesetIdByNameIdentifier(NameIdentifier identifier) {
+    NameIdentifierUtil.checkFileset(identifier);
+
+    EntityIdentifier fileIdentifier = EntityIdentifier.of(identifier, EntityType.FILESET);
+    return NameIdMappingService.getInstance()
+        .get(
+            fileIdentifier,
+            ident -> {
+              Long schemaId =
+                  CommonMetaService.getInstance()
+                      .getParentEntityIdByNamespace(ident.ident.namespace());
+              return getFilesetIdBySchemaIdAndName(schemaId, ident.ident.name());
+            });
+  }
+
+  private Long getFilesetIdBySchemaIdAndName(Long schemaId, String filesetName) {
     Long filesetId =
         SessionUtils.getWithoutCommit(
             FilesetMetaMapper.class,
@@ -223,12 +241,7 @@ public <E extends Entity & HasIdentifier> FilesetEntity updateFileset(
   public boolean deleteFileset(NameIdentifier identifier) {
     NameIdentifierUtil.checkFileset(identifier);
 
-    String filesetName = identifier.name();
-
-    Long schemaId =
-        CommonMetaService.getInstance().getParentEntityIdByNamespace(identifier.namespace());
-
-    Long filesetId = getFilesetIdBySchemaIdAndName(schemaId, filesetName);
+    Long filesetId = getFilesetIdByNameIdentifier(identifier);
 
     // We should delete meta and version info
     SessionUtils.doMultipleWithCommit(
@@ -300,24 +313,24 @@ public int deleteFilesetVersionsByRetentionCount(Long versionRetentionCount, int
 
   private void fillFilesetPOBuilderParentEntityId(FilesetPO.Builder builder, Namespace namespace) {
     NamespaceUtil.checkFileset(namespace);
-    Long parentEntityId = null;
+    Long entityId;
+
     for (int level = 0; level < namespace.levels().length; level++) {
-      String name = namespace.level(level);
+      String[] levels = ArrayUtils.subarray(namespace.levels(), 0, level + 1);
+      NameIdentifier nameIdentifier = NameIdentifier.of(levels);
       switch (level) {
         case 0:
-          parentEntityId = MetalakeMetaService.getInstance().getMetalakeIdByName(name);
-          builder.withMetalakeId(parentEntityId);
+          entityId =
+              MetalakeMetaService.getInstance().getMetalakeIdByNameIdentifier(nameIdentifier);
+          builder.withMetalakeId(entityId);
           continue;
         case 1:
-          parentEntityId =
-              CatalogMetaService.getInstance()
-                  .getCatalogIdByMetalakeIdAndName(parentEntityId, name);
-          builder.withCatalogId(parentEntityId);
+          entityId = CatalogMetaService.getInstance().getCatalogIdByNameIdentifier(nameIdentifier);
+          builder.withCatalogId(entityId);
           continue;
         case 2:
-          parentEntityId =
-              SchemaMetaService.getInstance().getSchemaIdByCatalogIdAndName(parentEntityId, name);
-          builder.withSchemaId(parentEntityId);
+          entityId = SchemaMetaService.getInstance().getSchemaIdByNameIdentifier(nameIdentifier);
+          builder.withSchemaId(entityId);
           break;
       }
     }