diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 8d71c6044f85..5989039391b7 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'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| @@ -122,8 +123,9 @@ Most metric values reset each emission period, as specified in `druid.monitoring |`mergeBuffer/queries`|Number of groupBy queries that acquired a batch of buffers from the merge buffer pool.|This metric is only available if the `GroupByStatsMonitor` module is included.|Depends on the number of groupBy queries needing merge buffers.| |`mergeBuffer/acquisitionTimeNs`|Total time in nanoseconds to acquire merge buffer for groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| |`mergeBuffer/maxAcquisitionTimeNs`|Maximum time in nanoseconds to acquire merge buffer for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| -|`mergeBuffer/bytesUsed`|Number of bytes used by merge buffers to process groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| -|`mergeBuffer/maxBytesUsed`|Maximum number of bytes used by merge buffers for any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| +|`mergeBuffer/bytesUsed`|Total merge buffer bytes used to process groupBy queries, summed across queries. Each query's value is itself the sum across the slices it held: a merge buffer is divided among `druid.processing.numThreads` concurrent query slices, so a query can spill once a single slice fills even though this summed usage looks well below `druid.processing.buffer.sizeBytes`. To gauge spill pressure, use `mergeBuffer/maxSpillProximity` rather than comparing this to `sizeBytes`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| +|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single groupBy query within the emission period, where each query's usage is the sum across the slices it held. Because the buffer is sliced among `druid.processing.numThreads` query slices, this value is not directly comparable to `druid.processing.buffer.sizeBytes`; use `mergeBuffer/maxSpillProximity` to gauge spill pressure.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| +|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to spilling within the emission period, as a fraction in [0.0, 1.0]. A merge buffer is divided into `druid.processing.numThreads` slices and a query spills as soon as its fullest single slice's hash table cannot allocate another bucket; this metric is that fullest slice's peak fill ratio (bucket count over the load-factor bucket limit), taken as a max across slices and across queries. **1.0 corresponds exactly to the spill trigger** — the value is bucket-count based, so it is not distorted by bucket width, offset-list overhead, or integer truncation. If you see 1.0, raise `druid.processing.buffer.sizeBytes` or lower `druid.processing.numThreads` to widen each slice. Read alongside `groupBy/spilledQueries`.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| |`groupBy/spilledQueries`|Number of groupBy queries that have spilled onto the disk.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| |`groupBy/spilledBytes`|Number of bytes spilled on the disk by the groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| |`groupBy/maxSpilledBytes`|Maximum number of bytes spilled to disk by any single groupBy query within the emission period.|This metric is only available if the `GroupByStatsMonitor` module is included.|Varies| @@ -154,8 +156,9 @@ to represent the task ID are deprecated and will be removed in a future release. |`mergeBuffer/queries`|Number of groupBy queries that acquired a batch of buffers from the merge buffer pool. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Depends on the number of groupBy queries needing merge buffers.| |`mergeBuffer/acquisitionTimeNs`|Total time in nanoseconds to acquire merge buffer for groupBy queries. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| |`mergeBuffer/maxAcquisitionTimeNs`|Maximum time in nanoseconds to acquire merge buffer for any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| -|`mergeBuffer/bytesUsed`|Number of bytes used by merge buffers to process groupBy queries.|This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| -|`mergeBuffer/maxBytesUsed`|Maximum number of bytes used by merge buffers for any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| +|`mergeBuffer/bytesUsed`|Total merge buffer bytes used to process groupBy queries, summed across queries. Each query's value is itself the sum across the slices it held: a merge buffer is divided among `druid.processing.numThreads` concurrent query slices, so a query can spill once a single slice fills even though this summed usage looks well below `druid.processing.buffer.sizeBytes`. To gauge spill pressure, use `mergeBuffer/maxSpillProximity` rather than comparing this to `sizeBytes`. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| +|`mergeBuffer/maxBytesUsed`|Maximum merge buffer bytes used by any single groupBy query within the emission period, where each query's usage is the sum across the slices it held. Because the buffer is sliced among `druid.processing.numThreads` query slices, this value is not directly comparable to `druid.processing.buffer.sizeBytes`; use `mergeBuffer/maxSpillProximity` to gauge spill pressure. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| +|`mergeBuffer/maxSpillProximity`|How close any single groupBy query came to spilling within the emission period, as a fraction in [0.0, 1.0]. A merge buffer is divided into `druid.processing.numThreads` slices and a query spills as soon as its fullest single slice's hash table cannot allocate another bucket; this metric is that fullest slice's peak fill ratio (bucket count over the load-factor bucket limit), taken as a max across slices and across queries. **1.0 corresponds exactly to the spill trigger** — the value is bucket-count based, so it is not distorted by bucket width, offset-list overhead, or integer truncation. If you see 1.0, raise `druid.processing.buffer.sizeBytes` or lower `druid.processing.numThreads` to widen each slice. Read alongside `groupBy/spilledQueries`. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| |`groupBy/spilledQueries`|Number of groupBy queries that have spilled onto the disk. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| |`groupBy/spilledBytes`|Number of bytes spilled on the disk by the groupBy queries. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| |`groupBy/maxSpilledBytes`|Maximum number of bytes spilled to disk by any single groupBy query within the emission period. This metric is only available if the `GroupByStatsMonitor` module is included.|`dataSource`, `taskId`|Varies| 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..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 @@ -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 @@ -73,6 +74,7 @@ public static class AggregateStats private long maxMergeBufferAcquisitionTimeNs = 0; private long totalMergeBufferUsedBytes = 0; private long maxMergeBufferUsedBytes = 0; + private double maxSpillProximity = 0.0; private long spilledQueries = 0; private long spilledBytes = 0; private long maxSpilledBytes = 0; @@ -91,6 +93,7 @@ public AggregateStats(AggregateStats aggregateStats) aggregateStats.maxMergeBufferAcquisitionTimeNs, aggregateStats.totalMergeBufferUsedBytes, aggregateStats.maxMergeBufferUsedBytes, + aggregateStats.maxSpillProximity, aggregateStats.spilledQueries, aggregateStats.spilledBytes, aggregateStats.maxSpilledBytes, @@ -105,6 +108,7 @@ public AggregateStats( long maxMergeBufferAcquisitionTimeNs, long totalMergeBufferUsedBytes, long maxMergeBufferUsedBytes, + double maxSpillProximity, long spilledQueries, long spilledBytes, long maxSpilledBytes, @@ -117,6 +121,7 @@ public AggregateStats( this.maxMergeBufferAcquisitionTimeNs = maxMergeBufferAcquisitionTimeNs; this.totalMergeBufferUsedBytes = totalMergeBufferUsedBytes; this.maxMergeBufferUsedBytes = maxMergeBufferUsedBytes; + this.maxSpillProximity = maxSpillProximity; this.spilledQueries = spilledQueries; this.spilledBytes = spilledBytes; this.maxSpilledBytes = maxSpilledBytes; @@ -149,6 +154,11 @@ public long getMaxMergeBufferUsedBytes() return maxMergeBufferUsedBytes; } + public double getMaxSpillProximity() + { + return maxSpillProximity; + } + public long getSpilledQueries() { return spilledQueries; @@ -174,6 +184,20 @@ 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 +207,9 @@ public void addQueryStats(PerQueryStats perQueryStats) maxMergeBufferAcquisitionTimeNs, perQueryStats.getMergeBufferAcquisitionTimeNs() ); - totalMergeBufferUsedBytes += perQueryStats.getMaxMergeBufferUsedBytes(); - maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, perQueryStats.getMaxMergeBufferUsedBytes()); + totalMergeBufferUsedBytes += perQueryStats.getMergeBufferUsedBytes(); + maxMergeBufferUsedBytes = Math.max(maxMergeBufferUsedBytes, perQueryStats.getMergeBufferUsedBytes()); + maxSpillProximity = Math.max(maxSpillProximity, perQueryStats.getSpillProximity()); } if (perQueryStats.getSpilledBytes() > 0) { @@ -204,6 +229,7 @@ public void reset() this.maxMergeBufferAcquisitionTimeNs = 0; this.totalMergeBufferUsedBytes = 0; this.maxMergeBufferUsedBytes = 0; + this.maxSpillProximity = 0.0; this.spilledQueries = 0; this.spilledBytes = 0; this.maxSpilledBytes = 0; @@ -215,7 +241,24 @@ public void reset() public static class PerQueryStats { private final AtomicLong mergeBufferAcquisitionTimeNs = new AtomicLong(0); - private final AtomicLong maxMergeBufferUsedBytes = new AtomicLong(0); + /** + * Sum of the peak merge-buffer usage of every grouper (slice) this query held. A + * {@code ConcurrentGrouper} slices a single merge buffer into one slice per processing thread, and each slice + * reports its own peak via + * {@link #addMergeBufferUsedBytes(long)} when closed, so the per-query value is the SUM across the query's slices. + */ + private final AtomicLong mergeBufferUsedBytes = new AtomicLong(0); + /** + * Spill proximity of the single fullest slice this query held, in [0.0, 1.0]. Each {@link #sliceUsage} call + * contributes one slice's peak {@code size / regrowthThreshold} ratio (tracked bucket-by-bucket by the underlying + * hash table, preserved across resets), and this keeps the MAX across the query's slices. A query spills as soon + * as one slice fills, so proximity is driven by the hottest slice, NOT the byte sum tracked by + * {@link #mergeBufferUsedBytes}. Keeping the ratio per slice (rather than maxing numerator and denominator + * independently) is what makes the metric correct when a single query mixes groupers with different spill + * thresholds — e.g. small sliced groupers from a {@code ConcurrentGrouper} alongside a full-buffer + * {@code SpillingGrouper} for subtotal/nested processing. 1.0 corresponds exactly to the actual spill trigger. + */ + private final DoubleAccumulator maxSpillProximity = new DoubleAccumulator(Math::max, 0.0); private final AtomicLong spilledBytes = new AtomicLong(0); private final AtomicLong mergeDictionarySize = new AtomicLong(0); @@ -224,9 +267,32 @@ public void mergeBufferAcquisitionTime(long delay) mergeBufferAcquisitionTimeNs.addAndGet(delay); } - public void maxMergeBufferUsedBytes(long bytes) + /** + * Accumulates the peak merge-buffer usage of one grouper (slice). Despite the previous "max" naming, this method + * sums across the slices a query holds; see {@link #mergeBufferUsedBytes}. + */ + public void addMergeBufferUsedBytes(long bytes) { - maxMergeBufferUsedBytes.addAndGet(bytes); + mergeBufferUsedBytes.addAndGet(bytes); + } + + /** + * Records one slice's peak fill ratio in [0.0, 1.0] — the underlying hash table's peak + * {@code size / regrowthThreshold} over the slice's lifetime, which reaches exactly 1.0 iff the slice actually + * spilled. Kept as a max across the query's slices, so after all slices close the value describes the single + * fullest slice — the one that drives spilling. Recording the ratio per slice (rather than maxing bytes and + * thresholds independently) keeps each slice's numerator paired with its own denominator, so a query that mixes + * groupers of different sizes still reports the true max proximity. Used to compute + * {@code mergeBuffer/maxSpillProximity}. Values are clamped defensively to [0, 1]; NaN is ignored so a + * never-initialized grouper contributes nothing. + */ + public void sliceUsage(double proximity) + { + if (Double.isNaN(proximity)) { + return; + } + final double clamped = proximity < 0.0 ? 0.0 : (proximity > 1.0 ? 1.0 : proximity); + maxSpillProximity.accumulate(clamped); } public void spilledBytes(long bytes) @@ -244,9 +310,19 @@ public long getMergeBufferAcquisitionTimeNs() return mergeBufferAcquisitionTimeNs.get(); } - public long getMaxMergeBufferUsedBytes() + public long getMergeBufferUsedBytes() + { + return mergeBufferUsedBytes.get(); + } + + /** + * Spill proximity for this query in [0.0, 1.0]: the fullest slice's peak {@code size / regrowthThreshold} over + * that slice's lifetime. 1.0 corresponds exactly to the spill trigger (a bucket allocation was rejected). Returns + * 0.0 when no slice usage was recorded (e.g. a grouper that never initialized). + */ + public double getSpillProximity() { - return maxMergeBufferUsedBytes.get(); + return maxSpillProximity.get(); } public long getSpilledBytes() 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/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 96e55907b21f..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 @@ -249,7 +249,15 @@ 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 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; 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..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 @@ -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. 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(0.05); + stats1.addMergeBufferUsedBytes(30); + stats1.sliceUsage(0.03); 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 0.05. + stats2.addMergeBufferUsedBytes(100); + stats2.sliceUsage(0.05); 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 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 void testMetricsWithMultipleQueries() QueryResourceId r1 = new QueryResourceId("r1"); GroupByStatsProvider.PerQueryStats stats1 = statsProvider.getPerQueryStatsContainer(r1); stats1.mergeBufferAcquisitionTime(2000); - stats1.maxMergeBufferUsedBytes(50); + stats1.addMergeBufferUsedBytes(50); + stats1.sliceUsage(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(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(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(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,212 @@ 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 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(0.01); + stats.addMergeBufferUsedBytes(20); + stats.sliceUsage(0.02); + stats.addMergeBufferUsedBytes(30); + stats.sliceUsage(0.03); + stats.addMergeBufferUsedBytes(40); + stats.sliceUsage(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.sliceUsage(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.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: 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.sliceUsage(0.2); + stats.sliceUsage(0.9); + stats.sliceUsage(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 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(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); + } + + @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(0.3); + agg.addQueryStats(low); + + GroupByStatsProvider.PerQueryStats high = new GroupByStatsProvider.PerQueryStats(); + high.mergeBufferAcquisitionTime(10); + high.sliceUsage(0.8); + agg.addQueryStats(high); + + GroupByStatsProvider.PerQueryStats mid = new GroupByStatsProvider.PerQueryStats(); + mid.mergeBufferAcquisitionTime(10); + mid.sliceUsage(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(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.sliceUsage(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 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/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..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 @@ -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(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(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(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);