This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 59c2883cf02 Give the funnel functions the query's null handling option
(#19332)
59c2883cf02 is described below
commit 59c2883cf021755684c9d4bc62d2e1179e150490
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Aug 21 16:08:28 2026 -0700
Give the funnel functions the query's null handling option (#19332)
---
.../aggregation/function/AggregationFunction.java | 62 +++---
.../function/AggregationFunctionFactory.java | 12 +-
.../function/HistogramAggregationFunction.java | 222 ++++++++++----------
.../function/funnel/AggregationStrategy.java | 158 +++++++++-----
.../function/funnel/BitmapAggregationStrategy.java | 4 +-
.../FunnelCountAggregationFunctionFactory.java | 12 +-
.../FunnelCountSortedAggregationFunction.java | 5 +-
.../function/funnel/SortedAggregationStrategy.java | 4 +-
.../funnel/ThetaSketchAggregationStrategy.java | 4 +-
.../window/FunnelBaseAggregationFunction.java | 134 +++++++-----
.../FunnelCompleteCountAggregationFunction.java | 4 +-
...unnelEventsFunctionEvalAggregationFunction.java | 151 ++++++++-----
.../window/FunnelMatchStepAggregationFunction.java | 4 +-
.../window/FunnelMaxStepAggregationFunction.java | 4 +-
...FunnelStepDurationStatsAggregationFunction.java | 4 +-
.../pinot/core/common/SyntheticBlockValSets.java | 69 +++++-
.../AggregationFunctionNullContractTest.java | 93 ++++++--
.../function/funnel/FunnelNullHandlingTest.java | 233 +++++++++++++++++++++
18 files changed, 840 insertions(+), 339 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
index b95b34a36f2..79ca2736e8a 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
@@ -64,9 +64,9 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
/// matching value was null. It carries no per-function meaning, which is
what makes it correct for the aggregation
/// methods to skip null rows outright rather than fold them in.
/// `null` is not the only way that state is represented, though: an empty
accumulator — an empty set, value list,
-/// digest, sketch or map — means the same thing, and still arrives. A peer
that serialized one sends it, and so do
-/// the functions in the first deviation below, which never receive the
option and substitute one rather than
-/// returning `null`. So [#extractFinalResult] accepts both.
+/// digest, sketch or map — means the same thing, and still arrives. A peer
that serialized one sends it, and so
+/// does any function whose disabled-mode answer is an empty accumulator
rather than a constant, since it
+/// substitutes one in that mode to keep the answer it has always given. So
[#extractFinalResult] accepts both.
/// `null` is the representation to produce where there is a choice, because
it is the only one always available:
/// `MAXSTRING`, `MINSTRING` and `ANYVALUE` are object-backed and have no
empty value to substitute.
/// - [#extractFinalResult] decides what that means for this aggregation, and
is the only method that does. `COUNT`
@@ -95,37 +95,31 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
/// these is therefore a behavioural change and not a documentation fix,
because it widens what the rest of the
/// engine must carry.
///
-/// TODO: Known deviations from the above.
-/// 1. Several aggregation functions never receive the query's null handling
option at all, so they cannot skip
-/// null rows and fold the column's default value into the aggregate
whatever the query asked for: the
-/// sketch-backed distinct counts (`DISTINCTCOUNTBITMAP`,
`DISTINCTCOUNTHLL`, `DISTINCTCOUNTTHETASKETCH`,
-/// `DISTINCTCOUNTCPCSKETCH`, `FASTHLL`,
`SEGMENTPARTITIONEDDISTINCTCOUNT` and the raw and smart variants of
-/// each), `HISTOGRAM`, `IDSET`, `STUNION`, the array sums, and the
funnel family. Whether a function takes the
-/// option is visible at its construction site, which is the reliable way
to tell.
-/// The family names are not a safe shorthand for this, in either
direction. The exact distinct functions
-/// (`DISTINCTCOUNT`, `DISTINCTSUM`, `DISTINCTAVG`,
`DISTINCTCOUNTOFFHEAP`) do take the option and skip null
-/// rows through their shared base, as do the variance,
standard-deviation and covariance functions and the
-/// first/last-with-time functions, so "the distinct-count family" and
"the statistical functions" both include
-/// members that honour the option and members that cannot.
-/// These same functions also substitute an empty accumulator in
[#extractAggregationResult],
-/// [#extractGroupByResult] or both, rather than returning `null` and
letting [#extractFinalResult] render the
-/// disabled-mode value. That is not a separate defect: without the
option [#extractFinalResult] cannot tell the
-/// two modes apart, so it has nothing to decide with and the
substitution is the only thing holding the answer.
-/// Which of the two paths substitutes is inconsistent across them and
sometimes within one, so such a function
-/// can answer differently depending on whether the query groups — a
filtered aggregation makes that observable,
-/// since one group key space is shared by every aggregation in a query
and a group created by one of them can
-/// hold no rows for another.
-/// Conforming them alongside the option is more than moving a branch.
The value to preserve is often not a
-/// constant, and the function that substitutes is frequently not the one
that renders: the raw variants
-/// delegate extraction to the plain function and serialize what it
returns, so conforming the plain function
-/// alone changes the raw variant's answer from a serialized empty sketch
to `NULL` without touching the raw
-/// variant at all.
-/// 2. The multi-stage engine constructs every aggregation function with
null handling enabled and never consults the
-/// query's null handling option, so a query that disables it still gets
enabled-mode semantics there. The two
-/// engines can therefore answer the same query differently: with null
handling disabled, `SUM` over a query
-/// whose segments are all pruned is `NULL` on the multi-stage engine and
`0` on the single-stage engine. This
-/// may be intended, the multi-stage engine being the SQL-conformant one,
but it means the mode described above
-/// is not actually per-query everywhere.
+/// ### Multi-input functions
+///
+/// A function that reads more than one column decides for itself which of
them a `null` disqualifies the row on,
+/// because that follows from what each column is for rather than from a
single rule. The covariances skip a row
+/// unless both value columns are non-null. A funnel skips a row whose
timestamp or correlation key is null, since
+/// neither an event with no position in the window nor a row belonging to no
key can contribute, but it does not
+/// skip a row for a null step: a step expression is a predicate, and a
predicate over a null operand is UNKNOWN,
+/// which SQL treats as not satisfied wherever a boolean is consumed, so a
null step already means that step did not
+/// match.
+///
+/// ### Functions outside this contract
+///
+/// Every user-facing aggregation receives the option. Three do not, and none
of them is an aggregate a query can
+/// ask for by name: `ParentExprMinMax` and `ChildExprMinMax` are produced by
the query rewriter rather than written
+/// by users, and `TimeSeriesAggregationFunction` is built from time-series
plan context rather than from a
+/// [org.apache.pinot.common.request.context.FunctionContext]. Whether a
function takes the option is visible at its
+/// construction site in [AggregationFunctionFactory], which is the reliable
way to tell.
+///
+/// TODO: Known deviation from the above.
+/// The multi-stage engine constructs every aggregation function with null
handling enabled and never consults the
+/// query's null handling option, so a query that disables it still gets
enabled-mode semantics there. The two
+/// engines can therefore answer the same query differently: with null
handling disabled, `SUM` over a query
+/// whose segments are all pruned is `NULL` on the multi-stage engine and
`0` on the single-stage engine. This
+/// may be intended, the multi-stage engine being the SQL-conformant one,
but it means the mode described above
+/// is not actually per-query everywhere.
///
/// @param <IntermediateResult> Intermediate result generated from segment
/// @param <FinalResult> Final result used in broker response
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
index 87b97a297e7..4c4165edbdb 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
@@ -511,17 +511,17 @@ public class AggregationFunctionFactory {
throw new IllegalArgumentException(
"Aggregation function: " + functionType + " is only supported
in selection without alias.");
case FUNNELCOUNT:
- return new FunnelCountAggregationFunctionFactory(arguments).get();
+ return new FunnelCountAggregationFunctionFactory(arguments,
nullHandlingEnabled).get();
case FUNNELMAXSTEP:
- return new FunnelMaxStepAggregationFunction(arguments);
+ return new FunnelMaxStepAggregationFunction(arguments,
nullHandlingEnabled);
case FUNNELMATCHSTEP:
- return new FunnelMatchStepAggregationFunction(arguments);
+ return new FunnelMatchStepAggregationFunction(arguments,
nullHandlingEnabled);
case FUNNELCOMPLETECOUNT:
- return new FunnelCompleteCountAggregationFunction(arguments);
+ return new FunnelCompleteCountAggregationFunction(arguments,
nullHandlingEnabled);
case FUNNELSTEPDURATIONSTATS:
- return new FunnelStepDurationStatsAggregationFunction(arguments);
+ return new FunnelStepDurationStatsAggregationFunction(arguments,
nullHandlingEnabled);
case FUNNELEVENTSFUNCTIONEVAL:
- return new FunnelEventsFunctionEvalAggregationFunction(arguments);
+ return new FunnelEventsFunctionEvalAggregationFunction(arguments,
nullHandlingEnabled);
case FREQUENTSTRINGSSKETCH:
return new FrequentStringsSketchAggregationFunction(arguments,
nullHandlingEnabled);
case FREQUENTLONGSSKETCH:
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
index 80830edcd92..6ca0f29891b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
@@ -259,76 +259,67 @@ public class HistogramAggregationFunction
}
@Override
- public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
if (blockValSet.isSingleValue()) {
- aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ aggregateSV(length, aggregationResultHolder, blockValSet);
} else {
- aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ aggregateMV(length, aggregationResultHolder, blockValSet);
}
}
- private void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
- BlockValSet blockValSet) {
+ private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ double[] histogram = new double[getNumBins()];
+ int numRows;
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[] values = blockValSet.getIntValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- int value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, values[i]);
}
+ return acum + to - from;
});
break;
}
case LONG: {
long[] values = blockValSet.getLongValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- long value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, values[i]);
}
+ return acum + to - from;
});
break;
}
case FLOAT: {
float[] values = blockValSet.getFloatValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- float value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, values[i]);
}
+ return acum + to - from;
});
break;
}
case DOUBLE: {
double[] values = blockValSet.getDoubleValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- double value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, values[i]);
}
+ return acum + to - from;
});
break;
}
case BIG_DECIMAL: {
BigDecimal[] values = blockValSet.getBigDecimalValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- double value = values[i].doubleValue();
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, values[i].doubleValue());
}
+ return acum + to - from;
});
break;
}
@@ -336,73 +327,75 @@ public class HistogramAggregationFunction
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
+ // The histogram is published only when a row reached it, so a block with
no non-null row leaves the holder
+ // untouched and extractFinalResult sees the null that means nothing was
aggregated. It is published once rather
+ // than per range, because the buffer accumulates across ranges and adding
it again would recount earlier rows.
+ if (numRows > 0) {
+ setAggregationResult(aggregationResultHolder, histogram);
+ }
}
- private void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
- BlockValSet blockValSet) {
+ private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ double[] histogram = new double[getNumBins()];
+ int numRows;
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[][] values = blockValSet.getIntValuesMV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (int value : values[i]) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, value);
}
}
+ return acum + to - from;
});
break;
}
case LONG: {
long[][] values = blockValSet.getLongValuesMV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (long value : values[i]) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, value);
}
}
+ return acum + to - from;
});
break;
}
case FLOAT: {
float[][] values = blockValSet.getFloatValuesMV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (float value : values[i]) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, value);
}
}
+ return acum + to - from;
});
break;
}
case DOUBLE: {
double[][] values = blockValSet.getDoubleValuesMV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (double value : values[i]) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
- }
+ increment(histogram, value);
}
}
+ return acum + to - from;
});
break;
}
case BIG_DECIMAL: {
BigDecimal[][] values = blockValSet.getBigDecimalValuesMV();
- forEachNotNull(length, blockValSet, (from, to) -> {
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (BigDecimal value : values[i]) {
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder,
value.doubleValue());
- }
+ increment(histogram, value.doubleValue());
}
}
+ return acum + to - from;
});
break;
}
@@ -410,6 +403,9 @@ public class HistogramAggregationFunction
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
+ if (numRows > 0) {
+ setAggregationResult(aggregationResultHolder, histogram);
+ }
}
@Override
@@ -546,80 +542,77 @@ public class HistogramAggregationFunction
}
}
- protected void setGroupByResult(int groupKey, GroupByResultHolder
groupByResultHolder, double val) {
- int binID = getBinId(val);
- DoubleArrayList byResultHolderResult =
groupByResultHolder.getResult(groupKey);
- if (byResultHolderResult == null) {
- byResultHolderResult =
DoubleVectorOpUtils.createAndInitialize(getNumBins());
- groupByResultHolder.setValueForKey(groupKey, byResultHolderResult);
- }
- if (binID != INVALID_BIN) {
- DoubleVectorOpUtils.incrementElementByOne(byResultHolderResult, binID);
- }
- }
-
@Override
- public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
+ public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
if (blockValSet.isSingleValue()) {
- aggregateSV(length, aggregationResultHolder, blockValSet);
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
} else {
- aggregateMV(length, aggregationResultHolder, blockValSet);
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
}
}
- private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
- double[] histogram = new double[getNumBins()];
- int numRows;
+ private void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[] values = blockValSet.getIntValuesSV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- increment(histogram, values[i]);
+ int value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- return acum + to - from;
});
break;
}
case LONG: {
long[] values = blockValSet.getLongValuesSV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- increment(histogram, values[i]);
+ long value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- return acum + to - from;
});
break;
}
case FLOAT: {
float[] values = blockValSet.getFloatValuesSV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- increment(histogram, values[i]);
+ float value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- return acum + to - from;
});
break;
}
case DOUBLE: {
double[] values = blockValSet.getDoubleValuesSV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- increment(histogram, values[i]);
+ double value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- return acum + to - from;
});
break;
}
case BIG_DECIMAL: {
BigDecimal[] values = blockValSet.getBigDecimalValuesSV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
- increment(histogram, values[i].doubleValue());
+ double value = values[i].doubleValue();
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- return acum + to - from;
});
break;
}
@@ -627,75 +620,73 @@ public class HistogramAggregationFunction
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
- // The histogram is published only when a row reached it, so a block with
no non-null row leaves the holder
- // untouched and extractFinalResult sees the null that means nothing was
aggregated. It is published once rather
- // than per range, because the buffer accumulates across ranges and adding
it again would recount earlier rows.
- if (numRows > 0) {
- setAggregationResult(aggregationResultHolder, histogram);
- }
}
- private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
- double[] histogram = new double[getNumBins()];
- int numRows;
+ private void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[][] values = blockValSet.getIntValuesMV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (int value : values[i]) {
- increment(histogram, value);
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
}
- return acum + to - from;
});
break;
}
case LONG: {
long[][] values = blockValSet.getLongValuesMV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (long value : values[i]) {
- increment(histogram, value);
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
}
- return acum + to - from;
});
break;
}
case FLOAT: {
float[][] values = blockValSet.getFloatValuesMV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (float value : values[i]) {
- increment(histogram, value);
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
}
- return acum + to - from;
});
break;
}
case DOUBLE: {
double[][] values = blockValSet.getDoubleValuesMV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (double value : values[i]) {
- increment(histogram, value);
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
}
- return acum + to - from;
});
break;
}
case BIG_DECIMAL: {
BigDecimal[][] values = blockValSet.getBigDecimalValuesMV();
- numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to && i < values.length; i++) {
for (BigDecimal value : values[i]) {
- increment(histogram, value.doubleValue());
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder,
value.doubleValue());
+ }
}
}
- return acum + to - from;
});
break;
}
@@ -703,8 +694,17 @@ public class HistogramAggregationFunction
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
- if (numRows > 0) {
- setAggregationResult(aggregationResultHolder, histogram);
+ }
+
+ protected void setGroupByResult(int groupKey, GroupByResultHolder
groupByResultHolder, double val) {
+ int binID = getBinId(val);
+ DoubleArrayList byResultHolderResult =
groupByResultHolder.getResult(groupKey);
+ if (byResultHolderResult == null) {
+ byResultHolderResult =
DoubleVectorOpUtils.createAndInitialize(getNumBins());
+ groupByResultHolder.setValueForKey(groupKey, byResultHolderResult);
+ }
+ if (binID != INVALID_BIN) {
+ DoubleVectorOpUtils.incrementElementByOne(byResultHolderResult, binID);
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/AggregationStrategy.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/AggregationStrategy.java
index 1de0f923b89..5cf401c333b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/AggregationStrategy.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/AggregationStrategy.java
@@ -21,12 +21,15 @@ package
org.apache.pinot.core.query.aggregation.function.funnel;
import com.google.common.base.Preconditions;
import java.util.List;
import java.util.Map;
+import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.utils.RoaringBitmapUtils;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.roaringbitmap.RoaringBitmap;
/// Interface for within segment aggregation strategy.
@@ -49,7 +52,11 @@ public abstract class AggregationStrategy<A> {
private final List<ExpressionContext> _correlateByExpressions;
private final ExpressionContext _primaryCorrelationCol;
- public AggregationStrategy(List<ExpressionContext> stepExpressions,
List<ExpressionContext> correlateByExpressions) {
+ protected final boolean _nullHandlingEnabled;
+
+ public AggregationStrategy(List<ExpressionContext> stepExpressions,
List<ExpressionContext> correlateByExpressions,
+ boolean nullHandlingEnabled) {
+ _nullHandlingEnabled = nullHandlingEnabled;
_stepExpressions = stepExpressions;
_correlateByExpressions = correlateByExpressions;
_primaryCorrelationCol = _correlateByExpressions.get(0);
@@ -117,13 +124,15 @@ public abstract class AggregationStrategy<A> {
final Dictionary dictionary = getPrimaryDictionary(blockValSetMap);
final int[] correlationIds = getPrimaryCorrelationIds(blockValSetMap);
final A aggResult = getAggregationResult(dictionary,
aggregationResultHolder);
- for (int i = 0; i < length; i++) {
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- add(dictionary, aggResult, n, correlationIds[i]);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ add(dictionary, aggResult, n, correlationIds[i]);
+ }
}
}
- }
+ });
}
private void aggregateMultiKey(int length, AggregationResultHolder
aggregationResultHolder,
@@ -132,16 +141,18 @@ public abstract class AggregationStrategy<A> {
final int[][] allCorrelationIds = getAllCorrelationDictIds(blockValSetMap);
final A aggResult = getAggregationResultMultiKey(dictionaries,
aggregationResultHolder);
final int[] rowDictIds = new int[_numCorrelateByKeys];
- for (int i = 0; i < length; i++) {
- for (int k = 0; k < _numCorrelateByKeys; k++) {
- rowDictIds[k] = allCorrelationIds[k][i];
- }
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int k = 0; k < _numCorrelateByKeys; k++) {
+ rowDictIds[k] = allCorrelationIds[k][i];
+ }
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ }
}
}
- }
+ });
}
/// Performs aggregation on the given group key array and block value sets
(aggregation group-by on single-value
@@ -160,15 +171,17 @@ public abstract class AggregationStrategy<A> {
Map<ExpressionContext, BlockValSet> blockValSetMap, int[][] steps) {
final Dictionary dictionary = getPrimaryDictionary(blockValSetMap);
final int[] correlationIds = getPrimaryCorrelationIds(blockValSetMap);
- for (int i = 0; i < length; i++) {
- final int groupKey = groupKeyArray[i];
- final A aggResult = getAggregationResultGroupBy(dictionary,
groupByResultHolder, groupKey);
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- add(dictionary, aggResult, n, correlationIds[i]);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ final int groupKey = groupKeyArray[i];
+ final A aggResult = getAggregationResultGroupBy(dictionary,
groupByResultHolder, groupKey);
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ add(dictionary, aggResult, n, correlationIds[i]);
+ }
}
}
- }
+ });
}
private void aggregateGroupBySVMultiKey(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
@@ -176,18 +189,20 @@ public abstract class AggregationStrategy<A> {
final Dictionary[] dictionaries = getAllDictionaries(blockValSetMap);
final int[][] allCorrelationIds = getAllCorrelationDictIds(blockValSetMap);
final int[] rowDictIds = new int[_numCorrelateByKeys];
- for (int i = 0; i < length; i++) {
- for (int k = 0; k < _numCorrelateByKeys; k++) {
- rowDictIds[k] = allCorrelationIds[k][i];
- }
- final int groupKey = groupKeyArray[i];
- final A aggResult = getAggregationResultGroupByMultiKey(dictionaries,
groupByResultHolder, groupKey);
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int k = 0; k < _numCorrelateByKeys; k++) {
+ rowDictIds[k] = allCorrelationIds[k][i];
+ }
+ final int groupKey = groupKeyArray[i];
+ final A aggResult = getAggregationResultGroupByMultiKey(dictionaries,
groupByResultHolder, groupKey);
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ }
}
}
- }
+ });
}
/// Performs aggregation on the given group keys array and block value sets
(aggregation group-by on multi-value
@@ -206,16 +221,18 @@ public abstract class AggregationStrategy<A> {
GroupByResultHolder groupByResultHolder, Map<ExpressionContext,
BlockValSet> blockValSetMap, int[][] steps) {
final Dictionary dictionary = getPrimaryDictionary(blockValSetMap);
final int[] correlationIds = getPrimaryCorrelationIds(blockValSetMap);
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- final A aggResult = getAggregationResultGroupBy(dictionary,
groupByResultHolder, groupKey);
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- add(dictionary, aggResult, n, correlationIds[i]);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ final A aggResult = getAggregationResultGroupBy(dictionary,
groupByResultHolder, groupKey);
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ add(dictionary, aggResult, n, correlationIds[i]);
+ }
}
}
}
- }
+ });
}
private void aggregateGroupByMVMultiKey(int length, int[][] groupKeysArray,
@@ -223,19 +240,21 @@ public abstract class AggregationStrategy<A> {
final Dictionary[] dictionaries = getAllDictionaries(blockValSetMap);
final int[][] allCorrelationIds = getAllCorrelationDictIds(blockValSetMap);
final int[] rowDictIds = new int[_numCorrelateByKeys];
- for (int i = 0; i < length; i++) {
- for (int k = 0; k < _numCorrelateByKeys; k++) {
- rowDictIds[k] = allCorrelationIds[k][i];
- }
- for (int groupKey : groupKeysArray[i]) {
- final A aggResult = getAggregationResultGroupByMultiKey(dictionaries,
groupByResultHolder, groupKey);
- for (int n = 0; n < _numSteps; n++) {
- if (steps[n][i] > 0) {
- addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ forEachNotNullCorrelation(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int k = 0; k < _numCorrelateByKeys; k++) {
+ rowDictIds[k] = allCorrelationIds[k][i];
+ }
+ for (int groupKey : groupKeysArray[i]) {
+ final A aggResult =
getAggregationResultGroupByMultiKey(dictionaries, groupByResultHolder,
groupKey);
+ for (int n = 0; n < _numSteps; n++) {
+ if (steps[n][i] > 0) {
+ addMultiKey(aggResult, n, dictionaries, rowDictIds);
+ }
}
}
}
- }
+ });
}
/// Adds a correlation id to the aggregation counter for a given step in the
funnel.
@@ -282,6 +301,49 @@ public abstract class AggregationStrategy<A> {
return allIds;
}
+ /// Runs the consumer over each range of rows whose correlation key is
entirely non-null, or over the whole block
+ /// when the option is disabled.
+ ///
+ /// Only the correlation key is consulted. A step expression is a predicate,
and a predicate over a null operand is
+ /// UNKNOWN, which SQL treats as not satisfied wherever a boolean is
consumed, so a null step already means that
+ /// step did not match. The correlation key is what the funnel counts
distinct values of, and a row whose key is
+ /// null belongs to no key at all; its dictionary id would otherwise be the
default's, counted as a real one. With
+ /// a composite key a null in any component leaves the whole key undefined,
so the row is skipped.
+ private void forEachNotNullCorrelation(int length, Map<ExpressionContext,
BlockValSet> blockValSetMap,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ RoaringBitmap nullBitmap = correlationNullBitmap(blockValSetMap);
+ if (nullBitmap == null) {
+ consumer.consume(0, length);
+ return;
+ }
+ // Skip if the entire block is null
+ if (!nullBitmap.contains(0, length)) {
+ RoaringBitmapUtils.forEachUnset(length, nullBitmap.getIntIterator(),
consumer);
+ }
+ }
+
+ /// Returns the union of the correlation columns' null bitmaps, or `null`
when no row is null.
+ @Nullable
+ private RoaringBitmap correlationNullBitmap(Map<ExpressionContext,
BlockValSet> blockValSetMap) {
+ if (!_nullHandlingEnabled) {
+ return null;
+ }
+ RoaringBitmap merged = null;
+ for (ExpressionContext correlateByExpression : _correlateByExpressions) {
+ RoaringBitmap nullBitmap =
blockValSetMap.get(correlateByExpression).getNullBitmap();
+ if (nullBitmap == null) {
+ continue;
+ }
+ // Copied before merging: the bitmap belongs to the block and must not
be mutated
+ if (merged == null) {
+ merged = nullBitmap.clone();
+ } else {
+ merged.or(nullBitmap);
+ }
+ }
+ return merged;
+ }
+
private int[][] getSteps(Map<ExpressionContext, BlockValSet> blockValSetMap)
{
final int[][] steps = new int[_numSteps][];
for (int n = 0; n < _numSteps; n++) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapAggregationStrategy.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapAggregationStrategy.java
index da648e13f39..3f76e8cdbe9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapAggregationStrategy.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapAggregationStrategy.java
@@ -26,8 +26,8 @@ import org.apache.pinot.segment.spi.index.reader.Dictionary;
/// Aggregation strategy leveraging roaring bitmap algebra
(unions/intersections).
class BitmapAggregationStrategy extends AggregationStrategy<DictIdsWrapper> {
public BitmapAggregationStrategy(List<ExpressionContext> stepExpressions,
- List<ExpressionContext> correlateByExpressions) {
- super(stepExpressions, correlateByExpressions);
+ List<ExpressionContext> correlateByExpressions, boolean
nullHandlingEnabled) {
+ super(stepExpressions, correlateByExpressions, nullHandlingEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunctionFactory.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunctionFactory.java
index 30b84f40b00..d59a610c94d 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunctionFactory.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunctionFactory.java
@@ -62,7 +62,10 @@ public class FunnelCountAggregationFunctionFactory
implements Supplier<Aggregati
final boolean _thetaSketchSetting;
final boolean _setSetting;
- public FunnelCountAggregationFunctionFactory(List<ExpressionContext>
expressions) {
+ private final boolean _nullHandlingEnabled;
+
+ public FunnelCountAggregationFunctionFactory(List<ExpressionContext>
expressions, boolean nullHandlingEnabled) {
+ _nullHandlingEnabled = nullHandlingEnabled;
_expressions = expressions;
Option.validate(expressions);
_correlateByExpressions =
Option.CORRELATE_BY.getInputExpressions(expressions);
@@ -120,7 +123,7 @@ public class FunnelCountAggregationFunctionFactory
implements Supplier<Aggregati
MergeStrategy<List<Long>> mergeStrategy) {
if (_sortingSetting) {
return new FunnelCountSortedAggregationFunction<>(_expressions,
_stepExpressions, _correlateByExpressions,
- aggregationStrategy, resultExtractionStrategy, mergeStrategy);
+ aggregationStrategy, resultExtractionStrategy, mergeStrategy,
_nullHandlingEnabled);
} else {
return new FunnelCountAggregationFunction<>(_expressions,
_stepExpressions, _correlateByExpressions,
aggregationStrategy, resultExtractionStrategy, mergeStrategy);
@@ -128,11 +131,12 @@ public class FunnelCountAggregationFunctionFactory
implements Supplier<Aggregati
}
AggregationStrategy<UpdatableThetaSketch[]> thetaSketchAggregationStrategy()
{
- return new ThetaSketchAggregationStrategy(_stepExpressions,
_correlateByExpressions, _nominalEntries);
+ return new ThetaSketchAggregationStrategy(_stepExpressions,
_correlateByExpressions, _nominalEntries,
+ _nullHandlingEnabled);
}
AggregationStrategy<DictIdsWrapper> bitmapAggregationStrategy() {
- return new BitmapAggregationStrategy(_stepExpressions,
_correlateByExpressions);
+ return new BitmapAggregationStrategy(_stepExpressions,
_correlateByExpressions, _nullHandlingEnabled);
}
MergeStrategy<List<ThetaSketch>> thetaSketchMergeStrategy() {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountSortedAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountSortedAggregationFunction.java
index 4a766e0ad4f..df4fc17b9a3 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountSortedAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountSortedAggregationFunction.java
@@ -54,10 +54,11 @@ public class FunnelCountSortedAggregationFunction<A>
extends FunnelCountAggregat
public FunnelCountSortedAggregationFunction(List<ExpressionContext>
expressions,
List<ExpressionContext> stepExpressions, List<ExpressionContext>
correlateByExpressions,
AggregationStrategy<A> aggregationStrategy, ResultExtractionStrategy<A,
List<Long>> resultExtractionStrategy,
- MergeStrategy<List<Long>> mergeStrategy) {
+ MergeStrategy<List<Long>> mergeStrategy, boolean nullHandlingEnabled) {
super(expressions, stepExpressions, correlateByExpressions,
aggregationStrategy, resultExtractionStrategy,
mergeStrategy);
- _sortedAggregationStrategy = new
SortedAggregationStrategy(stepExpressions, correlateByExpressions);
+ _sortedAggregationStrategy =
+ new SortedAggregationStrategy(stepExpressions, correlateByExpressions,
nullHandlingEnabled);
_sortedResultExtractionStrategy = SortedAggregationResult::extractResult;
_primaryCorrelationCol = correlateByExpressions.get(0);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SortedAggregationStrategy.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SortedAggregationStrategy.java
index 154e84837fa..7e423a10949 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SortedAggregationStrategy.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SortedAggregationStrategy.java
@@ -28,8 +28,8 @@ import org.apache.pinot.segment.spi.index.reader.Dictionary;
/// keys are handled within each primary-key group by
[SortedAggregationResult].
class SortedAggregationStrategy extends
AggregationStrategy<SortedAggregationResult> {
public SortedAggregationStrategy(List<ExpressionContext> stepExpressions,
- List<ExpressionContext> correlateByExpressions) {
- super(stepExpressions, correlateByExpressions);
+ List<ExpressionContext> correlateByExpressions, boolean
nullHandlingEnabled) {
+ super(stepExpressions, correlateByExpressions, nullHandlingEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/ThetaSketchAggregationStrategy.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/ThetaSketchAggregationStrategy.java
index 406c9db6617..23cc4043e60 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/ThetaSketchAggregationStrategy.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/ThetaSketchAggregationStrategy.java
@@ -30,8 +30,8 @@ class ThetaSketchAggregationStrategy extends
AggregationStrategy<UpdatableThetaS
final UpdatableThetaSketchBuilder _updateSketchBuilder;
public ThetaSketchAggregationStrategy(List<ExpressionContext>
stepExpressions,
- List<ExpressionContext> correlateByExpressions, int nominalEntries) {
- super(stepExpressions, correlateByExpressions);
+ List<ExpressionContext> correlateByExpressions, int nominalEntries,
boolean nullHandlingEnabled) {
+ super(stepExpressions, correlateByExpressions, nullHandlingEnabled);
_updateSketchBuilder = new
UpdatableThetaSketchBuilder().setNominalEntries(nominalEntries);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelBaseAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelBaseAggregationFunction.java
index f63350b0305..243cf4a71d8 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelBaseAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelBaseAggregationFunction.java
@@ -30,6 +30,7 @@ import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.common.utils.RoaringBitmapUtils;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
@@ -39,10 +40,12 @@ import
org.apache.pinot.core.query.aggregation.function.funnel.FunnelStepEvent;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.spi.query.QueryThreadContext;
+import org.roaringbitmap.RoaringBitmap;
public abstract class FunnelBaseAggregationFunction<F extends Comparable>
implements AggregationFunction<PriorityQueue<FunnelStepEvent>, F> {
+ protected final boolean _nullHandlingEnabled;
protected final ExpressionContext _timestampExpression;
protected final long _windowSize;
protected final List<ExpressionContext> _stepExpressions;
@@ -51,7 +54,8 @@ public abstract class FunnelBaseAggregationFunction<F extends
Comparable>
protected long _maxStepDuration = 0L;
protected final Map<String, String> _extraArguments = new HashMap<>();
- public FunnelBaseAggregationFunction(List<ExpressionContext> arguments) {
+ public FunnelBaseAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ _nullHandlingEnabled = nullHandlingEnabled;
int numArguments = arguments.size();
Preconditions.checkArgument(numArguments > 3,
"FUNNEL_AGG_FUNC expects >= 4 arguments, got: %s. The function can be
used as "
@@ -121,91 +125,121 @@ public abstract class FunnelBaseAggregationFunction<F
extends Comparable>
return new ObjectGroupByResultHolder(initialCapacity, maxCapacity);
}
+ /// Runs the consumer over each range of rows whose timestamp is not null,
or over the whole block when the
+ /// option is disabled.
+ ///
+ /// Only the timestamp is consulted. A step expression is a predicate, and a
predicate over a null operand is
+ /// UNKNOWN, which SQL treats as not satisfied wherever a boolean is
consumed, so a null step already means that
+ /// step did not match and the row still belongs to the funnel. A null
timestamp is different: the event has no
+ /// position in the window, and an aggregate ignores a row whose input is
null.
+ private void forEachNotNullTimestamp(int length, BlockValSet
timestampBlockValSet,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ RoaringBitmap nullBitmap = _nullHandlingEnabled ?
timestampBlockValSet.getNullBitmap() : null;
+ if (nullBitmap == null) {
+ consumer.consume(0, length);
+ return;
+ }
+ // Skip if the entire block is null
+ if (!nullBitmap.contains(0, length)) {
+ RoaringBitmapUtils.forEachUnset(length, nullBitmap.getIntIterator(),
consumer);
+ }
+ }
+
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
- PriorityQueue<FunnelStepEvent> stepEvents =
aggregationResultHolder.getResult();
- if (stepEvents == null) {
- stepEvents = new PriorityQueue<>();
- aggregationResultHolder.setValue(stepEvents);
+ PriorityQueue<FunnelStepEvent> existing =
aggregationResultHolder.getResult();
+ if (existing == null) {
+ existing = new PriorityQueue<>();
+ aggregationResultHolder.setValue(existing);
}
- for (int i = 0; i < length; i++) {
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
- stepFound = true;
- break;
+ PriorityQueue<FunnelStepEvent> stepEvents = existing;
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
}
}
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
- }
- }
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
- for (int i = 0; i < length; i++) {
- int groupKey = groupKeyArray[i];
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int groupKey = groupKeyArray[i];
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
- stepFound = true;
- break;
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
}
}
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
- }
- }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
- for (int i = 0; i < length; i++) {
- int[] groupKeys = groupKeysArray[i];
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int[] groupKeys = groupKeysArray[i];
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ for (int groupKey : groupKeys) {
+ PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
+ }
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
for (int groupKey : groupKeys) {
PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], j));
+ stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
}
- stepFound = true;
- break;
- }
- }
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- for (int groupKey : groupKeys) {
- PriorityQueue<FunnelStepEvent> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- stepEvents.add(new FunnelStepEvent(timestampBlock[i], -1));
}
}
- }
+ });
}
private static PriorityQueue<FunnelStepEvent>
getFunnelStepEvents(GroupByResultHolder groupByResultHolder,
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelCompleteCountAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelCompleteCountAggregationFunction.java
index f76e678cb14..02a0392d850 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelCompleteCountAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelCompleteCountAggregationFunction.java
@@ -31,8 +31,8 @@ import org.apache.pinot.spi.query.QueryThreadContext;
public class FunnelCompleteCountAggregationFunction extends
FunnelBaseAggregationFunction<Integer> {
- public FunnelCompleteCountAggregationFunction(List<ExpressionContext>
arguments) {
- super(arguments);
+ public FunnelCompleteCountAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
+ super(arguments, nullHandlingEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelEventsFunctionEvalAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelEventsFunctionEvalAggregationFunction.java
index f319a29b712..408d4fd75cb 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelEventsFunctionEvalAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelEventsFunctionEvalAggregationFunction.java
@@ -31,6 +31,7 @@ import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.RoaringBitmapUtils;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder;
@@ -41,12 +42,14 @@ import
org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.query.QueryThreadContext;
+import org.roaringbitmap.RoaringBitmap;
public class FunnelEventsFunctionEvalAggregationFunction
implements
AggregationFunction<PriorityQueue<FunnelStepEventWithExtraFields>,
ObjectArrayList<String>> {
private final static int INTERMEDIATE_RESULT_SERDE_VERSION = 0;
+ protected final boolean _nullHandlingEnabled;
protected final ExpressionContext _timestampExpression;
protected final long _windowSize;
protected final List<ExpressionContext> _stepExpressions;
@@ -56,7 +59,9 @@ public class FunnelEventsFunctionEvalAggregationFunction
protected final List<ExpressionContext> _extraExpressions;
protected long _maxStepDuration = 0L;
- public FunnelEventsFunctionEvalAggregationFunction(List<ExpressionContext>
arguments) {
+ public FunnelEventsFunctionEvalAggregationFunction(List<ExpressionContext>
arguments,
+ boolean nullHandlingEnabled) {
+ _nullHandlingEnabled = nullHandlingEnabled;
int numArguments = arguments.size();
Preconditions.checkArgument(numArguments > 3,
"FUNNEL_EVENTS_FUNCTION_EVAL expects >= 4 arguments, got: %s. The
function can be used as "
@@ -140,36 +145,67 @@ public class FunnelEventsFunctionEvalAggregationFunction
return new ObjectGroupByResultHolder(initialCapacity, maxCapacity);
}
+ /// Runs the consumer over each range of rows whose timestamp is not null,
or over the whole block when the
+ /// option is disabled.
+ ///
+ /// Only the timestamp is consulted, of the three kinds of column this
function reads.
+ ///
+ /// A step expression is a predicate, and a predicate over a null operand is
UNKNOWN, which SQL treats as not
+ /// satisfied wherever a boolean is consumed, so a null step already means
that step did not match and the row
+ /// still belongs to the funnel. A null timestamp is different: the event
has no position in the window, and an
+ /// aggregate ignores a row whose input is null.
+ ///
+ /// An extra field is neither. It is payload carried alongside a matched
event, so a null one does not make the
+ /// event invalid and dropping the row would lose an event that really
happened. It is therefore not gated on,
+ /// with one known limitation: the value is read positionally and a null row
yields the column default, so a null
+ /// extra field renders as `0` or the empty string rather than as NULL.
+ private void forEachNotNullTimestamp(int length, BlockValSet
timestampBlockValSet,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ RoaringBitmap nullBitmap = _nullHandlingEnabled ?
timestampBlockValSet.getNullBitmap() : null;
+ if (nullBitmap == null) {
+ consumer.consume(0, length);
+ return;
+ }
+ // Skip if the entire block is null
+ if (!nullBitmap.contains(0, length)) {
+ RoaringBitmapUtils.forEachUnset(length, nullBitmap.getIntIterator(),
consumer);
+ }
+ }
+
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
- PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
aggregationResultHolder.getResult();
- if (stepEvents == null) {
- stepEvents = new PriorityQueue<>();
- aggregationResultHolder.setValue(stepEvents);
+ PriorityQueue<FunnelStepEventWithExtraFields> existing =
aggregationResultHolder.getResult();
+ if (existing == null) {
+ existing = new PriorityQueue<>();
+ aggregationResultHolder.setValue(existing);
}
+ PriorityQueue<FunnelStepEventWithExtraFields> stepEvents = existing;
List<Object> extraFieldsBlocks = getExtraFieldsBlocks(blockValSetMap);
- for (int i = 0; i < length; i++) {
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ List<Object> extraFields = extractExtraFields(extraFieldsBlocks,
i);
+ stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
List<Object> extraFields = extractExtraFields(extraFieldsBlocks, i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
- stepFound = true;
- break;
+ stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
}
}
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- List<Object> extraFields = extractExtraFields(extraFieldsBlocks, i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
- }
- }
+ });
}
private List<Object> getExtraFieldsBlocks(Map<ExpressionContext,
BlockValSet> blockValSetMap) {
@@ -232,66 +268,75 @@ public class FunnelEventsFunctionEvalAggregationFunction
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
List<Object> extraFieldsBlocks = getExtraFieldsBlocks(blockValSetMap);
- for (int i = 0; i < length; i++) {
- int groupKey = groupKeyArray[i];
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int groupKey = groupKeyArray[i];
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
+ getFunnelStepEvents(groupByResultHolder, groupKey);
+ List<Object> extraFields = extractExtraFields(extraFieldsBlocks,
i);
+ stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
List<Object> extraFields = extractExtraFields(extraFieldsBlocks, i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
- stepFound = true;
- break;
+ stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
}
}
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- List<Object> extraFields = extractExtraFields(extraFieldsBlocks, i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
- }
- }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[] timestampBlock =
blockValSetMap.get(_timestampExpression).getLongValuesSV();
+ BlockValSet timestampBlockValSet =
blockValSetMap.get(_timestampExpression);
+ long[] timestampBlock = timestampBlockValSet.getLongValuesSV();
List<int[]> stepBlocks = new ArrayList<>(_numSteps);
for (ExpressionContext stepExpression : _stepExpressions) {
stepBlocks.add(blockValSetMap.get(stepExpression).getIntValuesSV());
}
List<Object> extraFieldsBlocks = getExtraFieldsBlocks(blockValSetMap);
- for (int i = 0; i < length; i++) {
- int[] groupKeys = groupKeysArray[i];
- boolean stepFound = false;
- for (int j = 0; j < _numSteps; j++) {
- if (stepBlocks.get(j)[i] == 1) {
+ forEachNotNullTimestamp(length, timestampBlockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int[] groupKeys = groupKeysArray[i];
+ boolean stepFound = false;
+ for (int j = 0; j < _numSteps; j++) {
+ if (stepBlocks.get(j)[i] == 1) {
+ for (int groupKey : groupKeys) {
+ PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
+ getFunnelStepEvents(groupByResultHolder, groupKey);
+ List<Object> extraFields = extractExtraFields(extraFieldsBlocks,
i);
+ stepEvents.add(
+ new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
+ }
+ stepFound = true;
+ break;
+ }
+ }
+ // If the mode is KEEP_ALL and no step is found, add a dummy step
event with step -1
+ if (_modes.hasKeepAll() && !stepFound) {
for (int groupKey : groupKeys) {
PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
List<Object> extraFields = extractExtraFields(extraFieldsBlocks,
i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], j), extraFields));
+ stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
}
- stepFound = true;
- break;
- }
- }
- // If the mode is KEEP_ALL and no step is found, add a dummy step event
with step -1
- if (_modes.hasKeepAll() && !stepFound) {
- for (int groupKey : groupKeys) {
- PriorityQueue<FunnelStepEventWithExtraFields> stepEvents =
getFunnelStepEvents(groupByResultHolder, groupKey);
- List<Object> extraFields = extractExtraFields(extraFieldsBlocks, i);
- stepEvents.add(new FunnelStepEventWithExtraFields(new
FunnelStepEvent(timestampBlock[i], -1), extraFields));
}
}
- }
+ });
}
private static PriorityQueue<FunnelStepEventWithExtraFields>
getFunnelStepEvents(
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMatchStepAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMatchStepAggregationFunction.java
index 9710f32ecd7..0c08dee146d 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMatchStepAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMatchStepAggregationFunction.java
@@ -32,8 +32,8 @@ import org.apache.pinot.spi.query.QueryThreadContext;
public class FunnelMatchStepAggregationFunction extends
FunnelBaseAggregationFunction<IntArrayList> {
- public FunnelMatchStepAggregationFunction(List<ExpressionContext> arguments)
{
- super(arguments);
+ public FunnelMatchStepAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ super(arguments, nullHandlingEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java
index 23ae788daa9..7cac01529c9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java
@@ -31,8 +31,8 @@ import org.apache.pinot.spi.query.QueryThreadContext;
public class FunnelMaxStepAggregationFunction extends
FunnelBaseAggregationFunction<Integer> {
- public FunnelMaxStepAggregationFunction(List<ExpressionContext> arguments) {
- super(arguments);
+ public FunnelMaxStepAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ super(arguments, nullHandlingEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
index d6ce01f983b..b162aabba43 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
@@ -49,8 +49,8 @@ public class FunnelStepDurationStatsAggregationFunction
extends FunnelBaseAggreg
private final List<String> _durationFunctions = new ArrayList<>();
private boolean _canSkipNonMatchedFunnel = true;
- public FunnelStepDurationStatsAggregationFunction(List<ExpressionContext>
arguments) {
- super(arguments);
+ public FunnelStepDurationStatsAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
+ super(arguments, nullHandlingEnabled);
if (_extraArguments.get("DURATIONFUNCTIONS") != null) {
String[] durationFunctions =
_extraArguments.get("DURATIONFUNCTIONS").split(",");
for (String durationFunction : durationFunctions) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
b/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
index ddb6dc6c314..dfb4824aea9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
@@ -32,6 +32,22 @@ import org.roaringbitmap.RoaringBitmap;
/// Synthetic [BlockValSet] for testing and benchmarking.
+///
+/// There is one fixture per stored type, single- and multi-value:
`Int`/`IntMV`, `Long`/`LongMV`, `Float`/`FloatMV`,
+/// `Double`/`DoubleMV`, `BigDec`/`BigDecMV`, `Str`/`StrMV` and
`Bytes`/`BytesMV`, plus `DictIds`/`DictIdsMV` for
+/// dictionary-encoded columns. Each takes an optional null bitmap and the raw
values.
+///
+/// Two of them are named for a JDK type they must not shadow. A nested class
called `String` or `BigDecimal` hides
+/// [String] or [BigDecimal] across the whole of this class, which silently
changes the signature of every `getString*`
+/// or `getBigDecimal*` method declared here so that it no longer implements
[BlockValSet]. Hence `Str` and `BigDec`;
+/// keep any future fixture clear of the same collision.
+///
+/// **The null bitmap is independent of the values.** A row marked null still
holds whatever the values array puts
+/// at that index, and these fixtures have no notion of a column's
`defaultNullValue`. That matches what a caller
+/// with null handling enabled sees, since it reads the bitmap and skips those
rows. It does **not** match a real
+/// segment with null handling disabled, where a null row reads as the column
default rather than as a neighbouring
+/// value. So a test that wants "nothing was aggregated" in the disabled mode
cannot get there with an all-null
+/// bitmap - the values are still aggregated - and needs a holder that was
never touched instead.
public class SyntheticBlockValSets {
private SyntheticBlockValSets() {
}
@@ -142,6 +158,55 @@ public class SyntheticBlockValSets {
/// A [BlockValSet] for a dictionary-encoded multi-value column, which
exposes dictionary ids rather than values.
///
/// Functions that collect dictionary ids take a different path from the one
that reads values, and resolve the ids
+ /// A simple [BlockValSet] for nullable, dictionary-encoded single-value
columns.
+ public static class DictIds extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final int[] _dictIds;
+ final Dictionary _dictionary;
+ final DataType _valueType;
+
+ private DictIds(@Nullable RoaringBitmap nullBitmap, int[] dictIds,
Dictionary dictionary, DataType valueType) {
+ _nullBitmap = nullBitmap;
+ _dictIds = dictIds;
+ _dictionary = dictionary;
+ _valueType = valueType;
+ }
+
+ public static DictIds create(@Nullable RoaringBitmap nullBitmap, int[]
dictIds, Dictionary dictionary,
+ DataType valueType) {
+ return new DictIds(nullBitmap, dictIds, dictionary, valueType);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return _valueType;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return true;
+ }
+
+ @Nullable
+ @Override
+ public Dictionary getDictionary() {
+ return _dictionary;
+ }
+
+ @Override
+ public int[] getDictionaryIdsSV() {
+ return _dictIds;
+ }
+ }
+
/// against the dictionary only when the result is extracted.
public static class DictIdsMV extends Base {
@@ -542,9 +607,7 @@ public class SyntheticBlockValSets {
/// A simple [BlockValSet] for nullable, not dictionary-encoded BigDecimal
values.
///
- /// Named `BigDec` rather than `BigDecimal` for the same reason [Str] is not
named `String`: a nested class of that
- /// name shadows [java.math.BigDecimal] across the whole enclosing class,
which silently changes the signature of
- /// every `getBigDecimal*` method declared here so that it no longer
implements [BlockValSet].
+ /// Named `BigDec` rather than `BigDecimal` to avoid shadowing [BigDecimal];
see the class comment.
public static class BigDec extends Base {
@Nullable
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
index 4e282e463a6..1a8df4c13da 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
@@ -18,6 +18,10 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.objects.ObjectArrayList;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -191,6 +195,27 @@ public class AggregationFunctionNullContractTest {
assertEquals(create("DISTINCTCOUNTBITMAP", "(column)",
true).extractFinalResult(null), 0);
}
+ /// A funnel over no events completed no steps, so every one of them answers
zero rather than `NULL`.
+ ///
+ /// They belong with the counting functions above rather than the value
functions below: the answer is a count, or
+ /// a per-step vector of counts, and zero is the meaningful value for it.
Pinned because the whole family renders
+ /// this from an initial accumulator rather than from a branch in
`extractFinalResult`, so it would move quietly.
+ @Test
+ public void testFunnelFunctionsReturnZeroWhenNothingAggregated() {
+ String steps = "(column, '1000', 2, column2 = 'a', column2 = 'b')";
+ assertEquals(create("FUNNELMAXSTEP", steps,
true).extractFinalResult(null), 0);
+ assertEquals(create("FUNNELCOMPLETECOUNT", steps,
true).extractFinalResult(null), 0);
+ assertEquals(create("FUNNELSTEPDURATIONSTATS",
+ "(column, '1000', 2, column2 = 'a', column2 = 'b',
'durationFunctions=count')", true)
+ .extractFinalResult(null), new DoubleArrayList());
+ assertEquals(create("FUNNELMATCHSTEP", steps,
true).extractFinalResult(null), new IntArrayList(new int[]{0, 0}));
+ assertEquals(create("FUNNELEVENTSFUNCTIONEVAL",
+ "(column, '1000', 2, column2 = 'a', column2 = 'b', 2, column,
column2)", true)
+ .extractFinalResult(null), new ObjectArrayList<String>());
+ assertEquals(create("FUNNELCOUNT", "(STEPS(column2 = 'a', column2 = 'b'),
CORRELATE_BY(column))", true)
+ .extractFinalResult(null), new LongArrayList(new long[]{0L, 0L}));
+ }
+
/// The functions that return SQL `NULL` when nothing was aggregated.
///
/// Both representations of that state are checked. A `null` intermediate
result is the obvious one, but several of
@@ -206,8 +231,8 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.PERCENTILEKLL,
AggregationFunctionType.PERCENTILESMARTTDIGEST,
// These carry a legacy sentinel with the option disabled - NaN, the
empty point, an all-zero histogram, an
// empty id set - so only the enabled answer is NULL, and only the
enabled answer is checked here
- AggregationFunctionType.SKEWNESS, AggregationFunctionType.KURTOSIS,
AggregationFunctionType.STUNION,
- AggregationFunctionType.HISTOGRAM, AggregationFunctionType.IDSET,
+ AggregationFunctionType.IDSET, AggregationFunctionType.HISTOGRAM,
AggregationFunctionType.SKEWNESS,
+ AggregationFunctionType.KURTOSIS, AggregationFunctionType.STUNION,
// These two answer NULL in both modes, which is why neither needs a
mode-aware branch
AggregationFunctionType.SUMARRAYLONG,
AggregationFunctionType.SUMARRAYDOUBLE}) {
// Built through the shared argument shapes: the percentile families
disagree on whether the percentile is a
@@ -229,9 +254,13 @@ public class AggregationFunctionNullContractTest {
/// which functions the option actually reaches.
///
/// The raw variants reach the option through the function they delegate to,
so threading it into one of those
- /// changes the raw variant alongside it. Absence is as meaningful as
presence: a function missing from here either
- /// never receives the option, which is the first known deviation on
[AggregationFunction], or does not skip null
- /// rows.
+ /// changes the raw variant alongside it.
+ ///
+ /// Absence no longer means a function was left out of the contract. Every
user-facing aggregation now receives the
+ /// option, so a function missing from here is one this harness cannot drive
— see
+ /// [#NOT_EXERCISABLE_BY_SYNTHETIC_BLOCK], which lists where each of those
is covered instead — or one that
+ /// genuinely answers the same in both modes, as the counting functions do
when the empty answer is `0` either
+ /// way.
///
/// Two bounds on how much this set proves. It only covers what the harness
can drive — see
/// [#NOT_EXERCISABLE_BY_SYNTHETIC_BLOCK] — and it compares the **rendered**
answers, because several functions
@@ -278,22 +307,31 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.DISTINCTCOUNTRAWTHETASKETCH,
AggregationFunctionType.DISTINCTCOUNTCPCSKETCH,
AggregationFunctionType.DISTINCTCOUNTRAWCPCSKETCH,
AggregationFunctionType.DISTINCTCOUNTBITMAPMV,
AggregationFunctionType.DISTINCTCOUNTHLLMV,
AggregationFunctionType.DISTINCTCOUNTRAWHLLMV,
- AggregationFunctionType.DISTINCTCOUNTHLLPLUSMV,
AggregationFunctionType.DISTINCTCOUNTRAWHLLPLUSMV
+ AggregationFunctionType.DISTINCTCOUNTHLLPLUSMV,
AggregationFunctionType.DISTINCTCOUNTRAWHLLPLUSMV,
+ // Reached once the multi-value shapes were added below; the array sums
take only an array column
+ AggregationFunctionType.SUMARRAYLONG,
AggregationFunctionType.SUMARRAYDOUBLE
);
/// Functions this test cannot drive with a one-column synthetic block,
pinned so that a silent drop-out is always a
/// reviewed decision. Derived from a run rather than predicted.
///
- /// The block implements only single-value `long` and `double`, so a
function lands here when it reads another value
- /// type, needs a multi-value block, or takes more than the one input column
the shared argument shapes supply. The
- /// exclusion is scoped to this test: the rest of the contract is still
checked against these functions, since the
- /// other cases construct them and call
[AggregationFunction#extractFinalResult] without aggregating first.
+ /// [#BLOCK_SHAPES] supplies one column at a time — `double`, `long`, `int`,
`String` or `byte[]`, single-value,
+ /// plus multi-value `long`, `double` and `int` — and gives every input
expression the same shape. A function
+ /// lands here when it needs a value type outside that list, a payload the
shape cannot fabricate (a serialized
+ /// sketch or geometry rather than an empty `byte[]`), a dictionary, or two
input columns of different types. That
+ /// last one is what rules out the funnels: their timestamp is a `long` and
their steps are `int` predicates, and
+ /// no single shape is both.
+ ///
+ /// The exclusion is scoped to this one check. The rest of the contract is
still enforced against these functions
+ /// here, since the other cases construct them and call
[AggregationFunction#extractFinalResult] without
+ /// aggregating first, and their null-row skipping is covered by a test
built for the shape each one needs:
+ /// `DistinctCountSketchNullHandlingTest`, `FrequentSketchNullHandlingTest`,
`ValueAggregationNullHandlingTest`
+ /// and `FunnelNullHandlingTest`. So membership here means "checked
elsewhere", not "unchecked".
private static final Set<AggregationFunctionType>
NOT_EXERCISABLE_BY_SYNTHETIC_BLOCK = Set.of(
AggregationFunctionType.FASTHLL,
AggregationFunctionType.DISTINCTCOUNTTUPLESKETCH,
AggregationFunctionType.DISTINCTCOUNTRAWINTEGERSUMTUPLESKETCH,
AggregationFunctionType.SUMVALUESINTEGERSUMTUPLESKETCH,
AggregationFunctionType.AVGVALUEINTEGERSUMTUPLESKETCH,
AggregationFunctionType.STUNION, AggregationFunctionType.BOOLAND,
AggregationFunctionType.BOOLOR,
- AggregationFunctionType.SUMARRAYLONG,
AggregationFunctionType.SUMARRAYDOUBLE,
AggregationFunctionType.FUNNELMAXSTEP,
AggregationFunctionType.FUNNELCOMPLETECOUNT,
AggregationFunctionType.FUNNELSTEPDURATIONSTATS,
AggregationFunctionType.FUNNELMATCHSTEP,
AggregationFunctionType.FUNNELEVENTSFUNCTIONEVAL,
AggregationFunctionType.FUNNELCOUNT
@@ -391,12 +429,39 @@ public class AggregationFunctionNullContractTest {
///
/// Probed rather than pinned per function: which value type an accumulator
reads is an implementation detail that
/// changes, and a hard-coded mapping silently drops a function out of the
census when it feeds the wrong width.
+ private static int[][] mvInts() {
+ int[][] rows = new int[NUM_DOCS][];
+ for (int i = 0; i < NUM_DOCS; i++) {
+ rows[i] = new int[]{0};
+ }
+ return rows;
+ }
+
+ private static long[][] mvLongs() {
+ long[][] rows = new long[NUM_DOCS][];
+ for (int i = 0; i < NUM_DOCS; i++) {
+ rows[i] = new long[]{0L};
+ }
+ return rows;
+ }
+
+ private static double[][] mvDoubles() {
+ double[][] rows = new double[NUM_DOCS][];
+ for (int i = 0; i < NUM_DOCS; i++) {
+ rows[i] = new double[]{0.0};
+ }
+ return rows;
+ }
+
private static final List<Supplier<BlockValSet>> BLOCK_SHAPES = List.of(
- () -> SyntheticBlockValSets.Double.create(NUM_DOCS, allNullBitmap(), ()
-> 0.0),
- () -> SyntheticBlockValSets.Long.create(NUM_DOCS, allNullBitmap(), () ->
0L),
() -> SyntheticBlockValSets.Int.create(NUM_DOCS, allNullBitmap(), () ->
0),
+ () -> SyntheticBlockValSets.Long.create(NUM_DOCS, allNullBitmap(), () ->
0L),
+ () -> SyntheticBlockValSets.Double.create(NUM_DOCS, allNullBitmap(), ()
-> 0.0),
() -> SyntheticBlockValSets.Str.create(NUM_DOCS, allNullBitmap(), () ->
""),
- () -> SyntheticBlockValSets.Bytes.create(NUM_DOCS, allNullBitmap(), ()
-> new byte[0])
+ () -> SyntheticBlockValSets.Bytes.create(NUM_DOCS, allNullBitmap(), ()
-> new byte[0]),
+ () -> SyntheticBlockValSets.IntMV.create(allNullBitmap(), mvInts()),
+ () -> SyntheticBlockValSets.LongMV.create(allNullBitmap(), mvLongs()),
+ () -> SyntheticBlockValSets.DoubleMV.create(allNullBitmap(), mvDoubles())
);
private static RoaringBitmap allNullBitmap() {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelNullHandlingTest.java
new file mode 100644
index 00000000000..fe1cb727cae
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelNullHandlingTest.java
@@ -0,0 +1,233 @@
+/**
+ * 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.query.aggregation.function.funnel;
+
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.FunctionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.funnel.window.FunnelMaxStepAggregationFunction;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+
+
+/// Null handling for the window funnel functions.
+///
+/// A funnel reads more than one column per row, so it never received the
option through
+///
[org.apache.pinot.core.query.aggregation.function.NullableSingleInputAggregationFunction]
and is invisible to
+///
[org.apache.pinot.core.query.aggregation.function.AggregationFunctionNullContractTest],
whose synthetic block
+/// supplies a single column.
+///
+/// Only the timestamp is consulted. A step expression is a predicate, and a
predicate over a null operand is
+/// UNKNOWN, which SQL treats as not satisfied wherever a boolean is consumed,
so a null step already means that
+/// step did not match. A null timestamp is different: the event has no
position in the window, and today it is read
+/// as the column default, which places a real step event at a fabricated time.
+public class FunnelNullHandlingTest {
+ private static final ExpressionContext TIMESTAMP =
ExpressionContext.forIdentifier("ts");
+ private static final ExpressionContext STEP_0 =
ExpressionContext.forIdentifier("step0");
+ private static final ExpressionContext STEP_1 =
ExpressionContext.forIdentifier("step1");
+
+ /// `funnelMaxStep(ts, '1000', 2, step0, step1)`, optionally with extra
arguments such as `MODE=KEEP_ALL`.
+ private static FunnelMaxStepAggregationFunction maxStep(boolean
nullHandlingEnabled, String... extraArguments) {
+ List<ExpressionContext> arguments = new ArrayList<>(List.of(TIMESTAMP,
+ ExpressionContext.forLiteral(Literal.longValue(1000)),
+ ExpressionContext.forLiteral(Literal.intValue(2)), STEP_0, STEP_1));
+ for (String extraArgument : extraArguments) {
+
arguments.add(ExpressionContext.forLiteral(Literal.stringValue(extraArgument)));
+ }
+ return new FunnelMaxStepAggregationFunction(arguments,
nullHandlingEnabled);
+ }
+
+ /// Timestamps carry the null bitmap; the two step columns are the
already-evaluated predicate results.
+ private static Map<ExpressionContext, BlockValSet> block(RoaringBitmap
timestampNulls, long[] timestamps,
+ int[] step0, int[] step1) {
+ return Map.of(
+ TIMESTAMP, SyntheticBlockValSets.Long.create(timestampNulls,
timestamps),
+ STEP_0, SyntheticBlockValSets.Int.create(null, step0),
+ STEP_1, SyntheticBlockValSets.Int.create(null, step1));
+ }
+
+ /// A row whose timestamp is null contributes no step event, so the funnel
that row would have completed does not
+ /// complete.
+ @Test
+ public void testNullTimestampRowContributesNoEvent() {
+ long[] timestamps = {100L, 200L};
+ int[] step0 = {1, 0};
+ int[] step1 = {0, 1};
+
+ FunnelMaxStepAggregationFunction enabled = maxStep(true);
+ AggregationResultHolder holder = enabled.createAggregationResultHolder();
+ enabled.aggregate(2, holder, block(RoaringBitmap.bitmapOf(1), timestamps,
step0, step1));
+
assertEquals(enabled.extractFinalResult(enabled.extractAggregationResult(holder)),
Integer.valueOf(1));
+
+ // With the option disabled the null bitmap is ignored and row 1 still
completes the funnel, which is the answer
+ // this mode has always given
+ FunnelMaxStepAggregationFunction disabled = maxStep(false);
+ AggregationResultHolder disabledHolder =
disabled.createAggregationResultHolder();
+ disabled.aggregate(2, disabledHolder, block(RoaringBitmap.bitmapOf(1),
timestamps, step0, step1));
+
assertEquals(disabled.extractFinalResult(disabled.extractAggregationResult(disabledHolder)),
Integer.valueOf(2));
+ }
+
+ /// Every row null leaves the queue empty rather than filling it with events
at fabricated timestamps.
+ @Test
+ public void testEveryRowNullYieldsNoEvents() {
+ long[] timestamps = {100L, 200L};
+ RoaringBitmap allNull = new RoaringBitmap();
+ allNull.add(0L, 2L);
+
+ FunnelMaxStepAggregationFunction function = maxStep(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(2, holder, block(allNull, timestamps, new int[]{1, 0},
new int[]{0, 1}));
+
+ PriorityQueue<FunnelStepEvent> events =
function.extractAggregationResult(holder);
+ assertNotNull(events, "the holder is created before the range, so it stays
non-null");
+ assertEquals(events.size(), 0);
+ assertEquals(function.extractFinalResult(events), Integer.valueOf(0));
+ }
+
+ /// In KEEP_ALL mode a row that matches no step still produces a dummy event
- but only if it has a timestamp to
+ /// place it at.
+ @Test
+ public void testKeepAllDoesNotFabricateAnEventForANullTimestamp() {
+ long[] timestamps = {100L, 200L};
+ int[] noMatch = {0, 0};
+
+ FunnelMaxStepAggregationFunction enabled = maxStep(true, "MODE=KEEP_ALL");
+ AggregationResultHolder holder = enabled.createAggregationResultHolder();
+ enabled.aggregate(2, holder, block(RoaringBitmap.bitmapOf(1), timestamps,
new int[]{1, 0}, noMatch));
+ assertEquals(enabled.extractAggregationResult(holder).size(), 1);
+
+ FunnelMaxStepAggregationFunction disabled = maxStep(false,
"MODE=KEEP_ALL");
+ AggregationResultHolder disabledHolder =
disabled.createAggregationResultHolder();
+ disabled.aggregate(2, disabledHolder, block(RoaringBitmap.bitmapOf(1),
timestamps, new int[]{1, 0}, noMatch));
+ assertEquals(disabled.extractAggregationResult(disabledHolder).size(), 2);
+ }
+
+ /// The group-by path skips the null row for its own group only.
+ @Test
+ public void testGroupBySVSkipsTheNullRow() {
+ long[] timestamps = {100L, 200L};
+
+ FunnelMaxStepAggregationFunction function = maxStep(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(2, new int[]{0, 1}, holder,
+ block(RoaringBitmap.bitmapOf(1), timestamps, new int[]{1, 1}, new
int[]{0, 0}));
+
+ PriorityQueue<FunnelStepEvent> group0 =
function.extractGroupByResult(holder, 0);
+ assertNotNull(group0);
+ assertEquals(group0.size(), 1);
+ assertNull(function.extractGroupByResult(holder, 1));
+ }
+
+ /// A null row is skipped for every group key it would have fed.
+ @Test
+ public void testGroupByMVSkipsTheNullRowForAllKeys() {
+ long[] timestamps = {100L, 200L};
+
+ FunnelMaxStepAggregationFunction function = maxStep(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupByMV(2, new int[][]{{0}, {0, 1}}, holder,
+ block(RoaringBitmap.bitmapOf(1), timestamps, new int[]{1, 1}, new
int[]{0, 0}));
+
+ PriorityQueue<FunnelStepEvent> group0 =
function.extractGroupByResult(holder, 0);
+ assertNotNull(group0);
+ assertEquals(group0.size(), 1);
+ assertNull(function.extractGroupByResult(holder, 1));
+ }
+
+ // ---------- FUNNELCOUNT ----------
+
+ private static final ExpressionContext USER =
ExpressionContext.forIdentifier("userId");
+
+ /// `funnelCount(steps(step0, step1), correlateby(userId))`, built through
the same factory
+ /// AggregationFunctionFactory uses.
+ private static AggregationFunction<?, LongArrayList> funnelCount(boolean
nullHandlingEnabled) {
+ List<ExpressionContext> arguments = List.of(
+ ExpressionContext.forFunction(
+ new FunctionContext(FunctionContext.Type.TRANSFORM, "steps",
List.of(STEP_0, STEP_1))),
+ ExpressionContext.forFunction(
+ new FunctionContext(FunctionContext.Type.TRANSFORM, "correlateby",
List.of(USER))));
+ return new FunnelCountAggregationFunctionFactory(arguments,
nullHandlingEnabled).get();
+ }
+
+ /// Two users, ids 0 and 1, so a dictionary id doubles as the user it stands
for.
+ private static Dictionary userDictionary() {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.length()).thenReturn(2);
+ when(dictionary.getValueType()).thenReturn(DataType.INT);
+ when(dictionary.getIntValue(anyInt())).thenAnswer(invocation ->
invocation.getArgument(0));
+ return dictionary;
+ }
+
+ private static Map<ExpressionContext, BlockValSet>
funnelCountBlock(RoaringBitmap userNulls, int[] userDictIds,
+ int[] step0, int[] step1) {
+ return Map.of(
+ USER, SyntheticBlockValSets.DictIds.create(userNulls, userDictIds,
userDictionary(), DataType.INT),
+ STEP_0, SyntheticBlockValSets.Int.create(null, step0),
+ STEP_1, SyntheticBlockValSets.Int.create(null, step1));
+ }
+
+ /// A row whose correlation key is null belongs to no user, so it must not
be counted as one.
+ ///
+ /// Its dictionary id is the default's, which is a real user's id, so
without the null check the funnel credits
+ /// that user with a step they never took.
+ @Test
+ public void testFunnelCountSkipsRowsWithANullCorrelationKey() {
+ // Row 0 is user 0 reaching step 0. Row 1 is null, and its dictionary id
happens to be user 1's.
+ int[] userDictIds = {0, 1};
+ int[] step0 = {1, 1};
+ int[] step1 = {0, 0};
+
+ AggregationFunction<?, LongArrayList> enabled = funnelCount(true);
+ AggregationResultHolder holder = enabled.createAggregationResultHolder();
+ enabled.aggregate(2, holder, funnelCountBlock(RoaringBitmap.bitmapOf(1),
userDictIds, step0, step1));
+ assertEquals(extractCounts(enabled, holder), List.of(1L, 0L));
+
+ AggregationFunction<?, LongArrayList> disabled = funnelCount(false);
+ AggregationResultHolder disabledHolder =
disabled.createAggregationResultHolder();
+ disabled.aggregate(2, disabledHolder,
funnelCountBlock(RoaringBitmap.bitmapOf(1), userDictIds, step0, step1));
+ assertEquals(extractCounts(disabled, disabledHolder), List.of(2L, 0L));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<Long> extractCounts(AggregationFunction<?,
LongArrayList> function,
+ AggregationResultHolder holder) {
+ AggregationFunction<Object, LongArrayList> typed =
(AggregationFunction<Object, LongArrayList>) function;
+ return typed.extractFinalResult(typed.extractAggregationResult(holder));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
