From 24a476dad7680779f7e3e6833b59c867eb3c459b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:20:58 +0800 Subject: [PATCH 1/3] [Pipe] Use buffered input for TsFile scan parser --- .../scan/TsFileInsertionEventScanParser.java | 36 ++++++++++++++++--- pom.xml | 2 +- 2 files changed, 32 insertions(+), 6 deletions(-) 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..1bbf7137c1046 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; @@ -156,11 +157,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 +190,35 @@ public TsFileInsertionEventScanParser( isWithMod); } + private static TsFileSequenceReader createTsFileSequenceReader( + final File tsFile, final boolean hasModifications) throws IOException { + if (hasModifications) { + return new TsFileSequenceReader(tsFile.getAbsolutePath(), true, true); + } + + final BufferedTsFileInput input = new BufferedTsFileInput(tsFile.toPath()); + try { + final ByteBuffer versionBuffer = ByteBuffer.allocate(Byte.BYTES); + if (input.read(versionBuffer, TSFileConfig.MAGIC_STRING.getBytes().length) == Byte.BYTES) { + versionBuffer.flip(); + if (versionBuffer.get() == TSFileConfig.VERSION_NUMBER) { + return new TsFileSequenceReader(input, false); + } + } + } catch (final IOException | RuntimeException | Error e) { + try { + input.close(); + } catch (final IOException closeException) { + e.addSuppressed(closeException); + } + throw e; + } + + // The TsFileInput constructor does not initialize the compatibility deserializer. + input.close(); + return new TsFileSequenceReader(tsFile.getAbsolutePath(), false, false); + } + @Override public Iterable toTabletInsertionEvents() { if (tabletInsertionIterable == null) { diff --git a/pom.xml b/pom.xml index 4b0f44bbad353..ac91cd70d47ee 100644 --- a/pom.xml +++ b/pom.xml @@ -145,7 +145,7 @@ 0.23.0 1.9 1.5.6-3 - 2.3.2-260616-SNAPSHOT + 2.3.2-260720-SNAPSHOT en From a1f2b5f4d3b61f9f25a39b183cb3522db96f5c28 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 21 Jul 2026 19:19:29 +0800 Subject: [PATCH 2/3] [Pipe] Use new TsFile input reader API --- .../scan/TsFileInsertionEventScanParser.java | 22 +------------------ pom.xml | 2 +- 2 files changed, 2 insertions(+), 22 deletions(-) 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 1bbf7137c1046..46e4c6893fe09 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 @@ -196,27 +196,7 @@ private static TsFileSequenceReader createTsFileSequenceReader( return new TsFileSequenceReader(tsFile.getAbsolutePath(), true, true); } - final BufferedTsFileInput input = new BufferedTsFileInput(tsFile.toPath()); - try { - final ByteBuffer versionBuffer = ByteBuffer.allocate(Byte.BYTES); - if (input.read(versionBuffer, TSFileConfig.MAGIC_STRING.getBytes().length) == Byte.BYTES) { - versionBuffer.flip(); - if (versionBuffer.get() == TSFileConfig.VERSION_NUMBER) { - return new TsFileSequenceReader(input, false); - } - } - } catch (final IOException | RuntimeException | Error e) { - try { - input.close(); - } catch (final IOException closeException) { - e.addSuppressed(closeException); - } - throw e; - } - - // The TsFileInput constructor does not initialize the compatibility deserializer. - input.close(); - return new TsFileSequenceReader(tsFile.getAbsolutePath(), false, false); + return new TsFileSequenceReader(new BufferedTsFileInput(tsFile.toPath()), false, false, null); } @Override diff --git a/pom.xml b/pom.xml index ac91cd70d47ee..79df3c2ebb172 100644 --- a/pom.xml +++ b/pom.xml @@ -145,7 +145,7 @@ 0.23.0 1.9 1.5.6-3 - 2.3.2-260720-SNAPSHOT + 2.3.2-260721-SNAPSHOT en From 0672b083016b2c6ab803b36a5120a71021ad6d29 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:42:43 +0800 Subject: [PATCH 3/3] Fix aligned TVList bitmap memory accounting --- .../db/utils/datastructure/AlignedTVList.java | 7 ++--- .../memtable/TsFileProcessorTest.java | 28 +++++++++---------- .../datastructure/AlignedTVListTest.java | 13 +++++++++ 3 files changed, 29 insertions(+), 19 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 0e44a3dbde5d3..97fab7a290bc6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -46,7 +46,6 @@ import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; -import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.utils.TsPrimitiveType; @@ -74,9 +73,7 @@ public abstract class AlignedTVList extends TVList { private static final long BITMAP_RAM_COST_PER_BLOCK = - RamUsageEstimator.shallowSizeOfInstance(BitMap.class) - + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE)) - + NUM_BYTES_OBJECT_REF; + new BitMap(ARRAY_SIZE).ramBytesUsed() + NUM_BYTES_OBJECT_REF; // Data types of this aligned tvList protected List dataTypes; @@ -1188,7 +1185,7 @@ public static long valueListArrayMemCost(TSDataType type) { long size = 0; // value array mem size size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize(); - // bitmap object, byte array, and reference in the bitmap list + // bitmap and its implementation, plus the reference in the bitmap list size += BITMAP_RAM_COST_PER_BLOCK; // array headers mem size size += NUM_BYTES_ARRAY_HEADER; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 040988997b2a3..f1d4a1572e6f3 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -536,21 +536,21 @@ public void alignedTvListRamCostTest() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1824552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(100, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1824552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(200, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1824552, memTable.getTVListsRamCost()); Assert.assertEquals(90000, memTable.getTotalPointsNum()); Assert.assertEquals(720360, memTable.memSize()); // Test records @@ -559,7 +559,7 @@ public void alignedTvListRamCostTest() record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); } - Assert.assertEquals(1778168, memTable.getTVListsRamCost()); + Assert.assertEquals(1826168, memTable.getTVListsRamCost()); Assert.assertEquals(90100, memTable.getTotalPointsNum()); Assert.assertEquals(721560, memTable.memSize()); } @@ -614,56 +614,56 @@ public void alignedTvListRamCostTest2() true, new long[5]); IMemTable memTable = processor.getWorkMemTable(); - Assert.assertEquals(1776552, memTable.getTVListsRamCost()); + Assert.assertEquals(1824552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(0, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(3552552, memTable.getTVListsRamCost()); + Assert.assertEquals(3648552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(100, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(3552552, memTable.getTVListsRamCost()); + Assert.assertEquals(3648552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(100, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(3552552, memTable.getTVListsRamCost()); + Assert.assertEquals(3648552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(200, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(3552552, memTable.getTVListsRamCost()); + Assert.assertEquals(3648552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(200, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(3552552, memTable.getTVListsRamCost()); + Assert.assertEquals(3648552, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNode(300, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(7105104, memTable.getTVListsRamCost()); + Assert.assertEquals(7297104, memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(300, true), Collections.singletonList(new int[] {0, 10}), new TSStatus[10], true, new long[5]); - Assert.assertEquals(7105104, memTable.getTVListsRamCost()); + Assert.assertEquals(7297104, memTable.getTVListsRamCost()); Assert.assertEquals(240000, memTable.getTotalPointsNum()); Assert.assertEquals(1920960, memTable.memSize()); @@ -673,14 +673,14 @@ public void alignedTvListRamCostTest2() record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); } - Assert.assertEquals(7106720, memTable.getTVListsRamCost()); + Assert.assertEquals(7298720, memTable.getTVListsRamCost()); // Test records for (int i = 1; i <= 100; i++) { TSRecord record = new TSRecord(deviceId, i); record.addTuple(DataPoint.getDataPoint(dataType, "s1", String.valueOf(i))); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); } - Assert.assertEquals(7108336, memTable.getTVListsRamCost()); + Assert.assertEquals(7300336, memTable.getTVListsRamCost()); Assert.assertEquals(240200, memTable.getTotalPointsNum()); Assert.assertEquals(1923360, memTable.memSize()); } 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 5ab976c13104b..b9f3428cd08ed 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 @@ -34,9 +34,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 + + new BitMap(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<>();