@Override
  public void updateIndices(SegmentDirectory.Writer segmentWriter)
      throws Exception {
    Map<String, List<Operation>> columnOperationsMap = computeOperations(segmentWriter);
    if (columnOperationsMap.isEmpty()) {
      return;
    }

    for (Map.Entry<String, List<Operation>> entry : columnOperationsMap.entrySet()) {
      String column = entry.getKey();
      List<Operation> operations = entry.getValue();

      for (Operation operation : operations) {
        switch (operation) {
          case DISABLE_FORWARD_INDEX:
            // Deletion of the forward index will be handled outside the index handler to ensure that other index
            // handlers that need the forward index to construct their own indexes will have it available.
            _tmpForwardIndexColumns.add(column);
            break;
          case ENABLE_FORWARD_INDEX:
            ColumnMetadata columnMetadata = createForwardIndexIfNeeded(segmentWriter, column, false);
            if (columnMetadata.hasDictionary()) {
              if (!segmentWriter.hasIndexFor(column, StandardIndexes.dictionary())) {
                throw new IllegalStateException(String.format(
                    "Dictionary should still exist after rebuilding forward index for dictionary column: %s", column));
              }
            } else {
              if (segmentWriter.hasIndexFor(column, StandardIndexes.dictionary())) {
                throw new IllegalStateException(
                    String.format("Dictionary should not exist after rebuilding forward index for raw column: %s",
                        column));
              }
            }
            break;
          case DISABLE_DICTIONARY:
            Set<String> newForwardIndexDisabledColumns =
                FieldIndexConfigsUtil.columnsWithIndexDisabled(_fieldIndexConfigs.keySet(), StandardIndexes.forward(),
                    _fieldIndexConfigs);
            if (newForwardIndexDisabledColumns.contains(column)) {
              removeDictionaryFromForwardIndexDisabledColumn(column, segmentWriter);
              if (segmentWriter.hasIndexFor(column, StandardIndexes.dictionary())) {
                throw new IllegalStateException(
                    String.format("Dictionary should not exist after disabling dictionary for column: %s", column));
              }
            } else {
              disableDictionaryAndCreateRawForwardIndex(column, segmentWriter);
            }
            break;
          case ENABLE_DICTIONARY:
            createDictBasedForwardIndex(column, segmentWriter);
            if (!segmentWriter.hasIndexFor(column, StandardIndexes.forward())) {
              throw new IllegalStateException(String.format("Forward index was not created for column: %s", column));
            }
            break;
          case CHANGE_INDEX_COMPRESSION_TYPE:
            rewriteForwardIndexForCompressionChange(column, segmentWriter);
            break;
          default:
            throw new IllegalStateException("Unsupported operation for column " + column);
        }
      }
    }
  }
@Test
  public void testChangeCompressionForMultipleColumns()
      throws Exception {
    // Setup
    SegmentMetadataImpl existingSegmentMetadata = new SegmentMetadataImpl(_segmentDirectory);
    SegmentDirectory segmentLocalFSDirectory =
        new SegmentLocalFSDirectory(_segmentDirectory, existingSegmentMetadata, ReadMode.mmap);
    SegmentDirectory.Writer writer = segmentLocalFSDirectory.createWriter();

    List<FieldConfig> fieldConfigs = new ArrayList<>(_tableConfig.getFieldConfigList());
    Random rand = new Random();
    int randomIdx = rand.nextInt(RAW_COMPRESSION_TYPES.size());
    CompressionCodec newCompressionType = RAW_COMPRESSION_TYPES.get(randomIdx);

    // Column 1
    String name;
    do {
      // Only try to change compression type for forward index enabled columns
      randomIdx = rand.nextInt(fieldConfigs.size());
      name = fieldConfigs.get(randomIdx).getName();
    } while (SV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name) || MV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name)
        || MV_FORWARD_INDEX_DISABLED_DUPLICATES_COLUMNS.contains(name) || FORWARD_INDEX_DISABLED_RAW_COLUMNS.contains(
        name) || DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITHOUT_INV_IDX.equals(name)
        || DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITH_RANGE_INDEX.equals(name));
    FieldConfig config1 = fieldConfigs.remove(randomIdx);
    String column1 = config1.getName();
    FieldConfig newConfig1 =
        new FieldConfig(column1, FieldConfig.EncodingType.RAW, Collections.emptyList(), newCompressionType, null);
    fieldConfigs.add(newConfig1);

    // Column 2
    do {
      // Only try to change compression type for forward index enabled columns
      randomIdx = rand.nextInt(fieldConfigs.size());
      name = fieldConfigs.get(randomIdx).getName();
    } while (SV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name) || MV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name)
        || MV_FORWARD_INDEX_DISABLED_DUPLICATES_COLUMNS.contains(name) || FORWARD_INDEX_DISABLED_RAW_COLUMNS.contains(
        name) || DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITHOUT_INV_IDX.equals(name)
        || DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITH_RANGE_INDEX.equals(name));
    FieldConfig config2 = fieldConfigs.remove(randomIdx);
    String column2 = config2.getName();
    FieldConfig newConfig2 =
        new FieldConfig(column2, FieldConfig.EncodingType.RAW, Collections.emptyList(), newCompressionType, null);
    fieldConfigs.add(newConfig2);

    TableConfig tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNoDictionaryColumns(_noDictionaryColumns)
            .setFieldConfigList(fieldConfigs).build();
    tableConfig.setFieldConfigList(fieldConfigs);

    IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig);
    ForwardIndexHandler fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, null);
    fwdIndexHandler.updateIndices(writer);
    fwdIndexHandler.postUpdateIndicesCleanup(writer);

    // Tear down before validation. Because columns.psf and index map cleanup happens at segmentDirectory.close()
    segmentLocalFSDirectory.close();

    ColumnMetadata metadata = existingSegmentMetadata.getColumnMetadataFor(column1);
    testIndexExists(column1, StandardIndexes.forward());
    validateIndexMap(column1, false, false);
    validateForwardIndex(column1, newCompressionType, metadata.isSorted());
    // Validate metadata properties. Nothing should change when a forwardIndex is rewritten for compressionType
    // change.
    validateMetadataProperties(column1, metadata.hasDictionary(), metadata.getColumnMaxLength(),
        metadata.getCardinality(), metadata.getTotalDocs(), metadata.getDataType(), metadata.getFieldType(),
        metadata.isSorted(), metadata.isSingleValue(), metadata.getMaxNumberOfMultiValues(),
        metadata.getTotalNumberOfEntries(), metadata.isAutoGenerated(), metadata.getMinValue(), metadata.getMaxValue(),
        false);

    metadata = existingSegmentMetadata.getColumnMetadataFor(column2);
    testIndexExists(column2, StandardIndexes.forward());
    validateIndexMap(column2, false, false);
    validateForwardIndex(column2, newCompressionType, metadata.isSorted());
    validateMetadataProperties(column2, metadata.hasDictionary(), metadata.getColumnMaxLength(),
        metadata.getCardinality(), metadata.getTotalDocs(), metadata.getDataType(), metadata.getFieldType(),
        metadata.isSorted(), metadata.isSingleValue(), metadata.getMaxNumberOfMultiValues(),
        metadata.getTotalNumberOfEntries(), metadata.isAutoGenerated(), metadata.getMinValue(), metadata.getMaxValue(),
        false);
  }