one collector per agg request instead per bucket #2759
Conversation
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
PSeitz-dd
force-pushed
the
bucket_id_agg
branch
2 times, most recently
from
a16d0ff to
d97bda9
Compare
PSeitz
force-pushed
the
bucket_id_agg
branch
4 times, most recently
from
59e9f5a to
856e5d5
Compare
In this refactoring a collector knows in which bucket of the parent their data is in. This allows to convert the previous approach of one collector per bucket to one collector per request. low card bucket optimization
use paged term map in term agg use special no sub agg term map impl
PSeitz
force-pushed
the
bucket_id_agg
branch
2 times, most recently
from
3356fc8 to
d76b315
Compare
increase cache to 2048
PSeitz
force-pushed
the
bucket_id_agg
branch
2 times, most recently
from
1f717da to
29add85
Compare
PSeitz-dd
force-pushed
the
bucket_id_agg
branch
from
29add85 to
411587a
Compare
PSeitz-dd
force-pushed
the
bucket_id_agg
branch
from
411587a to
1591344
Compare
fulmicoton
reviewed
Dec 11, 2025
Comment on lines
25
to
34
| /// Only used when LOWCARD is true. | ||
| /// Cache doc ids per bucket for sub-aggregations. | ||
| /// | ||
| /// The outer Vec is indexed by BucketId. | ||
| per_bucket_docs: Vec<Vec<DocId>>, | ||
| /// Only used when LOWCARD is false. | ||
| /// For higher cardinalities we use a partitioned approach to store | ||
| /// | ||
| /// partitioned Vec<(BucketId, DocId)> pairs to improve grouping locality. | ||
| partitions: [PartitionEntry; NUM_PARTITIONS], |
Collaborator
There was a problem hiding this comment.
!? why use a boolean for this. I don't understand?
Contributor
Author
There was a problem hiding this comment.
What boolean? Do you mean array?
It's done as a cheap inexact group_by on bucket_id
Collaborator
There was a problem hiding this comment.
no I meant LOWCARD.
It seems strange to have both partitions and per_bucket_docs, and switch usage basedon per_bucket_docs.
(Also why not Box<[Vec; NUM_PARTITIONS]>?)
PSeitz-dd
force-pushed
the
bucket_id_agg
branch
from
8a4286c to
71dc084
Compare
fulmicoton
reviewed
Jan 2, 2026
| // Note: We need to make sure that we don't lock ourselves into a situation where we hit | ||
| // the FLUSH_THRESHOLD, but never flush any buckets. (except the final flush) | ||
| let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); | ||
| const _: () = { |
Collaborator
There was a problem hiding this comment.
that's to make sure this is checked at compile time? That's fun.
fulmicoton
reviewed
Jan 2, 2026
| // The threshold above which we flush buckets individually. | ||
| // Note: We need to make sure that we don't lock ourselves into a situation where we hit | ||
| // the FLUSH_THRESHOLD, but never flush any buckets. (except the final flush) | ||
| let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); |
Collaborator
There was a problem hiding this comment.
Suggested change
| let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); | |
| let mut bucket_threshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); |
fulmicoton
reviewed
Jan 2, 2026
| let mut bucket_treshold = FLUSH_THRESHOLD / (self.per_bucket_docs.len().max(1) * 2); | ||
| const _: () = { | ||
| // MAX_NUM_TERMS_FOR_VEC == LOWCARD threshold | ||
| let bucket_treshold = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2); |
Collaborator
There was a problem hiding this comment.
Suggested change
| let bucket_treshold = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2); | |
| let bucket_threshold = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2); |
Collaborator
There was a problem hiding this comment.
Can you pick another name here?
fulmicoton
reviewed
Jan 2, 2026
| // MAX_NUM_TERMS_FOR_VEC == LOWCARD threshold | ||
| let bucket_treshold = FLUSH_THRESHOLD / (MAX_NUM_TERMS_FOR_VEC as usize * 2); | ||
| assert!( | ||
| bucket_treshold > 0, |
Collaborator
There was a problem hiding this comment.
Suggested change
| bucket_treshold > 0, | |
| bucket_threshold > 0, |
fulmicoton
reviewed
Jan 2, 2026
Comment on lines
+605
to
+606
| for doc_id in docs { | ||
| let doc_id = *doc_id; |
Collaborator
There was a problem hiding this comment.
Suggested change
| for doc_id in docs { | |
| let doc_id = *doc_id; | |
| for &doc_id in docs { |
fulmicoton
reviewed
Jan 2, 2026
| // TODO: this is terrible, a new vec is allocated for every doc | ||
| // We can fetch blocks instead | ||
| // We don't need to store the order for every value | ||
| let sorts: Vec<DocValueAndOrder> = req |
Collaborator
There was a problem hiding this comment.
can we just have a Vec buffer on the Self?
Collaborator
There was a problem hiding this comment.
no, currently the vec is stored in the TopNComputer
Collaborator
|
Do you know if the high card case is overall a regression or not? |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
In this refactoring a collector knows in which bucket of the parent their data is in. This allows to convert the previous approach of one collector per bucket to one collector per request.
Add
PagedTermMapas anotherTermAggregationMapto reduce memory usage compared to aHashMapIt contains an optimization for low cardinality bucket id
Remove Clone on the collector (we only have one instance now)
Future Work
PerRequestAggSegCtx, we can store now everything in the collectorPerformance
The heavy hitters are drastically reduced in terms of memory and CPU.
For term aggs with many terms, we use a lot less memory.
We use some more buffers to pass docs, which increases memory consumption for some aggs.
Biggest regression is
terms_zipf_1000_with_avg_sub_agg Avg: 9.1190ms (+39.79%)Which should be fixed when we fetch all values for all buckets at once.