[#12576] improvement(core): optimize schema write fencing

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

@@ -86,6 +86,16 @@ SchemaPO selectSchemaByFullQualifiedName(
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectSchemaMetaById")
   SchemaPO selectSchemaMetaById(@Param("schemaId") Long schemaId);
 
+  /**
+   * Returns one when an active table, view, fileset, function, model, or topic exists in the
+   * schema, and {@code null} otherwise.
+   *
+   * <p>Only a literal is selected because callers need an existence answer, not complete child
+   * metadata. The final limit also lets the database stop as soon as it finds the first child.
+   */
+  @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "selectActiveChildBySchemaId")
+  Integer selectActiveChildBySchemaId(@Param("schemaId") Long schemaId);
+
   /** Selects and locks an active schema by ID for the current transaction. */
   @SelectProvider(
       type = SchemaMetaSQLProviderFactory.class,

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

@@ -106,6 +106,11 @@ public static String selectSchemaMetaById(@Param("schemaId") Long schemaId) {
     return getProvider().selectSchemaMetaById(schemaId);
   }
 
+  /** Returns SQL that checks whether an active child exists in the schema. */
+  public static String selectActiveChildBySchemaId(@Param("schemaId") Long schemaId) {
+    return getProvider().selectActiveChildBySchemaId(schemaId);
+  }
+
   /** Returns SQL that selects and locks an active schema by ID. */
   public static String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) {
     return getProvider().selectSchemaMetaByIdForUpdate(schemaId);

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

@@ -22,7 +22,13 @@
 
 import java.util.List;
 import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FilesetMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.FunctionMetaMapper;
 import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TableMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.ViewMetaMapper;
 import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.ibatis.annotations.Param;
 
@@ -182,6 +188,31 @@ public String selectSchemaMetaById(@Param("schemaId") Long schemaId) {
         + " WHERE schema_id = #{schemaId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that checks whether an active child exists in the schema. */
+  public String selectActiveChildBySchemaId(@Param("schemaId") Long schemaId) {
+    // Each branch returns only the same literal, so UNION ALL avoids unnecessary duplicate
+    // elimination. LIMIT 1 lets the database stop as soon as any kind of child is found.
+    return "SELECT 1 FROM "
+        + TableMetaMapper.TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " UNION ALL SELECT 1 FROM "
+        + ViewMetaMapper.TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " UNION ALL SELECT 1 FROM "
+        + FilesetMetaMapper.META_TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " UNION ALL SELECT 1 FROM "
+        + FunctionMetaMapper.TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " UNION ALL SELECT 1 FROM "
+        + ModelMetaMapper.TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " UNION ALL SELECT 1 FROM "
+        + TopicMetaMapper.TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND deleted_at = 0"
+        + " LIMIT 1";
+  }
+
   /** Returns SQL that selects and locks an active schema by ID. */
   public String selectSchemaMetaByIdForUpdate(@Param("schemaId") Long schemaId) {
     return selectSchemaMetaById(schemaId) + " FOR UPDATE";

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

@@ -165,36 +165,33 @@ public void insertFileset(FilesetEntity filesetEntity, boolean overwrite) throws
       FilesetPO po = POConverters.initializeFilesetPOWithVersion(filesetEntity, builder);
 
       // insert both fileset meta table and version table
-      SessionUtils.doMultipleWithCommit(
-          // Hold the parent schema row until this transaction ends, so the fileset cannot be
-          // written below a schema that is being dropped.
-          () ->
-              SchemaMetaService.getInstance()
-                  .lockSchemaForEntityWrite(
-                      filesetEntity.nameIdentifier(),
-                      po.getSchemaId(),
-                      po.getCatalogId(),
-                      po.getMetalakeId()),
-          () ->
-              SessionUtils.doWithoutCommit(
-                  FilesetMetaMapper.class,
-                  mapper -> {
-                    if (overwrite) {
-                      mapper.insertFilesetMetaOnDuplicateKeyUpdate(po);
-                    } else {
-                      mapper.insertFilesetMeta(po);
-                    }
-                  }),
-          () ->
-              SessionUtils.doWithoutCommit(
-                  FilesetVersionMapper.class,
-                  mapper -> {
-                    if (overwrite) {
-                      mapper.insertFilesetVersionsOnDuplicateKeyUpdate(po.getFilesetVersionPOs());
-                    } else {
-                      mapper.insertFilesetVersions(po.getFilesetVersionPOs());
-                    }
-                  }));
+      SchemaMetaService.getInstance()
+          .doWithSchemaWriteLock(
+              filesetEntity.nameIdentifier(),
+              po.getSchemaId(),
+              po.getCatalogId(),
+              po.getMetalakeId(),
+              () ->
+                  SessionUtils.doWithoutCommit(
+                      FilesetMetaMapper.class,
+                      mapper -> {
+                        if (overwrite) {
+                          mapper.insertFilesetMetaOnDuplicateKeyUpdate(po);
+                        } else {
+                          mapper.insertFilesetMeta(po);
+                        }
+                      }),
+              () ->
+                  SessionUtils.doWithoutCommit(
+                      FilesetVersionMapper.class,
+                      mapper -> {
+                        if (overwrite) {
+                          mapper.insertFilesetVersionsOnDuplicateKeyUpdate(
+                              po.getFilesetVersionPOs());
+                        } else {
+                          mapper.insertFilesetVersions(po.getFilesetVersionPOs());
+                        }
+                      }));
     } catch (RuntimeException re) {
       ExceptionUtils.checkSQLException(
           re, Entity.EntityType.FILESET, filesetEntity.nameIdentifier().toString());
@@ -231,22 +228,33 @@ public <E extends Entity & HasIdentifier> FilesetEntity updateFileset(
         // back — including the version insert — and the update is treated as a conflict.
         int[] metaUpdateCountRef = new int[1];
         try {
-          SessionUtils.doMultipleWithCommit(
-              () ->
-                  SessionUtils.doWithoutCommit(
-                      FilesetVersionMapper.class,
-                      mapper -> mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())),
-              () -> {
-                metaUpdateCountRef[0] =
-                    SessionUtils.getWithoutCommit(
-                        FilesetMetaMapper.class,
-                        mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO));
-                if (metaUpdateCountRef[0] == 0) {
-                  throw new RuntimeException("Failed to update the entity: " + identifier);
-                }
-              });
+          SchemaMetaService.getInstance()
+              .doWithSchemaWriteLock(
+                  identifier,
+                  oldFilesetPO.getSchemaId(),
+                  oldFilesetPO.getCatalogId(),
+                  oldFilesetPO.getMetalakeId(),
+                  () ->
+                      SessionUtils.doWithoutCommit(
+                          FilesetVersionMapper.class,
+                          mapper ->
+                              mapper.insertFilesetVersions(newFilesetPO.getFilesetVersionPOs())),
+                  () -> {
+                    metaUpdateCountRef[0] =
+                        SessionUtils.getWithoutCommit(
+                            FilesetMetaMapper.class,
+                            mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO));
+                    if (metaUpdateCountRef[0] == 0) {
+                      throw new RuntimeException("Failed to update the entity: " + identifier);
+                    }
+                  });
           updateResult = 1;
         } catch (RuntimeException re) {
+          // The schema fence runs before the fileset update. Keep its missing-schema error rather
+          // than turning it into a fileset write conflict merely because the update count is zero.
+          if (re instanceof NoSuchEntityException) {
+            throw re;
+          }
           if (metaUpdateCountRef[0] == 0) {
             // The meta update matched no rows; the transaction was rolled back,
             // including the version insert above.
@@ -259,12 +267,17 @@ public <E extends Entity & HasIdentifier> FilesetEntity updateFileset(
         }
       } else {
         int[] metaUpdateCountRef = new int[1];
-        SessionUtils.doMultipleWithCommit(
-            () ->
-                metaUpdateCountRef[0] =
-                    SessionUtils.getWithoutCommit(
-                        FilesetMetaMapper.class,
-                        mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)));
+        SchemaMetaService.getInstance()
+            .doWithSchemaWriteLock(
+                identifier,
+                oldFilesetPO.getSchemaId(),
+                oldFilesetPO.getCatalogId(),
+                oldFilesetPO.getMetalakeId(),
+                () ->
+                    metaUpdateCountRef[0] =
+                        SessionUtils.getWithoutCommit(
+                            FilesetMetaMapper.class,
+                            mapper -> mapper.updateFilesetMeta(newFilesetPO, oldFilesetPO)));
         updateResult = metaUpdateCountRef[0];
       }
     } catch (RuntimeException re) {

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

@@ -114,29 +114,26 @@ public void insertFunction(FunctionEntity functionEntity, boolean overwrite) thr
       fillFunctionPOBuilderParentEntityId(builder, functionEntity.namespace());
       FunctionPO po = initializeFunctionPO(functionEntity, builder);
 
-      SessionUtils.doMultipleWithCommit(
-          // Hold the parent schema row until this transaction ends, so the function cannot be
-          // written below a schema that is being dropped.
-          () ->
-              SchemaMetaService.getInstance()
-                  .lockSchemaForEntityWrite(
-                      functionEntity.nameIdentifier(),
-                      po.schemaId(),
-                      po.catalogId(),
-                      po.metalakeId()),
-          () ->
-              SessionUtils.doWithoutCommit(
-                  FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)),
-          () ->
-              SessionUtils.doWithoutCommit(
-                  FunctionVersionMetaMapper.class,
-                  mapper -> {
-                    if (overwrite) {
-                      mapper.insertFunctionVersionMetaOnDuplicateKeyUpdate(po.functionVersionPO());
-                    } else {
-                      mapper.insertFunctionVersionMeta(po.functionVersionPO());
-                    }
-                  }));
+      SchemaMetaService.getInstance()
+          .doWithSchemaWriteLock(
+              functionEntity.nameIdentifier(),
+              po.schemaId(),
+              po.catalogId(),
+              po.metalakeId(),
+              () ->
+                  SessionUtils.doWithoutCommit(
+                      FunctionMetaMapper.class, mapper -> ops.insertPO(mapper, po, overwrite)),
+              () ->
+                  SessionUtils.doWithoutCommit(
+                      FunctionVersionMetaMapper.class,
+                      mapper -> {
+                        if (overwrite) {
+                          mapper.insertFunctionVersionMetaOnDuplicateKeyUpdate(
+                              po.functionVersionPO());
+                        } else {
+                          mapper.insertFunctionVersionMeta(po.functionVersionPO());
+                        }
+                      }));
     } catch (RuntimeException re) {
       ExceptionUtils.checkSQLException(
           re, Entity.EntityType.FUNCTION, functionEntity.nameIdentifier().toString());
@@ -271,33 +268,29 @@ public <E extends Entity & HasIdentifier> FunctionEntity updateFunction(
     try {
       FunctionPO newFunctionPO = updateFunctionPO(oldFunctionPO, newEntity);
       // Insert a new version and update function meta
-      SessionUtils.doMultipleWithCommit(
-          // The function was read before this transaction started. Lock its observed parent again
-          // before writing, so a schema drop cannot finish its function cleanup and then let this
-          // update add a new version below the deleted schema.
-          () ->
-              SchemaMetaService.getInstance()
-                  .lockSchemaForEntityWrite(
-                      identifier,
-                      oldFunctionPO.schemaId(),
-                      oldFunctionPO.catalogId(),
-                      oldFunctionPO.metalakeId()),
-          () ->
-              SessionUtils.doWithoutCommit(
-                  FunctionVersionMetaMapper.class,
-                  mapper -> mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())),
-          () -> {
-            int updated =
-                SessionUtils.getWithoutCommit(
-                    FunctionMetaMapper.class,
-                    mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO));
-            if (updated == 0) {
-              // The version row was inserted earlier in this transaction. Throwing here rolls the
-              // whole transaction back instead of leaving that version without an active function
-              // metadata row.
-              throw ExceptionUtils.concurrentModification(Entity.EntityType.FUNCTION, identifier);
-            }
-          });
+      SchemaMetaService.getInstance()
+          .doWithSchemaWriteLock(
+              identifier,
+              oldFunctionPO.schemaId(),
+              oldFunctionPO.catalogId(),
+              oldFunctionPO.metalakeId(),
+              () ->
+                  SessionUtils.doWithoutCommit(
+                      FunctionVersionMetaMapper.class,
+                      mapper ->
+                          mapper.insertFunctionVersionMeta(newFunctionPO.functionVersionPO())),
+              () -> {
+                int updated =
+                    SessionUtils.getWithoutCommit(
+                        FunctionMetaMapper.class,
+                        mapper -> ops.updatePO(mapper, newFunctionPO, oldFunctionPO));
+                if (updated == 0) {
+                  // The version was inserted above. Throwing here rolls it back instead of leaving
+                  // an active version without function metadata.
+                  throw ExceptionUtils.concurrentModification(
+                      Entity.EntityType.FUNCTION, identifier);
+                }
+              });
 
       return newEntity;
     } catch (RuntimeException re) {