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

catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java

@@ -216,7 +216,7 @@ private static CatalogInfo randomCatalogInfo(
   }
 
   @BeforeAll
-  public static void setUp() {
+  public static void setUp() throws IllegalAccessException {
     Config config = Mockito.mock(Config.class);
     when(config.get(ENTITY_STORE)).thenReturn(RELATIONAL_ENTITY_STORE);
     when(config.get(ENTITY_RELATIONAL_STORE)).thenReturn(DEFAULT_ENTITY_RELATIONAL_STORE);
@@ -336,6 +336,8 @@ public static void setUp() {
     schemaMetaServiceMockedStatic
         .when(SchemaMetaService::getInstance)
         .thenReturn(spySchemaMetaService);
+
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
   }
 
   @AfterAll

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

@@ -54,6 +54,7 @@
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.EntityStoreFactory;
+import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.Schema;
@@ -80,6 +81,7 @@
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
+import org.testcontainers.shaded.org.apache.commons.lang3.reflect.FieldUtils;
 
 public class TestKafkaCatalogOperations extends KafkaClusterEmbedded {
 
@@ -134,7 +136,7 @@ public PropertiesMetadata modelVersionPropertiesMetadata()
   private static KafkaCatalogOperations kafkaCatalogOperations;
 
   @BeforeAll
-  public static void setUp() {
+  public static void setUp() throws IllegalAccessException {
     Config config = Mockito.mock(Config.class);
     Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
     Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 * 1000L);
@@ -210,6 +212,8 @@ public static void setUp() {
                     .build())
             .build();
 
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+
     kafkaCatalogOperations = new KafkaCatalogOperations(store, idGenerator);
     kafkaCatalogOperations.initialize(
         MOCK_CATALOG_PROPERTIES, kafkaCatalogEntity.toCatalogInfo(), KAFKA_PROPERTIES_METADATA);

catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java

@@ -48,6 +48,7 @@
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.EntityStoreFactory;
+import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.Schema;
@@ -67,6 +68,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import org.testcontainers.shaded.org.apache.commons.lang3.reflect.FieldUtils;
 
 public class TestGenericCatalogOperations {
   private static final String STORE_PATH =
@@ -79,7 +81,7 @@ public class TestGenericCatalogOperations {
   private static GenericCatalogOperations ops;
 
   @BeforeAll
-  public static void setUp() throws IOException {
+  public static void setUp() throws IOException, IllegalAccessException {
     Config config = Mockito.mock(Config.class);
     when(config.get(ENTITY_STORE)).thenReturn(RELATIONAL_ENTITY_STORE);
     when(config.get(ENTITY_RELATIONAL_STORE)).thenReturn(DEFAULT_ENTITY_RELATIONAL_STORE);
@@ -132,6 +134,8 @@ public static void setUp() throws IOException {
             .build();
     store.put(catalog, false);
 
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+
     ops = new GenericCatalogOperations(store, idGenerator);
   }
 

catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java

@@ -52,6 +52,7 @@
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.EntityStoreFactory;
+import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.Schema;
@@ -83,6 +84,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import org.testcontainers.shaded.org.apache.commons.lang3.reflect.FieldUtils;
 
 public class TestModelCatalogOperations {
 
@@ -100,7 +102,7 @@ public class TestModelCatalogOperations {
   private static ModelCatalogOperations ops;
 
   @BeforeAll
-  public static void setUp() throws IOException {
+  public static void setUp() throws IOException, IllegalAccessException {
     Config config = Mockito.mock(Config.class);
     when(config.get(ENTITY_STORE)).thenReturn(RELATIONAL_ENTITY_STORE);
     when(config.get(ENTITY_RELATIONAL_STORE)).thenReturn(DEFAULT_ENTITY_RELATIONAL_STORE);
@@ -158,6 +160,8 @@ public static void setUp() throws IOException {
             .build();
     store.put(catalog, false);
 
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+
     ops = new ModelCatalogOperations(store);
     ops.initialize(
         Collections.emptyMap(),

common/src/main/java/org/apache/gravitino/config/ConfigConstants.java

@@ -83,6 +83,9 @@ private ConfigConstants() {}
   /** The version number for the 1.1.0 release. */
   public static final String VERSION_1_1_0 = "1.1.0";
 
+  /** The version number for the 1.2.0 release. */
+  public static final String VERSION_1_2_0 = "1.2.0";
+
   /** The current version of backend storage initialization script. */
-  public static final String CURRENT_SCRIPT_VERSION = VERSION_1_1_0;
+  public static final String CURRENT_SCRIPT_VERSION = VERSION_1_2_0;
 }

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

@@ -85,6 +85,33 @@ public interface FilesetMetaMapper {
   @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "listFilesetPOsBySchemaId")
   List<FilesetPO> listFilesetPOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @Results({
+    @Result(property = "filesetId", column = "fileset_id", id = true),
+    @Result(property = "filesetName", column = "fileset_name"),
+    @Result(property = "metalakeId", column = "metalake_id"),
+    @Result(property = "catalogId", column = "catalog_id"),
+    @Result(property = "schemaId", column = "schema_id"),
+    @Result(property = "type", column = "type"),
+    @Result(property = "auditInfo", column = "audit_info"),
+    @Result(property = "currentVersion", column = "current_version"),
+    @Result(property = "lastVersion", column = "last_version"),
+    @Result(property = "deletedAt", column = "deleted_at"),
+    @Result(
+        property = "filesetVersionPOs",
+        javaType = List.class,
+        column =
+            "{id,version_metalake_id,version_catalog_id,version_schema_id,version_fileset_id,version,"
+                + "fileset_comment,properties,storage_location_name,storage_location,version_deleted_at}",
+        many = @Many(resultMap = "mapToFilesetVersionPO"))
+  })
+  @SelectProvider(
+      type = FilesetMetaSQLProviderFactory.class,
+      method = "listFilesetPOsByFullQualifiedName")
+  List<FilesetPO> listFilesetPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName);
+
   @Results({
     @Result(property = "filesetId", column = "fileset_id", id = true),
     @Result(property = "filesetName", column = "fileset_name"),
@@ -160,6 +187,34 @@ FilesetPO selectFilesetMetaBySchemaIdAndName(
   @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "selectFilesetMetaById")
   FilesetPO selectFilesetMetaById(@Param("filesetId") Long filesetId);
 
+  @Results({
+    @Result(property = "filesetId", column = "fileset_id", id = true),
+    @Result(property = "filesetName", column = "fileset_name"),
+    @Result(property = "metalakeId", column = "metalake_id"),
+    @Result(property = "catalogId", column = "catalog_id"),
+    @Result(property = "schemaId", column = "schema_id"),
+    @Result(property = "type", column = "type"),
+    @Result(property = "auditInfo", column = "audit_info"),
+    @Result(property = "currentVersion", column = "current_version"),
+    @Result(property = "lastVersion", column = "last_version"),
+    @Result(property = "deletedAt", column = "deleted_at"),
+    @Result(
+        property = "filesetVersionPOs",
+        javaType = List.class,
+        column =
+            "{id,version_metalake_id,version_catalog_id,version_schema_id,version_fileset_id,version,"
+                + "fileset_comment,properties,storage_location_name,storage_location,version_deleted_at}",
+        many = @Many(resultMap = "mapToFilesetVersionPO"))
+  })
+  @SelectProvider(
+      type = FilesetMetaSQLProviderFactory.class,
+      method = "selectFilesetByFullQualifiedName")
+  FilesetPO selectFilesetByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("filesetName") String filesetName);
+
   @InsertProvider(type = FilesetMetaSQLProviderFactory.class, method = "insertFilesetMeta")
   void insertFilesetMeta(@Param("filesetMeta") FilesetPO filesetPO);
 

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

@@ -56,6 +56,13 @@ public static String listFilesetPOsBySchemaId(@Param("schemaId") Long schemaId)
     return getProvider().listFilesetPOsBySchemaId(schemaId);
   }
 
+  public static String listFilesetPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName) {
+    return getProvider().listFilesetPOsByFullQualifiedName(metalakeName, catalogName, schemaName);
+  }
+
   public static String listFilesetPOsByFilesetIds(@Param("filesetIds") List<Long> filesetIds) {
     return getProvider().listFilesetPOsByFilesetIds(filesetIds);
   }
@@ -74,6 +81,15 @@ public static String selectFilesetMetaById(@Param("filesetId") Long filesetId) {
     return getProvider().selectFilesetMetaById(filesetId);
   }
 
+  public static String selectFilesetByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("filesetName") String filesetName) {
+    return getProvider()
+        .selectFilesetByFullQualifiedName(metalakeName, catalogName, schemaName, filesetName);
+  }
+
   public static String insertFilesetMeta(@Param("filesetMeta") FilesetPO filesetPO) {
     return getProvider().insertFilesetMeta(filesetPO);
   }

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

@@ -40,6 +40,14 @@ public interface ModelMetaMapper {
   @SelectProvider(type = ModelMetaSQLProviderFactory.class, method = "listModelPOsBySchemaId")
   List<ModelPO> listModelPOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @SelectProvider(
+      type = ModelMetaSQLProviderFactory.class,
+      method = "listModelPOsByFullQualifiedName")
+  List<ModelPO> listModelPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName);
+
   @SelectProvider(type = ModelMetaSQLProviderFactory.class, method = "listModelPOsByModelIds")
   List<ModelPO> listModelPOsByModelIds(@Param("modelIds") List<Long> modelIds);
 
@@ -49,6 +57,15 @@ public interface ModelMetaMapper {
   ModelPO selectModelMetaBySchemaIdAndModelName(
       @Param("schemaId") Long schemaId, @Param("modelName") String modelName);
 
+  @SelectProvider(
+      type = ModelMetaSQLProviderFactory.class,
+      method = "selectModelByFullQualifiedName")
+  ModelPO selectModelByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("modelName") String modelName);
+
   @SelectProvider(
       type = ModelMetaSQLProviderFactory.class,
       method = "selectModelIdBySchemaIdAndModelName")

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

@@ -63,6 +63,13 @@ public static String listModelPOsBySchemaId(@Param("schemaId") Long schemaId) {
     return getProvider().listModelPOsBySchemaId(schemaId);
   }
 
+  public static String listModelPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName) {
+    return getProvider().listModelPOsByFullQualifiedName(metalakeName, catalogName, schemaName);
+  }
+
   public static String listModelPOsByModelIds(@Param("modelIds") List<Long> modelIds) {
     return getProvider().listModelPOsByModelIds(modelIds);
   }
@@ -72,6 +79,15 @@ public static String selectModelMetaBySchemaIdAndModelName(
     return getProvider().selectModelMetaBySchemaIdAndModelName(schemaId, modelName);
   }
 
+  public static String selectModelByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName,
+      @Param("modelName") String modelName) {
+    return getProvider()
+        .selectModelByFullQualifiedName(metalakeName, catalogName, schemaName, modelName);
+  }
+
   public static String selectModelIdBySchemaIdAndModelName(
       @Param("schemaId") Long schemaId, @Param("modelName") String modelName) {
     return getProvider().selectModelIdBySchemaIdAndModelName(schemaId, modelName);

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

@@ -42,6 +42,12 @@ public interface SchemaMetaMapper {
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByCatalogId")
   List<SchemaPO> listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId);
 
+  @SelectProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "listSchemaPOsByFullQualifiedName")
+  List<SchemaPO> listSchemaPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName);
+
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsBySchemaIds")
   List<SchemaPO> listSchemaPOsBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
@@ -57,6 +63,14 @@ Long selectSchemaIdByCatalogIdAndName(
   SchemaPO selectSchemaMetaByCatalogIdAndName(
       @Param("catalogId") Long catalogId, @Param("schemaName") String name);
 
+  @SelectProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "selectSchemaByFullQualifiedName")
+  SchemaPO selectSchemaByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName);
+
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectSchemaMetaById")
   SchemaPO selectSchemaMetaById(@Param("schemaId") Long schemaId);
 

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

@@ -51,6 +51,11 @@ static class SchemaMetaMySQLProvider extends SchemaMetaBaseSQLProvider {}
 
   static class SchemaMetaH2Provider extends SchemaMetaBaseSQLProvider {}
 
+  public static String listSchemaPOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) {
+    return getProvider().listSchemaPOsByFullQualifiedName(metalakeName, catalogName);
+  }
+
   public static String listSchemaPOsBySchemaIds(@Param("schemaIds") List<Long> schemaIds) {
     return getProvider().listSchemaPOsBySchemaIds(schemaIds);
   }
@@ -69,6 +74,13 @@ public static String selectSchemaMetaByCatalogIdAndName(
     return getProvider().selectSchemaMetaByCatalogIdAndName(catalogId, name);
   }
 
+  public static String selectSchemaByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName) {
+    return getProvider().selectSchemaByFullQualifiedName(metalakeName, catalogName, schemaName);
+  }
+
   public static String selectSchemaMetaById(@Param("schemaId") Long schemaId) {
     return getProvider().selectSchemaMetaById(schemaId);
   }

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

@@ -41,6 +41,14 @@ public interface TableMetaMapper {
   @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "listTablePOsBySchemaId")
   List<TablePO> listTablePOsBySchemaId(@Param("schemaId") Long schemaId);
 
+  @SelectProvider(
+      type = TableMetaSQLProviderFactory.class,
+      method = "listTablePOsByFullQualifiedName")
+  List<TablePO> listTablePOsByFullQualifiedName(
+      @Param("metalakeName") String metalakeName,
+      @Param("catalogName") String catalogName,
+      @Param("schemaName") String schemaName);
+
   @SelectProvider(type = TableMetaSQLProviderFactory.class, method = "listTablePOsByTableIds")
   List<TablePO> listTablePOsByTableIds(@Param("tableIds") List<Long> tableIds);
 
@@ -56,6 +64,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);