private static void validateIndexingConfig(IndexingConfig indexingConfig, @Nullable Schema schema) {
    if (schema == null) {
      return;
    }
    ArrayListMultimap<String, String> columnNameToConfigMap = ArrayListMultimap.create();
    Set<String> noDictionaryColumnsSet = new HashSet<>();

    if (indexingConfig.getNoDictionaryColumns() != null) {
      for (String columnName : indexingConfig.getNoDictionaryColumns()) {
        columnNameToConfigMap.put(columnName, "No Dictionary Column Config");
        noDictionaryColumnsSet.add(columnName);
      }
    }
    Set<String> bloomFilterColumns = new HashSet<>();
    if (indexingConfig.getBloomFilterColumns() != null) {
      bloomFilterColumns.addAll(indexingConfig.getBloomFilterColumns());
    }
    if (indexingConfig.getBloomFilterConfigs() != null) {
      bloomFilterColumns.addAll(indexingConfig.getBloomFilterConfigs().keySet());
    }
    for (String bloomFilterColumn : bloomFilterColumns) {
      columnNameToConfigMap.put(bloomFilterColumn, "Bloom Filter Config");
    }
    if (indexingConfig.getInvertedIndexColumns() != null) {
      for (String columnName : indexingConfig.getInvertedIndexColumns()) {
        if (noDictionaryColumnsSet.contains(columnName)) {
          throw new IllegalStateException("Cannot create an Inverted index on column " + columnName
              + " specified in the noDictionaryColumns config");
        }
        columnNameToConfigMap.put(columnName, "Inverted Index Config");
      }
    }

    if (indexingConfig.getOnHeapDictionaryColumns() != null) {
      for (String columnName : indexingConfig.getOnHeapDictionaryColumns()) {
        columnNameToConfigMap.put(columnName, "On Heap Dictionary Column Config");
      }
    }
    if (indexingConfig.getRangeIndexColumns() != null) {
      for (String columnName : indexingConfig.getRangeIndexColumns()) {
        columnNameToConfigMap.put(columnName, "Range Column Config");
      }
    }
    if (indexingConfig.getSortedColumn() != null) {
      for (String columnName : indexingConfig.getSortedColumn()) {
        columnNameToConfigMap.put(columnName, "Sorted Column Config");
      }
    }
    if (indexingConfig.getVarLengthDictionaryColumns() != null) {
      for (String columnName : indexingConfig.getVarLengthDictionaryColumns()) {
        columnNameToConfigMap.put(columnName, "Var Length Column Config");
      }
    }
    if (indexingConfig.getSegmentPartitionConfig() != null
        && indexingConfig.getSegmentPartitionConfig().getColumnPartitionMap() != null) {
      for (String columnName : indexingConfig.getSegmentPartitionConfig().getColumnPartitionMap().keySet()) {
        columnNameToConfigMap.put(columnName, "Segment Partition Config");
      }
    }
    Set<String> jsonIndexColumns = new HashSet<>();
    // Ignore jsonIndexColumns when jsonIndexConfigs is configured
    if (indexingConfig.getJsonIndexConfigs() != null) {
      jsonIndexColumns.addAll(indexingConfig.getJsonIndexConfigs().keySet());
    } else {
      if (indexingConfig.getJsonIndexColumns() != null) {
        jsonIndexColumns.addAll(indexingConfig.getJsonIndexColumns());
      }
    }
    for (String columnName : jsonIndexColumns) {
      columnNameToConfigMap.put(columnName, "Json Index Config");
    }

    List<StarTreeIndexConfig> starTreeIndexConfigList = indexingConfig.getStarTreeIndexConfigs();
    if (starTreeIndexConfigList != null) {
      for (StarTreeIndexConfig starTreeIndexConfig : starTreeIndexConfigList) {
        // Dimension split order cannot be null
        for (String columnName : starTreeIndexConfig.getDimensionsSplitOrder()) {
          columnNameToConfigMap.put(columnName, STAR_TREE_CONFIG_NAME);
        }
        List<String> functionColumnPairs = starTreeIndexConfig.getFunctionColumnPairs();
        List<StarTreeAggregationConfig> aggregationConfigs = starTreeIndexConfig.getAggregationConfigs();
        Preconditions.checkState(functionColumnPairs == null || aggregationConfigs == null,
            "Only one of 'functionColumnPairs' or 'aggregationConfigs' can be specified in StarTreeIndexConfig");
        Set<AggregationFunctionColumnPair> storedTypes = new HashSet<>();
        if (functionColumnPairs != null) {
          for (String functionColumnPair : functionColumnPairs) {
            AggregationFunctionColumnPair columnPair;
            try {
              columnPair = AggregationFunctionColumnPair.fromColumnName(functionColumnPair);
            } catch (Exception e) {
              throw new IllegalStateException("Invalid StarTreeIndex config: " + functionColumnPair + ". Must be"
                  + "in the form <Aggregation function>__<Column name>");
            }
            AggregationFunctionColumnPair storedType = AggregationFunctionColumnPair.resolveToStoredType(columnPair);
            if (!storedTypes.add(storedType)) {
              LOGGER.warn("StarTreeIndex config duplication: {} already matches existing function column pair: {}. ",
                  columnPair, storedType);
            }
            String columnName = columnPair.getColumn();
            if (!columnName.equals(AggregationFunctionColumnPair.STAR)) {
              columnNameToConfigMap.put(columnName, STAR_TREE_CONFIG_NAME);
            }
          }
        }
        if (aggregationConfigs != null) {
          for (StarTreeAggregationConfig aggregationConfig : aggregationConfigs) {
            AggregationFunctionColumnPair columnPair;
            try {
              columnPair = AggregationFunctionColumnPair.fromAggregationConfig(aggregationConfig);
            } catch (Exception e) {
              throw new IllegalStateException("Invalid StarTreeIndex config: " + aggregationConfig);
            }
            AggregationFunctionColumnPair storedType = AggregationFunctionColumnPair.resolveToStoredType(columnPair);
            if (!storedTypes.add(storedType)) {
              LOGGER.warn("StarTreeIndex config duplication: {} already matches existing function column pair: {}. ",
                  columnPair, storedType);
            }
            String columnName = columnPair.getColumn();
            if (!columnName.equals(AggregationFunctionColumnPair.STAR)) {
              columnNameToConfigMap.put(columnName, STAR_TREE_CONFIG_NAME);
            }
          }
        }
        List<String> skipDimensionList = starTreeIndexConfig.getSkipStarNodeCreationForDimensions();
        if (skipDimensionList != null) {
          for (String columnName : skipDimensionList) {
            columnNameToConfigMap.put(columnName, STAR_TREE_CONFIG_NAME);
          }
        }
      }
    }

    for (Map.Entry<String, String> entry : columnNameToConfigMap.entries()) {
      String columnName = entry.getKey();
      String configName = entry.getValue();
      FieldSpec columnFieldSpec = schema.getFieldSpecFor(columnName);
      Preconditions.checkState(columnFieldSpec != null,
          "Column Name " + columnName + " defined in " + configName + " must be a valid column defined in the schema");
      if (configName.equals(STAR_TREE_CONFIG_NAME)) {
        Preconditions.checkState(columnFieldSpec.isSingleValueField(),
            "Column Name " + columnName + " defined in " + configName + " must be a single value column");
      }
    }

    // Range index semantic validation
    // Range index can be defined on numeric columns and any column with a dictionary
    if (indexingConfig.getRangeIndexColumns() != null) {
      for (String rangeIndexCol : indexingConfig.getRangeIndexColumns()) {
        Preconditions.checkState(
            schema.getFieldSpecFor(rangeIndexCol).getDataType().isNumeric() || !noDictionaryColumnsSet.contains(
                rangeIndexCol), "Cannot create a range index on non-numeric/no-dictionary column " + rangeIndexCol);
      }
    }

    // Var length dictionary semantic validation
    if (indexingConfig.getVarLengthDictionaryColumns() != null) {
      for (String varLenDictCol : indexingConfig.getVarLengthDictionaryColumns()) {
        FieldSpec varLenDictFieldSpec = schema.getFieldSpecFor(varLenDictCol);
        switch (varLenDictFieldSpec.getDataType().getStoredType()) {
          case STRING:
          case BYTES:
            continue;
          default:
            throw new IllegalStateException(
                "var length dictionary can only be created for columns of type STRING and BYTES. Invalid for column "
                    + varLenDictCol);
        }
      }
    }

    for (String jsonIndexColumn : jsonIndexColumns) {
      FieldSpec fieldSpec = schema.getFieldSpecFor(jsonIndexColumn);
      Preconditions.checkState(
          fieldSpec.isSingleValueField() && fieldSpec.getDataType().getStoredType() == DataType.STRING,
          "Json index can only be created for single value String column. Invalid for column: %s", jsonIndexColumn);
    }
  }
