yashmayya commented on code in PR #19357:
URL: https://github.com/apache/pinot/pull/19357#discussion_r3855098923
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ParentExprMinMaxAggregationFunction.java:
##########
@@ -111,25 +116,26 @@ public GroupByResultHolder createGroupByResultHolder(int
initialCapacity, int ma
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- ExprMinMaxObject exprMinMaxObject = aggregationResultHolder.getResult();
+ initializeWithNewDataBlocks(blockValSetMap);
- if (exprMinMaxObject == null) {
- initializeWithNewDataBlocks(blockValSetMap);
- exprMinMaxObject = new ExprMinMaxObject(_measuringColumnSchema.get(),
_projectionColumnSchema.get());
- }
+ ExprMinMaxObject existing = aggregationResultHolder.getResult();
+ ExprMinMaxObject exprMinMaxObject = existing != null ? existing
+ : new ExprMinMaxObject(_measuringColumnSchema.get(),
_projectionColumnSchema.get());
List<Integer> rowIds = new ArrayList<>();
- for (int i = 0; i < length; i++) {
- int compareResult =
exprMinMaxObject.compareAndSetKey(_exprMinMaxWrapperMeasuringColumnSets.get(),
i, _isMax);
- if (compareResult == 0) {
- // same key, add the rowId to the list
- rowIds.add(i);
- } else if (compareResult > 0) {
- // new key is set, clear the list and add the new rowId
- rowIds.clear();
- rowIds.add(i);
+ forEachNotNullMeasuring(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int compareResult =
exprMinMaxObject.compareAndSetKey(_exprMinMaxWrapperMeasuringColumnSets.get(),
i, _isMax);
+ if (compareResult == 0) {
+ // same key, add the rowId to the list
+ rowIds.add(i);
+ } else if (compareResult > 0) {
+ // new key is set, clear the list and add the new rowId
+ rowIds.clear();
+ rowIds.add(i);
+ }
}
- }
+ });
// for all the rows that are associated with the extremum key, add the
projection columns
for (Integer rowId : rowIds) {
Review Comment:
`secondBlockIsReadWithItsOwnValues` fails on this commit. I reproduced it
locally, standalone:
```
ExprMinMaxNullHandlingTest.secondBlockIsReadWithItsOwnValues:101 expected
[1] but found [2]
```
The key is right; the row count is not. Rebinding the wrappers fixed half of
the second-block problem and uncovered the other half.
`rowIds` is cleared when a new extremum arrives, but it only holds rows of
the **current** block. The values added from earlier blocks stay in
`exprMinMaxObject`, and this loop only ever calls `addVal`. So block 1 leaves
`100` behind, block 2 wins with key `3`, and the object reports two rows for a
key only one row holds.
`updateGroupByResult` does not have this problem, because it calls
`setToNewVal` on `compareResult > 0`, and that clears
`_extremumProjectionValues` first. The batched path here has no equivalent.
Remembering whether the key was replaced in this block, and using
`setToNewVal` for the first surviving row, makes the test pass. All six tests
in the class pass, `ExprMinMaxTest` stays at 10/10, and
`AggregationFunctionNullContractTest` stays at 581.
The CI log for Unit Test Set 1 stops at the 4 MB cap before the summary,
which is why the failure is not visible there.
--
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]
