[#12452] improvement(core): add OCC for catalog writes

core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java

@@ -984,6 +984,12 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
 
           } catch (NoSuchMetalakeException | NoSuchCatalogException ignored) {
             return false;
+          } catch (NoSuchEntityException ignored) {
+            // Another server may have deleted the catalog after it was loaded but before this
+            // transaction reached the compare-and-set delete. Preserve the idempotent drop
+            // contract and discard the now-stale local cache entry.
+            catalogCache.invalidate(ident);
+            return false;
           } catch (GravitinoRuntimeException e) {
             throw e;
           } catch (Exception e) {

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

@@ -358,7 +358,9 @@ public boolean dropMetalake(NameIdentifier ident, boolean force)
             }
 
             return store.delete(ident, EntityType.METALAKE, true);
-          } catch (NoSuchMetalakeException e) {
+          } catch (NoSuchMetalakeException | NoSuchEntityException e) {
+            // Another server may have completed the drop after the initial existence check.
+            // Dropping an already-removed metalake remains an idempotent false result.
             return false;
 
           } catch (IOException e) {

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

@@ -47,6 +47,12 @@ public interface CatalogMetaMapper {
   @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByMetalakeId")
   List<CatalogPO> listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId);
 
+  /** Selects and locks all active catalogs in a metalake for the current transaction. */
+  @SelectProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "listCatalogPOsByMetalakeIdForUpdate")
+  List<CatalogPO> listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId);
+
   @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "listCatalogPOsByCatalogIds")
   List<CatalogPO> listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds);
 
@@ -73,6 +79,12 @@ CatalogPO selectCatalogMetaByName(
   @SelectProvider(type = CatalogMetaSQLProviderFactory.class, method = "selectCatalogMetaById")
   CatalogPO selectCatalogMetaById(@Param("catalogId") Long catalogId);
 
+  /** Selects and locks an active catalog by ID for the current transaction. */
+  @SelectProvider(
+      type = CatalogMetaSQLProviderFactory.class,
+      method = "selectCatalogMetaByIdForUpdate")
+  CatalogPO selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId);
+
   @InsertProvider(type = CatalogMetaSQLProviderFactory.class, method = "insertCatalogMeta")
   void insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO);
 
@@ -89,12 +101,18 @@ Integer updateCatalogMeta(
   @UpdateProvider(
       type = CatalogMetaSQLProviderFactory.class,
       method = "softDeleteCatalogMetasByCatalogId")
-  Integer softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId);
-
+  Integer softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion);
+
+  /**
+   * Soft-deletes catalogs whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
   @UpdateProvider(
       type = CatalogMetaSQLProviderFactory.class,
-      method = "softDeleteCatalogMetasByMetalakeId")
-  Integer softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
+      method = "softDeleteCatalogMetasWithVersion")
+  Integer softDeleteCatalogMetasWithVersion(@Param("catalogMetas") List<CatalogPO> catalogPOs);
 
   @DeleteProvider(
       type = CatalogMetaSQLProviderFactory.class,

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

@@ -61,6 +61,11 @@ public static String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metala
     return getProvider().listCatalogPOsByMetalakeId(metalakeId);
   }
 
+  /** Returns SQL that lists and locks all active catalogs in a metalake. */
+  public static String listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId) {
+    return getProvider().listCatalogPOsByMetalakeIdForUpdate(metalakeId);
+  }
+
   public static String listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds) {
     return getProvider().listCatalogPOsByCatalogIds(catalogIds);
   }
@@ -94,6 +99,11 @@ public static String selectCatalogMetaById(@Param("catalogId") Long catalogId) {
     return getProvider().selectCatalogMetaById(catalogId);
   }
 
+  /** Returns SQL that selects and locks an active catalog by ID. */
+  public static String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId) {
+    return getProvider().selectCatalogMetaByIdForUpdate(catalogId);
+  }
+
   public static String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) {
     return getProvider().insertCatalogMeta(catalogPO);
   }
@@ -109,12 +119,15 @@ public static String updateCatalogMeta(
     return getProvider().updateCatalogMeta(newCatalogPO, oldCatalogPO);
   }
 
-  public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) {
-    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId);
+  public static String softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId, currentVersion);
   }
 
-  public static String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteCatalogMetasByMetalakeId(metalakeId);
+  /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */
+  public static String softDeleteCatalogMetasWithVersion(
+      @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+    return getProvider().softDeleteCatalogMetasWithVersion(catalogPOs);
   }
 
   public static String deleteCatalogMetasByLegacyTimeline(

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

@@ -47,6 +47,18 @@ public interface MetalakeMetaMapper {
   @SelectProvider(type = MetalakeMetaSQLProviderFactory.class, method = "selectMetalakeMetaById")
   MetalakePO selectMetalakeMetaById(@Param("metalakeId") Long metalakeId);
 
+  /** Selects and locks an active metalake by ID for the current transaction. */
+  @SelectProvider(
+      type = MetalakeMetaSQLProviderFactory.class,
+      method = "selectMetalakeMetaByIdForUpdate")
+  MetalakePO selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId);
+
+  /** Selects and share-locks an active metalake by ID for the current transaction. */
+  @SelectProvider(
+      type = MetalakeMetaSQLProviderFactory.class,
+      method = "selectMetalakeMetaByIdForShare")
+  MetalakePO selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId);
+
   @SelectProvider(
       type = MetalakeMetaSQLProviderFactory.class,
       method = "listMetalakePOsByMetalakeIds")
