diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
index e919a1c5a5f2d..964a67155d881 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
@@ -178,6 +178,30 @@ public void testRejectedHotReloadDoesNotUpdateAppliedConfiguration() {
                     checkConfigFileContains(nodeWrapper, "heartbeat_interval_in_ms=1000")));
   }
 
+  @Test
+  public void testHotReloadRegionMigrationFileRemoveSpeedLimit() {
+    String key = "region_migration_file_remove_speed_limit_bytes_per_second";
+    int dataNodeId = EnvFactory.getEnv().getConfigNodeWrapperList().size();
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("set configuration \"" + key + "\"=\"65536\" on " + dataNodeId);
+      assertAppliedConfiguration(dataNodeId, key, "65536");
+      Assert.assertTrue(
+          checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key + "=65536"));
+    } catch (Exception e) {
+      Assert.fail(e.getMessage());
+    } finally {
+      try (Connection connection = EnvFactory.getEnv().getConnection();
+          Statement statement = connection.createStatement()) {
+        statement.execute("set configuration \"" + key + "\"=\"16777216\" on " + dataNodeId);
+      } catch (Exception e) {
+        Assert.fail(e.getMessage());
+      }
+    }
+    Assert.assertTrue(
+        checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key + "=16777216"));
+  }
+
   @Test
   public void testHotReloadContinuousQueryMinEveryInterval() {
     try (Connection connection = EnvFactory.getEnv().getConnection();
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 62f5af9c86f04..1e247ad599e1c 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -36,6 +36,7 @@
 import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
 import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
 import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.commons.utils.StatusUtils;
 import org.apache.iotdb.consensus.IConsensus;
 import org.apache.iotdb.consensus.IStateMachine;
@@ -378,7 +379,9 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException
     if (!exist.get()) {
       throw new ConsensusGroupNotExistException(groupId);
     }
-    FileUtils.deleteFileOrDirectory(new File(buildPeerDir(storageDir, groupId)));
+    FileUtils.deleteFileOrDirectoryWithRateLimiter(
+        new File(buildPeerDir(storageDir, groupId)),
+        RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
     KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
   }
 
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
index bb6b95b45ad56..f5607874019b4 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
@@ -19,22 +19,16 @@
 
 package org.apache.iotdb.consensus.iot.snapshot;
 
-import com.google.common.util.concurrent.RateLimiter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
 
 public class IoTConsensusRateLimiter {
-  private static final Logger logger = LoggerFactory.getLogger(IoTConsensusRateLimiter.class);
 
-  private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
+  private final RegionMigrationRateLimiter rateLimiter = RegionMigrationRateLimiter.getInstance();
 
   private IoTConsensusRateLimiter() {}
 
   public void init(long regionMigrationSpeedLimitBytesPerSecond) {
-    rateLimiter.setRate(
-        regionMigrationSpeedLimitBytesPerSecond <= 0
-            ? Double.MAX_VALUE
-            : regionMigrationSpeedLimitBytesPerSecond);
+    rateLimiter.init(regionMigrationSpeedLimitBytesPerSecond);
   }
 
   /**
@@ -43,15 +37,7 @@ public void init(long regionMigrationSpeedLimitBytesPerSecond) {
    * @param transitDataSize the size of the data to be sent
    */
   public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) {
-    while (transitDataSize > 0) {
-      if (transitDataSize > Integer.MAX_VALUE) {
-        rateLimiter.acquire(Integer.MAX_VALUE);
-        transitDataSize -= Integer.MAX_VALUE;
-      } else {
-        rateLimiter.acquire((int) transitDataSize);
-        return;
-      }
-    }
+    rateLimiter.acquire(transitDataSize);
   }
 
   private static final IoTConsensusRateLimiter INSTANCE = new IoTConsensusRateLimiter();
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
index 806f499116d5a..cf1d405639838 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
@@ -34,6 +34,7 @@
 import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
 import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
 import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.commons.utils.StatusUtils;
 import org.apache.iotdb.consensus.IConsensus;
 import org.apache.iotdb.consensus.IStateMachine;
@@ -323,7 +324,9 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException
         consensus.clear();
         stateMachineMap.remove(groupId);
 
-        FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
+        FileUtils.deleteFileOrDirectoryWithRateLimiter(
+            new File(getPeerDir(groupId)),
+            RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
         KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
         LOGGER.info(IoTConsensusV2Messages.FINISH_DELETE_LOCAL_PEER, CLASS_NAME, groupId);
       } finally {
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java
new file mode 100644
index 0000000000000..1124d04b8651f
--- /dev/null
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.grpc.GrpcFactory;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.leader.FollowerInfo;
+import org.apache.ratis.server.leader.LeaderState;
+import org.apache.ratis.server.leader.LogAppender;
+
+class RateLimitedGrpcFactory extends GrpcFactory {
+
+  RateLimitedGrpcFactory(Parameters parameters) {
+    super(parameters);
+  }
+
+  @Override
+  public LogAppender newLogAppender(
+      RaftServer.Division server, LeaderState leaderState, FollowerInfo follower) {
+    return new RateLimitedGrpcLogAppender(server, leaderState, follower);
+  }
+}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java
new file mode 100644
index 0000000000000..41830c85dc2f6
--- /dev/null
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java
@@ -0,0 +1,76 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
+
+import org.apache.ratis.grpc.server.GrpcLogAppender;
+import org.apache.ratis.proto.RaftProtos.FileChunkProto;
+import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.leader.FollowerInfo;
+import org.apache.ratis.server.leader.LeaderState;
+import org.apache.ratis.statemachine.SnapshotInfo;
+
+import java.util.Iterator;
+
+class RateLimitedGrpcLogAppender extends GrpcLogAppender {
+
+  private final RegionMigrationRateLimiter rateLimiter = RegionMigrationRateLimiter.getInstance();
+
+  RateLimitedGrpcLogAppender(
+      RaftServer.Division server, LeaderState leaderState, FollowerInfo follower) {
+    super(server, leaderState, follower);
+  }
+
+  @Override
+  public Iterable<InstallSnapshotRequestProto> newInstallSnapshotRequests(
+      String requestId, SnapshotInfo snapshot) {
+    final Iterable<InstallSnapshotRequestProto> requests =
+        super.newInstallSnapshotRequests(requestId, snapshot);
+    return () -> {
+      final Iterator<InstallSnapshotRequestProto> iterator = requests.iterator();
+      return new Iterator<InstallSnapshotRequestProto>() {
+        @Override
+        public boolean hasNext() {
+          return iterator.hasNext();
+        }
+
+        @Override
+        public InstallSnapshotRequestProto next() {
+          final InstallSnapshotRequestProto request = iterator.next();
+          rateLimiter.acquire(getSnapshotChunkDataSize(request));
+          return request;
+        }
+      };
+    };
+  }
+
+  static long getSnapshotChunkDataSize(InstallSnapshotRequestProto request) {
+    if (!request.hasSnapshotChunk()) {
+      return 0;
+    }
+
+    return request.getSnapshotChunk().getFileChunksList().stream()
+        .map(FileChunkProto::getData)
+        .mapToLong(data -> data == null ? 0 : data.size())
+        .sum();
+  }
+}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java
new file mode 100644
index 0000000000000..391ec2faef608
--- /dev/null
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java
@@ -0,0 +1,37 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.rpc.RpcFactory;
+import org.apache.ratis.rpc.RpcType;
+
+public class RateLimitedGrpcRpcType implements RpcType {
+
+  @Override
+  public String name() {
+    return RateLimitedGrpcRpcType.class.getName();
+  }
+
+  @Override
+  public RpcFactory newFactory(Parameters parameters) {
+    return new RateLimitedGrpcFactory(parameters);
+  }
+}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index b7369546e7cde..4cac559646fd8 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -56,6 +56,7 @@
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
+import org.apache.ratis.RaftConfigKeys;
 import org.apache.ratis.client.RaftClientRpc;
 import org.apache.ratis.conf.Parameters;
 import org.apache.ratis.conf.RaftProperties;
@@ -159,6 +160,7 @@ public RatisConsensus(ConsensusConfig config, IStateMachine.Registry registry) {
     this.storageDir = new File(config.getStorageDir());
 
     RaftServerConfigKeys.setStorageDir(properties, Collections.singletonList(storageDir));
+    RaftConfigKeys.Rpc.setType(properties, new RateLimitedGrpcRpcType());
     GrpcConfigKeys.Server.setHost(properties, config.getThisNodeEndPoint().getIp());
     GrpcConfigKeys.Server.setPort(properties, config.getThisNodeEndPoint().getPort());
 
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
index 273411a1b57a8..06fb39ba7e01e 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
@@ -26,6 +26,7 @@
 import org.apache.iotdb.commons.request.IConsensusRequest;
 import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.commons.utils.StatusUtils;
 import org.apache.iotdb.consensus.IConsensus;
 import org.apache.iotdb.consensus.IStateMachine;
@@ -197,7 +198,9 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException
         (k, v) -> {
           exist.set(true);
           v.stop();
-          FileUtils.deleteFileOrDirectory(new File(buildPeerDir(groupId)));
+          FileUtils.deleteFileOrDirectoryWithRateLimiter(
+              new File(buildPeerDir(groupId)),
+              RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
           return null;
         });
     if (!exist.get()) {
diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java
new file mode 100644
index 0000000000000..8b93bdcdd1a50
--- /dev/null
+++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.ratis.RaftConfigKeys;
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.proto.RaftProtos.FileChunkProto;
+import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
+import org.apache.ratis.rpc.RpcType;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RateLimitedGrpcLogAppenderTest {
+
+  @Test
+  public void testGetSnapshotChunkDataSize() {
+    final InstallSnapshotRequestProto request =
+        InstallSnapshotRequestProto.newBuilder()
+            .setSnapshotChunk(
+                InstallSnapshotRequestProto.SnapshotChunkProto.newBuilder()
+                    .addFileChunks(
+                        FileChunkProto.newBuilder()
+                            .setData(ByteString.copyFrom(new byte[3]))
+                            .build())
+                    .addFileChunks(
+                        FileChunkProto.newBuilder()
+                            .setData(ByteString.copyFrom(new byte[5]))
+                            .build()))
+            .build();
+
+    Assert.assertEquals(8, RateLimitedGrpcLogAppender.getSnapshotChunkDataSize(request));
+    Assert.assertEquals(
+        0,
+        RateLimitedGrpcLogAppender.getSnapshotChunkDataSize(
+            InstallSnapshotRequestProto.newBuilder().buildPartial()));
+  }
+
+  @Test
+  public void testRateLimitedGrpcRpcTypeIsResolvedByRatis() {
+    final RaftProperties properties = new RaftProperties();
+
+    RaftConfigKeys.Rpc.setType(properties, new RateLimitedGrpcRpcType());
+    final RpcType rpcType = RaftConfigKeys.Rpc.type(properties, ignored -> {});
+
+    Assert.assertTrue(rpcType instanceof RateLimitedGrpcRpcType);
+    Assert.assertTrue(rpcType.newFactory(new Parameters()) instanceof RateLimitedGrpcFactory);
+  }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index b0c78e496e706..fe74770d609ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -1101,6 +1101,7 @@ public class IoTDBConfig {
   private int maxPendingBatchesNum = 5;
   private double maxMemoryRatioForQueue = 0.6;
   private long regionMigrationSpeedLimitBytesPerSecond = 48 * 1024 * 1024L;
+  private long regionMigrationFileRemoveSpeedLimitBytesPerSecond = 16 * 1024 * 1024L;
   // Throttle the per-file snapshot-transmission progress log in IoTConsensus to at most once per
   // this interval (ms). A value <= 0 logs every file.
   private long dataRegionIotSnapshotTransmissionProgressLogIntervalMs = 5000L;
@@ -1278,6 +1279,16 @@ public void setRegionMigrationSpeedLimitBytesPerSecond(
     this.regionMigrationSpeedLimitBytesPerSecond = regionMigrationSpeedLimitBytesPerSecond;
   }
 
+  public long getRegionMigrationFileRemoveSpeedLimitBytesPerSecond() {
+    return regionMigrationFileRemoveSpeedLimitBytesPerSecond;
+  }
+
+  public void setRegionMigrationFileRemoveSpeedLimitBytesPerSecond(
+      long regionMigrationFileRemoveSpeedLimitBytesPerSecond) {
+    this.regionMigrationFileRemoveSpeedLimitBytesPerSecond =
+        regionMigrationFileRemoveSpeedLimitBytesPerSecond;
+  }
+
   public long getDataRegionIotSnapshotTransmissionProgressLogIntervalMs() {
     return dataRegionIotSnapshotTransmissionProgressLogIntervalMs;
   }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index ddbceab5cd7ef..bcccc7f95d727 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -33,6 +33,8 @@
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.utils.JVMCommonUtils;
 import org.apache.iotdb.commons.utils.NodeUrlUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
 import org.apache.iotdb.confignode.rpc.thrift.TCQConfig;
 import org.apache.iotdb.confignode.rpc.thrift.TGlobalConfig;
 import org.apache.iotdb.confignode.rpc.thrift.TRatisConfig;
@@ -1290,6 +1292,16 @@ private void loadIoTConsensusProps(TrimProperties properties) throws IOException
                 "region_migration_speed_limit_bytes_per_second",
                 ConfigurationFileUtils.getConfigurationDefaultValue(
                     "region_migration_speed_limit_bytes_per_second"))));
+    RegionMigrationRateLimiter.getInstance()
+        .init(conf.getRegionMigrationSpeedLimitBytesPerSecond());
+    conf.setRegionMigrationFileRemoveSpeedLimitBytesPerSecond(
+        Long.parseLong(
+            properties.getProperty(
+                "region_migration_file_remove_speed_limit_bytes_per_second",
+                ConfigurationFileUtils.getConfigurationDefaultValue(
+                    "region_migration_file_remove_speed_limit_bytes_per_second"))));
+    RegionMigrationFileRemoveRateLimiter.getInstance()
+        .init(conf.getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
     conf.setDataRegionIotSnapshotTransmissionProgressLogIntervalMs(
         Long.parseLong(
             properties.getProperty(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
index 7f82ce8e9a9a0..b608182f13419 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
@@ -32,7 +32,9 @@
 import org.apache.iotdb.commons.pipe.sink.payload.iotconsensusv2.response.IoTConsensusV2TransferFilePieceResp;
 import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
 import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId;
+import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq;
 import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferResp;
+import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq;
 import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferResp;
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics;
@@ -196,9 +198,9 @@ private void doTransfer() {
     try (final SyncIoTConsensusV2ServiceClient syncIoTConsensusV2ServiceClient =
         syncRetryClientManager.borrowClient(getFollowerUrl())) {
       final TIoTConsensusV2BatchTransferResp resp;
-      resp =
-          syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(
-              tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq());
+      final TIoTConsensusV2BatchTransferReq req =
+          tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq();
+      resp = syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(req);
 
       final List<TSStatus> statusList =
           resp.getBatchResps().stream()
@@ -258,14 +260,14 @@ private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent)
     try (final SyncIoTConsensusV2ServiceClient syncIoTConsensusV2ServiceClient =
         syncRetryClientManager.borrowClient(getFollowerUrl())) {
       progressIndex = pipeDeleteDataNodeEvent.getProgressIndex();
-      resp =
-          syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(
-              IoTConsensusV2DeleteNodeReq.toTIoTConsensusV2TransferReq(
-                  pipeDeleteDataNodeEvent.getDeleteDataNode(),
-                  tCommitId,
-                  tConsensusGroupId,
-                  progressIndex,
-                  thisDataNodeId));
+      final TIoTConsensusV2TransferReq req =
+          IoTConsensusV2DeleteNodeReq.toTIoTConsensusV2TransferReq(
+              pipeDeleteDataNodeEvent.getDeleteDataNode(),
+              tCommitId,
+              tConsensusGroupId,
+              progressIndex,
+              thisDataNodeId);
+      resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req);
     } catch (final Exception e) {
       throw new PipeRuntimeSinkRetryTimesConfigurableException(
           String.format(
@@ -330,10 +332,10 @@ private void doTransfer(PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletI
       insertNode = pipeInsertNodeTabletInsertionEvent.getInsertNode();
       progressIndex = pipeInsertNodeTabletInsertionEvent.getProgressIndex();
 
-      resp =
-          syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(
-              IoTConsensusV2TabletInsertNodeReq.toTIoTConsensusV2TransferReq(
-                  insertNode, tCommitId, tConsensusGroupId, progressIndex, thisDataNodeId));
+      final TIoTConsensusV2TransferReq req =
+          IoTConsensusV2TabletInsertNodeReq.toTIoTConsensusV2TransferReq(
+              insertNode, tCommitId, tConsensusGroupId, progressIndex, thisDataNodeId);
+      resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req);
     } catch (final Exception e) {
       throw new PipeRuntimeSinkRetryTimesConfigurableException(
           String.format(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
index 3b3ba14a238a7..df11b4b037ddd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
 
 import org.slf4j.Logger;
@@ -29,6 +30,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.util.Objects;
+import java.util.function.LongConsumer;
 
 public class SchemaRegionUtils {
 
@@ -38,6 +40,12 @@ private SchemaRegionUtils() {
 
   public static void deleteSchemaRegionFolder(String schemaRegionDirPath, Logger logger)
       throws MetadataException {
+    deleteSchemaRegionFolder(schemaRegionDirPath, logger, null);
+  }
+
+  public static void deleteSchemaRegionFolder(
+      String schemaRegionDirPath, Logger logger, LongConsumer deleteRateLimiter)
+      throws MetadataException {
     File schemaRegionDir = SystemFileFactory.INSTANCE.getFile(schemaRegionDirPath);
     File[] sgFiles = schemaRegionDir.listFiles();
     if (sgFiles == null) {
@@ -47,6 +55,9 @@ public static void deleteSchemaRegionFolder(String schemaRegionDirPath, Logger l
     }
     for (File file : sgFiles) {
       try {
+        if (deleteRateLimiter != null) {
+          deleteRateLimiter.accept(FileUtils.estimateFileOrDirectoryRemoveCost(file));
+        }
         Files.delete(file.toPath());
         logger.info(DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FILE, file.getAbsolutePath());
       } catch (IOException e) {
@@ -60,6 +71,9 @@ public static void deleteSchemaRegionFolder(String schemaRegionDirPath, Logger l
     }
 
     try {
+      if (deleteRateLimiter != null) {
+        deleteRateLimiter.accept(FileUtils.estimateFileOrDirectoryRemoveCost(schemaRegionDir));
+      }
       Files.delete(schemaRegionDir.toPath());
       logger.info(
           DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FOLDER, schemaRegionDir.getAbsolutePath());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
index 1f2408cca1237..c9841112df648 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -44,6 +44,7 @@
 import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -488,7 +489,8 @@ public synchronized void deleteSchemaRegion() throws MetadataException {
     clear();
 
     // delete all the schema region files
-    SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger);
+    SchemaRegionUtils.deleteSchemaRegionFolder(
+        schemaRegionDirPath, logger, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
     if (config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
       SystemInfo.getInstance()
           .decreaseDirectBufferMemoryCost(config.getSchemaRatisConsensusLogAppenderBufferSizeMax());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
index 8d87ef3f0618f..8fc6a95bbd221 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
@@ -32,6 +32,7 @@
 import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode;
 import org.apache.iotdb.commons.schema.template.Template;
 import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -491,7 +492,8 @@ public synchronized void deleteSchemaRegion() throws MetadataException {
     clear();
 
     // delete all the schema region files
-    SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger);
+    SchemaRegionUtils.deleteSchemaRegionFolder(
+        schemaRegionDirPath, logger, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
 
     if (config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS)) {
       SystemInfo.getInstance()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 972ed73bd41de..b7c6d6a07f0df 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -42,6 +42,7 @@
 import org.apache.iotdb.commons.service.IService;
 import org.apache.iotdb.commons.service.ServiceType;
 import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.commons.utils.StatusUtils;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -841,12 +842,8 @@ public void deleteDataRegion(DataRegionId regionId) {
                       dataDir + File.separator + IoTDBConstant.SNAPSHOT_FOLDER_NAME,
                       region.getDatabaseName() + FILE_NAME_SEPARATOR + regionId.getId());
               if (regionSnapshotDir.exists()) {
-                try {
-                  FileUtils.deleteDirectory(regionSnapshotDir);
-                } catch (IOException e) {
-                  LOGGER.error(
-                      StorageEngineMessages.FAILED_TO_DELETE_SNAPSHOT_DIR, regionSnapshotDir, e);
-                }
+                org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter(
+                    regionSnapshotDir, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
               }
             }
             break;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 9f4de75f7949e..54a3d9aa2f61c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -46,6 +46,7 @@
 import org.apache.iotdb.commons.service.metric.enums.Metric;
 import org.apache.iotdb.commons.service.metric.enums.Tag;
 import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.commons.utils.RetryUtils;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -2257,8 +2258,8 @@ public void deleteFolder(String systemDir) {
       File dataRegionSystemFolder =
           SystemFileFactory.INSTANCE.getFile(
               systemDir + File.separator + databaseName, dataRegionIdString);
-      org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent(
-          dataRegionSystemFolder);
+      org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(
+          dataRegionSystemFolder, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
     } finally {
       writeUnlock();
     }
@@ -2348,8 +2349,8 @@ private void deleteAllSGFolders(List<String> folder) {
         }
       } else {
         if (dataRegionDataFolder.exists()) {
-          org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent(
-              dataRegionDataFolder);
+          org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(
+              dataRegionDataFolder, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
         }
       }
     }
@@ -2392,7 +2393,8 @@ private void deleteAllObjectFiles(List<String> folders) {
         }
       } else {
         if (dataRegionObjectFolder.exists()) {
-          org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectory(dataRegionObjectFolder);
+          org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter(
+              dataRegionObjectFolder, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
         }
       }
     }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
index 13bc6ebe67517..23a1ae63e56bb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.wal.allocation;
 
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
 import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
 import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
@@ -88,7 +89,9 @@ public void deleteUniqueIdAndMayDeleteWALNode(String applicantUniqueId) {
         if (walNode != null) {
           walNode.close();
           if (walNode.getLogDirectory().exists()) {
-            FileUtils.deleteFileOrDirectory(walNode.getLogDirectory());
+            FileUtils.deleteFileOrDirectoryWithRateLimiter(
+                walNode.getLogDirectory(),
+                RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
           }
           WALManager.getInstance().subtractTotalDiskUsage(walNode.getDiskUsage());
           WALManager.getInstance().subtractTotalFileNum(walNode.getFileNum());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
index 219666773e729..f1451652c39a1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.wal.allocation;
 
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
 import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
 import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
@@ -98,7 +99,9 @@ public void deleteWALNode(String applicantUniqueId) {
         walNode.setDeleted(true);
         walNode.close();
         if (walNode.getLogDirectory().exists()) {
-          FileUtils.deleteFileOrDirectory(walNode.getLogDirectory());
+          FileUtils.deleteFileOrDirectoryWithRateLimiter(
+              walNode.getLogDirectory(),
+              RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
         }
         WALManager.getInstance().subtractTotalDiskUsage(walNode.getDiskUsage());
         WALManager.getInstance().subtractTotalFileNum(walNode.getFileNum());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
index 158692fd41bfc..f4daffbf303b9 100755
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.conf;
 
 import org.apache.iotdb.commons.conf.TrimProperties;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
 
 import com.tngtech.archunit.core.domain.JavaClasses;
 import com.tngtech.archunit.core.importer.ClassFileImporter;
@@ -35,6 +36,35 @@
 import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses;
 
 public class PropertiesTest {
+  @Test
+  public void testHotReloadRegionMigrationFileRemoveSpeedLimit() throws Exception {
+    IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance();
+    long originalLimit =
+        descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond();
+    try {
+      TrimProperties properties = new TrimProperties();
+      properties.setProperty("region_migration_file_remove_speed_limit_bytes_per_second", "65536");
+      descriptor.loadHotModifiedProps(properties);
+      Assert.assertEquals(
+          65536, descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
+      Assert.assertEquals(
+          65536.0, RegionMigrationFileRemoveRateLimiter.getInstance().getRate(), 0.0);
+
+      properties.setProperty("region_migration_file_remove_speed_limit_bytes_per_second", "0");
+      descriptor.loadHotModifiedProps(properties);
+      Assert.assertEquals(
+          0, descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
+      Assert.assertEquals(
+          Double.MAX_VALUE, RegionMigrationFileRemoveRateLimiter.getInstance().getRate(), 0.0);
+    } finally {
+      TrimProperties properties = new TrimProperties();
+      properties.setProperty(
+          "region_migration_file_remove_speed_limit_bytes_per_second",
+          Long.toString(originalLimit));
+      descriptor.loadHotModifiedProps(properties);
+    }
+  }
+
   @Test
   public void PropertiesWithSpace() {
     IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance();
diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index fdf7994325540..8d71f072db1bc 100644
--- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1730,12 +1730,20 @@ data_region_iot_max_pending_batches_num = 5
 # Datatype: double
 data_region_iot_max_memory_ratio_for_queue = 0.6
 
-# The maximum transit size in byte per second for region migration
+# The maximum size in byte per second for region migration transfer.
 # values less than or equal to 0 means no limit
 # effectiveMode: hot_reload
 # Datatype: long
 region_migration_speed_limit_bytes_per_second = 50331648
 
+# The maximum estimated remove cost in byte per second for region migration file cleanup.
+# Each regular file remove costs 64 KiB, and each directory remove costs 4 KiB.
+# The default value equals about 256 regular file removes per second.
+# values less than or equal to 0 means no limit
+# effectiveMode: hot_reload
+# Datatype: long
+region_migration_file_remove_speed_limit_bytes_per_second = 16777216
+
 # The minimum interval (in ms) between two per-file progress logs while transmitting a snapshot in
 # IoTConsensus. A snapshot may contain a huge number of files, so logging one line per file is
 # costly; this throttles it to at most once per interval. A value <= 0 logs every file.
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
index cc4bf13d9940f..dd799b0f9a7fa 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
@@ -48,11 +48,14 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Stack;
+import java.util.function.LongConsumer;
 
 public class FileUtils {
   private static final Logger LOGGER = LoggerFactory.getLogger(FileUtils.class);
 
   private static final int BUFFER_SIZE = 1024;
+  private static final long FILE_REMOVE_COST_IN_BYTES = 64 * 1024L;
+  private static final long DIRECTORY_REMOVE_COST_IN_BYTES = 4 * 1024L;
 
   private FileUtils() {}
 
@@ -111,15 +114,31 @@ public static void deleteFileOrDirectory(File file) {
   }
 
   public static void deleteFileOrDirectory(File file, boolean quietForNoSuchFile) {
+    deleteFileOrDirectory(file, quietForNoSuchFile, null);
+  }
+
+  public static void deleteFileOrDirectoryWithRateLimiter(
+      File file, LongConsumer deleteRateLimiter) {
+    deleteFileOrDirectory(file, false, deleteRateLimiter);
+  }
+
+  public static void deleteFileOrDirectoryWithRateLimiter(
+      File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) {
+    deleteFileOrDirectory(file, quietForNoSuchFile, deleteRateLimiter);
+  }
+
+  private static void deleteFileOrDirectory(
+      File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) {
     if (file.isDirectory()) {
       File[] files = file.listFiles();
       if (files != null) {
         for (File subfile : files) {
-          deleteFileOrDirectory(subfile, quietForNoSuchFile);
+          deleteFileOrDirectory(subfile, quietForNoSuchFile, deleteRateLimiter);
         }
       }
     }
     try {
+      acquireRemovePermit(file, deleteRateLimiter);
       Files.delete(file.toPath());
     } catch (NoSuchFileException e) {
       if (!quietForNoSuchFile) {
@@ -161,16 +180,39 @@ public static void deleteFileOrDirectoryWithRetry(File file) {
   }
 
   public static void deleteDirectoryAndEmptyParent(File folder) {
-    deleteFileOrDirectory(folder);
+    deleteDirectoryAndEmptyParent(folder, null);
+  }
+
+  public static void deleteDirectoryAndEmptyParentWithRateLimiter(
+      File folder, LongConsumer deleteRateLimiter) {
+    deleteDirectoryAndEmptyParent(folder, deleteRateLimiter);
+  }
+
+  private static void deleteDirectoryAndEmptyParent(File folder, LongConsumer deleteRateLimiter) {
+    deleteFileOrDirectory(folder, false, deleteRateLimiter);
     final File parentFolder = folder.getParentFile();
+    if (parentFolder == null) {
+      return;
+    }
     File[] files = parentFolder.listFiles();
     if (parentFolder.isDirectory() && (files == null || files.length == 0)) {
+      acquireRemovePermit(parentFolder, deleteRateLimiter);
       if (!parentFolder.delete()) {
         LOGGER.warn(UtilMessages.DELETE_FOLDER_FAILED, parentFolder.getAbsolutePath());
       }
     }
   }
 
+  public static long estimateFileOrDirectoryRemoveCost(File file) {
+    return file.isDirectory() ? DIRECTORY_REMOVE_COST_IN_BYTES : FILE_REMOVE_COST_IN_BYTES;
+  }
+
+  private static void acquireRemovePermit(File file, LongConsumer deleteRateLimiter) {
+    if (deleteRateLimiter != null && file.exists()) {
+      deleteRateLimiter.accept(estimateFileOrDirectoryRemoveCost(file));
+    }
+  }
+
   public static boolean copyDir(File sourceDir, File targetDir) throws IOException {
     if (!sourceDir.exists() || !sourceDir.isDirectory()) {
       LOGGER.error(UtilMessages.COPY_FOLDER_SOURCE_NOT_EXIST, sourceDir.getAbsolutePath());
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java
new file mode 100644
index 0000000000000..40f9ddf9c5448
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java
@@ -0,0 +1,60 @@
+/*
+ * 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.iotdb.commons.utils;
+
+import com.google.common.util.concurrent.RateLimiter;
+
+public class RegionMigrationFileRemoveRateLimiter {
+
+  private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
+
+  private RegionMigrationFileRemoveRateLimiter() {}
+
+  public void init(long regionMigrationFileRemoveSpeedLimitBytesPerSecond) {
+    rateLimiter.setRate(
+        regionMigrationFileRemoveSpeedLimitBytesPerSecond <= 0
+            ? Double.MAX_VALUE
+            : regionMigrationFileRemoveSpeedLimitBytesPerSecond);
+  }
+
+  public void acquire(long estimatedRemoveCostInBytes) {
+    while (estimatedRemoveCostInBytes > 0) {
+      if (estimatedRemoveCostInBytes > Integer.MAX_VALUE) {
+        rateLimiter.acquire(Integer.MAX_VALUE);
+        estimatedRemoveCostInBytes -= Integer.MAX_VALUE;
+      } else {
+        rateLimiter.acquire((int) estimatedRemoveCostInBytes);
+        return;
+      }
+    }
+  }
+
+  @TestOnly
+  public double getRate() {
+    return rateLimiter.getRate();
+  }
+
+  private static final RegionMigrationFileRemoveRateLimiter INSTANCE =
+      new RegionMigrationFileRemoveRateLimiter();
+
+  public static RegionMigrationFileRemoveRateLimiter getInstance() {
+    return INSTANCE;
+  }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java
new file mode 100644
index 0000000000000..0e3e48f75fc03
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java
@@ -0,0 +1,54 @@
+/*
+ * 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.iotdb.commons.utils;
+
+import com.google.common.util.concurrent.RateLimiter;
+
+public class RegionMigrationRateLimiter {
+
+  private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
+
+  private RegionMigrationRateLimiter() {}
+
+  public void init(long regionMigrationSpeedLimitBytesPerSecond) {
+    rateLimiter.setRate(
+        regionMigrationSpeedLimitBytesPerSecond <= 0
+            ? Double.MAX_VALUE
+            : regionMigrationSpeedLimitBytesPerSecond);
+  }
+
+  public void acquire(long sizeInBytes) {
+    while (sizeInBytes > 0) {
+      if (sizeInBytes > Integer.MAX_VALUE) {
+        rateLimiter.acquire(Integer.MAX_VALUE);
+        sizeInBytes -= Integer.MAX_VALUE;
+      } else {
+        rateLimiter.acquire((int) sizeInBytes);
+        return;
+      }
+    }
+  }
+
+  private static final RegionMigrationRateLimiter INSTANCE = new RegionMigrationRateLimiter();
+
+  public static RegionMigrationRateLimiter getInstance() {
+    return INSTANCE;
+  }
+}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
index 4c2210998b039..402634b084bde 100644
--- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
@@ -33,6 +33,8 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 public class FileUtilsTest {
   private File tmpDir;
@@ -72,6 +74,68 @@ public void testGetIllegalError4DirectoryRejectsEmptyPath() {
     Assert.assertNull(FileUtils.getIllegalError4Directory("valid_dir"));
   }
 
+  @Test
+  public void testDeleteFileOrDirectoryWithRateLimiter() throws IOException {
+    File deleteDir = new File(tmpDir, "deleteWithRateLimiter");
+    File subDir = new File(deleteDir, "subDir");
+    Assert.assertTrue(subDir.mkdirs());
+    Files.write(new File(deleteDir, "file1").toPath(), new byte[3]);
+    Files.write(new File(subDir, "file2").toPath(), new byte[7]);
+    Files.write(new File(subDir, "empty").toPath(), new byte[0]);
+
+    AtomicLong acquiredBytes = new AtomicLong();
+    AtomicInteger acquiredFiles = new AtomicInteger();
+    FileUtils.deleteFileOrDirectoryWithRateLimiter(
+        deleteDir,
+        removeCost -> {
+          acquiredFiles.incrementAndGet();
+          acquiredBytes.addAndGet(removeCost);
+        });
+
+    Assert.assertFalse(deleteDir.exists());
+    Assert.assertEquals(5, acquiredFiles.get());
+    Assert.assertEquals(
+        3 * FileUtils.estimateFileOrDirectoryRemoveCost(new File("file"))
+            + 2 * FileUtils.estimateFileOrDirectoryRemoveCost(tmpDir),
+        acquiredBytes.get());
+  }
+
+  @Test
+  public void testDeleteDirectoryAndEmptyParentWithRateLimiter() throws IOException {
+    File parentDir = new File(tmpDir, "parentDir");
+    File deleteDir = new File(parentDir, "deleteDir");
+    Assert.assertTrue(deleteDir.mkdirs());
+    Files.write(new File(deleteDir, "file").toPath(), new byte[5]);
+
+    AtomicLong acquiredBytes = new AtomicLong();
+    FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(deleteDir, acquiredBytes::addAndGet);
+
+    Assert.assertFalse(deleteDir.exists());
+    Assert.assertFalse(parentDir.exists());
+    Assert.assertTrue(tmpDir.exists());
+    Assert.assertEquals(
+        FileUtils.estimateFileOrDirectoryRemoveCost(new File("file"))
+            + 2 * FileUtils.estimateFileOrDirectoryRemoveCost(tmpDir),
+        acquiredBytes.get());
+  }
+
+  @Test
+  public void testDeleteDirectoryAndEmptyParentWithRateLimiterAndNoParent() throws IOException {
+    File deleteDir = new File("deleteDirWithoutParent-" + System.nanoTime());
+    try {
+      Assert.assertTrue(deleteDir.mkdirs());
+
+      AtomicLong acquiredBytes = new AtomicLong();
+      long directoryRemoveCost = FileUtils.estimateFileOrDirectoryRemoveCost(deleteDir);
+      FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(deleteDir, acquiredBytes::addAndGet);
+
+      Assert.assertFalse(deleteDir.exists());
+      Assert.assertEquals(directoryRemoveCost, acquiredBytes.get());
+    } finally {
+      FileUtils.deleteFileOrDirectory(deleteDir, true);
+    }
+  }
+
   private void generateFile(File tsfile) throws WriteProcessException, IOException {
     try (TsFileWriter writer = new TsFileWriter(tsfile)) {
       writer.registerAlignedTimeseries(