@VisibleForTesting
  Map<String, List<Operation>> computeOperations(SegmentDirectory.Reader segmentReader)
      throws Exception {
    Map<String, List<Operation>> columnOperationsMap = new HashMap<>();

    // Does not work for segment versions < V3.
    if (_segmentDirectory.getSegmentMetadata().getVersion().compareTo(SegmentVersion.v3) < 0) {
      return columnOperationsMap;
    }

    Set<String> existingAllColumns = _segmentDirectory.getSegmentMetadata().getAllColumns();
    Set<String> existingDictColumns = _segmentDirectory.getColumnsWithIndex(StandardIndexes.dictionary());
    Set<String> existingForwardIndexColumns = _segmentDirectory.getColumnsWithIndex(StandardIndexes.forward());

    for (String column : existingAllColumns) {
      if (_schema != null && !_schema.hasColumn(column)) {
        // _schema will be null only in tests
        LOGGER.info("Column {} is not in schema, skipping updating forward index", column);
        continue;
      }
      boolean existingHasDict = existingDictColumns.contains(column);
      boolean existingHasFwd = existingForwardIndexColumns.contains(column);
      FieldIndexConfigs newConf = _fieldIndexConfigs.get(column);
      boolean newIsFwd = newConf.getConfig(StandardIndexes.forward()).isEnabled();
      boolean newIsDict = newConf.getConfig(StandardIndexes.dictionary()).isEnabled();
      boolean newIsRange = newConf.getConfig(StandardIndexes.range()).isEnabled();

      if (existingHasFwd && !newIsFwd) {
        // Existing column has a forward index. New column config disables the forward index

        ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
        if (columnMetadata.isSorted()) {
          // Check if the column is sorted. If sorted, disabling forward index should be a no-op. Do not return an
          // operation for this column related to disabling forward index.
          LOGGER.warn("Trying to disable the forward index for a sorted column {}, ignoring", column);
          continue;
        }

        if (existingHasDict) {
          if (!newIsDict) {
            // Dictionary was also disabled. Just disable the dictionary and remove it along with the forward index
            // If range index exists, don't try to regenerate it on toggling the dictionary, throw an error instead
            Preconditions.checkState(!newIsRange, String.format(
                "Must disable range (enabled) index to disable the dictionary and forward index for column: %s or "
                    + "refresh / back-fill the forward index", column));
            columnOperationsMap.put(column,
                Arrays.asList(Operation.DISABLE_FORWARD_INDEX, Operation.DISABLE_DICTIONARY));
          } else {
            // Dictionary is still enabled, keep it but remove the forward index
            columnOperationsMap.put(column, Collections.singletonList(Operation.DISABLE_FORWARD_INDEX));
          }
        } else {
          if (!newIsDict) {
            // Dictionary remains disabled and we should not reconstruct temporary forward index as dictionary based
            columnOperationsMap.put(column, Collections.singletonList(Operation.DISABLE_FORWARD_INDEX));
          } else {
            // Dictionary is enabled, creation of dictionary and conversion to dictionary based forward index is needed
            columnOperationsMap.put(column,
                Arrays.asList(Operation.DISABLE_FORWARD_INDEX, Operation.ENABLE_DICTIONARY));
          }
        }
      } else if (!existingHasFwd && newIsFwd) {
        // Existing column does not have a forward index. New column config enables the forward index

        ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
        if (columnMetadata != null && columnMetadata.isSorted()) {
          // Check if the column is sorted. If sorted, disabling forward index should be a no-op and forward index
          // should already exist. Do not return an operation for this column related to enabling forward index.
          LOGGER.warn("Trying to enable the forward index for a sorted column {}, ignoring", column);
          continue;
        }

        // Get list of columns with inverted index
        Set<String> existingInvertedIndexColumns =
            segmentReader.toSegmentDirectory().getColumnsWithIndex(StandardIndexes.inverted());
        if (!existingHasDict || !existingInvertedIndexColumns.contains(column)) {
          // If either dictionary or inverted index is missing on the column there is no way to re-generate the forward
          // index. Treat this as a no-op and log a warning.
          LOGGER.warn("Trying to enable the forward index for a column {} missing either the dictionary ({}) and / or "
                  + "the inverted index ({}) is not possible. Either a refresh or back-fill is required to get the "
                  + "forward index, ignoring", column, existingHasDict ? "enabled" : "disabled",
              existingInvertedIndexColumns.contains(column) ? "enabled" : "disabled");
          continue;
        }

        columnOperationsMap.put(column, Collections.singletonList(Operation.ENABLE_FORWARD_INDEX));
      } else if (!existingHasFwd) {
        // Forward index is disabled for the existing column and should remain disabled based on the latest config
        // Need some checks to see whether the dictionary is being enabled or disabled here and take appropriate actions

        // If the dictionary is not enabled on the existing column it must be on the new noDictionary column list.
        // Cannot enable the dictionary for a column with forward index disabled.
        Preconditions.checkState(existingHasDict || !newIsDict,
            String.format("Cannot regenerate the dictionary for column %s with forward index disabled. Please "
                + "refresh or back-fill the data to add back the forward index", column));

        if (existingHasDict && !newIsDict) {
          // Dictionary is currently enabled on this column but is supposed to be disabled. Remove the dictionary
          // and update the segment metadata If the range index exists then throw an error since we are not
          // regenerating the range index on toggling the dictionary
          Preconditions.checkState(!newIsRange, String.format(
              "Must disable range (enabled) index to disable the dictionary for a forwardIndexDisabled column: %s or "
                  + "refresh / back-fill the forward index", column));
          columnOperationsMap.put(column, Collections.singletonList(Operation.DISABLE_DICTIONARY));
        }
      } else if (!existingHasDict && newIsDict) {
        // Existing column is RAW. New column is dictionary enabled.
        if (_schema == null || _tableConfig == null) {
          // This can only happen in tests.
          LOGGER.warn("Cannot enable dictionary for column={} as schema or tableConfig is null.", column);
          continue;
        }
        ColumnMetadata existingColumnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
        if (DictionaryIndexType.ignoreDictionaryOverride(_tableConfig.getIndexingConfig().isOptimizeDictionary(),
            _tableConfig.getIndexingConfig().isOptimizeDictionaryForMetrics(),
            _tableConfig.getIndexingConfig().getNoDictionarySizeRatioThreshold(), existingColumnMetadata.getFieldSpec(),
            _fieldIndexConfigs.get(column), existingColumnMetadata.getCardinality(),
            existingColumnMetadata.getTotalNumberOfEntries())) {
          columnOperationsMap.put(column, Collections.singletonList(Operation.ENABLE_DICTIONARY));
        }
      } else if (existingHasDict && !newIsDict) {
        // Existing column has dictionary. New config for the column is RAW.
        if (shouldDisableDictionary(column, _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column))) {
          columnOperationsMap.put(column, Collections.singletonList(Operation.DISABLE_DICTIONARY));
        }
      } else if (!existingHasDict) {
        // Both existing and new column is RAW forward index encoded. Check if compression needs to be changed.
        // TODO: Also check if raw index version needs to be changed
        if (shouldChangeRawCompressionType(column, segmentReader)) {
          columnOperationsMap.put(column, Collections.singletonList(Operation.CHANGE_INDEX_COMPRESSION_TYPE));
        }
      } else {
        // Both existing and new column is dictionary encoded. Check if compression needs to be changed.
        if (shouldChangeDictIdCompressionType(column, segmentReader)) {
          columnOperationsMap.put(column, Collections.singletonList(Operation.CHANGE_INDEX_COMPRESSION_TYPE));
        }
      }
    }

    return columnOperationsMap;
  }
