public void process()
      throws Exception {
    if (_segmentMetadata.getTotalDocs() == 0) {
      LOGGER.info("Skip preprocessing empty segment: {}", _segmentMetadata.getName());
      return;
    }

    // Segment processing has to be done with a local directory.
    File indexDir = new File(_indexDirURI);

    // This fixes the issue of temporary files not getting deleted after creating new inverted indexes.
    removeInvertedIndexTempFiles(indexDir);

    try (SegmentDirectory.Writer segmentWriter = _segmentDirectory.createWriter()) {
      // Update default columns according to the schema.
      if (_schema != null) {
        DefaultColumnHandler defaultColumnHandler = DefaultColumnHandlerFactory
            .getDefaultColumnHandler(indexDir, _segmentMetadata, _indexLoadingConfig, _schema, segmentWriter);
        defaultColumnHandler.updateDefaultColumns();
        _segmentMetadata = new SegmentMetadataImpl(indexDir);
        _segmentDirectory.reloadMetadata();
      } else {
        LOGGER.warn("Skip creating default columns for segment: {} without schema", _segmentMetadata.getName());
      }

      // Update single-column indices, like inverted index, json index etc.
      List<IndexHandler> indexHandlers = new ArrayList<>();

      // We cannot just create all the index handlers in a random order.
      // Specifically, ForwardIndexHandler needs to be executed first. This is because it modifies the segment metadata
      // while rewriting forward index to create a dictionary. Some other handlers (like the range one) assume that
      // metadata was already been modified by ForwardIndexHandler.
      IndexHandler forwardHandler = createHandler(StandardIndexes.forward());
      indexHandlers.add(forwardHandler);
      forwardHandler.updateIndices(segmentWriter);

      // Now that ForwardIndexHandler.updateIndices has been updated, we can run all other indexes in any order
      _segmentMetadata = new SegmentMetadataImpl(indexDir);
      _segmentDirectory.reloadMetadata();

      for (IndexType<?, ?, ?> type : IndexService.getInstance().getAllIndexes()) {
        if (type != StandardIndexes.forward()) {
          IndexHandler handler = createHandler(type);
          indexHandlers.add(handler);
          handler.updateIndices(segmentWriter);
          // Other IndexHandler classes may modify the segment metadata while creating a temporary forward
          // index to generate their respective indexes from if the forward index was disabled. This new metadata is
          // needed to construct other indexes like RangeIndex.
          _segmentMetadata = _segmentDirectory.getSegmentMetadata();
        }
      }

      // Perform post-cleanup operations on the index handlers.
      for (IndexHandler handler : indexHandlers) {
        handler.postUpdateIndicesCleanup(segmentWriter);
      }

      // Add min/max value to column metadata according to the prune mode.
      ColumnMinMaxValueGeneratorMode columnMinMaxValueGeneratorMode =
          _indexLoadingConfig.getColumnMinMaxValueGeneratorMode();
      if (columnMinMaxValueGeneratorMode != ColumnMinMaxValueGeneratorMode.NONE) {
        ColumnMinMaxValueGenerator columnMinMaxValueGenerator =
            new ColumnMinMaxValueGenerator(_segmentMetadata, segmentWriter, columnMinMaxValueGeneratorMode);
        columnMinMaxValueGenerator.addColumnMinMaxValue();
        // NOTE: This step may modify the segment metadata. When adding new steps after this, un-comment the next line.
        // _segmentMetadata = new SegmentMetadataImpl(indexDir);
      }

      segmentWriter.save();
    }

    // Startree creation will load the segment again, so we need to close and re-open the segment writer to make sure
    // that the other required indices (e.g. forward index) are up-to-date.
    try (SegmentDirectory.Writer segmentWriter = _segmentDirectory.createWriter()) {
      // Create/modify/remove star-trees if required.
      processStarTrees(indexDir);
      _segmentDirectory.reloadMetadata();
      segmentWriter.save();
    }
  }
