@@ -49,6 +49,7 @@ public class IndexedConsensusRequest implements IConsensusRequest {
   private long memorySize = 0;
   private long retainedMemorySize = 0;
   private boolean serializedRequestsBuilt = false;
+  private long queueReservedMemorySize = -1L;
   private boolean containsUserData = false;
   private final AtomicLong referenceCnt = new AtomicLong();
 
@@ -86,6 +87,11 @@ public ByteBuffer serializeToByteBuffer() {
     throw new UnsupportedOperationException();
   }
 
+  /** Returns whether any request of this entry materializes its bytes only when it is sent. */
+  public synchronized boolean hasDeferredRequests() {
+    return requests.stream().anyMatch(IConsensusRequest::isSerializationDeferred);
+  }
+
   public List<IConsensusRequest> getRequests() {
     return requests;
   }
@@ -112,6 +118,20 @@ public synchronized long getRetainedMemorySize() {
     return requests.stream().mapToLong(IConsensusRequest::getMemorySize).sum();
   }
 
+  /**
+   * Returns the amount to reserve for this entry while it waits in the replication queues.
+   *
+   * <p>The amount is captured on the first call, because a request that deferred its serialization
+   * materializes its bytes only when it is sent and its serialized buffers are accounted for by the
+   * batch that carries them. Releasing an entry therefore has to return exactly what was reserved.
+   */
+  public synchronized long getQueueReservedMemorySize() {
+    if (queueReservedMemorySize < 0) {
+      queueReservedMemorySize = getRetainedMemorySize();
+    }
+    return queueReservedMemorySize;
+  }
+
   /**
    * Releases the original request objects after their serialized buffers have been materialized.
    * Replication only needs the serialized buffers, while subscription delivery still needs the
 
@@ -38,6 +38,14 @@ public interface ConsensusReqReader {
 
   void setSafelyDeletedSearchIndex(long safelyDeletedSearchIndex);
 
+  /**
+   * @return the search index before which insert nodes can be deleted safely, or {@link
+   *     #DEFAULT_SAFELY_DELETED_SEARCH_INDEX} if the consensus layer never reported a watermark.
+   */
+  default long getSafelyDeletedSearchIndex() {
+    return DEFAULT_SAFELY_DELETED_SEARCH_INDEX;
+  }
+
   /**
    * Gets the consensus requests iterator from the specified start position.
    *
 
@@ -47,7 +47,9 @@ public boolean reserve(IndexedConsensusRequest request) {
     synchronized (request) {
       long prevRef = request.incRef();
       if (prevRef == 0) {
-        final long retainedMemorySize = request.getRetainedMemorySize();
+        // The amount is captured once, so that an entry that materializes its bytes after it was
+        // queued returns exactly what it reserved when it is released.
+        final long retainedMemorySize = request.getQueueReservedMemorySize();
         boolean reserved = reserve(retainedMemorySize, true);
         if (reserved) {
           if (logger.isDebugEnabled()) {
@@ -114,7 +116,7 @@ public void free(IndexedConsensusRequest request) {
     synchronized (request) {
       long prevRef = request.decRef();
       if (prevRef == 1) {
-        final long retainedMemorySize = request.getRetainedMemorySize();
+        final long retainedMemorySize = request.getQueueReservedMemorySize();
         free(retainedMemorySize, true);
         if (logger.isDebugEnabled()) {
           logger.debug(
 
@@ -190,8 +190,14 @@ public void offer(IndexedConsensusRequest request) {
   public void offer(IndexedConsensusRequest request, boolean keepRequestsForSubscription) {
     // we don't need to serialize and offer request when replicaNum is 1.
     if (!threads.isEmpty()) {
-      request.buildSerializedRequests();
-      if (!keepRequestsForSubscription) {
+      // A request that deferred its serialization keeps only its source objects, which stay small,
+      // so that the entries waiting in the queues to be replicated do not retain their payloads.
+      // Such requests materialize their bytes when a batch is built instead.
+      final boolean serializationDeferred = request.hasDeferredRequests();
+      if (!serializationDeferred) {
+        request.buildSerializedRequests();
+      }
+      if (!keepRequestsForSubscription && !serializationDeferred) {
         request.clearRequests();
       }
       synchronized (this) {
@@ -654,6 +660,9 @@ private boolean constructBatchFromWAL(long currentIndex, long maxIndex, Batch lo
 
     private void constructBatchIndexedFromConsensusRequest(
         IndexedConsensusRequest request, Batch logBatches) {
+      // A request that deferred its serialization keeps only its source objects while it is queued,
+      // so its bytes are materialized here, when the batch that carries them is built.
+      request.buildSerializedRequests();
       TLogEntry logEntry =
           new TLogEntry(
               request.getSerializedRequests(),
 
@@ -194,8 +194,12 @@ public synchronized boolean offer(final IndexedConsensusRequest indexedConsensus
       return false;
     }
 
-    // Subscription queues reserve request memory, including the serialized buffers.
-    indexedConsensusRequest.buildSerializedRequests();
+    // Subscription queues reserve request memory, including the serialized buffers. A request that
+    // deferred its serialization keeps only its source objects until it is sent, and subscription
+    // delivery reads those objects, so nothing is materialized here.
+    if (!indexedConsensusRequest.hasDeferredRequests()) {
+      indexedConsensusRequest.buildSerializedRequests();
+    }
 
     if (LOGGER.isDebugEnabled()) {
       LOGGER.debug(
 
@@ -0,0 +1,110 @@
+/*
+ * 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.common.request;
+
+import org.apache.iotdb.commons.request.IConsensusRequest;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+public class IndexedConsensusRequestTest {
+
+  @Test
+  public void testDeferredRequestIsMaterializedOnlyWhenItIsNeeded() {
+    final CountingRequest deferred = new CountingRequest(true);
+    final IndexedConsensusRequest request =
+        new IndexedConsensusRequest(1L, 1L, Collections.singletonList(deferred));
+
+    Assert.assertTrue(request.hasDeferredRequests());
+    Assert.assertEquals(
+        "a deferred request must keep its source object, not its bytes, while it is queued",
+        0,
+        deferred.serializationCount);
+    Assert.assertEquals(64L, request.getRetainedMemorySize());
+
+    // The bytes are materialized when a batch is built for sending
+    request.buildSerializedRequests();
+    Assert.assertFalse(request.getSerializedRequests().isEmpty());
+    Assert.assertEquals(1, deferred.serializationCount);
+    Assert.assertEquals(64L, request.getMemorySize());
+    Assert.assertEquals(96L, request.getRetainedMemorySize());
+
+    // and are reused afterwards instead of being rebuilt for every retry
+    Assert.assertFalse(request.getSerializedRequests().isEmpty());
+    Assert.assertEquals(1, deferred.serializationCount);
+  }
+
+  @Test
+  public void testQueueReservationSurvivesDeferredSerialization() {
+    final CountingRequest deferred = new CountingRequest(true);
+    final IndexedConsensusRequest request =
+        new IndexedConsensusRequest(1L, 1L, Collections.singletonList(deferred));
+
+    // The queue reserves the reference form
+    Assert.assertEquals(64L, request.getQueueReservedMemorySize());
+
+    // Materializing the payloads must not change what a release returns
+    request.buildSerializedRequests();
+    Assert.assertFalse(request.getSerializedRequests().isEmpty());
+    Assert.assertEquals(96L, request.getRetainedMemorySize());
+    Assert.assertEquals(64L, request.getQueueReservedMemorySize());
+  }
+
+  @Test
+  public void testRequestThatSerializesEagerlyIsNotTreatedAsDeferred() {
+    final CountingRequest eager = new CountingRequest(false);
+    final IndexedConsensusRequest request =
+        new IndexedConsensusRequest(1L, 1L, Collections.singletonList(eager));
+
+    Assert.assertFalse(request.hasDeferredRequests());
+    request.buildSerializedRequests();
+    Assert.assertEquals(1, eager.serializationCount);
+    Assert.assertEquals(64L, request.getMemorySize());
+  }
+
+  private static class CountingRequest implements IConsensusRequest {
+
+    private final boolean deferred;
+    private int serializationCount;
+
+    private CountingRequest(final boolean deferred) {
+      this.deferred = deferred;
+    }
+
+    @Override
+    public ByteBuffer serializeToByteBuffer() {
+      serializationCount++;
+      return ByteBuffer.allocate(32);
+    }
+
+    @Override
+    public long getMemorySize() {
+      return 64L;
+    }
+
+    @Override
+    public boolean isSerializationDeferred() {
+      return deferred;
+    }
+  }
+}
@@ -1038,6 +1038,10 @@ private DataNodeMiscMessages() {}
   // --- DataNodeInternalRPCServiceImpl ---
   public static final String LOAD_COMMAND_REQUIRES_TIME_PARTITION_TO_PROGRESS_INDEX_MAP =
       "Load command requires time partition to progress index map";
+  public static final String LOG_THE_LOAD_COMMAND_PROTOCOL_WAS_REMOVED_THE_REQUEST_OF_TASK_ARG_IS_REJECTED_76937A9C =
+      "The LOAD command protocol was removed; the request of task {} is rejected.";
+  public static final String MESSAGE_THE_LOAD_COMMAND_PROTOCOL_WAS_REMOVED_LOAD_TRAVELS_THROUGH_CONSENSUS_NOW_0165EE16 =
+      "The LOAD command protocol was removed, LOAD travels through consensus now.";
   public static final String TOPOLOGY_PROBING_TIMED_OUT_AFTER_S_MS =
       "Topology probing timed out after %sms";
   public static final String NO_SUCH_QUERY = "No such query";
@@ -1375,8 +1379,7 @@ private DataNodeMiscMessages() {}
           + "should be an integer, which is %s";
   public static final String MISC_EXCEPTION_ILLEGAL_DEFAULTDATABASELEVEL_D_SHOULD_1_03088B38 =
       "Illegal defaultDatabaseLevel: %d, should >= 1";
-  public static final String MISC_EXCEPTION_LOADTSFILESPILTPARTITIONMAXSIZE_SHOULD_BE_GREATER_THAN_OR_95B4DB23 =
-      "loadTsFileSpiltPartitionMaxSize should be greater than or equal to 0";
+  public static final String MISC_EXCEPTION_LOADTSFILESPILTPARTITIONMAXSIZE_SHOULD_BE_GREATER_THAN_0_17B75192 = "loadTsFileSpiltPartitionMaxSize should be greater than 0";
   public static final String MISC_EXCEPTION_STATEMENTID_SDOESN_T_EXIST_IN_THIS_SESSION_S_BD5B4733 =
       "StatementId: %sdoesn't exist in this session %s";
   public static final String MISC_EXCEPTION_INTERNALCLIENTSESSION_SHOULD_NEVER_CALL_PREPARE_STATEMENT_CCAB3CDC =