Presence vector

pinot-common/src/main/java/org/apache/pinot/common/config/IndexingConfig.java

@@ -87,6 +87,9 @@ public class IndexingConfig {
   @ConfigKey("aggregateMetrics")
   private boolean _aggregateMetrics;
 
+  @ConfigKey("nullHandlingEnabled")
+  private boolean _nullHandlingEnabled;
+
   /**
    * The list of columns for which the variable length dictionary needs to be enabled in offline
    * segments. This is only valid for string and bytes columns and has no impact for columns of
@@ -239,6 +242,14 @@ public void setVarLengthDictionaryColumns(List<String> varLengthDictionaryColumn
     _varLengthDictionaryColumns = varLengthDictionaryColumns;
   }
 
+  public boolean isNullHandlingEnabled() {
+    return _nullHandlingEnabled;
+  }
+
+  public void setNullHandlingEnabled(boolean nullHandlingEnabled) {
+    _nullHandlingEnabled = nullHandlingEnabled;
+  }
+
   @Override
   public String toString() {
     final StringBuilder result = new StringBuilder();
@@ -296,7 +307,9 @@ public boolean equals(Object o) {
         .isEqual(_starTreeIndexSpec, that._starTreeIndexSpec) && EqualityUtils
         .isEqual(_segmentPartitionConfig, that._segmentPartitionConfig) && EqualityUtils
         .isEqual(_bloomFilterColumns, that._bloomFilterColumns) && EqualityUtils
-        .isEqual(_varLengthDictionaryColumns, that._varLengthDictionaryColumns);
+        .isEqual(_varLengthDictionaryColumns, that._varLengthDictionaryColumns) && EqualityUtils
+        .isEqual(_aggregateMetrics, that._aggregateMetrics) && EqualityUtils
+        .isEqual(_nullHandlingEnabled, that._nullHandlingEnabled);
   }
 
   @Override
@@ -316,6 +329,8 @@ public int hashCode() {
     result = EqualityUtils.hashCodeOf(result, _segmentPartitionConfig);
     result = EqualityUtils.hashCodeOf(result, _bloomFilterColumns);
     result = EqualityUtils.hashCodeOf(result, _varLengthDictionaryColumns);
+    result = EqualityUtils.hashCodeOf(result, _aggregateMetrics);
+    result = EqualityUtils.hashCodeOf(result, _nullHandlingEnabled);
     return result;
   }
 }

pinot-common/src/main/java/org/apache/pinot/common/segment/SegmentMetadata.java

@@ -103,6 +103,8 @@ public interface SegmentMetadata {
 
   String getBloomFilterFileName(String column);
 
+  String getNullValueVectorFileName(String column);
+
   String getCreatorName();
 
   char getPaddingCharacter();

pinot-core/src/main/java/org/apache/pinot/core/common/DataSource.java

@@ -22,6 +22,7 @@
 import org.apache.pinot.core.segment.index.readers.BloomFilterReader;
 import org.apache.pinot.core.segment.index.readers.Dictionary;
 import org.apache.pinot.core.segment.index.readers.InvertedIndexReader;
+import org.apache.pinot.core.segment.index.readers.NullValueVectorReader;
 
 
 public abstract class DataSource extends BaseOperator {
@@ -45,4 +46,10 @@ public abstract class DataSource extends BaseOperator {
    * Returns the bloom filter for the data source if exists, or {@code null} if not.
    */
   public abstract BloomFilterReader getBloomFilter();
+
+  /**
+   * Returns null value vector for the data source if exists, or {@code null} if not.
+   */
+  public abstract NullValueVectorReader getNullValueVector();
+
 }

pinot-core/src/main/java/org/apache/pinot/core/data/GenericRow.java

@@ -90,6 +90,13 @@ public boolean isNullValue(String fieldName) {
     return _nullValueFields.contains(fieldName);
   }
 
+  /**
+   * Returns whether this row has null values for any of the columns
+   */
+  public boolean hasNullValues() {
+    return !_nullValueFields.isEmpty();
+  }
+
   /**
    * Sets the value for the given field.
    */

pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/HLRealtimeSegmentDataManager.java

@@ -188,7 +188,8 @@ public HLRealtimeSegmentDataManager(final RealtimeSegmentZKMetadata realtimeSegm
             getMemoryManager(realtimeTableDataManager.getConsumerDir(), segmentName,
                 indexLoadingConfig.isRealtimeOffheapAllocation(),
                 indexLoadingConfig.isDirectRealtimeOffheapAllocation(), serverMetrics))
-            .setStatsHistory(realtimeTableDataManager.getStatsHistory()).build();
+            .setStatsHistory(realtimeTableDataManager.getStatsHistory())
+            .setNullHandlingEnabled(indexingConfig.isNullHandlingEnabled()).build();
     realtimeSegment = new MutableSegmentImpl(realtimeSegmentConfig);
 
     notifier = realtimeTableDataManager;
@@ -272,7 +273,7 @@ public void run() {
               new RealtimeSegmentConverter(realtimeSegment, tempSegmentFolder.getAbsolutePath(), schema,
                   tableNameWithType, timeColumnName, realtimeSegmentZKMetadata.getSegmentName(), sortedColumn,
                   HLRealtimeSegmentDataManager.this.invertedIndexColumns, noDictionaryColumns,
-                  varLengthDictionaryColumns, null/*StarTreeIndexSpec*/); // Star tree not supported for HLC.
+                  varLengthDictionaryColumns, null/*StarTreeIndexSpec*/, indexingConfig.isNullHandlingEnabled()); // Star tree not supported for HLC.
 
           segmentLogger.info("Trying to build segment");
           final long buildStartTime = System.nanoTime();

pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java

