Skip to content

[Pipe] Use buffered input for TsFile scan parser #18254

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[Pipe] Use buffered input for TsFile scan parser #18254

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
24a476d
[Pipe] Use buffered input for TsFile scan parser
Caideyipi Jul 20, 2026
a1f2b5f
[Pipe] Use new TsFile input reader API
Caideyipi Jul 21, 2026
0672b08
Fix aligned TVList bitmap memory accounting
Caideyipi Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions .../apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -193,6 +190,15 @@ public TsFileInsertionEventScanParser(
isWithMod);
}

private static TsFileSequenceReader createTsFileSequenceReader(
final File tsFile, final boolean hasModifications) throws IOException {
if (hasModifications) {
return new TsFileSequenceReader(tsFile.getAbsolutePath(), true, true);
}

return new TsFileSequenceReader(new BufferedTsFileInput(tsFile.toPath()), false, false, null);

@jt2594838 jt2594838 Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The BufferedTsFilePInput uses a buffer of 8KB by default.
Better to account it in the pipe memory management.

}

@Override
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
if (tabletInsertionIterable == null) {
Expand Down
7 changes: 2 additions & 5 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<TSDataType> dataTypes;
Expand Down Expand Up @@ -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;
Expand Down
28 changes: 14 additions & 14 deletions .../test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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());
}
Expand Down Expand Up @@ -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());
Expand All @@ -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());
}
Expand Down
13 changes: 13 additions & 0 deletions ...ore/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TSDataType> dataTypes = new ArrayList<>();
Expand Down
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@
<thrift.version>0.23.0</thrift.version>
<xz.version>1.9</xz.version>
<zstd-jni.version>1.5.6-3</zstd-jni.version>
<tsfile.version>2.3.2-260616-SNAPSHOT</tsfile.version>
<tsfile.version>2.3.2-260721-SNAPSHOT</tsfile.version>
<i18n.locale>en</i18n.locale>
<tsfile.locale.opt/>
</properties>
Expand Down
Loading