[api][core] Parallelize format table overwrite commit

docs/generated/core_configuration.html

@@ -746,6 +746,18 @@
             <td>String</td>
             <td>Format table commit hive sync uri.</td>
         </tr>
+        <tr>
+            <td><h5>format-table.commit.cleanup-thread-num</h5></td>
+            <td style="word-wrap: break-word;">64</td>
+            <td>Integer</td>
+            <td>The maximum number of concurrent deletions of old data files during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup. This limit uses a separate thread pool and is independent of file-operation.thread-num, so the total file-operation concurrency in one process may be the sum of both limits.</td>
+        </tr>
+        <tr>
+            <td><h5>format-table.commit.publish-thread-num</h5></td>
+            <td style="word-wrap: break-word;">64</td>
+            <td>Integer</td>
+            <td>The maximum number of concurrent file publications during commits for a partitioned Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables publish serially.</td>
+        </tr>
         <tr>
             <td><h5>format-table.file.compression</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2654,6 +2654,29 @@ public String toString() {
                     .noDefaultValue()
                     .withDescription("Format table commit hive sync uri.");
 
+    public static final ConfigOption<Integer> FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM =
+            key("format-table.commit.cleanup-thread-num")
+                    .intType()
+                    .defaultValue(64)
+                    .withDescription(
+                            "The maximum number of concurrent deletions of old data files during "
+                                    + "overwrite commits for an internal Format Table with "
+                                    + "catalog-managed partitions. Supported values are 1 through "
+                                    + "64. Other Format Tables use serial cleanup. This limit uses "
+                                    + "a separate thread pool and is independent of "
+                                    + "file-operation.thread-num, so the total file-operation "
+                                    + "concurrency in one process may be the sum of both limits.");
+
+    public static final ConfigOption<Integer> FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM =
+            key("format-table.commit.publish-thread-num")
+                    .intType()
+                    .defaultValue(64)
+                    .withDescription(
+                            "The maximum number of concurrent file publications during commits "
+                                    + "for a partitioned Format Table with catalog-managed "
+                                    + "partitions. Supported values are 1 through 64. Other Format "
+                                    + "Tables publish serially.");
+
     @Immutable
     public static final ConfigOption<String> BLOB_FIELD =
             key("blob-field")
@@ -3302,6 +3325,26 @@ public String formatTableCommitSyncPartitionHiveUri() {
         return options.get(FORMAT_TABLE_COMMIT_HIVE_SYNC_URI);
     }
 
+    public int formatTableCommitCleanupThreadNum() {
+        int threadNum = options.get(FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM);
+        checkArgument(
+                threadNum >= 1 && threadNum <= 64,
+                "Option %s must be between 1 and 64, but was %s.",
+                FORMAT_TABLE_COMMIT_CLEANUP_THREAD_NUM.key(),
+                threadNum);
+        return threadNum;
+    }
+
+    public int formatTableCommitPublishThreadNum() {
+        int threadNum = options.get(FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM);
+        checkArgument(
+                threadNum >= 1 && threadNum <= 64,
+                "Option %s must be between 1 and 64, but was %s.",
+                FORMAT_TABLE_COMMIT_PUBLISH_THREAD_NUM.key(),
+                threadNum);
+        return threadNum;
+    }
+
     public MemorySize fileReaderAsyncThreshold() {
         return options.get(FILE_READER_ASYNC_THRESHOLD);
     }

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

@@ -71,13 +71,22 @@ public void commit(FileIO fileIO) throws IOException {
     @Override
     public void discard(FileIO fileIO) throws IOException {
         try {
-            MultiPartUploadStore<T, C> multiPartUploadStore = multiPartUploadStore(fileIO);
-            multiPartUploadStore.abortMultipartUpload(objectName, uploadId);
+            abortMultipartUpload(fileIO);
         } catch (Exception e) {
             LOG.warn("Failed to discard multipart upload with ID: {}", uploadId, e);
         }
     }
 
+    @Override
+    public void discardStaging(FileIO fileIO) throws IOException {
+        try {
+            // Aborting an upload never deletes a possibly completed object.
+            abortMultipartUpload(fileIO);
+        } catch (Exception e) {
+            throw new IOException("Failed to discard multipart upload with ID: " + uploadId, e);
+        }
+    }
+
     @Override
     public Path targetPath() {
         return this.targetPath;
@@ -91,6 +100,11 @@ public List<T> uploadedParts() {
     @Override
     public void clean(FileIO fileIO) throws IOException {}
 
+    private void abortMultipartUpload(FileIO fileIO) throws IOException {
+        MultiPartUploadStore<T, C> multiPartUploadStore = multiPartUploadStore(fileIO);
+        multiPartUploadStore.abortMultipartUpload(objectName, uploadId);
+    }
+
     private MultiPartUploadStore<T, C> multiPartUploadStore(FileIO fileIO) throws IOException {
         if (fileIO instanceof RESTTokenFileIO) {
             RESTTokenFileIO restTokenFileIO = (RESTTokenFileIO) fileIO;

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

@@ -53,6 +53,19 @@ public interface Committer extends Serializable {
          */
         void discard(FileIO fileIO) throws IOException;
 
+        /**
+         * Discards staged resources without deleting {@link #targetPath()}.
+         *
+         * <p>This is used when a commit may have taken effect and its target must therefore be
+         * preserved. The default delegates to {@link #clean}. Override this method if a failed or
+         * uncertain commit can leave staged resources that {@code clean} does not release.
+         *
+         * @throws IOException if an I/O error occurs during cleanup
+         */
+        default void discardStaging(FileIO fileIO) throws IOException {
+            clean(fileIO);
+        }
+
         Path targetPath();
 
         /**

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

@@ -78,6 +78,14 @@ public BatchTableCommit newCommit() {
         CoreOptions options = new CoreOptions(table.options());
         boolean formatTablePartitionOnlyValueInPath = options.formatTablePartitionOnlyValueInPath();
         String syncHiveUri = options.formatTableCommitSyncPartitionHiveUri();
+        int cleanupThreadNum =
+                table.partitionManager() != null && !table.partitionKeys().isEmpty()
+                        ? options.formatTableCommitCleanupThreadNum()
+                        : 1;
+        int publishThreadNum =
+                table.partitionManager() != null && !table.partitionKeys().isEmpty()
+                        ? options.formatTableCommitPublishThreadNum()
+                        : 1;
         return new FormatTableCommit(
                 table.location(),
                 table.partitionKeys(),
@@ -90,7 +98,9 @@ public BatchTableCommit newCommit() {
                 syncHiveUri,
                 table.catalogContext(),
                 table.partitionManager(),
-                options.dynamicPartitionOverwrite());
+                options.dynamicPartitionOverwrite(),
+                cleanupThreadNum,
+                publishThreadNum);
     }
 
     @Override