You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2025/08/22 01:48:46 UTC
(iotdb) branch master updated: Pipe: Replace fixed aligned series limit with memory-based estimation to avoid OOM in TsFileParser (#16201)
This is an automated email from the ASF dual-hosted git repository.
justinchen 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 9356150090d Pipe: Replace fixed aligned series limit with memory-based estimation to avoid OOM in TsFileParser (#16201)
9356150090d is described below
commit 9356150090d7fb1c998fc11cfd3038fc12dfa41d
Author: nanxiang xia <16...@users.noreply.github.com>
AuthorDate: Fri Aug 22 09:48:40 2025 +0800
Pipe: Replace fixed aligned series limit with memory-based estimation to avoid OOM in TsFileParser (#16201)
* chunk_size
* delete PipeMaxAlignedSeriesNumInOneBatch
* change logic
* fix
* fix
* close chunk
* fix comment
---
.../scan/TsFileInsertionEventScanParser.java | 45 +++++++++++++++++-----
.../table/TsFileInsertionEventTableParser.java | 15 ++++++--
...ileInsertionEventTableParserTabletIterator.java | 45 +++++++++++-----------
.../apache/iotdb/commons/conf/CommonConfig.java | 16 ++++----
.../iotdb/commons/pipe/config/PipeConfig.java | 8 ++--
.../iotdb/commons/pipe/config/PipeDescriptor.java | 8 ++--
6 files changed, 88 insertions(+), 49 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 064a2f22279..47aba940a72 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
@@ -64,9 +64,6 @@ import java.util.Objects;
public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
- private final int pipeMaxAlignedSeriesNumInOneBatch =
- PipeConfig.getInstance().getPipeMaxAlignedSeriesNumInOneBatch();
-
private final long startTime;
private final long endTime;
private final Filter filter;
@@ -74,6 +71,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
private IChunkReader chunkReader;
private BatchData data;
private final PipeMemoryBlock allocatedMemoryBlockForBatchData;
+ private final PipeMemoryBlock allocatedMemoryBlockForChunk;
private boolean currentIsMultiPage;
private IDeviceID currentDevice;
@@ -106,11 +104,14 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
this.endTime = endTime;
filter = Objects.nonNull(timeFilterExpression) ? timeFilterExpression.getFilter() : null;
- // Allocate empty memory block, will be resized later.
this.allocatedMemoryBlockForBatchData =
PipeDataNodeResourceManager.memory()
.forceAllocateForTabletWithRetry(
PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes());
+ this.allocatedMemoryBlockForChunk =
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTabletWithRetry(
+ PipeConfig.getInstance().getPipeMaxAlignedSeriesChunkSizeInOneBatch());
try {
tsFileSequenceReader = new TsFileSequenceReader(tsFile.getAbsolutePath(), false, false);
@@ -385,6 +386,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
private void moveToNextChunkReader() throws IOException, IllegalStateException {
ChunkHeader chunkHeader;
+ long valueChunkSize = 0;
final List<Chunk> valueChunkList = new ArrayList<>();
currentMeasurements.clear();
@@ -429,6 +431,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
break;
}
+ if (chunkHeader.getDataSize() > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlockForChunk, chunkHeader.getDataSize());
+ }
+
chunkReader =
currentIsMultiPage
? new ChunkReader(
@@ -462,17 +469,32 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
chunkHeader.getMeasurementID(),
(measurement, index) -> Objects.nonNull(index) ? index + 1 : 0);
- // Emit when encountered non-sequential value chunk, or the chunk list size exceeds
+ // Emit when encountered non-sequential value chunk, or the chunk size exceeds
// certain value to avoid OOM
// Do not record or end current value chunks when there are empty chunks
if (chunkHeader.getDataSize() == 0) {
break;
}
boolean needReturn = false;
- if (lastIndex >= 0
- && (valueIndex != lastIndex
- || valueChunkList.size() >= pipeMaxAlignedSeriesNumInOneBatch)) {
- needReturn = recordAlignedChunk(valueChunkList, marker);
+ final long timeChunkSize =
+ lastIndex >= 0
+ ? PipeMemoryWeightUtil.calculateChunkRamBytesUsed(timeChunkList.get(lastIndex))
+ : 0;
+ if (lastIndex >= 0) {
+ if (valueIndex != lastIndex) {
+ needReturn = recordAlignedChunk(valueChunkList, marker);
+ } else {
+ final long chunkSize = timeChunkSize + valueChunkSize;
+ if (chunkSize + chunkHeader.getDataSize()
+ > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+ if (valueChunkList.size() == 1
+ && chunkSize > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlockForChunk, chunkSize);
+ }
+ needReturn = recordAlignedChunk(valueChunkList, marker);
+ }
+ }
}
lastIndex = valueIndex;
if (needReturn) {
@@ -484,6 +506,7 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
firstChunkHeader4NextSequentialValueChunks = null;
}
+ valueChunkSize += chunkHeader.getDataSize();
valueChunkList.add(
new Chunk(
chunkHeader, tsFileSequenceReader.readChunk(-1, chunkHeader.getDataSize())));
@@ -541,5 +564,9 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
if (allocatedMemoryBlockForBatchData != null) {
allocatedMemoryBlockForBatchData.close();
}
+
+ if (allocatedMemoryBlockForChunk != null) {
+ allocatedMemoryBlockForChunk.close();
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java
index 1ed755f5660..61d0a968249 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParser.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.parser.table;
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.TablePattern;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParser;
@@ -67,15 +68,23 @@ public class TsFileInsertionEventTableParser extends TsFileInsertionEventParser
super(pipeName, creationTime, null, pattern, startTime, endTime, pipeTaskMeta, sourceEvent);
try {
- long tableSize = PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes();
+ long tableSize =
+ Math.min(
+ PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes(),
+ IoTDBDescriptor.getInstance().getConfig().getTargetChunkSize());
+
this.allocatedMemoryBlockForChunk =
- PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTabletWithRetry(
+ PipeConfig.getInstance().getPipeMaxAlignedSeriesChunkSizeInOneBatch());
this.allocatedMemoryBlockForBatchData =
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
this.allocatedMemoryBlockForChunkMeta =
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
this.allocatedMemoryBlockForTableSchemas =
- PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTabletWithRetry(
+ PipeConfig.getInstance().getPipeDataStructureTabletSizeInBytes());
this.startTime = startTime;
this.endTime = endTime;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
index 91f0f3baa2f..1665a8b3e0e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/table/TsFileInsertionEventTableParserTabletIterator.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.pipe.event.common.tsfile.parser.table;
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
@@ -59,9 +58,6 @@ import java.util.stream.Collectors;
public class TsFileInsertionEventTableParserTabletIterator implements Iterator<Tablet> {
- private final int pipeMaxAlignedSeriesNumInOneBatch =
- PipeConfig.getInstance().getPipeMaxAlignedSeriesNumInOneBatch();
-
private final long startTime;
private final long endTime;
@@ -231,9 +227,9 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T
deviceMetaIterator = metadataQuerier.deviceIterator(tableRoot, null);
final int columnSchemaSize = tableSchema.getColumnSchemas().size();
- dataTypeList = new ArrayList<>(pipeMaxAlignedSeriesNumInOneBatch);
- columnTypes = new ArrayList<>(pipeMaxAlignedSeriesNumInOneBatch);
- measurementList = new ArrayList<>(pipeMaxAlignedSeriesNumInOneBatch);
+ dataTypeList = new ArrayList<>();
+ columnTypes = new ArrayList<>();
+ measurementList = new ArrayList<>();
for (int i = 0; i < columnSchemaSize; i++) {
final IMeasurementSchema schema = tableSchema.getColumnSchemas().get(i);
@@ -341,15 +337,14 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T
timeChunk.getData().rewind();
long size = timeChunkSize;
- final List<Chunk> valueChunkList = new ArrayList<>(pipeMaxAlignedSeriesNumInOneBatch);
+ final List<Chunk> valueChunkList = new ArrayList<>();
// To ensure that the Tablet has the same alignedChunk column as the current one,
// you need to create a new Tablet to fill in the data.
isSameDeviceID = false;
// Need to ensure that columnTypes recreates an array
- final List<ColumnCategory> categories =
- new ArrayList<>(deviceIdSize + pipeMaxAlignedSeriesNumInOneBatch);
+ final List<ColumnCategory> categories = new ArrayList<>(deviceIdSize);
for (int i = 0; i < deviceIdSize; i++) {
categories.add(ColumnCategory.TAG);
}
@@ -359,25 +354,29 @@ public class TsFileInsertionEventTableParserTabletIterator implements Iterator<T
measurementList.subList(deviceIdSize, measurementList.size()).clear();
dataTypeList.subList(deviceIdSize, dataTypeList.size()).clear();
- final int startOffset = offset;
for (; offset < alignedChunkMetadata.getValueChunkMetadataList().size(); ++offset) {
final IChunkMetadata metadata = alignedChunkMetadata.getValueChunkMetadataList().get(offset);
if (metadata != null) {
- // Record the column information corresponding to Meta to fill in Tablet
- columnTypes.add(ColumnCategory.FIELD);
- measurementList.add(metadata.getMeasurementUid());
- dataTypeList.add(metadata.getDataType());
-
final Chunk chunk = reader.readMemChunk((ChunkMetadata) metadata);
size += PipeMemoryWeightUtil.calculateChunkRamBytesUsed(chunk);
- if (allocatedMemoryBlockForChunk.getMemoryUsageInBytes() < size) {
- PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForChunk, size);
+ if (size > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+ if (valueChunkList.isEmpty()) {
+ // If the first chunk exceeds the memory limit, we need to allocate more memory
+ PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForChunk, size);
+ columnTypes.add(ColumnCategory.FIELD);
+ measurementList.add(metadata.getMeasurementUid());
+ dataTypeList.add(metadata.getDataType());
+ valueChunkList.add(chunk);
+ ++offset;
+ }
+ break;
+ } else {
+ // Record the column information corresponding to Meta to fill in Tablet
+ columnTypes.add(ColumnCategory.FIELD);
+ measurementList.add(metadata.getMeasurementUid());
+ dataTypeList.add(metadata.getDataType());
+ valueChunkList.add(chunk);
}
-
- valueChunkList.add(chunk);
- }
- if (offset - startOffset >= pipeMaxAlignedSeriesNumInOneBatch) {
- break;
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 1d02fbf1ec7..1d7691cc35e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -308,7 +308,7 @@ public class CommonConfig {
private long pipeMemoryExpanderIntervalSeconds = (long) 3 * 60; // 3Min
private volatile long pipeCheckMemoryEnoughIntervalMs = 10L;
private float pipeLeaderCacheMemoryUsagePercentage = 0.1F;
- private int pipeMaxAlignedSeriesNumInOneBatch = 15;
+ private long pipeMaxAlignedSeriesChunkSizeInOneBatch = (long) 16 * 1024 * 1024; // 16MB;
private long pipeListeningQueueTransferSnapshotThreshold = 1000;
private int pipeSnapshotExecutionMaxBatchSize = 1000;
private long pipeRemainingTimeCommitRateAutoSwitchSeconds = 30;
@@ -1691,17 +1691,19 @@ public class CommonConfig {
"pipeLeaderCacheMemoryUsagePercentage is set to {}", pipeLeaderCacheMemoryUsagePercentage);
}
- public int getPipeMaxAlignedSeriesNumInOneBatch() {
- return pipeMaxAlignedSeriesNumInOneBatch;
+ public long getPipeMaxAlignedSeriesChunkSizeInOneBatch() {
+ return pipeMaxAlignedSeriesChunkSizeInOneBatch;
}
- public void setPipeMaxAlignedSeriesNumInOneBatch(int pipeMaxAlignedSeriesNumInOneBatch) {
- if (this.pipeMaxAlignedSeriesNumInOneBatch == pipeMaxAlignedSeriesNumInOneBatch) {
+ public void setPipeMaxAlignedSeriesChunkSizeInOneBatch(
+ long pipeMaxAlignedSeriesChunkSizeInOneBatch) {
+ if (this.pipeMaxAlignedSeriesChunkSizeInOneBatch == pipeMaxAlignedSeriesChunkSizeInOneBatch) {
return;
}
- this.pipeMaxAlignedSeriesNumInOneBatch = pipeMaxAlignedSeriesNumInOneBatch;
+ this.pipeMaxAlignedSeriesChunkSizeInOneBatch = pipeMaxAlignedSeriesChunkSizeInOneBatch;
logger.info(
- "pipeMaxAlignedSeriesNumInOneBatch is set to {}", pipeMaxAlignedSeriesNumInOneBatch);
+ "pipeMaxAlignedSeriesChunkSizeInOneBatch is set to {}",
+ pipeMaxAlignedSeriesChunkSizeInOneBatch);
}
public long getPipeListeningQueueTransferSnapshotThreshold() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index 7d1064c7462..2b47d4b525d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -233,8 +233,8 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeLeaderCacheMemoryUsagePercentage();
}
- public int getPipeMaxAlignedSeriesNumInOneBatch() {
- return COMMON_CONFIG.getPipeMaxAlignedSeriesNumInOneBatch();
+ public long getPipeMaxAlignedSeriesChunkSizeInOneBatch() {
+ return COMMON_CONFIG.getPipeMaxAlignedSeriesChunkSizeInOneBatch();
}
public long getPipeListeningQueueTransferSnapshotThreshold() {
@@ -488,7 +488,9 @@ public class PipeConfig {
isPipeConnectorRPCThriftCompressionEnabled());
LOGGER.info(
"PipeLeaderCacheMemoryUsagePercentage: {}", getPipeLeaderCacheMemoryUsagePercentage());
- LOGGER.info("PipeMaxAlignedSeriesNumInOneBatch: {}", getPipeMaxAlignedSeriesNumInOneBatch());
+ LOGGER.info(
+ "PipeMaxAlignedSeriesChunkSizeInOneBatch: {}",
+ getPipeMaxAlignedSeriesChunkSizeInOneBatch());
LOGGER.info(
"PipeListeningQueueTransferSnapshotThreshold: {}",
getPipeListeningQueueTransferSnapshotThreshold());
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
index 895b940ebf8..45a3678ec9f 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -473,11 +473,11 @@ public class PipeDescriptor {
properties.getProperty(
"pipe_leader_cache_memory_usage_percentage",
String.valueOf(config.getPipeLeaderCacheMemoryUsagePercentage()))));
- config.setPipeMaxAlignedSeriesNumInOneBatch(
- Integer.parseInt(
+ config.setPipeMaxAlignedSeriesChunkSizeInOneBatch(
+ Long.parseLong(
properties.getProperty(
- "pipe_max_aligned_series_num_in_one_batch",
- String.valueOf(config.getPipeMaxAlignedSeriesNumInOneBatch()))));
+ "pipe_max_aligned_series_chunk_size_in_one_batch",
+ String.valueOf(config.getPipeMaxAlignedSeriesChunkSizeInOneBatch()))));
config.setPipeTransferTsFileSync(
Boolean.parseBoolean(