public void build(@Nullable SegmentVersion segmentVersion, ServerMetrics serverMetrics)
      throws Exception {
    SegmentGeneratorConfig genConfig = new SegmentGeneratorConfig(_tableConfig, _dataSchema);

    // The segment generation code in SegmentColumnarIndexCreator will throw
    // exception if start and end time in time column are not in acceptable
    // range. We don't want the realtime consumption to stop (if an exception
    // is thrown) and thus the time validity check is explicitly disabled for
    // realtime segment generation
    genConfig.setSegmentTimeValueCheck(false);
    if (_columnIndicesForRealtimeTable.getInvertedIndexColumns() != null) {
      genConfig.setIndexOn(StandardIndexes.inverted(), IndexConfig.ENABLED,
          _columnIndicesForRealtimeTable.getInvertedIndexColumns());
    }

    if (_columnIndicesForRealtimeTable.getVarLengthDictionaryColumns() != null) {
      genConfig.setVarLengthDictionaryColumns(_columnIndicesForRealtimeTable.getVarLengthDictionaryColumns());
    }

    if (segmentVersion != null) {
      genConfig.setSegmentVersion(segmentVersion);
    }
    genConfig.setTableName(_tableName);
    genConfig.setOutDir(_outputPath);
    genConfig.setSegmentName(_segmentName);

    addIndexOrDefault(genConfig, StandardIndexes.text(), _columnIndicesForRealtimeTable.getTextIndexColumns(),
        new TextIndexConfigBuilder(genConfig.getFSTIndexType()).build());

    addIndexOrDefault(genConfig, StandardIndexes.fst(), _columnIndicesForRealtimeTable.getFstIndexColumns(),
        new FstIndexConfig(genConfig.getFSTIndexType()));

    SegmentPartitionConfig segmentPartitionConfig = _realtimeSegmentImpl.getSegmentPartitionConfig();
    genConfig.setSegmentPartitionConfig(segmentPartitionConfig);
    genConfig.setNullHandlingEnabled(_nullHandlingEnabled);
    genConfig.setSegmentZKPropsConfig(_segmentZKPropsConfig);

    // flush any artifacts to disk to improve mutable to immutable segment conversion
    _realtimeSegmentImpl.commit();

    SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
    try (PinotSegmentRecordReader recordReader = new PinotSegmentRecordReader()) {
      int[] sortedDocIds = _columnIndicesForRealtimeTable.getSortedColumn() != null
          ? _realtimeSegmentImpl.getSortedDocIdIterationOrderWithSortedColumn(
          _columnIndicesForRealtimeTable.getSortedColumn()) : null;
      recordReader.init(_realtimeSegmentImpl, sortedDocIds);
      RealtimeSegmentSegmentCreationDataSource dataSource =
          new RealtimeSegmentSegmentCreationDataSource(_realtimeSegmentImpl, recordReader);
      driver.init(genConfig, dataSource, RecordEnricherPipeline.getPassThroughPipeline(),
          TransformPipeline.getPassThroughPipeline());

      if (!_enableColumnMajor) {
        driver.build();
      } else {
        driver.buildByColumn(_realtimeSegmentImpl);
      }
    }

    if (segmentPartitionConfig != null) {
      Map<String, ColumnPartitionConfig> columnPartitionMap = segmentPartitionConfig.getColumnPartitionMap();
      for (String columnName : columnPartitionMap.keySet()) {
        int numPartitions = driver.getSegmentStats().getColumnProfileFor(columnName).getPartitions().size();
        serverMetrics.addValueToTableGauge(_tableName, ServerGauge.REALTIME_SEGMENT_NUM_PARTITIONS, numPartitions);
      }
    }
  }
