diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java index d4d6ebdc29383..576b243a4925a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java @@ -57,6 +57,7 @@ import org.apache.tsfile.read.common.BatchData; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.reader.BufferedTsFileInput; import org.apache.tsfile.read.reader.IChunkReader; import org.apache.tsfile.read.reader.chunk.AlignedChunkReader; import org.apache.tsfile.read.reader.chunk.ChunkReader; @@ -83,6 +84,8 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser { + private static final int TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES = 8 * 1024; + private final long startTime; private final long endTime; private final Filter filter; @@ -91,6 +94,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser { private BatchData data; private final PipeMemoryBlock allocatedMemoryBlockForBatchData; private final PipeMemoryBlock allocatedMemoryBlockForChunk; + private PipeMemoryBlock allocatedMemoryBlockForTsFileInput; private boolean currentIsMultiPage; private IDeviceID currentDevice; @@ -156,11 +160,7 @@ public TsFileInsertionEventScanParser( PipeDataNodeResourceManager.memory() .forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed()); - tsFileSequenceReader = - new TsFileSequenceReader( - tsFile.getAbsolutePath(), - !currentModifications.isEmpty(), - !currentModifications.isEmpty()); + tsFileSequenceReader = createTsFileSequenceReader(tsFile, !currentModifications.isEmpty()); tsFileSequenceReader.position((long) TSFileConfig.MAGIC_STRING.getBytes().length + 1); prepareData(); @@ -193,6 +193,22 @@ public TsFileInsertionEventScanParser( isWithMod); } + private TsFileSequenceReader createTsFileSequenceReader( + final File tsFile, final boolean hasModifications) throws IOException { + if (hasModifications) { + return new TsFileSequenceReader(tsFile.getAbsolutePath(), true, true); + } + + allocatedMemoryBlockForTsFileInput = + PipeDataNodeResourceManager.memory() + .forceAllocateForTabletWithRetry(TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES); + return new TsFileSequenceReader( + new BufferedTsFileInput(tsFile.toPath(), TS_FILE_INPUT_BUFFER_SIZE_IN_BYTES), + false, + false, + null); + } + @Override public Iterable toTabletInsertionEvents() { if (tabletInsertionIterable == null) { @@ -1098,6 +1114,10 @@ public void close() { if (allocatedMemoryBlockForChunk != null) { allocatedMemoryBlockForChunk.close(); } + + if (allocatedMemoryBlockForTsFileInput != null) { + allocatedMemoryBlockForTsFileInput.close(); + } } private Statistics findAlignedChunkStatistics( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java index 0b6feb7834951..06e422ad776a9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java @@ -39,9 +39,22 @@ import java.util.List; import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; +import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER; +import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF; public class AlignedTVListTest { + @Test + public void testValueListArrayMemCostIncludesBitmapImplementation() { + long expected = + (long) ARRAY_SIZE * Long.BYTES + + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed() + + NUM_BYTES_ARRAY_HEADER + + 2L * NUM_BYTES_OBJECT_REF; + + Assert.assertEquals(expected, AlignedTVList.valueListArrayMemCost(TSDataType.INT64)); + } + @Test public void testAlignedTVList1() { List dataTypes = new ArrayList<>();