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 a2f96c487ec Always support nulls in DataTable serialization (#19201)
a2f96c487ec is described below
commit a2f96c487ec403302648177e88466143d4331977
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Mon Aug 10 11:49:41 2026 -0700
Always support nulls in DataTable serialization (#19201)
---
.../common/datatable/BaseDataTableBuilder.java | 101 ++++++++-
.../core/common/datatable/DataTableBuilder.java | 13 +-
.../common/datatable/DataTableBuilderUtils.java | 9 +-
.../core/common/datatable/DataTableBuilderV4.java | 23 +--
.../apache/pinot/core/data/table/TableResizer.java | 2 +-
.../blocks/results/AggregationResultsBlock.java | 78 ++-----
.../blocks/results/GroupByResultsBlock.java | 77 ++-----
.../query/reduce/AggregationDataTableReducer.java | 83 ++------
.../core/query/reduce/GroupByDataTableReducer.java | 95 +++------
.../core/query/request/context/QueryContext.java | 10 +-
.../core/common/datatable/DataTableSerDeTest.java | 229 +++++++++++++++------
.../function/MaxStringAggregationFunctionTest.java | 70 +++++++
.../core/query/reduce/MergeDataTablesOnlyTest.java | 40 ++++
13 files changed, 494 insertions(+), 336 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/BaseDataTableBuilder.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/BaseDataTableBuilder.java
index db2f7f0214c..eba77e37300 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/BaseDataTableBuilder.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/BaseDataTableBuilder.java
@@ -23,19 +23,44 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.util.EnumSet;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.datatable.DataTableUtils;
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.query.aggregation.function.AggregationFunction;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.ByteArray;
import org.apache.pinot.spi.utils.MapUtils;
+import org.roaringbitmap.RoaringBitmap;
/// Base DataTableBuilder implementation.
+///
+/// Null values are always supported, independent of the query's null-handling
option. [#setNull] accepts any column
+/// type: types that carry their own in-band null encoding keep it, and every
other type is written as the column's
+/// null placeholder with the row recorded in a per-column null bitmap. The
bitmap section is appended by
+/// [#writeNullBitmaps], which subclasses must invoke from `build()` after the
last row.
+///
+/// The bitmap section is emitted only when at least one null was recorded, so
a table without nulls is byte-for-byte
+/// identical to one produced before null bitmaps existed. Readers detect the
section by buffer length and cannot
+/// distinguish "no section because no nulls" from "no section because the
writer never emitted one" -- both mean the
+/// same thing.
public abstract class BaseDataTableBuilder implements DataTableBuilder {
+ /// Column types whose null value is already representable in-band, and
which therefore need no bitmap entry:
+ /// - `OBJECT` and `UNKNOWN` are read back via `DataTable.getCustomObject`,
which returns `null` for the
+ /// [CustomObject#NULL_TYPE_VALUE] marker.
+ /// - `MAP` is read back via `DataTable.getMap`, which returns `null` for a
zero-length entry.
+ ///
+ /// These are also the only types for which the legacy `setNull` encoding
was ever correct: it writes an 8-byte
+ /// (offset, length) pair, which overflows or corrupts the 4-byte slot of an
`INT` / `FLOAT` / `STRING` column and
+ /// decodes as garbage in an 8-byte `LONG` / `DOUBLE` one.
+ private static final EnumSet<ColumnDataType> IN_BAND_NULL_TYPES =
+ EnumSet.of(ColumnDataType.OBJECT, ColumnDataType.UNKNOWN,
ColumnDataType.MAP);
+
protected final DataSchema _dataSchema;
protected final int _version;
protected final int[] _columnOffsets;
@@ -47,6 +72,14 @@ public abstract class BaseDataTableBuilder implements
DataTableBuilder {
protected final DataOutputStream _variableSizeDataOutputStream =
new DataOutputStream(_variableSizeDataByteArrayOutputStream);
+ private final ColumnDataType[] _storedColumnDataTypes;
+ /// Entries are allocated lazily so a table without nulls pays nothing.
+ private final RoaringBitmap[] _nullBitmaps;
+ private boolean _hasNulls;
+ private boolean _nullBitmapsWritten;
+ /// Cursor for the positional [#setNullRowIds] contract.
+ private int _nullRowIdsColId;
+
protected int _numRows;
protected ByteBuffer _currentRowDataByteBuffer;
@@ -55,6 +88,8 @@ public abstract class BaseDataTableBuilder implements
DataTableBuilder {
_version = version;
_columnOffsets = new int[dataSchema.size()];
_rowSizeInBytes = DataTableUtils.computeColumnOffsets(dataSchema,
_columnOffsets, _version);
+ _storedColumnDataTypes = dataSchema.getStoredColumnDataTypes();
+ _nullBitmaps = new RoaringBitmap[dataSchema.size()];
}
@Override
@@ -196,10 +231,27 @@ public abstract class BaseDataTableBuilder implements
DataTableBuilder {
@Override
public void setNull(int colId)
throws IOException {
- _currentRowDataByteBuffer.position(_columnOffsets[colId]);
-
_currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
- _currentRowDataByteBuffer.putInt(0);
- _variableSizeDataOutputStream.writeInt(CustomObject.NULL_TYPE_VALUE);
+ ColumnDataType storedColumnDataType = _storedColumnDataTypes[colId];
+ if (IN_BAND_NULL_TYPES.contains(storedColumnDataType)) {
+ _currentRowDataByteBuffer.position(_columnOffsets[colId]);
+
_currentRowDataByteBuffer.putInt(_variableSizeDataByteArrayOutputStream.size());
+ _currentRowDataByteBuffer.putInt(0);
+ _variableSizeDataOutputStream.writeInt(CustomObject.NULL_TYPE_VALUE);
+ return;
+ }
+ // Resolved on the logical type, not the stored type: UUID overrides
getNullPlaceholder() to return the nil UUID,
+ // whereas its stored type BYTES would yield a zero-length placeholder
that is not a valid UUID.
+ Object nullPlaceholder =
_dataSchema.getColumnDataType(colId).getNullPlaceholder();
+ assert nullPlaceholder != null;
+ DataTableBuilderUtils.setColumn(this, storedColumnDataType, colId,
nullPlaceholder);
+ RoaringBitmap nullBitmap = _nullBitmaps[colId];
+ if (nullBitmap == null) {
+ nullBitmap = new RoaringBitmap();
+ _nullBitmaps[colId] = nullBitmap;
+ _hasNulls = true;
+ }
+ // startRow() has already incremented _numRows for the row being written.
+ nullBitmap.add(_numRows - 1);
}
@Override
@@ -207,4 +259,45 @@ public abstract class BaseDataTableBuilder implements
DataTableBuilder {
throws IOException {
_fixedSizeDataByteArrayOutputStream.write(_currentRowDataByteBuffer.array());
}
+
+ @Override
+ public void setNullRowIds(@Nullable RoaringBitmap nullRowIds) {
+ int colId = _nullRowIdsColId++;
+ if (nullRowIds == null || nullRowIds.isEmpty()) {
+ return;
+ }
+ RoaringBitmap nullBitmap = _nullBitmaps[colId];
+ if (nullBitmap == null) {
+ // Copy rather than alias: the caller retains ownership of the bitmap it
passed in.
+ nullBitmap = new RoaringBitmap();
+ _nullBitmaps[colId] = nullBitmap;
+ _hasNulls = true;
+ }
+ nullBitmap.or(nullRowIds);
+ }
+
+ /// Appends the per-column null bitmap section to the fixed and variable
size buffers. Subclasses must invoke this
+ /// from `build()`, after every row has been written and before the buffers
are handed off.
+ ///
+ /// Writes nothing when no null was recorded, keeping such tables
byte-for-byte identical to the pre-null-bitmap
+ /// format. The section is all-or-nothing across columns because readers
index into it at a fixed stride.
+ ///
+ /// Idempotent, so that `build()` can be invoked more than once on the same
builder.
+ protected void writeNullBitmaps()
+ throws IOException {
+ if (!_hasNulls || _nullBitmapsWritten) {
+ return;
+ }
+ _nullBitmapsWritten = true;
+ for (RoaringBitmap nullBitmap : _nullBitmaps) {
+
_fixedSizeDataOutputStream.writeInt(_variableSizeDataByteArrayOutputStream.size());
+ if (nullBitmap == null) {
+ _fixedSizeDataOutputStream.writeInt(0);
+ } else {
+ byte[] bitmapBytes = RoaringBitmapUtils.serialize(nullBitmap);
+ _fixedSizeDataOutputStream.writeInt(bitmapBytes.length);
+ _variableSizeDataByteArrayOutputStream.write(bitmapBytes);
+ }
+ }
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
index 0348b2f0d4c..7682b225147 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilder.java
@@ -92,14 +92,23 @@ public interface DataTableBuilder {
void setColumn(int colId, AggregationFunction.SerializedIntermediateResult
value)
throws IOException;
+ /// Writes a `null` for the given column of the current row.
+ ///
+ /// Valid for every column type, independent of the query's null-handling
option. Types that can represent `null`
+ /// in-band (`OBJECT`, `UNKNOWN`, `MAP`) keep their own encoding; every
other type is written as the column's null
+ /// placeholder and the row is recorded in a per-column null bitmap that
`build()` appends to the table. Readers
+ /// restore the `null` via `DataTable.getNullRowIds`.
void setNull(int colId)
throws IOException;
void finishRow()
throws IOException;
- /// NOTE: When setting nullRowIds, we don't pass the colId currently, and
this method must be invoked for all columns.
- /// TODO: Revisit this
+ /// Merges a pre-computed null bitmap into the bitmap the builder maintains
for the next column.
+ ///
+ /// NOTE: The colId is positional -- the first call targets column 0, the
second column 1, and so on -- so callers
+ /// that use this method must invoke it once for every column, in order.
Callers that instead report nulls per cell
+ /// through [#setNull] need not call this at all.
void setNullRowIds(@Nullable RoaringBitmap nullRowIds)
throws IOException;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderUtils.java
index cee3fe689f2..456a0ecb63b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderUtils.java
@@ -35,10 +35,11 @@ public class DataTableBuilderUtils {
private DataTableBuilderUtils() {
}
- /// Writes a non-null value of the given stored column data type into the
[DataTableBuilder] at
- /// the given column. Supports all scalar and array stored types. OBJECT
columns are NOT handled here
- /// (callers serialize them via the owning aggregation function). Used by
the group-by result
- /// serialization on both the server (`GroupByResultsBlock`) and the
merge-only reduce path.
+ /// Writes a non-null value of the given stored column data type into the
[DataTableBuilder] at the given column.
+ ///
+ /// Supports all scalar and array stored types. `OBJECT`, `UNKNOWN` and
`MAP` are NOT handled here -- they are
+ /// written through the builder's own type-specific methods, since their
values are either serialized by an owning
+ /// aggregation function or already carry an in-band null encoding.
public static void setColumn(DataTableBuilder dataTableBuilder,
ColumnDataType storedColumnDataType,
int columnIndex, Object value)
throws IOException {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderV4.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderV4.java
index df18e73dc4b..f4da8848140 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderV4.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/common/datatable/DataTableBuilderV4.java
@@ -21,14 +21,12 @@ package org.apache.pinot.core.common.datatable;
import it.unimi.dsi.fastutil.objects.Object2IntMap;
import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import java.io.IOException;
-import javax.annotation.Nullable;
+import java.io.UncheckedIOException;
import org.apache.pinot.common.datatable.DataTable;
import org.apache.pinot.common.datatable.DataTableFactory;
import org.apache.pinot.common.datatable.DataTableImplV4;
import org.apache.pinot.common.utils.DataSchema;
-import org.apache.pinot.common.utils.RoaringBitmapUtils;
import org.apache.pinot.spi.utils.ByteArray;
-import org.roaringbitmap.RoaringBitmap;
public class DataTableBuilderV4 extends BaseDataTableBuilder {
@@ -67,21 +65,14 @@ public class DataTableBuilderV4 extends
BaseDataTableBuilder {
}
}
- @Override
- public void setNullRowIds(@Nullable RoaringBitmap nullRowIds)
- throws IOException {
-
_fixedSizeDataOutputStream.writeInt(_variableSizeDataByteArrayOutputStream.size());
- if (nullRowIds == null || nullRowIds.isEmpty()) {
- _fixedSizeDataOutputStream.writeInt(0);
- } else {
- byte[] bitmapBytes = RoaringBitmapUtils.serialize(nullRowIds);
- _fixedSizeDataOutputStream.writeInt(bitmapBytes.length);
- _variableSizeDataByteArrayOutputStream.write(bitmapBytes);
- }
- }
-
@Override
public DataTable build() {
+ try {
+ writeNullBitmaps();
+ } catch (IOException e) {
+ // Both buffers are in-memory byte arrays, so this cannot happen.
+ throw new UncheckedIOException(e);
+ }
String[] reverseDictionary = new String[_dictionary.size()];
for (Object2IntMap.Entry<String> entry : _dictionary.object2IntEntrySet())
{
reverseDictionary[entry.getIntValue()] = entry.getKey();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
index 82d43759546..d17a14b2772 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
@@ -107,7 +107,7 @@ public class TableResizer {
}
/// Grouping-set queries produce genuine null group-key values for
rolled-up columns regardless of the
/// user's null-handling option, so ORDER BY over a grouping column must
use the null-safe comparator.
- boolean nullHandlingEnabled =
queryContext.requiresNullAwareKeySerialization();
+ boolean nullHandlingEnabled =
queryContext.requiresNullAwareKeyEvaluation();
if (nullHandlingEnabled) {
_intermediateRecordComparator = (o1, o2) -> {
for (int i = 0; i < _numOrderByExpressions; i++) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
index 7e9ace4afdd..141367beb3e 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java
@@ -38,7 +38,6 @@ import
org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.spi.utils.ByteArray;
-import org.roaringbitmap.RoaringBitmap;
/// Results block for aggregation queries.
@@ -124,73 +123,32 @@ public class AggregationResultsBlock extends
BaseResultsBlock {
return dataTableBuilder.build();
}
- boolean returnFinalResult = _queryContext.isServerReturnFinalResult();
- if (_queryContext.isNullHandlingEnabled()) {
- RoaringBitmap[] nullBitmaps = new RoaringBitmap[numColumns];
+ // NOTE: Nulls are serialized through the builder's null-aware path
regardless of the query's null-handling
+ // option. Aggregation functions whose accumulator has no identity element
(MINSTRING, MAXSTRING, ANYVALUE)
+ // return a null intermediate result in both modes, and their result
column type is not OBJECT.
+ dataTableBuilder.startRow();
+ if (_queryContext.isServerReturnFinalResult()) {
for (int i = 0; i < numColumns; i++) {
- nullBitmaps[i] = new RoaringBitmap();
- }
- dataTableBuilder.startRow();
- if (returnFinalResult) {
- for (int i = 0; i < numColumns; i++) {
- Object result =
_aggregationFunctions[i].extractFinalResult(_results.get(i));
- if (result == null) {
- result = columnDataTypes[i].getNullPlaceholder();
- nullBitmaps[i].add(0);
- }
- assert result != null;
+ Object result =
_aggregationFunctions[i].extractFinalResult(_results.get(i));
+ if (result == null) {
+ dataTableBuilder.setNull(i);
+ } else {
setFinalResult(dataTableBuilder, columnDataTypes, i, result);
}
- } else {
- for (int i = 0; i < numColumns; i++) {
- Object result = _results.get(i);
- if (columnDataTypes[i] == ColumnDataType.OBJECT) {
- if (result == null) {
- dataTableBuilder.setNull(i);
- } else {
- dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
- }
- } else {
- if (result == null) {
- result = columnDataTypes[i].getNullPlaceholder();
- nullBitmaps[i].add(0);
- }
- assert result != null;
- AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
- }
- }
- }
- dataTableBuilder.finishRow();
- for (RoaringBitmap nullBitmap : nullBitmaps) {
- dataTableBuilder.setNullRowIds(nullBitmap);
}
} else {
- dataTableBuilder.startRow();
- if (returnFinalResult) {
- for (int i = 0; i < numColumns; i++) {
- Object result =
_aggregationFunctions[i].extractFinalResult(_results.get(i));
- if (result == null) {
- dataTableBuilder.setNull(i);
- } else {
- setFinalResult(dataTableBuilder, columnDataTypes, i, result);
- }
- }
- } else {
- for (int i = 0; i < numColumns; i++) {
- Object result = _results.get(i);
- if (result == null) {
- dataTableBuilder.setNull(i);
- } else {
- if (columnDataTypes[i] == ColumnDataType.OBJECT) {
- dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
- } else {
- AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
- }
- }
+ for (int i = 0; i < numColumns; i++) {
+ Object result = _results.get(i);
+ if (result == null) {
+ dataTableBuilder.setNull(i);
+ } else if (columnDataTypes[i] == ColumnDataType.OBJECT) {
+ dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
+ } else {
+ AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
}
}
- dataTableBuilder.finishRow();
}
+ dataTableBuilder.finishRow();
return dataTableBuilder.build();
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
index eecffa29433..1ffc5b8ee93 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/GroupByResultsBlock.java
@@ -38,7 +38,6 @@ import
org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import
org.apache.pinot.core.query.aggregation.groupby.AggregationGroupByResult;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.roaringbitmap.RoaringBitmap;
/// Results block for group-by queries.
@@ -201,66 +200,26 @@ public class GroupByResultsBlock extends BaseResultsBlock
{
int numKeyColumns = _queryContext.getNumGroupByKeyColumns();
Iterator<Record> iterator = _table.iterator();
int numRowsAdded = 0;
- /// Grouping sets produce NULL group keys (rolled-up columns) regardless
of the user's null-handling
- /// option, so they must be serialized through the null-aware path (null
bitmaps) to survive the DataTable
- /// round-trip -- the non-null path's setNull() writes to the variable
buffer that the reducer does not
- /// read back for fixed-width key columns.
- if (_queryContext.requiresNullAwareKeySerialization()) {
- RoaringBitmap[] nullBitmaps = new RoaringBitmap[numColumns];
- Object[] nullPlaceholders = new Object[numColumns];
- for (int colId = 0; colId < numColumns; colId++) {
- nullBitmaps[colId] = new RoaringBitmap();
- // Resolved on the logical type, not the stored type: UUID overrides
getNullPlaceholder() to return the nil
- // UUID, whereas its stored type BYTES would yield a zero-length
placeholder that is not a valid UUID.
- nullPlaceholders[colId] =
_dataSchema.getColumnDataType(colId).getNullPlaceholder();
- }
- int rowId = 0;
- while (iterator.hasNext()) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numRowsAdded,
"GroupByResultsBlock#getDataTable");
- dataTableBuilder.startRow();
- Object[] values = iterator.next().getValues();
- for (int i = 0; i < numColumns; i++) {
- Object value = values[i];
- if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
- if (value == null) {
- dataTableBuilder.setNull(i);
- } else {
- dataTableBuilder.setColumn(i, aggregationFunctions[i -
numKeyColumns].serializeIntermediateResult(value));
- }
- } else {
- if (value == null) {
- value = nullPlaceholders[i];
- nullBitmaps[i].add(rowId);
- }
- assert value != null;
- DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
- }
- }
- dataTableBuilder.finishRow();
- numRowsAdded++;
- rowId++;
- }
- for (RoaringBitmap nullBitmap : nullBitmaps) {
- dataTableBuilder.setNullRowIds(nullBitmap);
- }
- } else {
- while (iterator.hasNext()) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numRowsAdded,
"GroupByResultsBlock#getDataTable");
- dataTableBuilder.startRow();
- Object[] values = iterator.next().getValues();
- for (int i = 0; i < numColumns; i++) {
- Object value = values[i];
- if (value == null) {
- dataTableBuilder.setNull(i);
- } else if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
- dataTableBuilder.setColumn(i, aggregationFunctions[i -
numKeyColumns].serializeIntermediateResult(value));
- } else {
- DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
- }
+ // NOTE: Nulls are serialized through the builder's null-aware path
regardless of the query's null-handling
+ // option. Grouping sets produce NULL group keys for rolled-up columns,
and aggregation functions whose
+ // accumulator has no identity element (MINSTRING, MAXSTRING, ANYVALUE)
produce null intermediate results, in
+ // both modes.
+ while (iterator.hasNext()) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numRowsAdded,
"GroupByResultsBlock#getDataTable");
+ dataTableBuilder.startRow();
+ Object[] values = iterator.next().getValues();
+ for (int i = 0; i < numColumns; i++) {
+ Object value = values[i];
+ if (value == null) {
+ dataTableBuilder.setNull(i);
+ } else if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
+ dataTableBuilder.setColumn(i, aggregationFunctions[i -
numKeyColumns].serializeIntermediateResult(value));
+ } else {
+ DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
}
- dataTableBuilder.finishRow();
- numRowsAdded++;
}
+ dataTableBuilder.finishRow();
+ numRowsAdded++;
}
return dataTableBuilder.build();
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
index 3802a1064c0..7ddceb00b15 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
@@ -102,14 +102,11 @@ public class AggregationDataTableReducer implements
DataTableReducer {
AggregationFunction aggregationFunction = _aggregationFunctions[i];
Object intermediateResultToMerge;
ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
- if (_queryContext.isNullHandlingEnabled()) {
- RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
- if (nullBitmap != null && nullBitmap.contains(0)) {
- intermediateResultToMerge = null;
- } else {
- intermediateResultToMerge =
-
AggregationFunctionUtils.getIntermediateResult(aggregationFunction, dataTable,
columnDataType, 0, i);
- }
+ // Nulls are restored regardless of the query's null-handling option:
an aggregation function whose
+ // accumulator has no identity element returns a null intermediate
result in both modes.
+ RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
+ if (nullBitmap != null && nullBitmap.contains(0)) {
+ intermediateResultToMerge = null;
} else {
intermediateResultToMerge =
AggregationFunctionUtils.getIntermediateResult(aggregationFunction, dataTable,
columnDataType, 0, i);
@@ -154,48 +151,18 @@ public class AggregationDataTableReducer implements
DataTableReducer {
ColumnDataType[] columnDataTypes = dataSchema.getColumnDataTypes();
int numColumns = columnDataTypes.length;
DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
- if (_queryContext.isNullHandlingEnabled()) {
- RoaringBitmap[] nullBitmaps = new RoaringBitmap[numColumns];
- for (int i = 0; i < numColumns; i++) {
- nullBitmaps[i] = new RoaringBitmap();
- }
- dataTableBuilder.startRow();
- for (int i = 0; i < numColumns; i++) {
- Object result = intermediateResults[i];
- if (columnDataTypes[i] == ColumnDataType.OBJECT) {
- if (result == null) {
- dataTableBuilder.setNull(i);
- } else {
- dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
- }
- } else {
- if (result == null) {
- result = columnDataTypes[i].getNullPlaceholder();
- nullBitmaps[i].add(0);
- }
- AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
- }
- }
- dataTableBuilder.finishRow();
- for (RoaringBitmap nullBitmap : nullBitmaps) {
- dataTableBuilder.setNullRowIds(nullBitmap);
- }
- } else {
- dataTableBuilder.startRow();
- for (int i = 0; i < numColumns; i++) {
- Object result = intermediateResults[i];
- if (result == null) {
- dataTableBuilder.setNull(i);
- } else {
- if (columnDataTypes[i] == ColumnDataType.OBJECT) {
- dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
- } else {
- AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
- }
- }
+ dataTableBuilder.startRow();
+ for (int i = 0; i < numColumns; i++) {
+ Object result = intermediateResults[i];
+ if (result == null) {
+ dataTableBuilder.setNull(i);
+ } else if (columnDataTypes[i] == ColumnDataType.OBJECT) {
+ dataTableBuilder.setColumn(i,
_aggregationFunctions[i].serializeIntermediateResult(result));
+ } else {
+ AggregationFunctionUtils.setIntermediateResult(dataTableBuilder,
columnDataTypes[i], i, result);
}
- dataTableBuilder.finishRow();
}
+ dataTableBuilder.finishRow();
return dataTableBuilder.build();
}
@@ -205,13 +172,9 @@ public class AggregationDataTableReducer implements
DataTableReducer {
Object[] finalResults = new Object[numAggregationFunctions];
for (int i = 0; i < numAggregationFunctions; i++) {
ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
- if (_queryContext.isNullHandlingEnabled()) {
- RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
- if (nullBitmap != null && nullBitmap.contains(0)) {
- finalResults[i] = null;
- } else {
- finalResults[i] =
AggregationFunctionUtils.getConvertedFinalResult(dataTable, columnDataType, 0,
i);
- }
+ RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
+ if (nullBitmap != null && nullBitmap.contains(0)) {
+ finalResults[i] = null;
} else {
finalResults[i] =
AggregationFunctionUtils.getConvertedFinalResult(dataTable, columnDataType, 0,
i);
}
@@ -228,13 +191,9 @@ public class AggregationDataTableReducer implements
DataTableReducer {
QueryThreadContext.checkTerminationAndSampleUsage("AggregationDataTableReducer");
Comparable finalResultToMerge;
ColumnDataType columnDataType = dataSchema.getColumnDataType(i);
- if (_queryContext.isNullHandlingEnabled()) {
- RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
- if (nullBitmap != null && nullBitmap.contains(0)) {
- finalResultToMerge = null;
- } else {
- finalResultToMerge =
AggregationFunctionUtils.getFinalResult(dataTable, columnDataType, 0, i);
- }
+ RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
+ if (nullBitmap != null && nullBitmap.contains(0)) {
+ finalResultToMerge = null;
} else {
finalResultToMerge =
AggregationFunctionUtils.getFinalResult(dataTable, columnDataType, 0, i);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
index 3a987a4b5fd..c1add8ca708 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java
@@ -206,7 +206,7 @@ public class GroupByDataTableReducer implements
DataTableReducer {
if (havingFilter != null) {
rows = new ArrayList<>();
HavingFilterHandler havingFilterHandler = new
HavingFilterHandler(havingFilter, postAggregationHandler,
- _queryContext.requiresNullAwareKeySerialization());
+ _queryContext.requiresNullAwareKeyEvaluation());
int processedRows = 0;
while (rows.size() < limit && sortedIterator.hasNext()) {
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(processedRows++,
"GroupByDataTableReducer");
@@ -293,15 +293,15 @@ public class GroupByDataTableReducer implements
DataTableReducer {
public void runJob() {
try {
for (DataTable dataTable : reduceGroup) {
- /// Grouping-set queries serialize NULL group keys via null
bitmaps regardless of the user's
- /// null-handling option, so their key nulls must be restored
here as well.
- boolean restoreNulls =
_queryContext.requiresNullAwareKeySerialization();
- RoaringBitmap[] nullBitmaps = null;
- if (restoreNulls) {
- nullBitmaps = new RoaringBitmap[_numColumns];
- for (int i = 0; i < _numColumns; i++) {
- nullBitmaps[i] = dataTable.getNullRowIds(i);
- }
+ // Nulls are restored regardless of the query's null-handling
option: a DataTable carries a null bitmap
+ // whenever the producer wrote a null, which happens in both
modes. Tables without nulls have no bitmap
+ // section at all, so getNullRowIds() returns null for every
column and the per-row restore is skipped.
+ RoaringBitmap[] nullBitmaps = new RoaringBitmap[_numColumns];
+ boolean restoreNulls = false;
+ for (int i = 0; i < _numColumns; i++) {
+ RoaringBitmap nullBitmap = dataTable.getNullRowIds(i);
+ nullBitmaps[i] = nullBitmap;
+ restoreNulls |= nullBitmap != null;
}
int numRows = dataTable.getNumberOfRows();
@@ -454,7 +454,7 @@ public class GroupByDataTableReducer implements
DataTableReducer {
if (havingFilter != null) {
rows = new ArrayList<>();
HavingFilterHandler havingFilterHandler = new
HavingFilterHandler(havingFilter, postAggregationHandler,
- _queryContext.requiresNullAwareKeySerialization());
+ _queryContext.requiresNullAwareKeyEvaluation());
for (int i = 0; i < numRows; i++) {
Object[] row = getConvertedRowWithFinalResult(dataTable, i);
if (havingFilterHandler.isMatch(row)) {
@@ -546,64 +546,25 @@ public class GroupByDataTableReducer implements
DataTableReducer {
DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
ColumnDataType[] storedColumnDataTypes =
dataSchema.getStoredColumnDataTypes();
Iterator<Record> iterator = indexedTable.iterator();
- /// Grouping-set queries carry NULL group keys that must be serialized via
the null-aware path even when
- /// the user did not enable null handling.
- if (_queryContext.requiresNullAwareKeySerialization()) {
- RoaringBitmap[] nullBitmaps = new RoaringBitmap[_numColumns];
- Object[] nullPlaceholders = new Object[_numColumns];
- for (int colId = 0; colId < _numColumns; colId++) {
- nullBitmaps[colId] = new RoaringBitmap();
- // Resolved on the logical type, not the stored type: UUID overrides
getNullPlaceholder() to return the nil
- // UUID, whereas its stored type BYTES would yield a zero-length
placeholder that is not a valid UUID.
- nullPlaceholders[colId] =
dataSchema.getColumnDataType(colId).getNullPlaceholder();
- }
- int rowId = 0;
- while (iterator.hasNext()) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rowId,
"GroupByDataTableReducer#merge");
- dataTableBuilder.startRow();
- Object[] values = iterator.next().getValues();
- for (int i = 0; i < _numColumns; i++) {
- Object value = values[i];
- if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
- if (value == null) {
- dataTableBuilder.setNull(i);
- } else {
- dataTableBuilder.setColumn(i,
- _aggregationFunctions[i -
_numKeyColumns].serializeIntermediateResult(value));
- }
- } else {
- if (value == null) {
- value = nullPlaceholders[i];
- nullBitmaps[i].add(rowId);
- }
- DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
- }
- }
- dataTableBuilder.finishRow();
- rowId++;
- }
- for (RoaringBitmap nullBitmap : nullBitmaps) {
- dataTableBuilder.setNullRowIds(nullBitmap);
- }
- } else {
- int rowId = 0;
- while (iterator.hasNext()) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rowId++,
"GroupByDataTableReducer#merge");
- dataTableBuilder.startRow();
- Object[] values = iterator.next().getValues();
- for (int i = 0; i < _numColumns; i++) {
- Object value = values[i];
- if (value == null) {
- dataTableBuilder.setNull(i);
- } else if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
- dataTableBuilder.setColumn(i,
- _aggregationFunctions[i -
_numKeyColumns].serializeIntermediateResult(value));
- } else {
- DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
- }
+ // NOTE: Nulls are serialized through the builder's null-aware path
regardless of the query's null-handling
+ // option -- grouping sets carry NULL group keys, and aggregation
functions whose accumulator has no identity
+ // element produce null intermediate results, in both modes.
+ int rowId = 0;
+ while (iterator.hasNext()) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rowId++,
"GroupByDataTableReducer#merge");
+ dataTableBuilder.startRow();
+ Object[] values = iterator.next().getValues();
+ for (int i = 0; i < _numColumns; i++) {
+ Object value = values[i];
+ if (value == null) {
+ dataTableBuilder.setNull(i);
+ } else if (storedColumnDataTypes[i] == ColumnDataType.OBJECT) {
+ dataTableBuilder.setColumn(i, _aggregationFunctions[i -
_numKeyColumns].serializeIntermediateResult(value));
+ } else {
+ DataTableBuilderUtils.setColumn(dataTableBuilder,
storedColumnDataTypes[i], i, value);
}
- dataTableBuilder.finishRow();
}
+ dataTableBuilder.finishRow();
}
return dataTableBuilder.build();
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
index b3fcd8f9b06..c167d752217 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
@@ -257,10 +257,14 @@ public class QueryContext {
return numGroupByExpressions + getNumExtraGroupByKeyColumns();
}
- /// Returns whether group-by key columns must be serialized/deserialized
through the null-aware path (null
- /// bitmaps). True when the user enabled null handling, or for grouping-set
queries, which produce NULL keys
+ /// Returns whether group-by key columns must be compared and filtered
null-aware, i.e. whether ORDER BY
+ /// comparators and HAVING predicates over group keys have to treat `null`
as a distinct value rather than as the
+ /// type's default. True when the user enabled null handling, or for
grouping-set queries, which produce NULL keys
/// for rolled-up columns regardless of the user's null-handling option.
- public boolean requiresNullAwareKeySerialization() {
+ ///
+ /// NOTE: This does not gate serialization --
[org.apache.pinot.core.common.datatable.DataTableBuilder] always
+ /// supports nulls, so a null key or intermediate result round-trips through
a DataTable in either mode.
+ public boolean requiresNullAwareKeyEvaluation() {
return isNullHandlingEnabled() || isGroupingSets();
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/common/datatable/DataTableSerDeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/common/datatable/DataTableSerDeTest.java
index 117359b48a9..91e46c537c9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/common/datatable/DataTableSerDeTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/common/datatable/DataTableSerDeTest.java
@@ -37,10 +37,11 @@ import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.utils.ByteArray;
import org.apache.pinot.spi.utils.UuidUtils;
import org.roaringbitmap.RoaringBitmap;
-import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.testng.Assert.*;
+
/// Unit test for [DataTable] serialization/de-serialization.
public class DataTableSerDeTest {
@@ -83,11 +84,11 @@ public class DataTableSerDeTest {
DataTable dataTable = DataTableBuilderFactory.getEmptyDataTable();
dataTable.addException(QueryErrorCode.QUERY_EXECUTION, expected);
DataTable newDataTable =
DataTableFactory.getDataTable(dataTable.toBytes());
- Assert.assertNull(newDataTable.getDataSchema());
- Assert.assertEquals(newDataTable.getNumberOfRows(), 0);
+ assertNull(newDataTable.getDataSchema());
+ assertEquals(newDataTable.getNumberOfRows(), 0);
String actual =
newDataTable.getExceptions().get(QueryErrorCode.QUERY_EXECUTION.getId());
- Assert.assertEquals(actual, expected);
+ assertEquals(actual, expected);
}
@Test(dataProvider = "versionProvider")
@@ -176,7 +177,7 @@ public class DataTableSerDeTest {
} else if (emptyValue instanceof ByteArray) {
dataTableBuilder.setColumn(columnId, (ByteArray) emptyValue);
} else {
- Assert.fail();
+ fail();
}
}
dataTableBuilder.finishRow();
@@ -184,8 +185,8 @@ public class DataTableSerDeTest {
DataTable dataTable = dataTableBuilder.build();
DataTable newDataTable =
DataTableFactory.getDataTable(dataTable.toBytes());
- Assert.assertEquals(newDataTable.getDataSchema(), dataSchema);
- Assert.assertEquals(newDataTable.getNumberOfRows(), numRows);
+ assertEquals(newDataTable.getDataSchema(), dataSchema);
+ assertEquals(newDataTable.getNumberOfRows(), numRows);
for (int rowId = 0; rowId < numRows; rowId++) {
for (int columnId = 0; columnId < dataSchema.size(); columnId++) {
@@ -217,7 +218,7 @@ public class DataTableSerDeTest {
entry = newDataTable.getCustomObject(rowId, columnId);
break;
}
- Assert.assertEquals(entry, emptyValues[columnId]);
+ assertEquals(entry, emptyValues[columnId]);
}
}
}
@@ -239,8 +240,8 @@ public class DataTableSerDeTest {
DataTable dataTable = dataTableBuilder.build();
DataTable newDataTable =
DataTableFactory.getDataTable(dataTable.toBytes());
- Assert.assertEquals(newDataTable.getDataSchema(), dataSchema,
ERROR_MESSAGE);
- Assert.assertEquals(newDataTable.getNumberOfRows(), NUM_ROWS,
ERROR_MESSAGE);
+ assertEquals(newDataTable.getDataSchema(), dataSchema, ERROR_MESSAGE);
+ assertEquals(newDataTable.getNumberOfRows(), NUM_ROWS, ERROR_MESSAGE);
verifyDataIsSame(newDataTable, columnDataTypes, numColumns);
}
@@ -259,15 +260,136 @@ public class DataTableSerDeTest {
fillDataTableWithRandomData(dataTableBuilder, columnDataType, 1,
numRows);
DataTable dataTable = dataTableBuilder.build();
DataTable newDataTable =
DataTableFactory.getDataTable(dataTable.toBytes());
- Assert.assertEquals(newDataTable.getDataSchema(), dataSchema,
ERROR_MESSAGE);
- Assert.assertEquals(newDataTable.getNumberOfRows(), numRows,
ERROR_MESSAGE);
+ assertEquals(newDataTable.getDataSchema(), dataSchema, ERROR_MESSAGE);
+ assertEquals(newDataTable.getNumberOfRows(), numRows, ERROR_MESSAGE);
verifyDataIsSame(newDataTable, columnDataType, 1, numRows);
}
}
}
+ /// [DataTableBuilder#setNull] must round-trip a null for every column type,
not just the ones that can encode a
+ /// null in-band. A type stored in a 4-byte slot (`INT` / `FLOAT` / `STRING`
/ `BOOLEAN`) is the sharpest case:
+ /// writing the 8-byte custom-object encoding into it overflows the row
buffer when the column is last, and
+ /// silently overwrites the following column otherwise.
+ @Test(dataProvider = "versionProvider")
+ public void testSetNullForAllDataTypes(int dataTableVersion)
+ throws IOException {
+ DataTableBuilderFactory.setDataTableVersion(dataTableVersion);
+ for (DataSchema.ColumnDataType columnDataType :
DataSchema.ColumnDataType.values()) {
+ DataSchema dataSchema =
+ new DataSchema(new String[]{columnDataType.name()}, new
DataSchema.ColumnDataType[]{columnDataType});
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ dataTableBuilder.startRow();
+ dataTableBuilder.setNull(0);
+ dataTableBuilder.finishRow();
+ DataTable dataTable =
DataTableFactory.getDataTable(dataTableBuilder.build().toBytes());
+
+ String message = ERROR_MESSAGE + ", type: " + columnDataType;
+ assertEquals(dataTable.getNumberOfRows(), 1, message);
+ switch (columnDataType.getStoredType()) {
+ // Types that carry their own in-band null encoding need no bitmap
entry.
+ case OBJECT:
+ case UNKNOWN:
+ assertNull(dataTable.getCustomObject(0, 0), message);
+ assertNull(dataTable.getNullRowIds(0), message);
+ break;
+ case MAP:
+ assertNull(dataTable.getMap(0, 0), message);
+ assertNull(dataTable.getNullRowIds(0), message);
+ break;
+ default:
+ RoaringBitmap nullRowIds = dataTable.getNullRowIds(0);
+ assertNotNull(nullRowIds, message);
+ assertTrue(nullRowIds.contains(0), message);
+ break;
+ }
+ }
+ }
+
+ /// A null in a 4-byte column must not bleed into the column that follows it.
+ @Test(dataProvider = "versionProvider")
+ public void testSetNullDoesNotCorruptAdjacentColumn(int dataTableVersion)
+ throws IOException {
+ DataTableBuilderFactory.setDataTableVersion(dataTableVersion);
+ DataSchema dataSchema = new DataSchema(new String[]{"str", "dbl", "int"},
new DataSchema.ColumnDataType[]{
+ DataSchema.ColumnDataType.STRING, DataSchema.ColumnDataType.DOUBLE,
DataSchema.ColumnDataType.INT
+ });
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ dataTableBuilder.startRow();
+ dataTableBuilder.setNull(0);
+ dataTableBuilder.setColumn(1, 3.5d);
+ dataTableBuilder.setColumn(2, 42);
+ dataTableBuilder.finishRow();
+ DataTable dataTable =
DataTableFactory.getDataTable(dataTableBuilder.build().toBytes());
+
+ assertEquals(dataTable.getDouble(0, 1), 3.5d, ERROR_MESSAGE);
+ assertEquals(dataTable.getInt(0, 2), 42, ERROR_MESSAGE);
+ assertEquals(dataTable.getNullRowIds(0), RoaringBitmap.bitmapOf(0),
ERROR_MESSAGE);
+ assertNull(dataTable.getNullRowIds(1), ERROR_MESSAGE);
+ assertNull(dataTable.getNullRowIds(2), ERROR_MESSAGE);
+ }
+
+ /// A table without nulls must not grow a null bitmap section, so that its
bytes stay identical to what a writer
+ /// that predates always-on null support would have produced.
+ @Test(dataProvider = "versionProvider")
+ public void testNoNullBitmapSectionWithoutNulls(int dataTableVersion)
+ throws IOException {
+ DataTableBuilderFactory.setDataTableVersion(dataTableVersion);
+ DataSchema dataSchema = new DataSchema(new String[]{"str", "int"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.STRING,
DataSchema.ColumnDataType.INT});
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ dataTableBuilder.startRow();
+ dataTableBuilder.setColumn(0, "foo");
+ dataTableBuilder.setColumn(1, 42);
+ dataTableBuilder.finishRow();
+ // An all-empty bitmap carries no information and must not force the
section to be emitted either.
+ dataTableBuilder.setNullRowIds(new RoaringBitmap());
+ dataTableBuilder.setNullRowIds(null);
+ byte[] bytes = dataTableBuilder.build().toBytes();
+
+ DataTableBuilder controlBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ controlBuilder.startRow();
+ controlBuilder.setColumn(0, "foo");
+ controlBuilder.setColumn(1, 42);
+ controlBuilder.finishRow();
+ assertEquals(bytes, controlBuilder.build().toBytes(), ERROR_MESSAGE);
+
+ DataTable dataTable = DataTableFactory.getDataTable(bytes);
+ assertNull(dataTable.getNullRowIds(0), ERROR_MESSAGE);
+ assertNull(dataTable.getNullRowIds(1), ERROR_MESSAGE);
+ }
+
+ /// A pre-computed bitmap handed to [DataTableBuilder#setNullRowIds] must
merge with the per-cell nulls recorded
+ /// by [DataTableBuilder#setNull] for the same column rather than replacing
them.
+ @Test(dataProvider = "versionProvider")
+ public void testSetNullRowIdsMergesWithSetNull(int dataTableVersion)
+ throws IOException {
+ DataTableBuilderFactory.setDataTableVersion(dataTableVersion);
+ DataSchema dataSchema =
+ new DataSchema(new String[]{"int"}, new
DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.INT});
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(dataSchema);
+ for (int rowId = 0; rowId < 3; rowId++) {
+ dataTableBuilder.startRow();
+ if (rowId == 0) {
+ dataTableBuilder.setNull(0);
+ } else {
+ dataTableBuilder.setColumn(0, rowId);
+ }
+ dataTableBuilder.finishRow();
+ }
+ RoaringBitmap preComputed = RoaringBitmap.bitmapOf(2);
+ dataTableBuilder.setNullRowIds(preComputed);
+ DataTable dataTable =
DataTableFactory.getDataTable(dataTableBuilder.build().toBytes());
+
+ assertEquals(dataTable.getNullRowIds(0), RoaringBitmap.bitmapOf(0, 2),
ERROR_MESSAGE);
+ assertEquals(dataTable.getInt(1, 0), 1, ERROR_MESSAGE);
+ // The builder must not take ownership of the caller's bitmap.
+ assertEquals(preComputed, RoaringBitmap.bitmapOf(2), ERROR_MESSAGE);
+ }
+
@Test(dataProvider = "versionProvider")
- public void testThreadCPUMemMeasurement(int dataTableVersion) throws
IOException {
+ public void testThreadCPUMemMeasurement(int dataTableVersion)
+ throws IOException {
DataTableBuilderFactory.setDataTableVersion(dataTableVersion);
DataSchema.ColumnDataType[] columnDataTypes =
DataSchema.ColumnDataType.values();
int numColumns = columnDataTypes.length;
@@ -286,12 +408,11 @@ public class DataTableSerDeTest {
DataTable dataTable = dataTableBuilder.build();
DataTable newDataTable =
DataTableFactory.getDataTable(dataTable.toBytes());
// When ThreadCpuTimeMeasurement is enabled,
responseSerializationCpuTimeNs should be positive.
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_CPU_TIME_NS.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_MEM_ALLOCATED_BYTES.getName()));
- Assert.assertTrue(
-
Integer.parseInt(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_CPU_TIME_NS.getName()))
> 0);
- Assert.assertTrue(
+
assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_CPU_TIME_NS.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_MEM_ALLOCATED_BYTES.getName()));
+
assertTrue(Integer.parseInt(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_CPU_TIME_NS.getName()))
> 0);
+ assertTrue(
Integer.parseInt(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_MEM_ALLOCATED_BYTES.getName()))
> 0);
// Disable ThreadCpuTimeMeasurement, serialize/de-serialize data table.
@@ -300,11 +421,11 @@ public class DataTableSerDeTest {
dataTable = dataTableBuilder.build();
newDataTable = DataTableFactory.getDataTable(dataTable.toBytes());
// When measurement is disabled, response serialization metadata is absent.
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_CPU_TIME_NS.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_CPU_TIME_NS.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_MEM_ALLOCATED_BYTES.getName()));
-
Assert.assertNull(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_MEM_ALLOCATED_BYTES.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_CPU_TIME_NS.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_CPU_TIME_NS.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.THREAD_MEM_ALLOCATED_BYTES.getName()));
+
assertNull(newDataTable.getMetadata().get(MetadataKey.RESPONSE_SER_MEM_ALLOCATED_BYTES.getName()));
}
private void fillDataTableWithRandomData(DataTableBuilder dataTableBuilder,
@@ -370,8 +491,8 @@ public class DataTableSerDeTest {
dataTableBuilder.setColumn(colId, new ByteArray(BYTES[rowId]));
break;
case UUID:
- UUIDS[rowId] = isNull ? UuidUtils.nullUuidBytes()
- : UuidUtils.toBytes(new UUID(RANDOM.nextLong(),
RANDOM.nextLong()));
+ UUIDS[rowId] =
+ isNull ? UuidUtils.nullUuidBytes() : UuidUtils.toBytes(new
UUID(RANDOM.nextLong(), RANDOM.nextLong()));
dataTableBuilder.setColumn(colId, new ByteArray(UUIDS[rowId]));
break;
case INT_ARRAY:
@@ -505,87 +626,79 @@ public class DataTableSerDeTest {
boolean isNull = nullBitmaps[colId] != null &&
nullBitmaps[colId].contains(rowId);
switch (columnDataTypes[colId]) {
case INT:
- Assert.assertEquals(newDataTable.getInt(rowId, colId), isNull ? 0
: INTS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getInt(rowId, colId), isNull ? 0 :
INTS[rowId], ERROR_MESSAGE);
break;
case LONG:
- Assert.assertEquals(newDataTable.getLong(rowId, colId), isNull ? 0
: LONGS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getLong(rowId, colId), isNull ? 0 :
LONGS[rowId], ERROR_MESSAGE);
break;
case FLOAT:
- Assert.assertEquals(newDataTable.getFloat(rowId, colId), isNull ?
0 : FLOATS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getFloat(rowId, colId), isNull ? 0 :
FLOATS[rowId], ERROR_MESSAGE);
break;
case DOUBLE:
- Assert.assertEquals(newDataTable.getDouble(rowId, colId), isNull ?
0.0 : DOUBLES[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getDouble(rowId, colId), isNull ? 0.0 :
DOUBLES[rowId], ERROR_MESSAGE);
break;
case BIG_DECIMAL:
- Assert.assertEquals(newDataTable.getBigDecimal(rowId, colId),
- isNull ? BigDecimal.ZERO : BIG_DECIMALS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getBigDecimal(rowId, colId), isNull ?
BigDecimal.ZERO : BIG_DECIMALS[rowId],
+ ERROR_MESSAGE);
break;
case BOOLEAN:
- Assert.assertEquals(newDataTable.getInt(rowId, colId), isNull ? 0
: BOOLEANS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getInt(rowId, colId), isNull ? 0 :
BOOLEANS[rowId], ERROR_MESSAGE);
break;
case TIMESTAMP:
- Assert.assertEquals(newDataTable.getLong(rowId, colId), isNull ? 0
: TIMESTAMPS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getLong(rowId, colId), isNull ? 0 :
TIMESTAMPS[rowId], ERROR_MESSAGE);
break;
case STRING:
- Assert.assertEquals(newDataTable.getString(rowId, colId), isNull ?
"" : STRINGS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getString(rowId, colId), isNull ? "" :
STRINGS[rowId], ERROR_MESSAGE);
break;
case JSON:
- Assert.assertEquals(newDataTable.getString(rowId, colId), isNull ?
"" : JSONS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getString(rowId, colId), isNull ? "" :
JSONS[rowId], ERROR_MESSAGE);
break;
case BYTES:
- Assert.assertEquals(newDataTable.getBytes(rowId,
colId).getBytes(), isNull ? new byte[0] : BYTES[rowId],
+ assertEquals(newDataTable.getBytes(rowId, colId).getBytes(),
isNull ? new byte[0] : BYTES[rowId],
ERROR_MESSAGE);
break;
case UUID:
- Assert.assertEquals(newDataTable.getBytes(rowId, colId).getBytes(),
+ assertEquals(newDataTable.getBytes(rowId, colId).getBytes(),
isNull ? UuidUtils.nullUuidBytes() : UUIDS[rowId],
ERROR_MESSAGE);
break;
case INT_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getIntArray(rowId,
colId), INT_ARRAYS[rowId]), ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getIntArray(rowId, colId),
INT_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case LONG_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getLongArray(rowId,
colId), LONG_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getLongArray(rowId, colId),
LONG_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case FLOAT_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getFloatArray(rowId,
colId), FLOAT_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getFloatArray(rowId, colId),
FLOAT_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case DOUBLE_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getDoubleArray(rowId,
colId), DOUBLE_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getDoubleArray(rowId,
colId), DOUBLE_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case BIG_DECIMAL_ARRAY:
-
Assert.assertTrue(Arrays.equals(newDataTable.getBigDecimalArray(rowId, colId),
BIG_DECIMAL_ARRAYS[rowId]),
+ assertTrue(Arrays.equals(newDataTable.getBigDecimalArray(rowId,
colId), BIG_DECIMAL_ARRAYS[rowId]),
ERROR_MESSAGE);
break;
case BOOLEAN_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getIntArray(rowId,
colId), BOOLEAN_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getIntArray(rowId, colId),
BOOLEAN_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case TIMESTAMP_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getLongArray(rowId,
colId), TIMESTAMP_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getLongArray(rowId, colId),
TIMESTAMP_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case BYTES_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getBytesArray(rowId,
colId), BYTES_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getBytesArray(rowId, colId),
BYTES_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case UUID_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getBytesArray(rowId,
colId), UUID_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getBytesArray(rowId, colId),
UUID_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case STRING_ARRAY:
- Assert.assertTrue(Arrays.equals(newDataTable.getStringArray(rowId,
colId), STRING_ARRAYS[rowId]),
- ERROR_MESSAGE);
+ assertTrue(Arrays.equals(newDataTable.getStringArray(rowId,
colId), STRING_ARRAYS[rowId]), ERROR_MESSAGE);
break;
case MAP:
- Assert.assertEquals(newDataTable.getMap(rowId, colId),
MAPS[rowId], ERROR_MESSAGE);
+ assertEquals(newDataTable.getMap(rowId, colId), MAPS[rowId],
ERROR_MESSAGE);
break;
case OBJECT:
case UNKNOWN:
Object nulValue = newDataTable.getCustomObject(rowId, colId);
- Assert.assertNull(nulValue, ERROR_MESSAGE);
+ assertNull(nulValue, ERROR_MESSAGE);
break;
default:
throw new UnsupportedOperationException("Unable to generate random
data for: " + columnDataTypes[colId]);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/MaxStringAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/MaxStringAggregationFunctionTest.java
index 49666e554b0..459f548ae42 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/MaxStringAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/MaxStringAggregationFunctionTest.java
@@ -31,6 +31,7 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.exception.BadQueryRequestException;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
@@ -47,7 +48,13 @@ public class MaxStringAggregationFunctionTest extends
AbstractAggregationFunctio
/// This is used to simulate the DataTypeScenario concept from numeric
aggregation tests,
/// but fixed for the STRING data type.
protected FluentQueryTest.DeclaringTable getDeclaringTable(boolean
enableColumnBasedNullHandling) {
+ return getDeclaringTable(enableColumnBasedNullHandling, Map.of());
+ }
+
+ protected FluentQueryTest.DeclaringTable getDeclaringTable(boolean
enableColumnBasedNullHandling,
+ Map<String, String> extraQueryOptions) {
return FluentQueryTest.withBaseDir(_baseDir)
+ .withExtraQueryOptions(extraQueryOptions)
.givenTable(
new Schema.SchemaBuilder()
.setSchemaName("testTable")
@@ -307,4 +314,67 @@ public class MaxStringAggregationFunctionTest extends
AbstractAggregationFunctio
"tag3 | cherry" // Values for tag3: "cherry". Max is "cherry".
);
}
+
+ /// MAXSTRING has no identity element, so it yields a null intermediate
result when no row is aggregated at all --
+ /// in both null-handling modes. The intermediate result column is STRING,
whose 4-byte DataTable slot cannot hold
+ /// the custom-object null encoding, so this only round-trips if the builder
writes the null through a bitmap.
+ @Test
+ void aggregationNoMatchingRowsWithNullHandlingDisabled() {
+ getDeclaringTable(false)
+ .onFirstInstance("myField",
+ "alpha"
+ ).andOnSecondInstance("myField",
+ "beta"
+ ).whenQuery("select maxstring(myField) from testTable where myField =
'nomatch'")
+ .thenResultIs("STRING", "null");
+ }
+
+ @Test
+ void aggregationNoMatchingRowsWithNullHandlingEnabled() {
+ getDeclaringTable(true)
+ .onFirstInstance("myField",
+ "alpha"
+ ).andOnSecondInstance("myField",
+ "beta"
+ ).whenQueryWithNullHandlingEnabled("select maxstring(myField) from
testTable where myField = 'nomatch'")
+ .thenResultIs("STRING", "null");
+ }
+
+ /// The null intermediate result must survive alongside a neighbouring
aggregate: the STRING column occupies a
+ /// 4-byte slot, so an over-wide null encoding would corrupt the column that
follows it.
+ @Test
+ void aggregationNoMatchingRowsWithFollowingAggregateColumn() {
+ getDeclaringTable(false)
+ .onFirstInstance("myField",
+ "alpha"
+ ).andOnSecondInstance("myField",
+ "beta"
+ ).whenQuery("select maxstring(myField), count(*) from testTable where
myField = 'nomatch'")
+ .thenResultIs("STRING | LONG", "null | 0");
+ }
+
+ /// With `serverReturnFinalResult` the servers finalize before serializing,
so the null lands in a column typed by
+ /// `getFinalResultColumnType()` rather than the intermediate type -- STRING
here, a 4-byte slot either way. The
+ /// broker sets this option on its own for any single-server query, so this
is a routine path, not an edge case.
+ @Test
+ void aggregationNoMatchingRowsReturningFinalResultWithNullHandlingDisabled()
{
+ getDeclaringTable(false, Map.of(QueryOptionKey.SERVER_RETURN_FINAL_RESULT,
"true"))
+ .onFirstInstance("myField",
+ "alpha"
+ ).andOnSecondInstance("myField",
+ "beta"
+ ).whenQuery("select maxstring(myField) from testTable where myField =
'nomatch'")
+ .thenResultIs("STRING", "null");
+ }
+
+ @Test
+ void aggregationNoMatchingRowsReturningFinalResultWithNullHandlingEnabled() {
+ getDeclaringTable(true, Map.of(QueryOptionKey.SERVER_RETURN_FINAL_RESULT,
"true"))
+ .onFirstInstance("myField",
+ "alpha"
+ ).andOnSecondInstance("myField",
+ "beta"
+ ).whenQueryWithNullHandlingEnabled("select maxstring(myField) from
testTable where myField = 'nomatch'")
+ .thenResultIs("STRING", "null");
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/MergeDataTablesOnlyTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/MergeDataTablesOnlyTest.java
index 3efa9c0cb5b..6c5c5f2b2c4 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/MergeDataTablesOnlyTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/MergeDataTablesOnlyTest.java
@@ -269,6 +269,33 @@ public class MergeDataTablesOnlyTest {
assertRoundTrip(brokerRequest, serverTables);
}
+ /// A null intermediate result reaches the merge-only path with null
handling disabled too, so the merged
+ /// DataTable has to carry it.
+ ///
+ /// The aggregate has to be `MAXSTRING`: its accumulator has no identity
element, so a server that aggregated no
+ /// rows contributes null in either mode, and its `merge` is null-safe.
`MIN` / `SUM` cannot reach this state --
+ /// their accumulators are primitive and always hold an identity value -- so
a nulled DOUBLE fixture would be an
+ /// input no server produces, and `MinAggregationFunction#merge` would NPE
on it since it only null-guards when
+ /// null handling is enabled.
+ @Test
+ public void testNullAggregationRoundTripWithNullHandlingDisabled()
+ throws IOException {
+ String query = "SELECT MAXSTRING(s) FROM testTable";
+ DataSchema schema = new DataSchema(new String[]{"maxstring(s)"}, new
ColumnDataType[]{ColumnDataType.STRING});
+ // One server aggregated rows, the other matched none.
+ assertRoundTrip(query, List.of(buildNullableString(schema, "alpha"),
buildNullableString(schema, null)));
+ }
+
+ /// When every server contributes null, the merged intermediate DataTable
itself holds a null, so the null has to
+ /// survive `buildIntermediateDataTable`'s write as well as the restore on
the way back in.
+ @Test
+ public void testAllNullAggregationRoundTripWithNullHandlingDisabled()
+ throws IOException {
+ String query = "SELECT MAXSTRING(s) FROM testTable";
+ DataSchema schema = new DataSchema(new String[]{"maxstring(s)"}, new
ColumnDataType[]{ColumnDataType.STRING});
+ assertRoundTrip(query, List.of(buildNullableString(schema, null),
buildNullableString(schema, null)));
+ }
+
@Test
public void testConflictingSchemaSurfacedAsIncompleteMerge() {
// When one server's column types conflict with the first non-empty
table's, filterDataTablesAndPickSchema
@@ -574,6 +601,19 @@ public class MergeDataTablesOnlyTest {
return builder.build();
}
+ private static DataTable buildNullableString(DataSchema schema, String value)
+ throws IOException {
+ DataTableBuilder builder =
DataTableBuilderFactory.getDataTableBuilder(schema);
+ builder.startRow();
+ if (value == null) {
+ builder.setNull(0);
+ } else {
+ builder.setColumn(0, value);
+ }
+ builder.finishRow();
+ return builder.build();
+ }
+
private static void appendRow(DataTableBuilder builder, DataSchema schema,
Object[] values)
throws IOException {
builder.startRow();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
