(paimon) branch master updated: [core] Fix file operation thread limit (#9082)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 6f395abcb2 [core] Fix file operation thread limit (#9082)
6f395abcb2 is described below

commit 6f395abcb23ffcc5d2cf18455ec10fc1d087cc59
Author: huangxiaoping <1754789...@qq.com>
AuthorDate: Thu Aug 6 22:11:42 2026 +0800

    [core] Fix file operation thread limit (#9082)
---
 .../java/org/apache/paimon/utils/FileOperationThreadPool.java     | 8 ++++++--
 .../java/org/apache/paimon/operation/ListUnexistingFiles.java     | 4 ++--
 .../main/java/org/apache/paimon/table/sink/TableCommitImpl.java   | 3 +--
 3 files changed, 9 insertions(+), 6 deletions(-)

diff --git 
a/paimon-common/src/main/java/org/apache/paimon/utils/FileOperationThreadPool.java
 
b/paimon-common/src/main/java/org/apache/paimon/utils/FileOperationThreadPool.java
index 87c096c50d..0b875f1ad0 100644
--- 
a/paimon-common/src/main/java/org/apache/paimon/utils/FileOperationThreadPool.java
+++ 
b/paimon-common/src/main/java/org/apache/paimon/utils/FileOperationThreadPool.java
@@ -20,6 +20,7 @@ package org.apache.paimon.utils;
 
 import org.apache.paimon.fs.FileIO;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
@@ -32,10 +33,13 @@ public class FileOperationThreadPool {
     private static ThreadPoolExecutor executorService =
             createCachedThreadPool(Runtime.getRuntime().availableProcessors(), 
THREAD_NAME);
 
-    public static synchronized ThreadPoolExecutor getExecutorService(int 
threadNum) {
-        if (threadNum <= executorService.getMaximumPoolSize()) {
+    public static synchronized ExecutorService getExecutorService(int 
threadNum) {
+        if (threadNum <= 0 || threadNum == 
executorService.getMaximumPoolSize()) {
             return executorService;
         }
+        if (threadNum < executorService.getMaximumPoolSize()) {
+            return new SemaphoredDelegatingExecutor(executorService, 
threadNum, false);
+        }
         // we don't need to close previous pool
         // it is just cached pool
         executorService = createCachedThreadPool(threadNum, THREAD_NAME);
diff --git 
a/paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java
 
b/paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java
index dc3d029fb6..c824eb97c1 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java
@@ -38,14 +38,14 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ExecutorService;
 
 /** List what data files recorded in manifests are missing from the 
filesystem. */
 public class ListUnexistingFiles {
 
     private final FileStoreTable table;
     private final FileStorePathFactory pathFactory;
-    private final ThreadPoolExecutor executor;
+    private final ExecutorService executor;
 
     public ListUnexistingFiles(FileStoreTable table) {
         this.table = table;
diff --git 
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java 
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 7e47041db6..014b5e64da 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -61,7 +61,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -89,7 +88,7 @@ public class TableCommitImpl implements InnerTableCommit {
     private final AtomicReference<Throwable> maintainError;
     private final String tableName;
     private final boolean forceCreatingSnapshot;
-    private final ThreadPoolExecutor fileCheckExecutor;
+    private final ExecutorService fileCheckExecutor;
 
     @Nullable private Map<String, String> overwritePartitionSpec = null;
     @Nullable private List<BinaryRow> overwriteStaticPartitions = null;