@VisibleForTesting
  static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema) {
    if (tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE && (tableConfig.getDedupConfig() == null
        || !tableConfig.getDedupConfig().isDedupEnabled())) {
      return;
    }

    boolean isUpsertEnabled = tableConfig.getUpsertMode() != UpsertConfig.Mode.NONE;
    boolean isDedupEnabled = tableConfig.getDedupConfig() != null && tableConfig.getDedupConfig().isDedupEnabled();

    // check both upsert and dedup are not enabled simultaneously
    Preconditions.checkState(!(isUpsertEnabled && isDedupEnabled),
        "A table can have either Upsert or Dedup enabled, but not both");
    // check table type is realtime
    Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME,
        "Upsert/Dedup table is for realtime table only.");
    // primary key exists
    Preconditions.checkState(CollectionUtils.isNotEmpty(schema.getPrimaryKeyColumns()),
        "Upsert/Dedup table must have primary key columns in the schema");
    // replica group is configured for routing
    Preconditions.checkState(
        tableConfig.getRoutingConfig() != null && isRoutingStrategyAllowedForUpsert(tableConfig.getRoutingConfig()),
        "Upsert/Dedup table must use strict replica-group (i.e. strictReplicaGroup) based routing");
    Preconditions.checkState(tableConfig.getTenantConfig().getTagOverrideConfig() == null || (
            tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeConsuming() == null
                && tableConfig.getTenantConfig().getTagOverrideConfig().getRealtimeCompleted() == null),
        "Invalid tenant tag override used for Upsert/Dedup table");

    // specifically for upsert
    UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
    if (upsertConfig != null) {
      // no startree index
      Preconditions.checkState(CollectionUtils.isEmpty(tableConfig.getIndexingConfig().getStarTreeIndexConfigs())
              && !tableConfig.getIndexingConfig().isEnableDefaultStarTree(),
          "The upsert table cannot have star-tree index.");

      // comparison column exists
      List<String> comparisonColumns = upsertConfig.getComparisonColumns();
      if (comparisonColumns != null) {
        for (String column : comparisonColumns) {
          Preconditions.checkState(schema.hasColumn(column), "The comparison column does not exist on schema");
        }
      }

      // Delete record column exist and is a BOOLEAN field
      String deleteRecordColumn = upsertConfig.getDeleteRecordColumn();
      if (deleteRecordColumn != null) {
        FieldSpec fieldSpec = schema.getFieldSpecFor(deleteRecordColumn);
        Preconditions.checkState(fieldSpec != null,
            String.format("Column %s specified in deleteRecordColumn does not exist", deleteRecordColumn));
        Preconditions.checkState(fieldSpec.isSingleValueField(),
            String.format("The deleteRecordColumn - %s must be a single-valued column", deleteRecordColumn));
        DataType dataType = fieldSpec.getDataType();
        Preconditions.checkState(dataType == DataType.BOOLEAN || dataType == DataType.STRING || dataType.isNumeric(),
            String.format("The deleteRecordColumn - %s must be of type: String / Boolean / Numeric",
                deleteRecordColumn));
      }

      String outOfOrderRecordColumn = upsertConfig.getOutOfOrderRecordColumn();
      Preconditions.checkState(outOfOrderRecordColumn == null || !upsertConfig.isDropOutOfOrderRecord(),
          "outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist together for upsert table");

      if (outOfOrderRecordColumn != null) {
        FieldSpec fieldSpec = schema.getFieldSpecFor(outOfOrderRecordColumn);
        Preconditions.checkState(
            fieldSpec != null && fieldSpec.isSingleValueField() && fieldSpec.getDataType() == DataType.BOOLEAN,
            "The outOfOrderRecordColumn must be a single-valued BOOLEAN column");
      }
    }


    if (upsertConfig != null && upsertConfig.isEnableDeletedKeysCompactionConsistency()) {
      // enableDeletedKeysCompactionConsistency shouldn't exist with metadataTTL
      Preconditions.checkState(upsertConfig.getMetadataTTL() == 0,
          "enableDeletedKeysCompactionConsistency and metadataTTL shouldn't exist together for upsert table");

      // enableDeletedKeysCompactionConsistency shouldn't exist with enablePreload
      Preconditions.checkState(!upsertConfig.isEnablePreload(),
          "enableDeletedKeysCompactionConsistency and enablePreload shouldn't exist together for upsert table");

      // enableDeletedKeysCompactionConsistency should exist with deletedKeysTTL
      Preconditions.checkState(upsertConfig.getDeletedKeysTTL() > 0,
          "enableDeletedKeysCompactionConsistency should exist with deletedKeysTTL for upsert table");

      // enableDeletedKeysCompactionConsistency should exist with enableSnapshot
      Preconditions.checkState(upsertConfig.isEnableSnapshot(),
          "enableDeletedKeysCompactionConsistency should exist with enableSnapshot for upsert table");

      // enableDeletedKeysCompactionConsistency should exist with UpsertCompactionTask
      TableTaskConfig taskConfig = tableConfig.getTaskConfig();
      Preconditions.checkState(taskConfig != null
              && taskConfig.getTaskTypeConfigsMap().containsKey(UPSERT_COMPACTION_TASK_TYPE),
          "enableDeletedKeysCompactionConsistency should exist with UpsertCompactionTask for upsert table");
    }

    Preconditions.checkState(
        tableConfig.getInstanceAssignmentConfigMap() == null || !tableConfig.getInstanceAssignmentConfigMap()
            .containsKey(InstancePartitionsType.COMPLETED),
        "InstanceAssignmentConfig for COMPLETED is not allowed for upsert tables");
    validateAggregateMetricsForUpsertConfig(tableConfig);
    validateTTLForUpsertConfig(tableConfig, schema);
  }
