@VisibleForTesting
  static void validatePartialUpsertStrategies(TableConfig tableConfig, Schema schema) {
    if (tableConfig.getUpsertMode() != UpsertConfig.Mode.PARTIAL) {
      return;
    }

    Preconditions.checkState(tableConfig.getIndexingConfig().isNullHandlingEnabled(),
        "Null handling must be enabled for partial upsert tables");

    UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
    assert upsertConfig != null;
    Map<String, UpsertConfig.Strategy> partialUpsertStrategies = upsertConfig.getPartialUpsertStrategies();
    String partialUpsertMergerClass = upsertConfig.getPartialUpsertMergerClass();

    // check if partialUpsertMergerClass is provided then partialUpsertStrategies should be empty
    if (StringUtils.isNotBlank(partialUpsertMergerClass)) {
      Preconditions.checkState(MapUtils.isEmpty(partialUpsertStrategies),
          "If partialUpsertMergerClass is provided then partialUpsertStrategies should be empty");
    } else {
      List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
      // validate partial upsert column mergers
      for (Map.Entry<String, UpsertConfig.Strategy> entry : partialUpsertStrategies.entrySet()) {
        String column = entry.getKey();
        UpsertConfig.Strategy columnStrategy = entry.getValue();
        Preconditions.checkState(!primaryKeyColumns.contains(column),
            "Merger cannot be applied to primary key columns");

        if (upsertConfig.getComparisonColumns() != null) {
          Preconditions.checkState(!upsertConfig.getComparisonColumns().contains(column),
              "Merger cannot be applied to comparison column");
        } else {
          Preconditions.checkState(!tableConfig.getValidationConfig().getTimeColumnName().equals(column),
              "Merger cannot be applied to time column");
        }

        FieldSpec fieldSpec = schema.getFieldSpecFor(column);
        Preconditions.checkState(fieldSpec != null, "Merger cannot be applied to non-existing column: %s", column);

        if (columnStrategy == UpsertConfig.Strategy.INCREMENT) {
          Preconditions.checkState(fieldSpec.getDataType().getStoredType().isNumeric(),
              "INCREMENT merger cannot be applied to non-numeric column: %s", column);
          Preconditions.checkState(!schema.getDateTimeNames().contains(column),
              "INCREMENT merger cannot be applied to date time column: %s", column);
        } else if (columnStrategy == UpsertConfig.Strategy.APPEND || columnStrategy == UpsertConfig.Strategy.UNION) {
          Preconditions.checkState(!fieldSpec.isSingleValueField(),
              "%s merger cannot be applied to single-value column: %s", columnStrategy.toString(), column);
        }
      }
    }
  }
@Test
  public void testValidatePartialUpsertConfig() {
    Schema schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol1", FieldSpec.DataType.LONG)
            .addSingleValueDimension("myCol2", FieldSpec.DataType.STRING)
            .addDateTime("myTimeCol", FieldSpec.DataType.LONG, "1:DAYS:EPOCH", "1:DAYS")
            .setPrimaryKeyColumns(Lists.newArrayList("myCol1")).build();

    Map<String, String> streamConfigs = getStreamConfigs();
    streamConfigs.put("stream.kafka.consumer.type", "simple");
    Map<String, UpsertConfig.Strategy> partialUpsertStratgies = new HashMap<>();
    partialUpsertStratgies.put("myCol2", UpsertConfig.Strategy.IGNORE);
    UpsertConfig partialUpsertConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL);
    partialUpsertConfig.setPartialUpsertStrategies(partialUpsertStratgies);
    partialUpsertConfig.setDefaultPartialUpsertStrategy(UpsertConfig.Strategy.OVERWRITE);
    partialUpsertConfig.setComparisonColumn("myCol2");
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(partialUpsertConfig)
            .setNullHandlingEnabled(true).setRoutingConfig(
                new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
            .setStreamConfigs(streamConfigs).build();
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Merger cannot be applied to comparison column");
    }

    partialUpsertConfig = new UpsertConfig(UpsertConfig.Mode.PARTIAL);
    partialUpsertConfig.setPartialUpsertStrategies(partialUpsertStratgies);
    partialUpsertConfig.setDefaultPartialUpsertStrategy(UpsertConfig.Strategy.OVERWRITE);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName("myCol2")
        .setUpsertConfig(partialUpsertConfig).setNullHandlingEnabled(true).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).build();
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Merger cannot be applied to time column");
    }

    partialUpsertStratgies.put("myCol1", UpsertConfig.Strategy.INCREMENT);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName("timeCol")
        .setUpsertConfig(partialUpsertConfig).setNullHandlingEnabled(false).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).build();
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Null handling must be enabled for partial upsert tables");
    }

    tableConfig.getIndexingConfig().setNullHandlingEnabled(true);
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Merger cannot be applied to primary key columns");
    }

    partialUpsertStratgies.clear();
    partialUpsertStratgies.put("randomCol", UpsertConfig.Strategy.OVERWRITE);
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Merger cannot be applied to non-existing column: randomCol");
    }

    partialUpsertStratgies.clear();
    partialUpsertStratgies.put("myCol2", UpsertConfig.Strategy.INCREMENT);
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "INCREMENT merger cannot be applied to non-numeric column: myCol2");
    }

    partialUpsertStratgies.clear();
    partialUpsertStratgies.put("myCol2", UpsertConfig.Strategy.APPEND);
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "APPEND merger cannot be applied to single-value column: myCol2");
    }

    partialUpsertStratgies.clear();
    partialUpsertStratgies.put("myTimeCol", UpsertConfig.Strategy.INCREMENT);
    try {
      TableConfigUtils.validatePartialUpsertStrategies(tableConfig, schema);
      Assert.fail();
    } catch (Exception e) {
      Assert.assertEquals(e.getMessage(), "INCREMENT merger cannot be applied to date time column: myTimeCol");
    }
  }