This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch cp-592d6635-dev-1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 54411835cb213e2852d5dc6a4dcd3b7fd8e372ff Author: Caideyipi <[email protected]> AuthorDate: Thu Oct 22 16:52:00 2026 +0800 Fix same-region flush stalls during write admission (#18985) --- .../dataregion/memtable/TsFileProcessor.java | 17 +++ .../storageengine/dataregion/DataRegionTest.java | 142 +++++++++++++++++++++ 2 files changed, 159 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 7a72752af42..d2797df3e87 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -1204,6 +1204,10 @@ public class TsFileProcessor { if (workMemTable.shouldFlush()) { break; } + // The asynchronous submitter needs the region write lock held by this insertion. + // Submit other already-selected MemTables here so they can release memory while we + // wait. + submitOtherSelectedMemTablesToFlush(); try { TimeUnit.MILLISECONDS.sleep(config.getCheckPeriodWhenInsertBlocked()); if (System.currentTimeMillis() - startTime @@ -1227,6 +1231,19 @@ public class TsFileProcessor { workMemTable.addTextDataSize(textDataIncrement); } + private void submitOtherSelectedMemTablesToFlush() { + for (TsFileProcessor processor : dataRegionInfo.getAllReportedTsp()) { + // This processor still owns the pending write and its memory reservations. + if (processor == this) { + continue; + } + IMemTable memTable = processor.getWorkMemTable(); + if (memTable != null && memTable.shouldFlush()) { + processor.submitAFlushTask(); + } + } + } + private void rollbackMemoryInfo(long[] memIncrements) { long memTableIncrement = memIncrements[0]; long textDataIncrement = memIncrements[1]; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 0ed0abc11c8..7c2444a1cf0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -58,6 +58,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.constant import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionConfigRestorer; import org.apache.iotdb.db.storageengine.dataregion.flush.FlushManager; import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; @@ -76,6 +77,12 @@ import org.apache.tsfile.file.metadata.PlainDeviceID; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.TsFileReader; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.TSRecord; @@ -93,6 +100,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -181,6 +189,140 @@ public class DataRegionTest { return insertRowNode; } + @Test + public void testRejectedRowFlushesOnlySelectedOtherMemtable() throws Exception { + assertRejectedInsertionFlushesOnlySelectedOtherMemtable(false); + } + + @Test + public void testRejectedTabletFlushesOnlySelectedOtherMemtable() throws Exception { + assertRejectedInsertionFlushesOnlySelectedOtherMemtable(true); + } + + private void assertRejectedInsertionFlushesOnlySelectedOtherMemtable(boolean tablet) + throws Exception { + final SystemInfo systemInfo = SystemInfo.getInstance(); + final int originalWait = config.getMaxWaitingTimeWhenInsertBlocked(); + final int originalPeriod = config.getCheckPeriodWhenInsertBlocked(); + final boolean originalSeparateData = config.isEnableSeparateData(); + final long partitionInterval = TimePartitionUtils.getTimePartitionInterval(); + try { + config.setEnableSeparateData(true); + for (int i = 1; i <= 384; i++) { + insertMemoryAdmissionTestRow(i, i); + } + final TsFileProcessor selected = + dataRegion.getWorkSequenceTsFileProcessors().stream() + .max( + (left, right) -> + Long.compare(left.getWorkMemTableRamCost(), right.getWorkMemTableRamCost())) + .get(); + // Fill the first primitive arrays so that the rejected write owns a new reservation. + for (int i = 1; i <= 128; i++) { + insertMemoryAdmissionTestRow(partitionInterval + i, i); + } + final TsFileProcessor current = + dataRegion.getWorkSequenceTsFileProcessors().stream() + .filter(processor -> processor != selected) + .findFirst() + .get(); + insertMemoryAdmissionTestRow(2 * partitionInterval + 1, 1); + final TsFileProcessor unselected = + dataRegion.getWorkSequenceTsFileProcessors().stream() + .filter(processor -> processor != selected && processor != current) + .findFirst() + .get(); + final IMemTable currentMemTable = current.getWorkMemTable(); + final IMemTable unselectedMemTable = unselected.getWorkMemTable(); + final IMemTable selectedMemTable = selected.getWorkMemTable(); + + final Field regionInfoField = DataRegion.class.getDeclaredField("dataRegionInfo"); + regionInfoField.setAccessible(true); + final DataRegionInfo regionInfo = (DataRegionInfo) regionInfoField.get(dataRegion); + final long memoryBefore = regionInfo.getMemCost(); + setMemoryAdmissionTestThreshold("memorySizeForMemtable", 2 * memoryBefore); + setMemoryAdmissionTestThreshold( + "FLUSH_THRESHOLD", (double) memoryBefore - selected.getWorkMemTableRamCost() / 2); + setMemoryAdmissionTestThreshold("REJECT_THRESHOLD", (double) memoryBefore - 1); + config.setMaxWaitingTimeWhenInsertBlocked(5_000); + config.setCheckPeriodWhenInsertBlocked(10); + regionInfo.setNeedToReportToSystem(true); + + // Keep the region lock held until the write succeeds. The asynchronous flush submitter + // cannot acquire it, so the insertion must submit the selected processor itself. + dataRegion.writeLock("testMemoryAdmission"); + try { + if (tablet) { + dataRegion.insertTablet( + new InsertTabletNode( + new PlanNodeId("memoryAdmission"), + new PartialPath(deviceId), + false, + new String[] {measurementId}, + new TSDataType[] {TSDataType.INT32}, + new MeasurementSchema[] {new MeasurementSchema(measurementId, TSDataType.INT32)}, + new long[] {partitionInterval + 129}, + null, + new Object[] {new int[] {129}}, + 1)); + } else { + insertMemoryAdmissionTestRow(partitionInterval + 129, 129); + } + Assert.assertTrue(selectedMemTable.shouldFlush()); + Assert.assertNull(selected.getWorkMemTable()); + Assert.assertSame(currentMemTable, current.getWorkMemTable()); + Assert.assertSame(unselectedMemTable, unselected.getWorkMemTable()); + Assert.assertFalse(currentMemTable.shouldFlush()); + Assert.assertFalse(unselectedMemTable.shouldFlush()); + Assert.assertEquals(129, currentMemTable.getTotalPointsNum()); + Assert.assertFalse(systemInfo.isRejected()); + } finally { + dataRegion.writeUnlock(); + } + + // A later write still reuses the small current MemTable instead of creating another file. + insertMemoryAdmissionTestRow(partitionInterval + 130, 130); + Assert.assertSame(currentMemTable, current.getWorkMemTable()); + Assert.assertEquals(3, dataRegion.getTsFileManager().getTsFileList(true).size()); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + int points = 0; + for (TsFileResource resource : dataRegion.getTsFileManager().getTsFileList(true)) { + try (TsFileSequenceReader reader = new TsFileSequenceReader(resource.getTsFilePath()); + TsFileReader tsFileReader = new TsFileReader(reader)) { + final QueryDataSet dataSet = + tsFileReader.query( + QueryExpression.create( + Collections.singletonList(new Path(deviceId, measurementId, false)), null)); + while (dataSet.hasNext()) { + final RowRecord row = dataSet.next(); + Assert.assertEquals( + row.getTimestamp() % partitionInterval, row.getFields().get(0).getIntV()); + points++; + } + } + } + Assert.assertEquals(515, points); + } finally { + systemInfo.allocateWriteMemory(); + config.setMaxWaitingTimeWhenInsertBlocked(originalWait); + config.setCheckPeriodWhenInsertBlocked(originalPeriod); + config.setEnableSeparateData(originalSeparateData); + } + } + + private void insertMemoryAdmissionTestRow(long time, int value) throws Exception { + final TSRecord record = new TSRecord(deviceId, time); + record.addTuple( + DataPoint.getDataPoint(TSDataType.INT32, measurementId, Integer.toString(value))); + dataRegion.insert(buildInsertRowNodeByTSRecord(record)); + } + + private void setMemoryAdmissionTestThreshold(String name, Object value) throws Exception { + final Field field = SystemInfo.class.getDeclaredField(name); + field.setAccessible(true); + field.set(SystemInfo.getInstance(), value); + } + @Test public void testUnseqUnsealedDelete() throws WriteProcessException, IOException, MetadataException {
