This is an automated email from the ASF dual-hosted git repository.
aho135 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 6474032ba78 feat: Add mergeBuffer/maxSpillProximity metric for groupBy
spill diagnosis (#19627)
6474032ba78 is described below
commit 6474032ba7820f285fa7141ff9a5d93b35aa1016
Author: aho135 <[email protected]>
AuthorDate: Tue Aug 25 13:14:08 2026 -0700
feat: Add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis
(#19627)
* feat: add mergeBuffer/maxSpillProximity metric for groupBy spill diagnosis
ConcurrentGrouper divides the acquired merge buffer into
druid.processing.numThreads
slices; a groupBy spills as soon as its fullest single slice fills, which
can be far
below druid.processing.buffer.sizeBytes. The existing merge-buffer byte
metrics are
per-query sums discounted by the load factor, so they never approach
sizeBytes even
while queries spill and can't be used to reason about spill pressure.
Add mergeBuffer/maxSpillProximity, a gauge in [0.0, 1.0]: the max across
queries of the
max across a query's slices of that slice's peak
size/terminalRegrowthThreshold. It is
bucket-count based (the exact condition a BufferHashGrouper spills on), so
1.0
corresponds exactly to the spill trigger, independent of bucket width,
offset-list
overhead, and integer truncation.
- ByteBufferHashTable tracks the lifetime-max ratio, pinned to 1.0 when a
bucket
allocation is rejected in findBucketWithAutoGrowth and preserved across
reset(). The
denominator is the FIXED terminal-level regrowthThreshold
(computeSpillRegrowthThreshold()
replays the arena geometry, memoized), so the ratio is monotonic in size
and rises
smoothly to 1.0 rather than sawtoothing near (N-1)/N on every table
doubling.
- A recordsFillProximity() hook lets the alternating limit-push-down table
opt out of
fill recording: it heap-trims (never disk-spills) and would otherwise
saturate near
(T-1)/T; it now reads 0.0, and only a genuine bucket rejection reports
1.0.
- GroupByStatsProvider keeps the per-slice max (clamped to [0,1], NaN
ignored),
aggregated as a max across queries; SpillingGrouper reports each slice's
peak proximity
in close(); GroupByStatsMonitor emits the metric;
docs/operations/metrics.md documents
it and clarifies the slicing semantics of mergeBuffer/bytesUsed and
maxBytesUsed.
* refactor: share reset()/adjustTableWhenFull() geometry with
computeSpillRegrowthThreshold()
computeSpillRegrowthThreshold() replayed the arena-geometry math of reset()
(initial placement) and adjustTableWhenFull() (one growth step). Extract
both
pure decisions into shared private helpers so the geometry lives in one
place:
- initialTableStart(maxBuckets): the initial-placement loop, used by reset()
(which then slices the buffer) and by the replay.
- nextGrowthLevel(tableStart, buckets) -> GrowthLevel: the grow-up-or-wrap
decision, used by adjustTableWhenFull() (which then allocates + rehashes)
and
by the replay.
computeSpillRegrowthThreshold() now seeds with initialTableStart() and loops
nextGrowthLevel() to the terminal level, dropping its duplicated while(true)
placement loop. Behavior is unchanged (geometry asserted by
testLimitAndBufferSwapping / testMinBufferSize / BufferHashGrouperTest).
Also note that computeSpillRegrowthThreshold() is valid only for the
standard
grow-by-doubling layout and must not be reached by the alternating
limit-pushdown
table (guaranteed by recordsFillProximity() == false); a future fixed-layout
variant must override it rather than rely on the replay.
* docs: tighten verbose maxSpillProximity code comments
Consolidate the terminal-denominator/sawtooth rationale to a single place
(the
maxSpillProximity field) and make the method Javadocs
(updateMaxMergeBufferUsedBytes,
computeSpillRegrowthThreshold, recordsFillProximity, getMaxSpillProximity,
the
geometry helpers, and the GroupByStatsProvider/SpillingGrouper comments)
terse,
removing duplicated explanations. No logic changes; also fixes a few stale
'size / regrowthThreshold' references to reflect the terminal denominator.
* Register mergeBuffer/maxSpillProximity in the default Prometheus mapping
Per review: the emitter emits mergeBuffer/maxSpillProximity, but with the
default
dimensionMapPath the Prometheus emitter only exports metrics listed in
defaultMetrics.json, so the event was silently dropped for the default
config.
Add it as a no-dimension gauge and cover the mapping in MetricsTest.
* Make spill-regrowth-threshold helpers private
Per review: getSpillRegrowthThreshold() and computeSpillRegrowthThreshold()
are
only used within ByteBufferHashTable (no subclass overrides them — the
alternating
table opts out via recordsFillProximity()), so scope them private. Also
drop the
now-inapplicable 'must override it' note from the Javadoc.
* Simplify spill-proximity clamp with Math.min
Per review nit: use Math.min for the upper bound instead of a nested
ternary.
* Rename PerQueryStats.sliceUsage to spillProximity
Per review: sliceUsage(double) read like per-slice byte usage (it sits next
to
addMergeBufferUsedBytes). Rename to spillProximity(double), pairing with the
existing getSpillProximity() getter and matching the noun-style accumulators
(spilledBytes, dictionarySize). No behavior change.
* Fold merge-buffer stats reporting in SpillingGrouper.close() under one
isInitialized() check
Per review: getMaxMergeBufferUsedBytes() already guarded on
grouper.isInitialized(),
duplicating the check used to gate spill-proximity reporting. Report both
merge-buffer
usage and spill proximity under a single isInitialized() block and drop the
now
single-use private helper. Behavior-preserving: the uninitialized path
previously did an
inert addMergeBufferUsedBytes(0L).
---
docs/operations/metrics.md | 15 +-
.../src/main/resources/defaultMetrics.json | 1 +
.../druid/emitter/prometheus/MetricsTest.java | 15 ++
.../druid/query/groupby/GroupByStatsProvider.java | 80 ++++++-
.../epinephelinae/AbstractBufferHashGrouper.java | 10 +
.../groupby/epinephelinae/ByteBufferHashTable.java | 201 +++++++++++++---
.../epinephelinae/LimitedBufferHashGrouper.java | 13 ++
.../groupby/epinephelinae/SpillingGrouper.java | 12 +-
.../query/groupby/GroupByStatsProviderTest.java | 243 ++++++++++++++++++-
.../epinephelinae/BufferHashGrouperTest.java | 256 +++++++++++++++++++++
.../LimitedBufferHashGrouperTest.java | 80 +++++++
.../groupby/epinephelinae/SpillingGrouperTest.java | 127 +++++++++-
.../druid/server/metrics/GroupByStatsMonitor.java | 1 +
.../server/metrics/GroupByStatsMonitorTest.java | 38 ++-
14 files changed, 1025 insertions(+), 67 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index c05f6cc91f9..07a4922945f 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -94,8 +94,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`.| [...]
+|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single
groupBy query within the emission period, where each query's usage is the sum
across the slices it held. Because the buffer is sliced among
`druid.processing.numThreads` query slices, this value is not directly
comparable to `druid.processing.buffer.sizeBytes`; use
`mergeBuffer/maxSpillProximity` to gauge spill pressure.|This metric is only
available if the `GroupByStatsMonitor` module is included.|Varies|
+|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to
spilling within the emission period, as a fraction in [0.0, 1.0]. A merge
buffer is divided into `druid.processing.numThreads` slices and a query spills
as soon as its fullest single slice's hash table cannot allocate another
bucket; this metric is that fullest slice's peak fill ratio (bucket count over
the load-factor bucket limit), taken as a max across slices and across queries.
**1.0 corresponds exactly to th [...]
|`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|
@@ -124,8 +125,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`.| [...]
+|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single
groupBy query within the emission period, where each query's usage is the sum
across the slices it held. Because the buffer is sliced among
`druid.processing.numThreads` query slices, this value is not directly
comparable to `druid.processing.buffer.sizeBytes`; use
`mergeBuffer/maxSpillProximity` to gauge spill pressure.|This metric is only
available if the `GroupByStatsMonitor` module is included.|Varies|
+|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to
spilling within the emission period, as a fraction in [0.0, 1.0]. A merge
buffer is divided into `druid.processing.numThreads` slices and a query spills
as soon as its fullest single slice's hash table cannot allocate another
bucket; this metric is that fullest slice's peak fill ratio (bucket count over
the load-factor bucket limit), taken as a max across slices and across queries.
**1.0 corresponds exactly to th [...]
|`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|
@@ -157,8 +159,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`. [...]
+|`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`, `t
[...]
+|`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 th [...]
|`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/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
index ae7a4900dff..e1af9b007bd 100644
---
a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
+++
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
@@ -53,6 +53,7 @@
"mergeBuffer/maxAcquisitionTimeNs": { "dimensions": [], "type": "timer",
"help": "Maximum time in nanoseconds to acquire merge buffer for any single
groupBy query within the emission period."},
"mergeBuffer/bytesUsed" : { "dimensions": [], "type": "gauge", "help":
"Total number of bytes used by merge buffers to process groupBy queries."},
"mergeBuffer/maxBytesUsed" : { "dimensions": [], "type": "gauge", "help":
"Maximum number of bytes used by merge buffers for any single groupBy query
within the emission period."},
+ "mergeBuffer/maxSpillProximity" : { "dimensions": [], "type": "gauge",
"help": "How close any single groupBy query came to spilling within the
emission period, in [0,1]; 1.0 corresponds to the spill trigger."},
"mergeBuffer/queries": { "dimensions": [], "type": "gauge", "help":
"Number of groupBy queries that acquired a batch of buffers from the merge
buffer pool."},
"groupBy/spilledQueries": { "dimensions": [], "type": "gauge", "help":
"Number of groupBy queries that have spilled onto the disk."},
"groupBy/maxSpilledBytes": { "dimensions": [], "type": "gauge", "help":
"Maximum number of bytes spilled to disk by any single groupBy query within the
emission period."},
diff --git
a/extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/MetricsTest.java
b/extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/MetricsTest.java
index 6dfaaec1ee2..58479a029a5 100644
---
a/extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/MetricsTest.java
+++
b/extensions-contrib/prometheus-emitter/src/test/java/org/apache/druid/emitter/prometheus/MetricsTest.java
@@ -19,6 +19,7 @@
package org.apache.druid.emitter.prometheus;
+import io.prometheus.client.Gauge;
import io.prometheus.client.Histogram;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.ISE;
@@ -102,6 +103,20 @@ public class MetricsTest
Assertions.assertTrue(actualMessage.contains(expectedMessage));
}
+ @Test
+ public void testMergeBufferMaxSpillProximityRegisteredAsGauge()
+ {
+ // mergeBuffer/maxSpillProximity must be in the default mapping (a
no-dimension gauge); otherwise the default
+ // Prometheus configuration would silently drop the metric.
+ PrometheusEmitterConfig config = new PrometheusEmitterConfig(null,
"test_spill", null, null, null, true, true, null, null, null, null);
+ Metrics metrics = new Metrics(config);
+ DimensionsAndCollector dimensionsAndCollector =
metrics.getByName("mergeBuffer/maxSpillProximity", "broker");
+ Assertions.assertNotNull(dimensionsAndCollector);
+ Assertions.assertTrue(dimensionsAndCollector.getCollector() instanceof
Gauge);
+ // No metric-specific dimensions, only the standard service/host labels.
+ Assertions.assertArrayEquals(new String[]{"druid_service", "host_name"},
dimensionsAndCollector.getDimensions());
+ }
+
@Test
public void testMetricsConfigurationWithNonExistentMetric()
{
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 f6b92a7b62c..62c19ff183e 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 org.apache.druid.query.QueryResourceId;
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
@@ -73,6 +74,7 @@ public class GroupByStatsProvider
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 +93,7 @@ public class GroupByStatsProvider
aggregateStats.maxMergeBufferAcquisitionTimeNs,
aggregateStats.totalMergeBufferUsedBytes,
aggregateStats.maxMergeBufferUsedBytes,
+ aggregateStats.maxSpillProximity,
aggregateStats.spilledQueries,
aggregateStats.spilledBytes,
aggregateStats.maxSpilledBytes,
@@ -105,6 +108,7 @@ public class GroupByStatsProvider
long maxMergeBufferAcquisitionTimeNs,
long totalMergeBufferUsedBytes,
long maxMergeBufferUsedBytes,
+ double maxSpillProximity,
long spilledQueries,
long spilledBytes,
long maxSpilledBytes,
@@ -117,6 +121,7 @@ public class GroupByStatsProvider
this.maxMergeBufferAcquisitionTimeNs = maxMergeBufferAcquisitionTimeNs;
this.totalMergeBufferUsedBytes = totalMergeBufferUsedBytes;
this.maxMergeBufferUsedBytes = maxMergeBufferUsedBytes;
+ this.maxSpillProximity = maxSpillProximity;
this.spilledQueries = spilledQueries;
this.spilledBytes = spilledBytes;
this.maxSpilledBytes = maxSpilledBytes;
@@ -149,6 +154,11 @@ public class GroupByStatsProvider
return maxMergeBufferUsedBytes;
}
+ public double getMaxSpillProximity()
+ {
+ return maxSpillProximity;
+ }
+
public long getSpilledQueries()
{
return spilledQueries;
@@ -174,6 +184,18 @@ public class GroupByStatsProvider
return maxMergeDictionarySize;
}
+ /**
+ * Folds a completed query's stats into the running aggregate. For
merge-buffer usage:
+ * <ul>
+ * <li>{@code totalMergeBufferUsedBytes} (emitted as {@code
mergeBuffer/bytesUsed}) sums each query's usage
+ * across all queries, where each query's usage is itself the sum
across the query's slices.</li>
+ * <li>{@code maxMergeBufferUsedBytes} (emitted as {@code
mergeBuffer/maxBytesUsed}) is the max such per-query
+ * summed usage across queries.</li>
+ * <li>{@code maxSpillProximity} (emitted as {@code
mergeBuffer/maxSpillProximity}) is the max across queries of
+ * each query's fullest-slice peak fill ratio — a per-slice MAX (not
a byte sum), so it reflects the slice
+ * that drives a spill; 1.0 iff a slice actually spilled.</li>
+ * </ul>
+ */
public void addQueryStats(PerQueryStats perQueryStats)
{
if (perQueryStats.getMergeBufferAcquisitionTimeNs() > 0) {
@@ -183,8 +205,9 @@ public class GroupByStatsProvider
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 class GroupByStatsProvider
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,22 @@ public class GroupByStatsProvider
public static class PerQueryStats
{
private final AtomicLong mergeBufferAcquisitionTimeNs = new AtomicLong(0);
- private final AtomicLong maxMergeBufferUsedBytes = new AtomicLong(0);
+ /**
+ * Sum of the peak merge-buffer usage of every grouper (slice) this query
held. A
+ * {@code ConcurrentGrouper} slices a single merge buffer into one slice
per processing thread, and each slice
+ * reports its own peak via
+ * {@link #addMergeBufferUsedBytes(long)} when closed, so the per-query
value is the SUM across the query's slices.
+ */
+ private final AtomicLong mergeBufferUsedBytes = new AtomicLong(0);
+ /**
+ * Spill proximity of the fullest slice this query held, in [0.0, 1.0].
Each {@link #spillProximity} call contributes
+ * one slice's peak fill ratio and this keeps the MAX across slices: a
query spills as soon as its hottest slice
+ * fills, so proximity is driven by that slice, not the byte sum in {@link
#mergeBufferUsedBytes}. Keeping each
+ * slice's ratio intact (rather than maxing numerator and denominator
separately) stays correct when a query mixes
+ * groupers of different sizes (e.g. {@code ConcurrentGrouper} slices
alongside a full-buffer grouper). 1.0 iff a
+ * slice actually spilled.
+ */
+ private final DoubleAccumulator maxSpillProximity = new
DoubleAccumulator(Math::max, 0.0);
private final AtomicLong spilledBytes = new AtomicLong(0);
private final AtomicLong mergeDictionarySize = new AtomicLong(0);
@@ -224,9 +263,26 @@ public class GroupByStatsProvider
mergeBufferAcquisitionTimeNs.addAndGet(delay);
}
- public void maxMergeBufferUsedBytes(long bytes)
+ /**
+ * Accumulates the peak merge-buffer usage of one grouper (slice). Despite
the previous "max" naming, this method
+ * sums across the slices a query holds; see {@link #mergeBufferUsedBytes}.
+ */
+ public void addMergeBufferUsedBytes(long bytes)
{
- maxMergeBufferUsedBytes.addAndGet(bytes);
+ mergeBufferUsedBytes.addAndGet(bytes);
+ }
+
+ /**
+ * Records one slice's peak fill ratio (1.0 iff it spilled), kept as a max
across the query's slices; see
+ * {@link #maxSpillProximity}. Clamped to [0, 1]; NaN is ignored so a
never-initialized grouper contributes nothing.
+ */
+ public void spillProximity(double proximity)
+ {
+ if (Double.isNaN(proximity)) {
+ return;
+ }
+ final double clamped = proximity < 0.0 ? 0.0 : Math.min(proximity, 1.0);
+ maxSpillProximity.accumulate(clamped);
}
public void spilledBytes(long bytes)
@@ -244,9 +300,19 @@ public class GroupByStatsProvider
return mergeBufferAcquisitionTimeNs.get();
}
- public long getMaxMergeBufferUsedBytes()
+ public long getMergeBufferUsedBytes()
+ {
+ return mergeBufferUsedBytes.get();
+ }
+
+ /**
+ * Spill proximity for this query in [0.0, 1.0]: the fullest slice's peak
{@code size / regrowthThreshold} over
+ * that slice's lifetime. 1.0 corresponds exactly to the spill trigger (a
bucket allocation was rejected). Returns
+ * 0.0 when no slice usage was recorded (e.g. a grouper that never
initialized).
+ */
+ public double getSpillProximity()
{
- return maxMergeBufferUsedBytes.get();
+ return maxSpillProximity.get();
}
public long getSpilledBytes()
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 a5edb38cfa4..726852a6ec1 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,16 @@ public abstract class AbstractBufferHashGrouper<KeyType>
implements Grouper<KeyT
return hashTable.getMaxMergeBufferUsedBytes();
}
+ /**
+ * Peak fraction of the way to a spill over this grouper's lifetime, in
[0.0, 1.0]; 1.0 means it actually spilled.
+ * The value {@code SpillingGrouper} reports for {@code
mergeBuffer/maxSpillProximity}. Preserved across
+ * {@link #reset()}; 0.0 if the grouper was never initialized. See {@link
ByteBufferHashTable#getMaxSpillProximity()}.
+ */
+ 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/ByteBufferHashTable.java
b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/ByteBufferHashTable.java
index 0f64d08613c..550bbee08b1 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,17 @@ public class ByteBufferHashTable
// Tracks maximum bytes used for the entire lifecycle of this hash table.
protected long maxMergeBufferUsedBytes;
+ // Peak {@code size / terminalRegrowthThreshold} over this table's lifetime,
in [0.0, 1.0]; the denominator is the
+ // fixed bucket-count ceiling of the final growth level (see {@link
#getSpillRegrowthThreshold()}). A fixed denominator
+ // over monotonically-growing size makes this rise smoothly to exactly 1.0
at the real spill trigger; dividing by the
+ // CURRENT level's threshold would instead sawtooth and pin near (N-1)/N
after any growth. Preserved across
+ // {@link #reset()} so a grouper that spilled still reads 1.0 after size
returns to 0.
+ protected double maxSpillProximity;
+
+ // Cached denominator for {@link #maxSpillProximity} (terminal-level
regrowthThreshold). Zero = not yet computed;
+ // {@link #maxSizeForBuckets} is always >= 1, so zero is a safe sentinel.
Depends only on final geometry, so cached.
+ private int spillRegrowthThreshold;
+
public ByteBufferHashTable(
float maxLoadFactor,
int initialBuckets,
@@ -109,6 +120,8 @@ public class ByteBufferHashTable
this.tableArenaSize = buffer.capacity();
this.bucketUpdateHandler = bucketUpdateHandler;
this.maxMergeBufferUsedBytes = 0;
+ this.maxSpillProximity = 0.0;
+ this.spillRegrowthThreshold = 0;
}
public void reset()
@@ -127,25 +140,7 @@ public class ByteBufferHashTable
}
// Start table part-way through the buffer so the last growth can start
from zero and thereby use more space.
- tableStart = tableArenaSize - maxBuckets * bucketSizeWithHash;
- int nextBuckets = maxBuckets * 2;
- while (true) {
- long nextBucketsSize = (long) nextBuckets * bucketSizeWithHash;
- if (nextBucketsSize > Integer.MAX_VALUE) {
- break;
- }
- final int nextTableStart = tableStart - nextBuckets * bucketSizeWithHash;
- if (nextTableStart > tableArenaSize / 2) {
- tableStart = nextTableStart;
- nextBuckets = nextBuckets * 2;
- } else {
- break;
- }
- }
-
- if (tableStart < tableArenaSize / 2) {
- tableStart = 0;
- }
+ tableStart = initialTableStart(maxBuckets);
final ByteBuffer bufferDup = buffer.duplicate();
bufferDup.position(tableStart);
@@ -166,20 +161,10 @@ public class ByteBufferHashTable
return;
}
- final int newBuckets;
- final int newMaxSize;
- final int newTableStart;
-
- if (((long) maxBuckets * 3 * bucketSizeWithHash) > (long) tableArenaSize -
tableStart) {
- // Not enough space to grow upwards, start back from zero
- newTableStart = 0;
- newBuckets = tableStart / bucketSizeWithHash;
- newMaxSize = maxSizeForBuckets(newBuckets);
- } else {
- newTableStart = tableStart + tableBuffer.limit();
- newBuckets = maxBuckets * 2;
- newMaxSize = maxSizeForBuckets(newBuckets);
- }
+ final GrowthLevel next = nextGrowthLevel(tableStart, maxBuckets);
+ final int newTableStart = next.tableStart;
+ final int newBuckets = next.buckets;
+ final int newMaxSize = maxSizeForBuckets(newBuckets);
if (newBuckets < maxBuckets) {
throw new ISE("newBuckets[%,d] < maxBuckets[%,d]", newBuckets,
maxBuckets);
@@ -290,6 +275,12 @@ public class ByteBufferHashTable
}
}
+ if (bucket < 0) {
+ // Spill trigger: no bucket even after attempting to grow. Pin proximity
to 1.0 (also covers a rejection before
+ // size reaches the terminal threshold, e.g. a full-probe wraparound).
See {@link #maxSpillProximity}.
+ maxSpillProximity = 1.0;
+ }
+
return bucket;
}
@@ -442,12 +433,143 @@ public class ByteBufferHashTable
}
/**
- * 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.
+ * Called whenever {@link #size} or {@link #bucketSizeWithHash} changes, to
track {@link #maxMergeBufferUsedBytes} and
+ * the {@link #maxSpillProximity} peak. Proximity is recorded while {@code
size < regrowthThreshold} (the transient hit
+ * at intermediate growth boundaries is skipped, since the table then
grows); at the terminal level, parking at the
+ * threshold is the spill point and pins 1.0. Trim-and-swap tables ({@link
#recordsFillProximity()} == false) skip
+ * proximity entirely — see {@link #findBucketWithAutoGrowth} for their only
spill signal.
*/
protected void updateMaxMergeBufferUsedBytes()
{
maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, (long) size *
bucketSizeWithHash);
+ if (!recordsFillProximity()) {
+ return;
+ }
+ final int denominator = getSpillRegrowthThreshold();
+ if (denominator <= 0) {
+ return;
+ }
+ if (size < regrowthThreshold) {
+ // size < regrowthThreshold <= terminal denominator keeps this below
1.0; the clamp is defensive.
+ final double ratio = Math.min(1.0, (double) size / denominator);
+ if (ratio > maxSpillProximity) {
+ maxSpillProximity = ratio;
+ }
+ } else if (isTerminalTableLevel()) {
+ // At the load-factor limit with no room to grow: parking here IS the
spill point.
+ maxSpillProximity = 1.0;
+ }
+ }
+
+ /**
+ * Denominator for {@link #maxSpillProximity}: the {@code regrowthThreshold}
at the terminal growth level, where
+ * {@link #findBucketWithAutoGrowth} can no longer allocate a bucket.
Computed once from fixed geometry and cached.
+ */
+ private int getSpillRegrowthThreshold()
+ {
+ if (spillRegrowthThreshold == 0) {
+ spillRegrowthThreshold = computeSpillRegrowthThreshold();
+ }
+ return spillRegrowthThreshold;
+ }
+
+ /**
+ * Replays the arena geometry to the terminal growth level — via the same
{@link #initialTableStart} /
+ * {@link #nextGrowthLevel} primitives {@link #reset()} and {@link
#adjustTableWhenFull()} use — and returns
+ * {@link #maxSizeForBuckets} of its bucket count. Allocates no buffers.
+ *
+ * <p>Valid only for the standard grow-by-doubling layout. Fixed-layout
variants (the alternating limit-pushdown
+ * table) never reach this — guaranteed by {@link #recordsFillProximity()}
== false — so one that ever needs a spill
+ * denominator would have to revisit this method.
+ */
+ private int computeSpillRegrowthThreshold()
+ {
+ int buckets = Math.min(tableArenaSize / bucketSizeWithHash,
initialBuckets);
+ int start = initialTableStart(buckets);
+ while (start != 0) {
+ final GrowthLevel next = nextGrowthLevel(start, buckets);
+ start = next.tableStart;
+ buckets = next.buckets;
+ }
+ return maxSizeForBuckets(buckets);
+ }
+
+ /**
+ * Placement of the initial (smallest) table: far enough from the front that
successive doublings stack above it and
+ * the final growth can wrap to offset 0. Pure geometry, shared by {@link
#reset()} and
+ * {@link #computeSpillRegrowthThreshold()}. Returns 0 when no upward
doublings fit.
+ */
+ private int initialTableStart(int initialMaxBuckets)
+ {
+ int start = tableArenaSize - initialMaxBuckets * bucketSizeWithHash;
+ int nextBuckets = initialMaxBuckets * 2;
+ while (true) {
+ final long nextBucketsSize = (long) nextBuckets * bucketSizeWithHash;
+ if (nextBucketsSize > Integer.MAX_VALUE) {
+ break;
+ }
+ final int nextTableStart = start - nextBuckets * bucketSizeWithHash;
+ if (nextTableStart > tableArenaSize / 2) {
+ start = nextTableStart;
+ nextBuckets = nextBuckets * 2;
+ } else {
+ break;
+ }
+ }
+ return start < tableArenaSize / 2 ? 0 : start;
+ }
+
+ /**
+ * The next growth level given the current start offset and bucket count —
the pure decision shared by
+ * {@link #adjustTableWhenFull()} and {@link
#computeSpillRegrowthThreshold()}. Doubles upward while the arena has room
+ * for the next 3x, else wraps to offset 0 as the final level ({@code
tableStart == 0}).
+ */
+ private GrowthLevel nextGrowthLevel(int curTableStart, int curBuckets)
+ {
+ if (((long) curBuckets * 3 * bucketSizeWithHash) > (long) tableArenaSize -
curTableStart) {
+ // Not enough space to grow upwards; start back from zero.
+ return new GrowthLevel(0, curTableStart / bucketSizeWithHash);
+ } else {
+ // curBuckets * bucketSizeWithHash is the current table's byte length
(== tableBuffer.limit()).
+ return new GrowthLevel(curTableStart + curBuckets * bucketSizeWithHash,
curBuckets * 2);
+ }
+ }
+
+ /**
+ * A growth level's placement: the table's start offset within the arena and
its bucket count. {@code tableStart == 0}
+ * marks the terminal (final) level.
+ */
+ private static final class GrowthLevel
+ {
+ private final int tableStart;
+ private final int buckets;
+
+ GrowthLevel(int tableStart, int buckets)
+ {
+ this.tableStart = tableStart;
+ this.buckets = buckets;
+ }
+ }
+
+ /**
+ * True when the table can't grow further ({@link #tableStart} at the front
of the arena), matching
+ * {@link #adjustTableWhenFull()}'s early return. Overridden by
trim-and-swap variants whose "table full" is not a
+ * spill.
+ */
+ protected boolean isTerminalTableLevel()
+ {
+ return tableStart == 0;
+ }
+
+ /**
+ * Whether {@code size / terminalRegrowthThreshold} is a meaningful
proximity-to-spill signal. True for the base
+ * grow-by-doubling table (spills when it can't grow). Overridden false by
trim-and-swap variants (the alternating
+ * limit-pushdown table), which never spill via fill and would otherwise
saturate near 1.0; for those, only a
+ * rejection in {@link #findBucketWithAutoGrowth} pins 1.0.
+ */
+ protected boolean recordsFillProximity()
+ {
+ return true;
}
public long getMaxMergeBufferUsedBytes()
@@ -455,6 +577,15 @@ public class ByteBufferHashTable
return maxMergeBufferUsedBytes;
}
+ /**
+ * Peak fraction of the way to a spill over this table's lifetime, in [0.0,
1.0]; equals 1.0 iff the table actually
+ * spilled. See {@link #maxSpillProximity} for how it is computed.
+ */
+ 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 2bb544d0cda..167b21ee77c 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,18 @@ public class LimitedBufferHashGrouper<KeyType> extends
AbstractBufferHashGrouper
tableBuffer = newTableBuffer;
growthCount++;
}
+
+ /**
+ * The alternating table trims-and-swaps to keep the top-{@code limit}
entries in memory and never spills via the
+ * fill path (see base {@link #recordsFillProximity()}), so suppress
fill-proximity recording — such queries read
+ * 0.0. Its only spill signal is a rejection in {@link
#findBucketWithAutoGrowth}, which does NOT fire on ordinary
+ * swaps: post-swap {@code size == numCopied <= limit}, and construction
guarantees
+ * {@code regrowthThreshold >= limit + 1}, so the next insert always finds
a bucket.
+ */
+ @Override
+ protected boolean recordsFillProximity()
+ {
+ 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 96e55907b21..602aa2e0e9f 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
@@ -249,7 +249,12 @@ public class SpillingGrouper<KeyType> implements
Grouper<KeyType>
public void close()
{
perQueryStats.dictionarySize(getDictionarySizeEstimate());
- perQueryStats.maxMergeBufferUsedBytes(getMaxMergeBufferUsedBytes());
+ if (grouper.isInitialized()) {
+ // Merge-buffer usage and spill proximity are only meaningful once the
grouper touched the buffer;
+ // an untouched slice contributes nothing.
+
perQueryStats.addMergeBufferUsedBytes(grouper.getMaxMergeBufferUsedBytes());
+ perQueryStats.spillProximity(grouper.getMaxSpillProximity());
+ }
// Record spilled bytes before deleteFiles() decrements bytesUsed in
temporaryStorage.
long spilledBytes = 0;
for (final File file : files) {
@@ -267,11 +272,6 @@ public class SpillingGrouper<KeyType> implements
Grouper<KeyType>
deleteFiles();
}
- private long getMaxMergeBufferUsedBytes()
- {
- return grouper.isInitialized() ? grouper.getMaxMergeBufferUsedBytes() : 0L;
- }
-
private long getDictionarySizeEstimate()
{
return keySerde.getDictionarySize();
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 dafd381668d..ff39474f67f 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 @@ import org.junit.jupiter.api.Test;
public class GroupByStatsProviderTest
{
+ private static final double DELTA = 1e-9;
+
@Test
public void testMetricCollection()
{
@@ -35,7 +37,12 @@ public class GroupByStatsProviderTest
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. Both slices report
+ // their own peak fill ratio (bucket-count based, in [0,1]); the fullest
(0.05) drives proximity.
+ stats1.addMergeBufferUsedBytes(50);
+ stats1.spillProximity(0.05);
+ stats1.addMergeBufferUsedBytes(30);
+ stats1.spillProximity(0.03);
stats1.spilledBytes(200);
stats1.spilledBytes(400);
stats1.dictionarySize(100);
@@ -46,7 +53,9 @@ public class GroupByStatsProviderTest
stats2.mergeBufferAcquisitionTime(500);
stats2.mergeBufferAcquisitionTime(600);
- stats2.maxMergeBufferUsedBytes(100);
+ // Single slice at 0.05.
+ stats2.addMergeBufferUsedBytes(100);
+ stats2.spillProximity(0.05);
stats2.spilledBytes(400);
stats2.spilledBytes(600);
stats2.dictionarySize(300);
@@ -58,6 +67,7 @@ public class GroupByStatsProviderTest
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 class GroupByStatsProviderTest
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 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());
Assertions.assertEquals(1000L, aggregateStats.getMaxSpilledBytes());
@@ -88,28 +103,32 @@ public class GroupByStatsProviderTest
QueryResourceId r1 = new QueryResourceId("r1");
GroupByStatsProvider.PerQueryStats stats1 =
statsProvider.getPerQueryStatsContainer(r1);
stats1.mergeBufferAcquisitionTime(2000);
- stats1.maxMergeBufferUsedBytes(50);
+ stats1.addMergeBufferUsedBytes(50);
+ stats1.spillProximity(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.spillProximity(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.spillProximity(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.spillProximity(0.075);
stats4.spilledBytes(200);
stats4.dictionarySize(1500);
@@ -122,6 +141,8 @@ public class GroupByStatsProviderTest
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,212 @@ public class GroupByStatsProviderTest
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 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.spillProximity(0.01);
+ stats.addMergeBufferUsedBytes(20);
+ stats.spillProximity(0.02);
+ stats.addMergeBufferUsedBytes(30);
+ stats.spillProximity(0.03);
+ stats.addMergeBufferUsedBytes(40);
+ stats.spillProximity(0.04);
+
+ // Used bytes are summed across slices.
+ Assertions.assertEquals(100L, stats.getMergeBufferUsedBytes());
+ // 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 testSpillProximityClampsToRange()
+ {
+ // A slice at exactly the spill point.
+ GroupByStatsProvider.PerQueryStats atSpill = new
GroupByStatsProvider.PerQueryStats();
+ atSpill.spillProximity(1.0);
+ Assertions.assertEquals(1.0, atSpill.getSpillProximity(), DELTA);
+
+ // Defensive clamping: a caller that somehow passes >1.0 must not produce
>1.0.
+ GroupByStatsProvider.PerQueryStats over = new
GroupByStatsProvider.PerQueryStats();
+ over.spillProximity(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.spillProximity(-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.spillProximity(Double.NaN);
+ Assertions.assertEquals(0.0, nan.getSpillProximity(), DELTA);
+ }
+
+ @Test
+ public void testSpillProximityZeroWhenNoSliceUsageRecorded()
+ {
+ GroupByStatsProvider.PerQueryStats stats = new
GroupByStatsProvider.PerQueryStats();
+ // No spillProximity() call: proximity stays at its initial 0.0.
+ stats.addMergeBufferUsedBytes(500);
+ Assertions.assertEquals(0.0, stats.getSpillProximity(), DELTA);
+ }
+
+ @Test
+ public void testSpillProximityPicksFullestSliceWhenSlicesDiffer()
+ {
+ GroupByStatsProvider.PerQueryStats stats = new
GroupByStatsProvider.PerQueryStats();
+ // Three slices with differing fill; proximity is the fullest.
+ stats.spillProximity(0.2);
+ stats.spillProximity(0.9);
+ stats.spillProximity(0.1);
+ 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 (proximity 1.0) while a much
+ // larger full-buffer grouper stays lightly filled (proximity ~0.005).
Because spillProximity 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.spillProximity(1.0); // small sliced grouper at its spill
point
+ stats.spillProximity(0.005); // large full-buffer grouper barely
filled
+ Assertions.assertEquals(1.0, 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.spillProximity(0.3);
+ agg.addQueryStats(low);
+
+ GroupByStatsProvider.PerQueryStats high = new
GroupByStatsProvider.PerQueryStats();
+ high.mergeBufferAcquisitionTime(10);
+ high.spillProximity(0.8);
+ agg.addQueryStats(high);
+
+ GroupByStatsProvider.PerQueryStats mid = new
GroupByStatsProvider.PerQueryStats();
+ mid.mergeBufferAcquisitionTime(10);
+ mid.spillProximity(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.spillProximity(0.9);
+ 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 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()
+ {
+ 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)
+
+ 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 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.spillProximity(sliceProximity);
+ expectedUsed += sliceUsed;
+ }
+ stats.spilledBytes(1_000_000L);
+
+ statsProvider.closeQuery(id);
+ GroupByStatsProvider.AggregateStats aggregateStats =
statsProvider.getStatsSince();
+
+ // 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());
+ 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/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 9a491afe840..1333f612703 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,262 @@ public class BufferHashGrouperTest extends
InitializedNullHandlingTest
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<IntKey> 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<IntKey> 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<IntKey> 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<IntKey> 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 testMaxSpillProximityIsProportionalNotSawtooth()
+ {
+ // Regression guard for the sawtooth denominator bug. The old metric
divided size by the CURRENT growth level's
+ // regrowthThreshold. Since the table grows by doubling,
size/currentRegrowthThreshold resets to ~0.5 on every
+ // doubling and climbs back to ~1.0 before the next — so a table that
grows even a couple of times pinned its
+ // lifetime max near (N-1)/N ~ 0.99, regardless of how much headroom
remained before the real spill. The fix
+ // divides by the FIXED terminal-level regrowthThreshold, so the value is
the true fraction of the way to the spill.
+ final GroupByTestColumnSelectorFactory columnSelectorFactory =
GrouperTestUtil.newColumnSelectorFactory();
+ columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value",
1L)));
+
+ // First, discover the terminal threshold T by filling an identical
grouper until it rejects.
+ final BufferHashGrouper<IntKey> probe = new BufferHashGrouper<>(
+ Suppliers.ofInstance(ByteBuffer.allocate(100_000)),
+ GrouperTestUtil.intKeySerde(),
+ AggregatorAdapters.factorizeBuffered(
+ columnSelectorFactory,
+ ImmutableList.of(
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ )
+ ),
+ Integer.MAX_VALUE,
+ 0.75f,
+ 4,
+ true
+ );
+ probe.init();
+ int terminalThreshold = 0;
+ while (terminalThreshold < 100_000 && probe.aggregate(new
IntKey(terminalThreshold)).isOk()) {
+ terminalThreshold++;
+ }
+ Assertions.assertEquals(1.0, probe.getMaxSpillProximity(), 0.0);
+ probe.close();
+
+ // Now fill a fresh grouper to only a quarter of the terminal threshold.
This is enough distinct keys to force
+ // several doublings (so the OLD metric would have pinned near ~0.99), but
the table is nowhere near spilling.
+ final BufferHashGrouper<IntKey> grouper = new BufferHashGrouper<>(
+ Suppliers.ofInstance(ByteBuffer.allocate(100_000)),
+ GrouperTestUtil.intKeySerde(),
+ AggregatorAdapters.factorizeBuffered(
+ columnSelectorFactory,
+ ImmutableList.of(
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ )
+ ),
+ Integer.MAX_VALUE,
+ 0.75f,
+ 4,
+ true
+ );
+ grouper.init();
+
+ final int inserted = terminalThreshold / 4;
+ for (int i = 0; i < inserted; i++) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+
+ // The table must actually have grown multiple times — otherwise this
wouldn't exercise the sawtooth path.
+ Assertions.assertTrue(
+ grouper.getGrowthCount() >= 2,
+ "expected multiple growths to exercise the sawtooth path, got " +
grouper.getGrowthCount()
+ );
+
+ // The value tracks size/terminalThreshold ~ 0.25, NOT the ~0.99 the old
current-level denominator produced.
+ final double proximity = grouper.getMaxSpillProximity();
+ final double expected = (double) inserted / terminalThreshold;
+ Assertions.assertEquals(
+ expected,
+ proximity,
+ 0.05,
+ "proximity should be proportional to fill fraction, was " + proximity
+ );
+ Assertions.assertTrue(
+ proximity < 0.5,
+ "proximity must be far below 1.0 for a quarter-full table, was " +
proximity
+ );
+
+ grouper.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<IntKey> 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<Grouper<IntKey>> 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 0833a8ee490..aa3884cbcce 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
@@ -370,6 +370,86 @@ public class LimitedBufferHashGrouperTest extends
InitializedNullHandlingTest
Assertions.assertEquals(expectedPeakUsage,
grouper.getMaxMergeBufferUsedBytes());
}
+ @Test
+ public void testMaxSpillProximityIsZeroOnHeapTrimSwaps()
+ {
+ // A LimitedBufferHashGrouper's Alternating hash table swaps sub-buffers
on every "full" event and trims to `limit`
+ // entries — 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). This path never spills, so its
+ // fill ratio is not proximity to a spill:
AlternatingByteBufferHashTable.recordsFillProximity() returns false,
+ // suppressing fill-proximity recording entirely. A Limited grouper that
trims but never spills must report exactly
+ // 0.0 — not the ~(T-1)/T saturation that the raw fill ratio would produce.
+ final GroupByTestColumnSelectorFactory columnSelectorFactory =
GrouperTestUtil.newColumnSelectorFactory();
+ final LimitedBufferHashGrouper<IntKey> grouper =
makeGrouper(columnSelectorFactory, 12120);
+
+ columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value",
10L)));
+ for (int i = 0; i < NUM_ROWS; i++) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(i +
KEY_BASE)).isOk(), String.valueOf(i + KEY_BASE));
+ }
+
+ // Confirms this exercised the heap-trim swap path (matches
testMinBufferSize's observations).
+ Assertions.assertTrue(
+ grouper.getGrowthCount() > 100,
+ "expected multiple heap-trim swaps: " + grouper.getGrowthCount()
+ );
+ Assertions.assertEquals(101, grouper.getMaxSize());
+ Assertions.assertEquals(101, grouper.getSize());
+
+ // Despite size == regrowthThreshold at every swap, proximity must be
exactly 0.0: heap-trim swaps are not a spill,
+ // and no bucket rejection ever occurred (post-swap size <= limit <
regrowthThreshold guarantees the next insert
+ // succeeds), so nothing sets proximity above its 0.0 initial value.
+ Assertions.assertEquals(
+ 0.0,
+ grouper.getMaxSpillProximity(),
+ 0.0,
+ "heap-trim swaps must not register any spill proximity: " +
grouper.getMaxSpillProximity()
+ );
+ }
+
+ @Test
+ public void testMaxSpillProximityIsOneOnGenuineRejection()
+ {
+ // The only real spill signal for the Alternating table is a genuine
bucket rejection in findBucketWithAutoGrowth.
+ // Force one by capping bufferGrouperMaxSize (maxSizeForTesting) below
regrowthThreshold: once size reaches the cap
+ // no new bucket is allowed AND no swap is attempted (size <
maxSizeForTesting is false), so findBucketWithAutoGrowth
+ // returns -1 and pins proximity to exactly 1.0 — even though
recordsFillProximity() zeroes the ordinary heap-trim
+ // path.
+ final GroupByTestColumnSelectorFactory columnSelectorFactory =
GrouperTestUtil.newColumnSelectorFactory();
+ final int maxSize = 50;
+ final LimitedBufferHashGrouper<IntKey> grouper = new
LimitedBufferHashGrouper<>(
+ Suppliers.ofInstance(ByteBuffer.allocate(12120)),
+ GrouperTestUtil.intKeySerde(),
+ AggregatorAdapters.factorizeBuffered(
+ columnSelectorFactory,
+ ImmutableList.of(
+ new LongSumAggregatorFactory("valueSum", "value"),
+ new CountAggregatorFactory("count")
+ )
+ ),
+ maxSize,
+ 0.5f,
+ 2,
+ LIMIT,
+ false
+ );
+ grouper.init();
+
+ columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value",
10L)));
+
+ // Fill up to the cap; every insert is accepted and proximity stays 0.0
(heap-trim/fill path is suppressed).
+ for (int i = 0; i < maxSize; i++) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(i +
KEY_BASE)).isOk());
+ }
+ Assertions.assertEquals(0.0, grouper.getMaxSpillProximity(), 0.0);
+
+ // The next distinct key cannot be placed and cannot trigger a swap (size
== maxSizeForTesting): a genuine
+ // rejection. Proximity is pinned to exactly 1.0.
+ Assertions.assertFalse(grouper.aggregate(new IntKey(maxSize +
KEY_BASE)).isOk());
+ Assertions.assertEquals(1.0, grouper.getMaxSpillProximity(), 0.0);
+
+ grouper.close();
+ }
+
private static LimitedBufferHashGrouper<IntKey> makeGrouper(
GroupByTestColumnSelectorFactory columnSelectorFactory,
int bufferSize
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 1a96081e91c..a0f68646366 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 class SpillingGrouperTest extends
InitializedNullHandlingTest
}
}
+ @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/terminalRegrowthThreshold, so a
lightly-filled slice is well below 1.0.
+ final GroupByStatsProvider.PerQueryStats stats = new
GroupByStatsProvider.PerQueryStats();
+ final int bufferSize = 100_000;
+ final SpillingGrouper<IntKey> 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++) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+ grouper.close();
+
+ final double proximity = stats.getSpillProximity();
+ Assertions.assertTrue(proximity > 0.0, "proximity should be positive after
aggregating: " + proximity);
+ Assertions.assertTrue(proximity < 1.0, "a lightly-filled slice should be
well below the spill point: " + proximity);
+ }
+
+ @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<IntKey> grouper = makeGrouper(
+ 100_000,
+ new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024,
100, new GroupByStatsProvider.PerQueryStats()),
+ 1024 * 1024L,
+ stats,
+ false // do not init
+ );
+ grouper.close();
+
+ // No spillProximity() call was made, so proximity stays at its initial
0.0.
+ Assertions.assertEquals(0.0, stats.getSpillProximity(), 1e-9);
+ }
+
+ @Test
+ public void testSpillProximityStaysOneAfterSpillThenLightRefill() throws
IOException
+ {
+ // 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<IntKey> 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++) {
+ Assertions.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.
+ Assertions.assertTrue(grouper.aggregate(new IntKey(9001)).isOk());
+ Assertions.assertTrue(grouper.aggregate(new IntKey(9002)).isOk());
+ grouper.close();
+
+ Assertions.assertEquals(
+ 1.0,
+ stats.getSpillProximity(),
+ 0.0,
+ "spilled slice with post-spill light refill must still report 1.0
(peak preserved across reset)"
+ );
+ }
+
+ @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/terminalRegrowthThreshold 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<IntKey> grouper = makeGrouper(
+ bufferSize,
+ new LimitedTemporaryStorage(temporaryFolder.newFolder(), 1024 * 1024,
100, new GroupByStatsProvider.PerQueryStats()),
+ 1024 * 1024L,
+ stats,
+ true
+ );
+
+ for (int i = 0; i < 100; i++) {
+ Assertions.assertTrue(grouper.aggregate(new IntKey(i)).isOk());
+ }
+ grouper.close();
+
+ Assertions.assertEquals(
+ 1.0,
+ stats.getSpillProximity(),
+ 0.0,
+ "a slice that reached its spill trigger must report proximity == 1.0
exactly"
+ );
+ }
+
private SpillingGrouper<IntKey> makeGrouper(
int bufferSize,
File storageDir,
@@ -410,6 +520,17 @@ public class SpillingGrouperTest extends
InitializedNullHandlingTest
LimitedTemporaryStorage temporaryStorage,
long minSpillFileSize
)
+ {
+ return makeGrouper(bufferSize, temporaryStorage, minSpillFileSize, new
GroupByStatsProvider.PerQueryStats(), true);
+ }
+
+ private SpillingGrouper<IntKey> 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 @@ public class SpillingGrouperTest extends
InitializedNullHandlingTest
false,
bufferSize,
minSpillFileSize,
- new GroupByStatsProvider.PerQueryStats()
+ perQueryStats
);
- grouper.init();
+ if (init) {
+ grouper.init();
+ }
return grouper;
}
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 e5f46020fe0..9c69f182f8f 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 class GroupByStatsMonitor extends AbstractMonitor
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 ce5b4f13e9d..96021fb906e 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
@@ -65,6 +65,7 @@ public class GroupByStatsMonitorTest
100L,
200L,
200L,
+ 0.85,
2L,
200L,
200L,
@@ -95,7 +96,7 @@ public class GroupByStatsMonitorTest
// Trigger metric emission
monitor.doMonitor(emitter);
- Assertions.assertEquals(12, emitter.getNumEmittedEvents());
+ Assertions.assertEquals(13, emitter.getNumEmittedEvents());
emitter.verifyValue("mergeBuffer/pendingRequests", 0L);
emitter.verifyValue("mergeBuffer/used", 0L);
emitter.verifyValue("mergeBuffer/queries", 1L);
@@ -103,6 +104,7 @@ public class GroupByStatsMonitorTest
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);
@@ -139,6 +141,7 @@ public class GroupByStatsMonitorTest
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);
@@ -205,21 +208,24 @@ public class GroupByStatsMonitorTest
QueryResourceId r1 = new QueryResourceId("r1");
GroupByStatsProvider.PerQueryStats stats1 =
statsProvider.getPerQueryStatsContainer(r1);
stats1.mergeBufferAcquisitionTime(100);
- stats1.maxMergeBufferUsedBytes(50);
+ stats1.addMergeBufferUsedBytes(50);
+ stats1.spillProximity(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.spillProximity(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.spillProximity(0.1);
stats3.spilledBytes(800);
stats3.dictionarySize(200);
@@ -242,10 +248,34 @@ public class GroupByStatsMonitorTest
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);
+
+
Assertions.assertTrue(emitter.getMetricEvents("mergeBuffer/queries").isEmpty());
+
Assertions.assertTrue(emitter.getMetricEvents("mergeBuffer/maxBytesUsed").isEmpty());
+
Assertions.assertTrue(emitter.getMetricEvents("mergeBuffer/maxSpillProximity").isEmpty());
+ }
+
private void verifyMetricValue(StubServiceEmitter emitter, String
metricName, Map<String, Object> dimFilters, Number expectedValue)
{
final List<ServiceMetricEvent> observedMetricEvents =
emitter.getMetricEvents(metricName);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
