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 e57f94bce52 Honour the null handling option in EXPR_MIN/EXPR_MAX and
TIMESERIESAGGREGATE (#19357)
e57f94bce52 is described below
commit e57f94bce52d252410bc94a6fcca0509ba187b6b
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Tue Aug 25 14:04:03 2026 -0700
Honour the null handling option in EXPR_MIN/EXPR_MAX and
TIMESERIESAGGREGATE (#19357)
---
.../function/AggregationFunctionFactory.java | 6 +-
.../ParentExprMinMaxAggregationFunction.java | 121 ++++++++++---
.../function/TimeSeriesAggregationFunction.java | 139 ++++++++------
.../AggregationFunctionNullContractTest.java | 9 +-
.../function/ExprMinMaxNullHandlingTest.java | 201 +++++++++++++++++++++
.../function/StoredTypeDispatchTest.java | 4 +-
.../TimeSeriesAggregationNullHandlingTest.java | 143 +++++++++++++++
7 files changed, 534 insertions(+), 89 deletions(-)
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 4c4165edbdb..fdc93df43e8 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
@@ -499,9 +499,9 @@ public class AggregationFunctionFactory {
return new
AvgValueIntegerTupleSketchAggregationFunction(arguments,
IntegerSummary.Mode.Sum,
nullHandlingEnabled);
case PINOTPARENTAGGEXPRMAX:
- return new ParentExprMinMaxAggregationFunction(arguments, true);
+ return new ParentExprMinMaxAggregationFunction(arguments, true,
nullHandlingEnabled);
case PINOTPARENTAGGEXPRMIN:
- return new ParentExprMinMaxAggregationFunction(arguments, false);
+ return new ParentExprMinMaxAggregationFunction(arguments, false,
nullHandlingEnabled);
case PINOTCHILDAGGEXPRMAX:
return new ChildExprMinMaxAggregationFunction(arguments, true);
case PINOTCHILDAGGEXPRMIN:
@@ -535,7 +535,7 @@ public class AggregationFunctionFactory {
case DISTINCTCOUNTRAWULL:
return new DistinctCountRawULLAggregationFunction(arguments,
nullHandlingEnabled);
case TIMESERIESAGGREGATE:
- return new TimeSeriesAggregationFunction(arguments);
+ return new TimeSeriesAggregationFunction(arguments,
nullHandlingEnabled);
default:
throw new IllegalArgumentException("Unsupported aggregation
function type: " + functionType);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ParentExprMinMaxAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ParentExprMinMaxAggregationFunction.java
index 957ac86de29..00c4aeb191c 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ParentExprMinMaxAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ParentExprMinMaxAggregationFunction.java
@@ -26,6 +26,7 @@ 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.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;
@@ -36,6 +37,7 @@ import
org.apache.pinot.core.query.aggregation.utils.exprminmax.ExprMinMaxMeasur
import
org.apache.pinot.core.query.aggregation.utils.exprminmax.ExprMinMaxObject;
import
org.apache.pinot.core.query.aggregation.utils.exprminmax.ExprMinMaxProjectionValSetWrapper;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.roaringbitmap.RoaringBitmap;
public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFunction<ExprMinMaxObject, ExprMinMaxObject> {
@@ -54,6 +56,7 @@ public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFuncti
private final int _numMeasuringColumns;
// number of columns that we project based on the min/max value
private final int _numProjectionColumns;
+ private final boolean _nullHandlingEnabled;
// The following variable need to be initialized
@@ -68,10 +71,12 @@ public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFuncti
// If the schemas are initialized
private final ThreadLocal<Boolean> _schemaInitialized =
ThreadLocal.withInitial(() -> false);
- public ParentExprMinMaxAggregationFunction(List<ExpressionContext>
arguments, boolean isMax) {
+ public ParentExprMinMaxAggregationFunction(List<ExpressionContext>
arguments, boolean isMax,
+ boolean nullHandlingEnabled) {
super(arguments);
_isMax = isMax;
+ _nullHandlingEnabled = nullHandlingEnabled;
_functionIdContext = arguments.get(0);
_numMeasuringColumnContext = arguments.get(1);
@@ -110,33 +115,91 @@ public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFuncti
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
+ initializeWithNewDataBlocks(blockValSetMap);
- ExprMinMaxObject exprMinMaxObject = aggregationResultHolder.getResult();
-
- if (exprMinMaxObject == null) {
- initializeWithNewDataBlocks(blockValSetMap);
- exprMinMaxObject = new ExprMinMaxObject(_measuringColumnSchema.get(),
_projectionColumnSchema.get());
+ ExprMinMaxObject result = aggregationResultHolder.getResult();
+ if (result == null) {
+ result = new ExprMinMaxObject(_measuringColumnSchema.get(),
_projectionColumnSchema.get());
+ aggregationResultHolder.setValue(result);
}
+ ExprMinMaxObject exprMinMaxObject = result;
List<Integer> rowIds = new ArrayList<>();
- for (int i = 0; i < length; i++) {
- int compareResult =
exprMinMaxObject.compareAndSetKey(_exprMinMaxWrapperMeasuringColumnSets.get(),
i, _isMax);
- if (compareResult == 0) {
- // same key, add the rowId to the list
- rowIds.add(i);
- } else if (compareResult > 0) {
- // new key is set, clear the list and add the new rowId
- rowIds.clear();
- rowIds.add(i);
+ // Whether this block replaced the extremum key.
+ boolean[] keyReplaced = {false};
+ forEachNotNullMeasuring(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int compareResult =
exprMinMaxObject.compareAndSetKey(_exprMinMaxWrapperMeasuringColumnSets.get(),
i, _isMax);
+ if (compareResult == 0) {
+ // same key, add the rowId to the list
+ rowIds.add(i);
+ } else if (compareResult > 0) {
+ // new key is set, clear the list and add the new rowId
+ rowIds.clear();
+ rowIds.add(i);
+ keyReplaced[0] = true;
+ }
}
+ });
+
+ // For all the rows that are associated with the extremum key, add the
projection columns. The projection values
+ // are collected once here rather than per winning row, so that a row
later beaten within this block never reads
+ // the projection columns at all.
+ //
+ // Only matters across blocks. rowIds is local, so within one block
clearing it is enough and the object's value
+ // list is still empty; but the object outlives the call, and a block that
replaces the key has to discard what an
+ // earlier block published under the old one. rowIds.clear() cannot reach
those. Only setToNewVal clears them, so
+ // the first surviving row goes through it and the rows tying it are
appended.
+ List<ExprMinMaxProjectionValSetWrapper> projectionColumnSets =
_exprMinMaxWrapperProjectionColumnSets.get();
+ int firstToAppend = 0;
+ if (keyReplaced[0]) {
+ exprMinMaxObject.setToNewVal(projectionColumnSets, rowIds.get(0));
+ firstToAppend = 1;
}
+ for (int i = firstToAppend; i < rowIds.size(); i++) {
+ exprMinMaxObject.addVal(projectionColumnSets, rowIds.get(i));
+ }
+ }
- // for all the rows that are associated with the extremum key, add the
projection columns
- for (Integer rowId : rowIds) {
- exprMinMaxObject.addVal(_exprMinMaxWrapperProjectionColumnSets.get(),
rowId);
+ /// Runs `consumer` over the row ranges where every measuring column is
non-null.
+ ///
+ /// The measuring columns form a single composite key, so a null in any one
of them leaves the key undefined and the
+ /// row cannot take part in the comparison at all. The projection columns
are only payload carried out of the winning
+ /// row, so a null there does not disqualify it. With the option disabled
the whole block is one range, which is what
+ /// this function did unconditionally before.
+ private void forEachNotNullMeasuring(int length, Map<ExpressionContext,
BlockValSet> blockValSetMap,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ RoaringBitmap nullBitmap = measuringNullBitmap(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);
+ }
+ }
- aggregationResultHolder.setValue(exprMinMaxObject);
+ /// Returns the union of the measuring columns' null bitmaps, or `null` when
no row is null.
+ @Nullable
+ private RoaringBitmap measuringNullBitmap(Map<ExpressionContext,
BlockValSet> blockValSetMap) {
+ if (!_nullHandlingEnabled) {
+ return null;
+ }
+ RoaringBitmap merged = null;
+ for (ExpressionContext measuringColumn : _measuringColumns) {
+ RoaringBitmap nullBitmap =
blockValSetMap.get(measuringColumn).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;
}
// this method is called to initialize the schemas if they are not
initialized
@@ -225,10 +288,12 @@ public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFuncti
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
initializeWithNewDataBlocks(blockValSetMap);
- for (int i = 0; i < length; i++) {
- int groupKey = groupKeyArray[i];
- updateGroupByResult(groupByResultHolder, i, groupKey);
- }
+ forEachNotNullMeasuring(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int groupKey = groupKeyArray[i];
+ updateGroupByResult(groupByResultHolder, i, groupKey);
+ }
+ });
}
private void updateGroupByResult(GroupByResultHolder groupByResultHolder,
int i, int groupKey) {
@@ -249,11 +314,13 @@ public class ParentExprMinMaxAggregationFunction extends
ParentAggregationFuncti
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
initializeWithNewDataBlocks(blockValSetMap);
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupByResultHolder, i, groupKey);
+ forEachNotNullMeasuring(length, blockValSetMap, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ updateGroupByResult(groupByResultHolder, i, groupKey);
+ }
}
- }
+ });
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationFunction.java
index 1743256dfea..81ad63e0c75 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationFunction.java
@@ -33,6 +33,7 @@ import
org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.FunctionContext;
import org.apache.pinot.common.request.context.RequestContextUtils;
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;
@@ -82,6 +83,7 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
private final long _timeReferencePoint;
private final long _timeOffset;
private final long _timeBucketDivisor;
+ private final boolean _nullHandlingEnabled;
/// Arguments are as shown below:
///
@@ -89,7 +91,7 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
/// timeSeriesAggregate("m3ql", "MIN", valueExpr, timeExpr, timeUnit,
offsetSeconds, firstBucketValue,
/// bucketLenSeconds, numBuckets, "aggParam1=value1")
/// ```
- public TimeSeriesAggregationFunction(List<ExpressionContext> arguments) {
+ public TimeSeriesAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
// Initialize temporary variables.
Preconditions.checkArgument(arguments.size() == 10, "Expected 10 arguments
for time-series agg");
String language = arguments.get(0).getLiteral().getStringValue();
@@ -112,6 +114,7 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
_timeReferencePoint = timeUnit.convert(Duration.ofSeconds(firstBucketValue
- bucketWindowSeconds));
_timeOffset = timeUnit.convert(Duration.ofSeconds(offsetSeconds));
_timeBucketDivisor = timeUnit.convert(_timeBuckets.getBucketSize());
+ _nullHandlingEnabled = nullHandlingEnabled;
}
@Override
@@ -142,7 +145,7 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- final long[] timeValues =
blockValSetMap.get(_timeExpression).getLongValuesSV();
+ BlockValSet timeBlockValSet = blockValSetMap.get(_timeExpression);
BlockValSet valueBlockValSet = blockValSetMap.get(_valueExpression);
switch (valueBlockValSet.getValueType().getStoredType()) {
case INT:
@@ -150,10 +153,10 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
case FLOAT:
case DOUBLE:
case BIG_DECIMAL:
- aggregateNumericValues(length, timeValues, aggregationResultHolder,
valueBlockValSet);
+ aggregateNumericValues(length, timeBlockValSet,
aggregationResultHolder, valueBlockValSet);
break;
case STRING:
- aggregateStringValues(length, timeValues, aggregationResultHolder,
valueBlockValSet);
+ aggregateStringValues(length, timeBlockValSet,
aggregationResultHolder, valueBlockValSet);
break;
default:
throw new UnsupportedOperationException(String.format("Unsupported
type: %s in aggregate",
@@ -164,7 +167,7 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- final long[] timeValues =
blockValSetMap.get(_timeExpression).getLongValuesSV();
+ BlockValSet timeBlockValSet = blockValSetMap.get(_timeExpression);
BlockValSet valueBlockValSet = blockValSetMap.get(_valueExpression);
switch (valueBlockValSet.getValueType().getStoredType()) {
case INT:
@@ -172,10 +175,10 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
case FLOAT:
case DOUBLE:
case BIG_DECIMAL:
- aggregateGroupByNumericValues(length, groupKeyArray, timeValues,
groupByResultHolder, valueBlockValSet);
+ aggregateGroupByNumericValues(length, groupKeyArray, timeBlockValSet,
groupByResultHolder, valueBlockValSet);
break;
case STRING:
- aggregateGroupByStringValues(length, groupKeyArray, timeValues,
groupByResultHolder, valueBlockValSet);
+ aggregateGroupByStringValues(length, groupKeyArray, timeBlockValSet,
groupByResultHolder, valueBlockValSet);
break;
default:
throw new UnsupportedOperationException(String.format("Unsupported
type: %s in aggregate",
@@ -233,70 +236,96 @@ public class TimeSeriesAggregationFunction implements
AggregationFunction<BaseTi
return "TIME_SERIES";
}
- private void aggregateNumericValues(int length, long[] timeValues,
AggregationResultHolder resultHolder,
+ /// Runs `consumer` over the row ranges where both the value and the
timestamp are non-null.
+ ///
+ /// Both columns gate the row. A null value has nothing to add to its
bucket, and a null timestamp leaves the point
+ /// with no bucket to land in, so either one makes the row uncountable
rather than merely incomplete. With the option
+ /// disabled the whole block is one range, which is what this function did
unconditionally before.
+ private void forEachNotNull(int length, BlockValSet valueBlockValSet,
BlockValSet timeBlockValSet,
+ RoaringBitmapUtils.BatchConsumer consumer) {
+ RoaringBitmapUtils.forEachUnset(length,
+
NullableSingleInputAggregationFunction.orNullIterator(_nullHandlingEnabled,
valueBlockValSet, timeBlockValSet),
+ consumer);
+ }
+
+ /// Returns the holder's series builder, creating and storing one on first
use.
+ ///
+ /// Called from inside a non-null range rather than before the loop, so that
a block whose rows are all null leaves
+ /// the holder untouched and [#extractFinalResult] sees the `null` that
means nothing was aggregated.
+ private BaseTimeSeriesBuilder
getOrCreateSeriesBuilder(AggregationResultHolder resultHolder) {
+ BaseTimeSeriesBuilder seriesBuilder = resultHolder.getResult();
+ if (seriesBuilder == null) {
+ seriesBuilder = newSeriesBuilder();
+ resultHolder.setValue(seriesBuilder);
+ }
+ return seriesBuilder;
+ }
+
+ private BaseTimeSeriesBuilder newSeriesBuilder() {
+ return _factory.newTimeSeriesBuilder(_aggInfo, "TO_BE_REMOVED",
_timeBuckets,
+ BaseTimeSeriesBuilder.UNINITIALISED_TAG_NAMES,
BaseTimeSeriesBuilder.UNINITIALISED_TAG_VALUES);
+ }
+
+ private int timeIndex(long timeValue) {
+ return (int) (((timeValue + _timeOffset) - _timeReferencePoint - 1) /
_timeBucketDivisor);
+ }
+
+ private void aggregateNumericValues(int length, BlockValSet timeBlockValSet,
AggregationResultHolder resultHolder,
BlockValSet blockValSet) {
+ long[] timeValues = timeBlockValSet.getLongValuesSV();
double[] values = blockValSet.getDoubleValuesSV();
- BaseTimeSeriesBuilder currentSeriesBuilder = resultHolder.getResult();
- if (currentSeriesBuilder == null) {
- currentSeriesBuilder = _factory.newTimeSeriesBuilder(_aggInfo,
"TO_BE_REMOVED", _timeBuckets,
- BaseTimeSeriesBuilder.UNINITIALISED_TAG_NAMES,
BaseTimeSeriesBuilder.UNINITIALISED_TAG_VALUES);
- resultHolder.setValue(currentSeriesBuilder);
- }
- int timeIndex;
- for (int docIndex = 0; docIndex < length; docIndex++) {
- timeIndex = (int) (((timeValues[docIndex] + _timeOffset) -
_timeReferencePoint - 1) / _timeBucketDivisor);
- currentSeriesBuilder.addValueAtIndex(timeIndex, values[docIndex],
timeValues[docIndex]);
- }
+ forEachNotNull(length, blockValSet, timeBlockValSet, (from, to) -> {
+ BaseTimeSeriesBuilder currentSeriesBuilder =
getOrCreateSeriesBuilder(resultHolder);
+ for (int docIndex = from; docIndex < to; docIndex++) {
+ currentSeriesBuilder.addValueAtIndex(timeIndex(timeValues[docIndex]),
values[docIndex], timeValues[docIndex]);
+ }
+ });
}
- private void aggregateStringValues(int length, long[] timeValues,
AggregationResultHolder resultHolder,
+ private void aggregateStringValues(int length, BlockValSet timeBlockValSet,
AggregationResultHolder resultHolder,
BlockValSet blockValSet) {
+ long[] timeValues = timeBlockValSet.getLongValuesSV();
String[] values = blockValSet.getStringValuesSV();
- BaseTimeSeriesBuilder currentSeriesBuilder = resultHolder.getResult();
- if (currentSeriesBuilder == null) {
- currentSeriesBuilder = _factory.newTimeSeriesBuilder(_aggInfo,
"TO_BE_REMOVED", _timeBuckets,
- BaseTimeSeriesBuilder.UNINITIALISED_TAG_NAMES,
BaseTimeSeriesBuilder.UNINITIALISED_TAG_VALUES);
- resultHolder.setValue(currentSeriesBuilder);
- }
- int timeIndex;
- for (int docIndex = 0; docIndex < length; docIndex++) {
- timeIndex = (int) (((timeValues[docIndex] + _timeOffset) -
_timeReferencePoint - 1) / _timeBucketDivisor);
- currentSeriesBuilder.addValueAtIndex(timeIndex, values[docIndex],
timeValues[docIndex]);
- }
+ forEachNotNull(length, blockValSet, timeBlockValSet, (from, to) -> {
+ BaseTimeSeriesBuilder currentSeriesBuilder =
getOrCreateSeriesBuilder(resultHolder);
+ for (int docIndex = from; docIndex < to; docIndex++) {
+ currentSeriesBuilder.addValueAtIndex(timeIndex(timeValues[docIndex]),
values[docIndex], timeValues[docIndex]);
+ }
+ });
}
- private void aggregateGroupByNumericValues(int length, int[] groupKeyArray,
long[] timeValues,
+ private void aggregateGroupByNumericValues(int length, int[] groupKeyArray,
BlockValSet timeBlockValSet,
GroupByResultHolder resultHolder, BlockValSet blockValSet) {
+ long[] timeValues = timeBlockValSet.getLongValuesSV();
final double[] values = blockValSet.getDoubleValuesSV();
- int timeIndex;
- for (int docIndex = 0; docIndex < length; docIndex++) {
- int groupId = groupKeyArray[docIndex];
- BaseTimeSeriesBuilder currentSeriesBuilder =
resultHolder.getResult(groupId);
- if (currentSeriesBuilder == null) {
- currentSeriesBuilder = _factory.newTimeSeriesBuilder(_aggInfo,
"TO_BE_REMOVED", _timeBuckets,
- BaseTimeSeriesBuilder.UNINITIALISED_TAG_NAMES,
BaseTimeSeriesBuilder.UNINITIALISED_TAG_VALUES);
- resultHolder.setValueForKey(groupId, currentSeriesBuilder);
+ forEachNotNull(length, blockValSet, timeBlockValSet, (from, to) -> {
+ for (int docIndex = from; docIndex < to; docIndex++) {
+ int groupId = groupKeyArray[docIndex];
+ BaseTimeSeriesBuilder currentSeriesBuilder =
resultHolder.getResult(groupId);
+ if (currentSeriesBuilder == null) {
+ currentSeriesBuilder = newSeriesBuilder();
+ resultHolder.setValueForKey(groupId, currentSeriesBuilder);
+ }
+ currentSeriesBuilder.addValueAtIndex(timeIndex(timeValues[docIndex]),
values[docIndex], timeValues[docIndex]);
}
- timeIndex = (int) (((timeValues[docIndex] + _timeOffset) -
_timeReferencePoint - 1) / _timeBucketDivisor);
- currentSeriesBuilder.addValueAtIndex(timeIndex, values[docIndex],
timeValues[docIndex]);
- }
+ });
}
- private void aggregateGroupByStringValues(int length, int[] groupKeyArray,
long[] timeValues,
+ private void aggregateGroupByStringValues(int length, int[] groupKeyArray,
BlockValSet timeBlockValSet,
GroupByResultHolder resultHolder, BlockValSet blockValSet) {
+ long[] timeValues = timeBlockValSet.getLongValuesSV();
final String[] values = blockValSet.getStringValuesSV();
- int timeIndex;
- for (int docIndex = 0; docIndex < length; docIndex++) {
- int groupId = groupKeyArray[docIndex];
- BaseTimeSeriesBuilder currentSeriesBuilder =
resultHolder.getResult(groupId);
- if (currentSeriesBuilder == null) {
- currentSeriesBuilder = _factory.newTimeSeriesBuilder(_aggInfo,
"TO_BE_REMOVED", _timeBuckets,
- BaseTimeSeriesBuilder.UNINITIALISED_TAG_NAMES,
BaseTimeSeriesBuilder.UNINITIALISED_TAG_VALUES);
- resultHolder.setValueForKey(groupId, currentSeriesBuilder);
+ forEachNotNull(length, blockValSet, timeBlockValSet, (from, to) -> {
+ for (int docIndex = from; docIndex < to; docIndex++) {
+ int groupId = groupKeyArray[docIndex];
+ BaseTimeSeriesBuilder currentSeriesBuilder =
resultHolder.getResult(groupId);
+ if (currentSeriesBuilder == null) {
+ currentSeriesBuilder = newSeriesBuilder();
+ resultHolder.setValueForKey(groupId, currentSeriesBuilder);
+ }
+ currentSeriesBuilder.addValueAtIndex(timeIndex(timeValues[docIndex]),
values[docIndex], timeValues[docIndex]);
}
- timeIndex = (int) (((timeValues[docIndex] + _timeOffset) -
_timeReferencePoint - 1) / _timeBucketDivisor);
- currentSeriesBuilder.addValueAtIndex(timeIndex, values[docIndex],
timeValues[docIndex]);
- }
+ });
}
public static ExpressionContext create(String language, String
valueExpressionStr, ExpressionContext timeExpression,
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 1a8df4c13da..1de4e8cfb44 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
@@ -97,8 +97,13 @@ public class AggregationFunctionNullContractTest {
/// Types that cannot be constructed from a bare [FunctionContext] at all,
and so cannot be checked here.
///
/// - `EXPRMIN` / `EXPRMAX` are rejected by the factory outright; they are
only legal in a selection without an alias,
- /// and are rewritten into the parent/child pair above before execution.
- /// - `TIMESERIESAGGREGATE` needs time-series plan context that a bare
function context cannot supply.
+ /// and are rewritten into the parent/child pair above before execution.
Covered by `ExprMinMaxNullHandlingTest`,
+ /// which drives the parent directly.
+ /// - `TIMESERIESAGGREGATE` needs time-series plan context that a bare
function context cannot supply. Covered by
+ /// `TimeSeriesAggregationNullHandlingTest`.
+ ///
+ /// Being here means the contract is checked by the named test instead, not
that it goes unchecked. A new entry
+ /// needs a test of its own before it is added.
///
/// [#testEverySkippedTypeIsAccountedFor] pins this exactly, in both
directions, so a newly added function cannot drop
/// out of the contract unnoticed.
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ExprMinMaxNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ExprMinMaxNullHandlingTest.java
new file mode 100644
index 00000000000..508274ff870
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ExprMinMaxNullHandlingTest.java
@@ -0,0 +1,201 @@
+/**
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+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.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.utils.exprminmax.ExprMinMaxObject;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Null handling for `EXPR_MIN` and `EXPR_MAX`.
+///
+/// These cannot be reached through [AggregationFunctionNullContractTest]: the
factory rejects `EXPRMIN` and `EXPRMAX`
+/// outright, since they are only legal in a selection without an alias and
are rewritten into a parent/child pair
+/// before execution. So the parent is driven directly here.
+public class ExprMinMaxNullHandlingTest {
+ private static final ExpressionContext MEASURING =
ExpressionContext.forIdentifier("measuring");
+ private static final ExpressionContext PROJECTION =
ExpressionContext.forIdentifier("projection");
+
+ @Test
+ public void nullMeasuringValueDoesNotWinWhenNullHandlingEnabled() {
+ // Row 1 holds the smallest value but is null, so row 0 is the minimum
+ ExprMinMaxObject result = aggregate(false, true, nulls(1), new int[]{10,
3, 20}, new int[]{100, 101, 102});
+
+ assertEquals(result.getExtremumKey()[0], 10);
+ assertEquals(result.getNumberOfRows(), 1);
+ assertEquals(result.getField(0, 0), 100);
+ }
+
+ @Test
+ public void nullMeasuringValueWinsWhenNullHandlingDisabled() {
+ ExprMinMaxObject result = aggregate(false, false, nulls(1), new int[]{10,
3, 20}, new int[]{100, 101, 102});
+
+ assertEquals(result.getExtremumKey()[0], 3);
+ assertEquals(result.getField(0, 0), 101);
+ }
+
+ @Test
+ public void nullMeasuringValueDoesNotWinForMax() {
+ ExprMinMaxObject result = aggregate(true, true, nulls(2), new int[]{10, 3,
20}, new int[]{100, 101, 102});
+
+ assertEquals(result.getExtremumKey()[0], 10);
+ assertEquals(result.getField(0, 0), 100);
+ }
+
+ @Test
+ public void everyRowNullMeasuresNothing() {
+ ExprMinMaxObject result = aggregate(false, true, nulls(0, 1, 2), new
int[]{10, 3, 20}, new int[]{100, 101, 102});
+
+ assertEquals(result.getNumberOfRows(), 0);
+ }
+
+ @Test
+ public void tiedRowsAllProjectWhenTheirMeasuringValueIsNotNull() {
+ ExprMinMaxObject result = aggregate(false, true, nulls(), new int[]{10,
10, 20}, new int[]{100, 101, 102});
+
+ assertEquals(result.getExtremumKey()[0], 10);
+ assertEquals(result.getNumberOfRows(), 2);
+ assertEquals(result.getField(0, 0), 100);
+ assertEquals(result.getField(1, 0), 101);
+ }
+
+ /// A second block must be read with its own values.
+ ///
+ /// The wrappers hold the block they were last bound to, so a function that
only binds them while creating its
+ /// accumulator reads the first block's values at the second block's row
offsets.
+ @Test
+ public void secondBlockIsReadWithItsOwnValues() {
+ ParentExprMinMaxAggregationFunction function = function(false, true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+
+ function.aggregate(2, holder, block(nulls(), new int[]{10, 20}, new
int[]{100, 101}));
+ function.aggregate(2, holder, block(nulls(), new int[]{3, 30}, new
int[]{200, 201}));
+
+ ExprMinMaxObject result = function.extractAggregationResult(holder);
+ assertEquals(result.getExtremumKey()[0], 3);
+ assertEquals(result.getNumberOfRows(), 1);
+ assertEquals(result.getField(0, 0), 200);
+ }
+
+ /// A block that replaces the key must discard what the previous key
projected, and still keep the rows that tie the
+ /// new key. `rowIds.clear()` cannot do the first part: it only drops
candidates found within the current block.
+ @Test
+ public void
aReplacedKeyDiscardsTheEarlierBlocksProjectionButKeepsItsOwnTies() {
+ ParentExprMinMaxAggregationFunction function = function(false, true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+
+ function.aggregate(2, holder, block(nulls(), new int[]{10, 20}, new
int[]{100, 101}));
+ function.aggregate(3, holder, block(nulls(), new int[]{3, 3, 30}, new
int[]{200, 201, 202}));
+
+ ExprMinMaxObject result = function.extractAggregationResult(holder);
+ assertEquals(result.getExtremumKey()[0], 3);
+ assertEquals(result.getNumberOfRows(), 2);
+ assertEquals(result.getField(0, 0), 200);
+ assertEquals(result.getField(1, 0), 201);
+ }
+
+ @Test
+ public void nullMeasuringValueDoesNotWinInItsGroup() {
+ ParentExprMinMaxAggregationFunction function = function(false, true);
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+
+ // Rows 0 and 1 are group 0; row 1 holds the smaller value but is null
+ function.aggregateGroupBySV(4, new int[]{0, 0, 1, 1}, holder,
+ block(nulls(1), new int[]{10, 3, 20, 30}, new int[]{100, 101, 102,
103}));
+
+ ExprMinMaxObject group0 = function.extractGroupByResult(holder, 0);
+ assertEquals(group0.getExtremumKey()[0], 10);
+ assertEquals(group0.getNumberOfRows(), 1);
+ assertEquals(group0.getField(0, 0), 100);
+
+ ExprMinMaxObject group1 = function.extractGroupByResult(holder, 1);
+ assertEquals(group1.getExtremumKey()[0], 20);
+ assertEquals(group1.getField(0, 0), 102);
+ }
+
+ @Test
+ public void nullMeasuringValueDoesNotWinInAnyOfItsGroups() {
+ ParentExprMinMaxAggregationFunction function = function(false, true);
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+
+ // Row 1 belongs to both groups and holds the smaller value, but is null
+ function.aggregateGroupByMV(2, new int[][]{{0, 1}, {0, 1}}, holder,
+ block(nulls(1), new int[]{10, 3}, new int[]{100, 101}));
+
+ assertEquals(function.extractGroupByResult(holder, 0).getExtremumKey()[0],
10);
+ assertEquals(function.extractGroupByResult(holder, 0).getField(0, 0), 100);
+ assertEquals(function.extractGroupByResult(holder, 1).getExtremumKey()[0],
10);
+ assertEquals(function.extractGroupByResult(holder, 1).getField(0, 0), 100);
+ }
+
+ /// The group-by path publishes each winning row as it is found rather than
batching them, so a replaced key clears
+ /// the earlier block's projection through `setToNewVal` on its own. Pinned
because the non-group-by path needed a
+ /// fix to reach the same behaviour.
+ @Test
+ public void aReplacedKeyInAGroupDiscardsTheEarlierBlocksProjection() {
+ ParentExprMinMaxAggregationFunction function = function(false, true);
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+
+ function.aggregateGroupBySV(2, new int[]{0, 0}, holder, block(nulls(), new
int[]{10, 20}, new int[]{100, 101}));
+ function.aggregateGroupBySV(2, new int[]{0, 0}, holder, block(nulls(), new
int[]{3, 30}, new int[]{200, 201}));
+
+ ExprMinMaxObject group0 = function.extractGroupByResult(holder, 0);
+ assertEquals(group0.getExtremumKey()[0], 3);
+ assertEquals(group0.getNumberOfRows(), 1);
+ assertEquals(group0.getField(0, 0), 200);
+ }
+
+ private static ExprMinMaxObject aggregate(boolean isMax, boolean
nullHandlingEnabled, RoaringBitmap measuringNulls,
+ int[] measuring, int[] projection) {
+ ParentExprMinMaxAggregationFunction function = function(isMax,
nullHandlingEnabled);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(measuring.length, holder, block(measuringNulls,
measuring, projection));
+ return function.extractAggregationResult(holder);
+ }
+
+ private static ParentExprMinMaxAggregationFunction function(boolean isMax,
boolean nullHandlingEnabled) {
+ return new ParentExprMinMaxAggregationFunction(List.of(
+ ExpressionContext.forLiteral(Literal.intValue(0)),
+ ExpressionContext.forLiteral(Literal.intValue(1)),
+ MEASURING,
+ PROJECTION), isMax, nullHandlingEnabled);
+ }
+
+ private static Map<ExpressionContext, BlockValSet> block(RoaringBitmap
measuringNulls, int[] measuring,
+ int[] projection) {
+ return Map.of(
+ MEASURING, SyntheticBlockValSets.Int.create(measuringNulls, measuring),
+ PROJECTION, SyntheticBlockValSets.Int.create(null, projection)
+ );
+ }
+
+ private static RoaringBitmap nulls(int... indexes) {
+ return indexes.length == 0 ? null : RoaringBitmap.bitmapOf(indexes);
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StoredTypeDispatchTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StoredTypeDispatchTest.java
index 751ec0ba897..6afec5210a8 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StoredTypeDispatchTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StoredTypeDispatchTest.java
@@ -226,7 +226,7 @@ public class StoredTypeDispatchTest {
ExpressionContext.forLiteral(Literal.intValue(2)),
ExpressionContext.forLiteral(Literal.stringValue("")));
- TimeSeriesAggregationFunction function = new
TimeSeriesAggregationFunction(arguments);
+ TimeSeriesAggregationFunction function = new
TimeSeriesAggregationFunction(arguments, false);
AggregationResultHolder holder = function.createAggregationResultHolder();
function.aggregate(NUM_DOCS, holder, timeSeriesBlocks(valueBlock));
return function.extractAggregationResult(holder);
@@ -242,7 +242,7 @@ public class StoredTypeDispatchTest {
ExpressionContext.forLiteral(Literal.longValue(100)),
ExpressionContext.forLiteral(Literal.longValue(10)),
ExpressionContext.forLiteral(Literal.intValue(2)),
- ExpressionContext.forLiteral(Literal.stringValue(""))));
+ ExpressionContext.forLiteral(Literal.stringValue(""))), false);
}
private static Map<ExpressionContext, BlockValSet>
timeSeriesBlocks(BlockValSet valueBlock) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationNullHandlingTest.java
new file mode 100644
index 00000000000..787febc0c81
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/TimeSeriesAggregationNullHandlingTest.java
@@ -0,0 +1,143 @@
+/**
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+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.groupby.GroupByResultHolder;
+import org.apache.pinot.tsdb.spi.series.SimpleTimeSeriesBuilderFactory;
+import org.apache.pinot.tsdb.spi.series.TimeSeriesBuilderFactoryProvider;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+
+
+/// Null handling for `TIMESERIESAGGREGATE`.
+///
+/// This cannot be reached through [AggregationFunctionNullContractTest],
which builds every function from a bare
+/// [org.apache.pinot.common.request.context.FunctionContext]; this one needs
time-series plan context. It is not
+/// reachable from SQL either, but the time-series plan node passes the query
options straight through, so a
+/// time-series query that sets `enableNullHandling` arrives here with the
option on.
+///
+/// The rows are laid out as two buckets of two rows: timestamps `100, 100,
110, 110` fall into bucket `0` and bucket
+/// `1` respectively, and the aggregation is a `SUM`. The bucket values are
read off the series builder rather than
+/// through [TimeSeriesAggregationFunction#extractFinalResult], because that
is the path the time-series engine takes
+/// and because a bucket no row landed in stays `null`, which the final-result
conversion cannot represent.
+public class TimeSeriesAggregationNullHandlingTest {
+ private static final String LANGUAGE =
"TimeSeriesAggregationNullHandlingTest";
+ private static final ExpressionContext TIME =
ExpressionContext.forIdentifier("time");
+ private static final ExpressionContext VALUE =
ExpressionContext.forIdentifier("value");
+ private static final long[] TIMESTAMPS = {100L, 100L, 110L, 110L};
+ private static final double[] VALUES = {1.0, 2.0, 3.0, 4.0};
+
+ @BeforeClass
+ public void setUp() {
+ TimeSeriesBuilderFactoryProvider.registerSeriesBuilderFactory(LANGUAGE,
new SimpleTimeSeriesBuilderFactory());
+ }
+
+ @Test
+ public void nullValueIsNotAggregatedWhenNullHandlingEnabled() {
+ Double[] buckets = aggregate(true, RoaringBitmap.bitmapOf(0), null);
+
+ assertEquals(buckets[0].doubleValue(), 2.0);
+ assertEquals(buckets[1].doubleValue(), 7.0);
+ }
+
+ @Test
+ public void nullValueIsAggregatedWhenNullHandlingDisabled() {
+ Double[] buckets = aggregate(false, RoaringBitmap.bitmapOf(0), null);
+
+ assertEquals(buckets[0].doubleValue(), 3.0);
+ assertEquals(buckets[1].doubleValue(), 7.0);
+ }
+
+ /// A null timestamp leaves the point with no bucket to land in, so the row
is skipped even though its value is
+ /// perfectly good.
+ @Test
+ public void nullTimestampSkipsTheRow() {
+ Double[] buckets = aggregate(true, null, RoaringBitmap.bitmapOf(2));
+
+ assertEquals(buckets[0].doubleValue(), 3.0);
+ assertEquals(buckets[1].doubleValue(), 4.0);
+ }
+
+ @Test
+ public void aBucketWhoseRowsAreAllNullIsNeverWrittenTo() {
+ Double[] buckets = aggregate(true, RoaringBitmap.bitmapOf(0, 1), null);
+
+ assertNull(buckets[0]);
+ assertEquals(buckets[1].doubleValue(), 7.0);
+ }
+
+ @Test
+ public void everyRowNullLeavesNothingAggregated() {
+ TimeSeriesAggregationFunction function = function(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(VALUES.length, holder, block(RoaringBitmap.bitmapOf(0,
1, 2, 3), null));
+
+ assertNull(function.extractAggregationResult(holder), "the holder should
never have been touched");
+ }
+
+ @Test
+ public void nullValueIsNotAggregatedIntoItsGroup() {
+ TimeSeriesAggregationFunction function = function(true);
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(VALUES.length, new int[]{0, 0, 1, 1}, holder,
+ block(RoaringBitmap.bitmapOf(0), null));
+
+ assertEquals(function.extractGroupByResult(holder,
0).build().getDoubleValues()[0].doubleValue(), 2.0);
+ assertEquals(function.extractGroupByResult(holder,
1).build().getDoubleValues()[1].doubleValue(), 7.0);
+ }
+
+ private static Double[] aggregate(boolean nullHandlingEnabled, RoaringBitmap
valueNulls, RoaringBitmap timeNulls) {
+ TimeSeriesAggregationFunction function = function(nullHandlingEnabled);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(VALUES.length, holder, block(valueNulls, timeNulls));
+ return function.extractAggregationResult(holder).build().getDoubleValues();
+ }
+
+ private static TimeSeriesAggregationFunction function(boolean
nullHandlingEnabled) {
+ return new TimeSeriesAggregationFunction(List.of(
+ ExpressionContext.forLiteral(Literal.stringValue(LANGUAGE)),
+ ExpressionContext.forLiteral(Literal.stringValue("SUM")),
+ VALUE,
+ TIME,
+ ExpressionContext.forLiteral(Literal.stringValue("SECONDS")),
+ ExpressionContext.forLiteral(Literal.longValue(0)),
+ ExpressionContext.forLiteral(Literal.longValue(100)),
+ ExpressionContext.forLiteral(Literal.longValue(10)),
+ ExpressionContext.forLiteral(Literal.intValue(2)),
+ ExpressionContext.forLiteral(Literal.stringValue(""))),
nullHandlingEnabled);
+ }
+
+ private static Map<ExpressionContext, BlockValSet> block(RoaringBitmap
valueNulls, RoaringBitmap timeNulls) {
+ return Map.of(
+ TIME, SyntheticBlockValSets.Long.create(timeNulls, TIMESTAMPS),
+ VALUE, SyntheticBlockValSets.Double.create(valueNulls, VALUES)
+ );
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
