xiangfu0 commented on PR #19264:
URL: https://github.com/apache/pinot/pull/19264#issuecomment-5518549451
## Design rationale (for reviewers)
This summarizes the design decisions and their rationale, so the diff is
easier to follow. The commit history shows some evolution (an intermediate
cardinality gate that was later removed); this describes the **final** design.
### Problem
`GROUP BY GROUPING SETS / ROLLUP / CUBE` expanded every input row into one
group per grouping set **in each segment**, so per-set fan-out scaled with
scanned rows (`O(rows × numSets)`) and single-threaded per segment. ROLLUP/CUBE
were several times slower than a plain GROUP BY over the same columns.
### Core idea: base aggregation + merge-time derive
1. **Segment phase** aggregates only the *base grouping* (the union of all
grouping-set columns), exactly like a plain GROUP BY — one group per row, no
`$groupingId`. It emits the base groups.
2. **Server combine** (`GroupByCombineOperator`) merges base groups across
segments, then **derives** the individual grouping sets once, in
`mergeResults()`, **parallelized across the combine thread pool** by base-group
ranges into a shared concurrent table.
Total work is minimal (each row grouped once, never per set), and the
per-set fan-out is multi-threaded and runs *after* the row-collapsing base
merge.
**Why it is exact for every aggregation.** Deriving a coarser set = merging
the base groups' *intermediate* results. That is exactly
`AggregationFunction#merge`, which Pinot already requires for scatter-gather
(segment→server→broker). So every mergeable aggregation is correct by
construction — no per-function special-casing.
**Object-accumulator safety.** A base group's intermediate flows into every
grouping set (and across threads), and `merge` mutates/returns its argument. So
each base intermediate is **cloned per derived record** (serialize/deserialize
round-trip; scalar intermediates are immutable and skipped). This keeps AVG /
DISTINCTCOUNT / percentiles exact under concurrent cross-set merges.
### Trimming: accuracy vs. cost, made configurable
The derive is a bounded transformation of the already-bounded base groups
(base ≤ `numGroupsLimit` per segment ⇒ derived ≤ `numGroupsLimit × numSets`),
so it must **not** re-apply the per-segment `numGroupsLimit` cap — doing so
dropped derived groups **non-deterministically** under parallel upsert and
could starve a low-magnitude set (e.g. the grand total). The derive keeps all
groups by default and defers ORDER BY + LIMIT to the broker.
For high-cardinality unions where keeping everything is expensive, a new
**`groupingSetsServerTrimSize`** option applies a **per-set** (bucketed on
`$groupingId`) top-K on the server after the derive. Per-set bucketing is the
safe kind of trim: it never starves a set, only ranks within each. Default is
exact (defer to broker); set it for a bounded, approximate per-server top-K
(broker still does the exact final trim).
### Carve-outs (fall back to per-row expansion)
Only the standard `GroupByCombineOperator` performs the merge-time derive.
Base aggregation is disabled — segments expand per row and emit `$groupingId`
directly — when a different combine path runs:
- **Multi-value group-by columns** — an MV column fans a row across its
values in the base grouping, so rolling it up would over-count.
- **Filtered aggregations** — share a single group-key generator across
aggregation groups (distinct segment path).
- **Streaming combine** (MSE leaf stages, `streamingGroupByFlushThreshold >
0`) — flushes incrementally and cannot derive over the fully-merged base table.
- **Sorted-aggregate combine** — has its own merge that does not derive
(defensive: grouping-set queries always take the unsafe-trim path, so this is
not normally reached).
### Enablement
Enabled by default via `groupingSetsBaseAggregation`; set to `false` to
force the legacy per-row expansion path. Results are identical to expansion
(verified by equivalence tests). Open for reviewer discussion: whether a
behavior-changing computation path should default ON or OFF-until-validated.
### Supporting fix
`IndexedTable` now derives its key-column count from the schema (`columns −
aggregations`) rather than always from `getNumGroupByKeyColumns()`. The base
combine table's records omit the `$groupingId` column, so the old fixed count
merged aggregations at the wrong offset (dropping the first aggregation's
cross-segment contribution / reading out of bounds when a base key repeated
across segments). This is exercised by a new multi-segment test.
### Legacy-generator improvements (fallback path)
For the carve-out cases and `groupingSetsBaseAggregation=false`,
`GroupingSetsGroupKeyGenerator` now: resolves dict-encoded columns via native
dict-ids (no re-hash), packs composite keys into a `long` +
`Long2IntOpenHashMap` when they fit (no per-group object/array hashing), and
reuses per-row group-id buffers across blocks.
### Tests
- `GroupingSetsQueriesTest` — base-agg vs. expansion equivalence across
DISTINCTCOUNT/AVG/SUM over CUBE/ROLLUP/GROUPING SETS, with/without null
handling; server-trim bounds output without starving sets; existing
ROLLUP/CUBE/GROUPING()/null/MV cases.
- `DeriveGroupingSetsTest` — parallel derive identical across thread counts
{1,2,4,8}; cross-segment merge (reproduces the `IndexedTable` bug without the
fix); no group dropped / deterministic even when derived groups exceed a small
`numGroupsLimit`.
- `StreamingGroupByCombineOperatorTest` — grouping-set streaming path emits
the correct `$groupingId` layout.
### Benchmark
`BenchmarkGroupingSetsQueriesSSE` (full server→broker, 50 segments × 15k
rows, 2 forks × 5 iters): base aggregation is **1.8–2.7× faster** at
low/moderate cardinality (many rows collapse into few base groups) and roughly
**parity** at extreme cardinality (base groups ≈ rows, so little to collapse).
Numbers in the PR description.
--
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]
