[#9429] improvement(core): Use join to improve query performance on table, topic and so on.

core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java

@@ -20,14 +20,20 @@
 
 import static org.apache.gravitino.Metalake.PROPERTY_IN_USE;
 
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
 import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.Closeable;
 import java.io.IOException;
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
@@ -65,6 +71,19 @@ public class MetalakeManager implements MetalakeDispatcher, Closeable {
 
   private final IdGenerator idGenerator;
 
+  private static final Cache<NameIdentifier, BaseMetalake> metalakeCache =
+      Caffeine.newBuilder()
+          .expireAfterAccess(10 * 60 * 1000 /*10 minutes*/, TimeUnit.MILLISECONDS)
+          .scheduler(
+              Scheduler.forScheduledExecutorService(
+                  new ScheduledThreadPoolExecutor(
+                      1,
+                      new ThreadFactoryBuilder()
+                          .setDaemon(true)
+                          .setNameFormat("metalake-cleaner-%d")
+                          .build())))
+          .build();
+
   @Override
   public void close() {
     // do nothing
@@ -84,7 +103,8 @@ public MetalakeManager(EntityStore store, IdGenerator idGenerator) {
     // directly without list/get metalake first.
     BaseMetalake[] baseMetalakes = listMetalakes();
     for (BaseMetalake baseMetalake : baseMetalakes) {
-      loadMetalake(baseMetalake.nameIdentifier());
+      BaseMetalake newBaseMetalake = loadMetalake(baseMetalake.nameIdentifier());
+      metalakeCache.put(baseMetalake.nameIdentifier(), newBaseMetalake);
     }
   }
 
@@ -115,18 +135,24 @@ public static void checkMetalake(NameIdentifier ident, EntityStore store)
    */
   public static boolean metalakeInUse(EntityStore store, NameIdentifier ident)
       throws NoSuchMetalakeException {
-    try {
-      BaseMetalake metalake = store.get(ident, EntityType.METALAKE, BaseMetalake.class);
-      return (boolean)
-          metalake.propertiesMetadata().getOrDefault(metalake.properties(), PROPERTY_IN_USE);
-    } catch (NoSuchEntityException e) {
-      LOG.warn("Metalake {} does not exist", ident, e);
-      throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, ident);
-
-    } catch (IOException e) {
-      LOG.error("Failed to do store operation", e);
-      throw new RuntimeException(e);
-    }
+
+    BaseMetalake metalake =
+        metalakeCache.get(
+            ident,
+            key -> {
+              try {
+                return store.get(key, EntityType.METALAKE, BaseMetalake.class);
+              } catch (NoSuchEntityException e) {
+                LOG.warn("Metalake {} does not exist", key, e);
+                throw new RuntimeException(e);
+              } catch (IOException e) {
+                LOG.error("Failed to do store operation", e);
+                throw new RuntimeException(e);
+              }
+            });
+
+    return (boolean)
+        metalake.propertiesMetadata().getOrDefault(metalake.properties(), PROPERTY_IN_USE);
   }
 
   /**
@@ -165,16 +191,21 @@ public BaseMetalake loadMetalake(NameIdentifier ident) throws NoSuchMetalakeExce
         ident,
         LockType.READ,
         () -> {
-          try {
-            BaseMetalake baseMetalake = store.get(ident, EntityType.METALAKE, BaseMetalake.class);
-            return newMetalakeWithResolvedProperties(baseMetalake);
-          } catch (NoSuchEntityException e) {
-            LOG.warn("Metalake {} does not exist", ident, e);
-            throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, ident);
-          } catch (IOException ioe) {
-            LOG.error("Loading Metalake {} failed due to storage issues", ident, ioe);
-            throw new RuntimeException(ioe);
-          }
+          BaseMetalake baseMetalake =
+              metalakeCache.get(
+                  ident,
+                  key -> {
+                    try {
+                      return store.get(key, EntityType.METALAKE, BaseMetalake.class);
+                    } catch (NoSuchEntityException e) {
+                      LOG.warn("Metalake {} does not exist", key, e);
+                      throw new RuntimeException(e);
+                    } catch (IOException e) {
+                      LOG.error("Failed to do store operation", e);
+                      throw new RuntimeException(e);
+                    }
+                  });
+          return newMetalakeWithResolvedProperties(baseMetalake);
         });
   }
 
@@ -267,6 +298,8 @@ public BaseMetalake alterMetalake(NameIdentifier ident, MetalakeChange... change
                   "Metalake %s is not in use, please enable it first", ident);
             }
 
+            metalakeCache.invalidate(ident);
+
             return store.update(
                 ident,
                 BaseMetalake.class,
@@ -317,6 +350,8 @@ public boolean dropMetalake(NameIdentifier ident, boolean force)
                   "Metalake %s is in use, please disable it first or use force option", ident);
             }
 
+            metalakeCache.invalidate(ident);
+
             List<CatalogEntity> catalogEntities =
                 store.list(Namespace.of(ident.name()), CatalogEntity.class, EntityType.CATALOG);
             if (!catalogEntities.isEmpty() && !force) {
@@ -343,6 +378,7 @@ public void enableMetalake(NameIdentifier ident) throws NoSuchMetalakeException
           try {
             boolean inUse = metalakeInUse(store, ident);
             if (!inUse) {
+              metalakeCache.invalidate(ident);
               store.update(
                   ident,
                   BaseMetalake.class,
@@ -377,6 +413,7 @@ public void disableMetalake(NameIdentifier ident) throws NoSuchMetalakeException
           try {
             boolean inUse = metalakeInUse(store, ident);
             if (inUse) {
+              metalakeCache.invalidate(ident);
               store.update(
                   ident,
                   BaseMetalake.class,

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

@@ -56,6 +56,15 @@ Long selectTableIdBySchemaIdAndName(
   TablePO selectTableMetaBySchemaIdAndName(
       @Param("schemaId") Long schemaId, @Param("tableName") String name);
 
+  @SelectProvider(
+      type = TableMetaSQLProviderFactory.class,
+      method = "selectTableByFullQualifiedName")
+  TablePO selectTableByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("tableName") String tableName);
+
   @InsertProvider(type = TableMetaSQLProviderFactory.class, method = "insertTableMeta")
   void insertTableMeta(@Param("tableMeta") TablePO tablePO);
 

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

@@ -69,6 +69,15 @@ public static String selectTableMetaBySchemaIdAndName(
     return getProvider().selectTableMetaBySchemaIdAndName(schemaId, name);
   }
 
+  public static String selectTableByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("tableName") String tableName) {
+    return getProvider()
+        .selectTableByFullQualifiedName(metalakeName, catalogName, schemaName, tableName);
+  }
+
   public static String selectTableMetaById(@Param("tableId") Long tableId) {
     return getProvider().selectTableMetaById(tableId);
   }

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

@@ -21,6 +21,9 @@
 import static org.apache.gravitino.storage.relational.mapper.TableMetaMapper.TABLE_NAME;
 
 import java.util.List;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.TableVersionMapper;
 import org.apache.gravitino.storage.relational.po.TablePO;
 import org.apache.ibatis.annotations.Param;
@@ -230,4 +233,61 @@ public String deleteTableMetasByLegacyTimeline(
         + TABLE_NAME
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
+
+  public String selectTableByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("tableName") String tableName) {
+    return """
+            SELECT
+                tm.table_id AS tableId,
+                tm.table_name AS tableName,
+                tm.metalake_id AS metalakeId,
+                tm.catalog_id AS catalogId,
+                tm.schema_id AS schemaId,
+                tm.audit_info AS auditInfo,
+                tm.current_version AS currentVersion,
+                tm.last_version AS lastVersion,
+                tm.deleted_at AS deletedAt,
+                tvi.format AS format,
+                tvi.properties AS properties,
+                tvi.partitioning AS partitions,
+                tvi.sort_orders AS sortOrders,
+                tvi.distribution AS distribution,
+                tvi.indexes AS indexes,
+                tvi.comment AS comment
+            FROM
+                %s mm
+            INNER JOIN
+                %s cm ON mm.metalake_id = cm.metalake_id
+            INNER JOIN
+                %s sm ON cm.catalog_id = sm.catalog_id
+            INNER JOIN
+                %s tm ON sm.schema_id = tm.schema_id
+            LEFT JOIN
+                %s tvi ON tm.table_id = tvi.table_id
+                AND tm.current_version = tvi.version
+                AND tvi.deleted_at = 0
+            WHERE
+                mm.metalake_name = '%s'
+                AND mm.deleted_at = 0
+                AND cm.catalog_name = '%s'
+                AND cm.deleted_at = 0
+                AND sm.schema_name = '%s'
+                AND sm.deleted_at = 0
+                AND tm.table_name = '%s'
+                AND tm.deleted_at = 0;
+            """
+        .formatted(
+            MetalakeMetaMapper.TABLE_NAME,
+            CatalogMetaMapper.TABLE_NAME,
+            SchemaMetaMapper.TABLE_NAME,
+            TABLE_NAME,
+            TableVersionMapper.TABLE_NAME,
+            metalakeName,
+            catalogName,
+            schemaName,
+            tableName);
+  }
 }

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

@@ -85,11 +85,13 @@ public Long getTableIdBySchemaIdAndName(Long schemaId, String tableName) {
   public TableEntity getTableByIdentifier(NameIdentifier identifier) {
     NameIdentifierUtil.checkTable(identifier);
 
-    Long schemaId =
-        EntityIdService.getEntityId(
-            NameIdentifier.of(identifier.namespace().levels()), Entity.EntityType.SCHEMA);
+    String[] namespaceLevels = identifier.namespace().levels();
+    String metalakeName = namespaceLevels[0];
+    String catalogName = namespaceLevels[1];
+    String schemaName = namespaceLevels[2];
+    String tableName = identifier.name();
+    TablePO tablePO = getTableByFullQualifiedName(metalakeName, catalogName, schemaName, tableName);
 
-    TablePO tablePO = getTablePOBySchemaIdAndName(schemaId, identifier.name());
     List<ColumnPO> columnPOs =
         TableColumnMetaService.getInstance()
             .getColumnsByTableIdAndVersion(tablePO.getTableId(), tablePO.getCurrentVersion());
@@ -342,4 +344,21 @@ private TablePO getTablePOBySchemaIdAndName(Long schemaId, String tableName) {
     }
     return tablePO;
   }
+
+  private TablePO getTableByFullQualifiedName(
+      String metalakeName, String catalogName, String schemaName, String tableName) {
+    TablePO tablePO =
+        SessionUtils.getWithoutCommit(
+            TableMetaMapper.class,
+            mapper ->
+                mapper.selectTableByFullQualifiedName(
+                    metalakeName, catalogName, schemaName, tableName));
+    if (tablePO == null) {
+      throw new NoSuchEntityException(
+          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+          Entity.EntityType.TABLE.name().toLowerCase(),
+          tableName);
+    }
+    return tablePO;
+  }
 }

scripts/h2/schema-1.1.0-h2.sql

@@ -28,6 +28,7 @@ CREATE TABLE IF NOT EXISTS `metalake_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'metalake last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metalake deleted at',
     PRIMARY KEY (metalake_id),
+    KEY idx_name_da (metalake_name, deleted_at),
     CONSTRAINT uk_mn_del UNIQUE (metalake_name, deleted_at)
     ) ENGINE = InnoDB;
 
@@ -45,6 +46,7 @@ CREATE TABLE IF NOT EXISTS `catalog_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'catalog last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'catalog deleted at',
     PRIMARY KEY (catalog_id),
+    KEY idx_name_da (catalog_name, deleted_at),
     CONSTRAINT uk_mid_cn_del UNIQUE (metalake_id, catalog_name, deleted_at)
     ) ENGINE=InnoDB;
 
@@ -61,6 +63,7 @@ CREATE TABLE IF NOT EXISTS `schema_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'schema last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'schema deleted at',
     PRIMARY KEY (schema_id),
+    KEY idx_name_da (catalog_name, deleted_at),
     CONSTRAINT uk_cid_sn_del UNIQUE (catalog_id, schema_name, deleted_at),
     -- Aliases are used here, and indexes with the same name in H2 can only be created once.
     KEY idx_smid (metalake_id)
@@ -77,6 +80,7 @@ CREATE TABLE IF NOT EXISTS `table_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'table last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'table deleted at',
     PRIMARY KEY (table_id),
+    KEY idx_name_da (table_name, deleted_at),
     CONSTRAINT uk_sid_tn_del UNIQUE (schema_id, table_name, deleted_at),
     -- Aliases are used here, and indexes with the same name in H2 can only be created once.
     KEY idx_tmid (metalake_id),

scripts/h2/upgrade-1.0.0-to-1.1.0-h2.sql

@@ -30,3 +30,9 @@ CREATE TABLE IF NOT EXISTS `table_version_info` (
     `deleted_at`      BIGINT(20) UNSIGNED DEFAULT 0 COMMENT 'table deletion timestamp, 0 means not deleted',
     UNIQUE KEY `uk_table_id_version_deleted_at` (`table_id`, `version`, `deleted_at`)
 ) ENGINE=InnoDB COMMENT 'table detail information including format, location, properties, partition, distribution, sort order, index and so on';
+
+alter table `metalake_meta` add index idx_name_da (metalake_name, deleted_at);
+alter table `catalog_meta` add index idx_name_da (catalog_name, deleted_at);
+alter table `schema_meta` add index idx_name_da (schema_name, deleted_at);
+alter table `table_meta` add index idx_name_da (table_name, deleted_at);
+

scripts/mysql/schema-1.1.0-mysql.sql

@@ -28,6 +28,7 @@ CREATE TABLE IF NOT EXISTS `metalake_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'metalake last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'metalake deleted at',
     PRIMARY KEY (`metalake_id`),
+    KEY idx_name_da (metalake_name, deleted_at),
     UNIQUE KEY `uk_mn_del` (`metalake_name`, `deleted_at`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'metalake metadata';
 
@@ -44,6 +45,7 @@ CREATE TABLE IF NOT EXISTS `catalog_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'catalog last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'catalog deleted at',
     PRIMARY KEY (`catalog_id`),
+    KEY idx_name_da (catalog_name, deleted_at),
     UNIQUE KEY `uk_mid_cn_del` (`metalake_id`, `catalog_name`, `deleted_at`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'catalog metadata';
 
@@ -59,6 +61,7 @@ CREATE TABLE IF NOT EXISTS `schema_meta` (
     `last_version` INT UNSIGNED NOT NULL DEFAULT 1 COMMENT 'schema last version',
     `deleted_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'schema deleted at',
     PRIMARY KEY (`schema_id`),
+    KEY idx_name_da (schema_name, deleted_at),
     UNIQUE KEY `uk_cid_sn_del` (`catalog_id`, `schema_name`, `deleted_at`),
     KEY `idx_mid` (`metalake_id`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'schema metadata';
@@ -77,6 +80,7 @@ CREATE TABLE IF NOT EXISTS `table_meta` (
     UNIQUE KEY `uk_sid_tn_del` (`schema_id`, `table_name`, `deleted_at`),
     KEY `idx_mid` (`metalake_id`),
     KEY `idx_cid` (`catalog_id`)
+    KEY idx_name_da (table_name, deleted_at),
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table metadata';
 
 CREATE TABLE IF NOT EXISTS `table_column_version_info` (

scripts/mysql/upgrade-1.0.0-to-1.1.0-mysql.sql

@@ -30,3 +30,8 @@ CREATE TABLE IF NOT EXISTS `table_version_info` (
     `deleted_at`      BIGINT(20) UNSIGNED DEFAULT 0 COMMENT 'table deletion timestamp, 0 means not deleted',
     UNIQUE KEY `uk_table_id_version_deleted_at` (`table_id`, `version`, `deleted_at`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table detail information including format, location, properties, partition, distribution, sort order, index and so on';
+
+alter table `metalake_meta` add index idx_name_da (metalake_name, deleted_at);
+alter table `catalog_meta` add index idx_name_da (catalog_name, deleted_at);
+alter table `schema_meta` add index idx_name_da (schema_name, deleted_at);
+alter table `table_meta` add index idx_name_da (table_name, deleted_at);