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 76fca09afec47..f790116011736 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 @@ -1587,6 +1587,9 @@ private void checkMemoryPressure() throws WriteProcessRejectException { 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 @@ -1603,6 +1606,19 @@ private void checkMemoryPressure() throws WriteProcessRejectException { } } + 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(WriteMemoryDelta delta) { workMemTable.getArrayMemoryBlock().release(delta.arrayBytes); workMemTable.getBinaryMemoryBlock().release(delta.binaryBytes); 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 a70315cca3491..c82a65d7389fa 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 @@ -84,6 +84,7 @@ import org.apache.iotdb.db.storageengine.load.LoadTsFileManager; import org.apache.iotdb.db.storageengine.rescon.memory.MemTableManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; +import org.apache.iotdb.db.storageengine.rescon.memory.WriteMemoryBlock; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; import org.apache.iotdb.mpp.rpc.thrift.TDropTableModProgress; @@ -95,7 +96,13 @@ 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.common.TimeRange; +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; @@ -228,6 +235,143 @@ public static InsertRowNode buildInsertRowNodeByTSRecord(TSRecord record) 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()); + for (WriteMemoryBlock block : regionInfo.getWriteMemoryManager().getMemoryBlocks()) { + Assert.assertEquals(0, block.getReservedMemoryInBytes()); + } + 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.syncCloseAllWorkingAndClosingTsFileProcessors(); + 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.loadWriteMemory(); + 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 {