Repository: NetEase/arctic (java)

Issue:
[Bug]:  An error occurred in Mysql init sql 
### What happened?

I'm deploying the AMS. When I use `ams-mysql-init.sql` to init database, an error occurred.

### Affects Versions

master

### What engines are you seeing the problem on?

AMS

### How to reproduce

Just execute the SQL in the `ams-mysql-init.sql`.

### Relevant log output

```shell
SQL error [1171] [42000]: All parts of a PRIMARY KEY must be NOT NULL; if you need NULL in a key, use UNIQUE instead
```


### Anything else

_No response_

### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Pull request description:
[ARCTIC-1584] Fix Mysql init sql
## Why are the changes needed?
Fix #1584 .

## Brief change log

  - *Make `resource_id` not null in table `resource`.*

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? (not applicable)


Patch:
diff --git a/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java b/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java
index aeb02997c..9b655e66a 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java
@@ -48,38 +48,56 @@ public class ArcticManagementConf {
           .defaultValue("admin")
           .withDescription("The administrator password");
 
-  public static final ConfigOption<Long> EXTERNAL_CATALOG_REFRESH_INTERVAL =
-      ConfigOptions.key("refresh-external-catalog-interval")
+  public static final ConfigOption<Long> REFRESH_EXTERNAL_CATALOGS_INTERVAL =
+      ConfigOptions.key("refresh-external-catalogs.interval")
           .longType()
           .defaultValue(3 * 60 * 1000L)
           .withDescription("Interval to refresh the external catalog.");
 
-  public static final ConfigOption<Integer> EXPIRE_THREAD_POOL_SIZE =
-      ConfigOptions.key("expire-table-thread-count")
+  public static final ConfigOption<Boolean> EXPIRE_SNAPSHOTS_ENABLED =
+      ConfigOptions.key("expire-snapshots.enabled")
+          .booleanType()
+          .defaultValue(true)
+          .withDescription("Enable snapshots expiring.");
+
+  public static final ConfigOption<Integer> EXPIRE_SNAPSHOTS_THREAD_COUNT =
+      ConfigOptions.key("expire-snapshots.thread-count")
           .intType()
           .defaultValue(10)
-          .withDescription("The number of threads used for table expiring.");
+          .withDescription("The number of threads used for snapshots expiring.");
+
+  public static final ConfigOption<Boolean> CLEAN_ORPHAN_FILES_ENABLED =
+      ConfigOptions.key("clean-orphan-files.enabled")
+          .booleanType()
+          .defaultValue(true)
+          .withDescription("Enable orphan files cleaning.");
 
-  public static final ConfigOption<Integer> ORPHAN_CLEAN_THREAD_POOL_SIZE =
-      ConfigOptions.key("clean-orphan-file-thread-count")
+  public static final ConfigOption<Integer> CLEAN_ORPHAN_FILES_THREAD_COUNT =
+      ConfigOptions.key("clean-orphan-files.thread-count")
           .intType()
           .defaultValue(10)
           .withDescription("The number of threads used for orphan files cleaning.");
 
-  public static final ConfigOption<Integer> SUPPORT_HIVE_SYNC_THREAD_POOL_SIZE =
-      ConfigOptions.key("sync-hive-tables-thread-count")
+  public static final ConfigOption<Boolean> SYNC_HIVE_TABLES_ENABLED =
+      ConfigOptions.key("sync-hive-tables.enabled")
+          .booleanType()
+          .defaultValue(true)
+          .withDescription("Enable synchronizing Hive tables.");
+
+  public static final ConfigOption<Integer> SYNC_HIVE_TABLES_THREAD_COUNT =
+      ConfigOptions.key("sync-hive-tables.thread-count")
           .intType()
           .defaultValue(10)
           .withDescription("The number of threads used for synchronizing Hive tables.");
 
-  public static final ConfigOption<Integer> SNAPSHOTS_REFRESHING_THREAD_POOL_SIZE =
-      ConfigOptions.key("refresh-table-thread-count")
+  public static final ConfigOption<Integer> REFRESH_TABLES_THREAD_COUNT =
+      ConfigOptions.key("refresh-tables.thread-count")
           .intType()
           .defaultValue(10)
           .withDescription("The number of threads used for refreshing tables.");
 
-  public static final ConfigOption<Long> SNAPSHOTS_REFRESHING_INTERVAL =
-      ConfigOptions.key("refresh-table-interval")
+  public static final ConfigOption<Long> REFRESH_TABLES_INTERVAL =
+      ConfigOptions.key("refresh-tables.interval")
           .longType()
           .defaultValue(60000L)
           .withDescription("Interval for refreshing table metadata.");
@@ -144,7 +162,7 @@ public class ArcticManagementConf {
           .defaultValue(19090)
           .withDescription("Port that the Http server is bound to.");
 
-  public static final ConfigOption<Integer> OPTIMIZING_COMMIT_THREAD_POOL_SIZE =
+  public static final ConfigOption<Integer> OPTIMIZING_COMMIT_THREAD_COUNT =
       ConfigOptions.key("self-optimizing.commit-thread-count")
           .intType()
           .defaultValue(10)
diff --git a/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java b/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
index 4e8ec392e..7b21dcbdc 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
@@ -33,7 +33,9 @@ import com.netease.arctic.server.persistence.SqlSessionFactoryProvider;
 import com.netease.arctic.server.resource.ContainerMetadata;
 import com.netease.arctic.server.resource.ResourceContainers;
 import com.netease.arctic.server.table.DefaultTableService;
+import com.netease.arctic.server.table.RuntimeHandlerChain;
 import com.netease.arctic.server.table.executor.AsyncTableExecutors;
+import com.netease.arctic.server.utils.ConfigOption;
 import com.netease.arctic.server.utils.Configurations;
 import com.netease.arctic.server.utils.ThriftServiceProxy;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -113,14 +115,14 @@ public class ArcticServiceContainer {
 
     LOG.info("Setting up AMS table executors...");
     AsyncTableExecutors.getInstance().setup(tableService, serviceConfig);
-    tableService.addHandlerChain(optimizingService.getTableRuntimeHandler());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getSnapshotsExpiringExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingCommitExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingExpiringExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getBlockerExpiringExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getHiveCommitSyncExecutor());
-    tableService.addHandlerChain(AsyncTableExecutors.getInstance().getTableRefreshingExecutor());
+    addHandlerChain(optimizingService.getTableRuntimeHandler());
+    addHandlerChain(AsyncTableExecutors.getInstance().getSnapshotsExpiringExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getOrphanFilesCleaningExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingCommitExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getOptimizingExpiringExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getBlockerExpiringExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getHiveCommitSyncExecutor());
+    addHandlerChain(AsyncTableExecutors.getInstance().getTableRefreshingExecutor());
     tableService.initialize();
     LOG.info("AMS table service have been initialized");
 
@@ -129,6 +131,12 @@ public class ArcticServiceContainer {
     startHttpService();
   }
 
+  private void addHandlerChain(RuntimeHandlerChain chain) {
+    if (chain != null) {
+      tableService.addHandlerChain(chain);
+    }
+  }
+
   public void dispose() {
     if (server != null) {
       server.stop();
@@ -246,6 +254,7 @@ public class ArcticServiceContainer {
     private void initServiceConfig() throws IOException {
       LOG.info("initializing service configuration...");
       String configPath = Environments.getArcticHome() + SERVER_CONFIG_PATH;
+      LOG.info("load config from path: {}", configPath);
       yamlConfig = new JSONObject(new Yaml().loadAs(Files.newInputStream(Paths.get(configPath)), Map.class));
       JSONObject systemConfig = yamlConfig.getJSONObject(ArcticManagementConf.SYSTEM_CONFIG);
       Map<String, Object> expandedConfigurationMap = Maps.newHashMap();
@@ -257,7 +266,7 @@ public class ArcticServiceContainer {
 
     private void validateConfig(Map<String, Object> systemConfig) {
       if (!systemConfig.containsKey(ArcticManagementConf.SERVER_EXPOSE_HOST.key())) {
-        throw new RuntimeException(
+        throw new IllegalArgumentException(
             "configuration " + ArcticManagementConf.SERVER_EXPOSE_HOST.key() + " must be set");
       }
       InetAddress inetAddress = AmsUtil.lookForBindHost(
@@ -269,7 +278,7 @@ public class ArcticServiceContainer {
           .equalsIgnoreCase(ArcticManagementConf.DB_TYPE_MYSQL)) {
         if (!systemConfig.containsKey(ArcticManagementConf.DB_PASSWORD.key()) ||
             !systemConfig.containsKey(ArcticManagementConf.DB_USER_NAME.key())) {
-          throw new RuntimeException("username and password must be configured if the database type is mysql");
+          throw new IllegalArgumentException("username and password must be configured if the database type is mysql");
         }
       }
 
@@ -277,7 +286,7 @@ public class ArcticServiceContainer {
       if (systemConfig.containsKey(ArcticManagementConf.HA_ENABLE.key()) &&
           ((Boolean) systemConfig.get(ArcticManagementConf.HA_ENABLE.key()))) {
         if (!systemConfig.containsKey(ArcticManagementConf.HA_ZOOKEEPER_ADDRESS.key())) {
-          throw new RuntimeException(
+          throw new IllegalArgumentException(
               ArcticManagementConf.HA_ZOOKEEPER_ADDRESS.key() + " must be configured when you enable " +
                   "the ams high availability");
         }
@@ -286,9 +295,35 @@ public class ArcticServiceContainer {
       String terminalBackend = systemConfig.getOrDefault(ArcticManagementConf.TERMINAL_BACKEND.key(), "")
           .toString().toLowerCase();
       if (!Arrays.asList("local", "kyuubi", "custom").contains(terminalBackend)) {
-        throw new RuntimeException(
+        throw new IllegalArgumentException(
             String.format("Illegal terminal implement: %s, local, kyuubi, custom is available", terminalBackend));
       }
+
+      validateThreadCount(systemConfig, ArcticManagementConf.REFRESH_TABLES_THREAD_COUNT);
+      validateThreadCount(systemConfig, ArcticManagementConf.OPTIMIZING_COMMIT_THREAD_COUNT);
+
+      if (enabled(systemConfig, ArcticManagementConf.EXPIRE_SNAPSHOTS_ENABLED)) {
+        validateThreadCount(systemConfig, ArcticManagementConf.EXPIRE_SNAPSHOTS_THREAD_COUNT);
+      }
+
+      if (enabled(systemConfig, ArcticManagementConf.CLEAN_ORPHAN_FILES_ENABLED)) {
+        validateThreadCount(systemConfig, ArcticManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT);
+      }
+      if (enabled(systemConfig, ArcticManagementConf.SYNC_HIVE_TABLES_ENABLED)) {
+        validateThreadCount(systemConfig, ArcticManagementConf.SYNC_HIVE_TABLES_THREAD_COUNT);
+      }
+    }
+    
+    private boolean enabled(Map<String, Object> systemConfig, ConfigOption<Boolean> config) {
+      return (boolean) systemConfig.getOrDefault(config.key(), config.defaultValue());
+    }
+
+    private void validateThreadCount(Map<String, Object> systemConfig, ConfigOption<Integer> config) {
+      int threadCount = (int) systemConfig.getOrDefault(config.key(), config.defaultValue());
+      if (threadCount <= 0) {
+        throw new IllegalArgumentException(
+            String.format("%s(%s) must > 0, actual value = %d", config.key(), config.description(), threadCount));
+      }
     }
 
     @SuppressWarnings("unchecked")
diff --git a/ams/server/src/main/java/com/netease/arctic/server/table/DefaultTableService.java b/ams/server/src/main/java/com/netease/arctic/server/table/DefaultTableService.java
index 77ea83ba1..cfd619807 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/table/DefaultTableService.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/table/DefaultTableService.java
@@ -54,7 +54,7 @@ public class DefaultTableService extends StatedPersistentBase implements TableSe
 
   public DefaultTableService(Configurations configuration) {
     this.externalCatalogRefreshingInterval =
-        configuration.getLong(ArcticManagementConf.EXTERNAL_CATALOG_REFRESH_INTERVAL);
+        configuration.getLong(ArcticManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL);
     this.blockerTimeout = configuration.getLong(ArcticManagementConf.BLOCKER_TIMEOUT);
   }
 
diff --git a/ams/server/src/main/java/com/netease/arctic/server/table/executor/AsyncTableExecutors.java b/ams/server/src/main/java/com/netease/arctic/server/table/executor/AsyncTableExecutors.java
index a57595289..934ead844 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/table/executor/AsyncTableExecutors.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/table/executor/AsyncTableExecutors.java
@@ -20,19 +20,25 @@ public class AsyncTableExecutors {
   }
 
   public void setup(TableManager tableManager, Configurations conf) {
-    this.snapshotsExpiringExecutor = new SnapshotsExpiringExecutor(tableManager,
-        conf.getInteger(ArcticManagementConf.EXPIRE_THREAD_POOL_SIZE));
-    this.orphanFilesCleaningExecutor = new OrphanFilesCleaningExecutor(tableManager,
-        conf.getInteger(ArcticManagementConf.ORPHAN_CLEAN_THREAD_POOL_SIZE));
+    if (conf.getBoolean(ArcticManagementConf.EXPIRE_SNAPSHOTS_ENABLED)) {
+      this.snapshotsExpiringExecutor = new SnapshotsExpiringExecutor(tableManager,
+          conf.getInteger(ArcticManagementConf.EXPIRE_SNAPSHOTS_THREAD_COUNT));
+    }
+    if (conf.getBoolean(ArcticManagementConf.CLEAN_ORPHAN_FILES_ENABLED)) {
+      this.orphanFilesCleaningExecutor = new OrphanFilesCleaningExecutor(tableManager,
+          conf.getInteger(ArcticManagementConf.CLEAN_ORPHAN_FILES_THREAD_COUNT));
+    }
     this.optimizingCommitExecutor = new OptimizingCommitExecutor(tableManager,
-        conf.getInteger(ArcticManagementConf.OPTIMIZING_COMMIT_THREAD_POOL_SIZE));
+        conf.getInteger(ArcticManagementConf.OPTIMIZING_COMMIT_THREAD_COUNT));
     this.optimizingExpiringExecutor = new OptimizingExpiringExecutor(tableManager);
     this.blockerExpiringExecutor = new BlockerExpiringExecutor(tableManager);
-    this.hiveCommitSyncExecutor = new HiveCommitSyncExecutor(tableManager,
-        conf.getInteger(ArcticManagementConf.SUPPORT_HIVE_SYNC_THREAD_POOL_SIZE));
+    if (conf.getBoolean(ArcticManagementConf.SYNC_HIVE_TABLES_ENABLED)) {
+      this.hiveCommitSyncExecutor = new HiveCommitSyncExecutor(tableManager,
+          conf.getInteger(ArcticManagementConf.SYNC_HIVE_TABLES_THREAD_COUNT));
+    }
     this.tableRefreshingExecutor = new TableRuntimeRefreshExecutor(tableManager,
-        conf.getInteger(ArcticManagementConf.SNAPSHOTS_REFRESHING_THREAD_POOL_SIZE),
-        conf.getLong(ArcticManagementConf.SNAPSHOTS_REFRESHING_INTERVAL));
+        conf.getInteger(ArcticManagementConf.REFRESH_TABLES_THREAD_COUNT),
+        conf.getLong(ArcticManagementConf.REFRESH_TABLES_INTERVAL));
   }
 
   public SnapshotsExpiringExecutor getSnapshotsExpiringExecutor() {
diff --git a/ams/server/src/main/resources/mysql/ams-mysql-init.sql b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
index 7da878e41..4ebfb13c3 100644
--- a/ams/server/src/main/resources/mysql/ams-mysql-init.sql
+++ b/ams/server/src/main/resources/mysql/ams-mysql-init.sql
@@ -46,7 +46,7 @@ CREATE TABLE `optimizer`
 
 CREATE TABLE `resource`
 (
-    `resource_id`               varchar(100) DEFAULT NULL  COMMENT 'optimizer instance id',
+    `resource_id`               varchar(100) NOT NULL  COMMENT 'optimizer instance id',
     `resource_type`             tinyint(4) DEFAULT 0 COMMENT 'resource type like optimizer/ingestor',
     `container_name`            varchar(100) DEFAULT NULL  COMMENT 'container name',
     `group_name`                varchar(50) DEFAULT NULL COMMENT 'queue name',
diff --git a/core/src/main/java/com/netease/arctic/scan/BasicKeyedTableScan.java b/core/src/main/java/com/netease/arctic/scan/BasicKeyedTableScan.java
index 485780711..0ed33afe3 100644
--- a/core/src/main/java/com/netease/arctic/scan/BasicKeyedTableScan.java
+++ b/core/src/main/java/com/netease/arctic/scan/BasicKeyedTableScan.java
@@ -64,6 +64,7 @@ public class BasicKeyedTableScan implements KeyedTableScan {
   private final int lookBack;
   private final long openFileCost;
   private final long splitSize;
+  private Double splitTaskByDeleteRatio;
   private Expression expression;
 
   public BasicKeyedTableScan(BasicKeyedTable table) {
@@ -121,13 +122,20 @@ public class BasicKeyedTableScan implements KeyedTableScan {
         splitSize, lookBack, openFileCost);
   }
 
+  @Override
+  public KeyedTableScan enableSplitTaskByDeleteRatio(double splitTaskByDeleteRatio) {
+    this.splitTaskByDeleteRatio = splitTaskByDeleteRatio;
+    return this;
+  }
+
   private CloseableIterable<ArcticFileScanTask> planBaseFiles() {
     TableScan scan = table.baseTable().newScan();
     if (this.expression != null) {
       scan = scan.filter(this.expression);
     }
     CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles();
-    return CloseableIterable.transform(fileScanTasks,
+    return CloseableIterable.transform(
+        fileScanTasks,
         fileScanTask -> new BasicArcticFileScanTask(DefaultKeyedFile.parseBase(fileScanTask.file()),
             fileScanTask.deletes(), fileScanTask.spec(), expression));
   }
@@ -153,23 +161,44 @@ public class BasicKeyedTableScan implements KeyedTableScan {
   private void split() {
     fileScanTasks.forEach((structLike, fileScanTasks1) -> {
       for (NodeFileScanTask task : fileScanTasks1) {
-        if (task.cost() <= splitSize) {
+        if (task.dataTasks().size() < 2) {
           splitTasks.add(task);
           continue;
         }
-        if (task.dataTasks().size() < 2) {
+
+        if (splitTaskByDeleteRatio != null) {
+          long deleteWeight = task.arcticEquityDeletes().stream().mapToLong(s -> s.file().fileSizeInBytes())
+              .map(s -> s + openFileCost)
+              .sum();
+
+          long dataWeight = task.dataTasks().stream().mapToLong(s -> s.file().fileSizeInBytes())
+              .map(s -> s + openFileCost)
+              .sum();
+          double deleteRatio = deleteWeight * 1.0 / dataWeight;
+
+          if (deleteRatio < splitTaskByDeleteRatio) {
+            long targetSize = Math.min(new Double(deleteWeight / splitTaskByDeleteRatio).longValue(), splitSize);
+            split(task, targetSize);
+            continue;
+          }
+        }
+
+        if (task.cost() <= splitSize) {
           splitTasks.add(task);
           continue;
         }
-        CloseableIterable<NodeFileScanTask> tasksIterable = splitNode(CloseableIterable.withNoopClose(task.dataTasks()),
-            task.arcticEquityDeletes(), splitSize, lookBack, openFileCost);
-        List<NodeFileScanTask> tasks =
-            Lists.newArrayList(tasksIterable);
-        splitTasks.addAll(tasks);
+        split(task, splitSize);
       }
     });
   }
 
+  private void split(NodeFileScanTask task, long targetSize) {
+    CloseableIterable<NodeFileScanTask> tasksIterable =
+        splitNode(CloseableIterable.withNoopClose(task.dataTasks()),
+            task.arcticEquityDeletes(), targetSize, lookBack, openFileCost);
+    splitTasks.addAll(Lists.newArrayList(tasksIterable));
+  }
+
   public CloseableIterable<NodeFileScanTask> splitNode(
       CloseableIterable<ArcticFileScanTask> splitFiles,
       List<ArcticFileScanTask> deleteFiles,
diff --git a/core/src/main/java/com/netease/arctic/scan/KeyedTableScan.java b/core/src/main/java/com/netease/arctic/scan/KeyedTableScan.java
index 0e4d79b4d..8e7c64989 100644
--- a/core/src/main/java/com/netease/arctic/scan/KeyedTableScan.java
+++ b/core/src/main/java/com/netease/arctic/scan/KeyedTableScan.java
@@ -41,4 +41,11 @@ public interface KeyedTableScan {
    */
   CloseableIterable<CombinedScanTask> planTasks();
 
+  /**
+   * To enable the feature of splitting tasks based on the delete ratio,
+   * If the delete ratio under a node is less than splitTaskByDeleteRatio value,
+   * the plan will be split into as many tasks as possible during execution.
+   */
+  KeyedTableScan enableSplitTaskByDeleteRatio(double splitTaskByDeleteRatio);
+
 }
diff --git a/core/src/main/java/com/netease/arctic/utils/ManifestEntryFields.java b/core/src/main/java/com/netease/arctic/utils/ManifestEntryFields.java
index cc6472e42..83317336f 100644
--- a/core/src/main/java/com/netease/arctic/utils/ManifestEntryFields.java
+++ b/core/src/main/java/com/netease/arctic/utils/ManifestEntryFields.java
@@ -27,7 +27,9 @@ public class ManifestEntryFields {
   public static final Types.NestedField STATUS = required(0, "status", Types.IntegerType.get());
   public static final Types.NestedField SNAPSHOT_ID = optional(1, "snapshot_id", Types.LongType.get());
   public static final Types.NestedField SEQUENCE_NUMBER = optional(3, "sequence_number", Types.LongType.get());
+
   public static final int DATA_FILE_ID = 3;
+
   public static final String DATA_FILE_FIELD_NAME = "data_file";
 
   public enum Status {
diff --git a/dist/src/main/arctic-bin/conf/config.yaml b/dist/src/main/arctic-bin/conf/config.yaml
index 3e764feae..9cde1ddd6 100644
--- a/dist/src/main/arctic-bin/conf/config.yaml
+++ b/dist/src/main/arctic-bin/conf/config.yaml
@@ -3,15 +3,6 @@ ams:
   admin-password: admin
   server-bind-host: "0.0.0.0"
   server-expose-host: "127.0.0.1"
-  refresh-external-catalog-interval: 180000 # 3min
-  refresh-table-thread-count: 10
-  refresh-table-interval: 60000 #1min
-  expire-table-thread-count: 10
-  clean-orphan-file-thread-count: 10
-  sync-hive-tables-thread-count: 10
-
-  blocker:
-    timeout: 60000 # 1min
 
   thrift-server:
     bind-port: 1260
@@ -23,6 +14,13 @@ ams:
   http-server:
     bind-port: 1630
 
+  refresh-external-catalogs:
+    interval: 180000 # 3min
+
+  refresh-tables:
+    thread-count: 10
+    interval: 60000 # 1min
+    
   self-optimizing:
     commit-thread-count: 10
 
@@ -30,6 +28,22 @@ ams:
     heart-beat-timeout: 60000 # 1min
     task-ack-timeout: 30000 # 30s
 
+  blocker:
+    timeout: 60000 # 1min
+
+  # optional features
+  expire-snapshots:
+    enabled: true
+    thread-count: 10
+
+  clean-orphan-files:
+    enabled: true
+    thread-count: 10
+
+  sync-hive-tables:
+    enabled: true
+    thread-count: 10
+
   database:
     type: derby
     jdbc-driver-class: org.apache.derby.jdbc.EmbeddedDriver
diff --git a/site/docs/ch/configurations.md b/site/docs/ch/configurations.md
index f584abbd9..05aaa7d1c 100644
--- a/site/docs/ch/configurations.md
+++ b/site/docs/ch/configurations.md
@@ -44,11 +44,12 @@ Self-optimizing 配置对 Iceberg format, Mixed streaming format 都会生效。
 
 ### 表读取配置
 
-| 配置名称                            | 默认值             | 描述                                     |
-| ---------------------------------- | ---------------- | ----------------------------------       |
-| read.split.open-file-cost          | 4194304（4MB）    | 预估与读取开销等价的打开一个文件开销，一般不改                        |
-| read.split.planning-lookback       | 10               | 拆分读取任务所使用分桶算法里桶的个数               |
-| read.split.target-size              | 134217728（128MB）| 查询引擎中的读取任务会尽量拆分成这个大小                     |
+| 配置名称                            | 默认值             | 描述                                                                                                                                                       |
+| ---------------------------------- | ---------------- |----------------------------------------------------------------------------------------------------------------------------------------------------------|
+| read.split.open-file-cost          | 4194304（4MB）    | 预估与读取开销等价的打开一个文件开销，一般不改                                                                                                                                  |
+| read.split.planning-lookback       | 10               | 拆分读取任务所使用分桶算法里桶的个数                                                                                                                                       |
+| read.split.target-size              | 134217728（128MB）| 查询引擎中的读取任务会尽量拆分成这个大小                                                                                                                                     |
+| read.split.delete-ratio            | 0.05             |Delete文件占比小于此阈值时，读取任务会拆分成更多的任务以提高查询速度。 |
 
 ### 表写入配置
 
diff --git a/site/docs/ch/guides/deployment.md b/site/docs/ch/guides/deployment.md
index a51591b86..1cca7d9a9 100644
--- a/site/docs/ch/guides/deployment.md
+++ b/site/docs/ch/guides/deployment.md
@@ -189,15 +189,6 @@ ams:
   admin-password: admin
   server-bind-host: "0.0.0.0"
   server-expose-host: "127.0.0.1"
-  refresh-external-catalog-interval: 180000 # 3min
-  refresh-table-thread-count: 10
-  refresh-table-interval: 60000 #1min
-  expire-table-thread-count: 10
-  clean-orphan-file-thread-count: 10
-  sync-hive-tables-thread-count: 10
-  
-  blocker:
-    timeout: 60000 # 1min
 
   thrift-server:
     bind-port: 1260
@@ -208,6 +199,13 @@ ams:
 
   http-server:
     bind-port: 1630
+    
+  refresh-external-catalogs:
+    interval: 180000 # 3min
+
+  refresh-tables:
+    thread-count: 10
+    interval: 60000 # 1min
 
   self-optimizing:
     commit-thread-count: 10
@@ -215,6 +213,22 @@ ams:
   optimizer:
     heart-beat-timeout: 60000 # 1min
     task-ack-timeout: 30000 # 30s
+    
+  blocker:
+    timeout: 60000 # 1min
+    
+  # optional features
+  expire-snapshots:
+    enabled: true
+    thread-count: 10
+
+  clean-orphan-files:
+    enabled: true
+    thread-count: 10
+
+  sync-hive-tables:
+    enabled: true
+    thread-count: 10
 
   database:
     type: derby
diff --git a/trino/src/main/java/com/netease/arctic/trino/ArcticConfig.java b/trino/src/main/java/com/netease/arctic/trino/ArcticConfig.java
index 211208dd8..61ddfcb9a 100644
--- a/trino/src/main/java/com/netease/arctic/trino/ArcticConfig.java
+++ b/trino/src/main/java/com/netease/arctic/trino/ArcticConfig.java
@@ -30,6 +30,10 @@ public class ArcticConfig {
   private boolean hdfsImpersonationEnabled;
   private boolean tableStatisticsEnabled = true;
 
+  private Double splitTaskByDeleteRatio = 0.05;
+
+  private boolean enableSplitTaskByDeleteRatio = true;
+
   public String getCatalogUrl() {
     return catalogUrl;
   }
@@ -42,6 +46,14 @@ public class ArcticConfig {
     return tableStatisticsEnabled;
   }
 
+  public Double getSplitTaskByDeleteRatio() {
+    return splitTaskByDeleteRatio;
+  }
+
+  public boolean isEnableSplitTaskByDeleteRatio() {
+    return enableSplitTaskByDeleteRatio;
+  }
+
   @Config("arctic.url")
   public void setCatalogUrl(String catalogUrl) {
     this.catalogUrl = catalogUrl;
@@ -57,4 +69,14 @@ public class ArcticConfig {
   public void setTableStatisticsEnabled(boolean tableStatisticsEnabled) {
     this.tableStatisticsEnabled = tableStatisticsEnabled;
   }
+
+  @Config("arctic.enable-split-task-by-delete-ratio")
+  public void setEnableSplitTaskByDeleteRatio(boolean enableSplitTaskByDeleteRatio) {
+    this.enableSplitTaskByDeleteRatio = enableSplitTaskByDeleteRatio;
+  }
+
+  @Config("arctic.split-task-by-delete-ratio")
+  public void setSplitTaskByDeleteRatio(double splitTaskByDeleteRatio) {
+    this.splitTaskByDeleteRatio = splitTaskByDeleteRatio;
+  }
 }
diff --git a/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java b/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
index ccc54c3c0..62e24666b 100644
--- a/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
+++ b/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
@@ -29,6 +29,7 @@ import javax.inject.Inject;
 import java.util.List;
 
 import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+import static io.trino.spi.session.PropertyMetadata.doubleProperty;
 
 /**
  * Arctic supporting session properties
@@ -37,6 +38,9 @@ public final class ArcticSessionProperties
     implements SessionPropertiesProvider {
 
   private static final String ARCTIC_STATISTICS_ENABLED = "arctic_table_statistics_enabled";
+
+  private static final String ARCTIC_SPLIT_TASK_BY_DELETE_RATIO = "arctic_split_task_by_delete_ratio";
+  private static final String ARCTIC_ENABLE_SPLIT_TASK_BY_DELETE_RATIO = "arctic_enable_split_task_by_delete_ratio";
   private final List<PropertyMetadata<?>> sessionProperties;
 
   @Inject
@@ -50,6 +54,18 @@ public final class ArcticSessionProperties
             "Expose table statistics for Arctic table",
             arcticConfig.isTableStatisticsEnabled(),
             false))
+        .add(doubleProperty(
+            ARCTIC_SPLIT_TASK_BY_DELETE_RATIO,
+            "If task delete ratio less than this value will be split to more task",
+            arcticConfig.getSplitTaskByDeleteRatio(),
+            false
+        ))
+        .add(booleanProperty(
+          ARCTIC_ENABLE_SPLIT_TASK_BY_DELETE_RATIO,
+            "Enable task split by ratio",
+            arcticConfig.isEnableSplitTaskByDeleteRatio(),
+            false
+        ))
         .build();
   }
 
@@ -61,4 +77,12 @@ public final class ArcticSessionProperties
   public static boolean isArcticStatisticsEnabled(ConnectorSession session) {
     return session.getProperty(ARCTIC_STATISTICS_ENABLED, Boolean.class);
   }
+
+  public static boolean enableSplitTaskByDeleteRatio(ConnectorSession session) {
+    return session.getProperty(ARCTIC_ENABLE_SPLIT_TASK_BY_DELETE_RATIO, Boolean.class);
+  }
+
+  public static double splitTaskByDeleteRatio(ConnectorSession session) {
+    return session.getProperty(ARCTIC_SPLIT_TASK_BY_DELETE_RATIO, Double.class);
+  }
 }
diff --git a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
index 25109585d..3f1623996 100644
--- a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
+++ b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
@@ -24,6 +24,7 @@ import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.scan.KeyedTableScan;
 import com.netease.arctic.scan.KeyedTableScanTask;
 import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.trino.ArcticSessionProperties;
 import com.netease.arctic.trino.ArcticTransactionManager;
 import com.netease.arctic.trino.util.MetricUtil;
 import com.netease.arctic.trino.util.ObjectSerializerUtil;
@@ -91,6 +92,10 @@ public class KeyedConnectorSplitManager implements ConnectorSplitManager {
         .filter(toIcebergExpression(
             icebergTableHandle.getEnforcedPredicate().intersect(icebergTableHandle.getUnenforcedPredicate())));
 
+    if (ArcticSessionProperties.enableSplitTaskByDeleteRatio(session)) {
+      tableScan.enableSplitTaskByDeleteRatio(ArcticSessionProperties.splitTaskByDeleteRatio(session));
+    }
+
     ClassLoader pluginClassloader = arcticTable.getClass().getClassLoader();
 
     try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(pluginClassloader)) {


Test patch:
diff --git a/ams/server/src/test/java/com/netease/arctic/server/AmsEnvironment.java b/ams/server/src/test/java/com/netease/arctic/server/AmsEnvironment.java
index c5d096818..6177966c3 100644
--- a/ams/server/src/test/java/com/netease/arctic/server/AmsEnvironment.java
+++ b/ams/server/src/test/java/com/netease/arctic/server/AmsEnvironment.java
@@ -208,7 +208,7 @@ public class AmsEnvironment {
                 DynFields.builder().hiddenImpl(ArcticServiceContainer.class, "serviceConfig").build();
             serviceConfig = field.bind(arcticService).get();
             serviceConfig.set(ArcticManagementConf.THRIFT_BIND_PORT, thriftBindPort);
-            serviceConfig.set(ArcticManagementConf.EXTERNAL_CATALOG_REFRESH_INTERVAL, 1000L);
+            serviceConfig.set(ArcticManagementConf.REFRESH_EXTERNAL_CATALOGS_INTERVAL, 1000L);
             // when AMS is successfully running, this thread will wait here
             arcticService.startService();
             break;
diff --git a/core/src/test/java/com/netease/arctic/scan/TestScanSplitTask.java b/core/src/test/java/com/netease/arctic/scan/TestScanSplitTask.java
new file mode 100644
index 000000000..08918a7c6
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/scan/TestScanSplitTask.java
@@ -0,0 +1,77 @@
+/*
+ * 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 com.netease.arctic.scan;
+
+import com.netease.arctic.io.DataTestHelpers;
+import com.netease.arctic.io.TableDataTestBase;
+import com.netease.arctic.io.writer.GenericChangeTaskWriter;
+import com.netease.arctic.io.writer.GenericTaskWriters;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public class TestScanSplitTask extends TableDataTestBase {
+  @Test
+  public void testSplitTaskByDeleteRatio() throws IOException {
+    writeInsertFileIntoBaseStore();
+    writeInsertFileIntoBaseStore();
+    {
+      CloseableIterable<CombinedScanTask> combinedScanTasks =
+          getArcticTable().asKeyedTable().newScan().planTasks();
+      for (CombinedScanTask combinedScanTask : combinedScanTasks) {
+        Assert.assertEquals(combinedScanTask.tasks().size(), 7);
+      }
+    }
+    {
+      CloseableIterable<CombinedScanTask> combinedScanTasks =
+          getArcticTable().asKeyedTable().newScan().enableSplitTaskByDeleteRatio(0.04).planTasks();
+      for (CombinedScanTask combinedScanTask : combinedScanTasks) {
+        // If enableSplitTaskByDeleteRatio is turned on, tasks with delete content below the threshold will be split
+        // as much as possible. The above normal plan has 7 tasks, but two of them contain two files each and no
+        // delete content. Therefore, after enableSplitTaskByDeleteRatio is turned on, two additional tasks will be
+        // split out.
+        Assert.assertEquals(combinedScanTask.tasks().size(), 9);
+      }
+    }
+  }
+
+  private void writeInsertFileIntoBaseStore() throws IOException {
+    ImmutableList.Builder<Record> builder = ImmutableList.builder();
+    builder.add(DataTestHelpers.createRecord(7, "mary", 0, "2022-01-01T12:00:00"));
+    builder.add(DataTestHelpers.createRecord(8, "mack", 0, "2022-01-01T12:00:00"));
+    ImmutableList<Record> records = builder.build();
+
+    GenericChangeTaskWriter writer = GenericTaskWriters.builderFor(getArcticTable().asKeyedTable())
+        .withTransactionId(5L).buildChangeWriter();
+    for (Record record : records) {
+      writer.write(record);
+    }
+    WriteResult result = writer.complete();
+    AppendFiles baseAppend = getArcticTable().asKeyedTable().baseTable().newAppend();
+    Arrays.stream(result.dataFiles()).forEach(baseAppend::appendFile);
+    baseAppend.commit();
+  }
+}
\ No newline at end of file
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
index 3da0ea595..d129e15e1 100644
--- a/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
@@ -39,8 +39,12 @@ public class TestBaseArcticPrimaryTable extends TableTestBaseWithInitDataForTrin
     setupTables();
     initData();
     return ArcticQueryRunner.builder()
-        .setIcebergProperties(ImmutableMap.of("arctic.url",
-            String.format("thrift://localhost:%s/%s", AMS.port(), TEST_CATALOG_NAME)))
+        .setIcebergProperties(ImmutableMap.of(
+            "arctic.url",
+            String.format("thrift://localhost:%s/%s", AMS.port(), TEST_CATALOG_NAME),
+            "arctic.enable-split-task-by-delete-ratio",
+            "true"
+        ))
         .build();
   }