[#12343] improvement(core): add OCC for users and groups

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

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
@@ -116,9 +117,11 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
                     .build())
             .build();
     try {
-      store().put(schemaEntity, true /* overwrite */);
+      store().put(schemaEntity, false /* overwrite */);
     } catch (IOException ioe) {
       throw new RuntimeException("Failed to create schema " + ident, ioe);
+    } catch (EntityAlreadyExistsException e) {
+      throw new SchemaAlreadyExistsException(e, "Schema %s already exists", ident);
     } catch (NoSuchEntityException e) {
       throw new NoSuchCatalogException(e, "Catalog %s does not exist", ident.namespace());
     }

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

@@ -86,10 +86,30 @@ Integer updateCatalogMeta(
       @Param("newCatalogMeta") CatalogPO newCatalogPO,
       @Param("oldCatalogMeta") CatalogPO oldCatalogPO);
 
+  /**
+   * Advances the catalog version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = CatalogMetaSQLProviderFactory.class, method = "fenceCatalogMeta")
+  Integer fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion);
+
   @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 = "softDeleteCatalogMetasWithVersion")
+  Integer softDeleteCatalogMetasWithVersion(@Param("catalogMetas") List<CatalogPO> catalogPOs);
 
   @UpdateProvider(
       type = CatalogMetaSQLProviderFactory.class,

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

@@ -109,8 +109,21 @@ public static String updateCatalogMeta(
     return getProvider().updateCatalogMeta(newCatalogPO, oldCatalogPO);
   }
 
-  public static String softDeleteCatalogMetasByCatalogId(@Param("catalogId") Long catalogId) {
-    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId);
+  /** Returns SQL that advances a catalog OCC version conditionally. */
+  public static String fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceCatalogMeta(catalogId, currentVersion);
+  }
+
+  public static String softDeleteCatalogMetasByCatalogId(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteCatalogMetasByCatalogId(catalogId, currentVersion);
+  }
+
+  /** 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 softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {

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

@@ -76,7 +76,8 @@ List<ExtendedGroupPO> listExtendedGroupPOsByMetalakeIdAndNames(
   void insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") GroupPO groupPO);
 
   @UpdateProvider(type = GroupMetaSQLProviderFactory.class, method = "softDeleteGroupMetaByGroupId")
-  void softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId);
+  Integer softDeleteGroupMetaByGroupId(
+      @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion);
 
   @UpdateProvider(
       type = GroupMetaSQLProviderFactory.class,

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

@@ -72,8 +72,9 @@ public static String insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") Gro
     return getProvider().insertGroupMetaOnDuplicateKeyUpdate(groupPO);
   }
 
-  public static String softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId) {
-    return getProvider().softDeleteGroupMetaByGroupId(groupId);
+  public static String softDeleteGroupMetaByGroupId(
+      @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteGroupMetaByGroupId(groupId, currentVersion);
   }
 
   public static String softDeleteGroupMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {

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

@@ -70,10 +70,20 @@ Integer updateMetalakeMeta(
       @Param("newMetalakeMeta") MetalakePO newMetalakePO,
       @Param("oldMetalakeMeta") MetalakePO oldMetalakePO);
 
+  /**
+   * Advances the metalake version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = MetalakeMetaSQLProviderFactory.class, method = "fenceMetalakeMeta")
+  Integer fenceMetalakeMeta(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion);
+
   @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

@@ -88,8 +88,15 @@ public static String updateMetalakeMeta(
     return getProvider().updateMetalakeMeta(newMetalakePO, oldMetalakePO);
   }
 
-  public static String softDeleteMetalakeMetaByMetalakeId(@Param("metalakeId") Long metalakeId) {
-    return getProvider().softDeleteMetalakeMetaByMetalakeId(metalakeId);
+  /** Returns SQL that advances a metalake OCC version conditionally. */
+  public static String fenceMetalakeMeta(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceMetalakeMeta(metalakeId, currentVersion);
+  }
+
+  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

@@ -102,11 +102,36 @@ SchemaPO selectSchemaByFullQualifiedName(
   Integer updateSchemaMeta(
       @Param("newSchemaMeta") SchemaPO newSchemaPO, @Param("oldSchemaMeta") SchemaPO oldSchemaPO);
 
+  /**
+   * Advances the schema version when the expected OCC version still matches.
+   *
+   * @return the number of updated rows
+   */
+  @UpdateProvider(type = SchemaMetaSQLProviderFactory.class, method = "fenceSchemaMeta")
+  Integer fenceSchemaMeta(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion);
+
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "softDeleteSchemaMetasBySchemaIds")
   Integer softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
