Add opt-in off-heap group-by key tables and result holders for SSE

[xiangfu0]

Summary

Adds an opt-in off-heap storage mode for the SSE per-segment group-by state — the group-key
tables and the fixed-width aggregation result holders — targeting high-cardinality group-bys where
the on-heap maps/arrays drive GC pressure and heap-retention spikes. Default off; no behavior
change unless enabled.

• Server config: pinot.server.query.executor.groupby.offheap (default false)
• Query option: SET groupByOffHeap = true; (overrides the server config per query, same precedent
  as numGroupsLimit)
• Optional bounded per-thread buffer reuse: pinot.server.query.executor.groupby.offheap.pool.max.bytes.per.thread
  (default 0 = off)

Group ids remain dense ints in insertion order, so no AggregationFunction changes are needed,
and there are no wire or storage format changes. Grouping sets stay on-heap.

Design

New pinot-core package org.apache.pinot.core.query.aggregation.groupby.offheap, backed by
PinotDataBuffer.allocateDirect (so the bytes show up in the existing direct-buffer accounting),
with absolute-indexed direct ByteBuffer views on all hot paths (transparent wrapper fallback for
structures beyond the 2GB view limit — exercised in tests and at 100M groups):

Structure	Replaces	Layout
OffHeapIntGroupIdMap	IntGroupIdMap (dict tier), raw INT/FLOAT fastutil maps	8-byte slots [key+1][groupId], LF 0.5, linear probing, out-of-band -1 key
OffHeapLongGroupIdMap	Long2IntOpenHashMap (dict long tier, raw LONG/DOUBLE, packed two-int keys)	16-byte slots, zero key out-of-band
OffHeapBytesGroupIdMap	Object2IntOpenHashMap<String/…> (raw STRING/BYTES/BIG_DECIMAL, multi-column packed keys)	DuckDB-style two-part table: 8-byte directory entries (16-bit salt | 48-bit payload offset) over append-only 256KB payload chunks storing [hash][groupId][keyLen][key bytes]; the stored hash makes directory resize free of key reads
OffHeapDouble/Long/IntGroupByResultHolder	the on-heap array holders	fixed-width direct memory, identical semantics (defaults, ensureCapacity growth)

Doubles/floats are keyed via doubleToLongBits/floatToIntBits for exact fastutil parity (NaN
collapse, ±0.0 distinct). Strings are encoded with an inline UTF-8 encoder byte-identical to
String.getBytes(UTF_8) (including surrogate handling), so keys hash/compare identically to the
on-heap path.

Lifecycle: a ResourceTrackingGroupKeyGenerator wraps the generator and owns every off-heap
resource, so the existing generator-close() call sites release all direct memory — including the
FilteredGroupByOperator shared-generator case. This PR also hardens close paths that previously
leaked on exceptions (guards in GroupByOperator/FilteredGroupByOperator/DefaultGroupByExecutor
and widened finally coverage in the group-by combine operator). The streaming combine needs no
extra plumbing: since #19066 each per-segment result is detached and its generator closed on the
producing worker thread, which releases the off-heap state promptly as well.

Buffer pool: OffHeapGroupByBufferPool optionally caches freed buffers per thread with exact-size
free lists under a byte cap, mirroring the on-heap thread-local map reuse across queries; pooled bytes
stay visible in direct-buffer usage, and structures zero-fill on acquire so dirty reuse is safe.

Benchmarks

New in pinot-perf: BenchmarkOffHeapGroupBySSE (5 tiers × flag), BenchmarkOffHeapGroupByLargeSSE
(1M groups), BenchmarkOffHeapGroupByHugeSSE (10M groups, segment phase), BenchmarkOffHeapGroupIdMaps
(micro), and OffHeapGroupByMemoryFootprint (deterministic retained-heap/direct harness, up to 100M
groups). All numbers below were re-measured on this rebased branch (base 5e914c979b), sequentially on
an otherwise idle M-series Mac (32GB), JDK 25, fixed heaps, -prof gc; JMH rows are avgt ms/op ± 99.9% CI.

~80K groups per query (cache-resident, the off-heap worst case) — full-query latency, 4 segments ×
300K rows, 10×5s measurement iterations:

Scenario	On-heap	Off-heap	Δ latency	alloc/op	GC count
DICT_INT	11.9 ±0.4	15.5 ±1.2	+30%	55→46MB	49→32
DICT_TWO_COLS	23.4 ±2.4	31.1 ±9.7	+33%	80→65MB	36→23
RAW_INT	15.1 ±0.9	18.2 ±2.3	+21%	63→46MB	44→28
RAW_STRING	38.5 ±3.2	39.3 ±5.5	parity	144→147MB	39→39
RAW_MULTI	98.9 ±7.0	71.0 ±3.3	−28%	242→210MB	26→31

The dict/raw-int premium is why the feature is a per-query/per-server opt-in. (It reads wider than the
pre-rebase measurement mainly because the on-heap baseline itself got faster on current master; the
off-heap side is unchanged.)

1M groups (the trade inverts at scale) — 2 segments × 2M rows:
segment phase DICT_INT −18% (54.8 vs 67.1ms, alloc 65MB→0.3MB/op, GC 32→0ms); full query DICT_INT
−26% (253 vs 341ms, alloc 380→251MB/op); RAW_STRING latency parity with in-measurement GC time
3014→23ms (segment phase 4284→14ms — the on-heap arm spends seconds per iteration in GC that simply
disappears).

10M groups (segment phase, 1×12M-row segment) — DICT_INT −33% (624 vs 933ms, alloc
519MB→0.5MB/op, GC 101→0ms); RAW_STRING parity (+3.8%, within error) with GC time 3696→26ms and
alloc 1761→1193MB/op.