@Test
  public void testV1UpdateDefaultColumns()
      throws Exception {
    constructV1Segment(Collections.emptyList(), Collections.emptyList(), Collections.emptyList(),
        Collections.emptyList());
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setTransformConfigs(
        ImmutableList.of(
            new TransformConfig(NEW_INT_SV_DIMENSION_COLUMN_NAME, "plus(column1, 1)"),
            new TransformConfig(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME, "reverse(column3)"),
            // Ensure that null values returned by transform functions for derived columns are handled appropriately
            // during segment reload
            new TransformConfig(NEW_NULL_RETURN_STRING_SV_DIMENSION_COLUMN_NAME,
                "json_path_string(column21, 'non-existent-path', null)"),
            // Ensure that any transform function failures result in a null value if error on failure is false
            new TransformConfig(NEW_WRONG_ARG_DATE_TRUNC_DERIVED_COLUMN_NAME, "dateTrunc('abcd', column1)")
        ));
    _tableConfig.setIngestionConfig(ingestionConfig);
    _indexLoadingConfig.addInvertedIndexColumns(NEW_COLUMN_INVERTED_INDEX);
    _indexLoadingConfig.addNoDictionaryColumns(NEW_RAW_STRING_SV_DIMENSION_COLUMN_NAME);
    _indexLoadingConfig.setErrorOnColumnBuildFailure(false);
    checkUpdateDefaultColumns();

    // Try to use the third schema and update default value again.
    // For the third schema, we changed the default value for column 'newStringMVDimension' to 'notSameLength',
    // which is not the same length as before. This should be fine for segment format v1.
    // We added two new columns and also removed the NEW_INT_SV_DIMENSION_COLUMN_NAME from schema.
    // NEW_INT_SV_DIMENSION_COLUMN_NAME exists before processing but removed afterwards.
    SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir);
    assertNotNull(segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME));
    try (SegmentDirectory segmentDirectory = SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader()
        .load(_indexDir.toURI(),
            new SegmentDirectoryLoaderContext.Builder().setSegmentDirectoryConfigs(_configuration).build());
        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory, _indexLoadingConfig,
            _newColumnsSchema3)) {
      processor.process();
    }

    segmentMetadata = new SegmentMetadataImpl(_indexDir);
    assertNull(segmentMetadata.getColumnMetadataFor(NEW_INT_SV_DIMENSION_COLUMN_NAME));

    ColumnMetadata hllMetricMetadata = segmentMetadata.getColumnMetadataFor(NEW_HLL_BYTE_METRIC_COLUMN_NAME);
    FieldSpec expectedHllMetricFieldSpec = _newColumnsSchema3.getFieldSpecFor(NEW_HLL_BYTE_METRIC_COLUMN_NAME);
    assertEquals(hllMetricMetadata.getFieldSpec(), expectedHllMetricFieldSpec);
    ByteArray expectedDefaultValue = new ByteArray((byte[]) expectedHllMetricFieldSpec.getDefaultNullValue());
    assertEquals(hllMetricMetadata.getMinValue(), expectedDefaultValue);
    assertEquals(hllMetricMetadata.getMaxValue(), expectedDefaultValue);

    ColumnMetadata tDigestMetricMetadata = segmentMetadata.getColumnMetadataFor(NEW_TDIGEST_BYTE_METRIC_COLUMN_NAME);
    FieldSpec expectedTDigestMetricFieldSpec = _newColumnsSchema3.getFieldSpecFor(NEW_TDIGEST_BYTE_METRIC_COLUMN_NAME);
    assertEquals(tDigestMetricMetadata.getFieldSpec(), expectedTDigestMetricFieldSpec);
    expectedDefaultValue = new ByteArray((byte[]) expectedTDigestMetricFieldSpec.getDefaultNullValue());
    assertEquals(tDigestMetricMetadata.getMinValue(), expectedDefaultValue);
    assertEquals(tDigestMetricMetadata.getMaxValue(), expectedDefaultValue);
  }