@@ -73,7 +85,8 @@ Integer updateMetalakeMeta(
   @UpdateProvider(
       type = MetalakeMetaSQLProviderFactory.class,
       method = "softDeleteMetalakeMetaByMetalakeId")
-  Integer softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId);
+  Integer softDeleteMetalakeMetaByMetalakeId(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion);
 
   @DeleteProvider(
       type = MetalakeMetaSQLProviderFactory.class,

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

@@ -51,7 +51,13 @@ public static MetalakeMetaBaseSQLProvider getProvider() {
 
   static class MetalakeMetaMySQLProvider extends MetalakeMetaBaseSQLProvider {}
 
-  static class MetalakeMetaH2Provider extends MetalakeMetaBaseSQLProvider {}
+  static class MetalakeMetaH2Provider extends MetalakeMetaBaseSQLProvider {
+    @Override
+    public String selectMetalakeMetaByIdForShare(Long metalakeId) {
+      // H2 has no shared row-lock syntax, so use an exclusive lock in tests.
+      return selectMetalakeMetaByIdForUpdate(metalakeId);
+    }
+  }
 
   public String listMetalakePOs() {
     return getProvider().listMetalakePOs();
@@ -65,6 +71,16 @@ public static String selectMetalakeMetaById(@Param("metalakeId") Long metalakeId
     return getProvider().selectMetalakeMetaById(metalakeId);
   }
 
+  /** Returns SQL that selects and locks an active metalake by ID. */
+  public static String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId) {
+    return getProvider().selectMetalakeMetaByIdForUpdate(metalakeId);
+  }
+
+  /** Returns SQL that selects and share-locks an active metalake by ID. */
+  public static String selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId) {
+    return getProvider().selectMetalakeMetaByIdForShare(metalakeId);
+  }
+
   public static String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) {
     return getProvider().selectMetalakeIdMetaByName(metalakeName);
   }
@@ -88,8 +104,9 @@ public static String updateMetalakeMeta(
     return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO);
   }
 
-  public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId);
+  public static String softDeleteMetalakeMetaByMetalakeId(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId, currentVersion);
   }
 
   public static String deleteMetalakeMetasByLegacyTimeline(

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

@@ -42,6 +42,10 @@ public interface SchemaMetaMapper {
   @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByCatalogId")
   List<SchemaPO> listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId);
 
+  /** Lists all active schemas in a metalake. */
+  @SelectProvider(type = SchemaMetaSQLProviderFactory.class, method = "listSchemaPOsByMetalakeId")
+  List<SchemaPO> listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId);
+
   @SelectProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "listSchemaPOsByFullQualifiedName")
@@ -107,15 +111,15 @@ Integer updateSchemaMeta(
       method = "softDeleteSchemaMetasBySchemaIds")
   Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
+  /**
+   * Soft-deletes schemas whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
-      method = "softDeleteSchemaMetasByMetalakeId")
-  Integer softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
-
-  @UpdateProvider(
-      type = SchemaMetaSQLProviderFactory.class,
-      method = "softDeleteSchemaMetasByCatalogId")
-  Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId);
+      method = "softDeleteSchemaMetasWithVersion")
+  Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List<SchemaPO> schemaPOs);
 
   @DeleteProvider(
       type = SchemaMetaSQLProviderFactory.class,

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

@@ -72,6 +72,11 @@ public static String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId
     return getProvider().listSchemaPOsByCatalogId(catalogId);
   }
 
+  /** Returns SQL that lists all active schemas in a metalake. */
+  public static String listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
+    return getProvider().listSchemaPOsByMetalakeId(metalakeId);
+  }
+
   public static String selectSchemaIdByCatalogIdAndName(
       @Param("catalogId") Long catalogId, @Param("schemaName") String name) {
     return getProvider().selectSchemaIdByCatalogIdAndName(catalogId, name);
@@ -120,12 +125,10 @@ public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<L
     return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds);
   }
 
