feat(load): route decoded tsfile pieces through consensus

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequest.java

@@ -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,11 +87,19 @@ 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;
   }
 
-  public List<ByteBuffer> getSerializedRequests() {
+  public synchronized List<ByteBuffer> getSerializedRequests() {
+    // Requests that deferred their serialization materialize their bytes here, that is, when a
+    // batch is actually built for sending.
+    buildSerializedRequests();
     return serializedRequests;
   }
 
@@ -112,6 +121,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

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/log/ConsensusReqReader.java

@@ -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.
    *

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManager.java

@@ -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(

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java

@@ -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) {

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java

@@ -115,8 +115,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(

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/common/request/IndexedConsensusRequestTest.java

@@ -0,0 +1,108 @@
+/*
+ * 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
+    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
+    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;
+    }
+  }
+}

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistryTest.java

@@ -39,52 +39,95 @@ public class SubscriptionQueueRegistryTest {
   @Test
   public void testOfferWithoutQueuesDoesNotSerializeRequest() {
     final SubscriptionQueueRegistry registry = new SubscriptionQueueRegistry("test");
-    final IndexedConsensusRequest request = newRequest();
+    final ByteBufferConsensusRequest inner = new ByteBufferConsensusRequest(ByteBuffer.allocate(1));
+    final IndexedConsensusRequest request = newRequest(inner);
 
     assertFalse(registry.offer(request));
-    assertTrue(request.getSerializedRequests().isEmpty());
+    assertEquals(0, inner.getSerializationCount());
   }
 
   @Test
   public void testOfferSerializesRequestBeforeQueueAdmission() {
     final SubscriptionQueueRegistry registry = new SubscriptionQueueRegistry("test");
     final InspectingQueue queue = new InspectingQueue();
+    register(registry, queue);
+    final ByteBufferConsensusRequest inner = new ByteBufferConsensusRequest(ByteBuffer.allocate(1));
+    final IndexedConsensusRequest request = newRequest(inner);
+
+    assertTrue(registry.offer(request));
+    assertEquals(1, inner.getSerializationCount());
+    assertEquals(1, request.getSerializedRequests().size());
+    assertEquals(1, queue.getSerializedRequestCountAtOffer());
+    assertSame(request, queue.poll());
+  }
+
+  @Test
+  public void testOfferDefersSerializationOfDeferredRequest() {
+    final SubscriptionQueueRegistry registry = new SubscriptionQueueRegistry("test");
+    final ArrayBlockingQueue<IndexedConsensusRequest> queue = new ArrayBlockingQueue<>(1);
+    register(registry, queue);
+    final ByteBufferConsensusRequest inner =
+        new ByteBufferConsensusRequest(ByteBuffer.allocate(1), true);
+    final IndexedConsensusRequest request = newRequest(inner);
+
+    assertTrue(registry.offer(request));
+    assertEquals(
+        "a request that deferred its serialization must stay unexpanded while it waits in the queue",
+        0,
+        inner.getSerializationCount());
+    assertEquals(1, request.getSerializedRequests().size());
+    assertEquals(1, inner.getSerializationCount());
+  }
+
+  private static void register(
+      final SubscriptionQueueRegistry registry,
+      final ArrayBlockingQueue<IndexedConsensusRequest> queue) {
     registry.register(
         queue,
         new SubscriptionWalRetentionPolicy(
             "test",
             SubscriptionWalRetentionPolicy.UNBOUNDED,
             SubscriptionWalRetentionPolicy.UNBOUNDED));
-    final IndexedConsensusRequest request = newRequest();
-
-    assertTrue(registry.offer(request));
-    assertEquals(1, request.getSerializedRequests().size());
-    assertEquals(1, queue.getSerializedRequestCountAtOffer());
-    assertSame(request, queue.poll());
   }
 
-  private static IndexedConsensusRequest newRequest() {
-    return new IndexedConsensusRequest(
-        1, Collections.singletonList(new ByteBufferConsensusRequest(ByteBuffer.allocate(1))));
+  private static IndexedConsensusRequest newRequest(final IConsensusRequest inner) {
+    return new IndexedConsensusRequest(1, Collections.singletonList(inner));
   }
 
   private static final class ByteBufferConsensusRequest implements IConsensusRequest {
 
     private final ByteBuffer buffer;
+    private final boolean deferred;
+    private int serializationCount;
 
     private ByteBufferConsensusRequest(final ByteBuffer buffer) {
+      this(buffer, false);
+    }
+
+    private ByteBufferConsensusRequest(final ByteBuffer buffer, final boolean deferred) {
       this.buffer = buffer;
+      this.deferred = deferred;
     }
 
     @Override
     public ByteBuffer serializeToByteBuffer() {
+      serializationCount++;
       return buffer;
     }
 
     @Override
     public long getMemorySize() {
       return buffer.capacity();
     }
+
+    @Override
+    public boolean isSerializationDeferred() {
+      return deferred;
+    }
+
+    private int getSerializationCount() {
+      return serializationCount;
+    }
   }
 
   private static final class InspectingQueue extends ArrayBlockingQueue<IndexedConsensusRequest> {

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java

@@ -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";
@@ -1376,8 +1380,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 =