This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 8915a2e94c8 [Load] Isolate reused TsFile parser memory from Pipe pool
(#18419) (#18436)
8915a2e94c8 is described below
commit 8915a2e94c8f95cd568196c795e449ce32fc667c
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 11 17:10:46 2026 +0800
[Load] Isolate reused TsFile parser memory from Pipe pool (#18419) (#18436)
* fix(load): isolate reused tsfile parser memory from pipe pool
* test(load): cover repeated zero-size parser resize
(cherry picked from commit 85442f00f85ecab9bfaf903568ceb97712d1fc59)
---
.../container/TsFileInsertionDataContainer.java | 40 +++++--
.../query/TsFileInsertionQueryDataContainer.java | 65 ++++++++++--
.../TsFileInsertionQueryDataTabletIterator.java | 10 +-
.../scan/TsFileInsertionScanDataContainer.java | 94 +++++++++++++----
.../TsFileInsertionEventParserMemoryBlock.java | 37 +++++++
.../TsFileInsertionEventParserMemoryManager.java | 75 +++++++++++++
.../converter/LoadTreeTsFileTabletIterator.java | 23 +++-
.../load/memory/LoadTsFileMemoryBlock.java | 22 +++-
.../load/memory/LoadTsFileMemoryManager.java | 44 ++++++++
.../load/memory/LoadTsFileParserMemoryManager.java | 116 +++++++++++++++++++++
.../event/TsFileInsertionDataContainerTest.java | 55 +++++++---
.../LoadTsFileParserPipeMemoryIsolationTest.java | 66 ++++++++++++
.../load/memory/LoadTsFileMemoryManagerTest.java | 20 ++++
13 files changed, 605 insertions(+), 62 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 57757ec1f62..a02f82cd33e 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
@@ -23,9 +23,9 @@ import org.apache.iotdb.commons.path.PatternTreeMap;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryManager;
import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
-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;
import org.apache.iotdb.db.storageengine.dataregion.modification.Modification;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -55,14 +55,16 @@ public abstract class TsFileInsertionDataContainer
implements AutoCloseable {
protected final EnrichedEvent sourceEvent; // used to report progress
// mods entry
- protected PipeMemoryBlock allocatedMemoryBlockForModifications;
+ protected TsFileInsertionEventParserMemoryBlock
allocatedMemoryBlockForModifications;
protected PatternTreeMap<Modification, PatternTreeMapFactory.ModsSerializer>
currentModifications;
protected long parseStartTimeNano = -1;
protected boolean parseStartTimeRecorded = false;
protected boolean parseEndTimeRecorded = false;
- protected final PipeMemoryBlock allocatedMemoryBlockForTablet;
+ protected final TsFileInsertionEventParserMemoryBlock
allocatedMemoryBlockForTablet;
+
+ protected final TsFileInsertionEventParserMemoryManager memoryManager;
protected TsFileSequenceReader tsFileSequenceReader;
@@ -78,6 +80,30 @@ public abstract class TsFileInsertionDataContainer
implements AutoCloseable {
final PipeTaskMeta pipeTaskMeta,
final EnrichedEvent sourceEvent,
final boolean isWithMod) {
+ this(
+ tsFile,
+ pipeName,
+ creationTime,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod,
+ TsFileInsertionEventParserMemoryManager.pipe());
+ }
+
+ protected TsFileInsertionDataContainer(
+ final File tsFile,
+ final String pipeName,
+ final long creationTime,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final boolean isWithMod,
+ final TsFileInsertionEventParserMemoryManager memoryManager) {
this.pipeName = pipeName;
this.creationTime = creationTime;
@@ -89,10 +115,10 @@ public abstract class TsFileInsertionDataContainer
implements AutoCloseable {
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
+ this.memoryManager = memoryManager;
// Allocate empty memory block, will be resized later.
- this.allocatedMemoryBlockForTablet =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+ this.allocatedMemoryBlockForTablet =
memoryManager.forceAllocateForTabletWithRetry(0);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
@@ -166,7 +192,7 @@ public abstract class TsFileInsertionDataContainer
implements AutoCloseable {
protected void releaseTabletMemoryBlock() {
if (allocatedMemoryBlockForTablet != null
&& allocatedMemoryBlockForTablet.getMemoryUsageInBytes() > 0) {
-
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForTablet,
0);
+ allocatedMemoryBlockForTablet.forceResize(0);
}
}
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 7659c3c5555..1754b7a222e 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
@@ -27,9 +27,10 @@ import org.apache.iotdb.commons.utils.TestOnly;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryManager;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
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;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -65,7 +66,7 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
private static final Logger LOGGER =
LoggerFactory.getLogger(TsFileInsertionQueryDataContainer.class);
- private final PipeMemoryBlock allocatedMemoryBlock;
+ private final TsFileInsertionEventParserMemoryBlock allocatedMemoryBlock;
private final TsFileReader tsFileReader;
private final Iterator<Map.Entry<IDeviceID, List<String>>>
deviceMeasurementsMapIterator;
@@ -145,6 +146,35 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
final boolean isWithMod)
throws IOException {
+ this(
+ pipeName,
+ creationTime,
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ deviceIsAlignedMap,
+ deviceMeasurementsMapOverride,
+ isWithMod,
+ TsFileInsertionEventParserMemoryManager.pipe());
+ }
+
+ public TsFileInsertionQueryDataContainer(
+ final String pipeName,
+ final long creationTime,
+ final File tsFile,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final Map<IDeviceID, Boolean> deviceIsAlignedMap,
+ final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
+ final boolean isWithMod,
+ final TsFileInsertionEventParserMemoryManager memoryManager)
+ throws IOException {
super(
tsFile,
pipeName,
@@ -154,7 +184,8 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
endTime,
pipeTaskMeta,
sourceEvent,
- isWithMod);
+ isWithMod,
+ memoryManager);
try {
currentModifications =
@@ -162,8 +193,7 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
: PatternTreeMapFactory.getModsPatternTreeMap();
allocatedMemoryBlockForModifications =
- PipeDataNodeResourceManager.memory()
-
.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
+
memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
final PipeTsFileResourceManager tsFileResourceManager =
PipeDataNodeResourceManager.tsfile();
final Map<IDeviceID, List<String>> deviceMeasurementsMap;
@@ -224,8 +254,7 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
memoryRequiredInBytes +=
PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
}
- allocatedMemoryBlock =
-
PipeDataNodeResourceManager.memory().forceAllocate(memoryRequiredInBytes);
+ allocatedMemoryBlock =
memoryManager.forceAllocate(memoryRequiredInBytes);
final Iterator<Map.Entry<IDeviceID, List<String>>> iterator =
deviceMeasurementsMap.entrySet().iterator();
@@ -301,6 +330,25 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
final boolean isWithMod)
throws IOException {
+ this(
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ deviceMeasurementsMapOverride,
+ isWithMod,
+ TsFileInsertionEventParserMemoryManager.pipe());
+ }
+
+ public TsFileInsertionQueryDataContainer(
+ final File tsFile,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final Map<IDeviceID, List<String>> deviceMeasurementsMapOverride,
+ final boolean isWithMod,
+ final TsFileInsertionEventParserMemoryManager memoryManager)
+ throws IOException {
this(
null,
0,
@@ -312,7 +360,8 @@ public class TsFileInsertionQueryDataContainer extends
TsFileInsertionDataContai
null,
null,
deviceMeasurementsMapOverride,
- isWithMod);
+ isWithMod,
+ memoryManager);
}
private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(
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 2e81f4aa335..a763b45976e 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
@@ -22,9 +22,8 @@ package
org.apache.iotdb.db.pipe.event.common.tsfile.container.query;
import org.apache.iotdb.commons.path.PatternTreeMap;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
-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;
import org.apache.iotdb.db.storageengine.dataregion.modification.Modification;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -66,7 +65,7 @@ public class TsFileInsertionQueryDataTabletIterator
implements Iterator<Tablet>
private final QueryDataSet queryDataSet;
- private final PipeMemoryBlock allocatedBlockForTablet;
+ private final TsFileInsertionEventParserMemoryBlock allocatedBlockForTablet;
// Maintain sorted mods list and current index for each measurement
private final List<ModsOperationUtil.ModsInfo> measurementModsList;
@@ -79,7 +78,7 @@ public class TsFileInsertionQueryDataTabletIterator
implements Iterator<Tablet>
final String deviceId,
final List<String> measurements,
final IExpression timeFilterExpression,
- final PipeMemoryBlock allocatedBlockForTablet,
+ final TsFileInsertionEventParserMemoryBlock allocatedBlockForTablet,
final PatternTreeMap<Modification, PatternTreeMapFactory.ModsSerializer>
currentModifications,
final TabletStringInternPool tabletStringInternPool)
throws IOException {
@@ -160,8 +159,7 @@ public class TsFileInsertionQueryDataTabletIterator
implements Iterator<Tablet>
PipeMemoryWeightUtil.calculateTabletRowCountAndMemory(rowRecord);
tablet = new Tablet(deviceId, schemas,
rowCountAndMemorySize.getLeft());
if (allocatedBlockForTablet.getMemoryUsageInBytes() <
rowCountAndMemorySize.getRight()) {
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedBlockForTablet,
rowCountAndMemorySize.getRight());
+
allocatedBlockForTablet.forceResize(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 17d11c7ee7a..6572c92eb9a 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
@@ -29,9 +29,9 @@ import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryManager;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.util.ModsOperationUtil;
-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;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionPathUtils;
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
@@ -90,8 +90,8 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
private IChunkReader chunkReader;
private BatchData data;
- private final PipeMemoryBlock allocatedMemoryBlockForBatchData;
- private final PipeMemoryBlock allocatedMemoryBlockForChunk;
+ private final TsFileInsertionEventParserMemoryBlock
allocatedMemoryBlockForBatchData;
+ private final TsFileInsertionEventParserMemoryBlock
allocatedMemoryBlockForChunk;
private boolean currentIsMultiPage;
private String currentDevice;
@@ -123,6 +123,31 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
final EnrichedEvent sourceEvent,
final boolean isWithMod)
throws IOException {
+ this(
+ pipeName,
+ creationTime,
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod,
+ TsFileInsertionEventParserMemoryManager.pipe());
+ }
+
+ public TsFileInsertionScanDataContainer(
+ final String pipeName,
+ final long creationTime,
+ final File tsFile,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final boolean isWithMod,
+ final TsFileInsertionEventParserMemoryManager memoryManager)
+ throws IOException {
super(
tsFile,
pipeName,
@@ -132,16 +157,15 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
endTime,
pipeTaskMeta,
sourceEvent,
- isWithMod);
+ isWithMod,
+ memoryManager);
this.startTime = startTime;
this.endTime = endTime;
filter = Objects.nonNull(timeFilterExpression) ?
timeFilterExpression.getFilter() : null;
- this.allocatedMemoryBlockForBatchData =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
- this.allocatedMemoryBlockForChunk =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
+ this.allocatedMemoryBlockForBatchData =
memoryManager.forceAllocateForTabletWithRetry(0);
+ this.allocatedMemoryBlockForChunk =
memoryManager.forceAllocateForTabletWithRetry(0);
try {
currentModifications =
@@ -149,8 +173,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
? ModsOperationUtil.loadModificationsFromTsFile(tsFile)
: PatternTreeMapFactory.getModsPatternTreeMap();
allocatedMemoryBlockForModifications =
- PipeDataNodeResourceManager.memory()
-
.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
+
memoryManager.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
tsFileSequenceReader =
new TsFileSequenceReader(
@@ -178,7 +201,38 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
final EnrichedEvent sourceEvent,
final boolean isWithMod)
throws IOException {
- this(null, 0, tsFile, pattern, startTime, endTime, pipeTaskMeta,
sourceEvent, isWithMod);
+ this(
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod,
+ TsFileInsertionEventParserMemoryManager.pipe());
+ }
+
+ public TsFileInsertionScanDataContainer(
+ final File tsFile,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final boolean isWithMod,
+ final TsFileInsertionEventParserMemoryManager memoryManager)
+ throws IOException {
+ this(
+ null,
+ 0,
+ tsFile,
+ pattern,
+ startTime,
+ endTime,
+ pipeTaskMeta,
+ sourceEvent,
+ isWithMod,
+ memoryManager);
}
@Override
@@ -303,7 +357,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
if (!data.hasCurrent()) {
tablet = new Tablet(currentDevice, currentMeasurements, 1);
// Ignore the memory cost of tablet
-
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForTablet,
0);
+ allocatedMemoryBlockForTablet.forceResize(0);
return tablet;
}
@@ -319,8 +373,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
new Tablet(currentDevice, currentMeasurements,
rowCountAndMemorySize.getLeft());
if (allocatedMemoryBlockForTablet.getMemoryUsageInBytes()
< rowCountAndMemorySize.getRight()) {
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedMemoryBlockForTablet,
rowCountAndMemorySize.getRight());
+
allocatedMemoryBlockForTablet.forceResize(rowCountAndMemorySize.getRight());
}
isFirstRow = false;
}
@@ -422,8 +475,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
private void resizePageDataMemoryIfNeeded(final long
estimatedMemoryUsageInBytes) {
if (allocatedMemoryBlockForBatchData.getMemoryUsageInBytes() <
estimatedMemoryUsageInBytes) {
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedMemoryBlockForBatchData,
estimatedMemoryUsageInBytes);
+
allocatedMemoryBlockForBatchData.forceResize(estimatedMemoryUsageInBytes);
}
}
@@ -569,8 +621,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
}
if (chunkHeader.getDataSize() >
allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedMemoryBlockForChunk,
chunkHeader.getDataSize());
+
allocatedMemoryBlockForChunk.forceResize(chunkHeader.getDataSize());
}
final Chunk chunk =
@@ -927,7 +978,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
final long chunkSize = pendingAlignedChunkGroup.chunkSize +
valueChunk.valueChunkSize;
if (chunkSize > allocatedMemoryBlockForChunk.getMemoryUsageInBytes()) {
-
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForChunk,
chunkSize);
+ allocatedMemoryBlockForChunk.forceResize(chunkSize);
}
}
@@ -943,8 +994,7 @@ public class TsFileInsertionScanDataContainer extends
TsFileInsertionDataContain
calculateMaxAlignedPageMemorySizeWithBatchData(
pendingAlignedChunkGroup.timeChunkIndex, pendingAlignedChunkGroup,
valueChunk);
if (pageMemorySize > getPageDataMemoryLimitInBytes()) {
- PipeDataNodeResourceManager.memory()
- .forceResize(allocatedMemoryBlockForBatchData, pageMemorySize);
+ allocatedMemoryBlockForBatchData.forceResize(pageMemorySize);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryBlock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryBlock.java
new file mode 100644
index 00000000000..84a458028e0
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryBlock.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
+
+/**
+ * Memory block used by a tsfile parser.
+ *
+ * <p>The parser is shared by Pipe and Load. Keeping the block behind this
small interface allows
+ * the same parsing code to use the owning subsystem's memory pool instead of
hard-coding the Pipe
+ * pool.
+ */
+public interface TsFileInsertionEventParserMemoryBlock extends AutoCloseable {
+
+ long getMemoryUsageInBytes();
+
+ void forceResize(long newSizeInBytes);
+
+ @Override
+ void close();
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
new file mode 100644
index 00000000000..68669ce4b55
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParserMemoryManager.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.tsfile.parser;
+
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
+
+/** Allocates parser working memory from the pool owned by the caller. */
+public interface TsFileInsertionEventParserMemoryManager {
+
+ TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(long
sizeInBytes);
+
+ TsFileInsertionEventParserMemoryBlock forceAllocate(long sizeInBytes);
+
+ static TsFileInsertionEventParserMemoryManager pipe() {
+ return PipeHolder.INSTANCE;
+ }
+
+ final class PipeHolder {
+ private static final TsFileInsertionEventParserMemoryManager INSTANCE =
+ new TsFileInsertionEventParserMemoryManager() {
+ @Override
+ public TsFileInsertionEventParserMemoryBlock
forceAllocateForTabletWithRetry(
+ final long sizeInBytes) {
+ return new PipeBlock(
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(sizeInBytes));
+ }
+
+ @Override
+ public TsFileInsertionEventParserMemoryBlock forceAllocate(final
long sizeInBytes) {
+ return new
PipeBlock(PipeDataNodeResourceManager.memory().forceAllocate(sizeInBytes));
+ }
+ };
+ }
+
+ final class PipeBlock implements TsFileInsertionEventParserMemoryBlock {
+ private final PipeMemoryBlock delegate;
+
+ private PipeBlock(final PipeMemoryBlock delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public long getMemoryUsageInBytes() {
+ return delegate.getMemoryUsageInBytes();
+ }
+
+ @Override
+ public void forceResize(final long newSizeInBytes) {
+ PipeDataNodeResourceManager.memory().forceResize(delegate,
newSizeInBytes);
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
index 70ea903baf5..e65ba74244b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java
@@ -22,9 +22,11 @@ package org.apache.iotdb.db.storageengine.load.converter;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
+import
org.apache.iotdb.db.storageengine.load.memory.LoadTsFileParserMemoryManager;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.tsfile.file.metadata.IDeviceID;
@@ -153,10 +155,20 @@ class LoadTreeTsFileTabletIterator
try {
scanParser =
new TsFileInsertionScanDataContainer(
- file, LOAD_TREE_PATTERN, Long.MIN_VALUE, Long.MAX_VALUE, null,
null, isWithMod);
+ file,
+ LOAD_TREE_PATTERN,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE,
+ null,
+ null,
+ isWithMod,
+ LoadTsFileParserMemoryManager.getInstance());
activeIterator = scanParser.toTabletWithIsAligneds().iterator();
return;
} catch (final Exception e) {
+ if (shouldRethrow(e)) {
+ throw toRuntimeException(e);
+ }
if (!switchFromScanToQuery(e)) {
throw toRuntimeException(e);
}
@@ -323,7 +335,8 @@ class LoadTreeTsFileTabletIterator
activeQueryTask.startTime,
activeQueryTask.endTime,
activeQueryTask.toDeviceMeasurementsMap(),
- isWithMod);
+ isWithMod,
+ LoadTsFileParserMemoryManager.getInstance());
final Iterator<TabletInsertionEvent> tabletIterator =
activeQueryParser.toTabletInsertionEvents().iterator();
activeIterator =
@@ -349,6 +362,9 @@ class LoadTreeTsFileTabletIterator
};
return true;
} catch (final Exception e) {
+ if (shouldRethrow(e)) {
+ throw toRuntimeException(e);
+ }
LOGGER.warn(
"Load: Failed to initialize query fallback for device {}
measurements {} in TsFile {}. "
+ "Split or skip this query task and continue.",
@@ -386,7 +402,8 @@ class LoadTreeTsFileTabletIterator
Throwable current = e;
while (Objects.nonNull(current)) {
if (current instanceof InterruptedException
- || current instanceof PipeRuntimeOutOfMemoryCriticalException) {
+ || current instanceof PipeRuntimeOutOfMemoryCriticalException
+ || current instanceof LoadRuntimeOutOfMemoryException) {
return true;
}
current = current.getCause();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
index 623234859b9..2ba1fd7554e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryBlock.java
@@ -33,7 +33,7 @@ import java.util.concurrent.atomic.AtomicLong;
public class LoadTsFileMemoryBlock extends LoadTsFileAbstractMemoryBlock {
private static final Logger LOGGER =
LoggerFactory.getLogger(LoadTsFileMemoryBlock.class);
- private final long totalMemorySizeInBytes;
+ private long totalMemorySizeInBytes;
private final AtomicLong memoryUsageInBytes;
LoadTsFileMemoryBlock(long totalMemorySizeInBytes) {
@@ -79,13 +79,31 @@ public class LoadTsFileMemoryBlock extends
LoadTsFileAbstractMemoryBlock {
.decr(memoryInBytes);
}
+ synchronized long getMemoryUsageInBytes() {
+ return memoryUsageInBytes.get();
+ }
+
+ synchronized long getTotalMemorySizeInBytes() {
+ return totalMemorySizeInBytes;
+ }
+
+ synchronized void setTotalMemorySizeInBytes(final long
totalMemorySizeInBytes) {
+ this.totalMemorySizeInBytes = totalMemorySizeInBytes;
+ }
+
+ public synchronized void forceResize(final long newSizeInBytes) {
+ MEMORY_MANAGER.forceResize(this, newSizeInBytes);
+ }
+
@Override
protected synchronized void releaseAllMemory() {
if (memoryUsageInBytes.get() != 0) {
LOGGER.warn(
"Try to release memory from a memory block {} which has not released
all memory", this);
}
- MEMORY_MANAGER.releaseToQuery(totalMemorySizeInBytes);
+ if (totalMemorySizeInBytes > 0) {
+ MEMORY_MANAGER.releaseToQuery(totalMemorySizeInBytes);
+ }
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java
index c3f88815300..c62d6b5f26a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManager.java
@@ -121,6 +121,50 @@ public class LoadTsFileMemoryManager {
return new LoadTsFileMemoryBlock(sizeInBytes);
}
+ /**
+ * Resize an allocated Load TsFile memory block while keeping the query
memory accounting in sync.
+ *
+ * @throws LoadRuntimeOutOfMemoryException if the additional memory cannot
be allocated
+ */
+ synchronized void forceResize(final LoadTsFileMemoryBlock memoryBlock, final
long newSizeInBytes)
+ throws LoadRuntimeOutOfMemoryException {
+ if (newSizeInBytes < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Load: Invalid memory size %d bytes, must be non-negative",
newSizeInBytes));
+ }
+
+ final long oldSizeInBytes = memoryBlock.getTotalMemorySizeInBytes();
+ if (oldSizeInBytes == newSizeInBytes) {
+ return;
+ }
+
+ if (oldSizeInBytes > newSizeInBytes) {
+ if (memoryBlock.getMemoryUsageInBytes() > newSizeInBytes) {
+ LOGGER.error(
+ "Load: Memory block {} uses more memory than its resized limit {}
bytes",
+ memoryBlock,
+ newSizeInBytes);
+ }
+ releaseToQuery(oldSizeInBytes - newSizeInBytes);
+ memoryBlock.setTotalMemorySizeInBytes(newSizeInBytes);
+ return;
+ }
+
+ final long bytesNeeded = newSizeInBytes - oldSizeInBytes;
+ try {
+ forceAllocateFromQuery(bytesNeeded);
+ } catch (LoadRuntimeOutOfMemoryException e) {
+ if (dataCacheMemoryBlock == null ||
!dataCacheMemoryBlock.doShrink(bytesNeeded)) {
+ throw e;
+ }
+ LOGGER.info(
+ "Load: Query engine memory is insufficient; resized memory block
after shrinking data cache by {} bytes",
+ bytesNeeded);
+ }
+ memoryBlock.setTotalMemorySizeInBytes(newSizeInBytes);
+ }
+
public synchronized LoadTsFileDataCacheMemoryBlock
allocateDataCacheMemoryBlock()
throws LoadRuntimeOutOfMemoryException {
if (dataCacheMemoryBlock == null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java
new file mode 100644
index 00000000000..78b24dd3439
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileParserMemoryManager.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.load.memory;
+
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryManager;
+
+/**
+ * Allocates the working memory of TsFile parsers reused by Load from the
query engine memory pool.
+ */
+public class LoadTsFileParserMemoryManager implements
TsFileInsertionEventParserMemoryManager {
+
+ private static final LoadTsFileMemoryManager LOAD_MEMORY_MANAGER =
+ LoadTsFileMemoryManager.getInstance();
+
+ private LoadTsFileParserMemoryManager() {}
+
+ public static LoadTsFileParserMemoryManager getInstance() {
+ return LoadTsFileParserMemoryManagerHolder.INSTANCE;
+ }
+
+ @Override
+ public TsFileInsertionEventParserMemoryBlock forceAllocateForTabletWithRetry(
+ final long sizeInBytes) {
+ return new LoadParserMemoryBlock(sizeInBytes);
+ }
+
+ @Override
+ public TsFileInsertionEventParserMemoryBlock forceAllocate(final long
sizeInBytes) {
+ return new LoadParserMemoryBlock(sizeInBytes);
+ }
+
+ private static class LoadParserMemoryBlock implements
TsFileInsertionEventParserMemoryBlock {
+
+ private LoadTsFileMemoryBlock delegate;
+ private long memoryUsageInBytes;
+ private boolean isClosed;
+
+ private LoadParserMemoryBlock(final long sizeInBytes) {
+ checkNonNegative(sizeInBytes);
+ if (sizeInBytes > 0) {
+ delegate = LOAD_MEMORY_MANAGER.allocateMemoryBlock(sizeInBytes);
+ }
+ memoryUsageInBytes = sizeInBytes;
+ }
+
+ @Override
+ public synchronized long getMemoryUsageInBytes() {
+ return memoryUsageInBytes;
+ }
+
+ @Override
+ public synchronized void forceResize(final long newSizeInBytes) {
+ checkNonNegative(newSizeInBytes);
+ if (isClosed || memoryUsageInBytes == newSizeInBytes) {
+ return;
+ }
+
+ resizeDelegate(newSizeInBytes);
+ memoryUsageInBytes = newSizeInBytes;
+ }
+
+ private void resizeDelegate(final long newSizeInBytes) {
+ if (newSizeInBytes == 0) {
+ delegate.close();
+ delegate = null;
+ } else if (delegate == null) {
+ delegate = LOAD_MEMORY_MANAGER.allocateMemoryBlock(newSizeInBytes);
+ } else {
+ delegate.forceResize(newSizeInBytes);
+ }
+ }
+
+ @Override
+ public synchronized void close() {
+ if (isClosed) {
+ return;
+ }
+ isClosed = true;
+ memoryUsageInBytes = 0;
+ if (delegate != null) {
+ delegate.close();
+ delegate = null;
+ }
+ }
+
+ private static void checkNonNegative(final long sizeInBytes) {
+ if (sizeInBytes < 0) {
+ throw new IllegalArgumentException(
+ String.format("Load: Invalid memory size %d bytes, must be
non-negative", sizeInBytes));
+ }
+ }
+ }
+
+ private static class LoadTsFileParserMemoryManagerHolder {
+ private static final LoadTsFileParserMemoryManager INSTANCE =
+ new LoadTsFileParserMemoryManager();
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
index 2a2812dd395..dae32e56979 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
@@ -32,7 +32,7 @@ import
org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsert
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.AlignedSinglePageWholeChunkReader;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.SinglePageWholeChunkReader;
import
org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
-import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType;
@@ -174,11 +174,22 @@ public class TsFileInsertionDataContainerTest {
false)) {
replaceAllocatedTabletMemory(
container,
- new PipeMemoryBlock(0) {
+ new TsFileInsertionEventParserMemoryBlock() {
+ private long memoryUsageInBytes;
+
+ @Override
+ public long getMemoryUsageInBytes() {
+ return memoryUsageInBytes;
+ }
+
+ @Override
+ public void forceResize(final long newSizeInBytes) {
+ memoryUsageInBytes = newSizeInBytes;
+ }
+
@Override
public void close() {
- Assert.assertEquals(0, getMemoryUsageInBytes());
- super.close();
+ Assert.assertEquals(0, memoryUsageInBytes);
}
});
@@ -212,13 +223,25 @@ public class TsFileInsertionDataContainerTest {
final AtomicInteger memoryUsageReadCount = new AtomicInteger(0);
replaceAllocatedTabletMemory(
container,
- new PipeMemoryBlock(0) {
+ new TsFileInsertionEventParserMemoryBlock() {
+ private long memoryUsageInBytes;
+
@Override
public long getMemoryUsageInBytes() {
if (memoryUsageReadCount.incrementAndGet() == 2) {
throw new PipeRuntimeOutOfMemoryCriticalException("expected
oom");
}
- return super.getMemoryUsageInBytes();
+ return memoryUsageInBytes;
+ }
+
+ @Override
+ public void forceResize(final long newSizeInBytes) {
+ memoryUsageInBytes = newSizeInBytes;
+ }
+
+ @Override
+ public void close() {
+ memoryUsageInBytes = 0;
}
});
@@ -1336,37 +1359,41 @@ public class TsFileInsertionDataContainerTest {
return count;
}
- private PipeMemoryBlock getAllocatedChunkMemory(final
TsFileInsertionScanDataContainer parser)
+ private TsFileInsertionEventParserMemoryBlock getAllocatedChunkMemory(
+ final TsFileInsertionScanDataContainer parser)
throws NoSuchFieldException, IllegalAccessException {
final Field field =
TsFileInsertionScanDataContainer.class.getDeclaredField("allocatedMemoryBlockForChunk");
field.setAccessible(true);
- return (PipeMemoryBlock) field.get(parser);
+ return (TsFileInsertionEventParserMemoryBlock) field.get(parser);
}
- private PipeMemoryBlock getAllocatedBatchDataMemory(final
TsFileInsertionScanDataContainer parser)
+ private TsFileInsertionEventParserMemoryBlock getAllocatedBatchDataMemory(
+ final TsFileInsertionScanDataContainer parser)
throws NoSuchFieldException, IllegalAccessException {
final Field field =
TsFileInsertionScanDataContainer.class.getDeclaredField("allocatedMemoryBlockForBatchData");
field.setAccessible(true);
- return (PipeMemoryBlock) field.get(parser);
+ return (TsFileInsertionEventParserMemoryBlock) field.get(parser);
}
- private PipeMemoryBlock getAllocatedTabletMemory(final
TsFileInsertionDataContainer container)
+ private TsFileInsertionEventParserMemoryBlock getAllocatedTabletMemory(
+ final TsFileInsertionDataContainer container)
throws NoSuchFieldException, IllegalAccessException {
final Field field =
TsFileInsertionDataContainer.class.getDeclaredField("allocatedMemoryBlockForTablet");
field.setAccessible(true);
- return (PipeMemoryBlock) field.get(container);
+ return (TsFileInsertionEventParserMemoryBlock) field.get(container);
}
private void replaceAllocatedTabletMemory(
- final TsFileInsertionDataContainer container, final PipeMemoryBlock
replacement)
+ final TsFileInsertionDataContainer container,
+ final TsFileInsertionEventParserMemoryBlock replacement)
throws NoSuchFieldException, IllegalAccessException {
final Field field =
TsFileInsertionDataContainer.class.getDeclaredField("allocatedMemoryBlockForTablet");
field.setAccessible(true);
- ((PipeMemoryBlock) field.get(container)).close();
+ ((TsFileInsertionEventParserMemoryBlock) field.get(container)).close();
field.set(container, replacement);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileParserPipeMemoryIsolationTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileParserPipeMemoryIsolationTest.java
new file mode 100644
index 00000000000..2b403b079c3
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileParserPipeMemoryIsolationTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.load.converter;
+
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager;
+
+import org.apache.tsfile.utils.TsFileGeneratorUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.File;
+
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*",
"javax.management.*"})
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(PipeDataNodeResourceManager.class)
+public class LoadTsFileParserPipeMemoryIsolationTest {
+
+ @Test
+ public void testLoadParserDoesNotAccessPipeMemoryPool() throws Exception {
+ final File tsFile = new File("load-parser-pipe-memory-isolation.tsfile");
+ try {
+ TsFileGeneratorUtils.generateNonAlignedTsFile(tsFile.getPath(), 1, 1,
10, 0, 100, 10, 10);
+
+ PowerMockito.mockStatic(PipeDataNodeResourceManager.class);
+ PowerMockito.when(PipeDataNodeResourceManager.memory())
+ .thenThrow(new AssertionError("Load parser must not access Pipe
memory"));
+
+ final LoadTsFileMemoryManager loadMemoryManager =
LoadTsFileMemoryManager.getInstance();
+ final long loadMemoryBefore =
loadMemoryManager.getUsedMemorySizeInBytes();
+ try (final LoadTreeTsFileTabletIterator tabletIterator =
+ new LoadTreeTsFileTabletIterator(tsFile, true)) {
+ Assert.assertTrue(tabletIterator.hasNext());
+ Assert.assertTrue(loadMemoryManager.getUsedMemorySizeInBytes() >
loadMemoryBefore);
+ Assert.assertNotNull(tabletIterator.next());
+ }
+ Assert.assertEquals(loadMemoryBefore,
loadMemoryManager.getUsedMemorySizeInBytes());
+ } finally {
+ if (tsFile.exists()) {
+ Assert.assertTrue(tsFile.delete());
+ }
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java
index f3a1bf7e411..f0ffe1d5f99 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/memory/LoadTsFileMemoryManagerTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.load.memory;
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
+import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.TsFileInsertionEventParserMemoryBlock;
import org.junit.Assert;
import org.junit.Test;
@@ -83,6 +84,25 @@ public class LoadTsFileMemoryManagerTest {
}
}
+ @Test
+ public void testParserMemoryBlockGrowsAndReleasesFromQueryPool() throws
Exception {
+ final LoadTsFileMemoryManager manager =
LoadTsFileMemoryManager.getInstance();
+ final long usedMemoryBefore = manager.getUsedMemorySizeInBytes();
+ final TsFileInsertionEventParserMemoryBlock block =
+ LoadTsFileParserMemoryManager.getInstance().forceAllocate(0);
+
+ Assert.assertEquals(0L, block.getMemoryUsageInBytes());
+ block.forceResize(1024);
+ Assert.assertEquals(usedMemoryBefore + 1024,
manager.getUsedMemorySizeInBytes());
+ block.forceResize(0);
+ Assert.assertEquals(usedMemoryBefore, manager.getUsedMemorySizeInBytes());
+ block.forceResize(0);
+ Assert.assertEquals(0L, block.getMemoryUsageInBytes());
+ Assert.assertEquals(usedMemoryBefore, manager.getUsedMemorySizeInBytes());
+ block.close();
+ Assert.assertEquals(usedMemoryBefore, manager.getUsedMemorySizeInBytes());
+ }
+
private static LoadTsFileMemoryManager newMemoryManager() throws Exception {
final Constructor<LoadTsFileMemoryManager> constructor =
LoadTsFileMemoryManager.class.getDeclaredConstructor();
