[feat](spill) Support multi-level partition spilling - #61212
Merged
Conversation
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Member
Author
|
run buildall |
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
TPC-H: Total hot run time: 28022 ms |
TPC-DS: Total hot run time: 153851 ms |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
yiguolei
previously approved these changes
Mar 12, 2026
Member
Author
|
/review |
mrhhsg
force-pushed
the
spill_repartition
branch
from
March 13, 2026 10:00
cfcd7ad to
bedb76c
Compare
Member
Author
|
run buildall |
Member
Author
|
/review |
TPC-H: Total hot run time: 27280 ms |
TPC-DS: Total hot run time: 168100 ms |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
|
/review |
![github-actions[bot]](https://cdn.statically.io/img/avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review: feat Support multi-level partition spilling
Reviewed ~80 files (9124 additions, 3594 deletions). The overall architecture is sound — the new SpillFile/SpillFileWriter/SpillFileReader abstraction is well-designed, the SpillRepartitioner cleanly handles both partitioner (join) and column-index (aggregation) modes, and the recursive repartition approach is a solid solution for large partitions that don't fit in memory.
Critical Checkpoints (Part 1.3)
Goal & correctness: Multi-level spill repartitioning for hash join and aggregation. The core goal is achieved. Test coverage is strong for join probe repartition (level transitions, max depth, row conservation) and good for aggregation source repartition. SpillFile infrastructure and SpillRepartitioner have comprehensive unit tests.
Concurrency: Single-writer pattern for SpillFile is correct — no concurrent mutation of _part_count/_total_written_bytes. Pipeline task execution is single-threaded per task. Dependency block/set_ready patterns are properly paired.
Lifecycle: SpillFileWriter uses weak_ptr back-reference to SpillFile (no circular reference). SpillFile gc() handles disk cleanup on destruction. No SIOF concerns.
Configuration: New spill_file_part_size_bytes config is mutable (good). New session variables properly forwarded via thrift. One default value issue found (see inline comment).
Incompatible changes: New thrift field IDs (203-208) are additive, backward-compatible.
Parallel code paths: Sort operators correctly adapted to new SpillFile API. Multi-cast data streamer changes look correct.
Observability: Profile counters are present in SpillFileWriter and SpillFileReader. Some old counters (SpillTotalTime, etc.) were removed but replaced by more granular per-file counters — acceptable.
Test coverage: 10 test files with ~150+ test cases. Notable gap: no end-to-end multi-level escalation test for aggregation (level 0→1→2). Hash join multi-level is tested.
Performance: One O(n²) vector erasure pattern found (minor, small n). No other hot-path concerns.
Issues Found: 3 bugs, 1 minor
See inline comments below.
| description = {"一旦触发 spill 后，aggregation sink 的 revocable memory 超过该阈值就主动落盘（字节）。默认 64MB。", | ||
| "After spill is triggered, aggregation sink will proactively spill when revocable memory " | ||
| + "exceeds this threshold (in bytes). Default is 64GB."}) | ||
| public long spillAggregationSinkMemLimitBytes = 64L * 1024L * 1024L * 1024L; |
Contributor
There was a problem hiding this comment.
Bug (Critical): Default value is 64L * 1024L * 1024L * 1024L = 64GB, not 64MB.
All sibling variables use 64MB:
spillJoinBuildSinkMemLimitBytes = 64L * 1024L * 1024L(64MB)spillSortSinkMemLimitBytes = 64L * 1024L * 1024L(64MB)spillSortMergeMemLimitBytes = 64L * 1024L * 1024L(64MB)
The thrift default is also 64MB (67108864). The Chinese description says "默认 64MB" but the English says "Default is 64GB".
The BE accessor clamps to [1MB, 4GB], so the effective value becomes 4GB — 62x higher than the intended 64MB. This effectively disables proactive spilling for aggregation sinks, causing excessive memory usage or OOM in aggregation spill scenarios.
Fix:
public long spillAggregationSinkMemLimitBytes = 64L * 1024L * 1024L;Also fix the English description to say "Default is 64MB."
| int spill_repartition_max_depth() const { | ||
| if (_query_options.__isset.spill_repartition_max_depth) { | ||
| // Clamp to a reasonable range: [1, 128] | ||
| return std::min(_query_options.spill_repartition_max_depth, 128); |
Contributor
There was a problem hiding this comment.
Bug (Major): Comment says "Clamp to a reasonable range: [1, 128]" but only the upper bound is enforced. If a user sets spill_repartition_max_depth = 0, the function returns 0.
With depth=0, the repartition check if (new_level >= _repartition_max_depth) (in both partitioned_hash_join_probe_operator.cpp:424 and partitioned_aggregation_source_operator.cpp:450) will ALWAYS fail because new_level (which is partition.level + 1 >= 1) is always >= 0. This causes spill repartitioning to immediately return InternalError("exceeded max depth"), making queries fail under memory pressure.
Fix:
return std::max(1, std::min(_query_options.spill_repartition_max_depth, 128));| // Estimate rows that will land in the hash table so we can reserve | ||
| // enough for JoinHashTable::first[] + JoinHashTable::next[]. | ||
| size_t rows = std::max(static_cast<size_t>(state->batch_size()), | ||
| static_cast<size_t>(local_state._recovered_build_block->rows())); |
Contributor
There was a problem hiding this comment.
Bug (Edge case): _recovered_build_block can be nullptr when about_to_build is true, causing a null dereference here.
Scenario:
- At line 749,
_recovered_build_blockis reset when a new partition is popped recover_build_blocks_from_partitionis called (line 763) and reads the build file- If ALL blocks from the file happen to be empty (line 359
continue), the function reaches EOS without ever creating_recovered_build_block(lines 366-371 never execute) build_fileis reset (line 379), function returns OK- Back in
_pull_from_spill_queue, the function returns OK to the pipeline task - In the NEXT pipeline iteration,
get_reserve_mem_sizeis called. At this point:is_valid()=true,build_finished=false→about_to_build=true, but_recovered_build_blockis null - Line 909:
_recovered_build_block->rows()→ null dereference
While an all-empty-blocks spill file is an edge case, the code should be defensively correct. The simplest fix:
if (about_to_build && local_state._recovered_build_block) {Or add a null check:
size_t rows = local_state._recovered_build_block
? std::max(static_cast<size_t>(state->batch_size()),
static_cast<size_t>(local_state._recovered_build_block->rows()))
: static_cast<size_t>(state->batch_size());| while (!local_state._blocks.empty()) { | ||
| auto blk = std::move(local_state._blocks.front()); | ||
| merged_rows += blk.rows(); | ||
| local_state._blocks.erase(local_state._blocks.begin()); |
Contributor
There was a problem hiding this comment.
Minor (Performance): _blocks.erase(_blocks.begin()) on a std::vector is O(n) per call (shifts all elements), making this loop O(n²) overall.
With typical block sizes and 8MB buffer limits, n is usually small (single digits), so the practical impact is minimal. Still, a cleaner pattern would be index-based iteration:
for (size_t i = 0; i < local_state._blocks.size(); ++i) {
auto blk = std::move(local_state._blocks[i]);
merged_rows += blk.rows();
status = _agg_source_operator->merge_with_serialized_key_helper(...);
RETURN_IF_ERROR(status);
}
local_state._blocks.clear();
jacktengg
reviewed
Mar 16, 2026
| // number of spill partitions configured for this operator | ||
| size_t _partition_count = 0; | ||
| // max repartition depth (configured from session variable in FE) | ||
| size_t _repartition_max_depth = SpillRepartitioner::MAX_DEPTH; |
Contributor
There was a problem hiding this comment.
Should be type int, as in be/src/exec/operator/partitioned_hash_join_probe_operator.h
jacktengg
reviewed
Mar 16, 2026
| !state->is_cancelled()) { | ||
| const auto& key = iter.template get_key<typename HashTableType::key_type>(); | ||
| auto partition_index = Base::_shared_state->get_partition_index(hash_table.hash(key)); | ||
| auto partition_index = hash_table.hash(key) % parent._partition_count; |
Contributor
There was a problem hiding this comment.
What if parent._partition_count == 0?
jacktengg
reviewed
Mar 16, 2026
| void SpillFileReader::seek(size_t block_index) { | ||
| auto st = _seek_to_block(block_index); | ||
| DCHECK(st.ok()) << "SpillFileReader::seek failed, block_index=" << block_index | ||
| << ", error=" << st.to_string(); |
Contributor
There was a problem hiding this comment.
status should not be ignored.
jacktengg
reviewed
Mar 16, 2026
| void update_profile(RuntimeProfile* child_profile); | ||
| /// Flush the current in-memory hash table by draining it as blocks and routing | ||
| /// each block through the repartitioner into the output sub-spill-files. | ||
| Status flush_hash_table_to_sub_spill_files(RuntimeState* state); |
Contributor
There was a problem hiding this comment.
Declared but no definition, delete it.
jacktengg
reviewed
Mar 16, 2026
| /// unread spill files from `remaining_spill_files`, and push resulting sub-partitions into | ||
| /// `_partition_queue`. After this call the hash table is reset and | ||
| /// `remaining_spill_files` is cleared. | ||
| Status flush_and_repartition(RuntimeState* state); |
Contributor
There was a problem hiding this comment.
Declared but no definition, delete it.
Member
Author
|
run buildall |
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 27246 ms |
Contributor
|
PR approved by at least one committer and no changes requested. |
Contributor
|
PR approved by anyone and no changes requested. |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
[c7b56dbe6a5][2026-03-05][Hu Shenggang] fix ut [b3a486d14b9][2026-03-05][Hu Shenggang] tiny mod [6411b66][2026-03-05][yiguolei ] refactor code and add unit test [23854d8][2026-03-05][yiguolei ] add force spill logic in join sink operator [a631dac][2026-03-04][yiguolei ] update dir and file meta realtime [a3fd36e][2026-03-04][yiguolei ] fix agg profile bug [c0380ad][2026-03-04][yiguolei ] change spill file to shared ptr [05670c5][2026-03-04][yiguolei ] fix probe revokeable memory size bug [5595d90][2026-03-03][yiguolei ] fix compile bug [36e3004][2026-03-03][yiguolei ] simplify code [9094b6b][2026-03-03][yiguolei ] simplify agg code [fdb355f][2026-03-03][yiguolei ] simplify probe code enhancement probe operator [82e94bb][2026-03-03][yiguolei ] refactor spill file interface [7548fe8][2026-03-02][Hu Shenggang] some tiny fix [2329442][2026-03-01][Hu Shenggang] fix agg revocable mem size [2c23a07][2026-02-28][Hu Shenggang] disbale distinct streaming agg when spill enabled [5c27350][2026-02-28][Hu Shenggang] Using spill_buffer_size_bytes as read limit when recovering data [35f2c55][2026-02-28][Hu Shenggang] [pipeline] Proactively pause query for spill under memory pressure in PipelineTask [4df2277][2026-02-28][Hu Shenggang] Make spill stream RAII [6a99170][2026-02-28][Hu Shenggang] Clear revoked data in agg & avoid null pointer in join [995aeec][2026-02-27][Hu Shenggang] Make repartitioner level-aware [87cbf87][2026-02-27][yiguolei ] fix compile [8a250cd][2026-02-27][yiguolei ] fix compile [a92b150][2026-02-27][yiguolei ] remove some codes [37df1fd][2026-02-27][yiguolei ] f [dd6d1f0][2026-02-27][yiguolei ] f [eb8a827][2026-02-27][yiguolei ] f [f73c0e9][2026-02-27][yiguolei ] f [2c989cb][2026-02-27][yiguolei ] repartitioner
mrhhsg
force-pushed
the
spill_repartition
branch
from
March 17, 2026 09:00
13c0243 to
3991576
Compare
Member
Author
|
run buildall |
github-actions
Bot
removed
the
approved
Contributor
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 27155 ms |
TPC-DS: Total hot run time: 170013 ms |
github-actions
Bot
added
the
approved
Contributor
|
PR approved by at least one committer and no changes requested. |
yiguolei
approved these changes
Mar 18, 2026
Contributor
|
skip check_coverage |
mrhhsg
added a commit
to mrhhsg/doris
that referenced
this pull request
Mar 24, 2026
This PR adds support for spill repartition in partitioned operators.
When memory is still insufficient after a partition has been spilled,
the spilled partition can be repartitioned into smaller sub-partitions,
which further reduces peak memory usage. The repartitioner is made
level-aware so repartition can be applied recursively when needed.
This PR also integrates the new spill repartition flow into partitioned
hash join and partitioned aggregation, adds force-spill logic in the
join sink operator, refactors spill file interfaces, improves spill
metadata maintenance and memory-pressure handling, and fixes several
spill-related issues such as revocable memory accounting and profile
updates.
Related PR: #xxx
Problem Summary:
None
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---------
Co-authored-by: yiguolei <guolei@selectdb.com>
29 tasks
yiguolei
added a commit
that referenced
this pull request
Mar 25, 2026
…61677) Pick #61212 This PR adds support for spill repartition in partitioned operators. When memory is still insufficient after a partition has been spilled, the spilled partition can be repartitioned into smaller sub-partitions, which further reduces peak memory usage. The repartitioner is made level-aware so repartition can be applied recursively when needed. This PR also integrates the new spill repartition flow into partitioned hash join and partitioned aggregation, adds force-spill logic in the join sink operator, refactors spill file interfaces, improves spill metadata maintenance and memory-pressure handling, and fixes several spill-related issues such as revocable memory accounting and profile updates. Related PR: #xxx Problem Summary: None - Test <!-- At least one of them must be included. --> - [ ] Regression test - [ ] Unit Test - [ ] Manual test (add detailed scripts or steps below) - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason <!-- Add your reason? --> - Behavior changed: - [ ] No. - [ ] Yes. <!-- Explain the behavior change --> - Does this need documentation? - [ ] No. - [ ] Yes. <!-- Add document PR link here. eg: apache/doris-website#1214 --> - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into --> --------- ### What problem does this PR solve? Issue Number: close #xxx Related PR: #xxx Problem Summary: ### Release note None ### Check List (For Author) - Test <!-- At least one of them must be included. --> - [ ] Regression test - [ ] Unit Test - [ ] Manual test (add detailed scripts or steps below) - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason <!-- Add your reason? --> - Behavior changed: - [ ] No. - [ ] Yes. <!-- Explain the behavior change --> - Does this need documentation? - [ ] No. - [ ] Yes. <!-- Add document PR link here. eg: apache/doris-website#1214 --> ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into --> Co-authored-by: yiguolei <guolei@selectdb.com>
76 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
7 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
This PR adds support for spill repartition in partitioned operators.
When memory is still insufficient after a partition has been spilled, the spilled partition can be repartitioned into smaller sub-partitions, which further reduces peak memory usage. The repartitioner is made level-aware so repartition can be applied recursively when needed.
This PR also integrates the new spill repartition flow into partitioned hash join and partitioned aggregation, adds force-spill logic in the join sink operator, refactors spill file interfaces, improves spill metadata maintenance and memory-pressure handling, and fixes several spill-related issues such as revocable memory accounting and profile updates.
Related PR: #xxx
Problem Summary:
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)