This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 35cf94a780a Improve pipe runtime memory allocation (#18051)
35cf94a780a is described below
commit 35cf94a780ab6fe7f105cf1a021e2143f8b503bc
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 7 10:30:27 2026 +0800
Improve pipe runtime memory allocation (#18051)
* Improve pipe runtime memory allocation
* Fix pipe TsFile parser memory self-lock
* Fix Chinese pipe TsFile parser i18n message
* Fix TsFile parser OOM retry test setup
* Address pipe dynamic memory review comments
* Fix scan parser tablet memory release
* Deduplicate pipe meta heartbeat reporting
* Reduce pipe sync file transfer duplication
* Fix pipe i18n review comments
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 10 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 9 +-
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 455 ++++-----------------
.../subtask/processor/PipeProcessorSubtask.java | 2 +
.../common/tsfile/PipeTsFileInsertionEvent.java | 51 ++-
.../tsfile/parser/TsFileInsertionEventParser.java | 12 +-
.../query/TsFileInsertionEventQueryParser.java | 1 +
.../scan/TsFileInsertionEventScanParser.java | 14 +-
.../table/TsFileInsertionEventTableParser.java | 22 +-
.../processor/aggregate/AggregateProcessor.java | 3 +
.../downsampling/DownSamplingProcessor.java | 3 +
.../db/pipe/resource/memory/PipeMemoryManager.java | 17 +-
.../evolvable/batch/PipeTabletEventBatch.java | 69 +++-
.../evolvable/batch/PipeTabletEventPlainBatch.java | 51 ++-
.../batch/PipeTabletEventTsFileBatch.java | 30 +-
.../protocol/airgap/IoTDBDataRegionAirGapSink.java | 18 +-
.../iotconsensusv2/IoTConsensusV2SyncSink.java | 20 +-
.../IoTConsensusV2TsFileInsertionEventHandler.java | 34 +-
.../IoTConsensusV2TransferBatchReqBuilder.java | 109 +++--
.../sink/protocol/legacy/IoTDBLegacyPipeSink.java | 17 +-
.../async/handler/PipeTransferTsFileHandler.java | 42 +-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 136 ++++++
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 88 ----
.../pipe/event/TsFileInsertionEventParserTest.java | 98 +++++
24 files changed, 667 insertions(+), 644 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index e053b83d99d..dad8dcbc7f4 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -293,8 +293,8 @@ public final class DataNodePipeMessages {
+ "PipeInsertNodeTabletInsertionEvent({}) overlaps with the time
range: [{}, {}]. "
+ "Returning true to ensure data integrity.";
public static final String FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE =
- "{}: failed to allocate memory for parsing TsFile {}, tablet event no.
{}, retry count "
- + "is {}, will keep retrying.";
+ "{}: failed to allocate memory for parsing TsFile {}, tablet event no.
{}, "
+ + "will release parser memory and retry the TsFile event later.";
public static final String FAILED_TO_BUILD_TABLET = "Failed to build tablet";
public static final String FAILED_TO_CHECK_NEXT = "Failed to check next";
public static final String FAILED_TO_CLOSE_TSFILEREADER = "Failed to close
TsFileReader";
@@ -2473,5 +2473,11 @@ public final class DataNodePipeMessages {
"Invalid aligned value chunk index %d, while there are %d time chunks.";
public static final String
MESSAGE_FAILED_TO_ROLLBACK_CREATED_REALTIME_PIPE_ARG_STATUS_ARG_CE14334A =
"Failed to rollback created realtime pipe {}. Status: {}";
+ public static final String
LOG_REPORTING_PIPE_META_ARG_ISCOMPLETED_ARG_REMAININGEVENTCOUNT_ARG_8F996DF3 =
+ "Reporting pipe meta: %s, isCompleted: %s, remainingEventCount: %s";
+ public static final String LOG_REPORTED_ARG_PIPE_METAS_12068FC6 =
+ "Reported %s pipe metas.";
+ public static final String
MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD =
+ "Transfer file %s error, result status %s.";
}
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index e0f3fe3a744..b9b042aff52 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -279,7 +279,8 @@ public final class DataNodePipeMessages {
"判断 PipeInsertNodeTabletInsertionEvent({}) 的事件时间是否与时间范围 [{}, {}] 重叠时"
+ "发生异常。为保证数据完整性返回 true。";
public static final String FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE =
- "{}：为解析 TsFile {} 分配内存失败，tablet event 序号 {}，重试次数 {}，将持续重试。";
+ "{}：为解析 TsFile {} 分配内存失败，tablet 事件编号 {}，"
+ + "将释放解析器内存并稍后重试该 TsFile 事件。";
public static final String FAILED_TO_BUILD_TABLET = "构建 tablet 失败";
public static final String FAILED_TO_CHECK_NEXT = "check next 失败";
public static final String FAILED_TO_CLOSE_TSFILEREADER = "关闭 TsFileReader
失败";
@@ -2311,5 +2312,11 @@ public final class DataNodePipeMessages {
"对齐值 chunk 索引 %d 无效，而共有 %d 个 time chunk。";
public static final String
MESSAGE_FAILED_TO_ROLLBACK_CREATED_REALTIME_PIPE_ARG_STATUS_ARG_CE14334A =
"回滚已创建的 realtime pipe {} 失败。状态：{}";
+ public static final String
LOG_REPORTING_PIPE_META_ARG_ISCOMPLETED_ARG_REMAININGEVENTCOUNT_ARG_8F996DF3 =
+ "正在上报 pipe meta：%s，isCompleted：%s，remainingEventCount：%s";
+ public static final String LOG_REPORTED_ARG_PIPE_METAS_12068FC6 =
+ "已上报 %s 个 pipe meta。";
+ public static final String
MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD =
+ "传输文件 %s 出错，结果状态为 %s。";
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
index c069addeeaf..df08311151a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.pipe.agent.task;
-import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
@@ -39,7 +38,6 @@ import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
-import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
@@ -47,10 +45,8 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeType;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
-import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -59,7 +55,6 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.agent.task.builder.PipeDataNodeBuilder;
import org.apache.iotdb.db.pipe.agent.task.builder.PipeDataNodeTaskBuilder;
-import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtaskManager;
import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.metric.overview.PipeTsFileToTabletsMetrics;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
@@ -109,32 +104,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_END_TIME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_END_TIME_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_EXCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_INCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_EXCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_INCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_REALTIME_ENABLE_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_START_TIME_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_END_TIME_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_EXCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_INCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_EXCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_INCLUSION_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY;
-import static
org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_START_TIME_KEY;
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
@@ -199,9 +174,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
|| needConstructDataRegionTask
|| needConstructSchemaRegionTask) {
calculateMemoryUsage(
- pipeStaticMeta,
- Collections.singletonList(new Pair<>(consensusGroupId,
pipeTaskMeta)),
- false);
+ pipeStaticMeta, Collections.singletonList(new
Pair<>(consensusGroupId, pipeTaskMeta)));
final PipeDataNodeTask pipeTask =
new PipeDataNodeTaskBuilder(pipeStaticMeta, consensusGroupId,
pipeTaskMeta).build();
@@ -416,75 +389,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
PipeConfig.getInstance().getPipeMetaReportMaxLogIntervalRounds(),
pipeMetaKeeper.getPipeMetaCount());
- final Set<Integer> dataRegionIds =
- StorageEngine.getInstance().getAllDataRegionIds().stream()
- .map(DataRegionId::getId)
- .collect(Collectors.toSet());
-
- final List<ByteBuffer> pipeMetaBinaryList = new ArrayList<>();
- final List<Boolean> pipeCompletedList = new ArrayList<>();
- final List<Long> pipeRemainingEventCountList = new ArrayList<>();
- final List<Double> pipeRemainingTimeList = new ArrayList<>();
- final List<Integer> pipeDegradedStatusList = new ArrayList<>();
- try {
- for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
- pipeMetaBinaryList.add(pipeMeta.serialize());
-
- final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
-
- final Map<Integer, PipeTask> pipeTaskMap =
pipeTaskManager.getPipeTasks(staticMeta);
- final boolean isAllDataRegionCompleted =
- pipeTaskMap == null
- || pipeTaskMap.entrySet().stream()
- .filter(entry -> dataRegionIds.contains(entry.getKey()))
- .allMatch(entry -> ((PipeDataNodeTask)
entry.getValue()).isCompleted());
- final String sourceModeValue =
- pipeMeta
- .getStaticMeta()
- .getSourceParameters()
- .getStringOrDefault(
- Arrays.asList(
- PipeSourceConstant.EXTRACTOR_MODE_KEY,
PipeSourceConstant.SOURCE_MODE_KEY),
- PipeSourceConstant.EXTRACTOR_MODE_DEFAULT_VALUE);
- final boolean includeDataAndNeedDrop =
-
DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(
- pipeMeta.getStaticMeta().getSourceParameters())
- .getLeft()
- &&
(sourceModeValue.equalsIgnoreCase(PipeSourceConstant.EXTRACTOR_MODE_QUERY_VALUE)
- || sourceModeValue.equalsIgnoreCase(
- PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE));
-
- final boolean isCompleted = isAllDataRegionCompleted &&
includeDataAndNeedDrop;
- final Pair<Long, Double> remainingEventAndTime =
- PipeDataNodeSinglePipeMetrics.getInstance()
- .getRemainingEventAndTime(staticMeta.getPipeName(),
staticMeta.getCreationTime());
- pipeCompletedList.add(isCompleted);
- pipeRemainingEventCountList.add(remainingEventAndTime.getLeft());
- pipeRemainingTimeList.add(remainingEventAndTime.getRight());
- pipeDegradedStatusList.add(
- PipeTemporaryMeta.encodeTsFileEpochDegradedStatus(
- ((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta())
- .getGlobalTsFileEpochDegraded()));
-
- logger.ifPresent(
- l ->
- PipeLogger.log(
- l::info,
- "Reporting pipe meta: %s, isCompleted: %s,
remainingEventCount: %s",
- pipeMeta.coreReportMessage(),
- isCompleted,
- remainingEventAndTime.getLeft()));
- }
- logger.ifPresent(
- l -> PipeLogger.log(l::info, "Reported %s pipe metas.",
pipeMetaBinaryList.size()));
- } catch (final IOException | IllegalPathException e) {
- throw new TException(e);
- }
- resp.setPipeMetaList(pipeMetaBinaryList);
- resp.setPipeCompletedList(pipeCompletedList);
- resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
- resp.setPipeRemainingTimeList(pipeRemainingTimeList);
- resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ collectPipeMetaReport(logger, true).setTo(resp);
PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
}
@@ -506,19 +411,21 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
LOGGER.debug(
DataNodePipeMessages.RECEIVED_PIPE_HEARTBEAT_REQUEST_FROM_CONFIG_NODE,
req.heartbeatId);
+ collectPipeMetaReport(logger, false).setTo(resp);
+ PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
+ }
+
+ private PipeMetaReport collectPipeMetaReport(
+ final Optional<Logger> logger, final boolean includeQueryMode) throws
TException {
final Set<Integer> dataRegionIds =
StorageEngine.getInstance().getAllDataRegionIds().stream()
.map(DataRegionId::getId)
.collect(Collectors.toSet());
- final List<ByteBuffer> pipeMetaBinaryList = new ArrayList<>();
- final List<Boolean> pipeCompletedList = new ArrayList<>();
- final List<Long> pipeRemainingEventCountList = new ArrayList<>();
- final List<Double> pipeRemainingTimeList = new ArrayList<>();
- final List<Integer> pipeDegradedStatusList = new ArrayList<>();
+ final PipeMetaReport report = new PipeMetaReport();
try {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
- pipeMetaBinaryList.add(pipeMeta.serialize());
+ report.pipeMetaBinaryList.add(pipeMeta.serialize());
final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
@@ -528,21 +435,15 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
|| pipeTaskMap.entrySet().stream()
.filter(entry -> dataRegionIds.contains(entry.getKey()))
.allMatch(entry -> ((PipeDataNodeTask)
entry.getValue()).isCompleted());
-
- final boolean includeDataAndNeedDrop =
-
DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(
- pipeMeta.getStaticMeta().getSourceParameters())
- .getLeft()
- &&
isSnapshotMode(pipeMeta.getStaticMeta().getSourceParameters());
-
- final boolean isCompleted = isAllDataRegionCompleted &&
includeDataAndNeedDrop;
+ final boolean isCompleted =
+ isAllDataRegionCompleted && includeDataAndNeedDrop(pipeMeta,
includeQueryMode);
final Pair<Long, Double> remainingEventAndTime =
PipeDataNodeSinglePipeMetrics.getInstance()
.getRemainingEventAndTime(staticMeta.getPipeName(),
staticMeta.getCreationTime());
- pipeCompletedList.add(isCompleted);
- pipeRemainingEventCountList.add(remainingEventAndTime.getLeft());
- pipeRemainingTimeList.add(remainingEventAndTime.getRight());
- pipeDegradedStatusList.add(
+ report.pipeCompletedList.add(isCompleted);
+
report.pipeRemainingEventCountList.add(remainingEventAndTime.getLeft());
+ report.pipeRemainingTimeList.add(remainingEventAndTime.getRight());
+ report.pipeDegradedStatusList.add(
PipeTemporaryMeta.encodeTsFileEpochDegradedStatus(
((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta())
.getGlobalTsFileEpochDegraded()));
@@ -551,22 +452,66 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
l ->
PipeLogger.log(
l::info,
- "Reporting pipe meta: %s, isCompleted: %s,
remainingEventCount: %s",
+ DataNodePipeMessages
+
.LOG_REPORTING_PIPE_META_ARG_ISCOMPLETED_ARG_REMAININGEVENTCOUNT_ARG_8F996DF3,
pipeMeta.coreReportMessage(),
isCompleted,
remainingEventAndTime.getLeft()));
}
logger.ifPresent(
- l -> PipeLogger.log(l::info, "Reported %s pipe metas.",
pipeMetaBinaryList.size()));
+ l ->
+ PipeLogger.log(
+ l::info,
+ DataNodePipeMessages.LOG_REPORTED_ARG_PIPE_METAS_12068FC6,
+ report.pipeMetaBinaryList.size()));
} catch (final IOException | IllegalPathException e) {
throw new TException(e);
}
- resp.setPipeMetaList(pipeMetaBinaryList);
- resp.setPipeCompletedList(pipeCompletedList);
- resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
- resp.setPipeRemainingTimeList(pipeRemainingTimeList);
- resp.setPipeDegradedStatusList(pipeDegradedStatusList);
- PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
+ return report;
+ }
+
+ private boolean includeDataAndNeedDrop(final PipeMeta pipeMeta, final
boolean includeQueryMode)
+ throws IllegalPathException {
+ final PipeParameters sourceParameters =
pipeMeta.getStaticMeta().getSourceParameters();
+ if
(!DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(sourceParameters)
+ .getLeft()) {
+ return false;
+ }
+ if (!includeQueryMode) {
+ return isSnapshotMode(sourceParameters);
+ }
+
+ final String sourceModeValue =
+ sourceParameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSourceConstant.EXTRACTOR_MODE_KEY,
PipeSourceConstant.SOURCE_MODE_KEY),
+ PipeSourceConstant.EXTRACTOR_MODE_DEFAULT_VALUE);
+ return
sourceModeValue.equalsIgnoreCase(PipeSourceConstant.EXTRACTOR_MODE_QUERY_VALUE)
+ ||
sourceModeValue.equalsIgnoreCase(PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE);
+ }
+
+ private static class PipeMetaReport {
+ private final List<ByteBuffer> pipeMetaBinaryList = new ArrayList<>();
+ private final List<Boolean> pipeCompletedList = new ArrayList<>();
+ private final List<Long> pipeRemainingEventCountList = new ArrayList<>();
+ private final List<Double> pipeRemainingTimeList = new ArrayList<>();
+ private final List<Integer> pipeDegradedStatusList = new ArrayList<>();
+
+ private void setTo(final TDataNodeHeartbeatResp resp) {
+ resp.setPipeMetaList(pipeMetaBinaryList);
+ resp.setPipeCompletedList(pipeCompletedList);
+ resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
+ resp.setPipeRemainingTimeList(pipeRemainingTimeList);
+ resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ }
+
+ private void setTo(final TPipeHeartbeatResp resp) {
+ resp.setPipeMetaList(pipeMetaBinaryList);
+ resp.setPipeCompletedList(pipeCompletedList);
+ resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
+ resp.setPipeRemainingTimeList(pipeRemainingTimeList);
+ resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ }
}
///////////////////////// Terminate Logic /////////////////////////
@@ -820,21 +765,11 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
return;
}
- final PipeMeta pipeMetaInAgent =
pipeMetaKeeper.getPipeMeta(staticMeta.getPipeName());
- final boolean ignoreRegisteredSinkSubtasks =
- Objects.nonNull(pipeMetaInAgent)
- && (!pipeMetaInAgent.getStaticMeta().equals(staticMeta)
- ||
PipeStatus.DROPPED.equals(pipeMetaInAgent.getRuntimeMeta().getStatus().get()));
- calculateMemoryUsage(
- staticMeta,
- collectPipeTasksToBeCreated(pipeMetaFromCoordinator),
- ignoreRegisteredSinkSubtasks);
+ calculateMemoryUsage(staticMeta,
collectPipeTasksToBeCreated(pipeMetaFromCoordinator));
}
private void calculateMemoryUsage(
- final PipeStaticMeta staticMeta,
- final List<Pair<Integer, PipeTaskMeta>> pipeTasksToBeCreated,
- final boolean ignoreRegisteredSinkSubtasks)
+ final PipeStaticMeta staticMeta, final List<Pair<Integer, PipeTaskMeta>>
pipeTasksToBeCreated)
throws IllegalPathException {
if (!PipeConfig.getInstance().isPipeEnableMemoryCheck()
|| !isInnerSource(staticMeta.getSourceParameters())
@@ -848,8 +783,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
}
final MemoryEstimation memoryEstimation =
- calculateIncrementalMemoryUsage(
- staticMeta, pipeTasksToBeCreated, ignoreRegisteredSinkSubtasks);
+ calculateIncrementalMemoryUsage(staticMeta, pipeTasksToBeCreated);
calculateInsertNodeQueueMemory(
staticMeta.getSourceParameters(),
memoryEstimation.dataRegionTaskCount);
@@ -917,48 +851,19 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private MemoryEstimation calculateIncrementalMemoryUsage(
final PipeStaticMeta staticMeta,
- final List<Pair<Integer, PipeTaskMeta>> pipeTasksToBeCreated,
- final boolean ignoreRegisteredSinkSubtasks) {
- long needMemory = 0;
+ final List<Pair<Integer, PipeTaskMeta>> pipeTasksToBeCreated) {
int dataRegionTaskCount = 0;
- final Set<String> sinkSubtasksToBeCreated = new HashSet<>();
for (final Pair<Integer, PipeTaskMeta> regionIdAndTaskMeta :
pipeTasksToBeCreated) {
- final int regionId = regionIdAndTaskMeta.getLeft();
- final PipeTaskMeta pipeTaskMeta = regionIdAndTaskMeta.getRight();
- final PipeParameters sourceParameters =
- PipeDataNodeTaskBuilder.blendUserAndSystemParameters(
- staticMeta.getSourceParameters(), pipeTaskMeta);
- final PipeParameters sinkParameters =
- PipeDataNodeTaskBuilder.blendUserAndSystemParameters(
- staticMeta.getSinkParameters(), pipeTaskMeta);
- PipeDataNodeTaskBuilder.preprocessParameters(sourceParameters,
sinkParameters);
-
- final boolean isDataRegionTask = isDataRegionTask(regionId);
- if (isDataRegionTask) {
+ if (isDataRegionTask(regionIdAndTaskMeta.getLeft())) {
dataRegionTaskCount++;
- needMemory += calculateTsFileParserMemory(sourceParameters,
sinkParameters);
- }
-
- final String sinkSubtaskId =
- PipeSinkSubtaskManager.generateAttributeSortedString(sinkParameters,
regionId);
- if (isDataRegionTask
- && !sinkSubtasksToBeCreated.contains(sinkSubtaskId)
- && (ignoreRegisteredSinkSubtasks
- || !PipeSinkSubtaskManager.instance()
- .hasRegisteredSubtasks(sinkParameters, regionId))) {
- sinkSubtasksToBeCreated.add(sinkSubtaskId);
- final int sinkSubtaskNum =
- PipeSinkSubtaskManager.calculateSinkSubtaskNum(sinkParameters,
regionId);
- needMemory += calculateSinkBatchMemory(sinkParameters) *
sinkSubtaskNum;
- needMemory +=
- calculateSendTsFileReadBufferMemory(sourceParameters,
sinkParameters) * sinkSubtaskNum;
}
}
- if (dataRegionTaskCount > 0) {
- needMemory += calculateAssignerMemory(staticMeta.getSourceParameters());
- }
+ // TsFile parser, sink batch, and TsFile read buffer memory are allocated
dynamically
+ // from PipeMemoryManager only while they are active.
+ final long needMemory =
+ dataRegionTaskCount > 0 ?
calculateAssignerMemory(staticMeta.getSourceParameters()) : 0;
return new MemoryEstimation(needMemory, dataRegionTaskCount);
}
@@ -1029,212 +934,6 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent
{
}
}
- private long calculateTsFileParserMemory(
- final PipeParameters sourceParameters, final PipeParameters
sinkParameters) {
-
- // If the source is not history, we do not need to allocate memory
- boolean isExtractorHistory =
- sourceParameters.getBooleanOrDefault(
- SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
- SystemConstant.RESTART_OR_NEWLY_ADDED_DEFAULT_VALUE)
- || sourceParameters.getBooleanOrDefault(
- Arrays.asList(EXTRACTOR_HISTORY_ENABLE_KEY,
SOURCE_HISTORY_ENABLE_KEY),
- EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE);
-
- // If the source is history, and has start/end time, we need to allocate
memory
- boolean isTSFileParser =
- isExtractorHistory
- && sourceParameters.hasAnyAttributes(
- EXTRACTOR_HISTORY_START_TIME_KEY,
SOURCE_HISTORY_START_TIME_KEY);
-
- isTSFileParser =
- isTSFileParser
- || (isExtractorHistory
- && sourceParameters.hasAnyAttributes(
- EXTRACTOR_HISTORY_END_TIME_KEY,
SOURCE_HISTORY_END_TIME_KEY));
-
- // if the source has start/end time, we need to allocate memory
- isTSFileParser =
- isTSFileParser
- || sourceParameters.hasAnyAttributes(SOURCE_START_TIME_KEY,
EXTRACTOR_START_TIME_KEY);
-
- isTSFileParser =
- isTSFileParser
- || sourceParameters.hasAnyAttributes(SOURCE_END_TIME_KEY,
EXTRACTOR_END_TIME_KEY);
-
- // If the source has pattern or path, we need to allocate memory
- isTSFileParser =
- isTSFileParser
- || sourceParameters.hasAnyAttributes(
- EXTRACTOR_PATTERN_KEY,
- SOURCE_PATTERN_KEY,
- EXTRACTOR_PATTERN_INCLUSION_KEY,
- SOURCE_PATTERN_INCLUSION_KEY,
- EXTRACTOR_PATTERN_EXCLUSION_KEY,
- SOURCE_PATTERN_EXCLUSION_KEY);
-
- isTSFileParser =
- isTSFileParser
- || sourceParameters.hasAnyAttributes(
- EXTRACTOR_PATH_KEY,
- SOURCE_PATH_KEY,
- EXTRACTOR_PATH_INCLUSION_KEY,
- SOURCE_PATH_INCLUSION_KEY,
- EXTRACTOR_PATTERN_INCLUSION_KEY,
- SOURCE_PATTERN_INCLUSION_KEY,
- EXTRACTOR_PATH_EXCLUSION_KEY,
- SOURCE_PATH_EXCLUSION_KEY);
-
- // If the source is not hybrid, we do need to allocate memory
- isTSFileParser =
- isTSFileParser
- || !PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE.equals(
- sinkParameters.getStringOrDefault(
- Arrays.asList(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.SINK_FORMAT_KEY),
- PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE));
-
- if (!isTSFileParser) {
- return 0;
- }
-
- return PipeConfig.getInstance().getTsFileParserMemory();
- }
-
- private static long calculateSinkBatchMemory(final PipeParameters
sinkParameters) {
- final String format =
- sinkParameters.getStringOrDefault(
- Arrays.asList(PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.SINK_FORMAT_KEY),
- PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE);
- final boolean usingTsFileBatch =
PipeSinkConstant.CONNECTOR_FORMAT_TS_FILE_VALUE.equals(format);
-
- // TsFile format always uses a batch. Other formats only use a batch when
batch mode is enabled.
- final boolean needUseBatch =
- usingTsFileBatch
- || sinkParameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSinkConstant.CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY,
- PipeSinkConstant.SINK_IOTDB_BATCH_MODE_ENABLE_KEY),
-
PipeSinkConstant.CONNECTOR_IOTDB_BATCH_MODE_ENABLE_DEFAULT_VALUE);
-
- if (!needUseBatch) {
- return 0;
- }
-
- final long batchSizeInBytes =
- sinkParameters.getLongOrDefault(
- Arrays.asList(
- PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY,
- PipeSinkConstant.SINK_IOTDB_BATCH_SIZE_KEY),
- usingTsFileBatch
- ?
PipeSinkConstant.CONNECTOR_IOTDB_TS_FILE_BATCH_SIZE_DEFAULT_VALUE
- :
PipeSinkConstant.CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE);
-
- return batchSizeInBytes * calculateBatchShardCount(sinkParameters,
usingTsFileBatch);
- }
-
- private static long calculateBatchShardCount(
- final PipeParameters sinkParameters, final boolean usingTsFileBatch) {
- if (usingTsFileBatch
- || !sinkParameters.getBooleanOrDefault(
- Arrays.asList(
- PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_KEY,
- PipeSinkConstant.SINK_LEADER_CACHE_ENABLE_KEY),
- PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_DEFAULT_VALUE)) {
- return 1;
- }
-
- // Plain batches always allocate the default batch and may lazily allocate
one batch per target
- // endpoint when leader cache splits events by endpoint.
- return 1L + calculateTargetEndPointCount(sinkParameters);
- }
-
- private static int calculateTargetEndPointCount(final PipeParameters
sinkParameters) {
- final Set<TEndPoint> targetEndPoints = new HashSet<>();
- try {
- addTargetEndPoint(
- targetEndPoints,
- sinkParameters,
- PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY,
- PipeSinkConstant.CONNECTOR_IOTDB_HOST_KEY,
- PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY);
- addTargetEndPoint(
- targetEndPoints,
- sinkParameters,
- PipeSinkConstant.SINK_IOTDB_IP_KEY,
- PipeSinkConstant.SINK_IOTDB_HOST_KEY,
- PipeSinkConstant.SINK_IOTDB_PORT_KEY);
- if
(sinkParameters.hasAttribute(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY)) {
- targetEndPoints.addAll(
- NodeUrlUtils.parseTEndPointUrls(
- Arrays.asList(
- sinkParameters
-
.getStringByKeys(PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY)
- .replace(" ", "")
- .split(","))));
- }
- if
(sinkParameters.hasAttribute(PipeSinkConstant.SINK_IOTDB_NODE_URLS_KEY)) {
- targetEndPoints.addAll(
- NodeUrlUtils.parseTEndPointUrls(
- Arrays.asList(
- sinkParameters
-
.getStringByKeys(PipeSinkConstant.SINK_IOTDB_NODE_URLS_KEY)
- .replace(" ", "")
- .split(","))));
- }
- } catch (final Exception ignored) {
- return 1;
- }
- return Math.max(1, targetEndPoints.size());
- }
-
- private static void addTargetEndPoint(
- final Set<TEndPoint> targetEndPoints,
- final PipeParameters sinkParameters,
- final String ipKey,
- final String hostKey,
- final String portKey) {
- if (sinkParameters.hasAttribute(ipKey) &&
sinkParameters.hasAttribute(portKey)) {
- targetEndPoints.add(
- new TEndPoint(
- sinkParameters.getStringByKeys(ipKey),
sinkParameters.getIntByKeys(portKey)));
- }
- if (sinkParameters.hasAttribute(hostKey) &&
sinkParameters.hasAttribute(portKey)) {
- targetEndPoints.add(
- new TEndPoint(
- sinkParameters.getStringByKeys(hostKey),
sinkParameters.getIntByKeys(portKey)));
- }
- }
-
- private static long calculateSendTsFileReadBufferMemory(
- final PipeParameters sourceParameters, final PipeParameters
sinkParameters) {
- // If the source is history enable, we need to transfer tsfile
- boolean needTransferTsFile =
- sourceParameters.getBooleanOrDefault(
- SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
- SystemConstant.RESTART_OR_NEWLY_ADDED_DEFAULT_VALUE)
- || sourceParameters.getBooleanOrDefault(
- Arrays.asList(EXTRACTOR_HISTORY_ENABLE_KEY,
SOURCE_HISTORY_ENABLE_KEY),
- EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE);
-
- String format =
- sinkParameters.getStringOrDefault(
- Arrays.asList(PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.SINK_FORMAT_KEY),
- PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE);
-
- // If the sink format is tsfile and hybrid, we need to transfer tsfile
- needTransferTsFile =
- needTransferTsFile
- || PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE.equals(format)
- || PipeSinkConstant.CONNECTOR_FORMAT_TS_FILE_VALUE.equals(format);
-
- if (!needTransferTsFile) {
- return 0;
- }
-
- return PipeConfig.getInstance().getPipeSinkReadFileBufferSize();
- }
-
private long calculateAssignerMemory(final PipeParameters sourceParameters) {
try {
if (!PipeInsertionDataNodeListener.getInstance().isEmpty()
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index f9d02f7c2ca..1003c8a0ccb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -179,6 +179,8 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
event1 -> {
try {
pipeProcessor.process(event1, outputEventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
} catch (Exception e) {
ex.set(e);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 085d5388c64..d59b39ecca5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -726,38 +726,37 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
public void consumeTabletInsertionEventsWithRetry(
final TabletInsertionEventConsumer consumer, final String callerName)
throws Exception {
- final Iterable<TabletInsertionEvent> iterable = toTabletInsertionEvents();
- final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
int tabletEventCount = 0;
- while (iterator.hasNext()) {
- final TabletInsertionEvent parsedEvent = iterator.next();
- tabletEventCount++;
- int retryCount = 0;
- while (true) {
- // If failed due do insufficient memory, retry until success to avoid
race among multiple
- // processor threads
+ try {
+ final Iterable<TabletInsertionEvent> iterable =
toTabletInsertionEvents();
+ final Iterator<TabletInsertionEvent> iterator = iterable.iterator();
+ while (iterator.hasNext()) {
+ final TabletInsertionEvent parsedEvent = iterator.next();
+ tabletEventCount++;
try {
consumer.consume((PipeRawTabletInsertionEvent) parsedEvent);
- break;
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
- if (retryCount++ % 100 == 0) {
- LOGGER.warn(
-
DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
- callerName,
- getTsFile(),
- tabletEventCount,
- retryCount);
- } else if (LOGGER.isDebugEnabled()) {
- LOGGER.debug(
-
DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
- callerName,
- getTsFile(),
- tabletEventCount,
- retryCount,
- e);
- }
+ releaseParsedTabletEvent(parsedEvent);
+ throw e;
}
}
+ } catch (final PipeRuntimeOutOfMemoryCriticalException e) {
+ close();
+ LOGGER.warn(
+ DataNodePipeMessages.FAILED_TO_ALLOCATE_MEMORY_FOR_PARSING_TSFILE,
+ callerName,
+ getTsFile(),
+ tabletEventCount,
+ e);
+ throw e;
+ }
+ }
+
+ private void releaseParsedTabletEvent(final TabletInsertionEvent
parsedEvent) {
+ if (parsedEvent instanceof PipeRawTabletInsertionEvent
+ && ((PipeRawTabletInsertionEvent) parsedEvent).getReferenceCount() == 0
+ && !((PipeRawTabletInsertionEvent) parsedEvent).isReleased()) {
+ ((PipeRawTabletInsertionEvent)
parsedEvent).clearReferenceCount(getClass().getName());
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java
index 627731fa7ed..da5cc6b88db 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java
@@ -24,7 +24,6 @@ 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.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser;
@@ -109,9 +108,7 @@ public abstract class TsFileInsertionEventParser implements
AutoCloseable {
this.sourceEvent = sourceEvent;
this.allocatedMemoryBlockForTablet =
- PipeDataNodeResourceManager.memory()
- .forceAllocateForTabletWithRetry(
-
IoTDBDescriptor.getInstance().getConfig().getPipeDataStructureTabletSizeInBytes());
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
LOGGER.debug(
DataNodePipeMessages.TSFILE_HAS_INITIALIZED_PIPENAME_CREATION_TIME_PATTERN,
@@ -180,6 +177,13 @@ public abstract class TsFileInsertionEventParser
implements AutoCloseable {
}
}
+ protected void releaseTabletMemoryBlock() {
+ if (allocatedMemoryBlockForTablet != null
+ && allocatedMemoryBlockForTablet.getMemoryUsageInBytes() > 0) {
+
PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlockForTablet,
0);
+ }
+ }
+
@Override
public void close() {
tabletInsertionIterable = null;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
index 33652f3f3da..ba840b11c32 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java
@@ -524,6 +524,7 @@ public class TsFileInsertionEventQueryParser extends
TsFileInsertionEventParser
final Tablet tablet = tabletIterator.next();
// Record tablet metrics
recordTabletMetrics(tablet);
+ releaseTabletMemoryBlock();
final boolean isAligned =
deviceIsAlignedMap.getOrDefault(
IDeviceID.Factory.DEFAULT_FACTORY.create(tablet.getDeviceId()), false);
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 76d0a0916fb..d4d6ebdc293 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
@@ -29,7 +29,6 @@ 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.TreePattern;
import org.apache.iotdb.db.auth.AuthorityChecker;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -144,12 +143,9 @@ public class TsFileInsertionEventScanParser extends
TsFileInsertionEventParser {
filter = Objects.nonNull(timeFilterExpression) ?
timeFilterExpression.getFilter() : null;
this.allocatedMemoryBlockForBatchData =
- PipeDataNodeResourceManager.memory()
- .forceAllocateForTabletWithRetry(
-
IoTDBDescriptor.getInstance().getConfig().getPipeDataStructureTabletSizeInBytes());
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
this.allocatedMemoryBlockForChunk =
- PipeDataNodeResourceManager.memory()
-
.forceAllocateForTabletWithRetry(PipeConfig.getInstance().getPipeMaxReaderChunkSize());
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
try {
currentModifications =
@@ -225,6 +221,9 @@ public class TsFileInsertionEventScanParser extends
TsFileInsertionEventParser {
throw new NoSuchElementException();
}
+ // Release the previous parser-owned tablet buffer before
allocating the next
+ // tablet.
+ releaseTabletMemoryBlock();
// currentIsAligned is initialized when
TsFileInsertionEventScanParser is
// constructed.
// When the getNextTablet function is called,
currentIsAligned may be updated,
@@ -263,6 +262,9 @@ public class TsFileInsertionEventScanParser extends
TsFileInsertionEventParser {
sourceEvent,
isLast);
} finally {
+ // The raw event owns the generated tablet; only release
the parser-side memory
+ // accounting.
+ releaseTabletMemoryBlock();
if (isLast) {
recordParseEndTimeIfNecessary();
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 1aa7140bb75..187d55a22a2 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,11 +22,9 @@ package
org.apache.iotdb.db.pipe.event.common.tsfile.parser.table;
import org.apache.iotdb.commons.audit.IAuditEntity;
import org.apache.iotdb.commons.exception.auth.AccessDeniedException;
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.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.auth.AuthorityChecker;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -94,25 +92,14 @@ public class TsFileInsertionEventTableParser extends
TsFileInsertionEventParser
allocatedMemoryBlockForModifications =
PipeDataNodeResourceManager.memory()
.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
- long tableSize =
- Math.min(
-
IoTDBDescriptor.getInstance().getConfig().getPipeDataStructureTabletSizeInBytes(),
- IoTDBDescriptor.getInstance().getConfig().getTargetChunkSize());
-
this.allocatedMemoryBlockForChunk =
- PipeDataNodeResourceManager.memory()
- .forceAllocateForTabletWithRetry(
- PipeConfig.getInstance().getPipeMaxReaderChunkSize());
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
this.allocatedMemoryBlockForBatchData =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
this.allocatedMemoryBlockForChunkMeta =
-
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(tableSize);
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
this.allocatedMemoryBlockForTableSchemas =
- PipeDataNodeResourceManager.memory()
- .forceAllocateForTabletWithRetry(
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getPipeDataStructureTabletSizeInBytes());
+
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
this.startTime = startTime;
this.endTime = endTime;
@@ -240,6 +227,7 @@ public class TsFileInsertionEventTableParser extends
TsFileInsertionEventParser
final Tablet tablet = tabletIterator.next();
recordTabletMetrics(tablet);
+ releaseTabletMemoryBlock();
if (!PipeRawTabletInsertionEvent.isTabletEmpty(tablet)) {
bufferedTablet = tablet;
return true;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
index 8cbe56b4328..4d4ed0c4b42 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import
org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskProcessorRuntimeEnvironment;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -549,6 +550,8 @@ public class AggregateProcessor implements PipeProcessor {
event -> {
try {
process(event, eventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
} catch (Exception e) {
ex.set(e);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
index a8e0c270570..fcaa0feb058 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingProcessor.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.processor.downsampling;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskProcessorRuntimeEnvironment;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -156,6 +157,8 @@ public abstract class DownSamplingProcessor implements
PipeProcessor {
event -> {
try {
process(event, eventCollector);
+ } catch (PipeRuntimeOutOfMemoryCriticalException e) {
+ throw e;
} catch (Exception e) {
ex.set(e);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
index eb5e6b9ff0a..3bdda717bef 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -328,12 +328,6 @@ public class PipeMemoryManager {
}
final long oldSize = block.getMemoryUsageInBytes();
- if (oldSize == 0) {
- // If the memory block is not registered, we need to register it first.
- // Otherwise, the memory usage will be inconsistent.
- // See registerMemoryBlock for more details.
- allocatedBlocks.add(block);
- }
if (oldSize >= targetSize) {
memoryBlock.release(oldSize - targetSize);
@@ -349,6 +343,8 @@ public class PipeMemoryManager {
if (targetSize == 0) {
allocatedBlocks.remove(block);
}
+
+ this.notifyAll();
return;
}
@@ -358,6 +354,12 @@ public class PipeMemoryManager {
if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= sizeInBytes) {
memoryBlock.forceAllocateWithoutLimitation(sizeInBytes);
+ if (oldSize == 0) {
+ // If the memory block is not registered, we need to register it
first.
+ // Otherwise, the memory usage will be inconsistent.
+ // See registerMemoryBlock for more details.
+ allocatedBlocks.add(block);
+ }
if (block instanceof PipeTabletMemoryBlock) {
usedMemorySizeInBytesOfTablets += sizeInBytes;
}
@@ -485,6 +487,9 @@ public class PipeMemoryManager {
if (getTotalNonFloatingMemorySizeInBytes() -
memoryBlock.getUsedMemoryInBytes()
>= memoryInBytesNeededToBeAllocated) {
memoryBlock.forceAllocateWithoutLimitation(memoryInBytesNeededToBeAllocated);
+ if (block.getMemoryUsageInBytes() == 0) {
+ allocatedBlocks.add(block);
+ }
if (block instanceof PipeTabletMemoryBlock) {
usedMemorySizeInBytesOfTablets += memoryInBytesNeededToBeAllocated;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
index aede0e994d9..b850de896a1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventBatch.java
@@ -61,8 +61,7 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
// limit in buffer size
this.maxBatchSizeInBytes = requestMaxBatchSizeInBytes;
- this.allocatedMemoryBlock =
-
PipeDataNodeResourceManager.memory().forceAllocate(requestMaxBatchSizeInBytes);
+ this.allocatedMemoryBlock =
PipeDataNodeResourceManager.memory().forceAllocate(0);
if (recordMetric != null) {
this.recordMetric = recordMetric;
} else {
@@ -97,6 +96,10 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
events.add((EnrichedEvent) event);
}
} catch (final Exception e) {
+ if (events.isEmpty()) {
+ clearBatchData();
+ resetMemoryUsage();
+ }
// If the event is not added to the batch, we need to decrease the
reference count.
((EnrichedEvent) event)
.decreaseReferenceCount(PipeTransferBatchReqBuilder.class.getName(), false);
@@ -126,7 +129,27 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
protected abstract boolean constructBatch(final TabletInsertionEvent event)
throws WALPipeException, IOException;
+ protected void increaseTotalBufferSizeAndUpdateMemoryBlock(final long
bufferSize) {
+ if (bufferSize <= 0) {
+ return;
+ }
+
+ final long newTotalBufferSize = Math.min(totalBufferSize + bufferSize,
maxBatchSizeInBytes);
+ PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock,
newTotalBufferSize);
+ totalBufferSize = newTotalBufferSize;
+ }
+
+ protected void releaseAllocatedMemoryBlock() {
+ PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock, 0);
+ }
+
+ protected void clearBatchData() {}
+
public boolean shouldEmit() {
+ if (events.isEmpty()) {
+ return false;
+ }
+
final long diff = System.currentTimeMillis() - firstEventProcessingTime;
if (totalBufferSize >= maxBatchSizeInBytes || diff >= maxDelayInMs) {
recordMetric.accept(diff, totalBufferSize, events.size());
@@ -138,23 +161,26 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
public synchronized void onSuccess() {
events.clear();
- totalBufferSize = 0;
-
- firstEventProcessingTime = Long.MIN_VALUE;
+ resetMemoryUsage();
}
@Override
public synchronized void close() {
+ if (isClosed) {
+ return;
+ }
isClosed = true;
clearEventsReferenceCount(PipeTabletEventBatch.class.getName());
events.clear();
+ clearBatchData();
+ resetMemoryUsage();
allocatedMemoryBlock.close();
}
/**
- * Discard all events of the given pipe. This method only clears the
reference count of the events
- * and discard them, but do not modify other objects (such as buffers) for
simplicity.
+ * Discard all events of the given pipe. This method only clears the
reference count of the
+ * events. If some events remain, cached batch data is kept unchanged for
simplicity.
*/
public synchronized void discardEventsOfPipe(
final String pipeNameToDrop, final long creationTimeToDrop, final int
regionId) {
@@ -162,14 +188,27 @@ public abstract class PipeTabletEventBatch implements
AutoCloseable {
}
public synchronized void discardEventsOfPipe(final CommitterKey
committerKey) {
- events.removeIf(
- event -> {
- if (isEventFromPipe(event, committerKey)) {
-
event.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
- return true;
- }
- return false;
- });
+ final boolean hasDiscardedEvents =
+ events.removeIf(
+ event -> {
+ if (isEventFromPipe(event, committerKey)) {
+
event.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
+ return true;
+ }
+ return false;
+ });
+ if (hasDiscardedEvents && events.isEmpty()) {
+ clearBatchData();
+ resetMemoryUsage();
+ }
+ }
+
+ private void resetMemoryUsage() {
+ totalBufferSize = 0;
+
+ releaseAllocatedMemoryBlock();
+
+ firstEventProcessingTime = Long.MIN_VALUE;
}
private static boolean isEventFromPipe(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java
index abb9ae26608..bdf6ee1874a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventPlainBatch.java
@@ -75,7 +75,6 @@ public class PipeTabletEventPlainBatch extends
PipeTabletEventBatch {
@Override
protected boolean constructBatch(final TabletInsertionEvent event) throws
IOException {
final long bufferSize = buildTabletInsertionBuffer(event);
- totalBufferSize += bufferSize;
pipe2BytesAccumulated.compute(
new Pair<>(
((EnrichedEvent) event).getPipeName(), ((EnrichedEvent)
event).getCreationTime()),
@@ -86,8 +85,13 @@ public class PipeTabletEventPlainBatch extends
PipeTabletEventBatch {
@Override
public synchronized void onSuccess() {
+ clearBatchData();
+
super.onSuccess();
+ }
+ @Override
+ protected void clearBatchData() {
insertNodeBuffers.clear();
tabletBuffers.clear();
@@ -162,24 +166,21 @@ public class PipeTabletEventPlainBatch extends
PipeTabletEventBatch {
final InsertNode insertNode =
pipeInsertNodeTabletInsertionEvent.getInsertNode();
if (!(insertNode instanceof RelationalInsertTabletNode)) {
buffer = insertNode.serializeToByteBuffer();
+ final String databaseName =
+ pipeInsertNodeTabletInsertionEvent.isTableModelEvent()
+ ?
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName()
+ :
pipeInsertNodeTabletInsertionEvent.getTreeModelDatabaseName();
+ estimateSize = RamUsageEstimator.sizeOf(databaseName) + buffer.limit();
+ increaseTotalBufferSizeAndUpdateMemoryBlock(estimateSize);
insertNodeBuffers.add(buffer);
- if (pipeInsertNodeTabletInsertionEvent.isTableModelEvent()) {
- final String databaseName =
- pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName();
- estimateSize = RamUsageEstimator.sizeOf(databaseName);
- insertNodeDataBases.add(databaseName);
- } else {
- final String databaseName =
pipeInsertNodeTabletInsertionEvent.getTreeModelDatabaseName();
- estimateSize = RamUsageEstimator.sizeOf(databaseName);
- insertNodeDataBases.add(databaseName);
- }
- estimateSize += buffer.limit();
+ insertNodeDataBases.add(databaseName);
} else {
- for (final Tablet tablet :
- ((PipeInsertNodeTabletInsertionEvent) event).convertToTablets()) {
- estimateSize +=
- constructTabletBatch(
- tablet,
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
+ final List<Tablet> tablets =
pipeInsertNodeTabletInsertionEvent.convertToTablets();
+ estimateSize = calculateTabletsSizeInBytes(tablets);
+ increaseTotalBufferSizeAndUpdateMemoryBlock(estimateSize);
+ for (final Tablet tablet : tablets) {
+ constructTabletBatchWithoutMemoryReservation(
+ tablet,
pipeInsertNodeTabletInsertionEvent.getTableModelDatabaseName());
}
}
} else {
@@ -199,6 +200,7 @@ public class PipeTabletEventPlainBatch extends
PipeTabletEventBatch {
}
final String databaseName =
pipeRawTabletInsertionEvent.getTreeModelDatabaseName();
estimateSize = RamUsageEstimator.sizeOf(databaseName) + buffer.limit();
+ increaseTotalBufferSizeAndUpdateMemoryBlock(estimateSize);
tabletBuffers.add(buffer);
tabletDataBases.add(databaseName);
}
@@ -208,12 +210,27 @@ public class PipeTabletEventPlainBatch extends
PipeTabletEventBatch {
}
private long constructTabletBatch(final Tablet tablet, final String
databaseName) {
+ final long estimateSize = calculateTabletSizeInBytes(tablet);
+ increaseTotalBufferSizeAndUpdateMemoryBlock(estimateSize);
+ constructTabletBatchWithoutMemoryReservation(tablet, databaseName);
+ return estimateSize;
+ }
+
+ private void constructTabletBatchWithoutMemoryReservation(
+ final Tablet tablet, final String databaseName) {
final Pair<Integer, List<Tablet>> currentBatch =
tableModelTabletMap
.computeIfAbsent(databaseName, k -> new HashMap<>())
.computeIfAbsent(tablet.getTableName(), k -> new Pair<>(0, new
ArrayList<>()));
currentBatch.setLeft(currentBatch.getLeft() + tablet.getRowSize());
currentBatch.getRight().add(tablet);
+ }
+
+ private long calculateTabletsSizeInBytes(final List<Tablet> tablets) {
+ return
tablets.stream().mapToLong(PipeTabletEventPlainBatch::calculateTabletSizeInBytes).sum();
+ }
+
+ private static long calculateTabletSizeInBytes(final Tablet tablet) {
return PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) + 4;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
index 7b511e23fc6..053b42b2c78 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
@@ -86,6 +86,7 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
(PipeInsertNodeTabletInsertionEvent) event;
final boolean isTableModel =
insertNodeTabletInsertionEvent.isTableModelEvent();
final List<Tablet> tablets =
insertNodeTabletInsertionEvent.convertToTablets();
+
increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletsSizeInBytes(tablets));
for (int i = 0; i < tablets.size(); ++i) {
final Tablet tablet = tablets.get(i);
if (isTabletEmpty(tablet)) {
@@ -114,6 +115,7 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
if (isTabletEmpty(tablet)) {
return true;
}
+
increaseTotalBufferSizeAndUpdateMemoryBlock(calculateTabletSizeInBytes(tablet));
if (rawTabletInsertionEvent.isTableModelEvent()) {
// table Model
bufferTableModelTablet(
@@ -139,6 +141,17 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
return true;
}
+ private long calculateTabletsSizeInBytes(final List<Tablet> tablets) {
+ return tablets.stream()
+ .filter(tablet -> !isTabletEmpty(tablet))
+ .mapToLong(PipeTabletEventTsFileBatch::calculateTabletSizeInBytes)
+ .sum();
+ }
+
+ private static long calculateTabletSizeInBytes(final Tablet tablet) {
+ return PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet) * 2;
+ }
+
private void bufferTreeModelTablet(
final String pipeName,
final long creationTime,
@@ -146,11 +159,6 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
final boolean isAligned) {
new
PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
- // TODO: Currently, PipeTreeModelTsFileBuilderV2 still uses
PipeTreeModelTsFileBuilder as a
- // fallback builder, so memory table writing and storing temporary tablets
require double the
- // memory.
- totalBufferSize += PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet)
* 2;
-
pipeName2WeightMap.compute(
new Pair<>(pipeName, creationTime),
(pipe, weight) -> Objects.nonNull(weight) ? ++weight : 1);
@@ -162,11 +170,6 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
final String pipeName, final long creationTime, final Tablet tablet,
final String dataBase) {
new
PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByDevIdTimestamp();
- // TODO: Currently, PipeTableModelTsFileBuilderV2 still uses
PipeTableModelTsFileBuilder as a
- // fallback builder, so memory table writing and storing temporary tablets
require double the
- // memory.
- totalBufferSize += PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet)
* 2;
-
pipeName2WeightMap.compute(
new Pair<>(pipeName, creationTime),
(pipe, weight) -> Objects.nonNull(weight) ? ++weight : 1);
@@ -209,8 +212,13 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
@Override
public synchronized void onSuccess() {
+ clearBatchData();
+
super.onSuccess();
+ }
+ @Override
+ protected void clearBatchData() {
pipeName2WeightMap.clear();
tableModeTsFileBuilder.onSuccess();
treeModeTsFileBuilder.onSuccess();
@@ -220,8 +228,6 @@ public class PipeTabletEventTsFileBatch extends
PipeTabletEventBatch {
public synchronized void close() {
super.close();
- pipeName2WeightMap.clear();
-
tableModeTsFileBuilder.close();
treeModeTsFileBuilder.close();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
index f082a352182..8689069334e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java
@@ -33,6 +33,8 @@ import
org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventBatch;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventPlainBatch;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventTsFileBatch;
@@ -502,10 +504,13 @@ public class IoTDBDataRegionAirGapSink extends
IoTDBDataNodeAirGapSink {
final AirGapSocket socket,
final boolean isMultiFile)
throws PipeException, IOException {
- final int readFileBufferSize = PIPE_CONFIG.getPipeSinkReadFileBufferSize();
- final byte[] readBuffer = new byte[readFileBufferSize];
- long position = 0;
- try (final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
+ final int readFileBufferSize = getReadFileBufferSize(file);
+ try (final PipeTsFileMemoryBlock ignored =
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTsFileWithRetry(readFileBufferSize);
+ final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
+ final byte[] readBuffer = new byte[readFileBufferSize];
+ long position = 0;
while (true) {
mayLimitRateAndRecordIO(readFileBufferSize);
final int readLength = reader.read(readBuffer);
@@ -537,6 +542,11 @@ public class IoTDBDataRegionAirGapSink extends
IoTDBDataNodeAirGapSink {
}
}
+ private int getReadFileBufferSize(final File file) {
+ return (int)
+ Math.min((long) PIPE_CONFIG.getPipeSinkReadFileBufferSize(),
Math.max(file.length(), 1L));
+ }
+
private boolean sendBatch(
final AirGapSocket socket,
byte[] bytes,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
index 7f82ce8e9a9..5b47b92b092 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
@@ -39,6 +39,8 @@ import
org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.builder.IoTConsensusV2SyncBatchReqBuilder;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2DeleteNodeReq;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TabletInsertNodeReq;
@@ -439,10 +441,13 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
final TCommitId tCommitId,
final TConsensusGroupId tConsensusGroupId)
throws PipeException, IOException {
- final int readFileBufferSize =
PipeConfig.getInstance().getPipeSinkReadFileBufferSize();
- final byte[] readBuffer = new byte[readFileBufferSize];
- long position = 0;
- try (final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
+ final int readFileBufferSize = getReadFileBufferSize(file);
+ try (final PipeTsFileMemoryBlock ignored =
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTsFileWithRetry(readFileBufferSize);
+ final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
+ final byte[] readBuffer = new byte[readFileBufferSize];
+ long position = 0;
while (true) {
final int readLength = reader.read(readBuffer);
if (readLength == -1) {
@@ -508,6 +513,13 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
}
}
+ private int getReadFileBufferSize(final File file) {
+ return (int)
+ Math.min(
+ (long) PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
+ Math.max(file.length(), 1L));
+ }
+
private TEndPoint getFollowerUrl() {
// In current iotConsensusV2 design, one connector corresponds to one
follower, so the peers is
// actually a singleton list
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
index 4e269aaa7e8..52815e645bf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
@@ -31,6 +31,8 @@ import
org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferR
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TsFilePieceReq;
import
org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.request.IoTConsensusV2TsFilePieceWithModReq;
@@ -70,7 +72,8 @@ public class IoTConsensusV2TsFileInsertionEventHandler
private final boolean transferMod;
private final int readFileBufferSize;
- private final byte[] readBuffer;
+ private PipeTsFileMemoryBlock memoryBlock;
+ private byte[] readBuffer;
private long position;
private RandomAccessFile reader;
@@ -106,8 +109,15 @@ public class IoTConsensusV2TsFileInsertionEventHandler
transferMod = event.isWithMod();
currentFile = transferMod ? modFile : tsFile;
- readFileBufferSize =
PipeConfig.getInstance().getPipeSinkReadFileBufferSize();
- readBuffer = new byte[readFileBufferSize];
+ final long maxFileLength =
+ transferMod && Objects.nonNull(modFile)
+ ? Math.max(tsFile.length(), modFile.length())
+ : tsFile.length();
+ readFileBufferSize =
+ (int)
+ Math.min(
+ (long)
PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
+ Math.max(maxFileLength, 1L));
position = 0;
reader =
@@ -128,6 +138,12 @@ public class IoTConsensusV2TsFileInsertionEventHandler
this.client = client;
client.setShouldReturnSelf(false);
+ if (readBuffer == null) {
+ memoryBlock =
+
PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize);
+ readBuffer = new byte[readFileBufferSize];
+ }
+
final int readLength = reader.read(readBuffer);
if (readLength == -1) {
if (currentFile == modFile) {
@@ -246,6 +262,8 @@ public class IoTConsensusV2TsFileInsertionEventHandler
client.returnSelf();
}
+ releaseReadBufferMemoryBlock();
+
long duration = System.nanoTime() - createTime;
metric.recordConnectorTsFileTransferTimer(duration);
}
@@ -330,10 +348,20 @@ public class IoTConsensusV2TsFileInsertionEventHandler
connector.addFailureEventToRetryQueue(event);
metric.recordRetryCounter();
+ releaseReadBufferMemoryBlock();
+
if (client != null) {
client.setShouldReturnSelf(true);
client.returnSelf();
}
}
}
+
+ private void releaseReadBufferMemoryBlock() {
+ if (memoryBlock != null) {
+ memoryBlock.close();
+ memoryBlock = null;
+ readBuffer = null;
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java
index d8fece99a3e..fc387084a00 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/payload/builder/IoTConsensusV2TransferBatchReqBuilder.java
@@ -70,6 +70,7 @@ public abstract class IoTConsensusV2TransferBatchReqBuilder
implements AutoClose
protected long firstEventProcessingTime = Long.MIN_VALUE;
// limit in buffer size
+ protected final long maxBatchSizeInBytes;
protected final PipeMemoryBlock allocatedMemoryBlock;
protected long totalBufferSize = 0;
@@ -92,36 +93,12 @@ public abstract class IoTConsensusV2TransferBatchReqBuilder
implements AutoClose
this.consensusGroupId = consensusGroupId;
this.thisDataNodeId = thisDataNodeId;
- final long requestMaxBatchSizeInBytes =
+ maxBatchSizeInBytes =
parameters.getLongOrDefault(
Arrays.asList(CONNECTOR_IOTDB_BATCH_SIZE_KEY,
SINK_IOTDB_BATCH_SIZE_KEY),
CONNECTOR_IOTDB_PLAIN_BATCH_SIZE_DEFAULT_VALUE);
- allocatedMemoryBlock =
- PipeDataNodeResourceManager.memory()
- .tryAllocate(requestMaxBatchSizeInBytes)
- .setShrinkMethod(oldMemory -> Math.max(oldMemory / 2, 0))
- .setShrinkCallback(
- (oldMemory, newMemory) ->
- LOGGER.info(
-
DataNodePipeMessages.THE_BATCH_SIZE_LIMIT_HAS_SHRUNK_FROM,
- oldMemory,
- newMemory))
- .setExpandMethod(
- oldMemory -> Math.min(Math.max(oldMemory, 1) * 2,
requestMaxBatchSizeInBytes))
- .setExpandCallback(
- (oldMemory, newMemory) ->
- LOGGER.info(
-
DataNodePipeMessages.THE_BATCH_SIZE_LIMIT_HAS_EXPANDED_FROM,
- oldMemory,
- newMemory));
-
- if (getMaxBatchSizeInBytes() != requestMaxBatchSizeInBytes) {
- LOGGER.info(
-
DataNodePipeMessages.IOTCONSENSUSV2TRANSFERBATCHREQBUILDER_THE_MAX_BATCH_SIZE_IS_ADJUSTED,
- requestMaxBatchSizeInBytes,
- getMaxBatchSizeInBytes());
- }
+ allocatedMemoryBlock =
PipeDataNodeResourceManager.memory().forceAllocate(0);
}
/**
@@ -136,27 +113,74 @@ public abstract class
IoTConsensusV2TransferBatchReqBuilder implements AutoClose
return false;
}
- final long requestCommitId = ((EnrichedEvent)
event).getReplicateIndexForIoTV2();
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ final long requestCommitId = enrichedEvent.getReplicateIndexForIoTV2();
// The deduplication logic here is to avoid the accumulation of the same
event in a batch when
// retrying.
if ((events.isEmpty() || !events.get(events.size() - 1).equals(event))) {
- events.add((EnrichedEvent) event);
- requestCommitIds.add(requestCommitId);
- final int bufferSize = buildTabletInsertionBuffer(event);
-
- ((EnrichedEvent) event)
-
.increaseReferenceCount(IoTConsensusV2TransferBatchReqBuilder.class.getName());
+ if (!enrichedEvent.increaseReferenceCount(
+ IoTConsensusV2TransferBatchReqBuilder.class.getName())) {
+
LOGGER.warn(DataNodePipeMessages.CANNOT_INCREASE_REFERENCE_COUNT_FOR_EVENT_IGNORE,
event);
+ return shouldEmit();
+ }
- if (firstEventProcessingTime == Long.MIN_VALUE) {
- firstEventProcessingTime = System.currentTimeMillis();
+ final int previousEventsSize = events.size();
+ final int previousRequestCommitIdsSize = requestCommitIds.size();
+ final int previousBatchReqsSize = batchReqs.size();
+ try {
+ events.add(enrichedEvent);
+ requestCommitIds.add(requestCommitId);
+ final int bufferSize = buildTabletInsertionBuffer(event);
+ increaseTotalBufferSizeAndUpdateMemoryBlock(bufferSize);
+
+ if (firstEventProcessingTime == Long.MIN_VALUE) {
+ firstEventProcessingTime = System.currentTimeMillis();
+ }
+ } catch (final Exception e) {
+ rollbackTo(previousEventsSize, previousRequestCommitIdsSize,
previousBatchReqsSize);
+ if (events.isEmpty()) {
+ resetMemoryUsage();
+ }
+ enrichedEvent.decreaseReferenceCount(
+ IoTConsensusV2TransferBatchReqBuilder.class.getName(), false);
+ throw e;
}
+ }
+
+ return shouldEmit();
+ }
+
+ private boolean shouldEmit() {
+ return !events.isEmpty()
+ && (totalBufferSize >= getMaxBatchSizeInBytes()
+ || System.currentTimeMillis() - firstEventProcessingTime >=
maxDelayInMs);
+ }
- totalBufferSize += bufferSize;
+ private void increaseTotalBufferSizeAndUpdateMemoryBlock(final long
bufferSize) {
+ if (bufferSize <= 0) {
+ return;
}
- return totalBufferSize >= getMaxBatchSizeInBytes()
- || System.currentTimeMillis() - firstEventProcessingTime >=
maxDelayInMs;
+ final long newTotalBufferSize =
+ Math.min(totalBufferSize + bufferSize, getMaxBatchSizeInBytes());
+ PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock,
newTotalBufferSize);
+ totalBufferSize = newTotalBufferSize;
+ }
+
+ private void rollbackTo(
+ final int previousEventsSize,
+ final int previousRequestCommitIdsSize,
+ final int previousBatchReqsSize) {
+ events.subList(previousEventsSize, events.size()).clear();
+ requestCommitIds.subList(previousRequestCommitIdsSize,
requestCommitIds.size()).clear();
+ batchReqs.subList(previousBatchReqsSize, batchReqs.size()).clear();
+ }
+
+ private void resetMemoryUsage() {
+ firstEventProcessingTime = Long.MIN_VALUE;
+ totalBufferSize = 0;
+ PipeDataNodeResourceManager.memory().forceResize(allocatedMemoryBlock, 0);
}
public synchronized void onSuccess() {
@@ -165,9 +189,7 @@ public abstract class IoTConsensusV2TransferBatchReqBuilder
implements AutoClose
events.clear();
requestCommitIds.clear();
- firstEventProcessingTime = Long.MIN_VALUE;
-
- totalBufferSize = 0;
+ resetMemoryUsage();
}
public IoTConsensusV2TabletBatchReq toTIoTConsensusV2BatchTransferReq()
throws IOException {
@@ -175,7 +197,7 @@ public abstract class IoTConsensusV2TransferBatchReqBuilder
implements AutoClose
}
protected long getMaxBatchSizeInBytes() {
- return allocatedMemoryBlock.getMemoryUsageInBytes();
+ return maxBatchSizeInBytes;
}
public boolean isEmpty() {
@@ -219,6 +241,9 @@ public abstract class IoTConsensusV2TransferBatchReqBuilder
implements AutoClose
((EnrichedEvent) event).clearReferenceCount(this.getClass().getName());
}
}
+ batchReqs.clear();
+ events.clear();
+ requestCommitIds.clear();
allocatedMemoryBlock.close();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
index 5f850bc311f..bf37e02c8d5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
@@ -37,6 +37,8 @@ import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
@@ -421,8 +423,12 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
long position = 0;
// Try small piece to rebase the file position.
- final byte[] buffer = new
byte[PipeConfig.getInstance().getPipeSinkReadFileBufferSize()];
- try (final RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"r")) {
+ final int readFileBufferSize = getReadFileBufferSize(file);
+ try (final PipeTsFileMemoryBlock ignored =
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTsFileWithRetry(readFileBufferSize);
+ final RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"r")) {
+ final byte[] buffer = new byte[readFileBufferSize];
while (true) {
final int dataLength = randomAccessFile.read(buffer);
if (dataLength == -1) {
@@ -463,6 +469,13 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
}
}
+ private int getReadFileBufferSize(final File file) {
+ return (int)
+ Math.min(
+ (long) PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
+ Math.max(file.length(), 1L));
+ }
+
@Override
public void close() throws Exception {
if (client != null) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
index a0ee47f09e1..700f241eb1f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
@@ -124,11 +124,15 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
// the memory of the TsFile event is not released, so the memory is not
enough for slicing. This
// will cause a deadlock.
waitForResourceEnough4Slicing((long) ((1 + Math.random()) * 20 * 1000));
// 20 - 40 seconds
+ final long maxFileLength =
+ transferMod && Objects.nonNull(modFile)
+ ? Math.max(tsFile.length(), modFile.length())
+ : tsFile.length();
readFileBufferSize =
(int)
Math.min(
- PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
- transferMod ? Math.max(tsFile.length(), modFile.length()) :
tsFile.length());
+ (long)
PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
+ Math.max(maxFileLength, 1L));
position = 0;
isSealSignalSent = new AtomicBoolean(false);
@@ -142,21 +146,6 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
final IoTDBDataNodeAsyncClientManager clientManager,
final AsyncPipeDataTransferServiceClient client)
throws TException, IOException {
- // Delay creation of resources to avoid OOM or too many open files
- if (readBuffer == null) {
- memoryBlock =
- PipeDataNodeResourceManager.memory()
- .forceAllocateForTsFileWithRetry(
-
PipeConfig.getInstance().isPipeSinkReadFileBufferMemoryControlEnabled()
- ? readFileBufferSize
- : 0);
- readBuffer = new byte[readFileBufferSize];
- }
-
- if (reader == null) {
- reader = transferMod ? new RandomAccessFile(modFile, "r") : new
RandomAccessFile(tsFile, "r");
- }
-
this.clientManager = clientManager;
this.client = client;
@@ -173,6 +162,17 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
return;
}
+ // Delay creation of resources to avoid OOM or too many open files
+ if (readBuffer == null) {
+ memoryBlock =
+
PipeDataNodeResourceManager.memory().forceAllocateForTsFileWithRetry(readFileBufferSize);
+ readBuffer = new byte[readFileBufferSize];
+ }
+
+ if (reader == null) {
+ reader = transferMod ? new RandomAccessFile(modFile, "r") : new
RandomAccessFile(tsFile, "r");
+ }
+
client.setShouldReturnSelf(false);
client.setTimeoutDynamically(clientManager.getConnectionTimeout());
@@ -256,6 +256,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
super.onComplete(response);
} finally {
if (sink.isClosed()) {
+ releaseReadBufferMemoryBlock();
returnClientIfNecessary();
}
}
@@ -319,6 +320,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
referenceCount);
}
+ releaseReadBufferMemoryBlock();
returnClientIfNecessary();
}
@@ -361,6 +363,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
try {
super.onError(exception);
} finally {
+ releaseReadBufferMemoryBlock();
returnClientIfNecessary();
}
}
@@ -412,6 +415,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_CLOSE_FILE_READER_OR_DELETE,
e);
} finally {
try {
+ releaseReadBufferMemoryBlock();
returnClientIfNecessary();
} finally {
if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) {
@@ -473,10 +477,14 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
@Override
public void close() {
super.close();
+ releaseReadBufferMemoryBlock();
+ }
+ private void releaseReadBufferMemoryBlock() {
if (memoryBlock != null) {
memoryBlock.close();
memoryBlock = null;
+ readBuffer = null;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
index 72703aa57f2..d13b9664f8a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java
@@ -22,10 +22,12 @@ package org.apache.iotdb.db.pipe.sink.protocol.thrift.sync;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
import org.apache.iotdb.commons.pipe.sink.limiter.TsFileSendRateLimiter;
import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeTransferFilePieceReq;
+import
org.apache.iotdb.commons.pipe.sink.payload.thrift.response.PipeTransferFilePieceResp;
import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
@@ -36,6 +38,8 @@ import
org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.metric.overview.PipeResourceMetrics;
import org.apache.iotdb.db.pipe.metric.sink.PipeDataRegionSinkMetrics;
+import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
+import org.apache.iotdb.db.pipe.resource.memory.PipeTsFileMemoryBlock;
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventBatch;
import
org.apache.iotdb.db.pipe.sink.payload.evolvable.batch.PipeTabletEventPlainBatch;
@@ -71,6 +75,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
+import java.io.RandomAccessFile;
import java.nio.file.NoSuchFileException;
import java.util.Arrays;
import java.util.Collections;
@@ -610,6 +615,137 @@ public class IoTDBDataRegionSyncSink extends
IoTDBDataNodeSyncSink {
LOGGER.info(DataNodePipeMessages.SUCCESSFULLY_TRANSFERRED_FILE, tsFile);
}
+ @Override
+ protected void transferFilePieces(
+ final Map<Pair<String, Long>, Double> pipe2WeightMap,
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final boolean isMultiFile)
+ throws PipeException, IOException {
+ final int readFileBufferSize = getReadFileBufferSize(file);
+ try (final PipeTsFileMemoryBlock ignored =
+ PipeDataNodeResourceManager.memory()
+ .forceAllocateForTsFileWithRetry(readFileBufferSize);
+ final RandomAccessFile reader = new RandomAccessFile(file, "r")) {
+ final byte[] readBuffer = new byte[readFileBufferSize];
+ long position = 0;
+ int readLength;
+ while ((readLength = readNextFilePiece(reader, readBuffer,
readFileBufferSize)) != -1) {
+ position =
+ transferFilePiece(
+ pipe2WeightMap,
+ file,
+ clientAndStatus,
+ isMultiFile,
+ readBuffer,
+ position,
+ readLength,
+ reader);
+ }
+ }
+ }
+
+ private int readNextFilePiece(
+ final RandomAccessFile reader, final byte[] readBuffer, final int
readFileBufferSize)
+ throws IOException {
+ mayLimitRateAndRecordIO(readFileBufferSize);
+ return reader.read(readBuffer);
+ }
+
+ private long transferFilePiece(
+ final Map<Pair<String, Long>, Double> pipe2WeightMap,
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final boolean isMultiFile,
+ final byte[] readBuffer,
+ final long position,
+ final int readLength,
+ final RandomAccessFile reader)
+ throws PipeException, IOException {
+ final byte[] payLoad = buildFilePiecePayload(readBuffer, readLength);
+ final PipeTransferFilePieceResp resp =
+ doTransferFilePiece(pipe2WeightMap, file, clientAndStatus,
isMultiFile, payLoad, position);
+ return handleTransferFilePieceResp(file, clientAndStatus, reader, position
+ readLength, resp);
+ }
+
+ private byte[] buildFilePiecePayload(final byte[] readBuffer, final int
readLength) {
+ return readLength == readBuffer.length
+ ? readBuffer
+ : Arrays.copyOfRange(readBuffer, 0, readLength);
+ }
+
+ private PipeTransferFilePieceResp doTransferFilePiece(
+ final Map<Pair<String, Long>, Double> pipe2WeightMap,
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final boolean isMultiFile,
+ final byte[] payLoad,
+ final long position)
+ throws PipeException, IOException {
+ try {
+ final TPipeTransferReq req =
+ compressIfNeeded(
+ isMultiFile
+ ? getTransferMultiFilePieceReq(file.getName(), position,
payLoad)
+ : getTransferSingleFilePieceReq(file.getName(), position,
payLoad));
+ pipe2WeightMap.forEach(
+ (namePair, weight) ->
+ rateLimitIfNeeded(
+ namePair.getLeft(),
+ namePair.getRight(),
+ clientAndStatus.getLeft().getEndPoint(),
+ (long) (req.getBody().length * weight)));
+ return PipeTransferFilePieceResp.fromTPipeTransferResp(
+ clientAndStatus.getLeft().pipeTransfer(req));
+ } catch (final Exception e) {
+ clientAndStatus.setRight(false);
+ throw new PipeConnectionException(
+ String.format(
+ DataNodePipeMessages
+
.PIPE_EXCEPTION_NETWORK_ERROR_WHEN_TRANSFER_FILE_S_BECAUSE_S_3C673B7A,
+ file,
+ e.getMessage()),
+ e);
+ }
+ }
+
+ private long handleTransferFilePieceResp(
+ final File file,
+ final Pair<IoTDBSyncClient, Boolean> clientAndStatus,
+ final RandomAccessFile reader,
+ final long nextPosition,
+ final PipeTransferFilePieceResp resp)
+ throws PipeException, IOException {
+ final TSStatus status = resp.getStatus();
+ if (status.getCode() ==
TSStatusCode.PIPE_TRANSFER_FILE_OFFSET_RESET.getStatusCode()) {
+ reader.seek(resp.getEndWritingOffset());
+ LOGGER.info(DataNodePipeMessages.REDIRECT_FILE_POSITION_TO,
resp.getEndWritingOffset());
+ return resp.getEndWritingOffset();
+ }
+
+ if (status.getCode() ==
TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode()) {
+ getClientManager().sendHandshakeReq(clientAndStatus);
+ }
+
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && status.getCode() !=
TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
+ receiverStatusHandler.handle(
+ resp.getStatus(),
+ String.format(
+
DataNodePipeMessages.MESSAGE_TRANSFER_FILE_ARG_ERROR_RESULT_STATUS_ARG_E565D9FD,
+ file,
+ resp.getStatus()),
+ file.getName());
+ }
+ return nextPosition;
+ }
+
+ private int getReadFileBufferSize(final File file) {
+ return (int)
+ Math.min(
+ PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
Math.max(file.length(), 1L));
+ }
+
@Override
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws
IOException {
if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
index 0582a94170d..d3249933aa1 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgentTest.java
@@ -23,19 +23,13 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
-import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
-import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.junit.Assert;
import org.junit.Test;
-import java.lang.reflect.Method;
import java.util.HashMap;
-import java.util.Map;
public class PipeDataNodeTaskAgentTest {
@@ -74,86 +68,4 @@ public class PipeDataNodeTaskAgentTest {
.setPipeTotalFloatingMemoryProportion(originalPipeTotalFloatingMemoryProportion);
}
}
-
- @Test
- public void testPlainBatchMemoryIncludesLeaderCacheEndpointShards() throws
Exception {
- final Map<String, String> sinkAttributes = new HashMap<>();
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.CONNECTOR_FORMAT_TABLET_VALUE);
- sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY,
"1024");
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY, "127.0.0.1:6667,
127.0.0.2:6667");
- sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_IP_KEY, "127.0.0.3");
- sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_PORT_KEY, "6667");
-
- Assert.assertEquals(
- 4 * 1024L, invokeCalculateSinkBatchMemory(new
PipeParameters(sinkAttributes)));
-
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_LEADER_CACHE_ENABLE_KEY,
Boolean.FALSE.toString());
- Assert.assertEquals(1024L, invokeCalculateSinkBatchMemory(new
PipeParameters(sinkAttributes)));
- }
-
- @Test
- public void testTsFileBatchMemoryIgnoresLeaderCacheEndpointShards() throws
Exception {
- final Map<String, String> sinkAttributes = new HashMap<>();
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.CONNECTOR_FORMAT_TS_FILE_VALUE);
- sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY,
"2048");
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_IOTDB_NODE_URLS_KEY,
"127.0.0.1:6667,127.0.0.2:6667");
-
- Assert.assertEquals(2048L, invokeCalculateSinkBatchMemory(new
PipeParameters(sinkAttributes)));
- }
-
- @Test
- public void testPlainBatchMemoryReturnsZeroWhenBatchModeIsDisabled() throws
Exception {
- final Map<String, String> sinkAttributes = new HashMap<>();
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.CONNECTOR_FORMAT_TABLET_VALUE);
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY,
Boolean.FALSE.toString());
- sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY,
"1024");
-
- Assert.assertEquals(0L, invokeCalculateSinkBatchMemory(new
PipeParameters(sinkAttributes)));
- }
-
- @Test
- public void testSendTsFileReadBufferMemoryUsesSinkReadFileBufferSize()
throws Exception {
- final Map<String, String> sourceAttributes = new HashMap<>();
- sourceAttributes.put(PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
Boolean.FALSE.toString());
-
- final Map<String, String> sinkAttributes = new HashMap<>();
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.CONNECTOR_FORMAT_TABLET_VALUE);
- Assert.assertEquals(
- 0L,
- invokeCalculateSendTsFileReadBufferMemory(
- new PipeParameters(sourceAttributes), new
PipeParameters(sinkAttributes)));
-
- sinkAttributes.put(
- PipeSinkConstant.CONNECTOR_FORMAT_KEY,
PipeSinkConstant.CONNECTOR_FORMAT_HYBRID_VALUE);
- Assert.assertEquals(
- PipeConfig.getInstance().getPipeSinkReadFileBufferSize(),
- invokeCalculateSendTsFileReadBufferMemory(
- new PipeParameters(sourceAttributes), new
PipeParameters(sinkAttributes)));
- }
-
- private long invokeCalculateSinkBatchMemory(final PipeParameters
sinkParameters)
- throws Exception {
- final Method method =
- PipeDataNodeTaskAgent.class.getDeclaredMethod(
- "calculateSinkBatchMemory", PipeParameters.class);
- method.setAccessible(true);
- return (long) method.invoke(null, sinkParameters);
- }
-
- private long invokeCalculateSendTsFileReadBufferMemory(
- final PipeParameters sourceParameters, final PipeParameters
sinkParameters) throws Exception {
- final Method method =
- PipeDataNodeTaskAgent.class.getDeclaredMethod(
- "calculateSendTsFileReadBufferMemory", PipeParameters.class,
PipeParameters.class);
- method.setAccessible(true);
- return (long) method.invoke(null, sourceParameters, sinkParameters);
- }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
index 91b6d7d56c5..3d81e143ac8 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
@@ -155,6 +156,87 @@ public class TsFileInsertionEventParserTest {
System.out.println(System.currentTimeMillis() - startTime);
}
+ @Test
+ public void testScanParserReleasesTabletMemoryAfterRawTabletGenerated()
throws Exception {
+ nonalignedTsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(
+ "nonaligned-release-tablet-memory.tsfile", 1, 1, 10, 0, 100, 10,
10);
+
+ try (final TsFileInsertionEventScanParser parser =
+ new TsFileInsertionEventScanParser(
+ nonalignedTsFile,
+ new PrefixTreePattern("root"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE,
+ null,
+ null,
+ false)) {
+ final Iterator<TabletInsertionEvent> iterator =
parser.toTabletInsertionEvents().iterator();
+
+ Assert.assertTrue(iterator.hasNext());
+ final TabletInsertionEvent event = iterator.next();
+ Assert.assertTrue(event instanceof PipeRawTabletInsertionEvent);
+ Assert.assertEquals(0,
getAllocatedTabletMemory(parser).getMemoryUsageInBytes());
+
+ ((PipeRawTabletInsertionEvent)
event).clearReferenceCount(getClass().getName());
+ }
+ }
+
+ @Test
+ public void
testConsumeTabletInsertionEventsWithRetryReleasesParserOnOutOfMemory()
+ throws Exception {
+ nonalignedTsFile =
+ TsFileGeneratorUtils.generateNonAlignedTsFile(
+ "nonaligned-consume-oom.tsfile", 1, 1, 10, 0, 100, 10, 10);
+ resource = new TsFileResource(nonalignedTsFile);
+ resource.setStatusForTest(TsFileResourceStatus.NORMAL);
+
+ // The TsFile generator only creates the file, so mark the resource
non-empty explicitly.
+ final IDeviceID deviceID =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.testsg.d0");
+ resource.updateStartTime(deviceID, 0);
+ resource.updateEndTime(deviceID, 9);
+
+ final PipeTsFileInsertionEvent event =
+ new PipeTsFileInsertionEvent(
+ false,
+ "root",
+ resource,
+ null,
+ false,
+ false,
+ false,
+ null,
+ null,
+ 0,
+ null,
+ new PrefixTreePattern("root"),
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ final AtomicReference<PipeRawTabletInsertionEvent> parsedEventReference =
+ new AtomicReference<>();
+
+ final PipeRuntimeOutOfMemoryCriticalException exception =
+ Assert.assertThrows(
+ PipeRuntimeOutOfMemoryCriticalException.class,
+ () ->
+ event.consumeTabletInsertionEventsWithRetry(
+ parsedEvent -> {
+ parsedEventReference.set(parsedEvent);
+ throw new
PipeRuntimeOutOfMemoryCriticalException("expected oom");
+ },
+ "test"));
+
+ Assert.assertEquals("expected oom", exception.getMessage());
+ Assert.assertNotNull(parsedEventReference.get());
+ Assert.assertTrue(parsedEventReference.get().isReleased());
+ Assert.assertNull(getEventParser(event).get());
+ }
+
@Test
public void
testScanParserSplitNonAlignedSinglePageChunkByEstimatedPageMemory() throws
Exception {
final long originalPipeMaxReaderChunkSize =
@@ -1995,6 +2077,22 @@ public class TsFileInsertionEventParserTest {
return (PipeMemoryBlock) field.get(parser);
}
+ private PipeMemoryBlock getAllocatedTabletMemory(final
TsFileInsertionEventParser parser)
+ throws NoSuchFieldException, IllegalAccessException {
+ final Field field =
+
TsFileInsertionEventParser.class.getDeclaredField("allocatedMemoryBlockForTablet");
+ field.setAccessible(true);
+ return (PipeMemoryBlock) field.get(parser);
+ }
+
+ @SuppressWarnings("unchecked")
+ private AtomicReference<TsFileInsertionEventParser> getEventParser(
+ final PipeTsFileInsertionEvent event) throws NoSuchFieldException,
IllegalAccessException {
+ final Field field =
PipeTsFileInsertionEvent.class.getDeclaredField("eventParser");
+ field.setAccessible(true);
+ return (AtomicReference<TsFileInsertionEventParser>) field.get(event);
+ }
+
private long calculatePipeMaxReaderChunkSizeForSinglePageAlignedChunk(final
File tsFile)
throws Exception {
try (final TsFileSequenceReader reader = new
TsFileSequenceReader(tsFile.getAbsolutePath())) {
