feat: Add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis#19627
feat: Add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis#19627aho135 wants to merge 3 commits into
Conversation
…nosis The merge buffer is sliced into druid.processing.numThreads slices by ConcurrentGrouper, and a groupBy query spills as soon as its single fullest slice fills (~ sizeBytes/numThreads). Existing metrics could not explain this: mergeBuffer/maxBytesUsed is a per-query SUM across slices, discounted by the hash-table load factor, so it never approaches sizeBytes even while queries spill, making it impossible to compare against druid.processing.buffer.sizeBytes. Add mergeBuffer/maxSpillProximity, a dimensionless gauge in [0.0, 1.0]: the fullest single slice's used bytes divided by its spill threshold (sliceSize * maxLoadFactor), tracked as a max across slices and across queries. 1.0 means a query reached the point at which a slice spills to disk. - GroupByStatsProvider: track per-slice max used bytes and spill threshold; expose getSpillProximity() (clamped to [0,1]); aggregate as a max. - SpillingGrouper: report each slice's peak usage against its threshold. - BufferHashGrouper: expose resolveMaxLoadFactor() so the denominator matches the grouper's actual spill decision. - GroupByStatsMonitor: emit mergeBuffer/maxSpillProximity. - Clarify mergeBuffer/bytesUsed and maxBytesUsed docs (slicing semantics). Existing emitted metric names and values are unchanged.
aho135
changed the title
|
Hey @GWphua! We were using the |
FrankChen021
left a comment
FrankChen021
left a comment
There was a problem hiding this comment.
| Severity | Findings |
|---|---|
| P0 | 0 |
| P1 | 0 |
| P2 | 1 |
| P3 | 0 |
| Total | 1 |
Reviewed 7 of 7 changed files.
This is an automated review by Codex GPT-5.5
| { | ||
| maxMergeBufferUsedBytes.addAndGet(bytes); | ||
| maxSliceUsedBytes.accumulateAndGet(usedBytes, Math::max); | ||
| sliceSpillThresholdBytes.accumulateAndGet(spillThresholdBytes, Math::max); |
There was a problem hiding this comment.
[P2] Track spill proximity as a per-slice ratio
This stores the maximum used bytes and maximum threshold independently, which breaks when one query reports groupers with different thresholds. Nested/subtotal processing can pass the same PerQueryStats through a sliced ConcurrentGrouper and later full-buffer SpillingGroupers; if a small slice reaches its spill threshold, the larger full-buffer threshold can be retained here and getSpillProximity() will divide by that larger value, under-reporting the slice spill as roughly 1 / concurrencyHint. Please track the maximum usedBytes / spillThresholdBytes per sliceUsage call, or otherwise keep the used/threshold pair together, so mixed grouper sizes still report the true max proximity.
There was a problem hiding this comment.
Good catch — fixed in fc770ee. I replaced the two independent maxima (maxSliceUsedBytes and sliceSpillThresholdBytes) with a single DoubleAccumulator that maxes the per-slice ratio: sliceUsage() now computes min(1.0, usedBytes / spillThresholdBytes) at the point of measurement, so each slice's used bytes stay paired with its own threshold.
This closes exactly the case you described: when one PerQueryStats flows through both small ConcurrentGrouper slices and a full-buffer SpillingGrouper (subtotal/nested processing), a saturated small slice now reports the true 1.0 instead of being diluted to ~1/concurrencyHint by the larger threshold winning the separate max. Added testSpillProximityKeepsPerSliceRatioWhenThresholdsDiffer covering the mixed-threshold scenario.
Track spill proximity as a single per-slice ratio max rather than maxing used bytes and the spill threshold independently. When one query mixes groupers of different sizes (small ConcurrentGrouper slices alongside a full-buffer SpillingGrouper for subtotal/nested processing), the separate maxima could pair a saturated small slice's usage with the larger full-buffer threshold, under-reporting the spill as ~1/concurrencyHint. Computing min(1.0, used/threshold) per slice keeps each numerator paired with its own denominator.
FrankChen021
left a comment
FrankChen021
left a comment
There was a problem hiding this comment.
I reviewed the follow-up to the prior per-slice spill-proximity comment. The fc770ee update now computes and maxes the per-slice ratio at measurement time, with a mixed-threshold regression test covering the original issue; no further inline reply looks useful.
Reviewed 7 of 7 changed files.
This is an automated review by Codex GPT-5.5
FrankChen021
left a comment
FrankChen021
left a comment
There was a problem hiding this comment.
I reviewed the follow-up to the prior per-slice spill-proximity comment. The current fc770ee code now computes and maxes the per-slice ratio at measurement time, and the mixed-threshold regression test covers the original issue; no further inline reply looks useful.
Reviewed 7 of 7 changed files.
This is an automated review by Codex GPT-5.5
The byte-ratio formulation (peakUsedBytes / (mergeBufferSize * loadFactor)) could never reach 1.0 at the real spill point: the numerator counts only the hash-table arena while the denominator scales the whole buffer, and the table-doubling walk rarely fills the arena, so proximity topped out at a query-shape-dependent value well below 1.0 — defeating the metric's purpose. Rebase the metric on the bucket-count condition the grouper actually uses to spill. BufferHashGrouper spills when its hash table cannot allocate a new bucket, i.e. size == regrowthThreshold on a table that can no longer grow. ByteBufferHashTable now tracks a lifetime-max size/regrowthThreshold ratio, pinned to exactly 1.0 when a bucket allocation is rejected, and preserved across reset() so a slice that already spilled still reports 1.0 after its size returns to 0. To make 1.0 mean 'at or past the spill point' without false positives: - The running ratio is recorded strictly while size < regrowthThreshold, so the transient size == regrowthThreshold at intermediate growth boundaries (where the table then grows) is not mistaken for a spill. - A new overridable isTerminalTableLevel() (base: tableStart == 0) pins 1.0 when the table is parked at its threshold and cannot grow. AlternatingByte BufferHashTable overrides it to false, because a full limit-push-down table performs a heap-trim swap rather than spilling to disk. The value is now bucket-count based, so it is independent of bucket width, offset-list overhead, and integer truncation. sliceUsage() takes the ratio directly (clamped to [0,1], NaN ignored), keeping the per-slice max that lets a query mixing differently-sized groupers still report the true proximity. Docs updated to state 1.0 corresponds exactly to the spill trigger.
FrankChen021
left a comment
FrankChen021
left a comment
There was a problem hiding this comment.
I reviewed the latest update for correctness, edge cases, concurrency, lifecycle, and monitor integration risks; no new issues found.
Reviewed 12 of 12 changed files.
This is an automated review by Codex GPT-5.5
2 participants
Description
When a groupBy query runs,
ConcurrentGrouperdivides the single acquired merge buffer intodruid.processing.numThreadsequal slices (sliceSize = capacity / numThreads) and gives one slice to each processing thread. A query spills to disk as soon as its fullest single slice fills — at roughlysizeBytes / numThreads, which can be far below the configureddruid.processing.buffer.sizeBytes.The existing metrics do not let an operator see this.
mergeBuffer/maxBytesUsedis a per-query sum across slices, further discounted by the hash-table load factor, so it never approachessizeByteseven while queries are actively spilling — making it impossible to compare againstdruid.processing.buffer.sizeBytesor to reason about spill pressure.Concretely, an operator with
sizeBytes = 125 MiBandnumThreads = 240(slices ≈ 546 KiB) sawgroupBy/spilledQueriesclimbing whilemergeBuffer/maxBytesUsedsat around ~60 MB, which looks contradictory until you account for slicing.Change
This PR adds
mergeBuffer/maxSpillProximity, a dimensionless gauge in[0.0, 1.0]:sliceSize × maxLoadFactor(default load factor0.7), because aBufferHashGrouperspills when its bucket count reaches the load factor, not when the slice is byte-full. This makes1.0correspond to the real spill point.Operators can read
mergeBuffer/maxSpillProximityalongsidegroupBy/spilledQueries: a value near1.0means slices are saturating, and the fix is to widen each slice by raisingdruid.processing.buffer.sizeBytesor loweringdruid.processing.numThreads.Changed files
GroupByStatsProvider— track per-slice max used bytes and the per-slice spill threshold; addgetSpillProximity()(clamped to[0,1]); aggregate as a max across queries.SpillingGrouper— report each slice's peak usage against its spill threshold inclose().BufferHashGrouper— exposeresolveMaxLoadFactor()so the metric denominator matches the grouper's actual spill decision (including the default-resolution rule).GroupByStatsMonitor— emitmergeBuffer/maxSpillProximity.docs/operations/metrics.md— document the new metric and clarify the slicing semantics ofmergeBuffer/bytesUsedandmergeBuffer/maxBytesUsed.This PR has:
mergeBuffer/maxSpillProximity; no behavior or config changes.)