[#12650] improvement(core): add OCC for fileset writes

catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java

@@ -712,6 +712,14 @@ public boolean dropFileset(NameIdentifier ident) {
       FilesetEntity filesetEntity =
           store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
 
+      // Drop the metadata first. It is the only step that can still be rejected, because the drop
+      // is refused when another writer altered the fileset after the read above. Deleting the files
+      // first would leave the data gone while the rejected drop keeps the fileset row pointing at
+      // storage locations that no longer exist.
+      if (!store.delete(ident, Entity.EntityType.FILESET)) {
+        return false;
+      }
+
       // For managed fileset, we should delete the related files.
       if (!disableFSOps && filesetEntity.filesetType() == Fileset.Type.MANAGED) {
         AtomicReference<IOException> exception = new AtomicReference<>();
@@ -753,7 +761,7 @@ public boolean dropFileset(NameIdentifier ident) {
         }
       }
 
-      return store.delete(ident, Entity.EntityType.FILESET);
+      return true;
     } catch (NoSuchEntityException ne) {
       LOG.warn("Fileset {} does not exist", ident);
       return false;

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

@@ -76,6 +76,7 @@
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Config;
 import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.EntityStoreFactory;
 import org.apache.gravitino.GravitinoEnv;
@@ -105,6 +106,7 @@
 import org.apache.gravitino.exceptions.NoSuchFilesetException;
 import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.exceptions.NonEmptySchemaException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
 import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
 import org.apache.gravitino.file.FileInfo;
 import org.apache.gravitino.file.Fileset;
@@ -3386,6 +3388,41 @@ private static Stream<Arguments> testRenameArguments() {
             TEST_ROOT_PATH + "/fileset39"));
   }
 
