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 087982c22fc [To dev/1.3] Pipe: Replace fixed aligned series limit with
memory-based estimation to avoid OOM in TsFileParser (#16201) (#16232)
087982c22fc is described below
commit 087982c22fc31e9cf129e4a6589b466b659b47cf
Author: nanxiang xia <[email protected]>
AuthorDate: Fri Aug 22 12:16:23 2025 +0800
[To dev/1.3] Pipe: Replace fixed aligned series limit with memory-based
estimation to avoid OOM in TsFileParser (#16201) (#16232)
* chunk_size
* delete PipeMaxAlignedSeriesNumInOneBatch
* change logic
* fix
* fix
* close chunk
* fix comment
---
.../scan/TsFileInsertionScanDataContainer.java | 45 +++++++++++++++++-----
.../pipe/resource/memory/PipeMemoryWeightUtil.java | 5 +++
.../apache/iotdb/commons/conf/CommonConfig.java | 16 ++++----
.../iotdb/commons/pipe/config/PipeConfig.java | 8 ++--
.../iotdb/commons/pipe/config/PipeDescriptor.java | 8 ++--
5 files changed, 59 insertions(+), 23 deletions(-)
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 30006b35808..4223a5155f2 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
@@ -67,9 +67,6 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
private static final LocalDate EMPTY_DATE = LocalDate.of(1000, 1, 1);
- private final int pipeMaxAlignedSeriesNumInOneBatch =
- PipeConfig.getInstance().getPipeMaxAlignedSeriesNumInOneBatch();
-
private final long startTime;
private final long endTime;
private final Filter filter;
@@ -77,6 +74,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
private IChunkReader chunkReader;
private BatchData data;
private final PipeMemoryBlock allocatedMemoryBlockForBatchData;
+ private final PipeMemoryBlock allocatedMemoryBlockForChunk;
private boolean currentIsMultiPage;
private String currentDevice;
@@ -109,11 +107,14 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
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 TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
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 TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
break;
}
+ if (chunkHeader.getDataSize() >
allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
+ PipeDataNodeResourceManager.memory()
+ .forceResize(allocatedMemoryBlockForChunk,
chunkHeader.getDataSize());
+ }
+
chunkReader =
currentIsMultiPage
? new ChunkReader(
@@ -462,17 +469,32 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
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 TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
firstChunkHeader4NextSequentialValueChunks = null;
}
+ valueChunkSize += chunkHeader.getDataSize();
valueChunkList.add(
new Chunk(
chunkHeader, tsFileSequenceReader.readChunk(-1,
chunkHeader.getDataSize())));
@@ -544,5 +567,9 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
if (allocatedMemoryBlockForBatchData != null) {
allocatedMemoryBlockForBatchData.close();
}
+
+ if (allocatedMemoryBlockForChunk != null) {
+ allocatedMemoryBlockForChunk.close();
+ }
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
index d82c26144d6..c9b21d780ee 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.utils.MemUtils;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.BatchData;
+import org.apache.tsfile.read.common.Chunk;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
import org.apache.tsfile.utils.Binary;
@@ -280,6 +281,10 @@ public class PipeMemoryWeightUtil {
return batchData.length() * totalSizeInBytes;
}
+ public static long calculateChunkRamBytesUsed(Chunk chunk) {
+ return chunk != null ? chunk.getRetainedSizeInBytes() : 0L;
+ }
+
/**
* Rounds up the given integer num to the nearest multiple of n.
*
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 a97599fac74..1608303e46f 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
@@ -313,7 +313,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;
@@ -1698,17 +1698,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 39f93fea5af..00423730877 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
@@ -237,8 +237,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() {
@@ -492,7 +492,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 7bdfab83efa..640652345a1 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(
