public CompactedPinotSegmentRecordReader(File indexDir, RoaringBitmap validDocIds) {
    this(indexDir, validDocIds, null);
  }
@Test
  public void testCompactedPinotSegmentRecordReader()
      throws Exception {
    RoaringBitmap validDocIds = new RoaringBitmap();
    for (int i = 0; i < NUM_ROWS; i += 2) {
      validDocIds.add(i);
    }
    List<GenericRow> outputRows = new ArrayList<>();
    List<GenericRow> rewoundOuputRows = new ArrayList<>();

    CompactedPinotSegmentRecordReader compactedReader =
        new CompactedPinotSegmentRecordReader(_segmentIndexDir, validDocIds);
    while (compactedReader.hasNext()) {
      outputRows.add(compactedReader.next());
    }
    compactedReader.rewind();

    while (compactedReader.hasNext()) {
      rewoundOuputRows.add(compactedReader.next());
    }
    compactedReader.close();

    Assert.assertEquals(outputRows.size(), NUM_ROWS / 2,
        "Number of _rows returned by CompactedPinotSegmentRecordReader is incorrect");
    for (int i = 0; i < outputRows.size(); i++) {
      GenericRow outputRow = outputRows.get(i);
      GenericRow row = _rows.get(i * 2);
      Assert.assertEquals(outputRow.getValue(D_SV_1), row.getValue(D_SV_1));
      Assert.assertTrue(PinotSegmentUtil.compareMultiValueColumn(outputRow.getValue(D_MV_1), row.getValue(D_MV_1)));
      Assert.assertEquals(outputRow.getValue(M1), row.getValue(M1));
      Assert.assertEquals(outputRow.getValue(M2), row.getValue(M2));
      Assert.assertEquals(outputRow.getValue(TIME), row.getValue(TIME));
    }

    Assert.assertEquals(rewoundOuputRows.size(), NUM_ROWS / 2,
        "Number of _rows returned by CompactedPinotSegmentRecordReader is incorrect");
    for (int i = 0; i < rewoundOuputRows.size(); i++) {
      GenericRow outputRow = rewoundOuputRows.get(i);
      GenericRow row = _rows.get(i * 2);
      Assert.assertEquals(outputRow.getValue(D_SV_1), row.getValue(D_SV_1));
      Assert.assertTrue(PinotSegmentUtil.compareMultiValueColumn(outputRow.getValue(D_MV_1), row.getValue(D_MV_1)));
      Assert.assertEquals(outputRow.getValue(M1), row.getValue(M1));
      Assert.assertEquals(outputRow.getValue(M2), row.getValue(M2));
      Assert.assertEquals(outputRow.getValue(TIME), row.getValue(TIME));
    }
  }