@Test
  public void testValidateUpsertConfig() {
    Schema schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .build();
    UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Upsert/Dedup table is for realtime table only.");
    }

    tableConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Upsert/Dedup table must have primary key columns in the schema");
    }

    schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .setPrimaryKeyColumns(Lists.newArrayList("myCol")).build();
    Map<String, String> streamConfigs = getStreamConfigs();
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setStreamConfigs(streamConfigs).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "Upsert/Dedup table must use strict replica-group (i.e. strictReplicaGroup) based routing");
    }

    // invalid tag override with upsert
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL)).setStreamConfigs(getStreamConfigs())
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME", "T2_REALTIME")).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail("Tag override must not be allowed with upsert");
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Invalid tenant tag override used for Upsert/Dedup table");
    }

    // tag override even with same tag for CONSUMING and COMPLETED with upsert should fail
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL)).setStreamConfigs(getStreamConfigs())
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setTagOverrideConfig(new TagOverrideConfig("T1_REALTIME", "T1_REALTIME")).build();

    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail("Tag override must not be allowed with upsert");
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Invalid tenant tag override used for Upsert/Dedup table");
    }

    // empty tag override with upsert should pass
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
        .setUpsertConfig(new UpsertConfig(UpsertConfig.Mode.FULL)).setStreamConfigs(getStreamConfigs())
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setTagOverrideConfig(new TagOverrideConfig(null, null)).build();
    TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);

    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).build();
    TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);

    StarTreeIndexConfig starTreeIndexConfig = new StarTreeIndexConfig(Lists.newArrayList("myCol"), null,
        Collections.singletonList(
            new AggregationFunctionColumnPair(AggregationFunctionType.COUNT, "myCol").toColumnName()), null, 10);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStarTreeIndexConfigs(Lists.newArrayList(starTreeIndexConfig)).setStreamConfigs(streamConfigs).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "The upsert table cannot have star-tree index.");
    }

    //With Aggregate Metrics
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).setAggregateMetrics(true).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be enabled together");
    }

    //With aggregation Configs in Ingestion Config
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setAggregationConfigs(Collections.singletonList(new AggregationConfig("twiceSum", "SUM(twice)")));
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).setIngestionConfig(ingestionConfig).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Metrics aggregation and upsert cannot be enabled together");
    }

    //With aggregation Configs in Ingestion Config and IndexingConfig at the same time
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setUpsertConfig(upsertConfig)
        .setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .setStreamConfigs(streamConfigs).setAggregateMetrics(true).setIngestionConfig(ingestionConfig).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "Metrics aggregation cannot be enabled in the Indexing Config and Ingestion Config at the same time");
    }

    // Table upsert with delete column
    String stringTypeDelCol = "stringTypeDelCol";
    String delCol = "myDelCol";
    String mvCol = "mvCol";
    String timestampCol = "timestampCol";
    String invalidCol = "invalidCol";
    schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
        .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
        .addSingleValueDimension(stringTypeDelCol, FieldSpec.DataType.STRING)
        .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN)
        .addSingleValueDimension(timestampCol, FieldSpec.DataType.TIMESTAMP)
        .addMultiValueDimension(mvCol, FieldSpec.DataType.STRING).build();
    streamConfigs = getStreamConfigs();
    streamConfigs.put("stream.kafka.consumer.type", "simple");

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeleteRecordColumn(stringTypeDelCol);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.fail("Shouldn't fail table creation when delete column type is single-valued.");
    }

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeleteRecordColumn(delCol);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.fail("Shouldn't fail table creation when delete column type is single-valued.");
    }

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeleteRecordColumn(timestampCol);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail("Should have failed table creation when delete column type is timestamp.");
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "The deleteRecordColumn - timestampCol must be of type: String / Boolean / Numeric");
    }

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeleteRecordColumn(invalidCol);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail("Should have failed table creation when invalid delete column entered.");
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Column invalidCol specified in deleteRecordColumn does not exist");
    }

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeleteRecordColumn(mvCol);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
      Assert.fail("Should have failed table creation when delete column type is multi-valued.");
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "The deleteRecordColumn - mvCol must be a single-valued column");
    }

    // upsert deleted-keys-ttl configs with no deleted column
    schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
        .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
        .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN).build();
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDeletedKeysTTL(3600);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "Deleted Keys TTL can only be enabled with deleteRecordColumn set.");
    }

    upsertConfig.setDeleteRecordColumn(delCol);
    // multiple comparison columns set for deleted-keys-ttl
    schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
        .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
        .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
        .addSingleValueDimension(delCol, FieldSpec.DataType.BOOLEAN).build();
    upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN, "myCol"));
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "MetadataTTL / DeletedKeysTTL does not work with multiple comparison columns");
    }

    // comparison column with non-numeric type
    upsertConfig.setComparisonColumns(Lists.newArrayList("myCol"));
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)).build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "MetadataTTL / DeletedKeysTTL must have comparison column: myCol in numeric type, found: STRING");
    }

    // time column as comparison column
    upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN));
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false)).build();
    TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);

    // upsert out-of-order configs
    String outOfOrderRecordColumn = "outOfOrderRecordColumn";
    boolean dropOutOfOrderRecord = true;
    schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
        .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
        .addSingleValueDimension(outOfOrderRecordColumn, FieldSpec.DataType.BOOLEAN).build();
    streamConfigs = getStreamConfigs();
    streamConfigs.put("stream.kafka.consumer.type", "simple");

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setDropOutOfOrderRecord(dropOutOfOrderRecord);
    upsertConfig.setOutOfOrderRecordColumn(outOfOrderRecordColumn);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "outOfOrderRecordColumn and dropOutOfOrderRecord shouldn't exist together for upsert table");
    }

    // outOfOrderRecordColumn not of type BOOLEAN
    schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).setPrimaryKeyColumns(Lists.newArrayList("myPkCol"))
        .addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
        .addSingleValueDimension(outOfOrderRecordColumn, FieldSpec.DataType.STRING).build();
    streamConfigs = getStreamConfigs();
    streamConfigs.put("stream.kafka.consumer.type", "simple");

    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setOutOfOrderRecordColumn(outOfOrderRecordColumn);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(), "The outOfOrderRecordColumn must be a single-valued BOOLEAN column");
    }

    // test enableDeletedKeysCompactionConsistency shouldn't exist with metadataTTL
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
    upsertConfig.setMetadataTTL(1.0);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "enableDeletedKeysCompactionConsistency and metadataTTL shouldn't exist together for upsert table");
    }

    // test enableDeletedKeysCompactionConsistency shouldn't exist with enablePreload
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
    upsertConfig.setEnablePreload(true);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "enableDeletedKeysCompactionConsistency and enablePreload shouldn't exist together for upsert table");
    }

    // test enableDeletedKeysCompactionConsistency should exist with deletedKeysTTL
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
    upsertConfig.setDeletedKeysTTL(0);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "enableDeletedKeysCompactionConsistency should exist with deletedKeysTTL for upsert table");
    }

    // test enableDeletedKeysCompactionConsistency should exist with enableSnapshot
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
    upsertConfig.setDeletedKeysTTL(100);
    upsertConfig.setEnableSnapshot(false);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "enableDeletedKeysCompactionConsistency should exist with enableSnapshot for upsert table");
    }

    // test enableDeletedKeysCompactionConsistency should exist with UpsertCompactionTask
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setEnableDeletedKeysCompactionConsistency(true);
    upsertConfig.setDeletedKeysTTL(100);
    upsertConfig.setEnableSnapshot(true);
    tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setStreamConfigs(streamConfigs)
        .setUpsertConfig(upsertConfig).setRoutingConfig(
            new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE, false))
        .build();
    try {
      TableConfigUtils.validateUpsertAndDedupConfig(tableConfig, schema);
    } catch (IllegalStateException e) {
      Assert.assertEquals(e.getMessage(),
          "enableDeletedKeysCompactionConsistency should exist with UpsertCompactionTask for upsert table");
    }
  }