@VisibleForTesting
  static void validateTTLForUpsertConfig(TableConfig tableConfig, Schema schema) {
    UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
    if (upsertConfig == null || (upsertConfig.getMetadataTTL() == 0 && upsertConfig.getDeletedKeysTTL() == 0)) {
      return;
    }

    List<String> comparisonColumns = upsertConfig.getComparisonColumns();
    if (CollectionUtils.isNotEmpty(comparisonColumns)) {
      Preconditions.checkState(comparisonColumns.size() == 1,
          "MetadataTTL / DeletedKeysTTL does not work with multiple comparison columns");
      String comparisonColumn = comparisonColumns.get(0);
      DataType comparisonColumnDataType = schema.getFieldSpecFor(comparisonColumn).getDataType();
      Preconditions.checkState(comparisonColumnDataType.isNumeric(),
          "MetadataTTL / DeletedKeysTTL must have comparison column: %s in numeric type, found: %s", comparisonColumn,
          comparisonColumnDataType);
    }

    if (upsertConfig.getMetadataTTL() > 0) {
      Preconditions.checkState(upsertConfig.isEnableSnapshot(), "Upsert TTL must have snapshot enabled");
    }

    if (upsertConfig.getDeletedKeysTTL() > 0) {
      Preconditions.checkState(upsertConfig.getDeleteRecordColumn() != null,
          "Deleted Keys TTL can only be enabled with deleteRecordColumn set.");
    }
  }
@Test
  public void testValidateTTLConfigForUpsertConfig() {
    // Default comparison column (timestamp)
    Schema schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
            .setPrimaryKeyColumns(Lists.newArrayList("myCol")).build();
    UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setMetadataTTL(3600);
    upsertConfig.setEnableSnapshot(true);
    TableConfig tableConfigWithoutComparisonColumn =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
            .setUpsertConfig(upsertConfig).build();
    TableConfigUtils.validateTTLForUpsertConfig(tableConfigWithoutComparisonColumn, schema);

    // Invalid comparison columns: "myCol"
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setComparisonColumns(Collections.singletonList("myCol"));
    upsertConfig.setEnableSnapshot(true);
    upsertConfig.setMetadataTTL(3600);
    TableConfig tableConfigWithInvalidComparisonColumn =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
            .setUpsertConfig(upsertConfig).build();
    try {
      TableConfigUtils.validateTTLForUpsertConfig(tableConfigWithInvalidComparisonColumn, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      // Expected
    }

    // Invalid comparison columns: multiple comparison columns are not supported for TTL-enabled upsert table.
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setComparisonColumns(Lists.newArrayList(TIME_COLUMN, "myCol"));
    upsertConfig.setEnableSnapshot(true);
    upsertConfig.setMetadataTTL(3600);
    TableConfig tableConfigWithInvalidComparisonColumn2 =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
            .setUpsertConfig(upsertConfig).build();
    try {
      TableConfigUtils.validateTTLForUpsertConfig(tableConfigWithInvalidComparisonColumn2, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      // Expected
    }

    // Invalid config with TTLConfig but Snapshot is not enabled
    schema =
        new Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addSingleValueDimension("myCol", FieldSpec.DataType.STRING)
            .addDateTime(TIME_COLUMN, FieldSpec.DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
            .setPrimaryKeyColumns(Lists.newArrayList("myCol")).build();
    upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
    upsertConfig.setMetadataTTL(3600);
    upsertConfig.setEnableSnapshot(false);
    TableConfig tableConfigWithInvalidTTLConfig =
        new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME).setTimeColumnName(TIME_COLUMN)
            .setUpsertConfig(upsertConfig).build();
    try {
      TableConfigUtils.validateTTLForUpsertConfig(tableConfigWithInvalidTTLConfig, schema);
      Assert.fail();
    } catch (IllegalStateException e) {
      // Expected
    }
  }