xiangfu0 commented on code in PR #19264:
URL: https://github.com/apache/pinot/pull/19264#discussion_r4110249460
##########
pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java:
##########
@@ -87,6 +100,230 @@ 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());
+ int numTaskSlots = Math.max(1, Math.min(numTasks, numSets));
+ List<Map<Key, Record>> taskMaps;
+ if (numTaskSlots == 1) {
+ taskMaps = List.of(deriveSets(baseEntries, setContains, 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, taskIndex, numTaskSlots,
+ numUnionColumns, numAggregationFunctions, aggregationFunctions)));
+ }
+ taskMaps = new ArrayList<>(numTaskSlots);
+ try {
+ for (Future<Map<Key, Record>> future : futures) {
+ taskMaps.add(future.get());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while deriving grouping sets",
e);
+ } catch (ExecutionException e) {
+ for (Future<Map<Key, Record>> future : futures) {
+ future.cancel(true);
+ }
+ 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);
+ return buildIndexedTableFromRecords(groupingSetsSchema, queryContext,
kept, executorService);
+ }
+ 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,
+ int taskIndex, int numTaskSlots, int numUnionColumns, int
numAggregationFunctions,
+ AggregationFunction[] aggregationFunctions) {
+ int numSets = setContains.length;
+ // A single set can produce at most one derived group per base entry, so
this covers the largest owned set
+ // (typically the near-identity one) without rehashing; coarser owned sets
add far fewer distinct groups.
+ Map<Key, Record> taskMap = new
HashMap<>(HashUtil.getHashMapCapacity(baseEntries.size()));
+ // 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);
+ for (Map.Entry<Key, Record> baseEntry : baseEntries) {
Review Comment:
Fixed in 6be2ea7: `deriveSets` now calls
`QueryThreadContext.checkTerminationAndSampleUsagePeriodically` per projection
(same as the segment-merge loop), each `future.get` is bounded by the query
deadline, and the sibling futures are cancelled on interrupt, timeout and
execution failure alike.
##########
pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java:
##########
@@ -87,6 +100,230 @@ 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);
Review Comment:
Addressed in 6be2ea7 via the same cardinality gate as the totals issue: base
aggregation only runs when the estimated base-group count fits under
`groupingSetsBaseAggregationMaxGroups` (default `numGroupsLimit`), so the
derived output is bounded by that estimate x numSets only on workloads where
the base grouping provably collapses; everything else routes to expansion,
whose combine table stays capped at `numGroupsLimit` total groups. The opt-in
per-set trim (`groupingSetsMinServerTrimSize`) remains for bounding ORDER BY
queries further. If you still prefer the feature off by default for the first
release, happy to flip it — with the gate the remaining blast radius is
workloads whose base grouping fits under the group limit.
##########
pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java:
##########
@@ -87,6 +100,230 @@ 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());
+ int numTaskSlots = Math.max(1, Math.min(numTasks, numSets));
+ List<Map<Key, Record>> taskMaps;
+ if (numTaskSlots == 1) {
+ taskMaps = List.of(deriveSets(baseEntries, setContains, 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, taskIndex, numTaskSlots,
+ numUnionColumns, numAggregationFunctions, aggregationFunctions)));
+ }
+ taskMaps = new ArrayList<>(numTaskSlots);
+ try {
+ for (Future<Map<Key, Record>> future : futures) {
+ taskMaps.add(future.get());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while deriving grouping sets",
e);
+ } catch (ExecutionException e) {
+ for (Future<Map<Key, Record>> future : futures) {
+ future.cancel(true);
+ }
+ 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);
+ return buildIndexedTableFromRecords(groupingSetsSchema, queryContext,
kept, executorService);
+ }
+ 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,
+ int taskIndex, int numTaskSlots, int numUnionColumns, int
numAggregationFunctions,
+ AggregationFunction[] aggregationFunctions) {
+ int numSets = setContains.length;
+ // A single set can produce at most one derived group per base entry, so
this covers the largest owned set
+ // (typically the near-identity one) without rehashing; coarser owned sets
add far fewer distinct groups.
+ Map<Key, Record> taskMap = new
HashMap<>(HashUtil.getHashMapCapacity(baseEntries.size()));
Review Comment:
Sizing fixed in 6be2ea7: only tasks owning a full-union set presize to the
base-entry count; tasks owning only coarser sets start small and grow. On the
parallelism skew: agreed it is real (bounded by numSets, identity set
dominates). We measured the alternatives on the JMH benchmark:
range-partitioning into a shared concurrent table (the previous design) was
slower due to ConcurrentHashMap contention (stack profiles showed most derive
CPU in CHM machinery), and halving the identity-set task's hash work produced
no measurable end-to-end change, so the skew is not currently the wall-clock
bottleneck. Left as future work.
--
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]