-  public static String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteSchemaMetasByMetalakeId(metalakeId);
-  }
-
-  public static String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) {
-    return getProvider().softDeleteSchemaMetasByCatalogId(catalogId);
+  /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */
+  public static String softDeleteSchemaMetasWithVersion(
+      @Param("schemaMetas") List<SchemaPO> schemaPOs) {
+    return getProvider().softDeleteSchemaMetasWithVersion(schemaPOs);
   }
 
   public static String deleteSchemaMetasByLegacyTimeline(

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

@@ -53,6 +53,11 @@ public String listCatalogPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
         + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that lists and locks all active catalogs in a metalake. */
+  public String listCatalogPOsByMetalakeIdForUpdate(@Param("metalakeId") Long metalakeId) {
+    return listCatalogPOsByMetalakeId(metalakeId) + " FOR UPDATE";
+  }
+
   public String listCatalogPOsByCatalogIds(@Param("catalogIds") List<Long> catalogIds) {
     return "<script>"
         + "SELECT catalog_id as catalogId, catalog_name as catalogName,"
@@ -137,6 +142,11 @@ public String selectCatalogMetaById(@Param("catalogId") Long catalogId) {
         + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that selects and locks an active catalog by ID. */
+  public String selectCatalogMetaByIdForUpdate(@Param("catalogId") Long catalogId) {
+    return selectCatalogMetaById(catalogId) + " FOR UPDATE";
+  }
+
   public String insertCatalogMeta(@Param("catalogMeta") CatalogPO catalogPO) {
     return "INSERT INTO "
         + TABLE_NAME
@@ -206,33 +216,33 @@ public String updateCatalogMeta(
         + " last_version = #{newCatalogMeta.lastVersion},"
         + " deleted_at = #{newCatalogMeta.deletedAt}"
         + " WHERE catalog_id = #{oldCatalogMeta.catalogId}"
-        + " AND catalog_name = #{oldCatalogMeta.catalogName}"
-        + " AND metalake_id = #{oldCatalogMeta.metalakeId}"
-        + " AND type = #{oldCatalogMeta.type}"
-        + " AND provider = #{oldCatalogMeta.provider}"
-        + " AND (catalog_comment = #{oldCatalogMeta.catalogComment} "
-        + "   OR (catalog_comment IS NULL and #{oldCatalogMeta.catalogComment} IS NULL))"
-        + " AND properties = #{oldCatalogMeta.properties}"
-        + " AND audit_info = #{oldCatalogMeta.auditInfo}"
         + " AND current_version = #{oldCatalogMeta.currentVersion}"
-        + " AND last_version = #{oldCatalogMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 
-  public String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) {
+  public String softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
     return "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+        + " WHERE catalog_id = #{catalogId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
-  public String softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return "UPDATE "
+  /** Returns SQL that soft-deletes catalogs using identifier-and-version pairs. */
+  public String softDeleteCatalogMetasWithVersion(
+      @Param("catalogMetas") List<CatalogPO> catalogPOs) {
+    return "<script>"
+        + "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+        + " WHERE deleted_at = 0 AND "
+        + "<foreach collection='catalogMetas' item='item' separator=' OR ' open='(' close=')'>"
+        + "(catalog_id = #{item.catalogId} AND current_version = #{item.currentVersion})"
+        + "</foreach>"
+        + "</script>";
   }
 
   public String deleteCatalogMetasByLegacyTimeline(

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

@@ -59,6 +59,16 @@ public String selectMetalakeMetaById(@Param("metalakeId") Long metalakeId) {
         + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that selects and locks an active metalake by ID. */
+  public String selectMetalakeMetaByIdForUpdate(@Param("metalakeId") Long metalakeId) {
+    return selectMetalakeMetaById(metalakeId) + " FOR UPDATE";
+  }
+
+  /** Returns SQL that selects and share-locks an active metalake by ID. */
+  public String selectMetalakeMetaByIdForShare(@Param("metalakeId") Long metalakeId) {
+    return selectMetalakeMetaById(metalakeId) + " LOCK IN SHARE MODE";
+  }
+
   public String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) {
     return "SELECT metalake_id as metalakeId"
         + " FROM "
@@ -143,23 +153,18 @@ public String updateMetalakeMeta(
         + " current_version = #{newMetalakeMeta.currentVersion},"
         + " last_version = #{newMetalakeMeta.lastVersion}"
         + " WHERE metalake_id = #{oldMetalakeMeta.metalakeId}"
-        + " AND metalake_name = #{oldMetalakeMeta.metalakeName}"
-        + " AND (metalake_comment = #{oldMetalakeMeta.metalakeComment} "
-        + "  OR (metalake_comment IS NULL and #{oldMetalakeMeta.metalakeComment} IS NULL))"
-        + " AND properties = #{oldMetalakeMeta.properties}"
-        + " AND audit_info = #{oldMetalakeMeta.auditInfo}"
-        + " AND schema_version = #{oldMetalakeMeta.schemaVersion}"
         + " AND current_version = #{oldMetalakeMeta.currentVersion}"
-        + " AND last_version = #{oldMetalakeMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 
-  public String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) {
+  public String softDeleteMetalakeMetaByMetalakeId(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
     return "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+        + " WHERE metalake_id = #{metalakeId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
   public String deleteMetalakeMetasByLegacyTimeline(