@VisibleForTesting
  Map<String, DefaultColumnAction> computeDefaultColumnActionMap() {
    Map<String, DefaultColumnAction> defaultColumnActionMap = new HashMap<>();

    // Compute ADD and UPDATE actions.
    for (FieldSpec fieldSpecInSchema : _schema.getAllFieldSpecs()) {
      if (fieldSpecInSchema.isVirtualColumn()) {
        continue;
      }
      String column = fieldSpecInSchema.getName();
      FieldSpec.FieldType fieldTypeInSchema = fieldSpecInSchema.getFieldType();
      ColumnMetadata columnMetadata = _segmentMetadata.getColumnMetadataFor(column);

      if (columnMetadata != null) {
        // Column exists in the segment, check if we need to update the value.

        // Only check for auto-generated column.
        if (!columnMetadata.isAutoGenerated()) {
          continue;
        }

        // Check the field type matches.
        FieldSpec fieldSpecInMetadata = columnMetadata.getFieldSpec();
        FieldSpec.FieldType fieldTypeInMetadata = fieldSpecInMetadata.getFieldType();
        if (fieldTypeInMetadata != fieldTypeInSchema) {
          String failureMessage = "Field type: " + fieldTypeInMetadata + " for auto-generated column: " + column
              + " does not match field type: " + fieldTypeInSchema
              + " in schema, throw exception to drop and re-download the segment.";
          throw new RuntimeException(failureMessage);
        }

        // Check the data type and default value matches.
        DataType dataTypeInMetadata = fieldSpecInMetadata.getDataType();
        DataType dataTypeInSchema = fieldSpecInSchema.getDataType();
        boolean isSingleValueInMetadata = fieldSpecInMetadata.isSingleValueField();
        boolean isSingleValueInSchema = fieldSpecInSchema.isSingleValueField();
        String defaultValueInMetadata = fieldSpecInMetadata.getDefaultNullValueString();
        String defaultValueInSchema = fieldSpecInSchema.getDefaultNullValueString();

        if (fieldTypeInMetadata == DIMENSION) {
          if (dataTypeInMetadata != dataTypeInSchema) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_DATA_TYPE);
          } else if (!defaultValueInSchema.equals(defaultValueInMetadata)) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_DEFAULT_VALUE);
          } else if (isSingleValueInMetadata != isSingleValueInSchema) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DIMENSION_NUMBER_OF_VALUES);
          }
        } else if (fieldTypeInMetadata == METRIC) {
          if (dataTypeInMetadata != dataTypeInSchema) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_DATA_TYPE);
          } else if (!defaultValueInSchema.equals(defaultValueInMetadata)) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_DEFAULT_VALUE);
          } else if (isSingleValueInMetadata != isSingleValueInSchema) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_METRIC_NUMBER_OF_VALUES);
          }
        } else if (fieldTypeInMetadata == DATE_TIME) {
          if (dataTypeInMetadata != dataTypeInSchema) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DATE_TIME_DATA_TYPE);
          } else if (!defaultValueInSchema.equals(defaultValueInMetadata)) {
            defaultColumnActionMap.put(column, DefaultColumnAction.UPDATE_DATE_TIME_DEFAULT_VALUE);
          }
        }
      } else {
        // Column does not exist in the segment, add default value for it.

        switch (fieldTypeInSchema) {
          case DIMENSION:
            defaultColumnActionMap.put(column, DefaultColumnAction.ADD_DIMENSION);
            break;
          case METRIC:
            defaultColumnActionMap.put(column, DefaultColumnAction.ADD_METRIC);
            break;
          case DATE_TIME:
            defaultColumnActionMap.put(column, DefaultColumnAction.ADD_DATE_TIME);
            break;
          default:
            LOGGER.warn("Skip adding default column for column: {} with field type: {}", column, fieldTypeInSchema);
            break;
        }
      }
    }

    // Compute REMOVE actions.
    for (ColumnMetadata columnMetadata : _segmentMetadata.getColumnMetadataMap().values()) {
      String column = columnMetadata.getColumnName();
      // Only remove auto-generated columns
      if (!_schema.hasColumn(column) && columnMetadata.isAutoGenerated()) {
        FieldSpec.FieldType fieldTypeInMetadata = columnMetadata.getFieldSpec().getFieldType();
        if (fieldTypeInMetadata == DIMENSION) {
          defaultColumnActionMap.put(column, DefaultColumnAction.REMOVE_DIMENSION);
        } else if (fieldTypeInMetadata == METRIC) {
          defaultColumnActionMap.put(column, DefaultColumnAction.REMOVE_METRIC);
        } else if (fieldTypeInMetadata == DATE_TIME) {
          defaultColumnActionMap.put(column, DefaultColumnAction.REMOVE_DATE_TIME);
        }
      }
    }

    return defaultColumnActionMap;
  }
