@Nullable
  public static TimeHandlerConfig getTimeHandlerConfig(TableConfig tableConfig, Schema schema,
      Map<String, String> taskConfig) {
    String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
    if (timeColumn == null) {
      return null;
    }
    DateTimeFieldSpec fieldSpec = schema.getSpecForTimeColumn(timeColumn);
    Preconditions.checkState(fieldSpec != null, "No valid spec found for time column: %s in schema for table: %s",
        timeColumn, tableConfig.getTableName());

    TimeHandlerConfig.Builder timeHandlerConfigBuilder = new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH);

    String windowStartMs = taskConfig.get(MergeTask.WINDOW_START_MS_KEY);
    String windowEndMs = taskConfig.get(MergeTask.WINDOW_END_MS_KEY);
    if (windowStartMs != null && windowEndMs != null) {
      timeHandlerConfigBuilder.setTimeRange(Long.parseLong(windowStartMs), Long.parseLong(windowEndMs))
          .setNegateWindowFilter(Boolean.parseBoolean(taskConfig.get(MergeTask.NEGATE_WINDOW_FILTER)));
    }

    String roundBucketTimePeriod = taskConfig.get(MergeTask.ROUND_BUCKET_TIME_PERIOD_KEY);
    if (roundBucketTimePeriod != null) {
      timeHandlerConfigBuilder.setRoundBucketMs(TimeUtils.convertPeriodToMillis(roundBucketTimePeriod));
    }

    String partitionBucketTimePeriod = taskConfig.get(MergeTask.PARTITION_BUCKET_TIME_PERIOD_KEY);
    if (partitionBucketTimePeriod != null) {
      timeHandlerConfigBuilder.setPartitionBucketMs(TimeUtils.convertPeriodToMillis(partitionBucketTimePeriod));
    }

    return timeHandlerConfigBuilder.build();
  }
@Test
  public void testGetTimeHandlerConfig() {
    TableConfig tableConfig =
        new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").setTimeColumnName("dateTime").build();
    Schema schema = new Schema.SchemaBuilder()
        .addDateTime("dateTime", DataType.LONG, "1:SECONDS:SIMPLE_DATE_FORMAT:yyyyMMddHHmmss", "1:SECONDS").build();
    Map<String, String> taskConfig = new HashMap<>();
    long expectedWindowStartMs = 1625097600000L;
    long expectedWindowEndMs = 1625184000000L;
    taskConfig.put(MergeTask.WINDOW_START_MS_KEY, Long.toString(expectedWindowStartMs));
    taskConfig.put(MergeTask.WINDOW_END_MS_KEY, Long.toString(expectedWindowEndMs));
    long expectedRoundBucketMs = 6 * 3600 * 1000;
    taskConfig.put(MergeTask.ROUND_BUCKET_TIME_PERIOD_KEY, "6h");
    long expectedPartitionBucketMs = 24 * 3600 * 1000;
    taskConfig.put(MergeTask.PARTITION_BUCKET_TIME_PERIOD_KEY, "1d");

    TimeHandlerConfig timeHandlerConfig = MergeTaskUtils.getTimeHandlerConfig(tableConfig, schema, taskConfig);
    assertNotNull(timeHandlerConfig);
    assertEquals(timeHandlerConfig.getType(), TimeHandler.Type.EPOCH);
    assertEquals(timeHandlerConfig.getStartTimeMs(), expectedWindowStartMs);
    assertEquals(timeHandlerConfig.getEndTimeMs(), expectedWindowEndMs);
    assertEquals(timeHandlerConfig.getRoundBucketMs(), expectedRoundBucketMs);
    assertEquals(timeHandlerConfig.getPartitionBucketMs(), expectedPartitionBucketMs);

    // No time column in table config
    TableConfig tableConfigWithoutTimeColumn =
        new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build();
    assertNull(MergeTaskUtils.getTimeHandlerConfig(tableConfigWithoutTimeColumn, schema, taskConfig));

    // Time column does not exist in schema
    Schema schemaWithoutTimeColumn = new Schema.SchemaBuilder().build();
    try {
      MergeTaskUtils.getTimeHandlerConfig(tableConfig, schemaWithoutTimeColumn, taskConfig);
      fail();
    } catch (IllegalStateException e) {
      // Expected
    }
  }