Jackie-Jiang commented on code in PR #19211:
URL: https://github.com/apache/pinot/pull/19211#discussion_r3755927878
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/CovarianceAggregationFunction.java:
##########
@@ -54,11 +56,45 @@ public class CovarianceAggregationFunction implements
AggregationFunction<Covari
protected final ExpressionContext _expression1;
protected final ExpressionContext _expression2;
protected final boolean _isSample;
+ protected final boolean _nullHandlingEnabled;
- public CovarianceAggregationFunction(List<ExpressionContext> arguments,
boolean isSample) {
+ public CovarianceAggregationFunction(List<ExpressionContext> arguments,
boolean isSample,
+ boolean nullHandlingEnabled) {
_expression1 = arguments.get(0);
_expression2 = arguments.get(1);
_isSample = isSample;
+ _nullHandlingEnabled = nullHandlingEnabled;
+ }
+
+ /// Runs `consumer` over the row ranges where **both** input columns are
non-null.
+ ///
+ /// A covariance pairs two values per row, so a row contributes only when
neither is null; the ranges are therefore
+ /// taken from the union of the two null bitmaps. With null handling
disabled the whole block is consumed, matching
+ /// the single-input helper on [NullableSingleInputAggregationFunction].
+ private void forEachNotNull(int length, BlockValSet blockValSet1,
BlockValSet blockValSet2,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ if (!_nullHandlingEnabled) {
+ consumer.consume(0, length);
+ return;
+ }
+ RoaringBitmap nullBitmap1 = blockValSet1.getNullBitmap();
+ RoaringBitmap nullBitmap2 = blockValSet2.getNullBitmap();
+ RoaringBitmap nullBitmap;
+ if (nullBitmap1 == null) {
+ nullBitmap = nullBitmap2;
+ } else if (nullBitmap2 == null) {
+ nullBitmap = nullBitmap1;
+ } else {
+ // a new bitmap, so neither block's own bitmap is mutated
+ nullBitmap = RoaringBitmap.or(nullBitmap1, nullBitmap2);
Review Comment:
Right, and better than that — I duplicated something that already exists.
`NullableSingleInputAggregationFunction.orNullIterator` already merges two
blocks' null positions as a stream through `MinIntIterator`, with
`EmptyIntIterator` for the disabled case and no bitmap materialized. Covariance
does not extend that class, so I have exposed a static variant taking the
option explicitly and had the instance method delegate to it. No new merging
logic, and the `RoaringBitmap.or` allocation is gone.
I had claimed in the description that no multi-input nullable base existed
to inherit this from. I checked for a class and never grepped for a two-block
method, so that was wrong.
One consequence worth pinning: the merged stream can report a row twice when
both columns are null in it. `forEachUnset` absorbs that because `prev =
nextNull + 1` is idempotent, and `MinIntIterator.hasNext` uses `> 0`, so a
cached `0` is dropped — safe only because a cached `0` can only be a duplicate
already emitted. `testRowNullInBothColumnsIsDroppedOnce` covers exactly that,
at index 0.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/CovarianceAggregationFunction.java:
##########
@@ -135,31 +180,31 @@ public void aggregateGroupBySV(int length, int[]
groupKeyArray, GroupByResultHol
Map<ExpressionContext, BlockValSet> blockValSetMap) {
double[] values1 =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression1);
double[] values2 =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression2);
- for (int i = 0; i < length; i++) {
- setGroupByResult(groupKeyArray[i], groupByResultHolder, values1[i],
values2[i], values1[i] * values2[i], 1L);
- }
+ forEachNotNull(length, blockValSetMap.get(_expression1),
blockValSetMap.get(_expression2), (from, to) -> {
+ for (int i = from; i < to; i++) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, values1[i],
values2[i], values1[i] * values2[i], 1L);
+ }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
double[] values1 =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression1);
double[] values2 =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression2);
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, values1[i],
values2[i], values1[i] * values2[i], 1L);
+ forEachNotNull(length, blockValSetMap.get(_expression1),
blockValSetMap.get(_expression2), (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, values1[i],
values2[i], values1[i] * values2[i], 1L);
+ }
}
- }
+ });
}
+ @Nullable
@Override
public CovarianceTuple extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
- CovarianceTuple covarianceTuple = aggregationResultHolder.getResult();
- if (covarianceTuple == null) {
- return new CovarianceTuple(0.0, 0.0, 0.0, 0L);
- } else {
- return covarianceTuple;
- }
+ return aggregationResultHolder.getResult();
Review Comment:
Correct, and I have verified it. The pre-change `extractFinalResult(null)`
returned `null`, so a new server sending a `null` intermediate to an old broker
renders `NULL` where that broker previously received a zero-count tuple and
rendered `-Infinity`.
There is no NPE on that path: `AggregationFunctionUtils.merge` settles a
null on both sides, so the old broker never calls `merge` with one. It is a
rendered-answer difference, confined to the mixed-version window, and upgrading
brokers first avoids it.
I have not preserved the zero-count tuple for disabled mode, because doing
so keeps the substitution that this change exists to remove, and the contract
on `AggregationFunction` treats the substitution and the missing option as one
deviation rather than two. The description now states the mixed-version
behaviour explicitly instead of claiming the disabled mode is unaffected, which
is what it said before and was wrong.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java:
##########
@@ -458,9 +458,9 @@ public static AggregationFunction
getAggregationFunction(FunctionContext functio
case HISTOGRAM:
return new HistogramAggregationFunction(arguments);
case COVARPOP:
- return new CovarianceAggregationFunction(arguments, false);
+ return new CovarianceAggregationFunction(arguments, false,
nullHandlingEnabled);
case COVARSAMP:
- return new CovarianceAggregationFunction(arguments, true);
+ return new CovarianceAggregationFunction(arguments, true,
nullHandlingEnabled);
Review Comment:
Agreed. `NullHandlingEnabledQueriesTest` now covers this end to end over
real segments containing nulls, so the option is verified to reach the function
through the factory and the result to survive server-to-broker rendering: rows
dropped when either column is null, an all-null input answering `NULL`, and the
disabled mode still folding the column default.
One case did not survive there. That harness sets `_indexSegments =
asList(segment, segment)`, so a single-row fixture is queried twice and a
`COVAR_SAMP` over 'one row' is really two identical rows, whose sample
covariance is exactly `0.0`. That assertion lives in the unit test, where the
input size is exact.
--
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]