@Test(dataProvider = "reuseParams")
  public void testSegmentBuilderWithReuse(boolean columnMajorSegmentBuilder, String sortedColumn,
      boolean reuseMutableIndex, int luceneNRTCachingDirectoryMaxBufferSizeMB)
      throws Exception {
    File tmpDir = new File(TMP_DIR, "tmp_" + System.currentTimeMillis());
    FieldConfig textIndexFieldConfig =
        new FieldConfig.Builder(STRING_COLUMN1).withEncodingType(FieldConfig.EncodingType.RAW)
            .withIndexTypes(Collections.singletonList(FieldConfig.IndexType.TEXT)).build();
    List<FieldConfig> fieldConfigList = Collections.singletonList(textIndexFieldConfig);
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName("testTable").setTimeColumnName(DATE_TIME_COLUMN)
            .setInvertedIndexColumns(Lists.newArrayList(STRING_COLUMN1))
            .setSortedColumn(sortedColumn).setColumnMajorSegmentBuilderEnabled(columnMajorSegmentBuilder)
            .setFieldConfigList(fieldConfigList).build();
    Schema schema = new Schema.SchemaBuilder().addSingleValueDimension(STRING_COLUMN1, FieldSpec.DataType.STRING)
        .addDateTime(DATE_TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();

    String tableNameWithType = tableConfig.getTableName();
    String segmentName = "testTable__0__0__123456";
    IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
    TextIndexConfig textIndexConfig = new TextIndexConfigBuilder()
            .withUseANDForMultiTermQueries(false)
            .withReuseMutableIndex(reuseMutableIndex)
            .withLuceneNRTCachingDirectoryMaxBufferSizeMB(luceneNRTCachingDirectoryMaxBufferSizeMB)
            .build();

    RealtimeSegmentConfig.Builder realtimeSegmentConfigBuilder =
        new RealtimeSegmentConfig.Builder().setTableNameWithType(tableNameWithType).setSegmentName(segmentName)
            .setStreamName(tableNameWithType).setSchema(schema).setTimeColumnName(DATE_TIME_COLUMN).setCapacity(1000)
            .setIndex(Sets.newHashSet(STRING_COLUMN1), StandardIndexes.inverted(), IndexConfig.ENABLED)
            .setIndex(Sets.newHashSet(STRING_COLUMN1), StandardIndexes.text(), textIndexConfig)
            .setFieldConfigList(fieldConfigList).setSegmentZKMetadata(getSegmentZKMetadata(segmentName))
            .setOffHeap(true).setMemoryManager(new DirectMemoryManager(segmentName))
            .setStatsHistory(RealtimeSegmentStatsHistory.deserialzeFrom(new File(tmpDir, "stats")))
            .setConsumerDir(new File(tmpDir, "consumerDir").getAbsolutePath());

    // create mutable segment impl
    RealtimeLuceneTextIndexSearcherPool.init(1);
    RealtimeLuceneIndexRefreshManager.init(1, 10);
    MutableSegmentImpl mutableSegmentImpl = new MutableSegmentImpl(realtimeSegmentConfigBuilder.build(), null);
    List<GenericRow> rows = generateTestDataForReusePath();

    for (GenericRow row : rows) {
      mutableSegmentImpl.index(row, null);
    }

    // build converted segment
    File outputDir = new File(new File(tmpDir, segmentName), "tmp-" + segmentName + "-" + System.currentTimeMillis());
    SegmentZKPropsConfig segmentZKPropsConfig = new SegmentZKPropsConfig();
    segmentZKPropsConfig.setStartOffset("1");
    segmentZKPropsConfig.setEndOffset("100");
    ColumnIndicesForRealtimeTable cdc = new ColumnIndicesForRealtimeTable(sortedColumn,
        indexingConfig.getInvertedIndexColumns(), Collections.singletonList(STRING_COLUMN1), null,
        indexingConfig.getNoDictionaryColumns(), indexingConfig.getVarLengthDictionaryColumns());
    RealtimeSegmentConverter converter =
        new RealtimeSegmentConverter(mutableSegmentImpl, segmentZKPropsConfig, outputDir.getAbsolutePath(), schema,
            tableNameWithType, tableConfig, segmentName, cdc, false);
    converter.build(SegmentVersion.v3, null);

    File indexDir = new File(outputDir, segmentName);
    SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
    assertEquals(segmentMetadata.getVersion(), SegmentVersion.v3);
    assertEquals(segmentMetadata.getTotalDocs(), rows.size());
    assertEquals(segmentMetadata.getTimeColumn(), DATE_TIME_COLUMN);
    assertEquals(segmentMetadata.getTimeUnit(), TimeUnit.MILLISECONDS);

    long expectedStartTime = (long) rows.get(0).getValue(DATE_TIME_COLUMN);
    assertEquals(segmentMetadata.getStartTime(), expectedStartTime);
    long expectedEndTime = (long) rows.get(rows.size() - 1).getValue(DATE_TIME_COLUMN);
    assertEquals(segmentMetadata.getEndTime(), expectedEndTime);

    assertTrue(segmentMetadata.getAllColumns().containsAll(schema.getColumnNames()));
    assertEquals(segmentMetadata.getStartOffset(), "1");
    assertEquals(segmentMetadata.getEndOffset(), "100");

    // read converted segment
    SegmentLocalFSDirectory segmentDir = new SegmentLocalFSDirectory(indexDir, segmentMetadata, ReadMode.mmap);
    SegmentDirectory.Reader segmentReader = segmentDir.createReader();

    Map<String, ColumnIndexContainer> indexContainerMap = new HashMap<>();
    Map<String, ColumnMetadata> columnMetadataMap = segmentMetadata.getColumnMetadataMap();
    IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig);
    for (Map.Entry<String, ColumnMetadata> entry : columnMetadataMap.entrySet()) {
      indexContainerMap.put(entry.getKey(),
          new PhysicalColumnIndexContainer(segmentReader, entry.getValue(), indexLoadingConfig));
    }
    ImmutableSegmentImpl segmentFile = new ImmutableSegmentImpl(segmentDir, segmentMetadata, indexContainerMap, null);

    // test forward index contents
    GenericRow readRow = new GenericRow();
    int docId = 0;
    for (int i = 0; i < rows.size(); i++) {
      GenericRow row;
      if (sortedColumn == null) {
        row = rows.get(i);
      } else {
        row = rows.get(rows.size() - i - 1);
      }

      segmentFile.getRecord(docId, readRow);
      assertEquals(readRow.getValue(STRING_COLUMN1), row.getValue(STRING_COLUMN1));
      assertEquals(readRow.getValue(DATE_TIME_COLUMN), row.getValue(DATE_TIME_COLUMN));
      docId += 1;
    }

    // test docId conversion
    TextIndexReader textIndexReader = segmentFile.getIndex(STRING_COLUMN1, StandardIndexes.text());
    if (sortedColumn == null) {
      assertEquals(textIndexReader.getDocIds("str-8"), ImmutableRoaringBitmap.bitmapOf(0));
      assertEquals(textIndexReader.getDocIds("str-4"), ImmutableRoaringBitmap.bitmapOf(4));
    } else {
      assertEquals(textIndexReader.getDocIds("str-8"), ImmutableRoaringBitmap.bitmapOf(7));
      assertEquals(textIndexReader.getDocIds("str-4"), ImmutableRoaringBitmap.bitmapOf(3));
    }
  }