Retained memory for the per-segment state (map + 2 holders, GC-forced):

Scale	On-heap	Off-heap
int, 4M groups	137MB heap	0 heap / 125MB direct
string, 4M groups	388MB heap	0.2MB heap / 302MB direct, build/lookup ~20% faster
int, 100M groups	2.57GB heap, build 6.6s, lookup 3.5s	~0 heap / 3.57GB direct, build 4.9s, lookup 2.1s (1.65×)
string, 100M groups	8.70GB heap, build 31.8s, lookup 59.9s, GC 2.8s	4.4MB heap / 8.22GB direct, build 25.1s, lookup 21.0s (2.9×), GC 0.12s

(The int-100M direct total is larger than on-heap at that exact count — LF 0.5 vs 0.75 on a pow2
boundary; the point of the mode is where the bytes live, not always fewer of them.)

Included fix (first commit, affects the default on-heap path)

NoDictionarySingleColumnGroupKeyGenerator keeps the null group outside its key map for
primitive types but did not count it in getNumKeys() / getCurrentGroupKeyUpperBound(). Since the
null group takes the next dense id, the upper bound could equal an issued id — a latent
result-holder under-sizing (AIOOBE) and wrong numKeys bookkeeping wherever those counts are used
(e.g. TableResizer trim decisions). Split into its own commit so it can be evaluated independently,
with NoDictionaryNullGroupCountRegressionTest reproducing the AIOOBE through
DefaultGroupByExecutor.process() for all four primitive stored types (verified to fail with
ArrayIndexOutOfBoundsException when the fix is reverted).

Testing

• OffHeapGroupByQueriesTest — end-to-end differential battery running every query with the flag on
  and off and comparing results row-for-row, with per-query direct-memory leak assertions.
• OffHeapGroupKeyGeneratorParityTest — generator-level parity incl. null-group id bookkeeping.
• Per-structure unit tests for the three maps, three holders, pool, and UTF-8 encoder (differential
  vs the JDK, incl. surrogates), plus forced wrapper-fallback (>2GB view) runs for every structure.
• Streaming-combine leak tests (StreamingGroupByCombineOperatorTest): with off-heap enabled and
  high-cardinality segments, direct-buffer usage returns exactly to baseline both after full
  consumption and when the consumer abandons the stream after the first block.
• Existing group-by/null-handling/streaming suites (incl. the Fix data race in StreamingGroupByCombineOperator over reused thread-local group-by state #19066 streaming detach tests): 580+
  tests green; spotless/checkstyle/license clean on pinot-spi, pinot-common, pinot-core,
  pinot-perf.

Known limitations

• Off-heap bytes are not yet visible to the per-query resource accountant; the only bound is
  numGroupsLimit (see the config javadoc). Size -XX:MaxDirectMemorySize before enabling.
• The off-heap holders/maps guard group-id bounds with assert on the hot path (tests run with
  assertions enabled; production does not), trading the on-heap path's implicit AIOOBE for speed.
• On the streaming combine, per-segment results are materialized on-heap at hand-off (Fix data race in StreamingGroupByCombineOperator over reused thread-local group-by state #19066), so
  off-heap there only relieves the segment-execution phase itself.

Follow-ups (out of scope here)

• Off-heap IndexedTable for the cross-segment combine (dominates at 10M+ groups; the current
  combine is unchanged and mode-independent).
• MSE group-by operator support.
• ThreadAccountant integration for off-heap bytes (per-query budgeting) and a server gauge for
  pooled bytes.

🤖 Generated with Claude Code

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.09454% with 97 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.72%. Comparing base (bbbed25) to head (137b6c0).

Files with missing lines	Patch %	Lines
...gation/groupby/offheap/OffHeapBytesGroupIdMap.java	91.01%	12 Missing and 11 partials ⚠️
...pby/NoDictionarySingleColumnGroupKeyGenerator.java	92.25%	7 Missing and 4 partials ⚠️
...ry/aggregation/groupby/DefaultGroupByExecutor.java	67.74%	7 Missing and 3 partials ⚠️
...tion/groupby/offheap/OffHeapGroupByBufferPool.java	83.33%	6 Missing and 3 partials ⚠️
...regation/groupby/offheap/OffHeapIntGroupIdMap.java	94.96%	2 Missing and 5 partials ⚠️
...egation/groupby/offheap/OffHeapLongGroupIdMap.java	95.17%	2 Missing and 5 partials ⚠️
...pby/offheap/ResourceTrackingGroupKeyGenerator.java	79.31%	5 Missing and 1 partial ⚠️
...t/core/operator/query/FilteredGroupByOperator.java	42.85%	4 Missing ⚠️
.../core/operator/combine/GroupByCombineOperator.java	85.71%	1 Missing and 2 partials ⚠️
...che/pinot/core/operator/query/GroupByOperator.java	25.00%	3 Missing ⚠️
... and 8 more

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #19380       +/-   ##
=============================================
+ Coverage     57.71%   67.72%   +10.01%     
- Complexity        7     1430     +1423     
=============================================
  Files          2686     3495      +809     
  Lines        163987   225342    +61355     
  Branches      26627    35571     +8944     
=============================================
+ Hits          94640   152618    +57978     
+ Misses        61352    60665      -687     
- Partials       7995    12059     +4064     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.72% <92.09%> (+10.01%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.72% <92.09%> (+10.01%)	⬆️
unittests	67.72% <92.09%> (+10.01%)	⬆️
unittests1	57.96% <92.09%> (+0.25%)	⬆️
unittests2	39.12% <0.97%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.