[Storage] Cap the temporary flush memory estimate at the memtable size

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java

@@ -141,15 +141,30 @@ public MemTableFlushTask(
         writer.getFile().getName());
   }
 
+  /**
+   * Temporary memory of a flush: the bounded IO queue holds up to {@code ioTaskQueueSize} encoded
+   * series of average size. With no more series than queue slots the queue can't hold more than the
+   * memtable, so the estimate is capped at {@code memSize}. Without the cap a memtable of 3 series
+   * reserved 3.3 times its size, which can exceed the whole MemTable budget and make the reject
+   * threshold negative.
+   */
+  static long estimateTemporaryMemory(
+      long memSize, int seriesNumber, int ioTaskQueueSize, boolean encodingFasterThanIo) {
+    if (!encodingFasterThanIo || seriesNumber == 0) {
+      return 0;
+    }
+    return seriesNumber <= ioTaskQueueSize ? memSize : memSize / seriesNumber * ioTaskQueueSize;
+  }
+
   /** the function for flushing memtable. */
   @SuppressWarnings("squid:S3776")
   public void syncFlushMemTable() throws ExecutionException, InterruptedException {
     long estimatedBytes =
-        encodingFasterThanIo && memTable.getSeriesNumber() != 0
-            ? memTable.memSize()
-                / memTable.getSeriesNumber()
-                * config.getIoTaskQueueSizeForFlushing()
-            : 0;
+        estimateTemporaryMemory(
+            memTable.memSize(),
+            memTable.getSeriesNumber(),
+            config.getIoTaskQueueSizeForFlushing(),
+            encodingFasterThanIo);
     try (SystemInfo.FlushMemoryReservation ignored =
         SystemInfo.getInstance()
             .reserveTemporaryMemoryForFlushing(

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushMemoryEstimateTest.java

@@ -0,0 +1,51 @@
+/*
+ * 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.db.storageengine.dataregion.flush;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/** The temporary memory reserved for a flush (reduces the MemTable budget while it runs). */
+public class FlushMemoryEstimateTest {
+
+  private static final long MB = 1024 * 1024;
+
+  @Test
+  public void manySeriesReserveTheQueuedShare() {
+    // 300 series, a queue of 10 encoded chunks: 10 series' share of the memtable
+    assertEquals(
+        30 * MB / 300 * 10, MemTableFlushTask.estimateTemporaryMemory(30 * MB, 300, 10, true));
+  }
+
+  @Test
+  public void fewSeriesNeverReserveMoreThanTheMemTable() {
+    // 3 series cannot fill 10 queue slots: at most the memtable itself is queued
+    assertEquals(52 * MB, MemTableFlushTask.estimateTemporaryMemory(52 * MB, 3, 10, true));
+    assertEquals(52 * MB, MemTableFlushTask.estimateTemporaryMemory(52 * MB, 1, 10, true));
+    assertEquals(52 * MB, MemTableFlushTask.estimateTemporaryMemory(52 * MB, 10, 10, true));
+  }
+
+  @Test
+  public void noReservationWithoutABoundedQueueOrSeries() {
+    assertEquals(0, MemTableFlushTask.estimateTemporaryMemory(52 * MB, 3, 10, false));
+    assertEquals(0, MemTableFlushTask.estimateTemporaryMemory(52 * MB, 0, 10, true));
+  }
+}