+  @UpdateProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "softDeleteSchemaMetaBySchemaIdAndVersion")
+  Integer softDeleteSchemaMetaBySchemaIdAndVersion(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion);
+
+  /**
+   * Soft-deletes schemas whose identifiers and OCC versions still match.
+   *
+   * @return the number of deleted rows
+   */
+  @UpdateProvider(
+      type = SchemaMetaSQLProviderFactory.class,
+      method = "softDeleteSchemaMetasWithVersion")
+  Integer softDeleteSchemaMetasWithVersion(@Param("schemaMetas") List<SchemaPO> schemaPOs);
+
   @UpdateProvider(
       type = SchemaMetaSQLProviderFactory.class,
       method = "softDeleteSchemaMetasByMetalakeId")

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

@@ -116,10 +116,27 @@ public static String updateSchemaMeta(
     return getProvider().updateSchemaMeta(newSchemaPO, oldSchemaPO);
   }
 
+  /** Returns SQL that advances a schema OCC version conditionally. */
+  public static String fenceSchemaMeta(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().fenceSchemaMeta(schemaId, currentVersion);
+  }
+
   public static String softDeleteSchemaMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds) {
     return getProvider().softDeleteSchemaMetasBySchemaIds(schemaIds);
   }
 
+  public static String softDeleteSchemaMetaBySchemaIdAndVersion(
+      @Param("schemaId") Long schemaId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteSchemaMetaBySchemaIdAndVersion(schemaId, currentVersion);
+  }
+
+  /** 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 softDeleteSchemaMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return getProvider().softDeleteSchemaMetasByMetalakeId(metalakeId);
   }

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

@@ -71,7 +71,8 @@ UserPO selectUserMetaByMetalakeIdAndName(
   void insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserPO userPO);
 
   @UpdateProvider(type = UserMetaSQLProviderFactory.class, method = "softDeleteUserMetaByUserId")
-  void softDeleteUserMetaByUserId(@Param("userId") Long userId);
+  Integer softDeleteUserMetaByUserId(
+      @Param("userId") Long userId, @Param("currentVersion") Long currentVersion);
 
   @UpdateProvider(
       type = UserMetaSQLProviderFactory.class,

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

@@ -69,8 +69,9 @@ public static String insertUserMetaOnDuplicateKeyUpdate(@Param("userMeta") UserP
     return getProvider().insertUserMetaOnDuplicateKeyUpdate(userPO);
   }
 
-  public static String softDeleteUserMetaByUserId(@Param("userId") Long userId) {
-    return getProvider().softDeleteUserMetaByUserId(userId);
+  public static String softDeleteUserMetaByUserId(
+      @Param("userId") Long userId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteUserMetaByUserId(userId, currentVersion);
   }
 
   public static String softDeleteUserMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {

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

@@ -206,25 +206,43 @@ 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) {
+  /** Returns SQL that advances a catalog OCC version conditionally. */
+  public String fenceCatalogMeta(
+      @Param("catalogId") Long catalogId, @Param("currentVersion") Long currentVersion) {
+    return "UPDATE "
+        + TABLE_NAME
+        + " SET last_version = current_version + 1, current_version = current_version + 1"
+        + " WHERE catalog_id = #{catalogId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
+  }
+
+  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";
+  }
+
+  /** 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 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 softDeleteCatalogMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {

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

@@ -201,12 +201,14 @@ public String insertGroupMetaOnDuplicateKeyUpdate(@Param("groupMeta") GroupPO gr
         + " deleted_at = #{groupMeta.deletedAt}";
   }
 
-  public String softDeleteGroupMetaByGroupId(@Param("groupId") Long groupId) {
+  public String softDeleteGroupMetaByGroupId(
+      @Param("groupId") Long groupId, @Param("currentVersion") Long currentVersion) {
     return "UPDATE "
         + GROUP_TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE group_id = #{groupId} AND deleted_at = 0";
+        + " WHERE group_id = #{groupId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
   public String softDeleteGroupMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
@@ -229,11 +231,7 @@ public String updateGroupMeta(
         + " last_version = #{newGroupMeta.lastVersion},"
         + " deleted_at = #{newGroupMeta.deletedAt}"
         + " WHERE group_id = #{oldGroupMeta.groupId}"
-        + " AND group_name = #{oldGroupMeta.groupName}"
-        + " AND metalake_id = #{oldGroupMeta.metalakeId}"
-        + " AND audit_info = #{oldGroupMeta.auditInfo}"
         + " AND current_version = #{oldGroupMeta.currentVersion}"
-        + " AND last_version = #{oldGroupMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 

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

@@ -143,23 +143,28 @@ 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) {
+  /** Returns SQL that advances a metalake OCC version conditionally. */
+  public String fenceMetalakeMeta(
+      @Param("metalakeId") Long metalakeId, @Param("currentVersion") Long currentVersion) {
+    return "UPDATE "
+        + TABLE_NAME
+        + " SET last_version = current_version + 1, current_version = current_version + 1"
+        + " WHERE metalake_id = #{metalakeId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
+  }
+
+  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(