public File purgeSegment()
      throws Exception {
    SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir);
    String segmentName = segmentMetadata.getName();
    String tableNameWithType = _tableConfig.getTableName();
    LOGGER.info("Start purging table: {}, segment: {}", tableNameWithType, segmentName);

    try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) {
      // Make a first pass through the data to see if records need to be purged or modified
      while (purgeRecordReader.hasNext()) {
        purgeRecordReader.next();
      }

      if (_numRecordsModified == 0 && _numRecordsPurged == 0) {
        // Returns null if no records to be modified or purged
        return null;
      }

      SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, _schema);
      config.setOutDir(_workingDir.getPath());
      config.setSegmentName(segmentName);

      // Keep index creation time the same as original segment because both segments use the same raw data.
      // This way, for REFRESH case, when new segment gets pushed to controller, we can use index creation time to
      // identify if the new pushed segment has newer data than the existing one.
      config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime()));

      // The time column type info is not stored in the segment metadata.
      // Keep segment start/end time to properly handle time column type other than EPOCH (e.g.SIMPLE_FORMAT).
      if (segmentMetadata.getTimeInterval() != null) {
        config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName());
        config.setStartTime(Long.toString(segmentMetadata.getStartTime()));
        config.setEndTime(Long.toString(segmentMetadata.getEndTime()));
        config.setSegmentTimeUnit(segmentMetadata.getTimeUnit());
      }

      SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
      purgeRecordReader.rewind();
      driver.init(config, purgeRecordReader);
      driver.build();
    }

    LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", tableNameWithType,
        segmentName, _numRecordsPurged, _numRecordsModified);

    return new File(_workingDir, segmentName);
  }
@Test
  public void testPurgeSegment()
      throws Exception {
    // Purge records with d1 = 0
    SegmentPurger.RecordPurger recordPurger = row -> row.getValue(D1).equals(0);

    // Modify records with d2 = 0 to d2 = Integer.MAX_VALUE
    SegmentPurger.RecordModifier recordModifier = row -> {
      if (row.getValue(D2).equals(0)) {
        row.putValue(D2, Integer.MAX_VALUE);
        return true;
      } else {
        return false;
      }
    };

    SegmentPurger segmentPurger =
        new SegmentPurger(_originalIndexDir, PURGED_SEGMENT_DIR, _tableConfig, _schema, recordPurger, recordModifier);
    File purgedIndexDir = segmentPurger.purgeSegment();

    // Check the purge/modify counter in segment purger
    assertEquals(segmentPurger.getNumRecordsPurged(), _expectedNumRecordsPurged);
    assertEquals(segmentPurger.getNumRecordsModified(), _expectedNumRecordsModified);

    // Check crc and index creation time
    SegmentMetadataImpl purgedSegmentMetadata = new SegmentMetadataImpl(purgedIndexDir);
    SegmentMetadataImpl originalSegmentMetadata = new SegmentMetadataImpl(_originalIndexDir);
    assertNotEquals(purgedSegmentMetadata.getCrc(), originalSegmentMetadata.getCrc());
    assertEquals(purgedSegmentMetadata.getIndexCreationTime(), originalSegmentMetadata.getIndexCreationTime());

    try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(purgedIndexDir)) {
      int numRecordsRemaining = 0;
      int numRecordsModified = 0;

      GenericRow row = new GenericRow();
      while (pinotSegmentRecordReader.hasNext()) {
        row = pinotSegmentRecordReader.next(row);

        // Purged segment should not have any record with d1 = 0 or d2 = 0
        assertNotEquals(row.getValue(D1), 0);
        assertNotEquals(row.getValue(D2), 0);

        numRecordsRemaining++;
        if (row.getValue(D2).equals(Integer.MAX_VALUE)) {
          numRecordsModified++;
        }
      }

      assertEquals(numRecordsRemaining, NUM_ROWS - _expectedNumRecordsPurged);
      assertEquals(numRecordsModified, _expectedNumRecordsModified);
    }

    // Check inverted index
    Map<String, Object> props = new HashMap<>();
    props.put(IndexLoadingConfig.READ_MODE_KEY, ReadMode.mmap.toString());
    try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
        .load(purgedIndexDir.toURI(), new SegmentDirectoryLoaderContext.Builder().setTableConfig(_tableConfig)
            .setSegmentName(purgedSegmentMetadata.getName()).setSegmentDirectoryConfigs(new PinotConfiguration(props))
            .build()); SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
      assertTrue(reader.hasIndexFor(D1, StandardIndexes.inverted()));
      assertFalse(reader.hasIndexFor(D2, StandardIndexes.inverted()));
    }
  }