Skip to content

feat: Add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis #19627

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aho135 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

feat: Add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis #19627

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0c52f53
feat: add mergeBuffer/maxSpillProximity metric for groupBy spill diag…
aho135 Jun 24, 2026
fc770ee
fix: pair per-slice usage with its own threshold for maxSpillProximity
aho135 Jul 6, 2026
52ace82
fix: base maxSpillProximity on the hash table's actual spill trigger
aho135 Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions docs/operations/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,8 +93,9 @@ Most metric values reset each emission period, as specified in `druid.monitoring
|`mergeBuffer/queries`|Number of groupBy queries that acquired a batch of buffers from the merge buffer pool.|This metric is only available if the `GroupByStatsMonitor` module is included.|Depends on the number of groupBy queries needing merge buffers.|
|`mergeBuffer/acquisitionTimeNs`|Total time in nanoseconds to acquire merge buffer for groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxAcquisitionTimeNs`|Maximum time in nanoseconds to acquire merge buffer for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/bytesUsed`|Number of bytes used by merge buffers to process groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum number of bytes used by merge buffers for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/bytesUsed`|Total merge buffer bytes used to process groupBy queries, summed across queries. Each query's value is itself the sum across the slices it held: a merge buffer is divided among `druid.processing.numThreads` concurrent query slices, so a query can spill once a single slice fills even though this summed usage looks well below `druid.processing.buffer.sizeBytes`. To gauge spill pressure, use `mergeBuffer/maxSpillProximity` rather than comparing this to `sizeBytes`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single groupBy query within the emission period, where each query's usage is the sum across the slices it held. Because the buffer is sliced among `druid.processing.numThreads` query slices, this value is not directly comparable to `druid.processing.buffer.sizeBytes`; use `mergeBuffer/maxSpillProximity` to gauge spill pressure.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to spilling within the emission period, as a fraction in [0.0, 1.0]. A merge buffer is divided into `druid.processing.numThreads` slices and a query spills as soon as its fullest single slice's hash table cannot allocate another bucket; this metric is that fullest slice's peak fill ratio (bucket count over the load-factor bucket limit), taken as a max across slices and across queries. **1.0 corresponds exactly to the spill trigger** — the value is bucket-count based, so it is not distorted by bucket width, offset-list overhead, or integer truncation. If you see 1.0, raise `druid.processing.buffer.sizeBytes` or lower `druid.processing.numThreads` to widen each slice. Read alongside `groupBy/spilledQueries`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/spilledQueries`|Number of groupBy queries that have spilled onto the disk.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/spilledBytes`|Number of bytes spilled on the disk by the groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/maxSpilledBytes`|Maximum number of bytes spilled to disk by any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
Expand Down Expand Up @@ -122,8 +123,9 @@ Most metric values reset each emission period, as specified in `druid.monitoring
|`mergeBuffer/queries`|Number of groupBy queries that acquired a batch of buffers from the merge buffer pool.|This metric is only available if the `GroupByStatsMonitor` module is included.|Depends on the number of groupBy queries needing merge buffers.|
|`mergeBuffer/acquisitionTimeNs`|Total time in nanoseconds to acquire merge buffer for groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxAcquisitionTimeNs`|Maximum time in nanoseconds to acquire merge buffer for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/bytesUsed`|Number of bytes used by merge buffers to process groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum number of bytes used by merge buffers for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/bytesUsed`|Total merge buffer bytes used to process groupBy queries, summed across queries. Each query's value is itself the sum across the slices it held: a merge buffer is divided among `druid.processing.numThreads` concurrent query slices, so a query can spill once a single slice fills even though this summed usage looks well below `druid.processing.buffer.sizeBytes`. To gauge spill pressure, use `mergeBuffer/maxSpillProximity` rather than comparing this to `sizeBytes`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single groupBy query within the emission period, where each query's usage is the sum across the slices it held. Because the buffer is sliced among `druid.processing.numThreads` query slices, this value is not directly comparable to `druid.processing.buffer.sizeBytes`; use `mergeBuffer/maxSpillProximity` to gauge spill pressure.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to spilling within the emission period, as a fraction in [0.0, 1.0]. A merge buffer is divided into `druid.processing.numThreads` slices and a query spills as soon as its fullest single slice's hash table cannot allocate another bucket; this metric is that fullest slice's peak fill ratio (bucket count over the load-factor bucket limit), taken as a max across slices and across queries. **1.0 corresponds exactly to the spill trigger** — the value is bucket-count based, so it is not distorted by bucket width, offset-list overhead, or integer truncation. If you see 1.0, raise `druid.processing.buffer.sizeBytes` or lower `druid.processing.numThreads` to widen each slice. Read alongside `groupBy/spilledQueries`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/spilledQueries`|Number of groupBy queries that have spilled onto the disk.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/spilledBytes`|Number of bytes spilled on the disk by the groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
|`groupBy/maxSpilledBytes`|Maximum number of bytes spilled to disk by any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies|
Expand Down Expand Up @@ -154,8 +156,9 @@ to represent the task ID are deprecated and will be removed in a future release.
|`mergeBuffer/queries`|Number of groupBy queries that acquired a batch of buffers from the merge buffer pool. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Depends on the number of groupBy queries needing merge buffers.|
|`mergeBuffer/acquisitionTimeNs`|Total time in nanoseconds to acquire merge buffer for groupBy queries. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/maxAcquisitionTimeNs`|Maximum time in nanoseconds to acquire merge buffer for any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/bytesUsed`|Number of bytes used by merge buffers to process groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum number of bytes used by merge buffers for any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/bytesUsed`|Total merge buffer bytes used to process groupBy queries, summed across queries. Each query's value is itself the sum across the slices it held: a merge buffer is divided among `druid.processing.numThreads` concurrent query slices, so a query can spill once a single slice fills even though this summed usage looks well below `druid.processing.buffer.sizeBytes`. To gauge spill pressure, use `mergeBuffer/maxSpillProximity` rather than comparing this to `sizeBytes`. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single groupBy query within the emission period, where each query's usage is the sum across the slices it held. Because the buffer is sliced among `druid.processing.numThreads` query slices, this value is not directly comparable to `druid.processing.buffer.sizeBytes`; use `mergeBuffer/maxSpillProximity` to gauge spill pressure. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to spilling within the emission period, as a fraction in [0.0, 1.0]. A merge buffer is divided into `druid.processing.numThreads` slices and a query spills as soon as its fullest single slice's hash table cannot allocate another bucket; this metric is that fullest slice's peak fill ratio (bucket count over the load-factor bucket limit), taken as a max across slices and across queries. **1.0 corresponds exactly to the spill trigger** — the value is bucket-count based, so it is not distorted by bucket width, offset-list overhead, or integer truncation. If you see 1.0, raise `druid.processing.buffer.sizeBytes` or lower `druid.processing.numThreads` to widen each slice. Read alongside `groupBy/spilledQueries`. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`groupBy/spilledQueries`|Number of groupBy queries that have spilled onto the disk. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`groupBy/spilledBytes`|Number of bytes spilled on the disk by the groupBy queries. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
|`groupBy/maxSpilledBytes`|Maximum number of bytes spilled to disk by any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies|
Expand Down
90 changes: 83 additions & 7 deletions processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.DoubleAccumulator;

