From 97f1130367d78f126ba7a4570c5d88324492a27a Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Tue, 7 Apr 2026 03:43:12 -0700 Subject: [PATCH 1/4] Add non-blocking and partitioned group-by combine operators Introduce two new server-level group-by combine algorithms selectable via the `groupByAlgorithm` query option: - NONBLOCKING: Uses a lock-free ConcurrentHashMap to aggregate results across segments without holding a global lock, reducing contention under high concurrency. - PARTITIONED: Hash-partitions group keys across independent tables so that each partition can be aggregated and trimmed in parallel with zero cross-partition synchronization. Includes per-partition thread-local accumulation, cached key hashing, and parallel finish/trim to minimize overhead for high-cardinality queries. Also adds: - `normalizeGroupByAlgorithm` helper in QueryOptionsUtils - `IndexedTable.transferFrom` for bulk partition merging - JMH benchmark for comparing combine operator throughput - Unit and integration tests for both new algorithms --- .../utils/config/QueryOptionsUtils.java | 5 + .../pinot/core/data/table/IndexedTable.java | 13 + .../combine/GroupByCombineOperator.java | 165 +++++----- .../NonblockingGroupByCombineOperator.java | 113 +++++++ .../pinot/core/plan/CombinePlanNode.java | 38 ++- .../apache/pinot/queries/BaseQueriesTest.java | 2 +- ...BlockingGroupBySingleValueQueriesTest.java | 294 ++++++++++++++++++ .../perf/BenchmarkGroupByCombineOperator.java | 238 ++++++++++++++ .../pinot/spi/utils/CommonConstants.java | 1 + 9 files changed, 780 insertions(+), 89 deletions(-) create mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java create mode 100644 pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java create mode 100644 pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java index 4a39ab5d8e70..1048cfc59a57 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java @@ -450,6 +450,11 @@ public static Integer getMinInitialIndexedTableCapacity(Map quer return checkedParseIntPositive(QueryOptionKey.MIN_INITIAL_INDEXED_TABLE_CAPACITY, minInitialIndexedTableCapacity); } + @Nullable + public static String getGroupByAlgorithm(Map queryOptions) { + return queryOptions.get(QueryOptionKey.GROUP_BY_ALGORITHM); + } + public static boolean shouldDropResults(Map queryOptions) { return Boolean.parseBoolean(queryOptions.get(CommonConstants.Broker.Request.QueryOptionKey.DROP_RESULTS)); } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java index 2bb526f82aee..2edb2cc5961d 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java @@ -265,9 +265,22 @@ public int size() { @Override public Iterator iterator() { + if (_topRecords == null) { + return _lookupMap.values().iterator(); + } return _topRecords.iterator(); } + + /** + * Absorbs the resize/trim statistics from another {@link IndexedTable} into this table. + * Call this after merging tables to ensure resize metrics are accurately reported. + */ + public void absorbTrimStats(IndexedTable other) { + _numResizes += other._numResizes; + _resizeTimeNs += other._resizeTimeNs; + } + public int getNumResizes() { return _numResizes; } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java index 06835ce054ce..1fff1f0a0469 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java @@ -53,6 +53,8 @@ */ @SuppressWarnings("rawtypes") public class GroupByCombineOperator extends BaseSingleBlockCombineOperator { + public static final String ALGORITHM = "DEFAULT"; + private static final Logger LOGGER = LoggerFactory.getLogger(GroupByCombineOperator.class); private static final String EXPLAIN_NAME = "COMBINE_GROUP_BY"; @@ -63,12 +65,12 @@ public class GroupByCombineOperator extends BaseSingleBlockCombineOperator operators, QueryContext queryContext, ExecutorService executorService) { super(null, operators, overrideMaxExecutionThreads(queryContext, operators.size()), executorService); @@ -115,59 +117,11 @@ protected void processSegments() { if (_indexedTable == null) { synchronized (this) { if (_indexedTable == null) { - _indexedTable = GroupByUtils.createIndexedTableForCombineOperator(resultsBlock, _queryContext, _numTasks, - _executorService); - } - } - } - - if (resultsBlock.isGroupsTrimmed()) { - _groupsTrimmed = true; - } - // Set groups limit reached flag. - if (resultsBlock.isNumGroupsLimitReached()) { - _numGroupsLimitReached = true; - } - if (resultsBlock.isNumGroupsWarningLimitReached()) { - _numGroupsWarningLimitReached = true; - } - - // Merge aggregation group-by result. - // Iterate over the group-by keys, for each key, update the group-by result in the indexedTable - Collection intermediateRecords = resultsBlock.getIntermediateRecords(); - // Count the number of merged keys - int mergedKeys = 0; - // For now, only GroupBy OrderBy query has pre-constructed intermediate records - if (intermediateRecords == null) { - // Merge aggregation group-by result. - AggregationGroupByResult aggregationGroupByResult = resultsBlock.getAggregationGroupByResult(); - if (aggregationGroupByResult != null) { - // Iterate over the group-by keys, for each key, update the group-by result in the indexedTable - try { - Iterator dicGroupKeyIterator = aggregationGroupByResult.getGroupKeyIterator(); - while (dicGroupKeyIterator.hasNext()) { - QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, EXPLAIN_NAME); - GroupKeyGenerator.GroupKey groupKey = dicGroupKeyIterator.next(); - Object[] keys = groupKey._keys; - Object[] values = Arrays.copyOf(keys, _numColumns); - int groupId = groupKey._groupId; - for (int i = 0; i < _numAggregationFunctions; i++) { - values[_numKeyColumns + i] = aggregationGroupByResult.getResultForGroupId(i, groupId); - } - _indexedTable.upsert(new Key(keys), new Record(values)); - } - } finally { - // Release the resources used by the group key generator - aggregationGroupByResult.closeGroupKeyGenerator(); + _indexedTable = createIndexedTable(resultsBlock, _numTasks); } } - } else { - for (IntermediateRecord intermediateResult : intermediateRecords) { - QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, EXPLAIN_NAME); - //TODO: change upsert api so that it accepts intermediateRecord directly - _indexedTable.upsert(intermediateResult._key, intermediateResult._record); - } } + mergeGroupByResultsBlock(_indexedTable, resultsBlock, EXPLAIN_NAME); } catch (RuntimeException e) { throw wrapOperatorException(operator, e); } finally { @@ -178,6 +132,56 @@ protected void processSegments() { } } + protected IndexedTable createIndexedTable(GroupByResultsBlock resultsBlock, int numThreads) { + return GroupByUtils.createIndexedTableForCombineOperator(resultsBlock, _queryContext, numThreads, _executorService); + } + + protected void updateCombineResultsStats(GroupByResultsBlock resultsBlock) { + if (resultsBlock.isGroupsTrimmed()) { + _groupsTrimmed = true; + } + if (resultsBlock.isNumGroupsLimitReached()) { + _numGroupsLimitReached = true; + } + if (resultsBlock.isNumGroupsWarningLimitReached()) { + _numGroupsWarningLimitReached = true; + } + } + + protected void mergeGroupByResultsBlock(IndexedTable indexedTable, GroupByResultsBlock resultsBlock, + String explainName) { + updateCombineResultsStats(resultsBlock); + + Collection intermediateRecords = resultsBlock.getIntermediateRecords(); + int mergedKeys = 0; + if (intermediateRecords == null) { + AggregationGroupByResult aggregationGroupByResult = resultsBlock.getAggregationGroupByResult(); + if (aggregationGroupByResult != null) { + try { + Iterator dicGroupKeyIterator = aggregationGroupByResult.getGroupKeyIterator(); + while (dicGroupKeyIterator.hasNext()) { + QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, explainName); + GroupKeyGenerator.GroupKey groupKey = dicGroupKeyIterator.next(); + Object[] keys = groupKey._keys; + Object[] values = Arrays.copyOf(keys, _numColumns); + int groupId = groupKey._groupId; + for (int i = 0; i < _numAggregationFunctions; i++) { + values[_numKeyColumns + i] = aggregationGroupByResult.getResultForGroupId(i, groupId); + } + indexedTable.upsert(new Key(keys), new Record(values)); + } + } finally { + aggregationGroupByResult.closeGroupKeyGenerator(); + } + } + } else { + for (IntermediateRecord intermediateResult : intermediateRecords) { + QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, explainName); + indexedTable.upsert(intermediateResult._key, intermediateResult._record); + } + } + } + @Override public void onProcessSegmentsException(Throwable t) { _processingException.compareAndSet(null, t); @@ -191,13 +195,13 @@ public void onProcessSegmentsFinish() { /** * {@inheritDoc} * - *

Combines intermediate selection result blocks from underlying operators and returns a merged one. + *

Combines intermediate group-by results produced by underlying operators and returns a merged results block. *

    *
  • - * Merges multiple intermediate selection result blocks as a merged one. + * Merges multiple intermediate group-by result blocks into a single merged result. *
    • *
  • - * Set all exceptions encountered during execution into the merged result block + * Sets all exceptions encountered during execution into the merged result block. *
    • *
*/ @@ -207,33 +211,42 @@ public BaseResultsBlock mergeResults() long timeoutMs = _queryContext.getEndTimeMs() - System.currentTimeMillis(); boolean opCompleted = _operatorLatch.await(timeoutMs, TimeUnit.MILLISECONDS); if (!opCompleted) { - // If this happens, the broker side should already timed out, just log the error and return - String userError = "Timed out while combining group-by order-by results after " + timeoutMs + "ms"; - String logMsg = userError + ", queryContext = " + _queryContext; - LOGGER.error(logMsg); - return new ExceptionResultsBlock(new QueryErrorMessage(QueryErrorCode.EXECUTION_TIMEOUT, userError, logMsg)); + return getTimeoutResultsBlock(timeoutMs); } Throwable ex = _processingException.get(); if (ex != null) { - String userError = "Caught exception while processing group-by order-by query"; - String devError = userError + ": " + ex.getMessage(); - QueryErrorMessage errMsg; - if (ex instanceof QueryException) { - // If the exception is a QueryException, use the error code from the exception and trust the error message - errMsg = new QueryErrorMessage(((QueryException) ex).getErrorCode(), devError, devError); - } else { - // If the exception is not a QueryException, use the generic error code and don't expose the exception message - errMsg = new QueryErrorMessage(QueryErrorCode.QUERY_EXECUTION, userError, devError); - } - return new ExceptionResultsBlock(errMsg); + return getExceptionResultsBlock(ex); } - if (_indexedTable.isTrimmed() && _queryContext.isUnsafeTrim()) { + return getMergedResultsBlock(_indexedTable); + } + + protected ExceptionResultsBlock getTimeoutResultsBlock(long timeoutMs) { + long reportedTimeoutMs = Math.max(timeoutMs, 0L); + String userError = "Timed out while combining group-by results after " + reportedTimeoutMs + "ms"; + String logMsg = userError + ", queryContext = " + _queryContext; + LOGGER.error(logMsg); + return new ExceptionResultsBlock(new QueryErrorMessage(QueryErrorCode.EXECUTION_TIMEOUT, userError, logMsg)); + } + + protected ExceptionResultsBlock getExceptionResultsBlock(Throwable ex) { + String userError = "Caught exception while processing group-by query"; + String devError = userError + ": " + ex.getMessage(); + QueryErrorMessage errMsg; + if (ex instanceof QueryException) { + errMsg = new QueryErrorMessage(((QueryException) ex).getErrorCode(), devError, devError); + } else { + errMsg = new QueryErrorMessage(QueryErrorCode.QUERY_EXECUTION, userError, devError); + } + return new ExceptionResultsBlock(errMsg); + } + + protected GroupByResultsBlock getMergedResultsBlock(IndexedTable indexedTable) { + if (indexedTable.isTrimmed() && _queryContext.isUnsafeTrim()) { _groupsTrimmed = true; } - IndexedTable indexedTable = _indexedTable; if (_queryContext.isServerReturnFinalResult()) { indexedTable.finish(true, true); } else if (_queryContext.isServerReturnFinalResultKeyUnpartitioned()) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java new file mode 100644 index 000000000000..47c2a4a6e43d --- /dev/null +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java @@ -0,0 +1,113 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.operator.combine; + +import java.util.List; +import java.util.concurrent.ExecutorService; +import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.data.table.IndexedTable; +import org.apache.pinot.core.operator.AcquireReleaseColumnsSegmentOperator; +import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; +import org.apache.pinot.core.query.request.context.QueryContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Non-blocking combine operator for group-by queries. + *

+ * Parallelism is bounded by the configured max execution threads via {@link GroupByCombineOperator}. + */ +@SuppressWarnings("rawtypes") +public class NonblockingGroupByCombineOperator extends GroupByCombineOperator { + public static final String ALGORITHM = "NON-BLOCKING"; + + private static final Logger LOGGER = LoggerFactory.getLogger(NonblockingGroupByCombineOperator.class); + private static final String EXPLAIN_NAME = "COMBINE_GROUP_BY"; + + public NonblockingGroupByCombineOperator(List operators, QueryContext queryContext, + ExecutorService executorService) { + super(operators, queryContext, executorService); + LOGGER.debug("Using {} for group-by combine with {} tasks", ALGORITHM, _numTasks); + } + + @Override + public String toExplainString() { + return EXPLAIN_NAME; + } + + /** + * Executes query on one segment in a worker thread and merges the results into the indexed table. + */ + @Override + protected void processSegments() { + int operatorId; + IndexedTable indexedTable = null; + while (_processingException.get() == null && (operatorId = _nextOperatorId.getAndIncrement()) < _numOperators) { + Operator operator = _operators.get(operatorId); + try { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).acquire(); + } + GroupByResultsBlock resultsBlock = (GroupByResultsBlock) operator.nextBlock(); + if (indexedTable == null) { + synchronized (this) { + if (_indexedTable != null) { + indexedTable = _indexedTable; + _indexedTable = null; + } + } + if (indexedTable == null) { + indexedTable = createIndexedTable(resultsBlock, 1); + } + } + mergeGroupByResultsBlock(indexedTable, resultsBlock, EXPLAIN_NAME); + } catch (RuntimeException e) { + throw wrapOperatorException(operator, e); + } finally { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).release(); + } + } + } + + boolean setGroupByResult = false; + while (indexedTable != null && !setGroupByResult) { + IndexedTable indexedTableToMerge; + synchronized (this) { + if (_indexedTable == null) { + _indexedTable = indexedTable; + setGroupByResult = true; + continue; + } else { + indexedTableToMerge = _indexedTable; + _indexedTable = null; + } + } + if (indexedTable.size() > indexedTableToMerge.size()) { + indexedTable.merge(indexedTableToMerge); + indexedTable.absorbTrimStats(indexedTableToMerge); + } else { + indexedTableToMerge.merge(indexedTable); + indexedTableToMerge.absorbTrimStats(indexedTable); + indexedTable = indexedTableToMerge; + } + } + } +} diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java index 195b19e90176..96e0c34c5cf7 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java @@ -24,12 +24,14 @@ import javax.annotation.Nullable; import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.common.request.context.OrderByExpressionContext; +import org.apache.pinot.common.utils.config.QueryOptionsUtils; import org.apache.pinot.core.common.Operator; import org.apache.pinot.core.operator.combine.AggregationCombineOperator; import org.apache.pinot.core.operator.combine.BaseCombineOperator; import org.apache.pinot.core.operator.combine.DistinctCombineOperator; import org.apache.pinot.core.operator.combine.GroupByCombineOperator; import org.apache.pinot.core.operator.combine.MinMaxValueBasedSelectionOrderByCombineOperator; +import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; import org.apache.pinot.core.operator.combine.SelectionOnlyCombineOperator; import org.apache.pinot.core.operator.combine.SelectionOrderByCombineOperator; import org.apache.pinot.core.operator.combine.SequentialSortedGroupByCombineOperator; @@ -65,10 +67,10 @@ public class CombinePlanNode implements PlanNode { /** * Constructor for the class. * - * @param planNodes List of underlying plan nodes - * @param queryContext Query context + * @param planNodes List of underlying plan nodes + * @param queryContext Query context * @param executorService Executor service - * @param streamer Optional results block streamer for streaming query + * @param streamer Optional results block streamer for streaming query */ public CombinePlanNode(List planNodes, QueryContext queryContext, ExecutorService executorService, @Nullable ResultsBlockStreamer streamer) { @@ -146,15 +148,7 @@ private BaseCombineOperator getCombineOperator() { // Aggregation only return new AggregationCombineOperator(operators, _queryContext, _executorService); } else { - // Sorted aggregation group-by, when safeTrim and limit is not too large - if (_queryContext.shouldSortAggregateUnderSafeTrim()) { - if (operators.size() < _queryContext.getSortAggregateSequentialCombineNumSegmentsThreshold()) { - return new SequentialSortedGroupByCombineOperator(operators, _queryContext, _executorService); - } - return new SortedGroupByCombineOperator(operators, _queryContext, _executorService); - } - // Aggregation group-by - return new GroupByCombineOperator(operators, _queryContext, _executorService); + return getGroupByCombineOperator(operators); } } else if (QueryContextUtils.isSelectionQuery(_queryContext)) { if (_queryContext.getLimit() == 0 || _queryContext.getOrderByExpressions() == null) { @@ -175,4 +169,24 @@ private BaseCombineOperator getCombineOperator() { return new DistinctCombineOperator(operators, _queryContext, _executorService); } } + + private BaseCombineOperator getGroupByCombineOperator(List operators) { + // Sorted combine path for safe-trim with small limits + if (_queryContext.shouldSortAggregateUnderSafeTrim()) { + if (operators.size() < _queryContext.getSortAggregateSequentialCombineNumSegmentsThreshold()) { + return new SequentialSortedGroupByCombineOperator(operators, _queryContext, _executorService); + } + return new SortedGroupByCombineOperator(operators, _queryContext, _executorService); + } + + // Allow fallback to the legacy ConcurrentHashMap-based operator via query option + String algo = QueryOptionsUtils.getGroupByAlgorithm(_queryContext.getQueryOptions()); + if (GroupByCombineOperator.ALGORITHM.equalsIgnoreCase(algo)) { + return new GroupByCombineOperator(operators, _queryContext, _executorService); + } + + // Default to non-blocking combine which uses per-thread SimpleIndexedTables + // to eliminate ConcurrentHashMap contention (1.8x-3.1x faster than the legacy DEFAULT) + return new NonblockingGroupByCombineOperator(operators, _queryContext, _executorService); + } } diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/BaseQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/BaseQueriesTest.java index 6144f556bf21..657ce4ffdcfa 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/BaseQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/BaseQueriesTest.java @@ -174,7 +174,7 @@ protected BrokerResponseNative getBrokerResponse(@Language("sql") String query, * This can be particularly useful to test statistical aggregation functions. * @see StatisticalQueriesTest for an example use case. */ - private BrokerResponseNative getBrokerResponse(@Language("sql") String query, PlanMaker planMaker, + protected BrokerResponseNative getBrokerResponse(@Language("sql") String query, PlanMaker planMaker, @Nullable Map extraQueryOptions) { PinotQuery pinotQuery = CalciteSqlParser.compileToPinotQuery(query); if (extraQueryOptions != null) { diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java new file mode 100644 index 000000000000..9e890342de33 --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java @@ -0,0 +1,294 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.queries; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + + +/** + * Tests inter-segment GROUP BY queries with ORDER BY when using the non-blocking combine operator. + * Verifies ResultTable responses for both standard and trim-enabled execution paths. + */ +public class InterSegmentNonBlockingGroupBySingleValueQueriesTest extends BaseSingleValueQueriesTest { + private static final InstancePlanMakerImplV2 TRIM_ENABLED_PLAN_MAKER = new InstancePlanMakerImplV2(); + + static { + TRIM_ENABLED_PLAN_MAKER.setMinSegmentGroupTrimSize(1); + } + + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderBy(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(query, Map.of("groupByAlgorithm", "non-blocking")), + 120000L, 0L, expectedNumEntriesScannedPostFilter, + 120000L, expectedResultTable); + } + + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderByWithTrim(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, TRIM_ENABLED_PLAN_MAKER, Map.of("groupByAlgorithm", "non-blocking")), 120000L, 0L, + expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); + } + + /** + * Provides various combinations of order by in ResultTable. + * In order to calculate the expected results, the results from a group by were taken, and then ordered accordingly. + */ + @DataProvider + public Object[][] groupByOrderByDataProvider() { + List entries = new ArrayList<>(); + + // order by one of the group by columns + String query = "SELECT column11, SUM(column1) FROM testTable GROUP BY column11 ORDER BY column11"; + DataSchema dataSchema = new DataSchema(new String[]{"column11", "sum(column1)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + List results = Arrays.asList(new Object[]{"", 5935285005452.0}, new Object[]{"P", 88832999206836.0}, + new Object[]{"gFuH", 63202785888.0}, new Object[]{"o", 18105331533948.0}, new Object[]{"t", 16331923219264.0}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // order by one of the group by columns DESC + query = "SELECT column11, sum(column1) FROM testTable GROUP BY column11 ORDER BY column11 DESC"; + results = new ArrayList<>(results); + Collections.reverse(results); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // order by one of the group by columns, LIMIT less than default + query = "SELECT column11, Sum(column1) FROM testTable GROUP BY column11 ORDER BY column11 LIMIT 3"; + results = new ArrayList<>(results); + Collections.reverse(results); + results = results.subList(0, 3); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // group by 2 dimensions, order by both, tie breaker + query = "SELECT column11, column12, SUM(column1) FROM testTable" + + " GROUP BY column11, column12 ORDER BY column11, column12"; + dataSchema = new DataSchema(new String[]{"column11", "column12", "sum(column1)"}, new ColumnDataType[]{ + ColumnDataType.STRING, ColumnDataType.STRING, ColumnDataType.DOUBLE + }); + results = Arrays.asList(new Object[]{"", "HEuxNvH", 3789390396216.0}, + new Object[]{"", "KrNxpdycSiwoRohEiTIlLqDHnx", 733802350944.0}, + new Object[]{"", "MaztCmmxxgguBUxPti", 1333941430664.0}, new Object[]{"", "dJWwFk", 55470665124.0000}, + new Object[]{"", "oZgnrlDEtjjVpUoFLol", 22680162504.0}, new Object[]{"P", "HEuxNvH", 21998672845052.0}, + new Object[]{"P", "KrNxpdycSiwoRohEiTIlLqDHnx", 18069909216728.0}, + new Object[]{"P", "MaztCmmxxgguBUxPti", 27177029040008.0}, + new Object[]{"P", "TTltMtFiRqUjvOG", 4462670055540.0}, new Object[]{"P", "XcBNHe", 120021767504.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // group by 2 columns, order by both, LIMIT more than default + query = "SELECT column11, column12, SUM(column1) FROM testTable" + + " GROUP BY column11, column12 ORDER BY column11, column12 LIMIT 15"; + results = new ArrayList<>(results); + results.add(new Object[]{"P", "dJWwFk", 6224665921376.0}); + results.add(new Object[]{"P", "fykKFqiw", 1574451324140.0}); + results.add(new Object[]{"P", "gFuH", 860077643636.0}); + results.add(new Object[]{"P", "oZgnrlDEtjjVpUoFLol", 8345501392852.0}); + results.add(new Object[]{"gFuH", "HEuxNvH", 29872400856.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // group by 2 columns, order by both, one of them DESC + query = "SELECT column11, column12, SUM(column1) FROM testTable" + + " GROUP BY column11, column12 ORDER BY column11, column12 DESC"; + results = Arrays.asList(new Object[]{"", "oZgnrlDEtjjVpUoFLol", 22680162504.0}, + new Object[]{"", "dJWwFk", 55470665124.0000}, new Object[]{"", "MaztCmmxxgguBUxPti", 1333941430664.0}, + new Object[]{"", "KrNxpdycSiwoRohEiTIlLqDHnx", 733802350944.0}, new Object[]{"", "HEuxNvH", 3789390396216.0}, + new Object[]{"P", "oZgnrlDEtjjVpUoFLol", 8345501392852.0}, new Object[]{"P", "gFuH", 860077643636.0}, + new Object[]{"P", "fykKFqiw", 1574451324140.0}, new Object[]{"P", "dJWwFk", 6224665921376.0}, + new Object[]{"P", "XcBNHe", 120021767504.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // order by group by column and an aggregation + query = "SELECT column11, column12, SUM(column1) FROM testTable GROUP BY column11, column12 ORDER BY column11, sum" + + "(column1)"; + results = Arrays.asList(new Object[]{"", "oZgnrlDEtjjVpUoFLol", 22680162504.0}, + new Object[]{"", "dJWwFk", 55470665124.0000}, new Object[]{"", "KrNxpdycSiwoRohEiTIlLqDHnx", 733802350944.0}, + new Object[]{"", "MaztCmmxxgguBUxPti", 1333941430664.0}, new Object[]{"", "HEuxNvH", 3789390396216.0}, + new Object[]{"P", "XcBNHe", 120021767504.0}, new Object[]{"P", "gFuH", 860077643636.0}, + new Object[]{"P", "fykKFqiw", 1574451324140.0}, new Object[]{"P", "TTltMtFiRqUjvOG", 4462670055540.0}, + new Object[]{"P", "dJWwFk", 6224665921376.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // order by only aggregation, DESC, LIMIT + query = "SELECT column11, column12, SUM(column1) FROM testTable" + + " GROUP BY column11, column12 ORDER BY SUM(column1) DESC LIMIT 50"; + results = Arrays.asList(new Object[]{"P", "MaztCmmxxgguBUxPti", 27177029040008.0}, + new Object[]{"P", "HEuxNvH", 21998672845052.0}, + new Object[]{"P", "KrNxpdycSiwoRohEiTIlLqDHnx", 18069909216728.0}, + new Object[]{"P", "oZgnrlDEtjjVpUoFLol", 8345501392852.0}, + new Object[]{"o", "MaztCmmxxgguBUxPti", 6905624581072.0}, new Object[]{"P", "dJWwFk", 6224665921376.0}, + new Object[]{"o", "HEuxNvH", 5026384681784.0}, new Object[]{"t", "MaztCmmxxgguBUxPti", 4492405624940.0}, + new Object[]{"P", "TTltMtFiRqUjvOG", 4462670055540.0}, new Object[]{"t", "HEuxNvH", 4424489490364.0}, + new Object[]{"o", "KrNxpdycSiwoRohEiTIlLqDHnx", 4051812250524.0}, new Object[]{"", "HEuxNvH", 3789390396216.0}, + new Object[]{"t", "KrNxpdycSiwoRohEiTIlLqDHnx", 3529048341192.0}, + new Object[]{"P", "fykKFqiw", 1574451324140.0}, new Object[]{"t", "dJWwFk", 1349058948804.0}, + new Object[]{"", "MaztCmmxxgguBUxPti", 1333941430664.0}, new Object[]{"o", "dJWwFk", 1152689463360.0}, + new Object[]{"t", "oZgnrlDEtjjVpUoFLol", 1039101333316.0}, new Object[]{"P", "gFuH", 860077643636.0}, + new Object[]{"", "KrNxpdycSiwoRohEiTIlLqDHnx", 733802350944.0}, + new Object[]{"o", "oZgnrlDEtjjVpUoFLol", 699381633640.0}, new Object[]{"t", "TTltMtFiRqUjvOG", 675238030848.0}, + new Object[]{"t", "fykKFqiw", 480973878052.0}, new Object[]{"t", "gFuH", 330331507792.0}, + new Object[]{"o", "TTltMtFiRqUjvOG", 203835153352.0}, new Object[]{"P", "XcBNHe", 120021767504.0}, + new Object[]{"o", "fykKFqiw", 62975165296.0}, new Object[]{"", "dJWwFk", 55470665124.0000}, + new Object[]{"gFuH", "HEuxNvH", 29872400856.0}, new Object[]{"gFuH", "MaztCmmxxgguBUxPti", 29170832184.0}, + new Object[]{"", "oZgnrlDEtjjVpUoFLol", 22680162504.0}, new Object[]{"t", "XcBNHe", 11276063956.0}, + new Object[]{"gFuH", "KrNxpdycSiwoRohEiTIlLqDHnx", 4159552848.0}, new Object[]{"o", "gFuH", 2628604920.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // multiple aggregations (group-by column not in select) + query = "SELECT sum(column1), MIN(column6) FROM testTable GROUP BY column11 ORDER BY column11"; + dataSchema = new DataSchema(new String[]{"sum(column1)", "min(column6)"}, + new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); + results = Arrays.asList(new Object[]{5935285005452.0, 2.96467636E8}, new Object[]{88832999206836.0, 1689277.0}, + new Object[]{63202785888.0, 2.96467636E8}, new Object[]{18105331533948.0, 2.96467636E8}, + new Object[]{16331923219264.0, 1980174.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // order by aggregation with space/tab in order by + query = "SELECT column11, column12, SUM(column1) FROM testTable" + + " GROUP BY column11, column12 ORDER BY SUM (\tcolumn1) DESC LIMIT 3"; + dataSchema = new DataSchema(new String[]{"column11", "column12", "sum(column1)"}, new ColumnDataType[]{ + ColumnDataType.STRING, ColumnDataType.STRING, ColumnDataType.DOUBLE + }); + results = Arrays.asList(new Object[]{"P", "MaztCmmxxgguBUxPti", 27177029040008.0}, + new Object[]{"P", "HEuxNvH", 21998672845052.0}, + new Object[]{"P", "KrNxpdycSiwoRohEiTIlLqDHnx", 18069909216728.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // order by an aggregation DESC, and group by column + query = "SELECT column12, MIN(column6) FROM testTable GROUP BY column12 ORDER BY Min(column6) DESC, column12"; + dataSchema = new DataSchema(new String[]{"column12", "min(column6)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + results = Arrays.asList(new Object[]{"XcBNHe", 329467557.0}, new Object[]{"fykKFqiw", 296467636.0}, + new Object[]{"gFuH", 296467636.0}, new Object[]{"HEuxNvH", 6043515.0}, + new Object[]{"MaztCmmxxgguBUxPti", 6043515.0}, new Object[]{"dJWwFk", 6043515.0}, + new Object[]{"KrNxpdycSiwoRohEiTIlLqDHnx", 1980174.0}, new Object[]{"TTltMtFiRqUjvOG", 1980174.0}, + new Object[]{"oZgnrlDEtjjVpUoFLol", 1689277.0}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // aggregation in order-by but not in select + query = "SELECT column12 FROM testTable GROUP BY column12 ORDER BY Min(column6) DESC, column12"; + dataSchema = new DataSchema(new String[]{"column12"}, new ColumnDataType[]{ColumnDataType.STRING}); + results = + Arrays.asList(new Object[]{"XcBNHe"}, new Object[]{"fykKFqiw"}, new Object[]{"gFuH"}, new Object[]{"HEuxNvH"}, + new Object[]{"MaztCmmxxgguBUxPti"}, new Object[]{"dJWwFk"}, new Object[]{"KrNxpdycSiwoRohEiTIlLqDHnx"}, + new Object[]{"TTltMtFiRqUjvOG"}, new Object[]{"oZgnrlDEtjjVpUoFLol"}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // multiple aggregations in order-by but not in select + query = "SELECT column12 FROM testTable GROUP BY column12 ORDER BY Min(column6) DESC, SUM(column1) LIMIT 3"; + dataSchema = new DataSchema(new String[]{"column12"}, new ColumnDataType[]{ColumnDataType.STRING}); + results = Arrays.asList(new Object[]{"XcBNHe"}, new Object[]{"gFuH"}, new Object[]{"fykKFqiw"}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // multiple aggregations in order-by, some in select + query = "SELECT column12, MIN(column6) FROM testTable" + + " GROUP BY column12 ORDER BY Min(column6) DESC, SUM(column1) LIMIT 3"; + dataSchema = new DataSchema(new String[]{"column12", "min(column6)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + results = Arrays.asList(new Object[]{"XcBNHe", 329467557.0}, new Object[]{"gFuH", 296467636.0}, + new Object[]{"fykKFqiw", 296467636.0}); + entries.add(new Object[]{query, 360000L, new ResultTable(dataSchema, results)}); + + // numeric dimension should follow numeric ordering + query = "select column17, count(*) from testTable group by column17 order by column17 limit 15"; + dataSchema = new DataSchema(new String[]{"column17", "count(*)"}, + new ColumnDataType[]{ColumnDataType.INT, ColumnDataType.LONG}); + results = + Arrays.asList(new Object[]{83386499, 2924L}, new Object[]{217787432, 3892L}, new Object[]{227908817, 6564L}, + new Object[]{402773817, 7304L}, new Object[]{423049234, 6556L}, new Object[]{561673250, 7420L}, + new Object[]{635942547, 3308L}, new Object[]{638936844, 3816L}, new Object[]{939479517, 3116L}, + new Object[]{984091268, 3824L}, new Object[]{1230252339, 5620L}, new Object[]{1284373442, 7428L}, + new Object[]{1555255521, 2900L}, new Object[]{1618904660, 2744L}, new Object[]{1670085862, 3388L}); + entries.add(new Object[]{query, 120000L, new ResultTable(dataSchema, results)}); + + // group by UDF order by UDF + query = "SELECT sub(column1, 100000), COUNT(*) FROM testTable" + + " GROUP BY sub(column1, 100000) ORDER BY sub(column1, 100000) LIMIT 3"; + dataSchema = new DataSchema(new String[]{"sub(column1,'100000')", "count(*)"}, + new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.LONG}); + results = Arrays.asList(new Object[]{140528.0, 28L}, new Object[]{194355.0, 12L}, new Object[]{532157.0, 12L}); + entries.add(new Object[]{query, 120000L, new ResultTable(dataSchema, results)}); + + // space/tab in UDF + query = + "SELECT sub(column1, 100000), COUNT(*) FROM testTable GROUP BY sub(column1, 100000) ORDER BY SUB( column1, " + + "100000\t) LIMIT 3"; + dataSchema = new DataSchema(new String[]{"sub(column1,'100000')", "count(*)"}, + new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.LONG}); + results = Arrays.asList(new Object[]{140528.0, 28L}, new Object[]{194355.0, 12L}, new Object[]{532157.0, 12L}); + entries.add(new Object[]{query, 120000L, new ResultTable(dataSchema, results)}); + + // Object type aggregation - comparable intermediate results (AVG, MINMAXRANGE) + query = "SELECT column11, AVG(column6) FROM testTable GROUP BY column11 ORDER BY column11"; + dataSchema = new DataSchema(new String[]{"column11", "avg(column6)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + results = Arrays.asList(new Object[]{"", 296467636.0}, new Object[]{"P", 909380310.3521485}, + new Object[]{"gFuH", 296467636.0}, new Object[]{"o", 296467636.0}, new Object[]{"t", 526245333.3900426}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + query = "SELECT column11, AVG(column6) FROM testTable GROUP BY column11 ORDER BY AVG(column6), column11 DESC"; + dataSchema = new DataSchema(new String[]{"column11", "avg(column6)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + results = + Arrays.asList(new Object[]{"o", 296467636.0}, new Object[]{"gFuH", 296467636.0}, new Object[]{"", 296467636.0}, + new Object[]{"t", 526245333.3900426}, new Object[]{"P", 909380310.3521485}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // Object type aggregation - non comparable intermediate results (DISTINCTCOUNT) + query = "SELECT column12, DISTINCTCOUNT(column11) FROM testTable GROUP BY column12 ORDER BY column12"; + dataSchema = new DataSchema(new String[]{"column12", "distinctcount(column11)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.INT}); + results = Arrays.asList(new Object[]{"HEuxNvH", 5}, new Object[]{"KrNxpdycSiwoRohEiTIlLqDHnx", 5}, + new Object[]{"MaztCmmxxgguBUxPti", 5}, new Object[]{"TTltMtFiRqUjvOG", 3}, new Object[]{"XcBNHe", 2}, + new Object[]{"dJWwFk", 4}, new Object[]{"fykKFqiw", 3}, new Object[]{"gFuH", 3}, + new Object[]{"oZgnrlDEtjjVpUoFLol", 4}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + query = "SELECT column12, DISTINCTCOUNT(column11) FROM testTable" + + " GROUP BY column12 ORDER BY DistinctCount(column11), column12 DESC"; + dataSchema = new DataSchema(new String[]{"column12", "distinctcount(column11)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.INT}); + results = Arrays.asList(new Object[]{"XcBNHe", 2}, new Object[]{"gFuH", 3}, new Object[]{"fykKFqiw", 3}, + new Object[]{"TTltMtFiRqUjvOG", 3}, new Object[]{"oZgnrlDEtjjVpUoFLol", 4}, new Object[]{"dJWwFk", 4}, + new Object[]{"MaztCmmxxgguBUxPti", 5}, new Object[]{"KrNxpdycSiwoRohEiTIlLqDHnx", 5}, + new Object[]{"HEuxNvH", 5}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + // percentile + query = "SELECT column11, percentile90(column6) FROM testTable" + + " GROUP BY column11 ORDER BY PERCENTILE90(column6), column11 LIMIT 3"; + results = Arrays.asList(new Object[]{"", 2.96467636E8}, new Object[]{"gFuH", 2.96467636E8}, + new Object[]{"o", 2.96467636E8}); + dataSchema = new DataSchema(new String[]{"column11", "percentile90(column6)"}, + new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE}); + entries.add(new Object[]{query, 240000L, new ResultTable(dataSchema, results)}); + + return entries.toArray(new Object[0][]); + } +} diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java new file mode 100644 index 000000000000..979c68c21cab --- /dev/null +++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java @@ -0,0 +1,238 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.perf; + +import java.lang.reflect.Constructor; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.data.table.IntermediateRecord; +import org.apache.pinot.core.data.table.Key; +import org.apache.pinot.core.data.table.Record; +import org.apache.pinot.core.operator.BaseOperator; +import org.apache.pinot.core.operator.ExecutionStatistics; +import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock; +import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; +import org.apache.pinot.core.operator.combine.BaseCombineOperator; +import org.apache.pinot.core.operator.combine.GroupByCombineOperator; +import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; +import org.apache.pinot.core.query.request.context.QueryContext; +import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.infra.Blackhole; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.options.ChainedOptionsBuilder; +import org.openjdk.jmh.runner.options.OptionsBuilder; +import org.openjdk.jmh.runner.options.TimeValue; + + +/** + * Benchmarks end-to-end server-side group-by combine operators on synthetic segment-level result blocks. + * + *

The benchmark materializes fresh intermediate records per invocation so that repeated runs do not reuse mutable + * records from prior combines. + */ +@State(Scope.Benchmark) +@Fork(value = 1, jvmArgs = {"-server", "-Xms8G", "-Xmx8G", "-XX:MaxDirectMemorySize=8G"}) +public class BenchmarkGroupByCombineOperator { + private static final String QUERY = + "SELECT d1, d2, SUM(m1), MAX(m2) FROM testTable GROUP BY d1, d2 ORDER BY SUM(m1) DESC LIMIT 500"; + private static final int NUM_SEGMENTS = 32; + private static final int CARDINALITY_D1 = 4096; + private static final int CARDINALITY_D2 = 2048; + private static final long QUERY_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(10); + + @Param({ + GroupByCombineOperator.ALGORITHM, + NonblockingGroupByCombineOperator.ALGORITHM + }) + private String _algorithm; + + @Param({"10000", "100000", "500000"}) + private int _numRecordsPerSegment; + + @Param({"8"}) + private int _numThreads; + + private final String[] _d1Values = new String[CARDINALITY_D1]; + private final Integer[] _d2Values = new Integer[CARDINALITY_D2]; + + private DataSchema _dataSchema; + private SegmentData[] _segmentData; + private Constructor _intermediateRecordConstructor; + private ExecutorService _executorService; + + private List _operators; + private QueryContext _queryContext; + + @Setup(Level.Trial) + public void setup() + throws NoSuchMethodException { + for (int i = 0; i < CARDINALITY_D1; i++) { + _d1Values[i] = "d1_" + i; + } + for (int i = 0; i < CARDINALITY_D2; i++) { + _d2Values[i] = i; + } + + _dataSchema = new DataSchema(new String[]{"d1", "d2", "sum(m1)", "max(m2)"}, + new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.INT, + DataSchema.ColumnDataType.DOUBLE, DataSchema.ColumnDataType.DOUBLE}); + _segmentData = new SegmentData[NUM_SEGMENTS]; + Random random = new Random(8675309L); + for (int i = 0; i < NUM_SEGMENTS; i++) { + _segmentData[i] = new SegmentData(_numRecordsPerSegment); + for (int j = 0; j < _numRecordsPerSegment; j++) { + _segmentData[i]._d1Ids[j] = random.nextInt(CARDINALITY_D1); + _segmentData[i]._d2Ids[j] = random.nextInt(CARDINALITY_D2); + _segmentData[i]._sumValues[j] = random.nextInt(1000); + _segmentData[i]._maxValues[j] = random.nextInt(1000); + } + } + + _intermediateRecordConstructor = + IntermediateRecord.class.getDeclaredConstructor(Key.class, Record.class, Comparable[].class); + _intermediateRecordConstructor.setAccessible(true); + _executorService = Executors.newFixedThreadPool(_numThreads); + } + + @Setup(Level.Invocation) + public void setupInvocation() + throws Exception { + _queryContext = QueryContextConverterUtils.getQueryContext(QUERY); + _queryContext.setEndTimeMs(System.currentTimeMillis() + QUERY_TIMEOUT_MS); + _queryContext.setMaxExecutionThreads(_numThreads); + + _operators = new ArrayList<>(NUM_SEGMENTS); + for (SegmentData segmentData : _segmentData) { + _operators.add(new StaticGroupByOperator( + new GroupByResultsBlock(_dataSchema, createIntermediateRecords(segmentData), _queryContext))); + } + } + + @TearDown(Level.Trial) + public void tearDown() { + _executorService.shutdownNow(); + } + + @Benchmark + @BenchmarkMode(Mode.AverageTime) + @OutputTimeUnit(TimeUnit.MILLISECONDS) + public void combineGroupBy(Blackhole blackhole) { + BaseCombineOperator combineOperator = createCombineOperator(); + BaseResultsBlock resultsBlock = (BaseResultsBlock) combineOperator.nextBlock(); + if (!(resultsBlock instanceof GroupByResultsBlock)) { + throw new IllegalStateException("Expected GroupByResultsBlock but got " + resultsBlock.getClass().getName() + + " with errors: " + resultsBlock.getErrorMessages()); + } + GroupByResultsBlock groupByResultsBlock = (GroupByResultsBlock) resultsBlock; + blackhole.consume(groupByResultsBlock.getNumRows()); + blackhole.consume(groupByResultsBlock.getResizeTimeMs()); + blackhole.consume(groupByResultsBlock.getNumResizes()); + } + + private BaseCombineOperator createCombineOperator() { + switch (_algorithm) { + case NonblockingGroupByCombineOperator.ALGORITHM: + return new NonblockingGroupByCombineOperator(_operators, _queryContext, _executorService); + default: + return new GroupByCombineOperator(_operators, _queryContext, _executorService); + } + } + + private List createIntermediateRecords(SegmentData segmentData) + throws Exception { + List intermediateRecords = new ArrayList<>(_numRecordsPerSegment); + for (int i = 0; i < _numRecordsPerSegment; i++) { + Object d1 = _d1Values[segmentData._d1Ids[i]]; + Object d2 = _d2Values[segmentData._d2Ids[i]]; + Object[] keyValues = new Object[]{d1, d2}; + intermediateRecords.add(_intermediateRecordConstructor.newInstance(new Key(keyValues), + new Record(new Object[]{d1, d2, segmentData._sumValues[i], segmentData._maxValues[i]}), null)); + } + return intermediateRecords; + } + + public static void main(String[] args) + throws Exception { + ChainedOptionsBuilder options = new OptionsBuilder().include(BenchmarkGroupByCombineOperator.class.getSimpleName()) + .warmupIterations(1).warmupTime(TimeValue.seconds(5)).measurementIterations(3) + .measurementTime(TimeValue.seconds(10)).forks(1); + new Runner(options.build()).run(); + } + + private static final class SegmentData { + private final int[] _d1Ids; + private final int[] _d2Ids; + private final double[] _sumValues; + private final double[] _maxValues; + + private SegmentData(int numRecords) { + _d1Ids = new int[numRecords]; + _d2Ids = new int[numRecords]; + _sumValues = new double[numRecords]; + _maxValues = new double[numRecords]; + } + } + + private static final class StaticGroupByOperator extends BaseOperator { + private static final String EXPLAIN_NAME = "STATIC_GROUP_BY"; + + private final GroupByResultsBlock _resultsBlock; + + private StaticGroupByOperator(GroupByResultsBlock resultsBlock) { + _resultsBlock = resultsBlock; + } + + @Override + protected GroupByResultsBlock getNextBlock() { + return _resultsBlock; + } + + @Override + public List getChildOperators() { + return List.of(); + } + + @Override + public String toExplainString() { + return EXPLAIN_NAME; + } + + @Override + public ExecutionStatistics getExecutionStatistics() { + return new ExecutionStatistics(0, 0, 0, 0); + } + } +} diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index c4b6fa71c902..770c0ec5fe90 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -970,6 +970,7 @@ public static class QueryOptionKey { // possible. public static final String OPTIMIZE_MAX_INITIAL_RESULT_HOLDER_CAPACITY = "optimizeMaxInitialResultHolderCapacity"; + public static final String GROUP_BY_ALGORITHM = "groupByAlgorithm"; // Set to true if a cursor should be returned instead of the complete result set public static final String GET_CURSOR = "getCursor"; From 77ed9aafe1cf7113df8ae37692abda107b82c5e8 Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Thu, 9 Apr 2026 04:52:29 -0700 Subject: [PATCH 2/4] Rename DEFAULT algorithm to CONCURRENT, use query option constant in tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GroupByCombineOperator.ALGORITHM: "DEFAULT" → "CONCURRENT" (descriptive name for the shared ConcurrentHashMap-based operator; "DEFAULT" was misleading since it is no longer the default) - CombinePlanNode: update comment to reference CONCURRENT - InterSegmentNonBlockingGroupBySingleValueQueriesTest: use QueryOptionKey.GROUP_BY_ALGORITHM constant instead of raw string, use NonblockingGroupByCombineOperator.ALGORITHM / GroupByCombineOperator.ALGORITHM constants instead of hardcoded strings, add tests for the CONCURRENT fallback path --- .../combine/GroupByCombineOperator.java | 2 +- .../pinot/core/plan/CombinePlanNode.java | 2 +- ...BlockingGroupBySingleValueQueriesTest.java | 30 ++++++++++++++++--- 3 files changed, 28 insertions(+), 6 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java index 1fff1f0a0469..eeee6bf21856 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java @@ -53,7 +53,7 @@ */ @SuppressWarnings("rawtypes") public class GroupByCombineOperator extends BaseSingleBlockCombineOperator { - public static final String ALGORITHM = "DEFAULT"; + public static final String ALGORITHM = "CONCURRENT"; private static final Logger LOGGER = LoggerFactory.getLogger(GroupByCombineOperator.class); private static final String EXPLAIN_NAME = "COMBINE_GROUP_BY"; diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java index 96e0c34c5cf7..38c0f32daa4d 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java @@ -186,7 +186,7 @@ private BaseCombineOperator getGroupByCombineOperator(List operators) } // Default to non-blocking combine which uses per-thread SimpleIndexedTables - // to eliminate ConcurrentHashMap contention (1.8x-3.1x faster than the legacy DEFAULT) + // to eliminate ConcurrentHashMap contention (1.8x-3.1x faster than the CONCURRENT algorithm) return new NonblockingGroupByCombineOperator(operators, _queryContext, _executorService); } } diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java index 9e890342de33..4c1646ab9dd2 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java @@ -26,7 +26,10 @@ import org.apache.pinot.common.response.broker.ResultTable; import org.apache.pinot.common.utils.DataSchema; import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.core.operator.combine.GroupByCombineOperator; +import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2; +import org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -45,16 +48,35 @@ public class InterSegmentNonBlockingGroupBySingleValueQueriesTest extends BaseSi @Test(dataProvider = "groupByOrderByDataProvider") public void testGroupByOrderBy(String query, long expectedNumEntriesScannedPostFilter, ResultTable expectedResultTable) { - QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(query, Map.of("groupByAlgorithm", "non-blocking")), - 120000L, 0L, expectedNumEntriesScannedPostFilter, - 120000L, expectedResultTable); + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, + NonblockingGroupByCombineOperator.ALGORITHM)), + 120000L, 0L, expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); } @Test(dataProvider = "groupByOrderByDataProvider") public void testGroupByOrderByWithTrim(String query, long expectedNumEntriesScannedPostFilter, ResultTable expectedResultTable) { QueriesTestUtils.testInterSegmentsResult( - getBrokerResponse(query, TRIM_ENABLED_PLAN_MAKER, Map.of("groupByAlgorithm", "non-blocking")), 120000L, 0L, + getBrokerResponse(query, TRIM_ENABLED_PLAN_MAKER, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, + NonblockingGroupByCombineOperator.ALGORITHM)), 120000L, 0L, + expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); + } + + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderByWithConcurrentAlgorithm(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, GroupByCombineOperator.ALGORITHM)), + 120000L, 0L, expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); + } + + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderByWithConcurrentAlgorithmAndTrim(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, TRIM_ENABLED_PLAN_MAKER, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, + GroupByCombineOperator.ALGORITHM)), 120000L, 0L, expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); } From 0dbe5a15b5fb9499209d34f4eb8e2887e56d94f2 Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Fri, 10 Apr 2026 05:06:56 -0700 Subject: [PATCH 3/4] Fix review findings: instance-level config, encapsulation, Javadoc - Add Server.GROUP_BY_ALGORITHM config key so operators can set a fleet-wide default algorithm without modifying individual queries (MAJOR: no server-level rollback path). InstancePlanMakerImplV2 reads it at init and falls back to NON-BLOCKING. Per-query SET groupByAlgorithm=... still overrides. - Revert GroupByCombineOperator fields back to private (only _indexedTable was needed by the subclass). Replace direct field access with three protected exchange methods: stealSharedTable(), depositSharedTable(), hasSharedTable(). NonblockingGroupByCombineOperator now uses these methods instead of reaching into the parent's internal state (MINOR: overly wide protected field scope). - Make EXPLAIN_NAME protected in GroupByCombineOperator; remove the duplicate constant and the redundant toExplainString() override from the subclass (MINOR: duplicate constant). - Add Javadoc to IndexedTable.iterator() documenting the pre-finish semantics and to absorbTrimStats() documenting the non-thread-safe ownership requirement (MINOR: undocumented contract change). --- .../pinot/core/data/table/IndexedTable.java | 16 ++++++-- .../combine/GroupByCombineOperator.java | 39 ++++++++++++++++--- .../NonblockingGroupByCombineOperator.java | 27 +++++++------ .../pinot/core/plan/CombinePlanNode.java | 5 +-- .../plan/maker/InstancePlanMakerImplV2.java | 15 ++++++- .../query/request/context/QueryContext.java | 11 ++++++ .../pinot/spi/utils/CommonConstants.java | 5 +++ 7 files changed, 90 insertions(+), 28 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java index 2edb2cc5961d..36a02b2d0978 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java @@ -263,6 +263,13 @@ public int size() { return _topRecords != null ? _topRecords.size() : _lookupMap.size(); } + /** + * Returns an iterator over all records in this table. + *

+ * If {@link #finish} has already been called, iterates over the trimmed/sorted top records. Otherwise iterates + * directly over the internal lookup map. In the pre-finish case the caller is responsible for ensuring no + * concurrent modifications are made to this table while iterating. + */ @Override public Iterator iterator() { if (_topRecords == null) { @@ -271,10 +278,13 @@ public Iterator iterator() { return _topRecords.iterator(); } - /** - * Absorbs the resize/trim statistics from another {@link IndexedTable} into this table. - * Call this after merging tables to ensure resize metrics are accurately reported. + * Accumulates the resize/trim statistics from {@code other} into this table. + *

+ * This method is not thread-safe. It must only be called when both tables are exclusively owned by the + * calling thread (i.e., neither table is concurrently modified or iterated). + * + * @param other the table whose stats are absorbed; must be exclusively owned by the caller */ public void absorbTrimStats(IndexedTable other) { _numResizes += other._numResizes; diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java index eeee6bf21856..7f41052f1852 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java @@ -56,7 +56,7 @@ public class GroupByCombineOperator extends BaseSingleBlockCombineOperator operators, QueryContext queryContext, ExecutorService executorService) { super(null, operators, overrideMaxExecutionThreads(queryContext, operators.size()), executorService); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java index 47c2a4a6e43d..bb4e4872f08f 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/NonblockingGroupByCombineOperator.java @@ -32,6 +32,11 @@ /** * Non-blocking combine operator for group-by queries. *

+ * Each worker thread builds its own {@link org.apache.pinot.core.data.table.SimpleIndexedTable} (uncontended + * {@code HashMap}) and processes segments independently. After processing, threads merge their local tables via a + * lock-free tournament exchange protocol, depositing the final merged table into the shared slot for + * {@link #mergeResults()} to collect. + *

* Parallelism is bounded by the configured max execution threads via {@link GroupByCombineOperator}. */ @SuppressWarnings("rawtypes") @@ -39,7 +44,6 @@ public class NonblockingGroupByCombineOperator extends GroupByCombineOperator { public static final String ALGORITHM = "NON-BLOCKING"; private static final Logger LOGGER = LoggerFactory.getLogger(NonblockingGroupByCombineOperator.class); - private static final String EXPLAIN_NAME = "COMBINE_GROUP_BY"; public NonblockingGroupByCombineOperator(List operators, QueryContext queryContext, ExecutorService executorService) { @@ -47,13 +51,10 @@ public NonblockingGroupByCombineOperator(List operators, QueryContext LOGGER.debug("Using {} for group-by combine with {} tasks", ALGORITHM, _numTasks); } - @Override - public String toExplainString() { - return EXPLAIN_NAME; - } - /** - * Executes query on one segment in a worker thread and merges the results into the indexed table. + * Executes query on one segment in a worker thread and merges the results into a thread-local indexed table. + * After all segments are processed the thread merges its local table into the shared slot via a lock-free + * tournament exchange. */ @Override protected void processSegments() { @@ -68,9 +69,8 @@ protected void processSegments() { GroupByResultsBlock resultsBlock = (GroupByResultsBlock) operator.nextBlock(); if (indexedTable == null) { synchronized (this) { - if (_indexedTable != null) { - indexedTable = _indexedTable; - _indexedTable = null; + if (hasSharedTable()) { + indexedTable = stealSharedTable(); } } if (indexedTable == null) { @@ -91,13 +91,12 @@ protected void processSegments() { while (indexedTable != null && !setGroupByResult) { IndexedTable indexedTableToMerge; synchronized (this) { - if (_indexedTable == null) { - _indexedTable = indexedTable; + if (!hasSharedTable()) { + depositSharedTable(indexedTable); setGroupByResult = true; continue; } else { - indexedTableToMerge = _indexedTable; - _indexedTable = null; + indexedTableToMerge = stealSharedTable(); } } if (indexedTable.size() > indexedTableToMerge.size()) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java index 38c0f32daa4d..8f2bc026d32c 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java @@ -24,7 +24,6 @@ import javax.annotation.Nullable; import org.apache.pinot.common.request.context.ExpressionContext; import org.apache.pinot.common.request.context.OrderByExpressionContext; -import org.apache.pinot.common.utils.config.QueryOptionsUtils; import org.apache.pinot.core.common.Operator; import org.apache.pinot.core.operator.combine.AggregationCombineOperator; import org.apache.pinot.core.operator.combine.BaseCombineOperator; @@ -179,8 +178,8 @@ private BaseCombineOperator getGroupByCombineOperator(List operators) return new SortedGroupByCombineOperator(operators, _queryContext, _executorService); } - // Allow fallback to the legacy ConcurrentHashMap-based operator via query option - String algo = QueryOptionsUtils.getGroupByAlgorithm(_queryContext.getQueryOptions()); + // Allow fallback to the legacy ConcurrentHashMap-based operator via instance config or query option + String algo = _queryContext.getGroupByAlgorithm(); if (GroupByCombineOperator.ALGORITHM.equalsIgnoreCase(algo)) { return new GroupByCombineOperator(operators, _queryContext, _executorService); } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java index efb7aca3755b..d2c56fa77fb6 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java @@ -115,6 +115,7 @@ public class InstancePlanMakerImplV2 implements PlanMaker { private int _minSegmentGroupTrimSize = Server.DEFAULT_QUERY_EXECUTOR_MIN_SEGMENT_GROUP_TRIM_SIZE; private int _minServerGroupTrimSize = Server.DEFAULT_QUERY_EXECUTOR_MIN_SERVER_GROUP_TRIM_SIZE; private int _groupByTrimThreshold = Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD; + private String _defaultGroupByAlgorithm = Server.DEFAULT_QUERY_EXECUTOR_GROUP_BY_ALGORITHM; @Override public void init(PinotConfiguration queryExecutorConfig) { @@ -145,11 +146,13 @@ public void init(PinotConfiguration queryExecutorConfig) { Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD); Preconditions.checkState(_groupByTrimThreshold > 0, "Invalid configurable: groupByTrimThreshold: %d must be positive", _groupByTrimThreshold); + _defaultGroupByAlgorithm = queryExecutorConfig.getProperty(Server.GROUP_BY_ALGORITHM, + Server.DEFAULT_QUERY_EXECUTOR_GROUP_BY_ALGORITHM); LOGGER.info("Initialized plan maker with maxExecutionThreads: {}, defaultExecutionThreads: {}, " + "maxInitialResultHolderCapacity: {}, numGroupsLimit: {}, minSegmentGroupTrimSize: {}, " - + "minServerGroupTrimSize: {}, groupByTrimThreshold: {}", + + "minServerGroupTrimSize: {}, groupByTrimThreshold: {}, defaultGroupByAlgorithm: {}", _maxExecutionThreads, _defaultExecutionThreads, _maxInitialResultHolderCapacity, _numGroupsLimit, - _minSegmentGroupTrimSize, _minServerGroupTrimSize, _groupByTrimThreshold); + _minSegmentGroupTrimSize, _minServerGroupTrimSize, _groupByTrimThreshold, _defaultGroupByAlgorithm); } @VisibleForTesting @@ -202,6 +205,11 @@ public void setGroupByTrimThreshold(int groupByTrimThreshold) { _groupByTrimThreshold = groupByTrimThreshold; } + @VisibleForTesting + public void setDefaultGroupByAlgorithm(String defaultGroupByAlgorithm) { + _defaultGroupByAlgorithm = defaultGroupByAlgorithm; + } + @Override public Plan makeInstancePlan(List segmentContexts, QueryContext queryContext, ExecutorService executorService) { @@ -339,6 +347,9 @@ void applyQueryOptions(QueryContext queryContext) { if (streamingGroupByFlushThreshold != null) { queryContext.setStreamingGroupByFlushThreshold(streamingGroupByFlushThreshold); } + // Set groupByAlgorithm: per-query option takes precedence over instance default + String groupByAlgorithm = QueryOptionsUtils.getGroupByAlgorithm(queryOptions); + queryContext.setGroupByAlgorithm(groupByAlgorithm != null ? groupByAlgorithm : _defaultGroupByAlgorithm); } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java index 25c03ffebda3..cad7b6abeb0a 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java @@ -129,6 +129,8 @@ public class QueryContext { private int _minServerGroupTrimSize = Server.DEFAULT_QUERY_EXECUTOR_MIN_SERVER_GROUP_TRIM_SIZE; // Trim threshold to use for server combine for SQL GROUP BY private int _groupTrimThreshold = Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD; + // Group-by combine algorithm to use on the server; null means use the instance default + private String _groupByAlgorithm; private boolean _optimizeMaxInitialResultHolderCapacity; // Number of threads to use for final reduce private int _numThreadsExtractFinalResult = InstancePlanMakerImplV2.DEFAULT_NUM_THREADS_EXTRACT_FINAL_RESULT; @@ -514,6 +516,15 @@ public void setGroupTrimThreshold(int groupTrimThreshold) { _groupTrimThreshold = groupTrimThreshold; } + @Nullable + public String getGroupByAlgorithm() { + return _groupByAlgorithm; + } + + public void setGroupByAlgorithm(String groupByAlgorithm) { + _groupByAlgorithm = groupByAlgorithm; + } + public boolean isOptimizeMaxInitialResultHolderCapacity() { return _optimizeMaxInitialResultHolderCapacity; } diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index 770c0ec5fe90..9daa9f3c91a4 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -1464,6 +1464,11 @@ public static class Server { public static final String CONFIG_OF_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD = QUERY_EXECUTOR_CONFIG_PREFIX + "." + GROUPBY_TRIM_THRESHOLD; public static final int DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD = 1_000_000; + // Server-level default group-by combine algorithm; can be overridden per query via groupByAlgorithm option + public static final String GROUP_BY_ALGORITHM = "group.by.algorithm"; + public static final String CONFIG_OF_QUERY_EXECUTOR_GROUP_BY_ALGORITHM = + QUERY_EXECUTOR_CONFIG_PREFIX + "." + GROUP_BY_ALGORITHM; + public static final String DEFAULT_QUERY_EXECUTOR_GROUP_BY_ALGORITHM = "NON-BLOCKING"; // Do sort-aggregation when LIMIT is below this threshold public static final int DEFAULT_SORT_AGGREGATE_LIMIT_THRESHOLD = 10_000; // Use sequential instead of pair-wise combine for sort-aggr when numSegments is below this threshold From 83bae97e96a601b3086c96e84d300aa2bf4ccf59 Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Thu, 23 Apr 2026 03:05:58 -0700 Subject: [PATCH 4/4] Fix PartitionedGroupByCombineOperator: Knuth hash, trim loop, and comment accuracy - Replace h^h>>>16 secondary hash with Knuth multiplicative hash (0x9E3779B9) in partitionFor() to eliminate HashMap bucket clustering in partition-local tables - Fix trim storm in maybeTrimLocalPartitions: loop until actualTotal drops below threshold instead of trimming only the largest partition once - Clarify _indexedTable comment to accurately reflect volatile DCL access pattern --- .../pinot/core/data/table/IndexedTable.java | 10 + .../org/apache/pinot/core/data/table/Key.java | 12 +- .../combine/GroupByCombineOperator.java | 43 +- .../NonblockingGroupByCombineOperator.java | 29 +- .../PartitionedGroupByCombineOperator.java | 411 ++++++++++++++++++ .../pinot/core/plan/CombinePlanNode.java | 6 +- .../apache/pinot/core/util/GroupByUtils.java | 88 +++- ...PartitionedGroupByCombineOperatorTest.java | 197 +++++++++ .../pinot/core/util/GroupByUtilsTest.java | 146 +++++++ ...BlockingGroupBySingleValueQueriesTest.java | 19 + .../perf/BenchmarkGroupByCombineOperator.java | 167 +++++-- .../pinot/spi/utils/CommonConstants.java | 10 +- 12 files changed, 1077 insertions(+), 61 deletions(-) create mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/combine/PartitionedGroupByCombineOperator.java create mode 100644 pinot-core/src/test/java/org/apache/pinot/core/data/table/PartitionedGroupByCombineOperatorTest.java diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java index 36a02b2d0978..1a324024d901 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java @@ -155,6 +155,16 @@ protected void resize() { _resizeTimeNs += resizeTimeNs; } + /** + * Trims this table to at most {@code trimSize} entries if it has an ORDER BY clause. No-op otherwise. + * Intended for coordinated trim across multiple tables (e.g., in a partitioned combine). + */ + public void trim() { + if (_hasOrderBy && !_lookupMap.isEmpty()) { + resize(); + } + } + @Override public void finish(boolean sort, boolean storeFinalResult) { if (_hasOrderBy) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java index d1115c366ccb..b8b5560cdd59 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/table/Key.java @@ -38,6 +38,11 @@ */ public class Key implements Comparable { private final Object[] _values; + // Lazily cached hash code. Declared volatile so that threads reading a Key from a shared map (e.g. + // ConcurrentHashMap in the CONCURRENT algorithm) observe the cached value written by the inserting thread. + // 0 means not yet computed; if Arrays.hashCode returns 0 the field stays 0 and is recomputed on every + // call (extremely rare, not a correctness concern). + private volatile int _hashCode; public Key(Object[] values) { _values = values; @@ -56,7 +61,12 @@ public boolean equals(Object o) { @Override public int hashCode() { - return Arrays.hashCode(_values); + int h = _hashCode; + if (h == 0) { + h = Arrays.hashCode(_values); + _hashCode = h; + } + return h; } @Override diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java index 7f41052f1852..0e44fc702d40 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java @@ -58,17 +58,22 @@ public class GroupByCombineOperator extends BaseSingleBlockCombineOperator - * Each worker thread builds its own {@link org.apache.pinot.core.data.table.SimpleIndexedTable} (uncontended - * {@code HashMap}) and processes segments independently. After processing, threads merge their local tables via a - * lock-free tournament exchange protocol, depositing the final merged table into the shared slot for - * {@link #mergeResults()} to collect. - *

- * Parallelism is bounded by the configured max execution threads via {@link GroupByCombineOperator}. + * + *

Each worker thread builds its own thread-local {@link org.apache.pinot.core.data.table.SimpleIndexedTable} + * (uncontended {@code HashMap}) and processes segments without any cross-thread contention during accumulation. + * After all segments are processed, threads merge their local tables into a single result via a lock-free tournament + * exchange protocol: each thread tries to claim the empty shared slot, or steals the current occupant and merges + * (larger absorbs smaller), looping until the winning table is deposited. + * + *

Parallelism is bounded by the configured max execution threads via {@link GroupByCombineOperator}. */ @SuppressWarnings("rawtypes") public class NonblockingGroupByCombineOperator extends GroupByCombineOperator { @@ -52,9 +53,8 @@ public NonblockingGroupByCombineOperator(List operators, QueryContext } /** - * Executes query on one segment in a worker thread and merges the results into a thread-local indexed table. - * After all segments are processed the thread merges its local table into the shared slot via a lock-free - * tournament exchange. + * Processes all assigned segments into a thread-local indexed table, then deposits it into the shared slot via the + * lock-free tournament exchange. */ @Override protected void processSegments() { @@ -68,14 +68,7 @@ protected void processSegments() { } GroupByResultsBlock resultsBlock = (GroupByResultsBlock) operator.nextBlock(); if (indexedTable == null) { - synchronized (this) { - if (hasSharedTable()) { - indexedTable = stealSharedTable(); - } - } - if (indexedTable == null) { - indexedTable = createIndexedTable(resultsBlock, 1); - } + indexedTable = createIndexedTable(resultsBlock, 1); } mergeGroupByResultsBlock(indexedTable, resultsBlock, EXPLAIN_NAME); } catch (RuntimeException e) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/PartitionedGroupByCombineOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/PartitionedGroupByCombineOperator.java new file mode 100644 index 000000000000..3e0398fe5550 --- /dev/null +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/combine/PartitionedGroupByCombineOperator.java @@ -0,0 +1,411 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.operator.combine; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.data.table.IndexedTable; +import org.apache.pinot.core.data.table.IntermediateRecord; +import org.apache.pinot.core.data.table.Key; +import org.apache.pinot.core.data.table.Record; +import org.apache.pinot.core.operator.AcquireReleaseColumnsSegmentOperator; +import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock; +import org.apache.pinot.core.operator.blocks.results.ExceptionResultsBlock; +import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; +import org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByResult; +import org.apache.pinot.core.query.aggregation.groupby.GroupKeyGenerator; +import org.apache.pinot.core.query.request.context.QueryContext; +import org.apache.pinot.core.util.GroupByUtils; +import org.apache.pinot.spi.exception.QueryErrorCode; +import org.apache.pinot.spi.exception.QueryErrorMessage; +import org.apache.pinot.spi.query.QueryThreadContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Partitioned non-blocking combine operator for group-by queries. + * + *

Records from segment results are hash-routed to {@code P} independent partitions. Each worker thread builds + * {@code P} thread-local {@link org.apache.pinot.core.data.table.SimpleIndexedTable}s (one per partition) and + * processes its assigned segments without any cross-thread contention during the accumulation phase. After all + * segments are processed, a per-partition tournament-exchange protocol (using {@code P} independent lock objects) + * merges the per-thread partition tables into {@code P} single tables. Finally, the {@code P} partition tables are + * merged into one result in {@link #mergeResults()}. + * + *

Trim strategy: each thread maintains a raw record counter (including duplicate-key upserts) as a cheap proxy for + * unique group count. When it reaches {@code _globalGroupTrimThreshold} (the same threshold a non-partitioned combine + * would use), the actual unique group count (sum of all partition table sizes) is verified first. If the actual count + * is below the threshold (duplicate-key upserts inflated the counter), the counter is reset without trimming — + * matching the resize semantics of {@link NonblockingGroupByCombineOperator}. If the actual count also exceeds the + * threshold, all partition-local tables are trimmed simultaneously via {@link IndexedTable#trim()}. This avoids the + * over-trimming that would occur if each partition trimmed independently at {@code T/P}: a hot partition that receives + * most keys would trim at {@code T/P} entries even though the global total is far below {@code T}. + * + *

Compared with {@link NonblockingGroupByCombineOperator}: + *

    + *
  • Tournament merges for different partitions can proceed in parallel (different lock objects).
    • + *
  • Final merge of {@code P} tables in {@code mergeResults()} adds a small sequential overhead.
    • + *
+ */ +@SuppressWarnings("rawtypes") +public class PartitionedGroupByCombineOperator extends GroupByCombineOperator { + public static final String ALGORITHM = "PARTITIONED-NON-BLOCKING"; + // Shadows the parent's EXPLAIN_NAME so that toExplainString() and checkTermination sampling correctly + // attribute CPU time to this operator rather than to the generic COMBINE_GROUP_BY parent. + private static final String EXPLAIN_NAME = "COMBINE_GROUP_BY_PARTITIONED"; + + private static final Logger LOGGER = LoggerFactory.getLogger(PartitionedGroupByCombineOperator.class); + + private final int _numPartitions; + // Per-partition shared result slots used for the tournament exchange; _partitionLocks[p] guards slot p. + private final IndexedTable[] _partitionedSharedTables; + private final Object[] _partitionLocks; + // The threshold at which a worker thread trims one of its partition-local tables. Equal to the effective trim + // threshold a non-partitioned combine would use (Integer.MAX_VALUE when trim is disabled). Each thread + // independently tracks its own total across all P local tables and triggers a trim when the total reaches this + // value; only the largest partition is trimmed per event (cache-friendly, same total work as trimming all P). + private final int _globalGroupTrimThreshold; + // Initial-capacity hint for each partition table: max(T / P, 0). Pre-sizing to at least T/P entries prevents + // HashMap growth (and treeification) during normal per-thread accumulation and tournament merges. + private final int _partitionCapacityHint; + // Stores the first results block seen by any worker thread; used to construct an empty table when all partitions + // are empty (all segments produced no groups). Only one write can win via compareAndSet — all blocks for a given + // query share the same DataSchema so any non-null value is acceptable. The value is read only in mergeResults(), + // which runs after awaitOperatorLatch() provides the happens-before barrier that guarantees visibility. + private final AtomicReference _firstResultsBlock = new AtomicReference<>(); + + public PartitionedGroupByCombineOperator(List operators, QueryContext queryContext, + ExecutorService executorService) { + this(operators, queryContext, executorService, Math.max(2, Math.min(operators.size(), 8))); + } + + public PartitionedGroupByCombineOperator(List operators, QueryContext queryContext, + ExecutorService executorService, int numPartitions) { + super(operators, queryContext, executorService); + _numPartitions = numPartitions; + _partitionedSharedTables = new IndexedTable[numPartitions]; + _partitionLocks = new Object[numPartitions]; + for (int i = 0; i < numPartitions; i++) { + _partitionLocks[i] = new Object(); + } + // Use the same trim threshold a non-partitioned combine would use. Each thread tracks its own total entries + // across all P local partition tables and triggers a trim of the largest partition when the total reaches + // this value. This avoids over-trimming hot partitions (which would happen if each partition trimmed at T/P). + _globalGroupTrimThreshold = GroupByUtils.getEffectiveCombineTrimThreshold(queryContext); + // Pre-size each partition table to hold at least T/P entries without any HashMap growth, which prevents + // treeification of HashMap bins during per-thread accumulation and tournament-deposit merges. + _partitionCapacityHint = + _globalGroupTrimThreshold == Integer.MAX_VALUE ? 0 : _globalGroupTrimThreshold / _numPartitions; + LOGGER.debug("Using {} for group-by combine with {} tasks and {} partitions", ALGORITHM, _numTasks, _numPartitions); + } + + @Override + public String toExplainString() { + return EXPLAIN_NAME; + } + + /** + * Processes all assigned segments, routing each group-key record to one of {@code P} partition-local tables by + * {@code (key.hashCode() & Integer.MAX_VALUE) % P}. After all segments are processed the thread performs a + * per-partition tournament deposit, merging its local table for each partition into the shared per-partition slot. + */ + @Override + protected void processSegments() { + IndexedTable[] localTables = new IndexedTable[_numPartitions]; + // Raw record counter (includes duplicate-key upserts). Used as a cheap proxy for unique group count: + // when it reaches _globalGroupTrimThreshold, maybeTrimLocalPartitions() verifies the actual unique count + // before deciding whether to trim. This matches the trim semantics of NonblockingGroupByCombineOperator. + int threadLocalTotal = 0; + + int operatorId; + while (_processingException.get() == null && (operatorId = _nextOperatorId.getAndIncrement()) < _numOperators) { + Operator operator = _operators.get(operatorId); + try { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).acquire(); + } + GroupByResultsBlock resultsBlock = (GroupByResultsBlock) operator.nextBlock(); + _firstResultsBlock.compareAndSet(null, resultsBlock); + updateCombineResultsStats(resultsBlock); + threadLocalTotal = mergeIntoPartitions(localTables, resultsBlock, threadLocalTotal); + } catch (RuntimeException e) { + throw wrapOperatorException(operator, e); + } finally { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).release(); + } + } + } + + // Per-partition tournament: each local table competes for its partition slot + for (int p = 0; p < _numPartitions; p++) { + if (localTables[p] != null) { + depositPartition(localTables, p); + } + } + } + + /** + * Routes each record in {@code resultsBlock} to the appropriate partition table in {@code localTables}, creating + * the partition table lazily on first write. The initial-capacity hint is scaled by {@code 1 / _numPartitions} + * since each partition table holds approximately that fraction of the total groups. + * + *

The {@code threadLocalTotal} counter tracks the total records seen across all partition-local tables since + * the last trim (or since the thread started). When the counter reaches {@code _globalGroupTrimThreshold}, the + * actual unique group count (sum of all partition table sizes) is compared against the threshold. This + * two-phase check ensures trim fires on unique group count, not raw record count — matching the + * behavior of {@link NonblockingGroupByCombineOperator}. Records that update existing groups (duplicates) + * increment the counter cheaply but do not trigger a trim unless the actual unique count is also high. + * + * @return the updated {@code threadLocalTotal} after processing this results block + */ + private int mergeIntoPartitions(IndexedTable[] localTables, GroupByResultsBlock resultsBlock, int threadLocalTotal) { + // Scale the numGroups hint: each partition holds ~1/P of the total groups. + // Floor at _partitionCapacityHint (= T/P) so the underlying HashMap is pre-sized to hold all expected + // groups without growth, preventing HashMap bin treeification during accumulation and tournament merges. + // Guard getNumGroups(): it asserts that at least one of aggregationGroupByResult / intermediateRecords is + // non-null; use _partitionCapacityHint as the sole hint for blocks that carry neither (e.g. empty blocks). + Collection intermediateRecords = resultsBlock.getIntermediateRecords(); + AggregationGroupByResult aggregationGroupByResult = + intermediateRecords == null ? resultsBlock.getAggregationGroupByResult() : null; + int numGroupsHint = (intermediateRecords != null || aggregationGroupByResult != null) + ? Math.max(_partitionCapacityHint, resultsBlock.getNumGroups() / _numPartitions) : _partitionCapacityHint; + int mergedKeys = 0; + if (intermediateRecords == null) { + if (aggregationGroupByResult != null) { + try { + Iterator groupKeyIterator = aggregationGroupByResult.getGroupKeyIterator(); + while (groupKeyIterator.hasNext()) { + QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, EXPLAIN_NAME); + GroupKeyGenerator.GroupKey groupKey = groupKeyIterator.next(); + Object[] keys = groupKey._keys; + Object[] values = Arrays.copyOf(keys, _numColumns); + int groupId = groupKey._groupId; + for (int i = 0; i < _numAggregationFunctions; i++) { + values[_numKeyColumns + i] = aggregationGroupByResult.getResultForGroupId(i, groupId); + } + Key key = new Key(keys); + int p = partitionFor(key); + if (localTables[p] == null) { + localTables[p] = + GroupByUtils.createPartitionTableForCombineOperator(resultsBlock, _queryContext, numGroupsHint, + _executorService); + } + localTables[p].upsert(key, new Record(values)); + if (_globalGroupTrimThreshold != Integer.MAX_VALUE && ++threadLocalTotal >= _globalGroupTrimThreshold) { + threadLocalTotal = maybeTrimLocalPartitions(localTables); + } + } + } finally { + aggregationGroupByResult.closeGroupKeyGenerator(); + } + } + } else { + for (IntermediateRecord intermediateResult : intermediateRecords) { + QueryThreadContext.checkTerminationAndSampleUsagePeriodically(mergedKeys++, EXPLAIN_NAME); + int p = partitionFor(intermediateResult._key); + if (localTables[p] == null) { + localTables[p] = + GroupByUtils.createPartitionTableForCombineOperator(resultsBlock, _queryContext, numGroupsHint, + _executorService); + } + localTables[p].upsert(intermediateResult._key, intermediateResult._record); + if (_globalGroupTrimThreshold != Integer.MAX_VALUE && ++threadLocalTotal >= _globalGroupTrimThreshold) { + threadLocalTotal = maybeTrimLocalPartitions(localTables); + } + } + } + return threadLocalTotal; + } + + /** + * Two-phase trim: first computes the actual unique group count (sum of all partition table sizes). If the + * actual total is below {@code _globalGroupTrimThreshold} (the record counter over-counted due to + * duplicate-key upserts), trims are skipped and the counter is reset to the actual unique count. This matches + * the behavior of {@link NonblockingGroupByCombineOperator}, whose {@link IndexedTable} fires resize only when + * {@code _lookupMap.size()} — unique groups — reaches the threshold. + * + *

When trim is needed, partitions are trimmed (largest first) in a loop until the total drops below + * {@code _globalGroupTrimThreshold}. Trimming all necessary partitions in one invocation avoids a "trim storm" + * where returning a total that is still at or above the threshold would cause the caller to immediately + * re-invoke trim after the very next upsert — especially costly when {@code P} is large and trimming only one + * partition per event would fire P rapid-succession trims. + * + * @return the updated {@code threadLocalTotal}: actual unique count after trim (or without trim if skipped), + * guaranteed to be below {@code _globalGroupTrimThreshold} + */ + private int maybeTrimLocalPartitions(IndexedTable[] localTables) { + // Compute actual unique group count and locate the largest partition in one pass (O(P), each size() is O(1)). + int actualTotal = 0; + int maxSize = 0; + int maxIdx = -1; + for (int i = 0; i < localTables.length; i++) { + IndexedTable t = localTables[i]; + if (t != null) { + int sz = t.size(); + actualTotal += sz; + if (sz > maxSize) { + maxSize = sz; + maxIdx = i; + } + } + } + if (actualTotal < _globalGroupTrimThreshold) { + // Record count exceeded threshold but unique group count did not (duplicate-key upserts inflated the + // counter). Reset to actual unique count without trimming, matching NON-BLOCKING behavior. + return actualTotal; + } + // Trim partitions (largest first) until the total drops below the threshold. Trimming all needed partitions + // in one call prevents a "trim storm" where a single-partition trim still leaves the total at or above the + // threshold, causing the caller to re-fire trim after every subsequent upsert. + while (actualTotal >= _globalGroupTrimThreshold && maxIdx >= 0) { + int preTrimSize = maxSize; + localTables[maxIdx].trim(); + actualTotal = actualTotal - preTrimSize + localTables[maxIdx].size(); + // Find the new largest for the next iteration (if still needed). + if (actualTotal >= _globalGroupTrimThreshold) { + maxSize = 0; + maxIdx = -1; + for (int i = 0; i < localTables.length; i++) { + IndexedTable t = localTables[i]; + if (t != null) { + int sz = t.size(); + if (sz > maxSize) { + maxSize = sz; + maxIdx = i; + } + } + } + } + } + return actualTotal; + } + + /** + * Returns the partition index for the given key using Knuth's multiplicative hash. + * + *

A plain {@code h % P} or {@code (h ^ h>>>16) % P} would cause bucket clustering inside the partition + * tables: all partition-p keys would share the same low {@code log2(P)} bits in Java's HashMap bucket selector + * {@code (h ^ h>>>16) & (capacity-1)}, causing every partition table to use only {@code capacity/P} effective + * buckets and triggering {@link java.util.HashMap} bin treeification. + * + *

Multiplying by the 32-bit Fibonacci/golden-ratio constant {@code 0x9E3779B9} (Knuth's multiplicative hash) + * thoroughly mixes all 32 input bits into the result. The product's bit pattern is orthogonal to the + * {@code h ^ h>>>16} expression that Java's HashMap uses for bucket selection: keys that map to the same + * multiplicative-hash partition are uniformly scattered across all HashMap buckets, eliminating clustering and + * treeification regardless of key distribution or table capacity. + * + *

Uses {@code & Integer.MAX_VALUE} (clear sign bit) rather than {@code Math.abs()} because + * {@code Math.abs(Integer.MIN_VALUE)} returns {@code Integer.MIN_VALUE} (negative), while + * {@code Integer.MIN_VALUE & Integer.MAX_VALUE} = 0 (safe non-negative). + */ + private int partitionFor(Key key) { + // Knuth's multiplicative hash: multiply by the 32-bit Fibonacci constant to scatter bits. + int h = key.hashCode() * 0x9E3779B9; + return (h & Integer.MAX_VALUE) % _numPartitions; + } + + /** + * Performs a tournament deposit for partition {@code p}: tries to claim the empty shared slot or steal and merge + * with the current occupant. The winning (larger) table becomes the new local candidate and loops until it + * successfully claims the slot. Called at most once per partition index per thread. + */ + private void depositPartition(IndexedTable[] localTables, int p) { + // Use a local variable so we can re-assign after merges + IndexedTable table = localTables[p]; + boolean deposited = false; + while (!deposited) { + IndexedTable toMerge; + synchronized (_partitionLocks[p]) { + if (_partitionedSharedTables[p] == null) { + _partitionedSharedTables[p] = table; + deposited = true; + break; + } + toMerge = _partitionedSharedTables[p]; + _partitionedSharedTables[p] = null; + } + // Merge outside the lock: merge smaller INTO larger + if (table.size() > toMerge.size()) { + table.merge(toMerge); + table.absorbTrimStats(toMerge); + } else { + toMerge.merge(table); + toMerge.absorbTrimStats(table); + table = toMerge; + } + } + } + + /** + * Waits for all worker tasks to finish, then merges the {@code P} per-partition winner tables into a single result. + */ + @Override + public BaseResultsBlock mergeResults() + throws Exception { + long timeoutMs = _queryContext.getEndTimeMs() - System.currentTimeMillis(); + boolean opCompleted = awaitOperatorLatch(timeoutMs); + if (!opCompleted) { + return getTimeoutResultsBlock(timeoutMs); + } + + Throwable ex = _processingException.get(); + if (ex != null) { + return getExceptionResultsBlock(ex); + } + + // Merge P partition winner tables into one final table. + // absorbTrimStats is intentionally omitted here: each partition winner already accumulated stats from + // the per-partition tournament exchange (analogous to NonblockingGroupByCombineOperator). Adding the + // cross-partition stats on top would inflate _numResizes by a factor of P, misleading isTrimmed(). + IndexedTable finalTable = null; + for (int p = 0; p < _numPartitions; p++) { + IndexedTable partTable = _partitionedSharedTables[p]; + if (partTable == null) { + continue; + } + if (finalTable == null) { + finalTable = partTable; + } else { + finalTable.merge(partTable); + } + } + + if (finalTable == null) { + // All segments produced zero groups (e.g. WHERE clause matched no rows). Return an empty table. + GroupByResultsBlock firstResultsBlock = _firstResultsBlock.get(); + if (firstResultsBlock != null) { + finalTable = createIndexedTable(firstResultsBlock, 1); + } else { + // No segments were processed at all (zero operators, or all workers exited before setting _firstResultsBlock). + String msg = "No segments were processed by " + ALGORITHM; + LOGGER.warn(msg); + return new ExceptionResultsBlock(new QueryErrorMessage(QueryErrorCode.QUERY_EXECUTION, msg, msg)); + } + } + + return getMergedResultsBlock(finalTable); + } +} diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java index 8f2bc026d32c..bb0991613229 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java @@ -31,6 +31,7 @@ import org.apache.pinot.core.operator.combine.GroupByCombineOperator; import org.apache.pinot.core.operator.combine.MinMaxValueBasedSelectionOrderByCombineOperator; import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; +import org.apache.pinot.core.operator.combine.PartitionedGroupByCombineOperator; import org.apache.pinot.core.operator.combine.SelectionOnlyCombineOperator; import org.apache.pinot.core.operator.combine.SelectionOrderByCombineOperator; import org.apache.pinot.core.operator.combine.SequentialSortedGroupByCombineOperator; @@ -178,11 +179,14 @@ private BaseCombineOperator getGroupByCombineOperator(List operators) return new SortedGroupByCombineOperator(operators, _queryContext, _executorService); } - // Allow fallback to the legacy ConcurrentHashMap-based operator via instance config or query option + // Allow per-query algorithm override via instance config or query option String algo = _queryContext.getGroupByAlgorithm(); if (GroupByCombineOperator.ALGORITHM.equalsIgnoreCase(algo)) { return new GroupByCombineOperator(operators, _queryContext, _executorService); } + if (PartitionedGroupByCombineOperator.ALGORITHM.equalsIgnoreCase(algo)) { + return new PartitionedGroupByCombineOperator(operators, _queryContext, _executorService); + } // Default to non-blocking combine which uses per-thread SimpleIndexedTables // to eliminate ConcurrentHashMap contention (1.8x-3.1x faster than the CONCURRENT algorithm) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java index 36247442171e..67c5a0efe35e 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java @@ -143,8 +143,92 @@ static int getIndexedTableInitialCapacity(int maxRowsToKeep, int minNumGroups, i */ public static IndexedTable createIndexedTableForCombineOperator(GroupByResultsBlock resultsBlock, QueryContext queryContext, int numThreads, ExecutorService executorService) { + return createIndexedTableForCombineOperator(resultsBlock, queryContext, numThreads, executorService, + resultsBlock.getNumGroups()); + } + + /** + * Creates an indexed table for the combine operator given a sample results block, using {@code numGroupsHint} as the + * initial-capacity hint instead of {@code resultsBlock.getNumGroups()}. Use this overload when the table will hold + * only a subset of the segment groups (e.g., one partition out of P in a partitioned combine). + */ + public static IndexedTable createIndexedTableForCombineOperator(GroupByResultsBlock resultsBlock, + QueryContext queryContext, int numThreads, ExecutorService executorService, int numGroupsHint) { + return createIndexedTableForCombineOperator(resultsBlock, queryContext, numThreads, executorService, numGroupsHint, + queryContext.getGroupTrimThreshold()); + } + + /** + * Returns the effective trim threshold for a combine operator, accounting for disabled-trim sentinels. + * Returns {@link Integer#MAX_VALUE} when: + *

    + *
  • there is no ORDER BY (trim is only meaningful with top-K ordering), or
    • + *
  • {@code groupTrimThreshold} is non-positive or above {@link #MAX_TRIM_THRESHOLD}, or
    • + *
  • the computed trim size would cause integer overflow.
    • + *
+ * This value can be used as {@code _globalGroupTrimThreshold} in a partitioned combine to decide when to trigger a + * coordinated trim across all partition-local tables. + */ + public static int getEffectiveCombineTrimThreshold(QueryContext queryContext) { + if (queryContext.getOrderByExpressions() == null) { + return Integer.MAX_VALUE; + } + int minTrimSize = queryContext.getMinServerGroupTrimSize(); + int trimSize = minTrimSize > 0 ? getTableCapacity(queryContext.getLimit(), minTrimSize) : Integer.MAX_VALUE; + return getIndexedTableTrimThreshold(trimSize, queryContext.getGroupTrimThreshold()); + } + + /** + * Creates a partition-local indexed table for the partitioned group-by combine operator. + * + *

The returned table has auto-trim disabled ({@code trimThreshold = MAX_VALUE}) so the combine operator + * can coordinate trim across all partitions at the global threshold rather than per-partition. When an explicit + * {@link IndexedTable#trim()} call is made, the table trims to its {@code trimSize} — computed the same way as a + * regular combine table — rather than to {@link Integer#MAX_VALUE}. + * + * @param resultsBlock sample results block used to derive the data schema + * @param queryContext query context + * @param numGroupsHint expected number of groups for this partition (used for initial-capacity hint) + * @param executorService executor for multi-threaded final reduce + */ + public static IndexedTable createPartitionTableForCombineOperator(GroupByResultsBlock resultsBlock, + QueryContext queryContext, int numGroupsHint, ExecutorService executorService) { + DataSchema dataSchema = resultsBlock.getDataSchema(); + int limit = queryContext.getLimit(); + boolean hasOrderBy = queryContext.getOrderByExpressions() != null; + boolean hasHaving = queryContext.getHavingFilter() != null; + int minTrimSize = queryContext.getMinServerGroupTrimSize(); + int trimSize = minTrimSize > 0 ? getTableCapacity(limit, minTrimSize) : Integer.MAX_VALUE; + int minInitialCapacity = queryContext.getMinInitialIndexedTableCapacity(); + + int resultSize; + int tableTrimSize; + if (!hasOrderBy) { + resultSize = hasHaving ? trimSize : limit; + tableTrimSize = Integer.MAX_VALUE; // trim() is a no-op without ORDER BY + } else { + resultSize = (queryContext.isServerReturnFinalResult() && !hasHaving) ? limit : trimSize; + tableTrimSize = trimSize; // used when trim() is explicitly invoked + } + // Initial capacity based on expected groups per partition, uncapped by tableTrimSize + int initialCapacity = getIndexedTableInitialCapacity(Integer.MAX_VALUE, numGroupsHint, minInitialCapacity); + // trimThreshold=MAX_VALUE disables auto-trim; trim() calls still work via tableTrimSize + return new SimpleIndexedTable(dataSchema, false, queryContext, resultSize, tableTrimSize, Integer.MAX_VALUE, + initialCapacity, executorService); + } + + /** + * Creates an indexed table for the combine operator given a sample results block, using {@code numGroupsHint} as the + * initial-capacity hint and {@code groupTrimThresholdOverride} in place of + * {@code queryContext.getGroupTrimThreshold()}. Use this overload when the caller needs to supply an explicit trim + * threshold that differs from the one in {@code queryContext} — for example, passing + * {@link Integer#MAX_VALUE} to disable auto-trim on a partition-local table that will be trimmed externally. + */ + public static IndexedTable createIndexedTableForCombineOperator(GroupByResultsBlock resultsBlock, + QueryContext queryContext, int numThreads, ExecutorService executorService, int numGroupsHint, + int groupTrimThresholdOverride) { DataSchema dataSchema = resultsBlock.getDataSchema(); - int numGroups = resultsBlock.getNumGroups(); + int numGroups = numGroupsHint; int limit = queryContext.getLimit(); boolean hasOrderBy = queryContext.getOrderByExpressions() != null; boolean hasHaving = queryContext.getHavingFilter() != null; @@ -190,7 +274,7 @@ public static IndexedTable createIndexedTableForCombineOperator(GroupByResultsBl } else { resultSize = trimSize; } - int trimThreshold = getIndexedTableTrimThreshold(trimSize, queryContext.getGroupTrimThreshold()); + int trimThreshold = getIndexedTableTrimThreshold(trimSize, groupTrimThresholdOverride); int initialCapacity = getIndexedTableInitialCapacity(trimThreshold, numGroups, minInitialIndexedTableCapacity); if (trimThreshold == Integer.MAX_VALUE) { return getTrimDisabledIndexedTable(dataSchema, false, queryContext, resultSize, initialCapacity, numThreads, diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/table/PartitionedGroupByCombineOperatorTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/table/PartitionedGroupByCombineOperatorTest.java new file mode 100644 index 000000000000..1d0d59f9e16d --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/data/table/PartitionedGroupByCombineOperatorTest.java @@ -0,0 +1,197 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.data.table; + +// This test is in org.apache.pinot.core.data.table so it can access the +// package-private IntermediateRecord constructor for building synthetic segment results. + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import javax.annotation.Nullable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.operator.ExecutionStatistics; +import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock; +import org.apache.pinot.core.operator.blocks.results.ExceptionResultsBlock; +import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; +import org.apache.pinot.core.operator.combine.PartitionedGroupByCombineOperator; +import org.apache.pinot.core.plan.ExplainInfo; +import org.apache.pinot.core.query.request.context.QueryContext; +import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; + + +/** + * Unit tests for {@link PartitionedGroupByCombineOperator}. + * + *

Uses synthetic segment operators (each returning a single {@link IntermediateRecord}) to test the + * end-to-end combine behavior: correct top-K result selection, and correct handling of the global trim path. + */ +public class PartitionedGroupByCombineOperatorTest { + + private static final DataSchema DATA_SCHEMA = new DataSchema( + new String[]{"a", "sum(b)"}, + new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.DOUBLE}); + + /** + * Verifies that the operator returns at most {@code trimSize} results and that the actual top groups + * (those with the highest {@code SUM(b)} values) survive both the global-threshold trim and the final merge. + * + *

Setup: 500 groups, ORDER BY SUM(b) DESC LIMIT 5, trimSize=50, threshold=400. + * A single worker thread processes all 500 groups, so the global trim fires once at group 400, + * trimming each partition to the top 50. Groups added after the trim are retained. The final result + * must contain the top 5 groups by value. + */ + @Test + public void testTopKSurvivesGlobalTrim() + throws Exception { + QueryContext queryContext = QueryContextConverterUtils.getQueryContext( + "SELECT a, SUM(b) FROM t GROUP BY a ORDER BY SUM(b) DESC LIMIT 5"); + queryContext.setMinServerGroupTrimSize(50); + queryContext.setGroupTrimThreshold(400); + queryContext.setEndTimeMs(System.currentTimeMillis() + 30_000); + + int numGroups = 500; + List operators = new ArrayList<>(numGroups); + for (int i = 0; i < numGroups; i++) { + operators.add(new SingleGroupOperator(DATA_SCHEMA, queryContext, "group_" + i, (double) i)); + } + + // Use a single-threaded executor so one worker processes all 500 groups sequentially. + // This guarantees the thread-local total exceeds the threshold=400 and the global trim fires. + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + PartitionedGroupByCombineOperator combine = + new PartitionedGroupByCombineOperator(operators, queryContext, executor, 2); + BaseResultsBlock resultBlock = combine.nextBlock(); + + assertFalse(resultBlock instanceof ExceptionResultsBlock, "Expected successful result, got exception"); + Table table = ((GroupByResultsBlock) resultBlock).getTable(); + assertNotNull(table, "Result table must not be null"); + // After finish(), the table holds at most trimSize=50 records + assertTrue(table.size() <= 50, "Result size must not exceed trimSize=50, got " + table.size()); + + // The top 5 groups (group_499..group_495) must be present — if the global trim incorrectly + // over-trimmed a hot partition before the global total reached 400, these groups could be lost. + Set expectedTopGroups = new HashSet<>(); + for (int i = numGroups - 5; i < numGroups; i++) { + expectedTopGroups.add("group_" + i); + } + Set actualGroups = new HashSet<>(); + table.iterator().forEachRemaining(r -> actualGroups.add((String) r.getValues()[0])); + assertTrue(actualGroups.containsAll(expectedTopGroups), + "Top 5 groups must be present in results. Expected: " + expectedTopGroups + ", Actual: " + actualGroups); + } finally { + executor.shutdownNow(); + } + } + + /** + * Verifies correct combine behavior when there is no trim (total groups below the threshold). + * All groups are returned up to trimSize, and the top groups are correct. + */ + @Test + public void testNoTrimWhenBelowThreshold() + throws Exception { + QueryContext queryContext = QueryContextConverterUtils.getQueryContext( + "SELECT a, SUM(b) FROM t GROUP BY a ORDER BY SUM(b) DESC LIMIT 5"); + queryContext.setMinServerGroupTrimSize(50); + queryContext.setGroupTrimThreshold(1_000_000); + queryContext.setEndTimeMs(System.currentTimeMillis() + 30_000); + + int numGroups = 20; + List operators = new ArrayList<>(numGroups); + for (int i = 0; i < numGroups; i++) { + operators.add(new SingleGroupOperator(DATA_SCHEMA, queryContext, "group_" + i, (double) i)); + } + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + PartitionedGroupByCombineOperator combine = + new PartitionedGroupByCombineOperator(operators, queryContext, executor, 2); + BaseResultsBlock resultBlock = combine.nextBlock(); + + assertFalse(resultBlock instanceof ExceptionResultsBlock, "Expected successful result, got exception"); + Table table = ((GroupByResultsBlock) resultBlock).getTable(); + assertNotNull(table); + // All 20 groups fit within trimSize=50, so all should be present + assertEquals(table.size(), numGroups, "All groups must be present when below trimSize"); + } finally { + executor.shutdownNow(); + } + } + + /** + * Fake segment operator that returns a single group-by record. + * Placed in {@code org.apache.pinot.core.data.table} to access the package-private + * {@link IntermediateRecord} constructor. + */ + @SuppressWarnings("rawtypes") + private static class SingleGroupOperator implements Operator { + private final DataSchema _dataSchema; + private final QueryContext _queryContext; + private final String _groupKey; + private final double _value; + + SingleGroupOperator(DataSchema dataSchema, QueryContext queryContext, String groupKey, double value) { + _dataSchema = dataSchema; + _queryContext = queryContext; + _groupKey = groupKey; + _value = value; + } + + @Override + public GroupByResultsBlock nextBlock() { + Key key = new Key(new Object[]{_groupKey}); + Record record = new Record(new Object[]{_groupKey, _value}); + List records = List.of(new IntermediateRecord(key, record, new Comparable[]{_value})); + return new GroupByResultsBlock(_dataSchema, records, _queryContext); + } + + @Override + public List getChildOperators() { + return List.of(); + } + + @Nullable + @Override + public String toExplainString() { + return "SingleGroupOperator"; + } + + @Override + public ExplainInfo getExplainInfo() { + return null; + } + + @Override + public ExecutionStatistics getExecutionStatistics() { + return new ExecutionStatistics(0, 0, 0, 0); + } + } +} diff --git a/pinot-core/src/test/java/org/apache/pinot/core/util/GroupByUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/util/GroupByUtilsTest.java index f8117b69cea9..67ee90393f45 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/util/GroupByUtilsTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/util/GroupByUtilsTest.java @@ -18,6 +18,7 @@ */ package org.apache.pinot.core.util; +import java.lang.reflect.Field; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -29,6 +30,7 @@ import org.apache.pinot.core.common.datatable.DataTableBuilderFactory; import org.apache.pinot.core.data.table.ConcurrentIndexedTable; import org.apache.pinot.core.data.table.IndexedTable; +import org.apache.pinot.core.data.table.SimpleIndexedTable; import org.apache.pinot.core.data.table.UnboundedConcurrentIndexedTable; import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; import org.apache.pinot.core.query.reduce.DataTableReducerContext; @@ -38,6 +40,8 @@ import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertTrue; @@ -75,6 +79,148 @@ public void getIndexedTableTrimThreshold() { assertEquals(GroupByUtils.getIndexedTableTrimThreshold(500000001, 10), Integer.MAX_VALUE); } + /** + * Verifies that {@link GroupByUtils#createIndexedTableForCombineOperator} with an explicit + * {@code groupTrimThresholdOverride} correctly propagates or preserves disabled-trim sentinels. + *

    + *
  • A non-positive override (0, -1) or one above MAX_TRIM_THRESHOLD must produce a trim-disabled table + * ({@code _trimThreshold == Integer.MAX_VALUE}), not silently enable trimming at threshold = 1.
    • + *
  • A valid positive override within range must produce a trim-enabled table at that threshold.
    • + *
+ */ + @Test + public void testCreateIndexedTableForCombineOperatorTrimThresholdOverride() + throws Exception { + QueryContext queryContext = + QueryContextConverterUtils.getQueryContext("SELECT a, SUM(b) FROM t GROUP BY a ORDER BY SUM(b) LIMIT 100"); + queryContext.setMinServerGroupTrimSize(5000); // trimSize = max(100*5, 5000) = 5000 + DataSchema dataSchema = new DataSchema(new String[]{"a", "sum(b)"}, + new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.DOUBLE}); + GroupByResultsBlock block = new GroupByResultsBlock(dataSchema, queryContext); + + // Disabled-trim sentinel: 0 must NOT be converted to 1 (which would enable trim). + queryContext.setGroupTrimThreshold(0); // normally disabled + IndexedTable tableZero = GroupByUtils.createIndexedTableForCombineOperator(block, queryContext, 1, + Executors.newSingleThreadExecutor(), 100, 0); + assertTrue(tableZero instanceof SimpleIndexedTable, "Expected SimpleIndexedTable for numThreads=1"); + assertEquals(trimThresholdOf(tableZero), Integer.MAX_VALUE, + "groupTrimThresholdOverride=0 must produce a trim-disabled table"); + + // Disabled-trim sentinel: negative value. + IndexedTable tableNeg = GroupByUtils.createIndexedTableForCombineOperator(block, queryContext, 1, + Executors.newSingleThreadExecutor(), 100, -1); + assertEquals(trimThresholdOf(tableNeg), Integer.MAX_VALUE, + "groupTrimThresholdOverride=-1 must produce a trim-disabled table"); + + // Disabled-trim sentinel: above MAX_TRIM_THRESHOLD. + IndexedTable tableHigh = GroupByUtils.createIndexedTableForCombineOperator(block, queryContext, 1, + Executors.newSingleThreadExecutor(), 100, GroupByUtils.MAX_TRIM_THRESHOLD + 1); + assertEquals(trimThresholdOf(tableHigh), Integer.MAX_VALUE, + "groupTrimThresholdOverride above MAX_TRIM_THRESHOLD must produce a trim-disabled table"); + + // Valid override: should produce a trim-enabled table with the scaled threshold. + // trimSize=5000, override=100_000 → getIndexedTableTrimThreshold(5000, 100_000) = 100_000 + queryContext.setGroupTrimThreshold(1_000_000); // original, not used + IndexedTable tableValid = GroupByUtils.createIndexedTableForCombineOperator(block, queryContext, 1, + Executors.newSingleThreadExecutor(), 100, 100_000); + assertFalse(trimThresholdOf(tableValid) == Integer.MAX_VALUE, + "Valid groupTrimThresholdOverride must produce a trim-enabled table"); + assertEquals(trimThresholdOf(tableValid), 100_000, + "groupTrimThresholdOverride=100_000 with trimSize=5000 must produce trimThreshold=100_000"); + } + + private static int trimThresholdOf(IndexedTable table) + throws Exception { + Field f = IndexedTable.class.getDeclaredField("_trimThreshold"); + f.setAccessible(true); + return (int) f.get(table); + } + + private static int trimSizeOf(IndexedTable table) + throws Exception { + Field f = IndexedTable.class.getDeclaredField("_trimSize"); + f.setAccessible(true); + return (int) f.get(table); + } + + /** + * Verifies {@link GroupByUtils#getEffectiveCombineTrimThreshold}: + *
    + *
  • Queries without ORDER BY always return MAX_VALUE (trim is meaningless).
    • + *
  • Disabled-trim sentinels (0, -1, above MAX_TRIM_THRESHOLD) return MAX_VALUE.
    • + *
  • A valid threshold returns {@code max(threshold, 2 * trimSize)}.
    • + *
+ */ + @Test + public void testGetEffectiveCombineTrimThreshold() { + // No ORDER BY: always disabled regardless of threshold setting + QueryContext noOrderBy = + QueryContextConverterUtils.getQueryContext("SELECT a, SUM(b) FROM t GROUP BY a LIMIT 100"); + noOrderBy.setGroupTrimThreshold(1_000_000); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(noOrderBy), Integer.MAX_VALUE, + "No ORDER BY must return MAX_VALUE"); + + // With ORDER BY, disabled sentinels + QueryContext withOrderBy = QueryContextConverterUtils.getQueryContext( + "SELECT a, SUM(b) FROM t GROUP BY a ORDER BY SUM(b) LIMIT 100"); + withOrderBy.setMinServerGroupTrimSize(5000); + + withOrderBy.setGroupTrimThreshold(0); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(withOrderBy), Integer.MAX_VALUE, + "groupTrimThreshold=0 must return MAX_VALUE"); + + withOrderBy.setGroupTrimThreshold(-1); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(withOrderBy), Integer.MAX_VALUE, + "groupTrimThreshold=-1 must return MAX_VALUE"); + + withOrderBy.setGroupTrimThreshold(GroupByUtils.MAX_TRIM_THRESHOLD + 1); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(withOrderBy), Integer.MAX_VALUE, + "groupTrimThreshold above MAX must return MAX_VALUE"); + + // Valid threshold: result = max(threshold, 2*trimSize); trimSize=max(100*5,5000)=5000 → 2*trimSize=10000 + withOrderBy.setGroupTrimThreshold(1_000_000); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(withOrderBy), 1_000_000, + "valid threshold must be returned as-is when above 2*trimSize"); + + withOrderBy.setGroupTrimThreshold(100); + assertEquals(GroupByUtils.getEffectiveCombineTrimThreshold(withOrderBy), 10_000, + "threshold below 2*trimSize must be raised to 2*trimSize=10000"); + } + + /** + * Verifies {@link GroupByUtils#createPartitionTableForCombineOperator}: + *
    + *
  • The returned table must be a {@link SimpleIndexedTable} (single-threaded partition table).
    • + *
  • Auto-trim must be disabled: {@code _trimThreshold == Integer.MAX_VALUE}.
    • + *
  • For ORDER BY queries, {@code _trimSize} must equal the effective trim size so explicit + * {@link IndexedTable#trim()} calls work correctly.
    • + *
+ */ + @Test + public void testCreatePartitionTableForCombineOperator() + throws Exception { + QueryContext queryContext = QueryContextConverterUtils.getQueryContext( + "SELECT a, SUM(b) FROM t GROUP BY a ORDER BY SUM(b) LIMIT 100"); + queryContext.setMinServerGroupTrimSize(5000); + queryContext.setGroupTrimThreshold(1_000_000); + DataSchema dataSchema = new DataSchema(new String[]{"a", "sum(b)"}, + new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.DOUBLE}); + GroupByResultsBlock block = new GroupByResultsBlock(dataSchema, queryContext); + + IndexedTable table = GroupByUtils.createPartitionTableForCombineOperator(block, queryContext, 1000, + Executors.newSingleThreadExecutor()); + + assertTrue(table instanceof SimpleIndexedTable, "Partition table must be a SimpleIndexedTable"); + assertEquals(trimThresholdOf(table), Integer.MAX_VALUE, "Partition table must have auto-trim disabled"); + // trimSize = max(100*5, 5000) = 5000; explicit trim() calls reduce the table to this size + int expectedTrimSize = + GroupByUtils.getTableCapacity(queryContext.getLimit(), queryContext.getMinServerGroupTrimSize()); + assertNotEquals(trimSizeOf(table), Integer.MAX_VALUE, + "Partition table must have a valid trimSize for explicit trim() calls"); + assertEquals(trimSizeOf(table), expectedTrimSize, + "Partition table trimSize must equal the effective combine trim size"); + } + @Test public void testGetIndexedTableInitialCapacity() { assertEquals(GroupByUtils.getIndexedTableInitialCapacity(Integer.MAX_VALUE, 10, 128), 128); diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java index 4c1646ab9dd2..8939e531782b 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentNonBlockingGroupBySingleValueQueriesTest.java @@ -28,6 +28,7 @@ import org.apache.pinot.common.utils.DataSchema.ColumnDataType; import org.apache.pinot.core.operator.combine.GroupByCombineOperator; import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; +import org.apache.pinot.core.operator.combine.PartitionedGroupByCombineOperator; import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2; import org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey; import org.testng.annotations.DataProvider; @@ -80,6 +81,24 @@ public void testGroupByOrderByWithConcurrentAlgorithmAndTrim(String query, long expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); } + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderByWithPartitionedAlgorithm(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, + PartitionedGroupByCombineOperator.ALGORITHM)), + 120000L, 0L, expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); + } + + @Test(dataProvider = "groupByOrderByDataProvider") + public void testGroupByOrderByWithPartitionedAlgorithmAndTrim(String query, long expectedNumEntriesScannedPostFilter, + ResultTable expectedResultTable) { + QueriesTestUtils.testInterSegmentsResult( + getBrokerResponse(query, TRIM_ENABLED_PLAN_MAKER, Map.of(QueryOptionKey.GROUP_BY_ALGORITHM, + PartitionedGroupByCombineOperator.ALGORITHM)), + 120000L, 0L, expectedNumEntriesScannedPostFilter, 120000L, expectedResultTable); + } + /** * Provides various combinations of order by in ResultTable. * In order to calculate the expected results, the results from a group by were taken, and then ordered accordingly. diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java index 979c68c21cab..cc74ffbf0f6d 100644 --- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java +++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkGroupByCombineOperator.java @@ -27,9 +27,11 @@ import java.util.concurrent.TimeUnit; import org.apache.pinot.common.utils.DataSchema; import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.data.table.IndexedTable; import org.apache.pinot.core.data.table.IntermediateRecord; import org.apache.pinot.core.data.table.Key; import org.apache.pinot.core.data.table.Record; +import org.apache.pinot.core.operator.AcquireReleaseColumnsSegmentOperator; import org.apache.pinot.core.operator.BaseOperator; import org.apache.pinot.core.operator.ExecutionStatistics; import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock; @@ -37,6 +39,7 @@ import org.apache.pinot.core.operator.combine.BaseCombineOperator; import org.apache.pinot.core.operator.combine.GroupByCombineOperator; import org.apache.pinot.core.operator.combine.NonblockingGroupByCombineOperator; +import org.apache.pinot.core.operator.combine.PartitionedGroupByCombineOperator; import org.apache.pinot.core.query.request.context.QueryContext; import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils; import org.openjdk.jmh.annotations.Benchmark; @@ -60,36 +63,69 @@ /** * Benchmarks end-to-end server-side group-by combine operators on synthetic segment-level result blocks. * - *

The benchmark materializes fresh intermediate records per invocation so that repeated runs do not reuse mutable - * records from prior combines. + *

Group key cardinality: {@code CARDINALITY_D1 × CARDINALITY_D2 = 5000 × 2000 = 10M} unique groups. + * + *

Approximate unique groups in the combined result by (numSegments, numRecordsPerSegment): + *

    + *
  • (10, 100k) ≈ 1 M unique groups (10M total input records per combine)
    • + *
  • (100, 100k) ≈ 6 M unique groups (10M total input records per combine)
    • + *
  • (10, 1M) ≈ 6 M unique groups (10M total input records per combine)
    • + *
  • (100, 1M) ≈ 10 M unique groups (100M total input records per combine)
    • + *
+ * + *

Segment data is stored in a pool of {@value #SEGMENT_DATA_POOL_SIZE} unique {@link SegmentData} objects and + * cycled across all {@code numSegments} operators. Records are created lazily inside each operator's + * {@code getNextBlock()} call so that memory usage is proportional to concurrent thread count rather than + * total segment count — enabling large-scale tests (100 segments × 1M records) within an 8 GB heap. */ @State(Scope.Benchmark) -@Fork(value = 1, jvmArgs = {"-server", "-Xms8G", "-Xmx8G", "-XX:MaxDirectMemorySize=8G"}) +@Fork(value = 1, jvmArgs = {"-server", "-Xms16G", "-Xmx16G", "-XX:MaxDirectMemorySize=8G"}) public class BenchmarkGroupByCombineOperator { private static final String QUERY = "SELECT d1, d2, SUM(m1), MAX(m2) FROM testTable GROUP BY d1, d2 ORDER BY SUM(m1) DESC LIMIT 500"; - private static final int NUM_SEGMENTS = 32; - private static final int CARDINALITY_D1 = 4096; - private static final int CARDINALITY_D2 = 2048; + // 5000 × 2000 = 10M unique group keys + private static final int CARDINALITY_D1 = 5000; + private static final int CARDINALITY_D2 = 2000; + // Number of unique SegmentData objects in the pool; large numSegments cycle through this pool. + private static final int SEGMENT_DATA_POOL_SIZE = 10; private static final long QUERY_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(10); + // NON-BLOCKING-NO-STEAL: variant of NON-BLOCKING that always creates a fresh local table + // instead of stealing the shared table on first use, to quantify the steal optimization's benefit. + private static final String NON_BLOCKING_NO_STEAL = "NON-BLOCKING-NO-STEAL"; + @Param({ GroupByCombineOperator.ALGORITHM, - NonblockingGroupByCombineOperator.ALGORITHM + NonblockingGroupByCombineOperator.ALGORITHM, + NON_BLOCKING_NO_STEAL, + PartitionedGroupByCombineOperator.ALGORITHM }) private String _algorithm; - @Param({"10000", "100000", "500000"}) + // numSegments: number of segment operators presented to the combine. + // With SEGMENT_DATA_POOL_SIZE=10, numSegments > 10 cycles through the pool (simulates key overlap). + @Param({"10", "100"}) + private int _numSegments; + + // numRecordsPerSegment: approximate number of unique groups per segment block. + // With cardinality=10M, 100k records/seg ≈ 100k unique groups, 1M records/seg ≈ 950k unique groups. + @Param({"100000", "1000000"}) private int _numRecordsPerSegment; @Param({"8"}) private int _numThreads; + // numPartitions: only used by PARTITIONED-NON-BLOCKING; ignored by other algorithms. + @Param({"8", "32", "128"}) + private int _numPartitions; + + // Pre-created value pools; d2Values is Integer[] so boxes are allocated once at setup time, not per-record. private final String[] _d1Values = new String[CARDINALITY_D1]; private final Integer[] _d2Values = new Integer[CARDINALITY_D2]; private DataSchema _dataSchema; - private SegmentData[] _segmentData; + // Pool of unique segment datasets; cycled across all _numSegments operators. + private SegmentData[] _segmentDataPool; private Constructor _intermediateRecordConstructor; private ExecutorService _executorService; @@ -109,15 +145,16 @@ public void setup() _dataSchema = new DataSchema(new String[]{"d1", "d2", "sum(m1)", "max(m2)"}, new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.INT, DataSchema.ColumnDataType.DOUBLE, DataSchema.ColumnDataType.DOUBLE}); - _segmentData = new SegmentData[NUM_SEGMENTS]; + + _segmentDataPool = new SegmentData[SEGMENT_DATA_POOL_SIZE]; Random random = new Random(8675309L); - for (int i = 0; i < NUM_SEGMENTS; i++) { - _segmentData[i] = new SegmentData(_numRecordsPerSegment); + for (int i = 0; i < SEGMENT_DATA_POOL_SIZE; i++) { + _segmentDataPool[i] = new SegmentData(_numRecordsPerSegment); for (int j = 0; j < _numRecordsPerSegment; j++) { - _segmentData[i]._d1Ids[j] = random.nextInt(CARDINALITY_D1); - _segmentData[i]._d2Ids[j] = random.nextInt(CARDINALITY_D2); - _segmentData[i]._sumValues[j] = random.nextInt(1000); - _segmentData[i]._maxValues[j] = random.nextInt(1000); + _segmentDataPool[i]._d1Ids[j] = random.nextInt(CARDINALITY_D1); + _segmentDataPool[i]._d2Ids[j] = random.nextInt(CARDINALITY_D2); + _segmentDataPool[i]._sumValues[j] = random.nextInt(1000); + _segmentDataPool[i]._maxValues[j] = random.nextInt(1000); } } @@ -128,16 +165,14 @@ public void setup() } @Setup(Level.Invocation) - public void setupInvocation() - throws Exception { + public void setupInvocation() { _queryContext = QueryContextConverterUtils.getQueryContext(QUERY); _queryContext.setEndTimeMs(System.currentTimeMillis() + QUERY_TIMEOUT_MS); _queryContext.setMaxExecutionThreads(_numThreads); - _operators = new ArrayList<>(NUM_SEGMENTS); - for (SegmentData segmentData : _segmentData) { - _operators.add(new StaticGroupByOperator( - new GroupByResultsBlock(_dataSchema, createIntermediateRecords(segmentData), _queryContext))); + _operators = new ArrayList<>(_numSegments); + for (int i = 0; i < _numSegments; i++) { + _operators.add(new LazyGroupByOperator(_segmentDataPool[i % SEGMENT_DATA_POOL_SIZE])); } } @@ -166,6 +201,10 @@ private BaseCombineOperator createCombineOperator() { switch (_algorithm) { case NonblockingGroupByCombineOperator.ALGORITHM: return new NonblockingGroupByCombineOperator(_operators, _queryContext, _executorService); + case NON_BLOCKING_NO_STEAL: + return new NoStealNonblockingGroupByCombineOperator(_operators, _queryContext, _executorService); + case PartitionedGroupByCombineOperator.ALGORITHM: + return new PartitionedGroupByCombineOperator(_operators, _queryContext, _executorService, _numPartitions); default: return new GroupByCombineOperator(_operators, _queryContext, _executorService); } @@ -187,8 +226,8 @@ private List createIntermediateRecords(SegmentData segmentDa public static void main(String[] args) throws Exception { ChainedOptionsBuilder options = new OptionsBuilder().include(BenchmarkGroupByCombineOperator.class.getSimpleName()) - .warmupIterations(1).warmupTime(TimeValue.seconds(5)).measurementIterations(3) - .measurementTime(TimeValue.seconds(10)).forks(1); + .warmupIterations(1).warmupTime(TimeValue.seconds(10)).measurementIterations(3) + .measurementTime(TimeValue.seconds(15)).forks(1); new Runner(options.build()).run(); } @@ -206,18 +245,27 @@ private SegmentData(int numRecords) { } } - private static final class StaticGroupByOperator extends BaseOperator { - private static final String EXPLAIN_NAME = "STATIC_GROUP_BY"; + /** + * Segment operator that creates fresh IntermediateRecords on each {@link #getNextBlock()} call. + * This avoids accumulation bugs from mutable Record objects being reused across invocations, and keeps + * peak memory proportional to concurrent thread count rather than total segment count. + */ + private final class LazyGroupByOperator extends BaseOperator { + private static final String EXPLAIN_NAME = "LAZY_GROUP_BY"; - private final GroupByResultsBlock _resultsBlock; + private final SegmentData _segmentData; - private StaticGroupByOperator(GroupByResultsBlock resultsBlock) { - _resultsBlock = resultsBlock; + private LazyGroupByOperator(SegmentData segmentData) { + _segmentData = segmentData; } @Override protected GroupByResultsBlock getNextBlock() { - return _resultsBlock; + try { + return new GroupByResultsBlock(_dataSchema, createIntermediateRecords(_segmentData), _queryContext); + } catch (Exception e) { + throw new RuntimeException(e); + } } @Override @@ -235,4 +283,63 @@ public ExecutionStatistics getExecutionStatistics() { return new ExecutionStatistics(0, 0, 0, 0); } } + + /** + * Variant of {@link NonblockingGroupByCombineOperator} that always creates a fresh local table instead of first + * trying to steal the shared slot. Used to quantify the steal-before-create optimization's benefit. + */ + private static final class NoStealNonblockingGroupByCombineOperator extends GroupByCombineOperator { + NoStealNonblockingGroupByCombineOperator(List operators, QueryContext queryContext, + ExecutorService executorService) { + super(operators, queryContext, executorService); + } + + @Override + protected void processSegments() { + int operatorId; + IndexedTable indexedTable = null; + while (_processingException.get() == null && (operatorId = _nextOperatorId.getAndIncrement()) < _numOperators) { + Operator operator = _operators.get(operatorId); + try { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).acquire(); + } + GroupByResultsBlock resultsBlock = (GroupByResultsBlock) operator.nextBlock(); + // Always create a fresh local table (no steal from shared slot) + if (indexedTable == null) { + indexedTable = createIndexedTable(resultsBlock, 1); + } + mergeGroupByResultsBlock(indexedTable, resultsBlock, EXPLAIN_NAME); + } catch (RuntimeException e) { + throw wrapOperatorException(operator, e); + } finally { + if (operator instanceof AcquireReleaseColumnsSegmentOperator) { + ((AcquireReleaseColumnsSegmentOperator) operator).release(); + } + } + } + + boolean setGroupByResult = false; + while (indexedTable != null && !setGroupByResult) { + IndexedTable indexedTableToMerge; + synchronized (this) { + if (!hasSharedTable()) { + depositSharedTable(indexedTable); + setGroupByResult = true; + continue; + } else { + indexedTableToMerge = stealSharedTable(); + } + } + if (indexedTable.size() > indexedTableToMerge.size()) { + indexedTable.merge(indexedTableToMerge); + indexedTable.absorbTrimStats(indexedTableToMerge); + } else { + indexedTableToMerge.merge(indexedTable); + indexedTableToMerge.absorbTrimStats(indexedTable); + indexedTable = indexedTableToMerge; + } + } + } + } } diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index 9daa9f3c91a4..19ea76d0cd59 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -1464,11 +1464,13 @@ public static class Server { public static final String CONFIG_OF_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD = QUERY_EXECUTOR_CONFIG_PREFIX + "." + GROUPBY_TRIM_THRESHOLD; public static final int DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD = 1_000_000; - // Server-level default group-by combine algorithm; can be overridden per query via groupByAlgorithm option + // Server-level default group-by combine algorithm; can be overridden per query via groupByAlgorithm option. + // Valid values: "CONCURRENT" (stable, lock-based), "NON-BLOCKING" (tournament-exchange, ~2× faster), + // "PARTITIONED-NON-BLOCKING" (partitioned tournament-exchange, for high-thread-count scenarios). + // Defaults to "CONCURRENT" for safe rollout; operators can opt in to "NON-BLOCKING" via server config or query + // option after validating in their environment. public static final String GROUP_BY_ALGORITHM = "group.by.algorithm"; - public static final String CONFIG_OF_QUERY_EXECUTOR_GROUP_BY_ALGORITHM = - QUERY_EXECUTOR_CONFIG_PREFIX + "." + GROUP_BY_ALGORITHM; - public static final String DEFAULT_QUERY_EXECUTOR_GROUP_BY_ALGORITHM = "NON-BLOCKING"; + public static final String DEFAULT_QUERY_EXECUTOR_GROUP_BY_ALGORITHM = "CONCURRENT"; // Do sort-aggregation when LIMIT is below this threshold public static final int DEFAULT_SORT_AGGREGATE_LIMIT_THRESHOLD = 10_000; // Use sequential instead of pair-wise combine for sort-aggr when numSegments is below this threshold