This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new bee05d1afb2 [To dev/1.3] Pipe: Optimize TsFile parsing logic (#16173)
(#16207)
bee05d1afb2 is described below
commit bee05d1afb29e0ad9563b2392b37dfe2a028e6f7
Author: Zhenyu Luo <[email protected]>
AuthorDate: Wed Aug 20 09:44:10 2025 +0800
[To dev/1.3] Pipe: Optimize TsFile parsing logic (#16173) (#16207)
* Pipe: Optimize TsFile parsing logic
* update
(cherry picked from commit 0dc583e9c59433845d715351b6fa746a92abb5be)
---
.../container/TsFileInsertionDataContainer.java | 10 +-
.../query/TsFileInsertionQueryDataContainer.java | 145 +++++++++++----------
.../TsFileInsertionQueryDataTabletIterator.java | 14 +-
.../scan/TsFileInsertionScanDataContainer.java | 103 ++++++++-------
4 files changed, 150 insertions(+), 122 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
index 16ca1aff79d..77ef2c871cb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.event.common.tsfile.container;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
@@ -55,6 +56,8 @@ public abstract class TsFileInsertionDataContainer implements
AutoCloseable {
protected TsFileSequenceReader tsFileSequenceReader;
+ protected Iterable<TabletInsertionEvent> tabletInsertionIterable;
+
protected TsFileInsertionDataContainer(
final String pipeName,
final long creationTime,
@@ -77,7 +80,9 @@ public abstract class TsFileInsertionDataContainer implements
AutoCloseable {
// Allocate empty memory block, will be resized later.
this.allocatedMemoryBlockForTablet =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTabletWithRetry(
+
PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes());
}
/**
@@ -87,6 +92,9 @@ public abstract class TsFileInsertionDataContainer implements
AutoCloseable {
@Override
public void close() {
+
+ tabletInsertionIterable = null;
+
try {
if (pipeName != null && !timeUsageReported) {
PipeTsFileToTabletsMetrics.getInstance()
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
index d020228863a..897d820df90 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
@@ -276,76 +276,83 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
@Override
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
- return () ->
- new Iterator<TabletInsertionEvent>() {
-
- private TsFileInsertionQueryDataTabletIterator tabletIterator = null;
-
- @Override
- public boolean hasNext() {
- while (tabletIterator == null || !tabletIterator.hasNext()) {
- if (!deviceMeasurementsMapIterator.hasNext()) {
- close();
- return false;
- }
-
- final Map.Entry<IDeviceID, List<String>> entry =
deviceMeasurementsMapIterator.next();
-
- try {
- tabletIterator =
- new TsFileInsertionQueryDataTabletIterator(
- tsFileReader,
- measurementDataTypeMap,
- ((PlainDeviceID) entry.getKey()).toStringID(),
- entry.getValue(),
- timeFilterExpression,
- allocatedMemoryBlockForTablet);
- } catch (final Exception e) {
- close();
- throw new PipeException("failed to create
TsFileInsertionDataTabletIterator", e);
- }
- }
-
- return true;
- }
+ if (tabletInsertionIterable == null) {
+ tabletInsertionIterable =
+ () ->
+ new Iterator<TabletInsertionEvent>() {
+
+ private TsFileInsertionQueryDataTabletIterator tabletIterator
= null;
+
+ @Override
+ public boolean hasNext() {
+ while (tabletIterator == null || !tabletIterator.hasNext()) {
+ if (!deviceMeasurementsMapIterator.hasNext()) {
+ close();
+ return false;
+ }
+
+ final Map.Entry<IDeviceID, List<String>> entry =
+ deviceMeasurementsMapIterator.next();
+
+ try {
+ tabletIterator =
+ new TsFileInsertionQueryDataTabletIterator(
+ tsFileReader,
+ measurementDataTypeMap,
+ ((PlainDeviceID) entry.getKey()).toStringID(),
+ entry.getValue(),
+ timeFilterExpression,
+ allocatedMemoryBlockForTablet);
+ } catch (final Exception e) {
+ close();
+ throw new PipeException(
+ "failed to create
TsFileInsertionDataTabletIterator", e);
+ }
+ }
+
+ return true;
+ }
+
+ @Override
+ public TabletInsertionEvent next() {
+ if (!hasNext()) {
+ close();
+ throw new NoSuchElementException();
+ }
+
+ final Tablet tablet = tabletIterator.next();
+ final boolean isAligned =
+ deviceIsAlignedMap.getOrDefault(new
PlainDeviceID(tablet.deviceId), false);
+
+ final TabletInsertionEvent next;
+ if (!hasNext()) {
+ next =
+ new PipeRawTabletInsertionEvent(
+ tablet,
+ isAligned,
+ sourceEvent != null ? sourceEvent.getPipeName() :
null,
+ sourceEvent != null ?
sourceEvent.getCreationTime() : 0,
+ pipeTaskMeta,
+ sourceEvent,
+ true);
+ close();
+ } else {
+ next =
+ new PipeRawTabletInsertionEvent(
+ tablet,
+ isAligned,
+ sourceEvent != null ? sourceEvent.getPipeName() :
null,
+ sourceEvent != null ?
sourceEvent.getCreationTime() : 0,
+ pipeTaskMeta,
+ sourceEvent,
+ false);
+ }
+ return next;
+ }
+ };
+ }
- @Override
- public TabletInsertionEvent next() {
- if (!hasNext()) {
- close();
- throw new NoSuchElementException();
- }
-
- final Tablet tablet = tabletIterator.next();
- final boolean isAligned =
- deviceIsAlignedMap.getOrDefault(new
PlainDeviceID(tablet.deviceId), false);
-
- final TabletInsertionEvent next;
- if (!hasNext()) {
- next =
- new PipeRawTabletInsertionEvent(
- tablet,
- isAligned,
- sourceEvent != null ? sourceEvent.getPipeName() : null,
- sourceEvent != null ? sourceEvent.getCreationTime() : 0,
- pipeTaskMeta,
- sourceEvent,
- true);
- close();
- } else {
- next =
- new PipeRawTabletInsertionEvent(
- tablet,
- isAligned,
- sourceEvent != null ? sourceEvent.getPipeName() : null,
- sourceEvent != null ? sourceEvent.getCreationTime() : 0,
- pipeTaskMeta,
- sourceEvent,
- false);
- }
- return next;
- }
- };
+ return tabletInsertionIterable;
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataTabletIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataTabletIterator.java
index f9fa5a9d283..5fa252412d4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataTabletIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataTabletIterator.java
@@ -60,6 +60,8 @@ public class TsFileInsertionQueryDataTabletIterator
implements Iterator<Tablet>
private final PipeMemoryBlock allocatedBlockForTablet;
+ private RowRecord rowRecord;
+
TsFileInsertionQueryDataTabletIterator(
final TsFileReader tsFileReader,
final Map<String, TSDataType> measurementDataTypeMap,
@@ -130,22 +132,24 @@ public class TsFileInsertionQueryDataTabletIterator
implements Iterator<Tablet>
if (!queryDataSet.hasNext()) {
tablet = new Tablet(deviceId, schemas, 1);
tablet.initBitMaps();
- // Ignore the memory cost of tablet
-
PipeDataNodeResourceManager.memory().forceResize(allocatedBlockForTablet, 0);
return tablet;
}
boolean isFirstRow = true;
while (queryDataSet.hasNext()) {
- final RowRecord rowRecord = queryDataSet.next();
+ final RowRecord rowRecord = this.rowRecord != null ? this.rowRecord :
queryDataSet.next();
if (isFirstRow) {
// Calculate row count and memory size of the tablet based on the
first row
+ this.rowRecord = rowRecord; // Save the first row for later use
Pair<Integer, Integer> rowCountAndMemorySize =
PipeMemoryWeightUtil.calculateTabletRowCountAndMemory(rowRecord);
tablet = new Tablet(deviceId, schemas,
rowCountAndMemorySize.getLeft());
tablet.initBitMaps();
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedBlockForTablet,
rowCountAndMemorySize.getRight());
+ if (allocatedBlockForTablet.getMemoryUsageInBytes() <
rowCountAndMemorySize.getRight()) {
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedBlockForTablet,
rowCountAndMemorySize.getRight());
+ }
+ this.rowRecord = null; // Clear the saved first row
isFirstRow = false;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
index f93642cb9d1..30006b35808 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
@@ -111,7 +111,9 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
// Allocate empty memory block, will be resized later.
this.allocatedMemoryBlockForBatchData =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTabletWithRetry(
+
PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes());
try {
tsFileSequenceReader = new
TsFileSequenceReader(tsFile.getAbsolutePath(), false, false);
@@ -137,44 +139,50 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
@Override
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
- return () ->
- new Iterator<TabletInsertionEvent>() {
-
- @Override
- public boolean hasNext() {
- return Objects.nonNull(chunkReader);
- }
-
- @Override
- public TabletInsertionEvent next() {
- if (!hasNext()) {
- close();
- throw new NoSuchElementException();
- }
-
- // currentIsAligned is initialized when
TsFileInsertionEventScanParser is constructed.
- // When the getNextTablet function is called, currentIsAligned may
be updated, causing
- // the currentIsAligned information to be inconsistent with the
current Tablet
- // information.
- final boolean isAligned = currentIsAligned;
- final Tablet tablet = getNextTablet();
- final boolean hasNext = hasNext();
- try {
- return new PipeRawTabletInsertionEvent(
- tablet,
- isAligned,
- sourceEvent != null ? sourceEvent.getPipeName() : null,
- sourceEvent != null ? sourceEvent.getCreationTime() : 0,
- pipeTaskMeta,
- sourceEvent,
- !hasNext);
- } finally {
- if (!hasNext) {
- close();
- }
- }
- }
- };
+ if (tabletInsertionIterable == null) {
+ tabletInsertionIterable =
+ () ->
+ new Iterator<TabletInsertionEvent>() {
+
+ @Override
+ public boolean hasNext() {
+ return Objects.nonNull(chunkReader);
+ }
+
+ @Override
+ public TabletInsertionEvent next() {
+ if (!hasNext()) {
+ close();
+ throw new NoSuchElementException();
+ }
+
+ // currentIsAligned is initialized when
TsFileInsertionEventScanParser is
+ // constructed.
+ // When the getNextTablet function is called,
currentIsAligned may be updated,
+ // causing
+ // the currentIsAligned information to be inconsistent with
the current Tablet
+ // information.
+ final boolean isAligned = currentIsAligned;
+ final Tablet tablet = getNextTablet();
+ final boolean hasNext = hasNext();
+ try {
+ return new PipeRawTabletInsertionEvent(
+ tablet,
+ isAligned,
+ sourceEvent != null ? sourceEvent.getPipeName() : null,
+ sourceEvent != null ? sourceEvent.getCreationTime() :
0,
+ pipeTaskMeta,
+ sourceEvent,
+ !hasNext);
+ } finally {
+ if (!hasNext) {
+ close();
+ }
+ }
+ }
+ };
+ }
+ return tabletInsertionIterable;
}
public Iterable<Pair<Tablet, Boolean>> toTabletWithIsAligneds() {
@@ -233,8 +241,11 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
tablet =
new Tablet(currentDevice, currentMeasurements,
rowCountAndMemorySize.getLeft());
tablet.initBitMaps();
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedMemoryBlockForTablet,
rowCountAndMemorySize.getRight());
+ if (allocatedMemoryBlockForTablet.getMemoryUsageInBytes()
+ < rowCountAndMemorySize.getRight()) {
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlockForTablet,
rowCountAndMemorySize.getRight());
+ }
isFirstRow = false;
}
@@ -259,8 +270,6 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
if (tablet == null) {
tablet = new Tablet(currentDevice, currentMeasurements, 1);
tablet.initBitMaps();
- // Ignore the memory cost of tablet
-
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForTablet,
0);
}
// Switch chunk reader iff current chunk is all consumed
@@ -287,10 +296,10 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
do {
data = chunkReader.nextPageData();
- PipeDataNodeResourceManager.memory()
- .forceResize(
- allocatedMemoryBlockForBatchData,
- PipeMemoryWeightUtil.calculateBatchDataRamBytesUsed(data));
+ long size = PipeMemoryWeightUtil.calculateBatchDataRamBytesUsed(data);
+ if (allocatedMemoryBlockForBatchData.getMemoryUsageInBytes() < size) {
+
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForBatchData,
size);
+ }
} while (!data.hasCurrent() && chunkReader.hasNextSatisfiedPage());
} while (!data.hasCurrent());
}
