@Test
public void debugWatermark() throws Exception {

    String sql = "create table t1( " +
            "data_time STRING ," +
            "row1_time AS to_timestamp(data_time)," +
            "WATERMARK FOR row1_time AS row1_time - INTERVAL '5' SECOND " +
            ") with ( 'connector' = 'collection','data'='[]');" +
            "create table t2(stime TIMESTAMP(3),cnt bigint) with ( 'connector' = 'print');" +
            "insert into t2 select TUMBLE_START(row1_time, INTERVAL '1' MINUTE) as stime,count(1) cnt from t1 group by TUMBLE(row1_time, INTERVAL '1' MINUTE);";
    List<String> sourceData = Stream.of(
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:01").toJSONString(),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:02").toJSONString(),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:03").toJSONString(),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:01:01").toJSONString()
    ).collect(Collectors.toList());
    SqlDebugConfig sqlDebugConfig = new SqlDebugConfig();
    HashMap<String, SqlDebugConfig.SourceConfig> sourceConfigMap = new HashMap<>();
    sourceConfigMap.put("t1", new SqlDebugConfig.SourceConfig(sourceData));
    sqlDebugConfig.setSourceConfigMap(sourceConfigMap);
    sqlDebugConfig.setSql(sql);
    log.info("sqlDebugConfig={}", JSON.toJSONString(sqlDebugConfig));
    Map<String, List<String>> debugResult = SqlDebugDriver.debug(sqlDebugConfig);
    List<String> t2 = debugResult.get("t2");
    assert t2 != null && t2.size() > 0;
    assert "{\"stime\":\"2020-01-01 12:00:00\",\"cnt\":3}".equals(t2.get(0));
    assert "{\"stime\":\"2020-01-01 12:01:00\",\"cnt\":1}".equals(t2.get(1));
}