You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Caideyipi (via GitHub)" <gi...@apache.org> on 2026/06/29 02:46:33 UTC
[PR] Pipe: improve incremental memory estimation [iotdb]
Caideyipi opened a new pull request, #18049:
URL: https://github.com/apache/iotdb/pull/18049
## Description
This PR improves the pipe creation memory admission check from a coarse per-pipe estimate to an incremental estimate based on the tasks and sink subtasks that this DataNode will actually create.
Changes include:
- Estimate memory by local region tasks, and also check newly added single-region tasks.
- Account for sink subtask reuse so shared sink pending queues/subtasks are not charged repeatedly.
- Reuse the actual sink subtask key and parallelism logic from PipeSinkSubtaskManager.
- Match task builder parameter blending/preprocessing during admission estimation.
- Estimate batch memory from connector.batch.size-bytes/sink.batch.size-bytes, including leader-cache endpoint sharding.
- Estimate TsFile send read buffer with pipeSinkReadFileBufferSize.
- Skip admission checks for idempotent create no-ops, while checking before dropping old tasks during recreate.
## Verification
- mvn spotless:apply -pl iotdb-core/datanode,iotdb-core/node-commons
- git -c core.preloadIndex=false diff --check -- iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
- mvn "-Ddevelocity.off=true" "-Dcheckstyle.skip=true" "-Dspotless.check.skip=true" "-DskipTests" compile -pl iotdb-core/node-commons
Also ran datanode compile:
- mvn "-Ddevelocity.off=true" "-Dcheckstyle.skip=true" "-Dspotless.check.skip=true" "-DskipTests" compile -pl iotdb-core/datanode
- It reached javac but failed on existing generated-source/dependency issues such as missing IFill, Accumulator, and org.apache.commons.collections4.comparators, not in the modified pipe files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve incremental memory estimation [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #18049:
URL: https://github.com/apache/iotdb/pull/18049#issuecomment-4828543824
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=18049) **Quality Gate passed**
Issues
 [6 New issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=18049&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=18049&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=18049&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=18049&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=18049&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=18049)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve incremental memory estimation [iotdb]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #18049:
URL: https://github.com/apache/iotdb/pull/18049#issuecomment-4828733558
## [Codecov](https://app.codecov.io/gh/apache/iotdb/pull/18049?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `187 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 41.39%. Comparing base ([`7f41f06`](https://app.codecov.io/gh/apache/iotdb/commit/7f41f064858f9376d7c6f7e443dc541918d20ac8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`00962e6`](https://app.codecov.io/gh/apache/iotdb/commit/00962e625bd48503ee66453f396a9e8e8e93d7c9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/iotdb/pull/18049?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...otdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fpipe%2Fagent%2Ftask%2FPipeDataNodeTaskAgent.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcGlwZS9hZ2VudC90YXNrL1BpcGVEYXRhTm9kZVRhc2tBZ2VudC5qYXZh) | 0.00% | [139 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...gent/task/subtask/sink/PipeSinkSubtaskManager.java](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fpipe%2Fagent%2Ftask%2Fsubtask%2Fsink%2FPipeSinkSubtaskManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcGlwZS9hZ2VudC90YXNrL3N1YnRhc2svc2luay9QaXBlU2lua1N1YnRhc2tNYW5hZ2VyLmphdmE=) | 0.00% | [28 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/iotdb/commons/pipe/agent/task/PipeTaskAgent.java](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&filepath=iotdb-core%2Fnode-commons%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fcommons%2Fpipe%2Fagent%2Ftask%2FPipeTaskAgent.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9ub2RlLWNvbW1vbnMvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2lvdGRiL2NvbW1vbnMvcGlwZS9hZ2VudC90YXNrL1BpcGVUYXNrQWdlbnQuamF2YQ==) | 0.00% | [11 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pe/agent/task/builder/PipeDataNodeTaskBuilder.java](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fpipe%2Fagent%2Ftask%2Fbuilder%2FPipeDataNodeTaskBuilder.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvcGlwZS9hZ2VudC90YXNrL2J1aWxkZXIvUGlwZURhdGFOb2RlVGFza0J1aWxkZXIuamF2YQ==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/18049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18049 +/- ##
============================================
- Coverage 41.42% 41.39% -0.04%
Complexity 318 318
============================================
Files 5282 5286 +4
Lines 369246 369671 +425
Branches 47786 47847 +61
============================================
+ Hits 152978 153030 +52
- Misses 216268 216641 +373
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/iotdb/pull/18049?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve incremental memory estimation [iotdb]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #18049:
URL: https://github.com/apache/iotdb/pull/18049#issuecomment-4828941964
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=18049) **Quality Gate passed**
Issues
 [6 New issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=18049&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_iotdb&pullRequest=18049&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_iotdb&pullRequest=18049&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=18049&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=18049&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=18049)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve incremental memory estimation [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #18049:
URL: https://github.com/apache/iotdb/pull/18049
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Pipe: improve incremental memory estimation [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #18049:
URL: https://github.com/apache/iotdb/pull/18049#discussion_r3503002601
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -850,17 +990,17 @@ private void calculateInsertNodeQueueMemory(final PipeParameters sourceParameter
return;
}
+ final long needFloatingMemory =
+ PipeConfig.getInstance().getPipeInsertNodeQueueMemory() * dataRegionTaskCount;
final long allocatedMemorySizeInBytes = this.getAllFloatingMemoryUsageInByte();
final long remainingMemory =
PipeDataNodeResourceManager.memory().getTotalFloatingMemorySizeInBytes()
- allocatedMemorySizeInBytes;
- if (remainingMemory < PipeConfig.getInstance().getPipeInsertNodeQueueMemory()) {
+ if (remainingMemory < needFloatingMemory) {
final String message =
String.format(
"%s Need Floating memory: %d bytes, free Floating memory: %d bytes",
- PipeMessages.NOT_ENOUGH_MEMORY_FOR_PIPE,
- PipeConfig.getInstance().getPipeInsertNodeQueueMemory(),
- remainingMemory);
+ PipeMessages.NOT_ENOUGH_MEMORY_FOR_PIPE, needFloatingMemory, remainingMemory);
Review Comment:
Done. I moved the memory check details into `PipeMessages` with both English and Chinese i18n entries, and updated the code to use those message constants.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org