diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java index 7777e48be3a41..66b55cbaa0d41 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java +++ b/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( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushMemoryEstimateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushMemoryEstimateTest.java new file mode 100644 index 0000000000000..487e9c4822d5d --- /dev/null +++ b/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)); + } +}