[AMORO-2555] Tag dangling delete snapshot commits (#3058)

* Tag dangling delete snapshot commits

* make maintain commits public

* Re-gen table metastore java code via thrift

(cherry picked from commit 7066300)
Signed-off-by: zhoujinsong <[email protected]>

amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/maintainer/IcebergTableMaintainer.java

@@ -103,6 +103,12 @@ public class IcebergTableMaintainer implements TableMaintainer {
   public static final String EXPIRE_TIMESTAMP_MS = "TIMESTAMP_MS";
   public static final String EXPIRE_TIMESTAMP_S = "TIMESTAMP_S";
 
+  public static final Set<String> AMORO_MAINTAIN_COMMITS =
+      Sets.newHashSet(
+          CommitMetaProducer.OPTIMIZE.name(),
+          CommitMetaProducer.DATA_EXPIRATION.name(),
+          CommitMetaProducer.CLEAN_DANGLING_DELETE.name());
+
   protected Table table;
 
   public IcebergTableMaintainer(Table table) {
@@ -435,6 +441,9 @@ private int clearInternalTableDanglingDeleteFiles() {
         Collections.emptySet(),
         Collections.emptySet());
     try {
+      rewriteFiles.set(
+          org.apache.amoro.op.SnapshotSummary.SNAPSHOT_PRODUCER,
+          CommitMetaProducer.CLEAN_DANGLING_DELETE.name());
       rewriteFiles.commit();
     } catch (ValidationException e) {
       LOG.warn("Iceberg RewriteFiles commit failed on clear danglingDeleteFiles, but ignore", e);
@@ -498,10 +507,7 @@ public static Snapshot findLatestSnapshotContainsKey(Table table, String summary
   public static long fetchLatestNonOptimizedSnapshotTime(Table table) {
     Optional<Snapshot> snapshot =
         IcebergTableUtil.findFirstMatchSnapshot(
-            table,
-            s ->
-                !s.summary().containsValue(CommitMetaProducer.OPTIMIZE.name())
-                    && !s.summary().containsValue(CommitMetaProducer.DATA_EXPIRATION.name()));
+            table, s -> s.summary().values().stream().noneMatch(AMORO_MAINTAIN_COMMITS::contains));
     return snapshot.map(Snapshot::timestampMillis).orElse(Long.MAX_VALUE);
   }
 

amoro-core/src/main/thrift/amoro_table_metastore.thrift

@@ -99,7 +99,8 @@ struct PrimaryKeySpec {
 enum CommitMetaProducer {
     OPTIMIZE,
     INGESTION,
-    DATA_EXPIRATION
+    DATA_EXPIRATION,
+    CLEAN_DANGLING_DELETE
 }
 
 struct Blocker {