[#11103] feat(core): Add the hierarchy convention layer

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

@@ -325,7 +325,7 @@ public static void setUp() throws IllegalAccessException {
     MockedStatic<CatalogMetaService> catalogMetaServiceMockedStatic =
         Mockito.mockStatic(CatalogMetaService.class);
     MockedStatic<SchemaMetaService> schemaMetaServiceMockedStatic =
-        Mockito.mockStatic(SchemaMetaService.class);
+        Mockito.mockStatic(SchemaMetaService.class, Mockito.CALLS_REAL_METHODS);
 
     metalakeMetaServiceMockedStatic
         .when(MetalakeMetaService::getInstance)

core/src/main/java/org/apache/gravitino/SupportsRelationOperations.java

@@ -165,6 +165,34 @@ void insertRelation(
       boolean override)
       throws IOException;
 
+  /**
+   * Batch inserts the same relation from many source entities to one destination. Parameter order
+   * matches {@link #insertRelation}. All sources share one {@code srcType}. Semantics match
+   * repeated {@link #insertRelation} with the same {@code relType}; a relational backend may use
+   * fewer round-trips for some relation types (e.g. {@link Type#OWNER_REL}).
+   *
+   * @param relType the relation type (e.g. {@link Type#OWNER_REL})
+   * @param srcIdentifiers identifiers of the source side for each relation row
+   * @param srcType entity type shared by every identifier in {@code srcIdentifiers}
+   * @param dstIdentifier destination entity identifier (shared by all rows)
+   * @param dstType destination entity type
+   * @param override if true, replace existing relations of each source entity first, per {@link
+   *     #insertRelation}
+   * @throws IOException if the storage operation fails
+   */
+  default void batchInsertRelations(
+      Type relType,
+      List<NameIdentifier> srcIdentifiers,
+      Entity.EntityType srcType,
+      NameIdentifier dstIdentifier,
+      Entity.EntityType dstType,
+      boolean override)
+      throws IOException {
+    for (NameIdentifier srcIdent : srcIdentifiers) {
+      insertRelation(relType, srcIdent, srcType, dstIdentifier, dstType, override);
+    }
+  }
+
   /**
    * Updates the relations for a given entity by adding a set of new relations and removing another
    * set of relations.

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

@@ -733,6 +733,29 @@ public void insertRelation(
     }
   }
 
+  @Override
+  public void batchInsertRelations(
+      Type relType,
+      List<NameIdentifier> srcIdentifiers,
+      Entity.EntityType srcType,
+      NameIdentifier dstIdentifier,
+      Entity.EntityType dstType,
+      boolean override)
+      throws IOException {
+    if (srcIdentifiers == null || srcIdentifiers.isEmpty()) {
+      return;
+    }
+    switch (relType) {
+      case OWNER_REL:
+        OwnerMetaService.getInstance()
+            .batchSetOwners(srcIdentifiers, srcType, dstIdentifier, dstType);
+        break;
+      default:
+        throw new IllegalArgumentException(
+            String.format("Doesn't support batch insert for the relation type %s", relType));
+    }
+  }
+
   @Override
   public <E extends Entity & HasIdentifier> List<E> updateEntityRelations(
       Type relType,

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

@@ -55,7 +55,7 @@
 /**
  * Relation store to store entities. This means we can store entities in a relational store. I.e.,
  * MySQL, PostgreSQL, etc. If you want to use a different backend, you can implement the {@link
- * RelationalBackend} interface
+ * RelationalBackend} interface. The default JDBC backend is {@link JDBCBackend}.
  */
 public class RelationalEntityStore implements EntityStore, SupportsRelationOperations {
   private static final Logger LOGGER = LoggerFactory.getLogger(RelationalEntityStore.class);
@@ -327,6 +327,26 @@ public void insertRelation(
     cache.invalidate(dstIdentifier, dstType, relType);
   }
 
+  @Override
+  public void batchInsertRelations(
+      Type relType,
+      List<NameIdentifier> srcIdentifiers,
+      Entity.EntityType srcType,
+      NameIdentifier dstIdentifier,
+      Entity.EntityType dstType,
+      boolean override)
+      throws IOException {
+    if (srcIdentifiers == null || srcIdentifiers.isEmpty()) {
+      return;
+    }
+    backend.batchInsertRelations(
+        relType, srcIdentifiers, srcType, dstIdentifier, dstType, override);
+    for (NameIdentifier ident : srcIdentifiers) {
+      cache.invalidate(ident, srcType, relType);
+    }
+    cache.invalidate(dstIdentifier, dstType, relType);
+  }
+
   @Override
   public <E extends Entity & HasIdentifier> List<E> updateEntityRelations(
       Type relType,

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

@@ -0,0 +1,81 @@
+/*
+ * 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.service;
+
+import java.util.List;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+
+public abstract class BasePOStorageOps<PO, Mapper> {
+  public void insertPO(Mapper mapper, PO po, boolean overwrite) {
+    throw new UnsupportedOperationException(
+        "insertPO is not supported by " + getClass().getSimpleName());
+  }
+
+  public void batchInsertPOs(Mapper mapper, List<PO> pos, boolean overwrite) {
+    throw new UnsupportedOperationException(
+        "batchInsertPOs is not supported by " + getClass().getSimpleName());
+  }
+
+  public Integer updatePO(Mapper mapper, PO newPO, PO oldPO) {
+    throw new UnsupportedOperationException(
+        "updatePO is not supported by " + getClass().getSimpleName());
+  }
+
+  /**
+   * When {@code true} and the entity-id cache is enabled, callers may resolve rows by parent entity
+   * id plus short name via {@link #getPO} and {@link #listPOs}; see {@link POStorageReadRouting}.
+   */
+  public boolean supportsParentIdRelationalRead() {
+    return false;
+  }
+
+  public PO getPO(Mapper mapper, Long parentId, String name) {
+    throw new UnsupportedOperationException(
+        "getPO by parent id is not supported by " + getClass().getSimpleName());
+  }
+
+  public List<PO> listPOs(Mapper mapper, Long parentId) {
+    throw new UnsupportedOperationException(
+        "listPOs by parent id is not supported by " + getClass().getSimpleName());
+  }
+
+  public List<PO> listPOs(Mapper mapper, Namespace namespace, List<String> names) {
+    throw new UnsupportedOperationException(
+        "listPOs by namespace and names is not supported by " + getClass().getSimpleName());
+  }
+
+  public List<PO> listPOs(Mapper mapper, List<Long> entityIds) {
+    throw new UnsupportedOperationException(
+        "listPOs by entityIds is not supported by " + getClass().getSimpleName());
+  }
+
+  public PO getPOByFullName(Mapper mapper, NameIdentifier identifier) {
+    throw new UnsupportedOperationException(
+        "getPOByFullName is not supported by " + getClass().getSimpleName());
+  }
+
+  public List<PO> listPOsByNSFullName(Mapper mapper, Namespace namespace) {
+    throw new UnsupportedOperationException(
+        "listPOsByNSFullName is not supported by " + getClass().getSimpleName());
+  }
+
+  protected abstract Entity.EntityType entityType();
+}

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

@@ -33,7 +33,6 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.gravitino.Entity;
-import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.HasIdentifier;
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.NameIdentifier;
@@ -56,14 +55,17 @@
 import org.slf4j.LoggerFactory;
 
 public class FunctionMetaService {
+  private static final Logger LOG = LoggerFactory.getLogger(FunctionMetaService.class);
+  private static final FunctionMetaService INSTANCE = new FunctionMetaService();
+  private BasePOStorageOps<FunctionPO, FunctionMetaMapper> ops;
+
   public static FunctionMetaService getInstance() {
     return INSTANCE;
   }
 
-  private static final Logger LOG = LoggerFactory.getLogger(FunctionMetaService.class);
-  private static final FunctionMetaService INSTANCE = new FunctionMetaService();
-
-  private FunctionMetaService() {}
+  private FunctionMetaService() {
+    this.ops = new HierarchicalConventionPOStorageOps<>(new FunctionPOStorageOps());
+  }
 
   @Monitored(
       metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
@@ -87,18 +89,17 @@ public FunctionEntity getFunctionByIdentifier(NameIdentifier ident) {
       metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
       baseMetricName = "getFunctionIdBySchemaIdAndFunctionName")
   public Long getFunctionIdBySchemaIdAndFunctionName(Long schemaId, String functionName) {
-    Long functionId =
+    FunctionPO functionPO =
         SessionUtils.getWithoutCommit(
-            FunctionMetaMapper.class,
-            mapper -> mapper.selectFunctionIdBySchemaIdAndFunctionName(schemaId, functionName));
+            FunctionMetaMapper.class, mapper -> ops.getPO(mapper, schemaId, functionName));
 
-    if (functionId == null) {
+    if (functionPO == null) {
       throw new NoSuchEntityException(
           NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
           Entity.EntityType.FUNCTION.name().toLowerCase(Locale.ROOT),
           functionName);
     }
-    return functionId;
+    return functionPO.functionId();
   }
 
   @Monitored(
@@ -115,14 +116,7 @@ public void insertFunction(FunctionEntity functionEntity, boolean overwrite) thr
       SessionUtils.doMultipleWithCommit(
           () ->
               SessionUtils.doWithoutCommit(
-                  FunctionMetaMapper.class,
-                  mapper -> {
-                    if (overwrite) {
-                      mapper.insertFunctionMetaOnDuplicateKeyUpdate(po);
-                    } else {
-                      mapper.insertFunctionMeta(po);
-                    }
-                  }),
+                  FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)),
           () ->
               SessionUtils.doWithoutCommit(
                   FunctionVersionMetaMapper.class,
@@ -231,8 +225,18 @@ public int deleteFunctionVersionsByRetentionCount(Long versionRetentionCount, in
       baseMetricName = "getFunctionPOByIdentifier")
   FunctionPO getFunctionPOByIdentifier(NameIdentifier ident) {
     NameIdentifierUtil.checkFunction(ident);
+    FunctionPO functionPO =
+        SessionUtils.getWithoutCommit(
+            FunctionMetaMapper.class,
+            mapper -> POStorageReadRouting.getPO(mapper, ident, ops, Entity.EntityType.FUNCTION));
 
-    return functionPOFetcher().apply(ident);
+    if (functionPO == null) {
+      throw new NoSuchEntityException(
+          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+          Entity.EntityType.FUNCTION.name().toLowerCase(Locale.ROOT),
+          ident.name());
+    }
+    return functionPO;
   }
 
   @Monitored(
@@ -260,7 +264,7 @@ public <E extends Entity & HasIdentifier> FunctionEntity updateFunction(
           () ->
               SessionUtils.doWithoutCommit(
                   FunctionMetaMapper.class,
-                  mapper -> mapper.updateFunctionMeta(newFunctionPO, oldFunctionPO)));
+                  mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO)));
 
       return newEntity;
     } catch (RuntimeException re) {
@@ -270,89 +274,10 @@ public <E extends Entity & HasIdentifier> FunctionEntity updateFunction(
     }
   }
 
-  private Function<NameIdentifier, FunctionPO> functionPOFetcher() {
-    return GravitinoEnv.getInstance().cacheEnabled()
-        ? this::getFunctionPOBySchemaId
-        : this::getFunctionPOByFullQualifiedName;
-  }
-
-  private FunctionPO getFunctionPOBySchemaId(NameIdentifier ident) {
-    Long schemaId =
-        EntityIdService.getEntityId(
-            NameIdentifier.of(ident.namespace().levels()), Entity.EntityType.SCHEMA);
-
-    FunctionPO functionPO =
-        SessionUtils.getWithoutCommit(
-            FunctionMetaMapper.class,
-            mapper -> mapper.selectFunctionMetaBySchemaIdAndName(schemaId, ident.name()));
-
-    if (functionPO == null) {
-      throw new NoSuchEntityException(
-          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
-          Entity.EntityType.FUNCTION.name().toLowerCase(Locale.ROOT),
-          ident.toString());
-    }
-    return functionPO;
-  }
-
-  private FunctionPO getFunctionPOByFullQualifiedName(NameIdentifier ident) {
-    String[] namespaceLevels = ident.namespace().levels();
-    FunctionPO functionPO =
-        SessionUtils.getWithoutCommit(
-            FunctionMetaMapper.class,
-            mapper ->
-                mapper.selectFunctionMetaByFullQualifiedName(
-                    namespaceLevels[0], namespaceLevels[1], namespaceLevels[2], ident.name()));
-
-    if (functionPO == null || functionPO.functionId() == null) {
-      throw new NoSuchEntityException(
-          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
-          Entity.EntityType.FUNCTION.name().toLowerCase(Locale.ROOT),
-          ident.name());
-    }
-
-    if (functionPO.schemaId() == null) {
-      throw new NoSuchEntityException(
-          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
-          Entity.EntityType.SCHEMA.name().toLowerCase(),
-          namespaceLevels[2]);
-    }
-    return functionPO;
-  }
-
   private List<FunctionPO> listFunctionPOs(Namespace namespace) {
-    return functionListFetcher().apply(namespace);
-  }
-
-  private List<FunctionPO> listFunctionPOsBySchemaId(Namespace namespace) {
-    Long schemaId =
-        EntityIdService.getEntityId(
-            NameIdentifier.of(namespace.levels()), Entity.EntityType.SCHEMA);
     return SessionUtils.getWithoutCommit(
-        FunctionMetaMapper.class, mapper -> mapper.listFunctionPOsBySchemaId(schemaId));
-  }
-
-  private Function<Namespace, List<FunctionPO>> functionListFetcher() {
-    return GravitinoEnv.getInstance().cacheEnabled()
-        ? this::listFunctionPOsBySchemaId
-        : this::listFunctionPOsByFullQualifiedName;
-  }
-
-  private List<FunctionPO> listFunctionPOsByFullQualifiedName(Namespace namespace) {
-    String[] namespaceLevels = namespace.levels();
-    List<FunctionPO> functionPOs =
-        SessionUtils.getWithoutCommit(
-            FunctionMetaMapper.class,
-            mapper ->
-                mapper.listFunctionPOsByFullQualifiedName(
-                    namespaceLevels[0], namespaceLevels[1], namespaceLevels[2]));
-    if (functionPOs.isEmpty() || functionPOs.get(0).schemaId() == null) {
-      throw new NoSuchEntityException(
-          NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
-          Entity.EntityType.SCHEMA.name().toLowerCase(),
-          namespaceLevels[2]);
-    }
-    return functionPOs.stream().filter(po -> po.functionId() != null).collect(Collectors.toList());
+        FunctionMetaMapper.class,
+        mapper -> POStorageReadRouting.listPOs(mapper, namespace, ops, Entity.EntityType.FUNCTION));
   }
 
   private void fillFunctionPOBuilderParentEntityId(