@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 testDisableForwardIndexForSingleRawColumn()
      throws Exception {
    Set<String> forwardIndexDisabledColumns = new HashSet<>(SV_FORWARD_INDEX_DISABLED_COLUMNS);
    forwardIndexDisabledColumns.addAll(MV_FORWARD_INDEX_DISABLED_COLUMNS);
    forwardIndexDisabledColumns.addAll(MV_FORWARD_INDEX_DISABLED_DUPLICATES_COLUMNS);
    forwardIndexDisabledColumns.addAll(FORWARD_INDEX_DISABLED_RAW_COLUMNS);
    forwardIndexDisabledColumns.add(DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITHOUT_INV_IDX);
    forwardIndexDisabledColumns.add(DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITH_RANGE_INDEX);
    for (String column : _noDictionaryColumns) {
      if (FORWARD_INDEX_DISABLED_RAW_COLUMNS.contains(column) || RAW_SORTED_INDEX_COLUMNS.contains(column)) {
        // Forward index already disabled for these columns, skip them
        continue;
      }
      SegmentMetadataImpl existingSegmentMetadata = new SegmentMetadataImpl(_segmentDirectory);
      SegmentDirectory segmentLocalFSDirectory =
          new SegmentLocalFSDirectory(_segmentDirectory, existingSegmentMetadata, ReadMode.mmap);
      SegmentDirectory.Writer writer = segmentLocalFSDirectory.createWriter();

      IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
      forwardIndexDisabledColumns.add(column);
      indexLoadingConfig.setForwardIndexDisabledColumns(forwardIndexDisabledColumns);
      indexLoadingConfig.removeNoDictionaryColumns(forwardIndexDisabledColumns);
      indexLoadingConfig.addNoDictionaryColumns(FORWARD_INDEX_DISABLED_RAW_COLUMNS);
      Set<String> invertedIndexColumns = new HashSet<>(forwardIndexDisabledColumns);
      invertedIndexColumns.removeAll(FORWARD_INDEX_DISABLED_RAW_COLUMNS);
      invertedIndexColumns.remove(DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITHOUT_INV_IDX);
      invertedIndexColumns.remove(DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITH_RANGE_INDEX);
      indexLoadingConfig.setInvertedIndexColumns(invertedIndexColumns);
      ForwardIndexHandler fwdIndexHandler =
          new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
      fwdIndexHandler.updateIndices(writer);
      fwdIndexHandler.postUpdateIndicesCleanup(writer);

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

      validateIndexMap(column, true, true);
      validateIndexesForForwardIndexDisabledColumns(column);

      // In column metadata, nothing other than hasDictionary and dictionaryElementSize should change.
      int dictionaryElementSize = 0;
      ColumnMetadata metadata = existingSegmentMetadata.getColumnMetadataFor(column);
      FieldSpec.DataType dataType = metadata.getDataType();
      if (dataType == FieldSpec.DataType.STRING || dataType == FieldSpec.DataType.BYTES) {
        // This value is based on the rows in createTestData().
        dictionaryElementSize = 7;
      } else if (dataType == FieldSpec.DataType.BIG_DECIMAL) {
        dictionaryElementSize = 4;
      }
      validateMetadataProperties(column, true, dictionaryElementSize, metadata.getCardinality(),
          metadata.getTotalDocs(), dataType, metadata.getFieldType(), metadata.isSorted(), metadata.isSingleValue(),
          metadata.getMaxNumberOfMultiValues(), metadata.getTotalNumberOfEntries(), metadata.isAutoGenerated(),
          metadata.getMinValue(), metadata.getMaxValue(), false);
    }
  }