@Nullable
  @Override
  protected SelectionResultsBlock getNextBlock() {
    if (_numDocsScanned >= _limit) {
      // Already returned enough documents
      return null;
    }
    ValueBlock valueBlock = _projectOperator.nextBlock();
    if (valueBlock == null) {
      return null;
    }
    int numExpressions = _expressions.size();
    for (int i = 0; i < numExpressions; i++) {
      _blockValSets[i] = valueBlock.getBlockValueSet(_expressions.get(i));
    }
    RowBasedBlockValueFetcher blockValueFetcher = new RowBasedBlockValueFetcher(_blockValSets);
    int numDocs = valueBlock.getNumDocs();
    int numDocsToReturn = Math.min(_limit - _numDocsScanned, numDocs);
    List<Object[]> rows = new ArrayList<>(numDocsToReturn);
    if (_nullHandlingEnabled) {
      for (int i = 0; i < numExpressions; i++) {
        _nullBitmaps[i] = _blockValSets[i].getNullBitmap();
      }
      for (int docId = 0; docId < numDocsToReturn; docId++) {
        Object[] values = blockValueFetcher.getRow(docId);
        for (int colId = 0; colId < numExpressions; colId++) {
          if (_nullBitmaps[colId] != null && _nullBitmaps[colId].contains(docId)) {
            values[colId] = null;
          }
        }
        rows.add(values);
      }
    } else {
      for (int i = 0; i < numDocsToReturn; i++) {
        rows.add(blockValueFetcher.getRow(i));
      }
    }
    _numDocsScanned += numDocs;
    return new SelectionResultsBlock(_dataSchema, rows, _queryContext);
  }
@Test
  public void testNullHandling() {
    QueryContext queryContext =
        QueryContextConverterUtils.getQueryContext("SELECT * FROM testTable WHERE intColumn IS NULL");
    queryContext.setNullHandlingEnabled(true);
    List<ExpressionContext> expressions =
        SelectionOperatorUtils.extractExpressions(queryContext, _segmentWithNullValues);
    BaseProjectOperator<?> projectOperator =
        new ProjectPlanNode(new SegmentContext(_segmentWithNullValues), queryContext, expressions,
            DocIdSetPlanNode.MAX_DOC_PER_CALL).run();

    StreamingSelectionOnlyOperator operator = new StreamingSelectionOnlyOperator(
        _segmentWithNullValues,
        queryContext,
        expressions,
        projectOperator
    );
    SelectionResultsBlock block = operator.getNextBlock();
    List<Object[]> rows = block.getRows();
    assertNull(rows.get(0)[0], "Column value should be 'null' when null handling is enabled");
  }