/**
* Collects groupBy query metrics (spilled bytes, merge buffer usage, dictionary size) per-query, then
Expand Down Expand Up @@ -73,6 +74,7 @@ public static class AggregateStats
private long maxMergeBufferAcquisitionTimeNs = 0;
private long totalMergeBufferUsedBytes = 0;
private long maxMergeBufferUsedBytes = 0;
private double maxSpillProximity = 0.0;
private long spilledQueries = 0;
private long spilledBytes = 0;
private long maxSpilledBytes = 0;
Expand All @@ -91,6 +93,7 @@ public AggregateStats(AggregateStats aggregateStats)
aggregateStats.maxMergeBufferAcquisitionTimeNs,
aggregateStats.totalMergeBufferUsedBytes,
aggregateStats.maxMergeBufferUsedBytes,
aggregateStats.maxSpillProximity,
aggregateStats.spilledQueries,
aggregateStats.spilledBytes,
aggregateStats.maxSpilledBytes,
Expand All @@ -105,6 +108,7 @@ public AggregateStats(
long maxMergeBufferAcquisitionTimeNs,
long totalMergeBufferUsedBytes,
long maxMergeBufferUsedBytes,
double maxSpillProximity,
long spilledQueries,
long spilledBytes,
long maxSpilledBytes,
Expand All @@ -117,6 +121,7 @@ public AggregateStats(
this.maxMergeBufferAcquisitionTimeNs = maxMergeBufferAcquisitionTimeNs;
this.totalMergeBufferUsedBytes = totalMergeBufferUsedBytes;
this.maxMergeBufferUsedBytes = maxMergeBufferUsedBytes;
this.maxSpillProximity = maxSpillProximity;
this.spilledQueries = spilledQueries;
this.spilledBytes = spilledBytes;
this.maxSpilledBytes = maxSpilledBytes;
Expand Down Expand Up @@ -149,6 +154,11 @@ public long getMaxMergeBufferUsedBytes()
return maxMergeBufferUsedBytes;
}

public double getMaxSpillProximity()
{
return maxSpillProximity;
}

public long getSpilledQueries()
{
return spilledQueries;
Expand All @@ -174,6 +184,20 @@ public long getMaxMergeDictionarySize()
return maxMergeDictionarySize;
}

/**
* Folds a completed query's stats into the running aggregate. For merge-buffer usage:
* <ul>
* <li>{@code totalMergeBufferUsedBytes} (emitted as {@code mergeBuffer/bytesUsed}) sums each query's usage
* across all queries, where each query's usage is itself the sum across the query's slices.</li>
* <li>{@code maxMergeBufferUsedBytes} (emitted as {@code mergeBuffer/maxBytesUsed}) is the max such per-query
* summed usage across queries.</li>
* <li>{@code maxSpillProximity} (emitted as {@code mergeBuffer/maxSpillProximity}) is the max per-query spill
* proximity across queries, where each query's value is its fullest slice's peak
* {@code size / regrowthThreshold} (bucket-count based, tracked by the underlying hash table). Unlike the
* byte sums above, this is a per-slice MAX so it reflects the slice that drives a spill; 1.0 corresponds
* exactly to the spill trigger (a bucket allocation was rejected).</li>
* </ul>
*/
public void addQueryStats(PerQueryStats perQueryStats)
{
if (perQueryStats.getMergeBufferAcquisitionTimeNs() > 0) {
Expand All @@ -183,8 +207,9 @@ public void addQueryStats(PerQueryStats perQueryStats)
maxMergeBufferAcquisitionTimeNs,
perQueryStats.getMergeBufferAcquisitionTimeNs()
);
totalMergeBufferUsedBytes += perQueryStats.getMaxMergeBufferUsedBytes();
maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, perQueryStats.getMaxMergeBufferUsedBytes());
totalMergeBufferUsedBytes += perQueryStats.getMergeBufferUsedBytes();
maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, perQueryStats.getMergeBufferUsedBytes());
maxSpillProximity = Math.max(maxSpillProximity, perQueryStats.getSpillProximity());
}

