yashmayya commented on code in PR #19316:
URL: https://github.com/apache/pinot/pull/19316#discussion_r3815867976
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java:
##########
@@ -462,18 +512,21 @@ public void aggregate(int length, AggregationResultHolder
aggregationResultHolde
for (int i = 0; i < numFilters; i++) {
FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
ThetaSketchAccumulator thetaSketchAccumulator =
thetaSketchAccumulators.get(i + 1);
- for (int j = 0; j < length; j++) {
- if (filterEvaluator.evaluate(singleValues, valueTypes, valueArrays,
j)) {
- thetaSketchAccumulator.apply(sketches[j]);
+ forEachNotNull(length, mainBlockValSet, (from, to) -> {
Review Comment:
Seven lines up, the default sketch still takes every row:
```java
if (_includeDefaultSketch) {
for (ThetaSketch sketch : sketches) { // line 508
defaultThetaAccumulator.apply(sketch);
}
}
```
There is no `forEachNotNull`, so a null row folds its default into the
sketch. The filtered loop right here skips them, and `aggregateGroupBySV` (798)
and `aggregateGroupByMV` (1187) wrap the same default case. Plain
`DISTINCTCOUNTTHETASKETCH(sketchCol)` is the one path left counting nulls with
the option on.
`deserializeSketches` also wraps every row before any filter runs. The
default for a `BYTES` column is an empty array, which is not a serialized
sketch.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java:
##########
@@ -167,7 +168,8 @@ protected void aggregateSV(int length,
AggregationResultHolder aggregationResult
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
int[] dictIds = blockValSet.getDictionaryIdsSV();
- getDictIdBitmap(aggregationResultHolder, dictionary).addN(dictIds, 0,
length);
+ forEachNotNull(length, blockValSet,
Review Comment:
The `BYTES` branch of `aggregate` (line 146) never got the wrap:
```java
CpcSketch[] sketches = deserializeSketches(bytesValues, length);
for (CpcSketch sketch : sketches) {
if (sketch != null) {
cpcSketchAccumulator.apply(sketch);
}
}
```
`aggregateGroupBySV` (276) and `aggregateGroupByMV` (411) both wrap it. The
`bytes.length > 0` test in `deserializeSketches` hides this today, because the
default for `BYTES` is an empty array. It stops hiding it when the column sets
a non-empty `defaultNullValue`: that row deserializes to a real sketch and is
counted with the option on.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java:
##########
@@ -877,158 +987,193 @@ public void aggregateGroupByMV(int length, int[][]
groupKeysArray, GroupByResult
case INT:
int[][] intValues = (int[][]) valueArrays[0];
if (_includeDefaultSketch) {
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- UpdatableThetaSketch defaultSketch =
getUpdateSketches(groupByResultHolder, groupKey).get(0);
- for (int value : intValues[i]) {
- defaultSketch.update(value);
+ forEachNotNull(length, mainBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ UpdatableThetaSketch defaultSketch =
getUpdateSketches(groupByResultHolder, groupKey).get(0);
+ for (int value : intValues[i]) {
+ defaultSketch.update(value);
+ }
}
}
- }
+ });
}
for (int i = 0; i < numFilters; i++) {
FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
- for (int j = 0; j < length; j++) {
- if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
- for (int groupKey : groupKeysArray[i]) {
- UpdatableThetaSketch updateSketch =
getUpdateSketches(groupByResultHolder, groupKey).get(i + 1);
- for (int value : intValues[i]) {
- updateSketch.update(value);
+ int filterIndex = i;
+ forEachNotNull(length, mainBlockValSet, (from, to) -> {
+ for (int j = from; j < to; j++) {
+ if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
+ for (int groupKey : groupKeysArray[filterIndex]) {
Review Comment:
Pre-existing, not from this PR, but the rename makes it visible. `j` is the
row and `filterIndex` is the filter, and both the group keys and the values
read row `filterIndex`:
```java
for (int groupKey : groupKeysArray[filterIndex]) {
...
for (int value : intValues[filterIndex]) {
```
With one filter, every matching row updates row 0's groups with row 0's
values. `master` has the same code as `groupKeysArray[i]`. 17 spots, all in
`aggregateGroupByMV`. Worth its own fix.
--
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]
