You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "yashmayya (via GitHub)" <gi...@apache.org> on 2026/04/21 20:51:15 UTC
[PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
yashmayya opened a new pull request, #18277:
URL: https://github.com/apache/pinot/pull/18277
## Summary
Under heap pressure, the per-query heap accountant's `killAllQueries()` flags every in-flight query via `QueryExecutionContext.terminate(...)`, but worker threads only notice the flag at their next `checkTermination*` poll. Operator subclasses of `MultiStageOperator` already poll once per block in `nextBlock()`, and hot operators (`HashJoinOperator`, `WindowAggregateOperator`, `AsofJoinOperator`, `NonEquiJoinOperator`) additionally poll per-record inside their inner loops. The MSE exchange and splitter paths do not — `HashExchange.route`, `BroadcastExchange.route`, `BlockExchange.sendBlock`'s splitter loop, and `BlockSplitter.Default.visit(RowHeapDataBlock, ...)` each process an entire block worth of rows / chunks / mailboxes with zero poll points inside.
On large blocks (joins with duplicate keys can emit up to `DEFAULT_MAX_ROWS_IN_JOIN = 1_048_576` rows; aggregates can emit all groups in one block at EOS) the accountant's kill signal stays invisible while the exchange keeps allocating via `ArrayList.grow` (`Arrays.copyOf`) and splitter serialization, extending kill-to-reclaim latency. Observed in production `hs_err` with a top frame of `Arrays.copyOf → HashExchange.route → MailboxSendOperator.getNextBlock`.
This PR adds `QueryThreadContext.checkTerminationAndSampleUsagePeriodically` calls inside:
- `HashExchange.route` row-partition loop
- `BroadcastExchange.route` mailbox loop
- `BlockExchange.sendBlock` splitter loop
- `BlockSplitter.Default.visit(RowHeapDataBlock, ...)` chunk loop
Uses the existing static helper (sample mask `0x1FFF` = every 8192 records), matching the pattern already used in `HashJoinOperator`, `AsofJoinOperator`, etc. No behavior change on the happy path — only shortens the window between the accountant flagging a query and the worker noticing.
## Test plan
- [x] New unit tests in `HashExchangeTest`, `BroadcastExchangeTest`, `BlockExchangeTest`, and `BlockSplitterTest` that flag the query for termination via `QueryExecutionContext.terminate(...)` and assert each of the four poll points throws `TerminationException` without dispatching any rows / chunks / mailbox sends.
- [x] Existing tests in the affected classes pass unchanged.
- [x] `./mvnw -pl pinot-query-runtime spotless:apply checkstyle:check license:check` clean.
## Out of scope
Tracked separately: making `killAllQueries` call `Thread.interrupt`, lowering `min.memory.footprint.to.kill.bytes`, and direct-memory tracking in the per-query accountant.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18277:
URL: https://github.com/apache/pinot/pull/18277#issuecomment-4292082669
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18277?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 55.51%. Comparing base ([`0607efe`](https://app.codecov.io/gh/apache/pinot/commit/0607efe710d40a29852417e55f16c49daf657957?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`48e9579`](https://app.codecov.io/gh/apache/pinot/commit/48e9579deae8b2df4330bbca23424be6f27512bf?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (0607efe) and HEAD (48e9579). Click for more details.
>
> <details><summary>HEAD has 36 uploads less than BASE</summary>
>
>| Flag | BASE (0607efe) | HEAD (48e9579) |
>|------|------|------|
>|java-21|5|1|
>|unittests1|2|1|
>|unittests|4|1|
>|temurin|10|1|
>|java-11|5|0|
>|unittests2|2|0|
>|integration|6|0|
>|integration2|2|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18277 +/- ##
============================================
- Coverage 63.51% 55.51% -8.00%
+ Complexity 1658 837 -821
============================================
Files 3244 2536 -708
Lines 197440 146070 -51370
Branches 30557 23515 -7042
============================================
- Hits 125407 81093 -44314
+ Misses 61989 57986 -4003
+ Partials 10044 6991 -3053
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.51% <100.00%> (-7.98%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.51% <100.00%> (-8.00%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.51% <100.00%> (-8.00%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.51% <100.00%> (-0.01%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18277/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18277?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>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18277:
URL: https://github.com/apache/pinot/pull/18277#discussion_r3127325225
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -42,6 +43,7 @@ public abstract class BlockExchange implements AutoCloseable {
// TODO: Deduct this value via grpc config maximum byte size; and make it configurable with override.
// TODO: Max block size is a soft limit. only counts fixedSize datatable byte buffer
private static final int MAX_MAILBOX_CONTENT_SIZE_BYTES = 4 * 1024 * 1024;
+ private static final String SEND_BLOCK_SCOPE = "BlockExchange.sendBlock";
Review Comment:
(minor) Same for other places
```suggestion
private static final String SEND_BLOCK_SCOPE = "BlockExchange";
```
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java:
##########
@@ -41,7 +43,9 @@ protected BroadcastExchange(List<SendingMailbox> sendingMailboxes, BlockSplitter
@Override
protected void route(List<SendingMailbox> destinations, MseBlock.Data block) {
+ int i = 0;
for (SendingMailbox mailbox : destinations) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i++, ROUTE_SCOPE);
Review Comment:
We shouldn't do the periodic check here.
Should we consider adding the check into the mailbox send instead?
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/BlockSplitter.java:
##########
@@ -53,6 +54,7 @@ public interface BlockSplitter {
*/
class Default implements BlockSplitter, MseBlock.Data.Visitor<Iterator<MseBlock.Data>, Integer> {
private static final int MEDIAN_COLUMN_SIZE_BYTES = 8;
+ private static final String SPLIT_SCOPE = "BlockSplitter.Default.split";
Review Comment:
(minor) Current convention is to use the class name unless it involves multiple methods. It is inlined in most existing usage
```suggestion
private static final String SPLIT_SCOPE = "BlockSplitter.Default";
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #18277:
URL: https://github.com/apache/pinot/pull/18277#issuecomment-4303044655
@yashmayya Sorry I somehow edited your response instead of putting my own response :-(
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #18277:
URL: https://github.com/apache/pinot/pull/18277#issuecomment-4291764608
## Performance review
These changes are all on MSE query hot paths (every row of every block through every exchange), so I did a dedicated performance pass before finalizing. Summary:
### Fast-path cost (mask miss, 8191 of 8192 rows)
The static `QueryThreadContext.checkTerminationAndSampleUsagePeriodically(int, String)` does an `int & 0x1FFF` and branches on zero before doing any real work. The mask constant is an inlineable `public static final` and the method has no varargs or megamorphic dispatch. Per-iteration cost on the fast path is ~3 cycles (load counter, AND, branch-not-taken), which folds entirely into the row-routing loop that already does a hash-mod and a map put — relative overhead is sub-1%.
### Slow-path cost (mask hit, 1 of 8192 rows)
`ThreadLocal.get()`, a volatile read on `_executionContext.getTerminateException()`, `Thread.interrupted()`, `System.currentTimeMillis()`, and `_accountant.sampleUsage()`. This is exactly the cost every existing hot operator (`HashJoinOperator`, `AsofJoinOperator`, `NonEquiJoinOperator`, `WindowAggregateOperator`) already pays at the same 1/8192 rate inside their per-record loops, so it doesn't shift the perf envelope.
### Safety tweak applied
Switched `HashExchange.route` from indexed `rows.get(r)` back to enhanced-for with a counter variable. `RowHeapDataBlock.getRows()` returns `List<Object[]>` with no documented `RandomAccess` guarantee — in practice every producer passes an `ArrayList`, but the enhanced-for variant is fragility-free and matches the existing pattern in `HashJoinOperator`.
### Deliberately kept
- **`BlockExchange.sendBlock` poll + `BlockSplitter.Default.visit` poll**: these are complementary, not redundant. `BlockSplitter.Default.visit` is eager — the poll there fires during chunk list construction (before any serialization). The `sendBlock` poll fires between `sendingMailbox.send(split)` calls, which include serialization. They catch termination at different phases of the send pipeline.
- **`BroadcastExchange.route` per-mailbox poll**: broadcast fan-out is typically small, so the mask-miss is the common case and the slow path is essentially never hit from the counter alone. Free on fast path, keeps the loop cooperative if fan-out ever grows.
### Bottom line
No measurable regression expected. The pattern is consistent with how every other hot operator in the engine already polls for cooperative termination.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya merged PR #18277:
URL: https://github.com/apache/pinot/pull/18277
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18277:
URL: https://github.com/apache/pinot/pull/18277#discussion_r3123801787
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -162,7 +164,9 @@ protected void sendBlock(SendingMailbox sendingMailbox, MseBlock.Data block) {
sendingMailbox.send(block);
} else {
Iterator<? extends MseBlock.Data> splits = _splitter.split(block, MAX_MAILBOX_CONTENT_SIZE_BYTES);
+ int i = 0;
while (splits.hasNext()) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i++, SEND_BLOCK_SCOPE);
Review Comment:
`checkTerminationAndSampleUsagePeriodically` only checks when `(i & 0x1fff) == 0`, but here `i` counts 4MB splits rather than rows. If the heap accountant terminates the query just after the first split, this loop can still send up to 8191 more chunks before observing the kill signal, leaving a large kill-to-reclaim window in the OOM path this PR is trying to close. Please check termination for each split, or drive the periodic helper with a row/byte counter instead of the split index.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #18277:
URL: https://github.com/apache/pinot/pull/18277#issuecomment-4316157974
Whoops, forgot to change the PR title / squash commit message, we removed the check from the splitter loop
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #18277:
URL: https://github.com/apache/pinot/pull/18277#discussion_r3125433596
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java:
##########
@@ -162,7 +164,9 @@ protected void sendBlock(SendingMailbox sendingMailbox, MseBlock.Data block) {
sendingMailbox.send(block);
} else {
Iterator<? extends MseBlock.Data> splits = _splitter.split(block, MAX_MAILBOX_CONTENT_SIZE_BYTES);
+ int i = 0;
while (splits.hasNext()) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i++, SEND_BLOCK_SCOPE);
Review Comment:
Good point, addressed
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add cooperative termination polls inside MSE exchange and splitter loops [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #18277:
URL: https://github.com/apache/pinot/pull/18277#issuecomment-4300747433
> Instead of Exchange, should we consider adding the resource check into SendingMailbox.send()?
I think that makes sense for exchanges other than hash exchange (which does the per row routing).
> Does SendingMailbox honor interruptions? Is it converted into an IOException in gRPC?
No, I don't think so, the async stream observer uses its own cancellation mechanism via context / deadline.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org