@@ -259,6 +259,7 @@ public Map<String, File> getMetadataFiles() {
   private String _stopReason = null;
   private final Semaphore _segBuildSemaphore;
   private final boolean _isOffHeap;
+  private final boolean _nullHandlingEnabled;
 
   // TODO each time this method is called, we print reason for stop. Good to print only once.
   private boolean endCriteriaReached() {
@@ -687,7 +688,8 @@ protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) {
       RealtimeSegmentConverter converter =
           new RealtimeSegmentConverter(_realtimeSegment, tempSegmentFolder.getAbsolutePath(), _schema,
               _tableNameWithType, _timeColumnName, _segmentZKMetadata.getSegmentName(), _sortedColumn,
-              _invertedIndexColumns, _noDictionaryColumns, _varLengthDictionaryColumns, _starTreeIndexSpec);
+              _invertedIndexColumns, _noDictionaryColumns, _varLengthDictionaryColumns, _starTreeIndexSpec,
+              _nullHandlingEnabled);
       segmentLogger.info("Trying to build segment");
       try {
         converter.build(_segmentVersion, _serverMetrics);
@@ -1139,6 +1141,8 @@ public LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata,
 
     _isOffHeap = indexLoadingConfig.isRealtimeOffheapAllocation();
 
+    _nullHandlingEnabled = indexingConfig.isNullHandlingEnabled();
+
     // Start new realtime segment
     RealtimeSegmentConfig.Builder realtimeSegmentConfigBuilder =
         new RealtimeSegmentConfig.Builder().setSegmentName(_segmentNameStr).setStreamName(_streamTopic)
@@ -1149,14 +1153,15 @@ public LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata,
             .setInvertedIndexColumns(invertedIndexColumns).setRealtimeSegmentZKMetadata(segmentZKMetadata)
             .setOffHeap(_isOffHeap).setMemoryManager(_memoryManager)
             .setStatsHistory(realtimeTableDataManager.getStatsHistory())
-            .setAggregateMetrics(indexingConfig.isAggregateMetrics());
+            .setAggregateMetrics(indexingConfig.isAggregateMetrics())
+            .setNullHandlingEnabled(_nullHandlingEnabled);
 
     // Create message decoder
     _messageDecoder = StreamDecoderProvider.create(_partitionLevelStreamConfig, _schema);
     _clientId = _streamPartitionId + "-" + NetUtil.getHostnameOrAddress();
 
     // Create record transformer
-    _recordTransformer = CompositeTransformer.getDefaultTransformer(_schema);
+    _recordTransformer = CompositeTransformer.getDefaultTransformer(schema);
     makeStreamConsumer("Starting");
     makeStreamMetadataProvider("Starting");
 

pinot-core/src/main/java/org/apache/pinot/core/indexsegment/generator/SegmentGeneratorConfig.java

@@ -113,6 +113,7 @@ public enum TimeColumnType {
   // Use on-heap or off-heap memory to generate index (currently only affect inverted index and star-tree v2)
   private boolean _onHeap = false;
   private boolean _checkTimeColumnValidityDuringGeneration = true;
+  private boolean _nullHandlingEnabled = false;
 
   public SegmentGeneratorConfig() {
   }
@@ -162,6 +163,7 @@ public SegmentGeneratorConfig(SegmentGeneratorConfig config) {
     _onHeap = config._onHeap;
     _recordReaderPath = config._recordReaderPath;
     _checkTimeColumnValidityDuringGeneration = config._checkTimeColumnValidityDuringGeneration;
+    _nullHandlingEnabled = config._nullHandlingEnabled;
   }
 
   /**
@@ -218,6 +220,8 @@ public SegmentGeneratorConfig(@Nullable TableConfig tableConfig, Schema schema)
 
     SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
     _hllConfig = validationConfig.getHllConfig();
+
+    _nullHandlingEnabled = indexingConfig.isNullHandlingEnabled();
   }
 
   public SegmentGeneratorConfig(Schema schema) {
@@ -680,4 +684,12 @@ private String getQualifyingFields(FieldType type, boolean excludeVirtualColumns
     Collections.sort(fields);
     return StringUtils.join(fields, ",");
   }
+
+  public boolean isNullHandlingEnabled() {
+    return _nullHandlingEnabled;
+  }
+
+  public void setNullHandlingEnabled(boolean nullHandlingEnabled) {
+    _nullHandlingEnabled = nullHandlingEnabled;
+  }
 }

pinot-core/src/main/java/org/apache/pinot/core/indexsegment/mutable/MutableSegmentImpl.java

@@ -50,10 +50,12 @@
 import org.apache.pinot.core.realtime.impl.dictionary.BaseOffHeapMutableDictionary;
 import org.apache.pinot.core.realtime.impl.dictionary.MutableDictionaryFactory;
 import org.apache.pinot.core.realtime.impl.invertedindex.RealtimeInvertedIndexReader;
+import org.apache.pinot.core.realtime.impl.nullvalue.RealtimeNullValueVectorReaderWriter;
 import org.apache.pinot.core.segment.creator.impl.V1Constants;
 import org.apache.pinot.core.segment.index.SegmentMetadataImpl;
 import org.apache.pinot.core.segment.index.data.source.ColumnDataSource;
 import org.apache.pinot.core.segment.index.readers.BloomFilterReader;
+import org.apache.pinot.core.segment.index.readers.NullValueVectorReader;
 import org.apache.pinot.core.segment.virtualcolumn.VirtualColumnContext;
 import org.apache.pinot.core.segment.virtualcolumn.VirtualColumnProvider;
 import org.apache.pinot.core.segment.virtualcolumn.VirtualColumnProviderFactory;
@@ -86,12 +88,14 @@ public class MutableSegmentImpl implements MutableSegment {
   private final PinotDataBufferMemoryManager _memoryManager;
   private final RealtimeSegmentStatsHistory _statsHistory;
   private final SegmentPartitionConfig _segmentPartitionConfig;
+  private final boolean _nullHandlingEnabled;
 
   private final Map<String, BaseMutableDictionary> _dictionaryMap = new HashMap<>();
   private final Map<String, DataFileReader> _indexReaderWriterMap = new HashMap<>();
   private final Map<String, Integer> _maxNumValuesMap = new HashMap<>();
   private final Map<String, RealtimeInvertedIndexReader> _invertedIndexMap = new HashMap<>();
   private final Map<String, BloomFilterReader> _bloomFilterMap = new HashMap<>();
+  private final Map<String, RealtimeNullValueVectorReaderWriter> _nullValueVectorMap = new HashMap<>();
   private final IdMap<FixedIntArray> _recordIdMap;
   private boolean _aggregateMetrics;
 
@@ -142,6 +146,7 @@ public long getLatestIngestionTimestamp() {
     _memoryManager = config.getMemoryManager();
     _statsHistory = config.getStatsHistory();
     _segmentPartitionConfig = config.getSegmentPartitionConfig();
+    _nullHandlingEnabled = config.isNullHandlingEnabled();
 
     Collection<FieldSpec> allFieldSpecs = _schema.getAllFieldSpecs();
     List<FieldSpec> physicalFieldSpecs = new ArrayList<>(allFieldSpecs.size());
@@ -225,6 +230,10 @@ public long getLatestIngestionTimestamp() {
       if (invertedIndexColumns.contains(column)) {
         _invertedIndexMap.put(column, new RealtimeInvertedIndexReader());
       }
+
+      if (_nullHandlingEnabled) {
+        _nullValueVectorMap.put(column, new RealtimeNullValueVectorReaderWriter());
+      }
     }
 
     // Metric aggregation can be enabled only if config is specified, and all dimensions have dictionary,
@@ -262,6 +271,9 @@ public boolean index(GenericRow row, @Nullable RowMetadata rowMetadata) {
       // Add forward and inverted indices for new document.
       addForwardIndex(row, docId, dictIdMap);
       addInvertedIndex(docId, dictIdMap);
+      if (_nullHandlingEnabled) {
+        addNullValueVector(row, docId);
+      }
 
       // Update number of document indexed at last to make the latest record queryable
       canTakeMore = _numDocsIndexed++ < _capacity;
@@ -379,6 +391,22 @@ private void addInvertedIndex(int docId, Map<String, Object> dictIdMap) {
     }
   }
 
+  /**
+   * Check if the row has any null fields and update the
+   * column null value vectors accordingly
+   * @param row specifies row being ingested
+   * @param docId specified docId for this row
+   */
+  private void addNullValueVector(GenericRow row, int docId) {
+    if (!row.hasNullValues()) {
+      return;
+    }
+
+    for (String columnName : row.getNullValueFields()) {
+      _nullValueVectorMap.get(columnName).setNull(docId);
+    }
+  }
+
   private boolean aggregateMetrics(GenericRow row, int docId) {
     for (MetricFieldSpec metricFieldSpec : _physicalMetricFieldSpecs) {
       String column = metricFieldSpec.getName();
@@ -453,7 +481,7 @@ public ColumnDataSource getDataSource(String columnName) {
     } else {
       return new ColumnDataSource(fieldSpec, _numDocsIndexed, _maxNumValuesMap.get(columnName),
           _indexReaderWriterMap.get(columnName), _invertedIndexMap.get(columnName), _dictionaryMap.get(columnName),
-          _bloomFilterMap.get(columnName));
+          _bloomFilterMap.get(columnName), _nullValueVectorMap.get(columnName));
     }
   }
 
@@ -471,9 +499,18 @@ public List<StarTreeV2> getStarTrees() {
   public GenericRow getRecord(int docId, GenericRow reuse) {
     for (FieldSpec fieldSpec : _physicalFieldSpecs) {
       String column = fieldSpec.getName();
-      reuse.putField(column, IndexSegmentUtils
+      Object value = IndexSegmentUtils
           .getValue(docId, fieldSpec, _indexReaderWriterMap.get(column), _dictionaryMap.get(column),
-              _maxNumValuesMap.getOrDefault(column, 0)));
+              _maxNumValuesMap.getOrDefault(column, 0));
+      reuse.putValue(column, value);
+
+      if (_nullHandlingEnabled) {
+        NullValueVectorReader reader = _nullValueVectorMap.get(column);
+        // If column has null value for this docId, set that accordingly in GenericRow
+        if (reader != null && reader.isNull(docId)) {
+          reuse.putDefaultNullValue(column, value);
+        }
+      }
     }
     return reuse;
   }

pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/BitmapDocIdSet.java

@@ -18,8 +18,11 @@
  */
 package org.apache.pinot.core.operator.docidsets;
 
+import java.util.Arrays;
+import java.util.Iterator;
 import org.apache.pinot.core.common.BlockDocIdIterator;
 import org.apache.pinot.core.operator.dociditerators.BitmapDocIdIterator;
+import org.roaringbitmap.RoaringBitmap;
 import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
 import org.roaringbitmap.buffer.MutableRoaringBitmap;
 
@@ -33,7 +36,8 @@ public class BitmapDocIdSet implements FilterBlockDocIdSet {
   public BitmapDocIdSet(ImmutableRoaringBitmap[] bitmaps, int startDocId, int endDocId, boolean exclusive) {
     int numBitmaps = bitmaps.length;
     if (numBitmaps > 1) {
-      MutableRoaringBitmap orBitmap = MutableRoaringBitmap.or(bitmaps);
+      Iterator iterator = Arrays.asList(bitmaps).iterator();
+      MutableRoaringBitmap orBitmap = MutableRoaringBitmap.or(iterator, startDocId, endDocId + 1);
       if (exclusive) {
         orBitmap.flip(startDocId, endDocId + 1);
       }

pinot-core/src/main/java/org/apache/pinot/core/realtime/converter/RealtimeSegmentConverter.java

@@ -55,11 +55,12 @@ public class RealtimeSegmentConverter {
   private List<String> noDictionaryColumns;
   private StarTreeIndexSpec starTreeIndexSpec;
   private List<String> varLengthDictionaryColumns;
+  private final boolean _nullHandlingEnabled;
 
   public RealtimeSegmentConverter(MutableSegmentImpl realtimeSegment, String outputPath, Schema schema,
       String tableName, String timeColumnName, String segmentName, String sortedColumn,
       List<String> invertedIndexColumns, List<String> noDictionaryColumns,
-      List<String> varLengthDictionaryColumns, StarTreeIndexSpec starTreeIndexSpec) {
+      List<String> varLengthDictionaryColumns, StarTreeIndexSpec starTreeIndexSpec, boolean nullHandlingEnabled) {
     if (new File(outputPath).exists()) {
       throw new IllegalAccessError("path already exists:" + outputPath);
     }
@@ -77,12 +78,13 @@ public RealtimeSegmentConverter(MutableSegmentImpl realtimeSegment, String outpu
     this.noDictionaryColumns = noDictionaryColumns;
     this.varLengthDictionaryColumns = varLengthDictionaryColumns;
     this.starTreeIndexSpec = starTreeIndexSpec;
+    this._nullHandlingEnabled = nullHandlingEnabled;
   }
 
   public RealtimeSegmentConverter(MutableSegmentImpl realtimeSegment, String outputPath, Schema schema,
       String tableName, String timeColumnName, String segmentName, String sortedColumn) {
     this(realtimeSegment, outputPath, schema, tableName, timeColumnName, segmentName, sortedColumn, new ArrayList<>(),
-        new ArrayList<>(), new ArrayList<>(), null/*StarTreeIndexSpec*/);
+        new ArrayList<>(), new ArrayList<>(), null/*StarTreeIndexSpec*/, false/*nullHandlingEnabled*/);
   }
 
   public void build(@Nullable SegmentVersion segmentVersion, ServerMetrics serverMetrics)
@@ -135,6 +137,7 @@ public void build(@Nullable SegmentVersion segmentVersion, ServerMetrics serverM
     genConfig.setSegmentName(segmentName);
     SegmentPartitionConfig segmentPartitionConfig = realtimeSegmentImpl.getSegmentPartitionConfig();
     genConfig.setSegmentPartitionConfig(segmentPartitionConfig);
+    genConfig.setNullHandlingEnabled(_nullHandlingEnabled);
     final SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
     RealtimeSegmentSegmentCreationDataSource dataSource =
         new RealtimeSegmentSegmentCreationDataSource(realtimeSegmentImpl, reader, dataSchema);