@Test
  public void testComputeOperationEnableForwardIndex()
      throws Exception {
    // Setup
    SegmentMetadataImpl existingSegmentMetadata = new SegmentMetadataImpl(_segmentDirectory);
    SegmentDirectory segmentLocalFSDirectory =
        new SegmentLocalFSDirectory(_segmentDirectory, existingSegmentMetadata, ReadMode.mmap);
    SegmentDirectory.Writer writer = segmentLocalFSDirectory.createWriter();

    // TEST1: Try to change compression type for a forward index disabled column and enable forward index for it
    List<FieldConfig> fieldConfigs = new ArrayList<>(_tableConfig.getFieldConfigList());
    int randIdx;
    Random rand = new Random();
    String name;

    do {
      // Only try to change compression type for forward index disabled columns
      randIdx = rand.nextInt(fieldConfigs.size());
      name = fieldConfigs.get(randIdx).getName();
    } while (!SV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name) && !MV_FORWARD_INDEX_DISABLED_COLUMNS.contains(name));
    FieldConfig config = fieldConfigs.remove(randIdx);
    CompressionCodec newCompressionType = null;
    for (CompressionCodec type : RAW_COMPRESSION_TYPES) {
      if (config.getCompressionCodec() != type) {
        newCompressionType = type;
        break;
      }
    }
    FieldConfig newConfig =
        new FieldConfig(config.getName(), FieldConfig.EncodingType.RAW, Collections.emptyList(), newCompressionType,
            null);
    fieldConfigs.add(newConfig);
    List<String> noDictionaryColumns = new ArrayList<>(_noDictionaryColumns);
    noDictionaryColumns.add(config.getName());
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNoDictionaryColumns(noDictionaryColumns)
            .setInvertedIndexColumns(_invertedIndexColumns).setFieldConfigList(fieldConfigs).build();
    tableConfig.setFieldConfigList(fieldConfigs);

    IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(null, tableConfig);
    indexLoadingConfig.addNoDictionaryColumns(config.getName());
    indexLoadingConfig.removeInvertedIndexColumns(config.getName());
    ForwardIndexHandler fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, null);
    Map<String, List<ForwardIndexHandler.Operation>> operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 1);
    assertEquals(operationMap.get(config.getName()),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST2: Enable forward index in dictionary format for a column with forward index disabled
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_SV_FORWARD_INDEX_DISABLED_BYTES);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 1);
    assertEquals(operationMap.get(DIM_SV_FORWARD_INDEX_DISABLED_BYTES),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST3: Enable forward index in raw format for a column with forward index disabled. Remove column from inverted
    // index as well (inverted index needs dictionary)
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_MV_FORWARD_INDEX_DISABLED_INTEGER);
    indexLoadingConfig.addNoDictionaryColumns(DIM_MV_FORWARD_INDEX_DISABLED_INTEGER);
    indexLoadingConfig.removeInvertedIndexColumns(DIM_MV_FORWARD_INDEX_DISABLED_INTEGER);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 1);
    assertEquals(operationMap.get(DIM_MV_FORWARD_INDEX_DISABLED_INTEGER),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST4: Enable forward index in dictionary format for two columns with forward index disabled. Disable inverted
    // index for one of them
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_SV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_MV_FORWARD_INDEX_DISABLED_STRING);
    indexLoadingConfig.removeInvertedIndexColumns(DIM_SV_FORWARD_INDEX_DISABLED_LONG);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 2);
    assertEquals(operationMap.get(DIM_SV_FORWARD_INDEX_DISABLED_LONG),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));
    assertEquals(operationMap.get(DIM_MV_FORWARD_INDEX_DISABLED_STRING),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST5: Enable forward index in raw format for two columns with forward index disabled. Remove column from
    // inverted index as well (inverted index needs dictionary)
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_SV_FORWARD_INDEX_DISABLED_STRING);
    indexLoadingConfig.addNoDictionaryColumns(DIM_SV_FORWARD_INDEX_DISABLED_STRING);
    indexLoadingConfig.removeInvertedIndexColumns(DIM_SV_FORWARD_INDEX_DISABLED_STRING);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.addNoDictionaryColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.removeInvertedIndexColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 2);
    assertEquals(operationMap.get(DIM_SV_FORWARD_INDEX_DISABLED_STRING),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));
    assertEquals(operationMap.get(DIM_MV_FORWARD_INDEX_DISABLED_LONG),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST6: Enable forward index in dictionary format and one in raw format for columns with forward index disabled
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.addNoDictionaryColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.removeInvertedIndexColumns(DIM_MV_FORWARD_INDEX_DISABLED_LONG);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_SV_FORWARD_INDEX_DISABLED_BYTES);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap.size(), 2);
    assertEquals(operationMap.get(DIM_MV_FORWARD_INDEX_DISABLED_LONG),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));
    assertEquals(operationMap.get(DIM_SV_FORWARD_INDEX_DISABLED_BYTES),
        Collections.singletonList(ForwardIndexHandler.Operation.ENABLE_FORWARD_INDEX));

    // TEST7: Enable forward index for a raw column with forward index disabled and keep it as raw
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_RAW_MV_FORWARD_INDEX_DISABLED_INTEGER);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap, Collections.EMPTY_MAP);

    // TEST8: Enable forward index for a dictionary based column with forward index and inverted index disabled
    indexLoadingConfig = new IndexLoadingConfig(null, _tableConfig);
    indexLoadingConfig.removeForwardIndexDisabledColumns(DIM_SV_FORWARD_INDEX_DISABLED_INTEGER_WITHOUT_INV_IDX);
    fwdIndexHandler = new ForwardIndexHandler(segmentLocalFSDirectory, indexLoadingConfig, _schema);
    operationMap = fwdIndexHandler.computeOperations(writer);
    assertEquals(operationMap, Collections.EMPTY_MAP);

    // Tear down
    segmentLocalFSDirectory.close();
  }