if (perQueryStats.getSpilledBytes() > 0) {
Expand All @@ -204,6 +229,7 @@ public void reset()
this.maxMergeBufferAcquisitionTimeNs = 0;
this.totalMergeBufferUsedBytes = 0;
this.maxMergeBufferUsedBytes = 0;
this.maxSpillProximity = 0.0;
this.spilledQueries = 0;
this.spilledBytes = 0;
this.maxSpilledBytes = 0;
Expand All @@ -215,7 +241,24 @@ public void reset()
public static class PerQueryStats
{
private final AtomicLong mergeBufferAcquisitionTimeNs = new AtomicLong(0);
private final AtomicLong maxMergeBufferUsedBytes = new AtomicLong(0);
/**
* Sum of the peak merge-buffer usage of every grouper (slice) this query held. A
* {@code ConcurrentGrouper} slices a single merge buffer into one slice per processing thread, and each slice
* reports its own peak via
* {@link #addMergeBufferUsedBytes(long)} when closed, so the per-query value is the SUM across the query's slices.
*/
private final AtomicLong mergeBufferUsedBytes = new AtomicLong(0);
/**
* Spill proximity of the single fullest slice this query held, in [0.0, 1.0]. Each {@link #sliceUsage} call
* contributes one slice's peak {@code size / regrowthThreshold} ratio (tracked bucket-by-bucket by the underlying
* hash table, preserved across resets), and this keeps the MAX across the query's slices. A query spills as soon
* as one slice fills, so proximity is driven by the hottest slice, NOT the byte sum tracked by
* {@link #mergeBufferUsedBytes}. Keeping the ratio per slice (rather than maxing numerator and denominator
* independently) is what makes the metric correct when a single query mixes groupers with different spill
* thresholds — e.g. small sliced groupers from a {@code ConcurrentGrouper} alongside a full-buffer
* {@code SpillingGrouper} for subtotal/nested processing. 1.0 corresponds exactly to the actual spill trigger.
*/
private final DoubleAccumulator maxSpillProximity = new DoubleAccumulator(Math::max, 0.0);
private final AtomicLong spilledBytes = new AtomicLong(0);
private final AtomicLong mergeDictionarySize = new AtomicLong(0);

Expand All @@ -224,9 +267,32 @@ public void mergeBufferAcquisitionTime(long delay)
mergeBufferAcquisitionTimeNs.addAndGet(delay);
}

public void maxMergeBufferUsedBytes(long bytes)
/**
* Accumulates the peak merge-buffer usage of one grouper (slice). Despite the previous "max" naming, this method
* sums across the slices a query holds; see {@link #mergeBufferUsedBytes}.
*/
public void addMergeBufferUsedBytes(long bytes)
{
maxMergeBufferUsedBytes.addAndGet(bytes);
mergeBufferUsedBytes.addAndGet(bytes);
}

/**
* Records one slice's peak fill ratio in [0.0, 1.0] — the underlying hash table's peak
* {@code size / regrowthThreshold} over the slice's lifetime, which reaches exactly 1.0 iff the slice actually
* spilled. Kept as a max across the query's slices, so after all slices close the value describes the single
* fullest slice — the one that drives spilling. Recording the ratio per slice (rather than maxing bytes and
* thresholds independently) keeps each slice's numerator paired with its own denominator, so a query that mixes
* groupers of different sizes still reports the true max proximity. Used to compute
* {@code mergeBuffer/maxSpillProximity}. Values are clamped defensively to [0, 1]; NaN is ignored so a
* never-initialized grouper contributes nothing.
*/
public void sliceUsage(double proximity)
{
if (Double.isNaN(proximity)) {
return;
}
final double clamped = proximity < 0.0 ? 0.0 : (proximity > 1.0 ? 1.0 : proximity);
maxSpillProximity.accumulate(clamped);
}

public void spilledBytes(long bytes)
Expand All @@ -244,9 +310,19 @@ public long getMergeBufferAcquisitionTimeNs()
return mergeBufferAcquisitionTimeNs.get();
}

public long getMaxMergeBufferUsedBytes()
public long getMergeBufferUsedBytes()
{
return mergeBufferUsedBytes.get();
}

/**
* Spill proximity for this query in [0.0, 1.0]: the fullest slice's peak {@code size / regrowthThreshold} over
* that slice's lifetime. 1.0 corresponds exactly to the spill trigger (a bucket allocation was rejected). Returns
* 0.0 when no slice usage was recorded (e.g. a grouper that never initialized).
*/
public double getSpillProximity()
{
return maxMergeBufferUsedBytes.get();
return maxSpillProximity.get();
}

public long getSpilledBytes()
Expand Down
Loading
Loading