yashmayya commented on code in PR #19264:
URL: https://github.com/apache/pinot/pull/19264#discussion_r4127080325
##########
pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java:
##########
@@ -252,6 +253,26 @@ public int getNumExtraGroupByKeyColumns() {
return isGroupingSets() ? 1 : 0;
}
+ /// Returns whether grouping-set queries should aggregate the base grouping
(union columns) once per segment
+ /// and derive the individual grouping-set records from those base groups,
instead of expanding every input
+ /// row into one group per grouping set. Enabled by default; the
`groupingSetsBaseAggregation=false` query
+ /// option forces the legacy per-row expansion path. See
[CommonConstants.Broker.Request.QueryOptionKey].
+ ///
+ /// Only the standard
[org.apache.pinot.core.operator.combine.GroupByCombineOperator] performs the
merge-time
+ /// derive that turns the emitted base groups back into grouping-set
records. Base aggregation is therefore
+ /// disabled whenever a different combine path would run: filtered
aggregations (distinct shared group-key
+ /// generator), the streaming combine used on MSE leaf stages
(`streamingGroupByFlushThreshold > 0`, which
+ /// flushes incrementally and cannot derive over the fully-merged base
table), and the sorted-aggregate combine
+ /// ([#shouldSortAggregateUnderSafeTrim()]). In all these cases the segment
expands per row so the emitted rows
+ /// already carry the $groupingId discriminator.
+ public boolean isGroupingSetsBaseAggregation() {
+ if (_hasFilteredAggregations || _streamingGroupByFlushThreshold > 0 ||
shouldSortAggregateUnderSafeTrim()) {
Review Comment:
**Blocker: wrong subtotals for aggregations that merge only across segments**
`SEGMENTPARTITIONEDDISTINCTCOUNT` and `FUNNEL_COUNT(...
SETTINGS('partitioned'))` keep a count that is final for one segment and one
group. Their merge adds the counts. That is correct across segments. But the
derive also merges base groups of the same segment. If one user is in two base
groups, the subtotal counts that user two times.
Repro, one segment: user 1 is in (US, ios) and (US, android). User 2 is in
(US, ios). User 3 is in (CA, ios) and (CA, web). `GROUP BY ROLLUP(country,
device)`:
| | Base path | Expansion path |
|---|---|---|
| `SEGMENTPARTITIONEDDISTINCTCOUNT(userId)`: US / CA / total | 3 / 2 / 5 | 2
/ 1 / 3 |
| partitioned `FUNNEL_COUNT`: CA / total | [1, 0] / [3, 0] | [1, 1] / [3, 2]
|
The broker rewrites a plain `DISTINCTCOUNT(col)` to
`SEGMENTPARTITIONEDDISTINCTCOUNT` when the column is segment-partitioned. So
normal queries on these tables return wrong numbers by default, with no flag.
Fix idea: use the expansion path when any aggregation cannot merge across
groups (for example, a new flag on `AggregationFunction`).
##########
pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java:
##########
@@ -87,6 +102,280 @@ public static GroupByResultsBlock
buildGroupingSetsResultsBlock(QueryContext que
return resultsBlock;
}
+ /// Derives the individual grouping sets from a merged BASE-grouping
[IndexedTable] (union columns aggregated
+ /// once, like a plain GROUP BY), for a GROUP BY GROUPING SETS / ROLLUP /
CUBE query using base aggregation.
+ /// Each base group is projected into every grouping set: its rolled-up
(non-participating) columns are set to
+ /// `null`, the `$groupingId` discriminator is inserted after the union
columns, and the base group's
+ /// aggregation intermediates are merged into the derived group. This moves
the per-set fan-out from O(rows) to
+ /// O(base groups) and runs it here -- after the row-collapsing base merge
-- across the combine's threads,
+ /// rather than expanding every scanned row.
+ ///
+ /// The grouping sets are partitioned round-robin across up to `numTasks`
worker tasks; each task derives its
+ /// owned sets over all base entries into a task-local plain HashMap. Task
key spaces are disjoint (every
+ /// derived key carries its set ordinal), so the heavy merge work runs
contention-free -- no concurrent table,
+ /// no bin locks, no shared size counters -- and the task maps are then
unioned single-threaded into the result
+ /// table with plain stores (never merges). Profiling showed the previous
shared-concurrent-table derive spent
+ /// most of its time in ConcurrentHashMap machinery rather than in the
projection itself.
+ ///
+ /// Clone discipline: a base group's intermediate flows into every grouping
set (and is read concurrently by
+ /// other tasks), while [AggregationFunction#merge] mutates/returns its
arguments. Each derived group's stored
+ /// accumulator is therefore a clone (see [#cloneIntermediate]), and OBJECT
intermediates are re-cloned per
+ /// merge so a merge can never mutate the shared base accumulator another
task is still reading. Scalar
+ /// intermediates are immutable and clone-free.
+ public static IndexedTable
deriveGroupingSetsFromMergedBaseTable(IndexedTable baseTable, QueryContext
queryContext,
+ int numTasks, ExecutorService executorService) {
+ AggregationFunction[] aggregationFunctions =
queryContext.getAggregationFunctions();
+ assert aggregationFunctions != null;
+ int numAggregationFunctions = aggregationFunctions.length;
+ List<int[]> groupingSets = queryContext.getGroupingSets();
+ int numSets = groupingSets.size();
+ int numUnionColumns = queryContext.getGroupByExpressions().size();
+ // Per grouping set: membership mask over the union columns (true =
participates, false = rolled up to NULL).
+ boolean[][] setContains = new boolean[numSets][numUnionColumns];
+ for (int s = 0; s < numSets; s++) {
+ for (int columnIndex : groupingSets.get(s)) {
+ setContains[s][columnIndex] = true;
+ }
+ }
+
+ // Grouping-set output schema: the base schema with the synthetic
$groupingId INT column inserted right after
+ // the union group-by columns (mirroring GroupByOperator's grouping-set
schema layout).
+ DataSchema groupingSetsSchema =
insertGroupingIdColumn(baseTable.getDataSchema(), numUnionColumns);
+ // The derive must not drop groups: it is a bounded transformation of the
already-bounded base groups (whose
+ // count is capped at numGroupsLimit per segment), so the derived count is
at most numGroupsLimit * numSets --
+ // a finite amount. Capping the derived table at numGroupsLimit would drop
derived groups NON-DETERMINISTICALLY
+ // under the parallel upsert (whichever threads fill the quota first win),
which can starve an entire
+ // low-magnitude grouping set such as the grand total. Use an unbounded
result size here and defer the real
+ // ORDER BY + LIMIT to the broker (per-set trim below still bounds it when
explicitly configured).
+ int derivedUpperBound = (int) Math.min((long) baseTable.size() * numSets,
Integer.MAX_VALUE);
+ int initialCapacity = getIndexedTableInitialCapacity(derivedUpperBound,
derivedUpperBound,
+ queryContext.getMinInitialIndexedTableCapacity());
+
+ List<Map.Entry<Key, Record>> baseEntries = new
ArrayList<>(baseTable.getRecordEntries());
+ // A full-union set (one that contains every union column, i.e. the
identity grouping) maps base groups to
+ // derived groups 1:1: keys are unique so no merge ever runs on its
records, and its stored intermediates can
+ // safely be the base objects themselves (all other readers of a base
intermediate treat it as read-only).
+ boolean[] isFullUnionSet = new boolean[numSets];
+ for (int s = 0; s < numSets; s++) {
+ boolean fullUnion = true;
+ for (int col = 0; col < numUnionColumns; col++) {
+ fullUnion &= setContains[s][col];
+ }
+ isFullUnionSet[s] = fullUnion;
+ }
+ int numTaskSlots = Math.max(1, Math.min(numTasks, numSets));
+ List<Map<Key, Record>> taskMaps;
+ if (numTaskSlots == 1) {
+ taskMaps = List.of(deriveSets(baseEntries, setContains, isFullUnionSet,
0, 1, numUnionColumns,
+ numAggregationFunctions, aggregationFunctions));
+ } else {
+ List<Future<Map<Key, Record>>> futures = new ArrayList<>(numTaskSlots);
+ for (int t = 0; t < numTaskSlots; t++) {
+ int taskIndex = t;
+ futures.add(executorService.submit(() -> deriveSets(baseEntries,
setContains, isFullUnionSet, taskIndex,
+ numTaskSlots, numUnionColumns, numAggregationFunctions,
aggregationFunctions)));
+ }
+ taskMaps = new ArrayList<>(numTaskSlots);
+ try {
+ // Bound each wait by the query deadline (when one is set) so a stuck
task cannot hold the combine past
+ // the timeout.
+ long endTimeMs = queryContext.getEndTimeMs();
+ for (Future<Map<Key, Record>> future : futures) {
+ if (endTimeMs > 0) {
+ taskMaps.add(future.get(Math.max(endTimeMs -
System.currentTimeMillis(), 0), TimeUnit.MILLISECONDS));
+ } else {
+ taskMaps.add(future.get());
+ }
+ }
+ } catch (InterruptedException e) {
+ cancelAll(futures);
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while deriving grouping sets",
e);
+ } catch (TimeoutException e) {
+ cancelAll(futures);
+ throw new RuntimeException("Timed out while deriving grouping sets",
e);
+ } catch (ExecutionException e) {
+ cancelAll(futures);
+ throw new RuntimeException("Caught exception while deriving grouping
sets", e.getCause());
+ }
+ }
+
+ // Union the disjoint task maps into the result table. Since task key
spaces are disjoint, the union never
+ // merges records. In the common case, adopt the largest task map as the
table's backing map and putAll the
+ // rest into it, so the largest map's keys are never re-hashed. The
deterministic accurate-group-by mode
+ // needs a sorted (skip-list) backing map for deterministic iteration
order, so it keeps the upsert-based
+ // union into the table produced by getTrimDisabledIndexedTable
(single-threaded, numThreads = 1).
+ IndexedTable derivedTable;
+ if (useDeterministicIndexedTable(queryContext)) {
+ derivedTable = getTrimDisabledIndexedTable(groupingSetsSchema, false,
queryContext, Integer.MAX_VALUE,
+ initialCapacity, 1, executorService);
+ for (Map<Key, Record> taskMap : taskMaps) {
+ for (Map.Entry<Key, Record> entry : taskMap.entrySet()) {
+ derivedTable.upsert(entry.getKey(), entry.getValue());
+ }
+ }
+ } else {
+ Map<Key, Record> mergedMap = taskMaps.get(0);
+ for (Map<Key, Record> taskMap : taskMaps) {
+ if (taskMap.size() > mergedMap.size()) {
+ mergedMap = taskMap;
+ }
+ }
+ for (Map<Key, Record> taskMap : taskMaps) {
+ if (taskMap != mergedMap) {
+ mergedMap.putAll(taskMap);
+ }
+ }
+ derivedTable = new SimpleIndexedTable(groupingSetsSchema, false,
queryContext, Integer.MAX_VALUE,
+ Integer.MAX_VALUE, Integer.MAX_VALUE, mergedMap, executorService);
+ }
+
+ /// Optional server-side per-set trim: when configured (and there is an
ORDER BY), keep at most K groups
+ /// within each grouping set (bucketed by $groupingId), bounding this
server's derived output for
+ /// high-cardinality unions. Bucketing per set means a global top-K cannot
starve a low-magnitude set. This
+ /// is an approximate top-K (deferred exact ORDER BY + LIMIT still runs at
the broker); unset -> keep all.
+ int serverTrimSize = queryContext.getGroupingSetServerTrimSize();
+ if (serverTrimSize > 0 && derivedTable.size() > (long) serverTrimSize *
numSets) {
+ derivedTable.finish(false);
+ TableResizer tableResizer = new TableResizer(groupingSetsSchema,
queryContext);
+ List<IntermediateRecord> kept =
tableResizer.trimTableByGroupingSet(derivedTable, serverTrimSize,
+ numUnionColumns);
+
ServerMetrics.get().addMeteredGlobalValue(ServerMeter.AGGREGATE_TIMES_GROUPS_TRIMMED,
1);
+ IndexedTable trimmedTable =
+ buildIndexedTableFromRecords(groupingSetsSchema, queryContext, kept,
executorService);
+ // Surface the approximation: groups were dropped on the server, so the
merged block (and ultimately the
+ // broker response) must be flagged as trimmed.
+ trimmedTable.markTrimmed();
+ return trimmedTable;
+ }
+ return derivedTable;
+ }
+
+ /// Builds a trim-disabled grouping-set [IndexedTable] pre-populated with
the given (already unique) records.
+ /// Used to materialize the per-set-trim survivors back into the table the
combine returns.
+ private static IndexedTable buildIndexedTableFromRecords(DataSchema
dataSchema, QueryContext queryContext,
+ List<IntermediateRecord> records, ExecutorService executorService) {
+ int numRecords = records.size();
+ int initialCapacity =
+ getIndexedTableInitialCapacity(numRecords, numRecords,
queryContext.getMinInitialIndexedTableCapacity());
+ // Populated single-threaded, so use a plain-HashMap-backed table
(numThreads = 1).
+ IndexedTable table = getTrimDisabledIndexedTable(dataSchema, false,
queryContext, Integer.MAX_VALUE,
+ initialCapacity, 1, executorService);
+ for (IntermediateRecord record : records) {
+ table.upsert(record._key, record._record);
+ }
+ return table;
+ }
+
+ /// One derive task: projects every base group into the grouping sets owned
by `taskIndex` (set `s` is owned
+ /// when `s % numTaskSlots == taskIndex`) and aggregates the derived groups
into a task-local map. Each task
+ /// runs single-threaded over its own map, so a plain HashMap suffices; task
key spaces are disjoint because
+ /// every derived key ends with its set ordinal.
+ ///
+ /// The first record stored for a derived key holds cloned intermediates (it
becomes the group's accumulator,
+ /// mutated by later merges); on merge, OBJECT intermediates from the shared
base group are cloned again so the
+ /// merge can never mutate a base accumulator that other tasks are still
reading. Scalar intermediates are
+ /// immutable and pass through uncloned.
+ private static Map<Key, Record> deriveSets(List<Map.Entry<Key, Record>>
baseEntries, boolean[][] setContains,
+ boolean[] isFullUnionSet, int taskIndex, int numTaskSlots, int
numUnionColumns, int numAggregationFunctions,
+ AggregationFunction[] aggregationFunctions) {
+ int numSets = setContains.length;
+ // A full-union owned set produces exactly one derived group per base
entry, so presize for it (it never
+ // rehashes); tasks owning only coarser sets start small and grow, since
their distinct-group counts are
+ // usually far below the base size and a full-size bucket array per task
would be wasted memory.
+ boolean ownsFullUnionSet = false;
+ for (int s = taskIndex; s < numSets; s += numTaskSlots) {
+ ownsFullUnionSet |= isFullUnionSet[s];
+ }
+ Map<Key, Record> taskMap =
+ ownsFullUnionSet ? new
HashMap<>(HashUtil.getHashMapCapacity(baseEntries.size())) : new HashMap<>();
+ // Flyweight probe: reuse one key buffer for lookups (HashMap.get does not
retain its argument) and copy it
+ // into a fresh array only when inserting a new derived group. Most
projections into coarse sets hit an
+ // existing group, so this avoids a key-array + Key allocation per
projection on the merge path.
+ Object[] probeValues = new Object[numUnionColumns + 1];
+ Key probeKey = new Key(probeValues);
+ int numProjections = 0;
+ for (Map.Entry<Key, Record> baseEntry : baseEntries) {
+ Object[] baseKeys = baseEntry.getKey().getValues();
+ Object[] baseValues = baseEntry.getValue().getValues();
+ for (int s = taskIndex; s < numSets; s += numTaskSlots) {
+ // Keep the derive responsive to query timeout / cancellation and
visible to resource accounting, like
+ // the segment-merge loop in GroupByCombineOperator.
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numProjections++,
"GroupByUtils#deriveSets");
+ boolean[] contains = setContains[s];
+ for (int col = 0; col < numUnionColumns; col++) {
+ probeValues[col] = contains[col] ? baseKeys[col] : null;
+ }
+ probeValues[numUnionColumns] = s;
+ // Full-union sets never see a duplicate key, so skip the probe and
insert blindly. Their records can
+ // also share the base intermediates without cloning: no merge will
ever mutate them (this set never
+ // merges, and every merge into OTHER sets clones its second argument
before use).
+ boolean fullUnion = isFullUnionSet[s];
+ Record existing = fullUnion ? null : taskMap.get(probeKey);
+ if (existing == null) {
+ Object[] keyValues = probeValues.clone();
+ Object[] values = new Object[numUnionColumns + 1 +
numAggregationFunctions];
+ System.arraycopy(keyValues, 0, values, 0, numUnionColumns + 1);
+ for (int i = 0; i < numAggregationFunctions; i++) {
+ Object intermediate = baseValues[numUnionColumns + i];
+ values[numUnionColumns + 1 + i] =
+ fullUnion ? intermediate :
cloneIntermediate(aggregationFunctions[i], intermediate);
+ }
+ taskMap.put(new Key(keyValues), new Record(values));
+ } else {
+ Object[] values = existing.getValues();
+ for (int i = 0; i < numAggregationFunctions; i++) {
+ int valueIndex = numUnionColumns + 1 + i;
+ // The first argument is this set's owned accumulator (mutating it
is safe). The second argument is
+ // cloned because AggregationFunction#merge may RETURN its second
argument as the new accumulator
+ // (e.g. HyperLogLog merge with mismatched sizes), which would
then be mutated by later merges while
+ // other tasks are still reading the shared base intermediate.
+ values[valueIndex] =
AggregationFunctionUtils.merge(aggregationFunctions[i], values[valueIndex],
+ cloneIntermediate(aggregationFunctions[i],
baseValues[numUnionColumns + i]));
+ }
+ }
+ }
+ }
+ return taskMap;
+ }
+
+ private static void cancelAll(List<Future<Map<Key, Record>>> futures) {
+ for (Future<Map<Key, Record>> future : futures) {
+ future.cancel(true);
+ }
+ }
+
+ /// Returns `schema` with a synthetic `$groupingId` INT column inserted at
`index` (after the union group-by
+ /// columns), producing the grouping-set output schema from the
base-grouping schema.
+ private static DataSchema insertGroupingIdColumn(DataSchema baseSchema, int
index) {
+ String[] baseNames = baseSchema.getColumnNames();
+ ColumnDataType[] baseTypes = baseSchema.getColumnDataTypes();
+ int numColumns = baseNames.length + 1;
+ String[] names = new String[numColumns];
+ ColumnDataType[] types = new ColumnDataType[numColumns];
+ System.arraycopy(baseNames, 0, names, 0, index);
+ System.arraycopy(baseTypes, 0, types, 0, index);
+ names[index] = GroupingSets.GROUPING_ID_COLUMN;
+ types[index] = ColumnDataType.INT;
+ System.arraycopy(baseNames, index, names, index + 1, baseNames.length -
index);
+ System.arraycopy(baseTypes, index, types, index + 1, baseTypes.length -
index);
+ return new DataSchema(names, types);
+ }
+
+ /// Returns a defensive copy of an aggregation intermediate result so that a
subsequent
+ /// [AggregationFunction#merge] into a derived grouping-set group cannot
mutate a base group's shared
+ /// accumulator. Scalar (non-OBJECT) intermediates are immutable boxed
values and are returned as-is; `null`
+ /// (nothing aggregated) is the merge identity and needs no copy; OBJECT
accumulators are cloned via the
+ /// function's own serialize/deserialize round-trip.
+ private static Object cloneIntermediate(AggregationFunction
aggregationFunction, Object intermediate) {
+ if (intermediate == null ||
aggregationFunction.getIntermediateResultColumnType() != ColumnDataType.OBJECT)
{
+ return intermediate;
+ }
+ SerializedIntermediateResult serialized =
aggregationFunction.serializeIntermediateResult(intermediate);
Review Comment:
**Blocker: data race on shared base intermediates**
The derive tasks serialize the same base object at the same time. For some
functions, serialization changes the object:
- `PercentileTDigestAccumulator.serialize()` calls `compress()`, which
rewrites the centroid arrays. Its Javadoc says that it is not safe for
concurrent use.
- `ThetaSketchAccumulator`, `CpcSketchAccumulator` and
`TupleIntSketchAccumulator` flush in `getResult()`: they union and clear the
pending list.
Repro: 3 real segments, 2 threads, `SELECT a, b, c, PERCENTILETDIGEST(m, 50)
FROM t GROUP BY CUBE(a, b, c)`. 10 of 10 runs fail with error 450 (`Caught
exception while deriving grouping sets`). The root causes include `Invalid
TDigest centroid`, `BufferOverflowException` and
`ArrayIndexOutOfBoundsException`. `DISTINCTCOUNTTHETASKETCH` fails 5 of 10 runs
(`ConcurrentModificationException`). With `groupingSetsBaseAggregation=false`,
all runs pass.
This is the default path for ROLLUP(a, b) or larger, and for CUBE, on a
server with 2 or more segments.
Fix idea: serialize each OBJECT base intermediate one time, on one thread
(or split by base-group ranges). Then the tasks deserialize only from those
bytes. This also cuts the serde work.
Related, not a race: with `ExprMinMaxRewriter` on, `EXPR_MAX` fails 10 of 10
runs on this path, also with 1 thread. The deserialized `ExprMinMaxObject` is
immutable, and the next `toBytes()` or merge throws an NPE.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java:
##########
@@ -172,6 +192,36 @@ protected void processSegments() {
}
}
+ private IndexedTable ensureIndexedTable(GroupByResultsBlock resultsBlock) {
+ IndexedTable indexedTable = _indexedTable;
+ if (indexedTable == null) {
+ synchronized (this) {
+ indexedTable = _indexedTable;
+ if (indexedTable == null) {
+ indexedTable =
GroupByUtils.createIndexedTableForCombineOperator(resultsBlock, _queryContext,
_numTasks,
+ _executorService);
+ _indexedTable = indexedTable;
+ }
+ }
+ }
+ return indexedTable;
+ }
+
+ private IndexedTable ensureBaseIndexedTable(GroupByResultsBlock
resultsBlock) {
+ IndexedTable baseIndexedTable = _baseIndexedTable;
+ if (baseIndexedTable == null) {
+ synchronized (this) {
+ baseIndexedTable = _baseIndexedTable;
+ if (baseIndexedTable == null) {
+ baseIndexedTable =
GroupByUtils.createIndexedTableForCombineOperator(resultsBlock, _queryContext,
_numTasks,
Review Comment:
**Blocker: the server combine still drops base groups, so totals are wrong**
The new gate is per segment. But this base table gets `resultSize =
numGroupsLimit` for the whole server (`createIndexedTableForCombineOperator`).
Without ORDER BY, it stops taking new keys when it is full. A dropped base
group is lost from every derived set, including the grand total. The expansion
path keeps the grand total and subtotals exact, because it creates those groups
first.
Repro with default settings, one server, 12 daily segments with 10k users
each:
`SELECT day, userId, COUNT(*) FROM t GROUP BY ROLLUP(day, userId) LIMIT
1000000`
- Base path: grand total = 100,000.
- Expansion path (same as master): grand total = 120,000 = `COUNT(*)`.
Each segment passes the gate (estimate 10k), but the union is 120k.
Time-partitioned tables hit this often. The response sets
`numGroupsLimitReached`, but master returns exact totals for this query.
`testNumGroupsLimitPressureKeepsTotalsExactByDefault` does not catch this.
Both of its segments hold the same 4 base keys, and the gate sends both to the
expansion path.
Fix ideas: when the base table is full, keep the extra base records and
merge them only into derived groups that already exist (like expansion). Or
apply the gate to the whole server, not per segment.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java:
##########
@@ -252,6 +253,26 @@ public int getNumExtraGroupByKeyColumns() {
return isGroupingSets() ? 1 : 0;
}
+ /// Returns whether grouping-set queries should aggregate the base grouping
(union columns) once per segment
+ /// and derive the individual grouping-set records from those base groups,
instead of expanding every input
+ /// row into one group per grouping set. Enabled by default; the
`groupingSetsBaseAggregation=false` query
+ /// option forces the legacy per-row expansion path. See
[CommonConstants.Broker.Request.QueryOptionKey].
+ ///
+ /// Only the standard
[org.apache.pinot.core.operator.combine.GroupByCombineOperator] performs the
merge-time
+ /// derive that turns the emitted base groups back into grouping-set
records. Base aggregation is therefore
+ /// disabled whenever a different combine path would run: filtered
aggregations (distinct shared group-key
+ /// generator), the streaming combine used on MSE leaf stages
(`streamingGroupByFlushThreshold > 0`, which
+ /// flushes incrementally and cannot derive over the fully-merged base
table), and the sorted-aggregate combine
+ /// ([#shouldSortAggregateUnderSafeTrim()]). In all these cases the segment
expands per row so the emitted rows
+ /// already carry the $groupingId discriminator.
+ public boolean isGroupingSetsBaseAggregation() {
+ if (_hasFilteredAggregations || _streamingGroupByFlushThreshold > 0 ||
shouldSortAggregateUnderSafeTrim()) {
+ return false;
+ }
+ String option =
_queryOptions.get(CommonConstants.Broker.Request.QueryOptionKey.GROUPING_SETS_BASE_AGGREGATION);
+ return option == null || Boolean.parseBoolean(option);
Review Comment:
**Blocker: default-on makes some queries much slower and larger**
The gate checks only the estimated base-group count. It does not check how
many rows collapse into each base group, or how much each merge costs. Measured
on this PR: 1 segment of 200k rows, `CUBE(D1, D2, D3)`, about 80k base groups
(passes the gate). Server p50:
| Aggregation | Base path | Expansion path |
|---|---|---|
| `COUNT(*), SUM(M1)` | 53 ms | 40 ms |
| `PERCENTILEKLL(M1, 50)` | 485 ms, 4.1 GB allocated | 92 ms, 160 MB
allocated |
| `PERCENTILEEST(M1, 50)` | 2,260 ms | 428 ms |
The derive does one clone and one merge per base group for each non-full
set. For sketch results, one merge costs much more than one row add on the
expansion path.
Also, `numGroupsLimit` no longer bounds the server output. With default
settings, `CUBE` over 12 columns of cardinality 2 (estimate 4,096) sends
531,441 rows (32.7 MB). The expansion path sends 100,000 rows (6.2 MB) and sets
`numGroupsLimitReached`. `ORDER BY COUNT(*) DESC LIMIT 10` still sends all
531,441 rows.
The PR benchmark does not show this. It uses only COUNT and SUM. At head,
its high-cardinality rows run the expansion path in both arms (0 of 50 segments
take the base path).
Please keep this off by default until the gate accounts for the collapse
ratio and the merge cost, and the derived output is bounded.
--
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]
