[#12451] improvement(core): add OCC for metalake writes

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);
 
@@ -91,10 +97,15 @@ Integer updateCatalogMeta(
       method = "softDeleteCatalogMetasByCatalogId")
   Integer softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId);
 
+  /**
+   * 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);
   }
@@ -113,8 +118,10 @@ public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long
     return getProvider().softDeleteCatalogMetasByCatalogId(catalogId);
   }
 
-  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,12 @@ 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);
+
   @SelectProvider(
       type = MetalakeMetaSQLProviderFactory.class,
       method = "listMetalakePOsByMetalakeIds")
@@ -73,7 +79,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

@@ -65,6 +65,11 @@ 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);
+  }
+
   public static String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) {
     return getProvider().selectMetalakeIdMetaByName(metalakeName);
   }
@@ -88,8 +93,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")
@@ -109,13 +113,18 @@ Integer updateSchemaMeta(
 
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
-      method = "softDeleteSchemaMetasByMetalakeId")
-  Integer softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
+      method = "softDeleteSchemaMetasByCatalogId")
+  Integer softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId);
 
+  /**
+   * Soft-deletes schemas whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
   @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,14 +125,16 @@ 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(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit) {
     return getProvider().deleteSchemaMetasByLegacyTimeline(legacyTimeline, limit);

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,"
@@ -227,12 +232,19 @@ public String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalog
         + " WHERE catalog_id = #{catalogId} 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,11 @@ 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";
+  }
+
   public String selectMetalakeIdMetaByName(@Param("metalakeName") String metalakeName) {
     return "SELECT metalake_id as metalakeId"
         + " FROM "
@@ -143,23 +148,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(

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

@@ -38,6 +38,18 @@ public String listSchemaPOsByCatalogId(@Param("catalogId") Long catalogId) {
         + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
+  /** Returns SQL that lists all active schemas in a metalake. */
+  public String listSchemaPOsByMetalakeId(@Param("metalakeId") Long metalakeId) {
+    return "SELECT schema_id as schemaId, schema_name as schemaName,"
+        + " metalake_id as metalakeId, catalog_id as catalogId,"
+        + " schema_comment as schemaComment, properties, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + TABLE_NAME
+        + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+  }
+
   public String listSchemaPOsByFullQualifiedName(
       @Param("metalakeName") String metalakeName, @Param("catalogName") String catalogName) {
     return """
@@ -299,20 +311,26 @@ public String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> sc
         + "</script>";
   }
 
-  public String softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
+  public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) {
     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 catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
-  public String softDeleteSchemaMetasByCatalogId(@Param("catalogId") Long catalogId) {
-    return "UPDATE "
+  /** Returns SQL that soft-deletes schemas using identifier-and-version pairs. */
+  public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List<SchemaPO> schemaPOs) {
+    return "<script>"
+        + "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 deleted_at = 0 AND "
+        + "<foreach collection='schemaMetas' item='item' separator=' OR ' open='(' close=')'>"
+        + "(schema_id = #{item.schemaId} AND current_version = #{item.currentVersion})"
+        + "</foreach>"
+        + "</script>";
   }
 
   public String deleteSchemaMetasByLegacyTimeline(

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java

@@ -20,6 +20,7 @@
 
 import static org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper.TABLE_NAME;
 
+import java.util.List;
 import org.apache.gravitino.storage.relational.mapper.provider.base.CatalogMetaBaseSQLProvider;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.ibatis.annotations.Param;
@@ -33,12 +34,18 @@ public String softDeleteCatalogMetasByCatalogId(Long catalogId) {
         + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
+  /** {@inheritDoc} */
   @Override
-  public String softDeleteCatalogMetasByMetalakeId(Long metalakeId) {
-    return "UPDATE "
+  public String softDeleteCatalogMetasWithVersion(List<CatalogPO> catalogPOs) {
+    return "<script>"
+        + "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
-        + " 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>";
   }
 
   @Override

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/MetalakeMetaPostgreSQLProvider.java

@@ -26,11 +26,12 @@
 
 public class MetalakeMetaPostgreSQLProvider extends MetalakeMetaBaseSQLProvider {
   @Override
-  public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId) {
+  public String softDeleteMetalakeMetaByMetalakeId(Long metalakeId, Long currentVersion) {
     return "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
-        + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+        + " WHERE metalake_id = #{metalakeId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
   @Override
@@ -75,15 +76,7 @@ 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 (CAST(metalake_comment AS VARCHAR) IS NULL AND "
-        + "  CAST(#{oldMetalakeMeta.metalakeComment} AS VARCHAR) 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";
   }
 

core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java

@@ -126,19 +126,25 @@ public String softDeleteSchemaMetasBySchemaIds(List<Long> schemaIds) {
   }
 
   @Override
-  public String softDeleteSchemaMetasByMetalakeId(Long metalakeId) {
+  public String softDeleteSchemaMetasByCatalogId(Long catalogId) {
     return "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
-        + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0";
+        + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
   }
 
+  /** {@inheritDoc} */
   @Override
-  public String softDeleteSchemaMetasByCatalogId(Long catalogId) {
-    return "UPDATE "
+  public String softDeleteSchemaMetasWithVersion(List<SchemaPO> schemaPOs) {
+    return "<script>"
+        + "UPDATE "
         + TABLE_NAME
         + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
-        + " WHERE catalog_id = #{catalogId} AND deleted_at = 0";
+        + " WHERE deleted_at = 0 AND "
+        + "<foreach collection='schemaMetas' item='item' separator=' OR ' open='(' close=')'>"
+        + "(schema_id = #{item.schemaId} AND current_version = #{item.currentVersion})"
+        + "</foreach>"
+        + "</script>";
   }
 
   @Override