From 0c52f534c031ecba612b663c02fbf6b4287335f0 Mon Sep 17 00:00:00 2001 From: Andrew Ho Date: Wed, 24 Jun 2026 11:21:12 -0700 Subject: [PATCH 1/3] feat: add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis 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. --- docs/operations/metrics.md | 15 +- .../query/groupby/GroupByStatsProvider.java | 85 ++++++- .../epinephelinae/BufferHashGrouper.java | 15 +- .../epinephelinae/SpillingGrouper.java | 15 +- .../groupby/GroupByStatsProviderTest.java | 224 +++++++++++++++++- .../server/metrics/GroupByStatsMonitor.java | 1 + .../metrics/GroupByStatsMonitorTest.java | 38 ++- 7 files changed, 366 insertions(+), 27 deletions(-) diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 8d71c6044f85..812e5d4d043c 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| @@ -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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| @@ -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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| diff --git a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java index f6b92a7b62c1..ad68c649c409 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java @@ -73,6 +73,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; @@ -91,6 +92,7 @@ public AggregateStats(AggregateStats aggregateStats) aggregateStats.maxMergeBufferAcquisitionTimeNs, aggregateStats.totalMergeBufferUsedBytes, aggregateStats.maxMergeBufferUsedBytes, + aggregateStats.maxSpillProximity, aggregateStats.spilledQueries, aggregateStats.spilledBytes, aggregateStats.maxSpilledBytes, @@ -105,6 +107,7 @@ public AggregateStats( long maxMergeBufferAcquisitionTimeNs, long totalMergeBufferUsedBytes, long maxMergeBufferUsedBytes, + double maxSpillProximity, long spilledQueries, long spilledBytes, long maxSpilledBytes, @@ -117,6 +120,7 @@ public AggregateStats( this.maxMergeBufferAcquisitionTimeNs = maxMergeBufferAcquisitionTimeNs; this.totalMergeBufferUsedBytes = totalMergeBufferUsedBytes; this.maxMergeBufferUsedBytes = maxMergeBufferUsedBytes; + this.maxSpillProximity = maxSpillProximity; this.spilledQueries = spilledQueries; this.spilledBytes = spilledBytes; this.maxSpilledBytes = maxSpilledBytes; @@ -149,6 +153,11 @@ public long getMaxMergeBufferUsedBytes() return maxMergeBufferUsedBytes; } + public double getMaxSpillProximity() + { + return maxSpillProximity; + } + public long getSpilledQueries() { return spilledQueries; @@ -174,6 +183,19 @@ public long getMaxMergeDictionarySize() return maxMergeDictionarySize; } + /** + * Folds a completed query's stats into the running aggregate. For merge-buffer usage: + * + */ public void addQueryStats(PerQueryStats perQueryStats) { if (perQueryStats.getMergeBufferAcquisitionTimeNs() > 0) { @@ -183,8 +205,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) { @@ -204,6 +227,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; @@ -215,7 +239,25 @@ 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); + /** + * Peak used bytes of the single fullest slice this query held, tracked as a MAX across the query's slices. A query + * spills as soon as one slice fills, so spill proximity is driven by the hottest slice, NOT the sum tracked by + * {@link #mergeBufferUsedBytes}. Compared against {@link #sliceSpillThresholdBytes}. + */ + private final AtomicLong maxSliceUsedBytes = new AtomicLong(0); + /** + * Per-slice spill threshold in bytes, i.e. {@code sliceSize * maxLoadFactor}. A slice's hash table spills once its + * bucket count reaches the max load factor, so this (not the raw slice size) is the denominator that makes a + * usage ratio of 1.0 mean "at the spill point". Tracked as a max; all slices of a query share the same value. + */ + private final AtomicLong sliceSpillThresholdBytes = new AtomicLong(0); private final AtomicLong spilledBytes = new AtomicLong(0); private final AtomicLong mergeDictionarySize = new AtomicLong(0); @@ -224,9 +266,24 @@ 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) + { + mergeBufferUsedBytes.addAndGet(bytes); + } + + /** + * Records one slice's peak used bytes and its spill threshold ({@code sliceSize * maxLoadFactor}). Both are tracked + * as maxima, so after all slices close the pair describes the single fullest slice — the one that drives spilling. + * Used to compute {@code mergeBuffer/maxSpillProximity}. + */ + public void sliceUsage(long usedBytes, long spillThresholdBytes) { - maxMergeBufferUsedBytes.addAndGet(bytes); + maxSliceUsedBytes.accumulateAndGet(usedBytes, Math::max); + sliceSpillThresholdBytes.accumulateAndGet(spillThresholdBytes, Math::max); } public void spilledBytes(long bytes) @@ -244,9 +301,23 @@ public long getMergeBufferAcquisitionTimeNs() return mergeBufferAcquisitionTimeNs.get(); } - public long getMaxMergeBufferUsedBytes() + public long getMergeBufferUsedBytes() { - return maxMergeBufferUsedBytes.get(); + return mergeBufferUsedBytes.get(); + } + + /** + * Spill proximity for this query in [0.0, 1.0]: the fullest slice's used bytes divided by that slice's spill + * threshold. 1.0 means a slice reached the point at which it spills to disk. Returns 0.0 when no slice usage was + * recorded (e.g. a grouper that never initialized). + */ + public double getSpillProximity() + { + long threshold = sliceSpillThresholdBytes.get(); + if (threshold <= 0) { + return 0.0; + } + return Math.min(1.0, (double) maxSliceUsedBytes.get() / threshold); } public long getSpilledBytes() diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java index 670a03cb2dee..e5ab63a83d82 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java @@ -36,7 +36,8 @@ public class BufferHashGrouper extends AbstractBufferHashGrouper 0 ? maxLoadFactor : DEFAULT_MAX_LOAD_FACTOR; + this.maxLoadFactor = resolveMaxLoadFactor(maxLoadFactor); this.initialBuckets = initialBuckets > 0 ? Math.max(MIN_INITIAL_BUCKETS, initialBuckets) : DEFAULT_INITIAL_BUCKETS; if (this.maxLoadFactor >= 1.0f) { @@ -74,6 +75,16 @@ public BufferHashGrouper( this.useDefaultSorting = useDefaultSorting; } + /** + * Resolves the effective max load factor, applying {@link #DEFAULT_MAX_LOAD_FACTOR} when a non-positive value is + * configured. A hash table spills once its bucket count reaches this fraction of capacity, so this is the value to + * compare per-slice usage against when computing spill proximity (see {@code SpillingGrouper}). + */ + static float resolveMaxLoadFactor(float maxLoadFactor) + { + return maxLoadFactor > 0 ? maxLoadFactor : DEFAULT_MAX_LOAD_FACTOR; + } + @Override public void init() { diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java index 96e55907b21f..df43c5e934e8 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java @@ -84,6 +84,11 @@ public class SpillingGrouper implements Grouper private final Comparator> defaultOrderKeyObjComparator; private final GroupByStatsProvider.PerQueryStats perQueryStats; private final long minSpillFileSize; + // Per-slice spill threshold in bytes: the slice's capacity scaled by the resolved max load factor. A slice's hash + // table spills once its bucket count reaches the load factor, so this (not the raw slice size) is what the slice's + // peak usage is compared against to produce mergeBuffer/maxSpillProximity. For a ConcurrentGrouper slice the slice + // size is the per-thread fraction of the merge buffer, not the full configured buffer. + private final long spillThresholdBytes; private final List files = new ArrayList<>(); private final List dictionaryFiles = new ArrayList<>(); @@ -178,6 +183,8 @@ public SpillingGrouper( this.sortHasNonGroupingFields = sortHasNonGroupingFields; this.minSpillFileSize = minSpillFileSize; this.perQueryStats = perQueryStats; + final float resolvedLoadFactor = BufferHashGrouper.resolveMaxLoadFactor(bufferGrouperMaxLoadFactor); + this.spillThresholdBytes = (long) (mergeBufferSize * resolvedLoadFactor); } @Override @@ -249,7 +256,13 @@ public void reset() public void close() { perQueryStats.dictionarySize(getDictionarySizeEstimate()); - perQueryStats.maxMergeBufferUsedBytes(getMaxMergeBufferUsedBytes()); + final long sliceUsedBytes = getMaxMergeBufferUsedBytes(); + perQueryStats.addMergeBufferUsedBytes(sliceUsedBytes); + if (grouper.isInitialized()) { + // Report this slice's peak usage against its spill threshold so the provider can compute spill proximity. Only + // recorded when the grouper was initialized, so a grouper that never touched the merge buffer is not counted. + perQueryStats.sliceUsage(sliceUsedBytes, spillThresholdBytes); + } // Record spilled bytes before deleteFiles() decrements bytesUsed in temporaryStorage. long spilledBytes = 0; for (final File file : files) { diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java index dafd381668d8..4153c2faf3e5 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java @@ -25,6 +25,8 @@ public class GroupByStatsProviderTest { + private static final double DELTA = 1e-9; + @Test public void testMetricCollection() { @@ -35,7 +37,12 @@ public void testMetricCollection() stats1.mergeBufferAcquisitionTime(300); stats1.mergeBufferAcquisitionTime(400); - stats1.maxMergeBufferUsedBytes(50); + // Two slices of the same query: usage SUMS to 80, while spill proximity is the per-slice MAX. Slice 1 is at + // 50/1000 of its threshold, slice 2 at 30/1000; the fullest slice (0.05) drives proximity. + stats1.addMergeBufferUsedBytes(50); + stats1.sliceUsage(50, 1000); + stats1.addMergeBufferUsedBytes(30); + stats1.sliceUsage(30, 1000); stats1.spilledBytes(200); stats1.spilledBytes(400); stats1.dictionarySize(100); @@ -46,7 +53,9 @@ public void testMetricCollection() stats2.mergeBufferAcquisitionTime(500); stats2.mergeBufferAcquisitionTime(600); - stats2.maxMergeBufferUsedBytes(100); + // Single slice at 100/2000 = 0.05. + stats2.addMergeBufferUsedBytes(100); + stats2.sliceUsage(100, 2000); stats2.spilledBytes(400); stats2.spilledBytes(600); stats2.dictionarySize(300); @@ -58,6 +67,7 @@ public void testMetricCollection() Assertions.assertEquals(0L, aggregateStats.getMaxMergeBufferAcquisitionTimeNs()); Assertions.assertEquals(0L, aggregateStats.getTotalMergeBufferUsedBytes()); Assertions.assertEquals(0L, aggregateStats.getMaxMergeBufferUsedBytes()); + Assertions.assertEquals(0.0, aggregateStats.getMaxSpillProximity(), DELTA); Assertions.assertEquals(0L, aggregateStats.getSpilledQueries()); Assertions.assertEquals(0L, aggregateStats.getSpilledBytes()); Assertions.assertEquals(0L, aggregateStats.getMaxSpilledBytes()); @@ -71,8 +81,13 @@ public void testMetricCollection() Assertions.assertEquals(2, aggregateStats.getMergeBufferQueries()); Assertions.assertEquals(1800L, aggregateStats.getMergeBufferAcquisitionTimeNs()); Assertions.assertEquals(1100L, aggregateStats.getMaxMergeBufferAcquisitionTimeNs()); - Assertions.assertEquals(150L, aggregateStats.getTotalMergeBufferUsedBytes()); + // bytesUsed sums across queries AND across each query's slices: (50 + 30) + 100 = 180. + Assertions.assertEquals(180L, aggregateStats.getTotalMergeBufferUsedBytes()); + // maxBytesUsed is the max per-query summed usage: max(80, 100) = 100. Assertions.assertEquals(100L, aggregateStats.getMaxMergeBufferUsedBytes()); + // maxSpillProximity is the max per-query proximity, where each query's proximity is its fullest slice's + // fill fraction: q1 -> max(50/1000, 30/1000) = 0.05, q2 -> 100/2000 = 0.05, so max = 0.05. + Assertions.assertEquals(0.05, aggregateStats.getMaxSpillProximity(), DELTA); Assertions.assertEquals(2L, aggregateStats.getSpilledQueries()); Assertions.assertEquals(1600L, aggregateStats.getSpilledBytes()); Assertions.assertEquals(1000L, aggregateStats.getMaxSpilledBytes()); @@ -88,28 +103,32 @@ public void testMetricsWithMultipleQueries() QueryResourceId r1 = new QueryResourceId("r1"); GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); stats1.mergeBufferAcquisitionTime(2000); - stats1.maxMergeBufferUsedBytes(50); + stats1.addMergeBufferUsedBytes(50); + stats1.sliceUsage(50, 1000); // 0.05 stats1.spilledBytes(100); stats1.dictionarySize(200); QueryResourceId r2 = new QueryResourceId("r2"); GroupByStatsProvider.PerQueryStats stats2 = statsProvider.getPerQueryStatsContainer(r2); stats2.mergeBufferAcquisitionTime(100); - stats2.maxMergeBufferUsedBytes(500); + stats2.addMergeBufferUsedBytes(500); + stats2.sliceUsage(500, 1000); // 0.5 stats2.spilledBytes(150); stats2.dictionarySize(250); QueryResourceId r3 = new QueryResourceId("r3"); GroupByStatsProvider.PerQueryStats stats3 = statsProvider.getPerQueryStatsContainer(r3); stats3.mergeBufferAcquisitionTime(200); - stats3.maxMergeBufferUsedBytes(100); + stats3.addMergeBufferUsedBytes(100); + stats3.sliceUsage(100, 1000); // 0.1 stats3.spilledBytes(3000); stats3.dictionarySize(300); QueryResourceId r4 = new QueryResourceId("r4"); GroupByStatsProvider.PerQueryStats stats4 = statsProvider.getPerQueryStatsContainer(r4); stats4.mergeBufferAcquisitionTime(300); - stats4.maxMergeBufferUsedBytes(75); + stats4.addMergeBufferUsedBytes(75); + stats4.sliceUsage(75, 1000); // 0.075 stats4.spilledBytes(200); stats4.dictionarySize(1500); @@ -122,6 +141,8 @@ public void testMetricsWithMultipleQueries() Assertions.assertEquals(2000L, aggregateStats.getMaxMergeBufferAcquisitionTimeNs()); Assertions.assertEquals(500L, aggregateStats.getMaxMergeBufferUsedBytes()); + // Max per-query proximity across the four queries: max(0.05, 0.5, 0.1, 0.075) = 0.5. + Assertions.assertEquals(0.5, aggregateStats.getMaxSpillProximity(), DELTA); Assertions.assertEquals(3000L, aggregateStats.getMaxSpilledBytes()); Assertions.assertEquals(1500L, aggregateStats.getMaxMergeDictionarySize()); @@ -132,4 +153,193 @@ public void testMetricsWithMultipleQueries() Assertions.assertEquals(3450L, aggregateStats.getSpilledBytes()); Assertions.assertEquals(2250L, aggregateStats.getMergeDictionarySize()); } + + @Test + public void testPerQueryUsedBytesSumsWhileSpillProximityTakesSliceMax() + { + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + + // Simulate a ConcurrentGrouper closing four equally-sized slices. Each slice reports its own peak usage and the + // shared per-slice spill threshold. Used bytes accumulate (sum); spill proximity is driven by the fullest slice. + stats.addMergeBufferUsedBytes(10); + stats.sliceUsage(10, 1000); + stats.addMergeBufferUsedBytes(20); + stats.sliceUsage(20, 1000); + stats.addMergeBufferUsedBytes(30); + stats.sliceUsage(30, 1000); + stats.addMergeBufferUsedBytes(40); + stats.sliceUsage(40, 1000); + + // Used bytes are summed across slices. + Assertions.assertEquals(100L, stats.getMergeBufferUsedBytes()); + // Proximity is the fullest slice's fill fraction (40/1000), NOT the summed fraction (100/1000 = 0.1). + Assertions.assertEquals(0.04, stats.getSpillProximity(), DELTA); + } + + @Test + public void testSpillProximityClampsToOneAtSpillPoint() + { + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + // A slice that reached its threshold exactly. + stats.sliceUsage(700, 700); + Assertions.assertEquals(1.0, stats.getSpillProximity(), DELTA); + + // The offset-list term can push raw used bytes slightly above the threshold; proximity must clamp at 1.0. + GroupByStatsProvider.PerQueryStats over = new GroupByStatsProvider.PerQueryStats(); + over.sliceUsage(750, 700); + Assertions.assertEquals(1.0, over.getSpillProximity(), DELTA); + } + + @Test + public void testSpillProximityZeroWhenNoSliceUsageRecorded() + { + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + // No sliceUsage() call: threshold is 0, so proximity is 0.0 rather than dividing by zero. + stats.addMergeBufferUsedBytes(500); + Assertions.assertEquals(0.0, stats.getSpillProximity(), DELTA); + } + + @Test + public void testSpillProximityPicksFullestSliceWhenSlicesDiffer() + { + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + // Slices with differing usage AND thresholds. maxSliceUsedBytes and sliceSpillThresholdBytes are each tracked + // as independent maxima; in practice all slices of a query share the same threshold, so this exercises the + // max-of-each behaviour for the dominating slice. + stats.sliceUsage(200, 1000); + stats.sliceUsage(900, 1000); + stats.sliceUsage(100, 1000); + // Fullest slice is 900/1000 = 0.9. + Assertions.assertEquals(0.9, stats.getSpillProximity(), DELTA); + } + + @Test + public void testAggregateStatsResetZeroesSpillProximity() + { + GroupByStatsProvider.AggregateStats aggregateStats = new GroupByStatsProvider.AggregateStats( + 1L, + 100L, + 100L, + 200L, + 200L, + 0.75, + 2L, + 200L, + 200L, + 300L, + 300L + ); + Assertions.assertEquals(0.75, aggregateStats.getMaxSpillProximity(), DELTA); + + aggregateStats.reset(); + Assertions.assertEquals(0.0, aggregateStats.getMaxSpillProximity(), DELTA); + } + + @Test + public void testAggregateStatsCopyConstructorRoundTripsSpillProximity() + { + GroupByStatsProvider.AggregateStats original = new GroupByStatsProvider.AggregateStats( + 1L, + 100L, + 100L, + 200L, + 200L, + 0.42, + 2L, + 200L, + 200L, + 300L, + 300L + ); + GroupByStatsProvider.AggregateStats copy = new GroupByStatsProvider.AggregateStats(original); + + Assertions.assertEquals(0.42, copy.getMaxSpillProximity(), DELTA); + // Spill proximity is the 6th ctor arg, sitting between maxBytesUsed and spilledQueries; verify neighbours + // did not shift position. + Assertions.assertEquals(200L, copy.getMaxMergeBufferUsedBytes()); + Assertions.assertEquals(2L, copy.getSpilledQueries()); + } + + @Test + public void testAggregateStatsTakesMaxSpillProximityAcrossQueries() + { + GroupByStatsProvider.AggregateStats agg = new GroupByStatsProvider.AggregateStats(); + + GroupByStatsProvider.PerQueryStats low = new GroupByStatsProvider.PerQueryStats(); + low.mergeBufferAcquisitionTime(10); + low.sliceUsage(300, 1000); // 0.3 + agg.addQueryStats(low); + + GroupByStatsProvider.PerQueryStats high = new GroupByStatsProvider.PerQueryStats(); + high.mergeBufferAcquisitionTime(10); + high.sliceUsage(800, 1000); // 0.8 + agg.addQueryStats(high); + + GroupByStatsProvider.PerQueryStats mid = new GroupByStatsProvider.PerQueryStats(); + mid.mergeBufferAcquisitionTime(10); + mid.sliceUsage(500, 1000); // 0.5 + agg.addQueryStats(mid); + + Assertions.assertEquals(0.8, agg.getMaxSpillProximity(), DELTA); + } + + @Test + public void testSpillProximityDroppedWhenNoAcquisitionTimeRecorded() + { + // A PerQueryStats with slice usage but no acquisition time is not folded into the mergeBuffer block, mirroring + // the monitor guard. In practice this never happens: acquisition time is recorded in + // GroupByResourcesReservationPool.reserve() before any grouper initializes. + GroupByStatsProvider.AggregateStats agg = new GroupByStatsProvider.AggregateStats(); + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + stats.sliceUsage(900, 1000); + agg.addQueryStats(stats); + + Assertions.assertEquals(0L, agg.getMergeBufferQueries()); + Assertions.assertEquals(0.0, agg.getMaxSpillProximity(), DELTA); + } + + /** + * End-to-end through {@link GroupByStatsProvider} reproducing the user's scenario: a 125MiB merge buffer divided + * into 240 per-thread slices (sliceSize ~= 546KiB). Each slice fills well below the configured buffer size, yet the + * fullest slice reaches its spill threshold (sliceSize * 0.7) and the query spills. The summed {@code bytesUsed} is + * far below {@code sizeBytes}, which is exactly why comparing it to {@code sizeBytes} was misleading; {@code + * maxSpillProximity} instead reports ~1.0, correctly indicating the query was at the spill point. + */ + @Test + public void testEndToEndSlicedBufferSpillScenario() + { + final long sizeBytes = 125L * 1024 * 1024; // druid.processing.buffer.sizeBytes (125MiB) + final int numThreads = 240; // concurrencyHint / numThreads + final long sliceSize = sizeBytes / numThreads; // per-slice capacity (~546KiB) + final float loadFactor = 0.7f; // BufferHashGrouper.DEFAULT_MAX_LOAD_FACTOR + final long sliceThreshold = (long) (sliceSize * loadFactor); + + GroupByStatsProvider statsProvider = new GroupByStatsProvider(); + QueryResourceId id = new QueryResourceId("spilly"); + GroupByStatsProvider.PerQueryStats stats = statsProvider.getPerQueryStatsContainer(id); + + stats.mergeBufferAcquisitionTime(42); + long expectedUsed = 0; + for (int i = 0; i < numThreads; i++) { + // Most slices stay light; one slice (i == 0) reaches its threshold and triggers the spill. + final long sliceUsed = (i == 0) ? sliceThreshold : sliceThreshold / 10; + stats.addMergeBufferUsedBytes(sliceUsed); + stats.sliceUsage(sliceUsed, sliceThreshold); + expectedUsed += sliceUsed; + } + stats.spilledBytes(1_000_000L); + + statsProvider.closeQuery(id); + GroupByStatsProvider.AggregateStats aggregateStats = statsProvider.getStatsSince(); + + // The fullest slice is at its threshold, so proximity is 1.0: the query spilled. + Assertions.assertEquals(1.0, aggregateStats.getMaxSpillProximity(), DELTA); + // ...even though the summed usage across slices is a tiny fraction of the configured buffer size. + Assertions.assertEquals(expectedUsed, aggregateStats.getTotalMergeBufferUsedBytes()); + Assertions.assertTrue( + aggregateStats.getTotalMergeBufferUsedBytes() < sizeBytes / 2, + "summed bytesUsed should look small next to sizeBytes, despite the spill" + ); + Assertions.assertEquals(1L, aggregateStats.getSpilledQueries()); + } } diff --git a/server/src/main/java/org/apache/druid/server/metrics/GroupByStatsMonitor.java b/server/src/main/java/org/apache/druid/server/metrics/GroupByStatsMonitor.java index e5f46020fe09..9c69f182f8f5 100644 --- a/server/src/main/java/org/apache/druid/server/metrics/GroupByStatsMonitor.java +++ b/server/src/main/java/org/apache/druid/server/metrics/GroupByStatsMonitor.java @@ -68,6 +68,7 @@ public boolean doMonitor(ServiceEmitter emitter) emitter.emit(builder.setMetric("mergeBuffer/maxAcquisitionTimeNs", statsContainer.getMaxMergeBufferAcquisitionTimeNs())); emitter.emit(builder.setMetric("mergeBuffer/bytesUsed", statsContainer.getTotalMergeBufferUsedBytes())); emitter.emit(builder.setMetric("mergeBuffer/maxBytesUsed", statsContainer.getMaxMergeBufferUsedBytes())); + emitter.emit(builder.setMetric("mergeBuffer/maxSpillProximity", statsContainer.getMaxSpillProximity())); } if (statsContainer.getSpilledQueries() > 0) { diff --git a/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java b/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java index eaca043e02e7..8d02e64e3b49 100644 --- a/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java +++ b/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java @@ -63,6 +63,7 @@ public synchronized AggregateStats getStatsSince() 100L, 200L, 200L, + 0.85, 2L, 200L, 200L, @@ -93,7 +94,7 @@ public void testMonitor() // Trigger metric emission monitor.doMonitor(emitter); - Assert.assertEquals(12, emitter.getNumEmittedEvents()); + Assert.assertEquals(13, emitter.getNumEmittedEvents()); emitter.verifyValue("mergeBuffer/pendingRequests", 0L); emitter.verifyValue("mergeBuffer/used", 0L); emitter.verifyValue("mergeBuffer/queries", 1L); @@ -101,6 +102,7 @@ public void testMonitor() emitter.verifyValue("mergeBuffer/maxAcquisitionTimeNs", 100L); emitter.verifyValue("mergeBuffer/bytesUsed", 200L); emitter.verifyValue("mergeBuffer/maxBytesUsed", 200L); + emitter.verifyValue("mergeBuffer/maxSpillProximity", 0.85); emitter.verifyValue("groupBy/spilledQueries", 2L); emitter.verifyValue("groupBy/spilledBytes", 200L); emitter.verifyValue("groupBy/maxSpilledBytes", 200L); @@ -137,6 +139,7 @@ public void testMonitorWithServiceDimensions() verifyMetricValue(emitter, "mergeBuffer/maxAcquisitionTimeNs", dimFilters, 100L); verifyMetricValue(emitter, "mergeBuffer/bytesUsed", dimFilters, 200L); verifyMetricValue(emitter, "mergeBuffer/maxBytesUsed", dimFilters, 200L); + verifyMetricValue(emitter, "mergeBuffer/maxSpillProximity", dimFilters, 0.85); verifyMetricValue(emitter, "groupBy/spilledQueries", dimFilters, 2L); verifyMetricValue(emitter, "groupBy/spilledBytes", dimFilters, 200L); verifyMetricValue(emitter, "groupBy/maxSpilledBytes", dimFilters, 200L); @@ -202,21 +205,24 @@ public void testMonitoringWithMultipleResources() QueryResourceId r1 = new QueryResourceId("r1"); GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); stats1.mergeBufferAcquisitionTime(100); - stats1.maxMergeBufferUsedBytes(50); + stats1.addMergeBufferUsedBytes(50); + stats1.sliceUsage(50, 1000); // 0.05 stats1.spilledBytes(200); stats1.dictionarySize(100); QueryResourceId r2 = new QueryResourceId("r2"); GroupByStatsProvider.PerQueryStats stats2 = statsProvider.getPerQueryStatsContainer(r2); stats2.mergeBufferAcquisitionTime(500); - stats2.maxMergeBufferUsedBytes(30); + stats2.addMergeBufferUsedBytes(30); + stats2.sliceUsage(30, 2000); // 0.015 stats2.spilledBytes(100); stats2.dictionarySize(300); QueryResourceId r3 = new QueryResourceId("r3"); GroupByStatsProvider.PerQueryStats stats3 = statsProvider.getPerQueryStatsContainer(r3); stats3.mergeBufferAcquisitionTime(200); - stats3.maxMergeBufferUsedBytes(150); + stats3.addMergeBufferUsedBytes(150); + stats3.sliceUsage(150, 1500); // 0.1 stats3.spilledBytes(800); stats3.dictionarySize(200); @@ -239,10 +245,34 @@ public void testMonitoringWithMultipleResources() emitter.verifyValue("mergeBuffer/maxAcquisitionTimeNs", 500L); emitter.verifyValue("mergeBuffer/maxBytesUsed", 150L); + // Spill proximity is the MAX per-query fullest-slice fill fraction: max(0.05, 0.015, 0.1) = 0.1. + emitter.verifyValue("mergeBuffer/maxSpillProximity", 0.1); emitter.verifyValue("groupBy/maxSpilledBytes", 800L); emitter.verifyValue("groupBy/maxMergeDictionarySize", 300L); } + @Test + public void testMaxSpillProximityNotEmittedWhenNoMergeBufferQueries() + { + // No query records any merge-buffer acquisition time, so the entire mergeBuffer/* block is skipped. + GroupByStatsProvider statsProvider = new GroupByStatsProvider(); + + QueryResourceId r1 = new QueryResourceId("r1"); + GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); + // dictionary-only activity, no merge buffer acquisition + stats1.dictionarySize(100); + statsProvider.closeQuery(r1); + + final GroupByStatsMonitor monitor = new GroupByStatsMonitor(statsProvider, mergeBufferPool); + final StubServiceEmitter emitter = new StubServiceEmitter("service", "host"); + emitter.start(); + monitor.doMonitor(emitter); + + Assert.assertTrue(emitter.getMetricEvents("mergeBuffer/queries").isEmpty()); + Assert.assertTrue(emitter.getMetricEvents("mergeBuffer/maxBytesUsed").isEmpty()); + Assert.assertTrue(emitter.getMetricEvents("mergeBuffer/maxSpillProximity").isEmpty()); + } + private void verifyMetricValue(StubServiceEmitter emitter, String metricName, Map dimFilters, Number expectedValue) { final List observedMetricEvents = emitter.getMetricEvents(metricName); From fc770ee5353aa509670e4140332ae4ba0d710916 Mon Sep 17 00:00:00 2001 From: Andrew Ho Date: Mon, 6 Jul 2026 16:10:19 -0700 Subject: [PATCH 2/3] fix: pair per-slice usage with its own threshold for maxSpillProximity 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. --- .../query/groupby/GroupByStatsProvider.java | 41 +++++++++---------- .../groupby/GroupByStatsProviderTest.java | 19 +++++++-- 2 files changed, 36 insertions(+), 24 deletions(-) diff --git a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java index ad68c649c409..882f515627a7 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java @@ -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 @@ -247,17 +248,15 @@ public static class PerQueryStats */ private final AtomicLong mergeBufferUsedBytes = new AtomicLong(0); /** - * Peak used bytes of the single fullest slice this query held, tracked as a MAX across the query's slices. A query - * spills as soon as one slice fills, so spill proximity is driven by the hottest slice, NOT the sum tracked by - * {@link #mergeBufferUsedBytes}. Compared against {@link #sliceSpillThresholdBytes}. + * Spill proximity of the single fullest slice this query held, in [0.0, 1.0]. Each {@link #sliceUsage} call + * computes one slice's {@code usedBytes / spillThresholdBytes} ratio, 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 (rather than the numerator and denominator as + * independent maxima) 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. */ - private final AtomicLong maxSliceUsedBytes = new AtomicLong(0); - /** - * Per-slice spill threshold in bytes, i.e. {@code sliceSize * maxLoadFactor}. A slice's hash table spills once its - * bucket count reaches the max load factor, so this (not the raw slice size) is the denominator that makes a - * usage ratio of 1.0 mean "at the spill point". Tracked as a max; all slices of a query share the same value. - */ - private final AtomicLong sliceSpillThresholdBytes = new AtomicLong(0); + private final DoubleAccumulator maxSpillProximity = new DoubleAccumulator(Math::max, 0.0); private final AtomicLong spilledBytes = new AtomicLong(0); private final AtomicLong mergeDictionarySize = new AtomicLong(0); @@ -276,14 +275,18 @@ public void addMergeBufferUsedBytes(long bytes) } /** - * Records one slice's peak used bytes and its spill threshold ({@code sliceSize * maxLoadFactor}). Both are tracked - * as maxima, so after all slices close the pair describes the single fullest slice — the one that drives spilling. - * Used to compute {@code mergeBuffer/maxSpillProximity}. + * Records one slice's peak used bytes against its spill threshold ({@code sliceSize * maxLoadFactor}). The ratio is + * computed and clamped per slice, then kept as a max, so after all slices close the value describes the single + * fullest slice — the one that drives spilling. Computing the ratio per call (rather than maxing used bytes and + * threshold separately) 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}. A non-positive threshold contributes nothing (avoids divide-by-zero). */ public void sliceUsage(long usedBytes, long spillThresholdBytes) { - maxSliceUsedBytes.accumulateAndGet(usedBytes, Math::max); - sliceSpillThresholdBytes.accumulateAndGet(spillThresholdBytes, Math::max); + if (spillThresholdBytes > 0) { + maxSpillProximity.accumulate(Math.min(1.0, (double) usedBytes / spillThresholdBytes)); + } } public void spilledBytes(long bytes) @@ -307,17 +310,13 @@ public long getMergeBufferUsedBytes() } /** - * Spill proximity for this query in [0.0, 1.0]: the fullest slice's used bytes divided by that slice's spill + * Spill proximity for this query in [0.0, 1.0]: the fullest slice's used bytes divided by that slice's own spill * threshold. 1.0 means a slice reached the point at which it spills to disk. Returns 0.0 when no slice usage was * recorded (e.g. a grouper that never initialized). */ public double getSpillProximity() { - long threshold = sliceSpillThresholdBytes.get(); - if (threshold <= 0) { - return 0.0; - } - return Math.min(1.0, (double) maxSliceUsedBytes.get() / threshold); + return maxSpillProximity.get(); } public long getSpilledBytes() diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java index 4153c2faf3e5..c4964b144707 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java @@ -203,9 +203,7 @@ public void testSpillProximityZeroWhenNoSliceUsageRecorded() public void testSpillProximityPicksFullestSliceWhenSlicesDiffer() { GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - // Slices with differing usage AND thresholds. maxSliceUsedBytes and sliceSpillThresholdBytes are each tracked - // as independent maxima; in practice all slices of a query share the same threshold, so this exercises the - // max-of-each behaviour for the dominating slice. + // Slices sharing a threshold but with differing usage; proximity is the fullest slice's ratio. stats.sliceUsage(200, 1000); stats.sliceUsage(900, 1000); stats.sliceUsage(100, 1000); @@ -213,6 +211,21 @@ public void testSpillProximityPicksFullestSliceWhenSlicesDiffer() Assertions.assertEquals(0.9, stats.getSpillProximity(), DELTA); } + @Test + public void testSpillProximityKeepsPerSliceRatioWhenThresholdsDiffer() + { + // A single query can pass one PerQueryStats through both small sliced groupers (from a ConcurrentGrouper) and a + // full-buffer SpillingGrouper (subtotal/nested processing). A small slice can saturate its own tiny threshold + // while a much larger full-buffer grouper stays lightly filled. Proximity must pair each slice's used bytes with + // its OWN threshold, so the saturated small slice reports 1.0 and is not diluted by the large threshold. + GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + stats.sliceUsage(1000, 1000); // small sliced grouper at its spill point -> 1.0 + stats.sliceUsage(5000, 1_000_000); // large full-buffer grouper barely filled -> 0.005 + // If used bytes and thresholds were maxed independently, this would report 5000/1_000_000 = 0.005 and hide the + // spill. Pairing per slice keeps the true max of 1.0. + Assertions.assertEquals(1.0, stats.getSpillProximity(), DELTA); + } + @Test public void testAggregateStatsResetZeroesSpillProximity() { From 52ace82275c6767f1de26f61081940a307fdb703 Mon Sep 17 00:00:00 2001 From: Andrew Ho Date: Tue, 7 Jul 2026 13:35:25 -0700 Subject: [PATCH 3/3] fix: base maxSpillProximity on the hash table's actual spill trigger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- docs/operations/metrics.md | 6 +- .../query/groupby/GroupByStatsProvider.java | 44 +++-- .../AbstractBufferHashGrouper.java | 11 ++ .../epinephelinae/BufferHashGrouper.java | 15 +- .../epinephelinae/ByteBufferHashTable.java | 64 ++++++- .../LimitedBufferHashGrouper.java | 22 +++ .../epinephelinae/SpillingGrouper.java | 15 +- .../groupby/GroupByStatsProviderTest.java | 108 ++++++----- .../epinephelinae/BufferHashGrouperTest.java | 174 ++++++++++++++++++ .../LimitedBufferHashGrouperTest.java | 31 ++++ .../epinephelinae/SpillingGrouperTest.java | 127 ++++++++++++- .../metrics/GroupByStatsMonitorTest.java | 6 +- 12 files changed, 521 insertions(+), 102 deletions(-) diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 812e5d4d043c..5989039391b7 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -95,7 +95,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring |`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`|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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| +|`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| @@ -125,7 +125,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring |`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`|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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| +|`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| @@ -158,7 +158,7 @@ to represent the task ID are deprecated and will be removed in a future release. |`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`|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 reaches the hash-table load factor; this metric is that fullest slice's fill fraction of its spill threshold (max across slices, then across queries). 1.0 means at least one query reached the spill point — 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| +|`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| diff --git a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java index 882f515627a7..6fb05154ba68 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/GroupByStatsProvider.java @@ -192,9 +192,10 @@ public long getMaxMergeDictionarySize() *
  • {@code maxMergeBufferUsedBytes} (emitted as {@code mergeBuffer/maxBytesUsed}) is the max such per-query * summed usage across queries.
    • *
  • {@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 fill fraction of the spill - * threshold. Unlike the byte sums above, this is a per-slice MAX so it reflects the slice that drives a - * spill; 1.0 means at least one query reached the spill point.
    • + * 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). * */ public void addQueryStats(PerQueryStats perQueryStats) @@ -249,12 +250,13 @@ public static class PerQueryStats 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 - * computes one slice's {@code usedBytes / spillThresholdBytes} ratio, 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 (rather than the numerator and denominator as - * independent maxima) is what makes the metric correct when a single query mixes groupers with different spill + * 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. + * {@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); @@ -275,18 +277,22 @@ public void addMergeBufferUsedBytes(long bytes) } /** - * Records one slice's peak used bytes against its spill threshold ({@code sliceSize * maxLoadFactor}). The ratio is - * computed and clamped per slice, then kept as a max, so after all slices close the value describes the single - * fullest slice — the one that drives spilling. Computing the ratio per call (rather than maxing used bytes and - * threshold separately) keeps each slice's numerator paired with its own denominator, so a query that mixes + * 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}. A non-positive threshold contributes nothing (avoids divide-by-zero). + * {@code mergeBuffer/maxSpillProximity}. Values are clamped defensively to [0, 1]; NaN is ignored so a + * never-initialized grouper contributes nothing. */ - public void sliceUsage(long usedBytes, long spillThresholdBytes) + public void sliceUsage(double proximity) { - if (spillThresholdBytes > 0) { - maxSpillProximity.accumulate(Math.min(1.0, (double) usedBytes / spillThresholdBytes)); + 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) @@ -310,9 +316,9 @@ public long getMergeBufferUsedBytes() } /** - * Spill proximity for this query in [0.0, 1.0]: the fullest slice's used bytes divided by that slice's own spill - * threshold. 1.0 means a slice reached the point at which it spills to disk. Returns 0.0 when no slice usage was - * recorded (e.g. a grouper that never initialized). + * 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() { diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java index a5edb38cfa4b..40f5865c4dd2 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/AbstractBufferHashGrouper.java @@ -185,6 +185,17 @@ public long getMaxMergeBufferUsedBytes() return hashTable.getMaxMergeBufferUsedBytes(); } + /** + * Peak {@code size / regrowthThreshold} observed over this grouper's lifetime, in [0.0, 1.0]. 1.0 means the grouper + * actually hit its spill trigger (a bucket allocation was rejected) — the value {@code SpillingGrouper} reports for + * {@code mergeBuffer/maxSpillProximity}. Preserved across {@link #reset()}, so a slice that spilled and then had + * more rows added still reports 1.0 at close time. Returns 0.0 when the grouper has never been initialized. + */ + public double getMaxSpillProximity() + { + return hashTable == null ? 0.0 : hashTable.getMaxSpillProximity(); + } + /** * Populate a {@link ReusableEntry} with values from a particular bucket. */ diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java index e5ab63a83d82..670a03cb2dee 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouper.java @@ -36,8 +36,7 @@ public class BufferHashGrouper extends AbstractBufferHashGrouper 0 ? maxLoadFactor : DEFAULT_MAX_LOAD_FACTOR; this.initialBuckets = initialBuckets > 0 ? Math.max(MIN_INITIAL_BUCKETS, initialBuckets) : DEFAULT_INITIAL_BUCKETS; if (this.maxLoadFactor >= 1.0f) { @@ -75,16 +74,6 @@ public BufferHashGrouper( this.useDefaultSorting = useDefaultSorting; } - /** - * Resolves the effective max load factor, applying {@link #DEFAULT_MAX_LOAD_FACTOR} when a non-positive value is - * configured. A hash table spills once its bucket count reaches this fraction of capacity, so this is the value to - * compare per-slice usage against when computing spill proximity (see {@code SpillingGrouper}). - */ - static float resolveMaxLoadFactor(float maxLoadFactor) - { - return maxLoadFactor > 0 ? maxLoadFactor : DEFAULT_MAX_LOAD_FACTOR; - } - @Override public void init() { diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferHashTable.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferHashTable.java index 0f64d08613c5..d5ae7739b5b0 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferHashTable.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferHashTable.java @@ -90,6 +90,12 @@ public static int calculateTableArenaSizeWithFixedAdditionalSize( // Tracks maximum bytes used for the entire lifecycle of this hash table. protected long maxMergeBufferUsedBytes; + // Peak {@code size / regrowthThreshold} over this table's lifetime, in [0.0, 1.0]. The table can accept a new bucket + // only while {@code size < regrowthThreshold}, so this equals 1.0 exactly at the moment a new-bucket insert would be + // rejected — the spill trigger in {@link SpillingGrouper#aggregate}. Preserved across {@link #reset()} so a grouper + // that spilled retains the 1.0 peak even after {@code size} returns to 0. + protected double maxSpillProximity; + public ByteBufferHashTable( float maxLoadFactor, int initialBuckets, @@ -109,6 +115,7 @@ public ByteBufferHashTable( this.tableArenaSize = buffer.capacity(); this.bucketUpdateHandler = bucketUpdateHandler; this.maxMergeBufferUsedBytes = 0; + this.maxSpillProximity = 0.0; } public void reset() @@ -290,6 +297,13 @@ protected int findBucketWithAutoGrowth( } } + if (bucket < 0) { + // This is the caller's spill trigger: no bucket could be allocated even after attempting to grow. Pin the + // proximity peak to exactly 1.0 here so callers see "at the spill point" independent of bucket width or + // offset-list overhead. See {@link #maxSpillProximity}. + maxSpillProximity = 1.0; + } + return bucket; } @@ -443,11 +457,46 @@ public int getGrowthCount() /** * To maintain an accurate tracking of the maximum bytes used per query, this function is to be called immediately - * whenever either of {@link #size} or {@link #bucketSizeWithHash} is changed. + * whenever either of {@link #size} or {@link #bucketSizeWithHash} is changed. Also updates {@link #maxSpillProximity} + * on every size mutation, so the peak {@code size / regrowthThreshold} ratio is tracked without the caller having to + * remember to observe it. Preserving the peak across resets is what lets a grouper that already spilled report + * proximity 1.0 even though {@code size} is currently 0. + * + *

    The ratio is captured strictly while {@code size < regrowthThreshold}. At intermediate growth boundaries + * {@code size} hits {@code regrowthThreshold} transiently — but the arena still has room to grow, so the very next + * mutation triggers {@link #adjustTableWhenFull()} which enlarges {@code regrowthThreshold}. Recording 1.0 there + * would falsely mark ordinary growth as a spill. The exception is the TERMINAL growth level, guarded by + * {@link #isTerminalTableLevel()}: at that level no further growth is possible, so parking at + * {@code size == regrowthThreshold} IS the spill point and 1.0 is recorded. The other definitive spill trigger, + * where 1.0 is also pinned, is in {@link #findBucketWithAutoGrowth} when a bucket rejection actually occurs.

    */ protected void updateMaxMergeBufferUsedBytes() { maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, (long) size * bucketSizeWithHash); + if (regrowthThreshold <= 0) { + return; + } + if (size < regrowthThreshold) { + final double ratio = (double) size / regrowthThreshold; + if (ratio > maxSpillProximity) { + maxSpillProximity = ratio; + } + } else if (isTerminalTableLevel()) { + // Table is at the load-factor limit and no further growth is possible: functionally the spill point. + maxSpillProximity = 1.0; + } + } + + /** + * True when the current table level cannot be enlarged any further. Base implementation: {@link #tableStart} has + * reached the front of the arena, matching {@link #adjustTableWhenFull()}'s early-return. Overridden by + * hash-table variants (e.g. the alternating heap-trim variant in {@code LimitedBufferHashGrouper}) whose + * "table full" event is NOT a spill trigger — those return false so proximity is only pinned via the explicit + * spill path in {@link #findBucketWithAutoGrowth}. + */ + protected boolean isTerminalTableLevel() + { + return tableStart == 0; } public long getMaxMergeBufferUsedBytes() @@ -455,6 +504,19 @@ public long getMaxMergeBufferUsedBytes() return maxMergeBufferUsedBytes; } + /** + * Peak {@code size / regrowthThreshold} observed over this table's lifetime, in [0.0, 1.0]. Equals 1.0 exactly when + * a new-bucket insert has been rejected inside {@link #findBucketWithAutoGrowth} — the true spill trigger. This is + * bucket-count based, so the value is independent of bucket width, offset-list overhead, and integer truncation in + * the arena-size calculation. Preserved across {@link #reset()} so a grouper that already spilled retains the 1.0 + * peak even after {@code size} returns to 0. Ordinary table growth (when {@link #adjustTableWhenFull()} enlarges + * {@code regrowthThreshold} instead of triggering a spill) does not push this to 1.0. + */ + public double getMaxSpillProximity() + { + return maxSpillProximity; + } + public interface BucketUpdateHandler { void handleNewBucket(int bucketOffset); diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java index 2bb544d0cda7..b428c31816c5 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouper.java @@ -587,5 +587,27 @@ public void adjustTableWhenFull() tableBuffer = newTableBuffer; growthCount++; } + + /** + * Alternating sub-buffers can always accept another heap-trim swap; hitting {@code size == regrowthThreshold} here + * is not a spill trigger (the limit push-down grouper doesn't spill from this path). Return false so the base + * class's terminal-level 1.0 pin does not fire on ordinary swaps. + * + *

    Note that a raw {@code tableStart == 0} check (the base implementation) would ALSO be wrong here: + * {@code tableStart} stays at 0 for this table's entire life (it's never set by the ctor or {@link #reset()}), + * so a base implementation would spuriously fire on every swap where {@code size == regrowthThreshold}. The + * override is what keeps limit-push-down queries from a false-positive 1.0.

    + * + *

    Why the Alternating table's true spill trigger is safe to rely on {@link #findBucketWithAutoGrowth} + * returning -1: after a swap, {@code size = numCopied} and {@code numCopied <= limit}. The grouper is + * validated at construction to have {@code regrowthThreshold >= limit + 1}, so + * {@code size < regrowthThreshold} always holds post-swap and the subsequent findBucket succeeds. Only a + * genuinely-full-and-cannot-swap condition would return -1 — the real spill case.

    + */ + @Override + protected boolean isTerminalTableLevel() + { + return false; + } } } diff --git a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java index df43c5e934e8..b149519b1428 100644 --- a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java +++ b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouper.java @@ -84,11 +84,6 @@ public class SpillingGrouper implements Grouper private final Comparator> defaultOrderKeyObjComparator; private final GroupByStatsProvider.PerQueryStats perQueryStats; private final long minSpillFileSize; - // Per-slice spill threshold in bytes: the slice's capacity scaled by the resolved max load factor. A slice's hash - // table spills once its bucket count reaches the load factor, so this (not the raw slice size) is what the slice's - // peak usage is compared against to produce mergeBuffer/maxSpillProximity. For a ConcurrentGrouper slice the slice - // size is the per-thread fraction of the merge buffer, not the full configured buffer. - private final long spillThresholdBytes; private final List files = new ArrayList<>(); private final List dictionaryFiles = new ArrayList<>(); @@ -183,8 +178,6 @@ public SpillingGrouper( this.sortHasNonGroupingFields = sortHasNonGroupingFields; this.minSpillFileSize = minSpillFileSize; this.perQueryStats = perQueryStats; - final float resolvedLoadFactor = BufferHashGrouper.resolveMaxLoadFactor(bufferGrouperMaxLoadFactor); - this.spillThresholdBytes = (long) (mergeBufferSize * resolvedLoadFactor); } @Override @@ -259,9 +252,11 @@ public void close() final long sliceUsedBytes = getMaxMergeBufferUsedBytes(); perQueryStats.addMergeBufferUsedBytes(sliceUsedBytes); if (grouper.isInitialized()) { - // Report this slice's peak usage against its spill threshold so the provider can compute spill proximity. Only - // recorded when the grouper was initialized, so a grouper that never touched the merge buffer is not counted. - perQueryStats.sliceUsage(sliceUsedBytes, spillThresholdBytes); + // Report this slice's peak fill ratio (bucket-count based, tracked by the underlying hash table across the + // grouper's lifetime and preserved across reset()). 1.0 means the grouper actually hit its spill trigger; a + // lightly-filled slice reports <1.0 exactly. Only recorded when the grouper was initialized, so an untouched + // slice contributes nothing. + perQueryStats.sliceUsage(grouper.getMaxSpillProximity()); } // Record spilled bytes before deleteFiles() decrements bytesUsed in temporaryStorage. long spilledBytes = 0; diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java index c4964b144707..1a726dfccb44 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByStatsProviderTest.java @@ -37,12 +37,12 @@ public void testMetricCollection() stats1.mergeBufferAcquisitionTime(300); stats1.mergeBufferAcquisitionTime(400); - // Two slices of the same query: usage SUMS to 80, while spill proximity is the per-slice MAX. Slice 1 is at - // 50/1000 of its threshold, slice 2 at 30/1000; the fullest slice (0.05) drives proximity. + // Two slices of the same query: usage SUMS to 80, while spill proximity is the per-slice MAX. Both slices report + // their own peak fill ratio (bucket-count based, in [0,1]); the fullest (0.05) drives proximity. stats1.addMergeBufferUsedBytes(50); - stats1.sliceUsage(50, 1000); + stats1.sliceUsage(0.05); stats1.addMergeBufferUsedBytes(30); - stats1.sliceUsage(30, 1000); + stats1.sliceUsage(0.03); stats1.spilledBytes(200); stats1.spilledBytes(400); stats1.dictionarySize(100); @@ -53,9 +53,9 @@ public void testMetricCollection() stats2.mergeBufferAcquisitionTime(500); stats2.mergeBufferAcquisitionTime(600); - // Single slice at 100/2000 = 0.05. + // Single slice at 0.05. stats2.addMergeBufferUsedBytes(100); - stats2.sliceUsage(100, 2000); + stats2.sliceUsage(0.05); stats2.spilledBytes(400); stats2.spilledBytes(600); stats2.dictionarySize(300); @@ -86,7 +86,7 @@ public void testMetricCollection() // maxBytesUsed is the max per-query summed usage: max(80, 100) = 100. Assertions.assertEquals(100L, aggregateStats.getMaxMergeBufferUsedBytes()); // maxSpillProximity is the max per-query proximity, where each query's proximity is its fullest slice's - // fill fraction: q1 -> max(50/1000, 30/1000) = 0.05, q2 -> 100/2000 = 0.05, so max = 0.05. + // fill ratio: q1 -> max(0.05, 0.03) = 0.05, q2 -> 0.05, so max = 0.05. Assertions.assertEquals(0.05, aggregateStats.getMaxSpillProximity(), DELTA); Assertions.assertEquals(2L, aggregateStats.getSpilledQueries()); Assertions.assertEquals(1600L, aggregateStats.getSpilledBytes()); @@ -104,7 +104,7 @@ public void testMetricsWithMultipleQueries() GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); stats1.mergeBufferAcquisitionTime(2000); stats1.addMergeBufferUsedBytes(50); - stats1.sliceUsage(50, 1000); // 0.05 + stats1.sliceUsage(0.05); stats1.spilledBytes(100); stats1.dictionarySize(200); @@ -112,7 +112,7 @@ public void testMetricsWithMultipleQueries() GroupByStatsProvider.PerQueryStats stats2 = statsProvider.getPerQueryStatsContainer(r2); stats2.mergeBufferAcquisitionTime(100); stats2.addMergeBufferUsedBytes(500); - stats2.sliceUsage(500, 1000); // 0.5 + stats2.sliceUsage(0.5); stats2.spilledBytes(150); stats2.dictionarySize(250); @@ -120,7 +120,7 @@ public void testMetricsWithMultipleQueries() GroupByStatsProvider.PerQueryStats stats3 = statsProvider.getPerQueryStatsContainer(r3); stats3.mergeBufferAcquisitionTime(200); stats3.addMergeBufferUsedBytes(100); - stats3.sliceUsage(100, 1000); // 0.1 + stats3.sliceUsage(0.1); stats3.spilledBytes(3000); stats3.dictionarySize(300); @@ -128,7 +128,7 @@ public void testMetricsWithMultipleQueries() GroupByStatsProvider.PerQueryStats stats4 = statsProvider.getPerQueryStatsContainer(r4); stats4.mergeBufferAcquisitionTime(300); stats4.addMergeBufferUsedBytes(75); - stats4.sliceUsage(75, 1000); // 0.075 + stats4.sliceUsage(0.075); stats4.spilledBytes(200); stats4.dictionarySize(1500); @@ -159,42 +159,52 @@ public void testPerQueryUsedBytesSumsWhileSpillProximityTakesSliceMax() { GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - // Simulate a ConcurrentGrouper closing four equally-sized slices. Each slice reports its own peak usage and the - // shared per-slice spill threshold. Used bytes accumulate (sum); spill proximity is driven by the fullest slice. + // Simulate a ConcurrentGrouper closing four equally-sized slices. Each slice reports its own peak fill ratio (in + // [0, 1]) and its own peak used bytes. Used bytes accumulate (sum); spill proximity is driven by the fullest slice. stats.addMergeBufferUsedBytes(10); - stats.sliceUsage(10, 1000); + stats.sliceUsage(0.01); stats.addMergeBufferUsedBytes(20); - stats.sliceUsage(20, 1000); + stats.sliceUsage(0.02); stats.addMergeBufferUsedBytes(30); - stats.sliceUsage(30, 1000); + stats.sliceUsage(0.03); stats.addMergeBufferUsedBytes(40); - stats.sliceUsage(40, 1000); + stats.sliceUsage(0.04); // Used bytes are summed across slices. Assertions.assertEquals(100L, stats.getMergeBufferUsedBytes()); - // Proximity is the fullest slice's fill fraction (40/1000), NOT the summed fraction (100/1000 = 0.1). + // Proximity is the fullest slice's fill ratio (0.04), NOT anything computed from the summed usage. Assertions.assertEquals(0.04, stats.getSpillProximity(), DELTA); } @Test - public void testSpillProximityClampsToOneAtSpillPoint() + public void testSpillProximityClampsToRange() { - GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - // A slice that reached its threshold exactly. - stats.sliceUsage(700, 700); - Assertions.assertEquals(1.0, stats.getSpillProximity(), DELTA); + // A slice at exactly the spill point. + GroupByStatsProvider.PerQueryStats atSpill = new GroupByStatsProvider.PerQueryStats(); + atSpill.sliceUsage(1.0); + Assertions.assertEquals(1.0, atSpill.getSpillProximity(), DELTA); - // The offset-list term can push raw used bytes slightly above the threshold; proximity must clamp at 1.0. + // Defensive clamping: a caller that somehow passes >1.0 must not produce >1.0. GroupByStatsProvider.PerQueryStats over = new GroupByStatsProvider.PerQueryStats(); - over.sliceUsage(750, 700); + over.sliceUsage(1.5); Assertions.assertEquals(1.0, over.getSpillProximity(), DELTA); + + // Defensive clamping on the low end: a negative ratio is treated as 0.0 (never contributes to the max). + GroupByStatsProvider.PerQueryStats neg = new GroupByStatsProvider.PerQueryStats(); + neg.sliceUsage(-0.25); + Assertions.assertEquals(0.0, neg.getSpillProximity(), DELTA); + + // NaN is ignored entirely so a never-initialized grouper does not corrupt the accumulator. + GroupByStatsProvider.PerQueryStats nan = new GroupByStatsProvider.PerQueryStats(); + nan.sliceUsage(Double.NaN); + Assertions.assertEquals(0.0, nan.getSpillProximity(), DELTA); } @Test public void testSpillProximityZeroWhenNoSliceUsageRecorded() { GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - // No sliceUsage() call: threshold is 0, so proximity is 0.0 rather than dividing by zero. + // No sliceUsage() call: proximity stays at its initial 0.0. stats.addMergeBufferUsedBytes(500); Assertions.assertEquals(0.0, stats.getSpillProximity(), DELTA); } @@ -203,11 +213,10 @@ public void testSpillProximityZeroWhenNoSliceUsageRecorded() public void testSpillProximityPicksFullestSliceWhenSlicesDiffer() { GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - // Slices sharing a threshold but with differing usage; proximity is the fullest slice's ratio. - stats.sliceUsage(200, 1000); - stats.sliceUsage(900, 1000); - stats.sliceUsage(100, 1000); - // Fullest slice is 900/1000 = 0.9. + // Three slices with differing fill; proximity is the fullest. + stats.sliceUsage(0.2); + stats.sliceUsage(0.9); + stats.sliceUsage(0.1); Assertions.assertEquals(0.9, stats.getSpillProximity(), DELTA); } @@ -215,14 +224,12 @@ public void testSpillProximityPicksFullestSliceWhenSlicesDiffer() public void testSpillProximityKeepsPerSliceRatioWhenThresholdsDiffer() { // A single query can pass one PerQueryStats through both small sliced groupers (from a ConcurrentGrouper) and a - // full-buffer SpillingGrouper (subtotal/nested processing). A small slice can saturate its own tiny threshold - // while a much larger full-buffer grouper stays lightly filled. Proximity must pair each slice's used bytes with - // its OWN threshold, so the saturated small slice reports 1.0 and is not diluted by the large threshold. + // full-buffer SpillingGrouper (subtotal/nested processing). A small slice can saturate (proximity 1.0) while a much + // larger full-buffer grouper stays lightly filled (proximity ~0.005). Because sliceUsage records the ratio directly, + // the saturated slice's 1.0 is preserved verbatim — there is no shared byte threshold to dilute it. GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - stats.sliceUsage(1000, 1000); // small sliced grouper at its spill point -> 1.0 - stats.sliceUsage(5000, 1_000_000); // large full-buffer grouper barely filled -> 0.005 - // If used bytes and thresholds were maxed independently, this would report 5000/1_000_000 = 0.005 and hide the - // spill. Pairing per slice keeps the true max of 1.0. + stats.sliceUsage(1.0); // small sliced grouper at its spill point + stats.sliceUsage(0.005); // large full-buffer grouper barely filled Assertions.assertEquals(1.0, stats.getSpillProximity(), DELTA); } @@ -280,17 +287,17 @@ public void testAggregateStatsTakesMaxSpillProximityAcrossQueries() GroupByStatsProvider.PerQueryStats low = new GroupByStatsProvider.PerQueryStats(); low.mergeBufferAcquisitionTime(10); - low.sliceUsage(300, 1000); // 0.3 + low.sliceUsage(0.3); agg.addQueryStats(low); GroupByStatsProvider.PerQueryStats high = new GroupByStatsProvider.PerQueryStats(); high.mergeBufferAcquisitionTime(10); - high.sliceUsage(800, 1000); // 0.8 + high.sliceUsage(0.8); agg.addQueryStats(high); GroupByStatsProvider.PerQueryStats mid = new GroupByStatsProvider.PerQueryStats(); mid.mergeBufferAcquisitionTime(10); - mid.sliceUsage(500, 1000); // 0.5 + mid.sliceUsage(0.5); agg.addQueryStats(mid); Assertions.assertEquals(0.8, agg.getMaxSpillProximity(), DELTA); @@ -304,7 +311,7 @@ public void testSpillProximityDroppedWhenNoAcquisitionTimeRecorded() // GroupByResourcesReservationPool.reserve() before any grouper initializes. GroupByStatsProvider.AggregateStats agg = new GroupByStatsProvider.AggregateStats(); GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); - stats.sliceUsage(900, 1000); + stats.sliceUsage(0.9); agg.addQueryStats(stats); Assertions.assertEquals(0L, agg.getMergeBufferQueries()); @@ -314,9 +321,9 @@ public void testSpillProximityDroppedWhenNoAcquisitionTimeRecorded() /** * End-to-end through {@link GroupByStatsProvider} reproducing the user's scenario: a 125MiB merge buffer divided * into 240 per-thread slices (sliceSize ~= 546KiB). Each slice fills well below the configured buffer size, yet the - * fullest slice reaches its spill threshold (sliceSize * 0.7) and the query spills. The summed {@code bytesUsed} is - * far below {@code sizeBytes}, which is exactly why comparing it to {@code sizeBytes} was misleading; {@code - * maxSpillProximity} instead reports ~1.0, correctly indicating the query was at the spill point. + * fullest slice reaches its spill trigger (peak size/regrowthThreshold == 1.0) and the query spills. The summed + * {@code bytesUsed} is far below {@code sizeBytes}, which is exactly why comparing it to {@code sizeBytes} was + * misleading; {@code maxSpillProximity} instead reports 1.0, correctly indicating the query was at the spill point. */ @Test public void testEndToEndSlicedBufferSpillScenario() @@ -324,8 +331,6 @@ public void testEndToEndSlicedBufferSpillScenario() final long sizeBytes = 125L * 1024 * 1024; // druid.processing.buffer.sizeBytes (125MiB) final int numThreads = 240; // concurrencyHint / numThreads final long sliceSize = sizeBytes / numThreads; // per-slice capacity (~546KiB) - final float loadFactor = 0.7f; // BufferHashGrouper.DEFAULT_MAX_LOAD_FACTOR - final long sliceThreshold = (long) (sliceSize * loadFactor); GroupByStatsProvider statsProvider = new GroupByStatsProvider(); QueryResourceId id = new QueryResourceId("spilly"); @@ -334,10 +339,11 @@ public void testEndToEndSlicedBufferSpillScenario() stats.mergeBufferAcquisitionTime(42); long expectedUsed = 0; for (int i = 0; i < numThreads; i++) { - // Most slices stay light; one slice (i == 0) reaches its threshold and triggers the spill. - final long sliceUsed = (i == 0) ? sliceThreshold : sliceThreshold / 10; + // Most slices stay light; one slice (i == 0) reaches its spill trigger (proximity 1.0). + final long sliceUsed = (i == 0) ? sliceSize / 2 : sliceSize / 20; + final double sliceProximity = (i == 0) ? 1.0 : 0.1; stats.addMergeBufferUsedBytes(sliceUsed); - stats.sliceUsage(sliceUsed, sliceThreshold); + stats.sliceUsage(sliceProximity); expectedUsed += sliceUsed; } stats.spilledBytes(1_000_000L); @@ -345,7 +351,7 @@ public void testEndToEndSlicedBufferSpillScenario() statsProvider.closeQuery(id); GroupByStatsProvider.AggregateStats aggregateStats = statsProvider.getStatsSince(); - // The fullest slice is at its threshold, so proximity is 1.0: the query spilled. + // The fullest slice hit its spill trigger, so proximity is exactly 1.0. Assertions.assertEquals(1.0, aggregateStats.getMaxSpillProximity(), DELTA); // ...even though the summed usage across slices is a tiny fraction of the configured buffer size. Assertions.assertEquals(expectedUsed, aggregateStats.getTotalMergeBufferUsedBytes()); diff --git a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouperTest.java b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouperTest.java index 9a491afe840a..4d8ac80f3405 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouperTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/BufferHashGrouperTest.java @@ -202,6 +202,180 @@ public void testMaxMergeBufferUsedBytes() grouper.close(); } + @Test + public void testMaxSpillProximityAtSpillTrigger() + { + // A tiny fixed-size table (maxSizeForTesting=1) forces a spill trigger on the 2nd distinct key: no bucket can be + // allocated even after growth attempts, so findBucketWithAutoGrowth returns -1 and pins proximity to exactly 1.0. + // The invariant we care about: 1.0 corresponds to the real spill point, and is independent of bucket width / + // offset-list overhead / integer truncation. Contrast with the byte-based numerator, which topped out below 1.0. + final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); + columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 1L))); + final BufferHashGrouper grouper = new BufferHashGrouper<>( + Suppliers.ofInstance(ByteBuffer.allocate(1000)), + GrouperTestUtil.intKeySerde(), + AggregatorAdapters.factorizeBuffered( + columnSelectorFactory, + ImmutableList.of( + new LongSumAggregatorFactory("valueSum", "value"), + new CountAggregatorFactory("count") + ) + ), + /* bufferGrouperMaxSize */ 1, + 0, + 0, + true + ); + grouper.init(); + + // Before any aggregation, proximity is 0.0 (empty table). + Assertions.assertEquals(0.0, grouper.getMaxSpillProximity(), 0.0); + + // First key fits. Proximity is still strictly below 1.0 (size < regrowthThreshold after growth). + Assertions.assertTrue(grouper.aggregate(new IntKey(1)).isOk()); + Assertions.assertTrue( + grouper.getMaxSpillProximity() < 1.0, + "proximity should stay below 1.0 while the table can still accept more keys: " + grouper.getMaxSpillProximity() + ); + + // Second key triggers a spill (findBucketWithAutoGrowth returns -1). Proximity is pinned to exactly 1.0. + Assertions.assertFalse(grouper.aggregate(new IntKey(2)).isOk()); + Assertions.assertEquals(1.0, grouper.getMaxSpillProximity(), 0.0); + + // Reset preserves the peak: the grouper spilled at some point in its life. + grouper.reset(); + Assertions.assertEquals(1.0, grouper.getMaxSpillProximity(), 0.0); + + grouper.close(); + } + + @Test + public void testMaxSpillProximityBelowOneWhenNoSpill() + { + // A generously-sized table that never rejects a bucket. Proximity should be strictly below 1.0 for the entire + // aggregation. This is the "operator reads <1.0, so no spill happened" invariant. + final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); + columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 1L))); + final BufferHashGrouper grouper = new BufferHashGrouper<>( + Suppliers.ofInstance(ByteBuffer.allocate(10_000)), + GrouperTestUtil.intKeySerde(), + AggregatorAdapters.factorizeBuffered( + columnSelectorFactory, + ImmutableList.of( + new LongSumAggregatorFactory("valueSum", "value"), + new CountAggregatorFactory("count") + ) + ), + Integer.MAX_VALUE, + 0, + 0, + true + ); + grouper.init(); + + for (int i = 0; i < 20; i++) { + Assertions.assertTrue(grouper.aggregate(new IntKey(i)).isOk()); + Assertions.assertTrue( + grouper.getMaxSpillProximity() < 1.0, + "no spill occurred; proximity must remain strictly < 1.0: " + grouper.getMaxSpillProximity() + ); + } + + grouper.close(); + } + + @Test + public void testMaxSpillProximityAtTerminalThresholdWithoutRejection() + { + // Adversarial "at exactly the spill point, but no further insert attempted" case. Fill the grouper until one more + // insert would trigger a rejection, then stop calling aggregate. Because the table is at its terminal growth level + // (arena exhausted, no room to enlarge regrowthThreshold), size == regrowthThreshold means the next insert would + // fail — that's the spill point. Proximity must be exactly 1.0, not (T-1)/T. The base-class updateMax records 1.0 + // at the terminal level via isTerminalTableLevel(). + final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); + columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 1L))); + final BufferHashGrouper grouper = new BufferHashGrouper<>( + Suppliers.ofInstance(ByteBuffer.allocate(10_000)), + GrouperTestUtil.intKeySerde(), + AggregatorAdapters.factorizeBuffered( + columnSelectorFactory, + ImmutableList.of( + new LongSumAggregatorFactory("valueSum", "value"), + new CountAggregatorFactory("count") + ) + ), + Integer.MAX_VALUE, + 0.75f, + 4, + true + ); + grouper.init(); + + // Fill until the first rejection. + int inserted = 0; + while (inserted < 10_000 && grouper.aggregate(new IntKey(inserted)).isOk()) { + inserted++; + } + // A rejection occurred, so the grouper is definitively at its spill trigger. + Assertions.assertEquals(1.0, grouper.getMaxSpillProximity(), 0.0); + + // The stricter Case #5 check: rebuild and STOP one insert before the rejection. size == regrowthThreshold, but + // aggregate() was never called after that, so findBucketWithAutoGrowth was never invoked with a rejection. Still, + // being at the terminal level means proximity must be 1.0. + grouper.close(); + + final BufferHashGrouper parked = new BufferHashGrouper<>( + Suppliers.ofInstance(ByteBuffer.allocate(10_000)), + GrouperTestUtil.intKeySerde(), + AggregatorAdapters.factorizeBuffered( + columnSelectorFactory, + ImmutableList.of( + new LongSumAggregatorFactory("valueSum", "value"), + new CountAggregatorFactory("count") + ) + ), + Integer.MAX_VALUE, + 0.75f, + 4, + true + ); + parked.init(); + for (int i = 0; i < inserted; i++) { + Assertions.assertTrue(parked.aggregate(new IntKey(i)).isOk()); + } + // Confirm the arithmetic: `inserted` == regrowthThreshold_final, so after `inserted` successful inserts the parked + // grouper's size sits exactly at getMaxSize() (i.e. regrowthThreshold_final). The next new-key aggregate() would + // find no bucket and return not-ok — this IS the spill point. + Assertions.assertEquals(inserted, parked.getSize()); + Assertions.assertEquals(inserted, parked.getMaxSize()); + // No further aggregate call. Grouper is parked exactly at its terminal threshold; the next insert WOULD spill. + Assertions.assertEquals(1.0, parked.getMaxSpillProximity(), 0.0); + parked.close(); + } + + @Test + public void testMaxSpillProximityBeforeInitIsZero() + { + // Grouper never initialized: no hash table has been created, so proximity is 0.0 rather than NaN or a crash. + final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); + final BufferHashGrouper grouper = new BufferHashGrouper<>( + Suppliers.ofInstance(ByteBuffer.allocate(1000)), + GrouperTestUtil.intKeySerde(), + AggregatorAdapters.factorizeBuffered( + columnSelectorFactory, + ImmutableList.of( + new LongSumAggregatorFactory("valueSum", "value"), + new CountAggregatorFactory("count") + ) + ), + Integer.MAX_VALUE, + 0, + 0, + true + ); + Assertions.assertEquals(0.0, grouper.getMaxSpillProximity(), 0.0); + } + private ResourceHolder> makeGrouper( GroupByTestColumnSelectorFactory columnSelectorFactory, int bufferSize, diff --git a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java index df9851ba829b..c187a87a7add 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/LimitedBufferHashGrouperTest.java @@ -164,6 +164,37 @@ public void testMinBufferSize() Assert.assertEquals(expected, entriesToList(grouper.iterator(true))); } + @Test + public void testMaxSpillProximityStaysBelowOneOnHeapTrimSwaps() + { + // A LimitedBufferHashGrouper's Alternating hash table swaps sub-buffers on every "full" event and trims to `limit` + // entries — this is a heap trim for limit push-down, NOT a disk spill. size hits regrowthThreshold on every swap + // (see testMinBufferSize: size == getMaxSize() == 101 at steady state after 899 swaps). A raw + // "tableStart == 0" pin in the base class would falsely mark this as a spill; the isTerminalTableLevel() override + // in AlternatingByteBufferHashTable returns false to prevent that. This test locks in that invariant: a Limited + // grouper that trims but never spills must report strictly < 1.0. + final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); + final LimitedBufferHashGrouper grouper = makeGrouper(columnSelectorFactory, 12120); + + columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 10L))); + for (int i = 0; i < NUM_ROWS; i++) { + Assert.assertTrue(String.valueOf(i + KEY_BASE), grouper.aggregate(new IntKey(i + KEY_BASE)).isOk()); + } + + // Confirms this exercised the heap-trim swap path (matches testMinBufferSize's observations). + Assert.assertTrue("expected multiple heap-trim swaps: " + grouper.getGrowthCount(), grouper.getGrowthCount() > 100); + Assert.assertEquals(101, grouper.getMaxSize()); + Assert.assertEquals(101, grouper.getSize()); + + // Proximity must be < 1.0 despite size == regrowthThreshold at every swap. The upper bound is (T-1)/T = 100/101 + // ≈ 0.9901 (heap trims land at numCopied == limit == 100, then the next new bucket brings size to 101 which is + // regrowthThreshold — but that hit is NOT recorded because isTerminalTableLevel() is false for Alternating). + Assert.assertTrue( + "heap-trim swap must not report 1.0: " + grouper.getMaxSpillProximity(), + grouper.getMaxSpillProximity() < 1.0 + ); + } + @Test public void testAggregateAfterIterated() { diff --git a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java index 3ec121568234..67e73ad7caf5 100644 --- a/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java +++ b/processing/src/test/java/org/apache/druid/query/groupby/epinephelinae/SpillingGrouperTest.java @@ -372,6 +372,116 @@ public void testMaxSpillFileCount() throws IOException } } + @Test + public void testSpillProximityReflectsSliceFillOnClose() throws IOException + { + // Fill a grouper part-way (no spill) and confirm close() records a proximity strictly in (0, 1). The proximity is + // the underlying hash table's peak size/regrowthThreshold, so a lightly-filled slice is well below 1.0. + final GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + final int bufferSize = 100_000; + final SpillingGrouper grouper = makeGrouper( + bufferSize, + new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024, 100, new GroupByStatsProvider.PerQueryStats()), + 1024 * 1024L, + stats, + true + ); + + // A handful of distinct keys: some buckets used, well short of the spill threshold. + for (int i = 0; i < 10; i++) { + Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk()); + } + grouper.close(); + + final double proximity = stats.getSpillProximity(); + Assert.assertTrue("proximity should be positive after aggregating: " + proximity, proximity > 0.0); + Assert.assertTrue("a lightly-filled slice should be well below the spill point: " + proximity, proximity < 1.0); + } + + @Test + public void testSpillProximityNotRecordedWhenGrouperNeverInitialized() throws IOException + { + // A grouper that never initialized (never touched the merge buffer) must not contribute to spill proximity, per the + // isInitialized() gate in close(). Otherwise idle slices would report a spurious 0-of-threshold data point. + final GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + final SpillingGrouper grouper = makeGrouper( + 100_000, + new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024, 100, new GroupByStatsProvider.PerQueryStats()), + 1024 * 1024L, + stats, + false // do not init + ); + grouper.close(); + + // No sliceUsage() call was made, so proximity stays at its initial 0.0. + Assert.assertEquals(0.0, stats.getSpillProximity(), 1e-9); + } + + @Test + public void testSpillProximityStaysOneAfterSpillThenLightRefill() throws IOException + { + // Case #4: force a real spill, then aggregate a few more keys that don't refill the table, then close. The + // underlying hash table gets reset() by spill() (size returns to 0, regrowthThreshold shrinks to its initial small + // value), so the ratio in isolation at close time would be tiny. What must save us is peak preservation: the 1.0 + // pinned inside findBucketWithAutoGrowth (or at the terminal threshold) before the reset survives it. Assert the + // reported proximity is exactly 1.0 despite the low post-spill fill. + final GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + final int bufferSize = 50; + final SpillingGrouper grouper = makeGrouper( + bufferSize, + new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024, 100, new GroupByStatsProvider.PerQueryStats()), + 1024 * 1024L, + stats, + true + ); + + // Enough keys to trigger multiple spills. + for (int i = 0; i < 50; i++) { + Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk()); + } + // Just a couple more keys — table has been reset() by the last spill so it's lightly filled at close. + Assert.assertTrue(grouper.aggregate(new IntKey(9001)).isOk()); + Assert.assertTrue(grouper.aggregate(new IntKey(9002)).isOk()); + grouper.close(); + + Assert.assertEquals( + "spilled slice with post-spill light refill must still report 1.0 (peak preserved across reset)", + 1.0, + stats.getSpillProximity(), + 0.0 + ); + } + + @Test + public void testSpillProximityExactlyOneWhenSliceSpills() throws IOException + { + // A tiny 50-byte buffer with 100 unique keys forces the underlying BufferHashGrouper to reject a bucket allocation + // (findBucketWithAutoGrowth returns -1), which is the real spill trigger. SpillingGrouper.aggregate then invokes + // spill(); the peak size/regrowthThreshold at that instant is pinned to exactly 1.0 and preserved across the + // subsequent grouper.reset(). This is the "1.0 <=> actually spilled" invariant. + final GroupByStatsProvider.PerQueryStats stats = new GroupByStatsProvider.PerQueryStats(); + final int bufferSize = 50; + final SpillingGrouper grouper = makeGrouper( + bufferSize, + new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024, 100, new GroupByStatsProvider.PerQueryStats()), + 1024 * 1024L, + stats, + true + ); + + for (int i = 0; i < 100; i++) { + Assert.assertTrue(grouper.aggregate(new IntKey(i)).isOk()); + } + grouper.close(); + + Assert.assertEquals( + "a slice that reached its spill trigger must report proximity == 1.0 exactly", + 1.0, + stats.getSpillProximity(), + 0.0 + ); + } + private SpillingGrouper makeGrouper( int bufferSize, File storageDir, @@ -410,6 +520,17 @@ private SpillingGrouper makeGrouper( LimitedTemporaryStorage temporaryStorage, long minSpillFileSize ) + { + return makeGrouper(bufferSize, temporaryStorage, minSpillFileSize, new GroupByStatsProvider.PerQueryStats(), true); + } + + private SpillingGrouper makeGrouper( + int bufferSize, + LimitedTemporaryStorage temporaryStorage, + long minSpillFileSize, + GroupByStatsProvider.PerQueryStats perQueryStats, + boolean init + ) { final GroupByTestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory(); columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 1L))); @@ -429,9 +550,11 @@ private SpillingGrouper makeGrouper( false, bufferSize, minSpillFileSize, - new GroupByStatsProvider.PerQueryStats() + perQueryStats ); - grouper.init(); + if (init) { + grouper.init(); + } return grouper; } diff --git a/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java b/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java index 8d02e64e3b49..df6e4b6771bd 100644 --- a/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java +++ b/server/src/test/java/org/apache/druid/server/metrics/GroupByStatsMonitorTest.java @@ -206,7 +206,7 @@ public void testMonitoringWithMultipleResources() GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); stats1.mergeBufferAcquisitionTime(100); stats1.addMergeBufferUsedBytes(50); - stats1.sliceUsage(50, 1000); // 0.05 + stats1.sliceUsage(0.05); stats1.spilledBytes(200); stats1.dictionarySize(100); @@ -214,7 +214,7 @@ public void testMonitoringWithMultipleResources() GroupByStatsProvider.PerQueryStats stats2 = statsProvider.getPerQueryStatsContainer(r2); stats2.mergeBufferAcquisitionTime(500); stats2.addMergeBufferUsedBytes(30); - stats2.sliceUsage(30, 2000); // 0.015 + stats2.sliceUsage(0.015); stats2.spilledBytes(100); stats2.dictionarySize(300); @@ -222,7 +222,7 @@ public void testMonitoringWithMultipleResources() GroupByStatsProvider.PerQueryStats stats3 = statsProvider.getPerQueryStatsContainer(r3); stats3.mergeBufferAcquisitionTime(200); stats3.addMergeBufferUsedBytes(150); - stats3.sliceUsage(150, 1500); // 0.1 + stats3.sliceUsage(0.1); stats3.spilledBytes(800); stats3.dictionarySize(200);