xiangfu0 opened a new pull request, #19380:
URL: https://github.com/apache/pinot/pull/19380
## 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`,
widened `finally` coverage in both combine operators, and a drain for
abandoned queued blocks in the
streaming combine — gated to off-heap blocks, because on-heap generator
`close()` trims a per-worker
thread-local map and must not run cross-thread).
**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). Measured on an M-series Mac, JDK 25, fixed heaps, `-prof gc`:
**~80K groups per segment (cache-resident, the off-heap worst case)** —
latency `RAW_MULTI` −20%,
other tiers +10–22% premium; allocation −13..−27%; GC counts −10..−40%. This
premium is why the
feature is a per-query/per-server opt-in.
**1M groups (latency inverts at scale)** — segment phase `DICT_INT` −12% (49
vs 56ms, alloc
65MB→0.3MB/op); full query `DICT_INT` −50% (160 vs 323ms), `RAW_STRING` −21%
(515 vs 648ms, GC time
in measurement 3057→20ms). On-heap error bars were ±80–250ms (GC-driven) vs
±2–6ms off-heap.
**10M groups (segment phase)** — `DICT_INT` −15% (657 vs 770ms, alloc
519MB→0.4MB/op, GC 67→0ms);
`RAW_STRING` latency parity with GC time 2946→45ms.
**Retained memory for the per-segment state (map + 2 holders, GC-forced)**:
| Scale | On-heap | Off-heap |
|---|---|---|
| int, 4M groups | 136MB heap | 0 heap / 125MB direct |
| string, 4M groups | 389MB heap | 0.2MB heap / 298MB direct |
| int, 100M groups | 2.57GB heap | ~0 heap / 3.57GB direct, lookups 1.9×
faster |
| string, 100M groups | 8.7GB heap, build 146s, lookup 124s (GC-bound) |
4.4MB heap / 8.2GB direct, build 39.5s (3.7×), lookup 31.8s (3.9×), 0 GC |
(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.
## 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.
- `StreamingGroupByCombineOperatorTest` drain tests pinning the
abandoned-block release behavior.
- Existing group-by/null-handling/streaming suites: 577+ tests green;
spotless/checkstyle/license
clean on `pinot-spi`, `pinot-common`, `pinot-core`, `pinot-perf`.
## 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](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
