Jackie-Jiang commented on code in PR #20061:
URL: https://github.com/apache/pinot/pull/20061#discussion_r4485520629
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java:
##########
@@ -141,9 +141,23 @@ public static PinotOperatorTable instance(boolean
nullHandlingEnabled) {
/// [org.apache.pinot.calcite.rel.rules.PinotRuleUtils#isRelocatable]
ignores this flag, so ARRAY_TO_MV filters are
/// still pushed below joins.
public static final PinotSqlFunction ARRAY_TO_MV =
- new PinotSqlFunction("ARRAY_TO_MV", opBinding ->
opBinding.getOperandType(0).getComponentType(),
+ new PinotSqlFunction(ArrayToMvFunctionName.NAME,
+ opBinding -> arrayToMvReturnType(opBinding.getOperandType(0)),
OperandTypes.ARRAY, false);
+ /// The type of `ARRAY_TO_MV(array)`: the component type of `arrayType`. The
MV aggregate gives its MV keys the same
+ /// type, so that it can replace a plain aggregate over `ARRAY_TO_MV(array)`
without changing the row type.
+ ///
+ /// The type does not depend on the MV aggregate mode, because the `auto`
mode optimizes the already converted plan
+ /// again with the MV aggregate. A null multi-value row still goes into the
null group at runtime, as in the
+ /// single-stage engine.
+ public static RelDataType arrayToMvReturnType(RelDataType arrayType) {
+ RelDataType componentType = arrayType.getComponentType();
+ Preconditions.checkArgument(componentType != null, "ARRAY_TO_MV requires
an array, got: %s",
+ arrayType.getFullTypeString());
+ return componentType;
Review Comment:
[P1] Preserve nullability for MV group keys
The array container can be nullable while its component type is NOT NULL
(`TypeFactory` applies nullability after constructing the array). Returning the
component unchanged therefore tells Calcite that a group key cannot be null,
but `MvKeyExpander` expands a null array into one null element. Calcite
consequently removes `COALESCE` and folds null checks incorrectly. The new
self-join case in `MvAggregate.json` demonstrates this: `SELECT
COALESCE(ARRAY_TO_MV(t1.strMv), 'none'), COUNT(*) FROM tbl t1 JOIN tbl t2 ON
t1.k = t2.k GROUP BY 1` expects `[null, 1]` for the null input, although it
should return `['none', 1]`. The leaf-only issue predates this PR, but the
self-join query previously failed planning and now becomes executable through
the default `auto` fallback. Please make function inference and the MV
aggregate output nullable when either the array or its component is nullable,
and correct the regression expectations.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -544,7 +552,37 @@ private void checkAuthorization(RequesterIdentity
requesterIdentity, RequestCont
}
}
- private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders
httpHeaders, Map<String, String> queryOptions,
+ /// Applies [CommonConstants.Broker#CONFIG_OF_USE_MV_AGGREGATE] from the
cluster config, so that a change applies to
+ /// the next query without a broker restart. The cluster config wins over
the broker config. A removed key restores
+ /// the broker config value. An invalid value is logged and falls back to
the broker config value.
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ super.onChange(changedConfigs, clusterConfigs);
+ if
(!changedConfigs.contains(CommonConstants.Broker.CONFIG_OF_USE_MV_AGGREGATE)) {
+ return;
+ }
+ String value =
clusterConfigs.get(CommonConstants.Broker.CONFIG_OF_USE_MV_AGGREGATE);
Review Comment:
[P2] Honor the global cluster-config alias
An alias-only cluster setting such as
`pinot.all.multistage.use.mv.aggregate=false` silently leaves the mode at
`auto`. `ServiceStartableUtils.applyClusterConfig` documents and implements
generic `pinot.all.*` expansion to the role-specific key, but the new startup
pinning installs the broker key before that expansion can apply. This listener
then ignores the raw global key, so the setting is ineffective both at startup
and during live updates, including after a restart. Please handle changes to
both keys and resolve broker-specific cluster value, then global cluster value,
then broker configuration, as the fingerprinting listener does. Add alias-only
startup, update, and removal coverage.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -629,4 +667,171 @@ private int[] generateGroupByKeys(DataBlock dataBlock,
int numMatchedRows, Roari
}
return intKeys;
}
+
+ private void processMvAggregateWithoutFilter(MseBlock.Data block) {
+ int[][] mvKeys = generateMvGroupByKeys(block, null);
+ int numGroups = _groupIdGenerator.getNumGroups();
+ for (int i = 0; i < _aggFunctions.length; i++) {
+ AggregationFunction aggFunction = _aggFunctions[i];
+ Map<ExpressionContext, BlockValSet> blockValSetMap =
AggregateOperator.getBlockValSetMap(aggFunction, block);
+ GroupByResultHolder groupByResultHolder = _aggregateResultHolders[i];
+ groupByResultHolder.ensureCapacity(numGroups);
+ aggFunction.aggregateGroupByMV(block.getNumRows(), mvKeys,
groupByResultHolder, blockValSetMap);
+ }
+ }
+
+ /// The MV counterpart of [#processAggregateWithFilter(MseBlock.Data)]: a
filtered aggregation receives the group
+ /// ids of the matched rows only, in matched order.
+ private void processMvAggregateWithFilter(MseBlock.Data block) {
+ // In the first loop, generate all the group keys, cache the matching rows
+ int[][] mvKeys = _filteredAggregationsSkipEmptyGroups ? null :
generateMvGroupByKeys(block, null);
+ RoaringBitmap[] matchedBitmaps = new RoaringBitmap[_maxFilterArgId + 1];
+ int[] numMatchedRowsArray = new int[_maxFilterArgId + 1];
+ int[][][] filteredMvKeysArray = new int[_maxFilterArgId + 1][][];
+ for (int filterArgId : _filterArgIds) {
+ if (filterArgId < 0) {
+ // No filter for this aggregation function
+ if (mvKeys == null) {
+ mvKeys = generateMvGroupByKeys(block, null);
+ }
+ } else {
+ // Need to filter the block before aggregation
+ if (matchedBitmaps[filterArgId] == null) {
+ RoaringBitmap matchedBitmap =
AggregateOperator.getMatchedBitmap(block, filterArgId);
+ matchedBitmaps[filterArgId] = matchedBitmap;
+ numMatchedRowsArray[filterArgId] = matchedBitmap.getCardinality();
+ filteredMvKeysArray[filterArgId] = generateMvGroupByKeys(block,
matchedBitmap);
+ }
+ }
+ }
+
+ // In the second loop, aggregate the values
+ int numGroups = _groupIdGenerator.getNumGroups();
+ for (int i = 0; i < _aggFunctions.length; i++) {
+ AggregationFunction aggFunction = _aggFunctions[i];
+ GroupByResultHolder groupByResultHolder = _aggregateResultHolders[i];
+ groupByResultHolder.ensureCapacity(numGroups);
+ int filterArgId = _filterArgIds[i];
+ if (filterArgId < 0) {
+ Map<ExpressionContext, BlockValSet> blockValSetMap =
AggregateOperator.getBlockValSetMap(aggFunction, block);
+ aggFunction.aggregateGroupByMV(block.getNumRows(), mvKeys,
groupByResultHolder, blockValSetMap);
+ } else {
+ Map<ExpressionContext, BlockValSet> blockValSetMap =
+ AggregateOperator.getFilteredBlockValSetMap(aggFunction, block,
numMatchedRowsArray[filterArgId],
+ matchedBitmaps[filterArgId]);
+ aggFunction.aggregateGroupByMV(numMatchedRowsArray[filterArgId],
filteredMvKeysArray[filterArgId],
+ groupByResultHolder, blockValSetMap);
+ }
+ }
+ }
+
+ /// Creates the group ids of each row in MV mode: one id per combination of
the row's MV key elements.
+ ///
+ /// @param matchedBitmap the rows to read, or `null` for every row of the
block
+ /// @return for each selected row, in order, its group ids
([GroupIdGenerator#INVALID_ID] when the groups limit is
+ /// reached), or an empty array when one of its MV keys is an empty
array
+ private int[][] generateMvGroupByKeys(MseBlock.Data block, @Nullable
RoaringBitmap matchedBitmap) {
+ int numKeys = _groupKeyIds.length;
+ int numRows = matchedBitmap == null ? block.getNumRows() :
matchedBitmap.getCardinality();
+ Object[][] columns = new Object[numKeys][];
+ if (block.isRowHeap()) {
+ List<Object[]> rows = block.asRowHeap().getRows();
+ for (int i = 0; i < numKeys; i++) {
+ columns[i] = new Object[numRows];
+ }
+ if (matchedBitmap == null) {
+ for (int rowId = 0; rowId < numRows; rowId++) {
+ Object[] row = rows.get(rowId);
+ for (int i = 0; i < numKeys; i++) {
+ columns[i][rowId] = row[_groupKeyIds[i]];
+ }
+ }
+ } else {
+ PeekableIntIterator iterator = matchedBitmap.getIntIterator();
+ for (int matchedRowId = 0; matchedRowId < numRows; matchedRowId++) {
+ Object[] row = rows.get(iterator.next());
+ for (int i = 0; i < numKeys; i++) {
+ columns[i][matchedRowId] = row[_groupKeyIds[i]];
+ }
+ }
+ }
+ } else {
+ DataBlock dataBlock = block.asSerialized().getDataBlock();
+ for (int i = 0; i < numKeys; i++) {
+ columns[i] = matchedBitmap == null
+ ? DataBlockExtractUtils.extractKey(dataBlock, _groupKeyIds[i])
+ : DataBlockExtractUtils.extractKey(dataBlock, _groupKeyIds[i],
numRows, matchedBitmap);
+ }
+ }
+
+ int[][] mvKeys = new int[numRows][];
+ Object[][] elements = new Object[numKeys][];
+ Object[] key = numKeys > 1 ? new Object[numKeys] : null;
+ for (int rowId = 0; rowId < numRows; rowId++) {
+ mvKeys[rowId] = getMvGroupIds(columns, rowId, elements, key);
+ }
+ return mvKeys;
+ }
+
+ private int countMvKeys() {
+ int numMvKeys = 0;
+ for (boolean isMvKey : _isMvKey) {
+ if (isMvKey) {
+ numMvKeys++;
+ }
+ }
+ return numMvKeys;
+ }
+
+ /// Returns the group ids of one row. The combinations are generated as in
the single-stage engine
+ /// (`DictionaryBasedGroupKeyGenerator.getIntRawKeys`): the last key varies
fastest, the first key slowest. This order
+ /// is an implementation detail, not a contract: without ORDER BY, the
groups that survive the groups limit are not
+ /// defined.
+ ///
+ /// @param elements scratch storage for the elements of each MV key
+ /// @param key scratch storage for a multi-column key, `null` when there is
one key
+ private int[] getMvGroupIds(Object[][] columns, int rowId, Object[][]
elements, @Nullable Object[] key) {
+ int numKeys = columns.length;
+ int numCombinations = 1;
+ for (int i = 0; i < numKeys; i++) {
+ if (_isMvKey[i]) {
+ Object[] keyElements = MvKeyExpander.elements(columns[i][rowId]);
+ if (keyElements.length == 0) {
+ return NO_GROUP_IDS;
+ }
+ elements[i] = keyElements;
+ try {
+ numCombinations = Math.multiplyExact(numCombinations,
keyElements.length);
+ } catch (ArithmeticException e) {
+ throw QueryErrorCode.SERVER_RESOURCE_LIMIT_EXCEEDED.asException(
+ "Too many MV group key combinations in one row: more than " +
Integer.MAX_VALUE + " for "
+ + countMvKeys() + " MV keys", e);
+ }
+ }
+ }
+ int[] groupIds = new int[numCombinations];
Review Comment:
[P1] Bound Cartesian expansion before allocating group IDs
The integer-overflow check still allows a huge allocation before any
group-limit or termination check. One row with three 1,000-element MV keys
allocates `int[1_000_000_000]` here, about 4 GB, even with `numGroupsLimit=1`.
Repeated elements can make the actual result contain only one group.
`generateMvGroupByKeys` also retains each row's expanded array until block
aggregation finishes, and `consumeGroupBy` checks termination only after
`processBlock` returns. Thus a small joined input can exhaust server heap or
spend a billion iterations processing combinations after cancellation. Please
bound expansion before allocation or process the product in bounded batches
with periodic termination checks. The current overflow test covers products
above Integer.MAX_VALUE but misses this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
