[BugFix] Use a separate thread pool for dictionary collection

Use a separate dictionary to avoid occupying the default ForkJoinPool leading to potential deadlocks. Signed-off-by: stdpain <[email protected]>
StarRocks · Dec 27, 2024 · d0c5300 · d0c5300
1 parent 571cf78
commit d0c5300
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/common/Config.java b/fe/fe-core/src/main/java/com/starrocks/common/Config.java
@@ -2047,6 +2047,9 @@ public class Config extends ConfigBase {
     @ConfField
     public static long statistic_dict_columns = 100000;
 
+    @ConfField
+    public static int dict_collect_thread_pool_size = 16;
+
     /**
      * The column statistic cache update interval
      */

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/statistics/CacheDictManager.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/statistics/CacheDictManager.java
@@ -25,6 +25,7 @@
 import com.starrocks.common.Config;
 import com.starrocks.common.Pair;
 import com.starrocks.common.Status;
+import com.starrocks.common.ThreadPoolManager;
 import com.starrocks.memory.MemoryTrackable;
 import com.starrocks.qe.ConnectContext;
 import com.starrocks.server.GlobalStateMgr;
@@ -44,6 +45,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
 
 import static com.starrocks.statistic.StatisticExecutor.queryDictSync;
 
@@ -107,9 +109,12 @@ public CompletableFuture<Optional<ColumnDict>> asyncReload(
 
     private final AsyncLoadingCache<ColumnIdentifier, Optional<ColumnDict>> dictStatistics = Caffeine.newBuilder()
             .maximumSize(Config.statistic_dict_columns)
+            .executor(ThreadPoolManager.newDaemonCacheThreadPool(Config.dict_collect_thread_pool_size, "cache-dict",
+                    false))
             .buildAsync(dictLoader);
 
-    private Optional<ColumnDict> deserializeColumnDict(long tableId, ColumnId columnName, TStatisticData statisticData) {
+    private Optional<ColumnDict> deserializeColumnDict(long tableId, ColumnId columnName,
+                                                       TStatisticData statisticData) {
         if (statisticData.dict == null) {
             throw new RuntimeException("Collect dict error in BE");
         }
@@ -167,7 +172,8 @@ public boolean hasGlobalDict(long tableId, ColumnId columnName, long versionTime
         Set<Long> dbIds = ConnectContext.get().getCurrentSqlDbIds();
         for (Long id : dbIds) {
             Database db = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb(id);
-            if (db != null && GlobalStateMgr.getCurrentState().getLocalMetastore().getTable(db.getId(), tableId) != null) {
+            if (db != null &&
+                    GlobalStateMgr.getCurrentState().getLocalMetastore().getTable(db.getId(), tableId) != null) {
                 columnIdentifier.setDbId(db.getId());
                 break;
             }