+  @Test
+  public void testDropFilesetKeepsFilesWhenMetadataDropIsRejected() throws IOException {
+    String schemaName = "schema_drop_rejected";
+    String filesetName = "fileset_drop_rejected";
+    String catalogPath = TEST_ROOT_PATH + "/catalog_drop_rejected";
+    createSchema(schemaName, "comment", catalogPath, null);
+    Fileset fileset =
+        createFileset(filesetName, schemaName, "comment", Fileset.Type.MANAGED, catalogPath, null);
+
+    Path filesetPath = new Path(fileset.storageLocation());
+    FileSystem fs = filesetPath.getFileSystem(new Configuration());
+    Assertions.assertTrue(fs.exists(filesetPath));
+
+    NameIdentifier filesetIdent = NameIdentifier.of("m1", "c1", schemaName, filesetName);
+    EntityStore rejectingStore = Mockito.spy(store);
+    Mockito.doThrow(new OptimisticLockException("fileset was modified concurrently"))
+        .when(rejectingStore)
+        .delete(filesetIdent, Entity.EntityType.FILESET);
+
+    try (FilesetCatalogOperations ops =
+        new FilesetCatalogOperations(rejectingStore, secretManager)) {
+      ops.initialize(
+          ImmutableMap.of(LOCATION, catalogPath),
+          randomCatalogInfo("m1", "c1"),
+          FILESET_PROPERTIES_METADATA);
+      Assertions.assertThrows(OptimisticLockException.class, () -> ops.dropFileset(filesetIdent));
+    }
+
+    // The drop was refused, so the fileset row still advertises this location. Deleting the files
+    // anyway would leave that row pointing at data that is gone.
+    Assertions.assertTrue(fs.exists(filesetPath));
+
+    fs.delete(filesetPath, true);
+  }
+
   private Schema createSchema(String name, String comment, String catalogPath, String schemaPath)
       throws IOException {
     return createSchema(name, comment, catalogPath, schemaPath, false);

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

@@ -187,6 +187,30 @@ FilesetPO selectFilesetMetaBySchemaIdAndName(
   @SelectProvider(type = FilesetMetaSQLProviderFactory.class, method = "selectFilesetMetaById")
   FilesetPO selectFilesetMetaById(@Param("filesetId") Long filesetId);
 
+  /**
+   * Selects and exclusively locks an active fileset metadata row.
+   *
+   * @param filesetId the fileset ID
+   * @return the active fileset metadata, or {@code null} when it no longer exists
+   */
+  @SelectProvider(
+      type = FilesetMetaSQLProviderFactory.class,
+      method = "selectFilesetMetaByIdForUpdate")
+  FilesetPO selectFilesetMetaByIdForUpdate(@Param("filesetId") Long filesetId);
+
+  /**
+   * Selects an active fileset metadata row by schema and name in the current transaction.
+   *
+   * @param schemaId the schema ID
+   * @param filesetName the fileset name
+   * @return the active fileset metadata, or {@code null} when it does not exist
+   */
+  @SelectProvider(
+      type = FilesetMetaSQLProviderFactory.class,
+      method = "selectFilesetMetaBySchemaIdAndNameForUpdate")
+  FilesetPO selectFilesetMetaBySchemaIdAndNameForUpdate(
+      @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName);
+
   @Results({
     @Result(property = "filesetId", column = "fileset_id", id = true),
     @Result(property = "filesetName", column = "fileset_name"),
@@ -243,10 +267,18 @@ Integer updateFilesetMeta(
       method = "softDeleteFilesetMetasBySchemaIds")
   Integer softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List<Long> schemaIds);
 
+  /**
+   * Soft-deletes a fileset only if its version has not changed since the caller read it.
+   *
+   * @param filesetId the fileset ID
+   * @param currentVersion the version observed by the caller
+   * @return the number of deleted rows; zero means the fileset changed or disappeared
+   */
   @UpdateProvider(
       type = FilesetMetaSQLProviderFactory.class,
       method = "softDeleteFilesetMetasByFilesetId")
-  Integer softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId);
+  Integer softDeleteFilesetMetasByFilesetId(
+      @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion);
 
   @DeleteProvider(
       type = FilesetMetaSQLProviderFactory.class,

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

@@ -81,6 +81,28 @@ public static String selectFilesetMetaById(@Param("filesetId") Long filesetId) {
     return getProvider().selectFilesetMetaById(filesetId);
   }
 
+  /**
+   * Returns SQL that locks an active fileset metadata row by ID.
+   *
+   * @param filesetId the fileset ID
+   * @return the locking select SQL
+   */
+  public static String selectFilesetMetaByIdForUpdate(@Param("filesetId") Long filesetId) {
+    return getProvider().selectFilesetMetaByIdForUpdate(filesetId);
+  }
+
+  /**
+   * Returns SQL that selects an active fileset metadata row by schema and name.
+   *
+   * @param schemaId the schema ID
+   * @param filesetName the fileset name
+   * @return the metadata-only select SQL
+   */
+  public static String selectFilesetMetaBySchemaIdAndNameForUpdate(
+      @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName) {
+    return getProvider().selectFilesetMetaBySchemaIdAndNameForUpdate(schemaId, filesetName);
+  }
+
   public static String selectFilesetByFullQualifiedName(
       @Param("metalakeName") String metalakeName,
       @Param("catalogName") String catalogName,
@@ -117,8 +139,16 @@ public static String softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List<
     return getProvider().softDeleteFilesetMetasBySchemaIds(schemaIds);
   }
 
-  public String softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId) {
-    return getProvider().softDeleteFilesetMetasByFilesetId(filesetId);
+  /**
+   * Returns SQL that soft-deletes a fileset by ID and expected version.
+   *
+   * @param filesetId the fileset ID
+   * @param currentVersion the version observed by the caller
+   * @return the version-checked delete SQL
+   */
+  public static String softDeleteFilesetMetasByFilesetId(
+      @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteFilesetMetasByFilesetId(filesetId, currentVersion);
   }
 
   public String deleteFilesetMetasByLegacyTimeline(

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

@@ -74,6 +74,15 @@ void insertFilesetVersionsOnDuplicateKeyUpdate(
   Integer deleteFilesetVersionsByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
 
+  /**
+   * Returns the highest live version recorded for a fileset, or {@code null} when it has none.
+   *
+   * @param filesetId the fileset whose versions are inspected
+   * @return the highest version still present in the version table
+   */
+  @SelectProvider(type = FilesetVersionSQLProviderFactory.class, method = "selectMaxFilesetVersion")
+  Long selectMaxFilesetVersion(@Param("filesetId") Long filesetId);
+
   @SelectProvider(
       type = FilesetVersionSQLProviderFactory.class,
       method = "selectFilesetVersionsByRetentionCount")

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

@@ -83,6 +83,10 @@ public static String deleteFilesetVersionsByLegacyTimeline(
     return getProvider().deleteFilesetVersionsByLegacyTimeline(legacyTimeline, limit);
   }
 
+  public static String selectMaxFilesetVersion(@Param("filesetId") Long filesetId) {
+    return getProvider().selectMaxFilesetVersion(filesetId);
+  }
+
   public static String selectFilesetVersionsByRetentionCount(
       @Param("versionRetentionCount") Long versionRetentionCount) {
     return getProvider().selectFilesetVersionsByRetentionCount(versionRetentionCount);

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

@@ -223,6 +223,50 @@ public String selectFilesetMetaById(@Param("filesetId") Long filesetId) {
         + " AND fm.deleted_at = 0 AND vi.deleted_at = 0";
   }
 
+  /**
+   * Returns an active fileset metadata row and locks it for the current transaction.
+   *
+   * <p>This query reads {@code fileset_meta} alone. The regular ID lookup returns one joined row
+   * per storage location and would also lock version rows, neither of which is needed when
+   * classifying a failed metadata CAS.
+   *
+   * @param filesetId the fileset ID
+   * @return the locking select SQL
+   */
+  public String selectFilesetMetaByIdForUpdate(@Param("filesetId") Long filesetId) {
+    return "SELECT fileset_id as filesetId, fileset_name as filesetName,"
+        + " metalake_id as metalakeId, catalog_id as catalogId, schema_id as schemaId,"
+        + " type as type, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + META_TABLE_NAME
+        + " WHERE fileset_id = #{filesetId} AND deleted_at = 0 FOR UPDATE";
+  }
+
+  /**
+   * Returns the active fileset metadata row selected by its natural key.
+   *
+   * <p>An overwrite may match the natural key instead of the incoming ID. Reading the stored row
+   * after the upsert tells dependent version rows which ID and database-generated version to use.
+   *
+   * @param schemaId the schema ID
+   * @param filesetName the fileset name
+   * @return the metadata-only select SQL
+   */
+  public String selectFilesetMetaBySchemaIdAndNameForUpdate(
+      @Param("schemaId") Long schemaId, @Param("filesetName") String filesetName) {
+    return "SELECT fileset_id as filesetId, fileset_name as filesetName,"
+        + " metalake_id as metalakeId, catalog_id as catalogId, schema_id as schemaId,"
+        + " type as type, audit_info as auditInfo,"
+        + " current_version as currentVersion, last_version as lastVersion,"
+        + " deleted_at as deletedAt"
+        + " FROM "
+        + META_TABLE_NAME
+        + " WHERE schema_id = #{schemaId} AND fileset_name = #{filesetName}"
+        + " AND deleted_at = 0 FOR UPDATE";
+  }
+
   public String insertFilesetMeta(@Param("filesetMeta") FilesetPO filesetPO) {
     return "INSERT INTO "
         + META_TABLE_NAME
@@ -268,11 +312,24 @@ public String insertFilesetMetaOnDuplicateKeyUpdate(@Param("filesetMeta") Filese
         + " schema_id = #{filesetMeta.schemaId},"
         + " type = #{filesetMeta.type},"
         + " audit_info = #{filesetMeta.auditInfo},"
-        + " current_version = #{filesetMeta.currentVersion},"
-        + " last_version = #{filesetMeta.lastVersion},"
+        // An overwrite is also a write observed by OCC. Advance from the stored value instead of
+        // resetting the row to the initial version carried by the incoming create request.
+        + " last_version = current_version + 1,"
+        + " current_version = current_version + 1,"
         + " deleted_at = #{filesetMeta.deletedAt}";
   }
 
+  /**
+   * Returns SQL that updates a fileset only while its OCC version is unchanged.
+   *
+   * <p>The version is the concurrency token, so payload, name, and audit columns are deliberately
+   * excluded from the predicate. This also detects change-then-change-back races that a full-row
+   * comparison would miss.
+   *
+   * @param newFilesetPO the new fileset values
+   * @param oldFilesetPO the fileset values and version observed by the caller
+   * @return the version-checked update SQL
+   */
   public String updateFilesetMeta(
       @Param("newFilesetMeta") FilesetPO newFilesetPO,
       @Param("oldFilesetMeta") FilesetPO oldFilesetPO) {
@@ -288,14 +345,7 @@ public String updateFilesetMeta(
         + " last_version = #{newFilesetMeta.lastVersion},"
         + " deleted_at = #{newFilesetMeta.deletedAt}"
         + " WHERE fileset_id = #{oldFilesetMeta.filesetId}"
-        + " AND fileset_name = #{oldFilesetMeta.filesetName}"
-        + " AND metalake_id = #{oldFilesetMeta.metalakeId}"
-        + " AND catalog_id = #{oldFilesetMeta.catalogId}"
-        + " AND schema_id = #{oldFilesetMeta.schemaId}"
-        + " AND type = #{oldFilesetMeta.type}"
-        + " AND audit_info = #{oldFilesetMeta.auditInfo}"
         + " AND current_version = #{oldFilesetMeta.currentVersion}"
-        + " AND last_version = #{oldFilesetMeta.lastVersion}"
         + " AND deleted_at = 0";
   }
 
@@ -329,12 +379,21 @@ public String softDeleteFilesetMetasBySchemaIds(@Param("schemaIds") List<Long> s
         + "</script>";
   }
 
-  public String softDeleteFilesetMetasByFilesetId(@Param("filesetId") Long filesetId) {
+  /**
+   * Returns SQL that deletes only the fileset version observed by the caller.
+   *
+   * @param filesetId the fileset ID
+   * @param currentVersion the version observed by the caller
+   * @return the version-checked delete SQL
+   */
+  public String softDeleteFilesetMetasByFilesetId(
+      @Param("filesetId") Long filesetId, @Param("currentVersion") Long currentVersion) {
     return "UPDATE "
         + META_TABLE_NAME
         + " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
         + " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
-        + " WHERE fileset_id = #{filesetId} AND deleted_at = 0";
+        + " WHERE fileset_id = #{filesetId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
   public String deleteFilesetMetasByLegacyTimeline(

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

@@ -116,6 +116,13 @@ public String deleteFilesetVersionsByLegacyTimeline(
         + " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT #{limit}";
   }
 
+  public String selectMaxFilesetVersion(@Param("filesetId") Long filesetId) {
+    return "SELECT MAX(version)"
+        + " FROM "
+        + VERSION_TABLE_NAME
+        + " WHERE fileset_id = #{filesetId} AND deleted_at = 0";
+  }
+
   public String selectFilesetVersionsByRetentionCount(
       @Param("versionRetentionCount") Long versionRetentionCount) {
     return "SELECT fileset_id as filesetId,"

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

@@ -57,11 +57,12 @@ public String softDeleteFilesetMetasBySchemaIds(List<Long> schemaIds) {
   }
 
   @Override
-  public String softDeleteFilesetMetasByFilesetId(Long filesetId) {
+  public String softDeleteFilesetMetasByFilesetId(Long filesetId, Long currentVersion) {
     return "UPDATE "
         + META_TABLE_NAME
         + " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)"
-        + " WHERE fileset_id = #{filesetId} AND deleted_at = 0";
+        + " WHERE fileset_id = #{filesetId}"
+        + " AND current_version = #{currentVersion} AND deleted_at = 0";
   }
 
   @Override
@@ -100,8 +101,13 @@ public String insertFilesetMetaOnDuplicateKeyUpdate(FilesetPO filesetPO) {
         + " schema_id = #{filesetMeta.schemaId},"
         + " type = #{filesetMeta.type},"
         + " audit_info = #{filesetMeta.auditInfo},"
-        + " current_version = #{filesetMeta.currentVersion},"
-        + " last_version = #{filesetMeta.lastVersion},"
+        // PostgreSQL requires the stored row to be qualified on the update side of ON CONFLICT.
+        + " current_version = "
+        + META_TABLE_NAME
+        + ".current_version + 1,"
+        + " last_version = "
+        + META_TABLE_NAME
+        + ".current_version + 1,"
         + " deleted_at = #{filesetMeta.deletedAt}";
   }
 }