(iotdb) branch master updated: [Pipe] Use buffered input for TsFile scan parser (#18254)

[jiangtian]

This is an automated email from the ASF dual-hosted git repository.

jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 1eb7c30faa2 [Pipe] Use buffered input for TsFile scan parser (#18254)
1eb7c30faa2 is described below

commit 1eb7c30faa2de8c46bb51fafc9d6aa6b71202d65
Author: Caideyipi <87789683+caidey...@users.noreply.github.com>
AuthorDate: Mon Jul 27 09:58:16 2026 +0800

    [Pipe] Use buffered input for TsFile scan parser (#18254)
    
    * [Pipe] Use buffered input for TsFile scan parser
    
    * [Pipe] Use new TsFile input reader API
    
    * Fix aligned TVList bitmap memory accounting
---
 .../scan/TsFileInsertionEventScanParser.java       | 30 ++++++++++++++++++----
 .../db/utils/datastructure/AlignedTVListTest.java  | 13 ++++++++++
 2 files changed, 38 insertions(+), 5 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 d4d6ebdc293..576b243a492 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.TsFileSequenceReader;
 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 @@ import java.util.Objects;
 
 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 class TsFileInsertionEventScanParser extends 
TsFileInsertionEventParser {
           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 class TsFileInsertionEventScanParser extends 
TsFileInsertionEventParser {
         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<TabletInsertionEvent> toTabletInsertionEvents() {
     if (tabletInsertionIterable == null) {
@@ -1098,6 +1114,10 @@ public class TsFileInsertionEventScanParser extends 
TsFileInsertionEventParser {
     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 0b6feb78349..06e422ad776 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.Arrays;
 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<TSDataType> dataTypes = new ArrayList<>();