From 506a7705a911ccb178da52df98f8a3abcb2c0f09 Mon Sep 17 00:00:00 2001 From: Christofer Dutz Date: Tue, 3 Nov 2026 20:48:14 +0100 Subject: [PATCH] [Storage] Cap the temporary flush memory estimate at the memtable size A flush reserves memSize / seriesNumber * io_task_queue_size_for_flushing and takes it out of the MemTable budget. With fewer series than queue slots (default 10) the estimate exceeds the memtable itself: a memtable with 3 series reserved 3.3 times its size, more than the whole MemTable budget, so the reject threshold became negative and writes were rejected. The bounded IO queue can never hold more than the memtable, so the estimate is now capped at memSize when there are no more series than queue slots. With more series the estimate is unchanged. --- .../dataregion/flush/MemTableFlushTask.java | 25 +++++++-- .../flush/FlushMemoryEstimateTest.java | 51 +++++++++++++++++++ 2 files changed, 71 insertions(+), 5 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/flush/FlushMemoryEstimateTest.java 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)); + } +}