add clean for committer

Author: jerry-024

paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java

@@ -83,4 +83,7 @@ public void discard(FileIO fileIO) throws IOException {
     public Path targetFilePath() {
         return new Path(objectName);
     }
+
+    @Override
+    public void clean(FileIO fileIO) throws IOException {}
 }

paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java

@@ -29,7 +29,7 @@
  */
 @Public
 public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream {
-    public static final String TEMP_DIR_NAME = "_temporary";
+    private static final String TEMP_DIR_NAME = "_temporary";
 
     private final Path targetPath;
     private final Path tempPath;
@@ -133,5 +133,13 @@ public void discard(FileIO fileIO) throws IOException {
         public Path targetFilePath() {
             return targetPath;
         }
+
+        @Override
+        public void clean(FileIO fileIO) throws IOException {
+            Path path = tempPath.getParent();
+            if (fileIO.exists(path)) {
+                fileIO.deleteQuietly(path);
+            }
+        }
     }
 }

paimon-common/src/main/java/org/apache/paimon/fs/TwoPhaseOutputStream.java

@@ -53,5 +53,7 @@ public interface Committer extends Serializable {
         void discard(FileIO fileIO) throws IOException;
 
         Path targetFilePath();
+
+        void clean(FileIO fileIO) throws IOException;
     }
 }

paimon-common/src/test/java/org/apache/paimon/fs/MultiPartUploadTwoPhaseOutputStreamTest.java

@@ -322,6 +322,9 @@ public void discard(FileIO fileIO) throws IOException {
         public Path targetFilePath() {
             return new Path(objectName);
         }
+
+        @Override
+        public void clean(FileIO fileIO) throws IOException {}
     }
 
     private static final class TestPart {

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java

@@ -21,7 +21,6 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.RenamingTwoPhaseOutputStream;
 import org.apache.paimon.fs.TwoPhaseOutputStream;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.stats.Statistics;
@@ -102,21 +101,10 @@ public void commit(List<CommitMessage> commitMessages) {
             for (TwoPhaseOutputStream.Committer committer : committers) {
                 committer.commit(this.fileIO);
             }
-            if (committers.stream()
-                    .filter(c -> c instanceof RenamingTwoPhaseOutputStream.TempFileCommitter)
-                    .findAny()
-                    .isPresent()) {
-                if (partitionPaths.size() > 1) {
-                    for (Path partitionPath : partitionPaths) {
-                        Path tempPath =
-                                new Path(partitionPath, RenamingTwoPhaseOutputStream.TEMP_DIR_NAME);
-                        fileIO.deleteQuietly(tempPath);
-                    }
-                } else {
-                    Path tempPath = new Path(location, RenamingTwoPhaseOutputStream.TEMP_DIR_NAME);
-                    fileIO.deleteQuietly(tempPath);
-                }
+            for (TwoPhaseOutputStream.Committer committer : committers) {
+                committer.clean(this.fileIO);
             }
+
         } catch (Exception e) {
             this.abort(commitMessages);
             throw new RuntimeException(e);