@Test
  public void testValidateIndexingConfig() {
    Schema schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .addSingleValueDimension("bytesCol", FieldSpec.DataType.BYTES)
            .addSingleValueDimension("intCol", FieldSpec.DataType.INT)
            .addMultiValueDimension("multiValCol", FieldSpec.DataType.STRING).build();
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setBloomFilterColumns(Arrays.asList("myCol2")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Bloom filter column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setInvertedIndexColumns(Arrays.asList(""))
            .build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setInvertedIndexColumns(Arrays.asList("myCol2")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Inverted Index column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNoDictionaryColumns(Arrays.asList(""))
            .build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setNoDictionaryColumns(Arrays.asList("myCol2")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid No Dictionary column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setOnHeapDictionaryColumns(Arrays.asList(""))
            .build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setOnHeapDictionaryColumns(Arrays.asList("myCol2")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid On Heap Dictionary column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setRangeIndexColumns(Arrays.asList(""))
            .build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setRangeIndexColumns(Arrays.asList("myCol2"))
            .build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Range Index column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setSortedColumn("").build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setSortedColumn("myCol2").build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Sorted column name");
    } catch (Exception e) {
      // expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setVarLengthDictionaryColumns(Arrays.asList("")).build();
    TableConfigUtils.validate(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setVarLengthDictionaryColumns(Arrays.asList("myCol2")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Var Length Dictionary column name");
    } catch (Exception e) {
      // expected
    }

    ColumnPartitionConfig columnPartitionConfig = new ColumnPartitionConfig("Murmur", 4);
    Map<String, ColumnPartitionConfig> partitionConfigMap = new HashMap<>();
    partitionConfigMap.put("myCol2", columnPartitionConfig);
    SegmentPartitionConfig partitionConfig = new SegmentPartitionConfig(partitionConfigMap);
    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setSegmentPartitionConfig(partitionConfig)
            .build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid Segment Partition column name");
    } catch (Exception e) {
      // expected
    }

    // Although this config makes no sense, it should pass the validation phase
    StarTreeIndexConfig starTreeIndexConfig =
        new StarTreeIndexConfig(List.of("myCol"), List.of("myCol"), List.of("SUM__myCol"), null, 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
    } catch (Exception e) {
      Assert.fail("Should not fail for valid StarTreeIndex config column name");
    }

    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol2"), List.of("myCol"), List.of("SUM__myCol"), null, 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid StarTreeIndex config column name in dimension split order");
    } catch (Exception e) {
      // expected
    }

    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), List.of("myCol2"), List.of("SUM__myCol"), null, 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid StarTreeIndex config column name in skip star node for dimension");
    } catch (Exception e) {
      // expected
    }

    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), List.of("myCol"), List.of("SUM__myCol2"), null, 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid StarTreeIndex config column name in function column pair");
    } catch (Exception e) {
      // expected
    }

    starTreeIndexConfig =
        new StarTreeIndexConfig(List.of("myCol"), null, null, List.of(new StarTreeAggregationConfig("myCol2", "SUM")),
            1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid StarTreeIndex config column name in aggregation config");
    } catch (Exception e) {
      // expected
    }

    starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), null, List.of("SUM__myCol"),
        List.of(new StarTreeAggregationConfig("myCol", "SUM")), 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid StarTreeIndex config with both function column pair and aggregation config");
    } catch (Exception e) {
      // expected
    }

    starTreeIndexConfig =
        new StarTreeIndexConfig(List.of("multiValCol"), List.of("multiValCol"), List.of("SUM__multiValCol"), null, 1);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setStarTreeIndexConfigs(List.of(starTreeIndexConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for multi-value column name in StarTreeIndex config");
    } catch (Exception e) {
      // expected
    }

    FieldConfig fieldConfig = new FieldConfig("myCol2", null, Collections.emptyList(), null, null);
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setFieldConfigList(Arrays.asList(fieldConfig)).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for invalid column name in Field Config List");
    } catch (Exception e) {
      // expected
    }

    List<String> columnList = Arrays.asList("myCol");
    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNoDictionaryColumns(columnList)
        .setInvertedIndexColumns(columnList).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for valid column name in both no dictionary and inverted index column config");
    } catch (Exception e) {
      // expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setJsonIndexColumns(Arrays.asList("non-existent-column")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for non existent column in Json index config");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setJsonIndexColumns(Arrays.asList("intCol"))
            .build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for Json index defined on non string column");
    } catch (Exception e) {
      // expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setJsonIndexColumns(Arrays.asList("multiValCol")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for Json index defined on multi-value column");
    } catch (Exception e) {
      // expected
    }

    tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setRangeIndexColumns(columnList).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
    } catch (Exception e) {
      Assert.fail("Should work for range index defined on dictionary encoded string column");
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setRangeIndexColumns(columnList)
        .setNoDictionaryColumns(columnList).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for range index defined on non numeric/no-dictionary column");
    } catch (Exception e) {
      // Expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setVarLengthDictionaryColumns(Arrays.asList("intCol")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for Var length dictionary defined for non string/bytes column");
    } catch (Exception e) {
      // expected
    }

    tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
        .setJsonIndexColumns(Arrays.asList("multiValCol")).build();
    try {
      TableConfigUtils.validate(tableConfig, schema);
      Assert.fail("Should fail for Json Index defined on a multi value column");
    } catch (Exception e) {
      // expected
    }
  }