[core][common][oss] Parallelize format table overwrite commit and batch delete old files

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 old-data file deletions during overwrite commits for an internal Format Table with catalog-managed partitions. Supported values are 1 through 64. Other Format Tables use serial cleanup.</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 an internal 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,26 @@ 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 old-data file deletions during "
+                                    + "overwrite commits for an internal Format Table with "
+                                    + "catalog-managed partitions. Supported values are 1 through "
+                                    + "64. Other Format Tables use serial cleanup.");
+
+    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 an internal 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 +3322,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/BatchDeleteResult.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+import org.apache.paimon.annotation.Public;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Immutable result of a successful strict batch delete.
+ *
+ * @since 2.1
+ */
+@Public
+public final class BatchDeleteResult {
+
+    private final List<Path> deletedOrNotFound;
+
+    public BatchDeleteResult(List<Path> deletedOrNotFound) {
+        this.deletedOrNotFound = Collections.unmodifiableList(new ArrayList<>(deletedOrNotFound));
+    }
+
+    /** Files confirmed deleted or not found, in request order. */
+    public List<Path> deletedOrNotFound() {
+        return deletedOrNotFound;
+    }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.fs;
+
+import org.apache.paimon.annotation.Public;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Deletes files in one provider request without falling back to individual deletes.
+ *
+ * <p>A successful invocation confirms every requested file as deleted or not found. A failure or
+ * timeout only means that the complete batch was not confirmed; the provider may already have
+ * deleted some files. If a caller retries, it must retry the same complete batch. Implementations
+ * must validate the complete request before accessing storage.
+ *
+ * @since 2.1
+ */
+@Public
+public interface BatchFileDeleter {
+
+    /** Maximum number of files accepted by one {@link #delete(List)} invocation. */
+    int maxBatchSize();
+
+    /**
+     * Deletes one non-empty batch.
+     *
+     * @return files confirmed deleted or not found
+     * @throws IOException if any requested file cannot be confirmed
+     */
+    BatchDeleteResult delete(List<Path> files) throws IOException;
+}

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

@@ -217,6 +217,18 @@ default FileStatus[] listDirectories(Path path) throws IOException {
      */
     boolean exists(Path path) throws IOException;
 
+    /**
+     * Returns a strict batch-delete capability for the provider serving the given path.
+     *
+     * <p>An empty result is the only signal that callers may use individual deletes instead. The
+     * default performs no storage access and preserves compatibility with existing providers.
+     *
+     * @since 2.1
+     */
+    default Optional<BatchFileDeleter> batchFileDeleter(Path path) throws IOException {
+        return Optional.empty();
+    }
+
     /**
      * Delete a file.
      *

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

@@ -24,6 +24,9 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
 
 /**
  * A {@link FileIO} for plugin jar. {@link FileIO} is serializable, so plugin FileIO should be
@@ -78,6 +81,28 @@ public boolean exists(Path path) throws IOException {
         return wrap(() -> fileIO(path).exists(path));
     }
 
+    @Override
+    public Optional<BatchFileDeleter> batchFileDeleter(Path path) throws IOException {
+        Optional<BatchFileDeleter> capability = wrap(() -> fileIO(path).batchFileDeleter(path));
+        if (!capability.isPresent()) {
+            return Optional.empty();
+        }
+
+        BatchFileDeleter delegate = capability.get();
+        return Optional.of(
+                new BatchFileDeleter() {
+                    @Override
+                    public int maxBatchSize() {
+                        return wrapUnchecked(delegate::maxBatchSize);
+                    }
+
+                    @Override
+                    public BatchDeleteResult delete(List<Path> files) throws IOException {
+                        return wrap(() -> delegate.delete(files));
+                    }
+                });
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return wrap(() -> fileIO(path).delete(path, recursive));
@@ -132,6 +157,16 @@ private <T> T wrap(Func<T> func) throws IOException {
         }
     }
 
+    private <T> T wrapUnchecked(Supplier<T> supplier) {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(pluginClassLoader());
+            return supplier.get();
+        } finally {
+            Thread.currentThread().setContextClassLoader(cl);
+        }
+    }
+
     /** Apply function with wrapping classloader. */
     @FunctionalInterface
     protected interface Func<T> {

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

@@ -26,10 +26,14 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.net.URI;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
 
 import static org.apache.paimon.options.CatalogOptions.RESOLVING_FILE_IO_ENABLED;
 
@@ -94,6 +98,30 @@ public boolean exists(Path path) throws IOException {
         return wrap(() -> fileIO(path).exists(path));
     }
 
+    @Override
+    public Optional<BatchFileDeleter> batchFileDeleter(Path path) throws IOException {
+        Optional<BatchFileDeleter> capability = wrap(() -> fileIO(path).batchFileDeleter(path));
+        if (!capability.isPresent()) {
+            return Optional.empty();
+        }
+
+        URI provider = path.toUri();
+        BatchFileDeleter delegate = capability.get();
+        return Optional.of(
+                new BatchFileDeleter() {
+                    @Override
+                    public int maxBatchSize() {
+                        return wrapUnchecked(delegate::maxBatchSize);
+                    }
+
+                    @Override
+                    public BatchDeleteResult delete(List<Path> files) throws IOException {
+                        validateProvider(files, provider);
+                        return wrap(() -> delegate.delete(files));
+                    }
+                });
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return wrap(() -> fileIO(path).delete(path, recursive));
@@ -149,6 +177,30 @@ private <T> T wrap(Func<T> func) throws IOException {
         }
     }
 
+    private <T> T wrapUnchecked(Supplier<T> supplier) {
+        ClassLoader cl = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(ResolvingFileIO.class.getClassLoader());
+            return supplier.get();
+        } finally {
+            Thread.currentThread().setContextClassLoader(cl);
+        }
+    }
+
+    private static void validateProvider(List<Path> files, URI provider) {
+        if (files == null) {
+            throw new IllegalArgumentException("Batch delete files must not be null.");
+        }
+        for (Path file : files) {
+            if (file == null
+                    || !Objects.equals(provider.getScheme(), file.toUri().getScheme())
+                    || !Objects.equals(provider.getAuthority(), file.toUri().getAuthority())) {
+                throw new IllegalArgumentException(
+                        "Batch delete files must use the capability provider's scheme and authority.");
+            }
+        }
+    }
+
     /** Apply function with wrapping classloader. */
     @FunctionalInterface
     protected interface Func<T> {

paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.fs.BatchFileDeleter;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
@@ -40,6 +41,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -165,6 +167,11 @@ public boolean exists(Path path) throws IOException {
         return delegate.exists(path);
     }
 
+    @Override
+    public Optional<BatchFileDeleter> batchFileDeleter(Path path) throws IOException {
+        return delegate.batchFileDeleter(path);
+    }
+
     @Override
     public boolean delete(Path path, boolean recursive) throws IOException {
         return delegate.delete(path, recursive);