[core] Delete old format table files in batches during dynamic overwrite

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-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java

@@ -21,6 +21,9 @@
 import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
 
+import java.security.AccessControlContext;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -134,7 +137,7 @@ private void advanceIfNeeded() {
     }
 
     /**
-     * Parallel processes one bounded batch at a time and returns results in input order.
+     * Processes a bounded number of inputs in parallel and returns results in input order.
      *
      * <p>The caller must close the iterator to cancel unstarted tasks and wait for running tasks.
      */
@@ -143,10 +146,36 @@ public static <T, U> CloseableBatchIterator<T> sequentialBatchedExecuteCloseable
             Function<U, List<T>> processor,
             List<U> input,
             int queueSize) {
+        return newSequentialBatchIterator(executor, processor, input.iterator(), queueSize, true);
+    }
+
+    /**
+     * As {@link #sequentialBatchedExecuteCloseable}, but closing waits for a task that has already
+     * started instead of interrupting it.
+     *
+     * <p>Use this when a task changes stored state. Interrupting a delete or a write halfway leaves
+     * the caller unable to say whether it took effect, so a caller that has to know the outcome of
+     * everything it handed out cannot let close cancel work that is already running.
+     */
+    public static <T, U> CloseableBatchIterator<T> sequentialBatchedExecuteAwaitRunningTasksOnClose(
+            ExecutorService executor,
+            Function<U, List<T>> processor,
+            Iterator<U> input,
+            int queueSize) {
+        return newSequentialBatchIterator(executor, processor, input, queueSize, false);
+    }
+
+    private static <T, U> CloseableBatchIterator<T> newSequentialBatchIterator(
+            ExecutorService executor,
+            Function<U, List<T>> processor,
+            Iterator<U> input,
+            int queueSize,
+            boolean cancelRunningOnClose) {
         if (queueSize <= 0) {
             throw new NegativeArraySizeException("queue size should not be negative");
         }
-        return new SequentialBatchIterator<>(executor, processor, input, queueSize);
+        return new SequentialBatchIterator<>(
+                executor, processor, input, queueSize, cancelRunningOnClose);
     }
 
     public static <U> void randomlyOnlyExecute(
@@ -224,21 +253,28 @@ private static class SequentialBatchIterator<T, U> implements CloseableBatchIter
 
         private final ExecutorService executor;
         private final Function<U, List<T>> processor;
-        private final Queue<List<U>> batches;
+        private final Iterator<U> input;
+        private final int queueSize;
+        private final boolean cancelRunningOnClose;
         private final Queue<BatchTask<T, U>> activeTasks = new ArrayDeque<>();
+        private final Object submissionLock = new Object();
 
         private Iterator<T> activeResults = Collections.<T>emptyList().iterator();
         private T next;
         private boolean closed;
+        private boolean submissionStopped;
 
         private SequentialBatchIterator(
                 ExecutorService executor,
                 Function<U, List<T>> processor,
-                List<U> input,
-                int queueSize) {
+                Iterator<U> input,
+                int queueSize,
+                boolean cancelRunningOnClose) {
             this.executor = executor;
             this.processor = processor;
-            this.batches = new ArrayDeque<>(Lists.partition(input, queueSize));
+            this.input = input;
+            this.queueSize = queueSize;
+            this.cancelRunningOnClose = cancelRunningOnClose;
         }
 
         @Override
@@ -263,32 +299,51 @@ private void advanceIfNeeded() {
             while (next == null) {
                 if (activeResults.hasNext()) {
                     next = activeResults.next();
-                } else if (!activeTasks.isEmpty()) {
-                    BatchTask<T, U> task = activeTasks.peek();
-                    try {
-                        List<T> results = task.result();
+                    continue;
+                }
+                fillWindow();
+                if (activeTasks.isEmpty()) {
+                    return;
+                }
+                BatchTask<T, U> task = activeTasks.peek();
+                try {
+                    List<T> results = task.result();
+                    activeTasks.poll();
+                    activeResults = results.iterator();
+                } catch (RuntimeException | Error failure) {
+                    if (task.failureReported()) {
                         activeTasks.poll();
-                        activeResults = results.iterator();
-                    } catch (RuntimeException | Error failure) {
-                        if (task.failureReported()) {
-                            activeTasks.poll();
-                        }
-                        throw failure;
                     }
-                } else if (batches.isEmpty()) {
-                    return;
-                } else {
-                    submitBatch(batches.poll());
+                    throw failure;
                 }
             }
         }
 
-        private void submitBatch(List<U> batch) {
+        /** Does not consume more input than the active-task window can hold. */
+        private void fillWindow() {
             ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-            for (U input : batch) {
-                BatchTask<T, U> task = new BatchTask<>(processor, input, classLoader);
-                executor.execute(task);
-                activeTasks.add(task);
+            AccessControlContext accessControlContext = AccessController.getContext();
+            while (activeTasks.size() < queueSize) {
+                synchronized (submissionLock) {
+                    if (submissionStopped || !input.hasNext()) {
+                        return;
+                    }
+                    BatchTask<T, U> task =
+                            new BatchTask<>(
+                                    processor,
+                                    input.next(),
+                                    classLoader,
+                                    accessControlContext,
+                                    this::stopSubmission);
+                    executor.execute(task);
+                    activeTasks.add(task);
+                }
+            }
+        }
+
+        private void stopSubmission() {
+            synchronized (submissionLock) {
+                submissionStopped = true;
             }
         }
 
@@ -298,17 +353,25 @@ public synchronized void close() {
                 return;
             }
             closed = true;
-            batches.clear();
 
             Throwable failure = null;
             boolean interrupted = Thread.interrupted();
             for (BatchTask<T, U> task : activeTasks) {
                 try {
-                    task.cancel();
+                    task.cancelIfUnstarted();
                 } catch (Throwable cleanupFailure) {
                     failure = firstOrSuppressed(cleanupFailure, failure);
                 }
             }
+            if (cancelRunningOnClose) {
+                for (BatchTask<T, U> task : activeTasks) {
+                    try {
+                        task.interruptIfRunning();
+                    } catch (Throwable cleanupFailure) {
+                        failure = firstOrSuppressed(cleanupFailure, failure);
+                    }
+                }
+            }
             for (BatchTask<T, U> task : activeTasks) {
                 while (true) {
                     try {
@@ -346,6 +409,8 @@ private static class BatchTask<T, U> implements Runnable {
         private final Function<U, List<T>> processor;
         private final U input;
         private final ClassLoader classLoader;
+        private final AccessControlContext accessControlContext;
+        private final Runnable stopSubmission;
         private final CountDownLatch completion = new CountDownLatch(1);
 
         private int state = CREATED;
@@ -354,10 +419,17 @@ private static class BatchTask<T, U> implements Runnable {
         private Throwable failure;
         private volatile boolean failureReported;
 
-        private BatchTask(Function<U, List<T>> processor, U input, ClassLoader classLoader) {
+        private BatchTask(
+                Function<U, List<T>> processor,
+                U input,
+                ClassLoader classLoader,
+                AccessControlContext accessControlContext,
+                Runnable stopSubmission) {
             this.processor = processor;
             this.input = input;
             this.classLoader = classLoader;
+            this.accessControlContext = accessControlContext;
+            this.stopSubmission = stopSubmission;
         }
 
         @Override
@@ -372,12 +444,26 @@ public void run() {
                 runner = Thread.currentThread();
             }
 
+            Thread currentThread = Thread.currentThread();
+            boolean interruptedOnEntry = currentThread.isInterrupted();
+            ClassLoader originalClassLoader = currentThread.getContextClassLoader();
             try {
-                Thread.currentThread().setContextClassLoader(classLoader);
-                result = processor.apply(input);
+                currentThread.setContextClassLoader(classLoader);
+                result =
+                        AccessController.doPrivileged(
+                                (PrivilegedAction<List<T>>) () -> processor.apply(input),
+                                accessControlContext);
             } catch (RuntimeException | Error taskFailure) {
                 failure = taskFailure;
+                stopSubmission.run();
             } finally {
+                currentThread.setContextClassLoader(originalClassLoader);
+                // Reset the flag to its entry state so a cancelled task cannot leak an interrupt
+                // to a reused worker and a direct executor cannot clear its caller's interrupt.
+                Thread.interrupted();
+                if (interruptedOnEntry) {
+                    currentThread.interrupt();
+                }
                 synchronized (this) {
                     runner = null;
                     state = FINISHED;
@@ -386,21 +472,27 @@ public void run() {
             }
         }
 
-        private synchronized void cancel() {
+        private synchronized void interruptIfRunning() {
+            if (state == RUNNING) {
+                runner.interrupt();
+            }
+        }
+
+        private synchronized void cancelIfUnstarted() {
             if (state == CREATED) {
                 state = CANCELLED;
                 completion.countDown();
-            } else if (state == RUNNING) {
-                runner.interrupt();
             }
         }
 
         private List<T> result() {
-            try {
-                completion.await();
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new RuntimeException(e);
+            if (completion.getCount() != 0) {
+                try {
+                    completion.await();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException(e);
+                }
             }
             if (failure != null) {
                 failureReported = true;