@Test
  public void testComputeDefaultColumnActionMapForCommittedSegment() {
    // Dummy IndexLoadingConfig
    IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig();

    // Same schema
    Schema schema0 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT).build();

    BaseDefaultColumnHandler defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema0, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(), Collections.EMPTY_MAP);

    // Add single-value dimension in the schema
    Schema schema1 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column11", FieldSpec.DataType.INT) // add column11
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT).build();
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema1, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(),
        ImmutableMap.of("column11", BaseDefaultColumnHandler.DefaultColumnAction.ADD_DIMENSION));

    // Add multi-value dimension in the schema
    Schema schema2 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addMultiValueDimension("column11", FieldSpec.DataType.INT) // add column11
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT).build();
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema2, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(),
        ImmutableMap.of("column11", BaseDefaultColumnHandler.DefaultColumnAction.ADD_DIMENSION));

    // Add metric in the schema
    Schema schema3 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT)
            .addMetric("column11", FieldSpec.DataType.INT).build(); // add column11
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema3, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(),
        ImmutableMap.of("column11", BaseDefaultColumnHandler.DefaultColumnAction.ADD_METRIC));

    // Add metric in the schema
    Schema schema4 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT)
            .addDateTime("column11", FieldSpec.DataType.INT, "1:HOURS:EPOCH", "1:HOURS").build(); // add column11
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema4, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(),
        ImmutableMap.of("column11", BaseDefaultColumnHandler.DefaultColumnAction.ADD_DATE_TIME));

    // Do not remove non-autogenerated column in the segmentMetadata
    Schema schema5 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING) // remove column2
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT).build();
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema5, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(), Collections.EMPTY_MAP);

    // Do not update non-autogenerated column in the schema
    Schema schema6 =
        new Schema.SchemaBuilder().setSchemaName("testTable").addSingleValueDimension("column1", FieldSpec.DataType.INT)
            .addSingleValueDimension("column2", FieldSpec.DataType.STRING) // update datatype
            .addSingleValueDimension("column3", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column4", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column5", FieldSpec.DataType.STRING)
            .addSingleValueDimension("column6", FieldSpec.DataType.INT)
            .addSingleValueDimension("column7", FieldSpec.DataType.INT)
            .addSingleValueDimension("column8", FieldSpec.DataType.INT)
            .addSingleValueDimension("column9", FieldSpec.DataType.INT)
            .addSingleValueDimension("column10", FieldSpec.DataType.INT)
            .addSingleValueDimension("column13", FieldSpec.DataType.INT)
            .addSingleValueDimension("count", FieldSpec.DataType.INT)
            .addSingleValueDimension("daysSinceEpoch", FieldSpec.DataType.INT)
            .addSingleValueDimension("weeksSinceEpochSunday", FieldSpec.DataType.INT).build();
    defaultColumnHandler =
        new V3DefaultColumnHandler(_segmentDirectory, _committedSegmentMetadata, indexLoadingConfig, schema6, _writer);
    Assert.assertEquals(defaultColumnHandler.computeDefaultColumnActionMap(), Collections.EMPTY_MAP);
  }