@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)); }
public static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig) throws Exception { Preconditions.checkNotNull(sqlDebugConfig, "sqlDebugConfig should be not null"); Preconditions.checkNotNull(sqlDebugConfig.getSql(), "sql should be not null"); String identifier = UUID.randomUUID().toString(); String debugSql = handleDebugSql(identifier, sqlDebugConfig.getSql(), sqlDebugConfig); log.debug("start sql debug,sql:{}", debugSql); SqlConfig config = SqlConfig.builder().sql(debugSql).jobName("sql_job_debug_test").build(); SqlJob sqlJob = new SqlJob(config); try { CollectionDataWarehouse.registerLock(identifier); sqlJob.start(); return CollectionDataWarehouse.getData(identifier); } finally { CollectionDataWarehouse.remove(identifier); } }
SqlDebugDriver { public static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig) throws Exception { Preconditions.checkNotNull(sqlDebugConfig, "sqlDebugConfig should be not null"); Preconditions.checkNotNull(sqlDebugConfig.getSql(), "sql should be not null"); String identifier = UUID.randomUUID().toString(); String debugSql = handleDebugSql(identifier, sqlDebugConfig.getSql(), sqlDebugConfig); log.debug("start sql debug,sql:{}", debugSql); SqlConfig config = SqlConfig.builder().sql(debugSql).jobName("sql_job_debug_test").build(); SqlJob sqlJob = new SqlJob(config); try { CollectionDataWarehouse.registerLock(identifier); sqlJob.start(); return CollectionDataWarehouse.getData(identifier); } finally { CollectionDataWarehouse.remove(identifier); } } }
SqlDebugDriver { public static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig) throws Exception { Preconditions.checkNotNull(sqlDebugConfig, "sqlDebugConfig should be not null"); Preconditions.checkNotNull(sqlDebugConfig.getSql(), "sql should be not null"); String identifier = UUID.randomUUID().toString(); String debugSql = handleDebugSql(identifier, sqlDebugConfig.getSql(), sqlDebugConfig); log.debug("start sql debug,sql:{}", debugSql); SqlConfig config = SqlConfig.builder().sql(debugSql).jobName("sql_job_debug_test").build(); SqlJob sqlJob = new SqlJob(config); try { CollectionDataWarehouse.registerLock(identifier); sqlJob.start(); return CollectionDataWarehouse.getData(identifier); } finally { CollectionDataWarehouse.remove(identifier); } }  }
SqlDebugDriver { public static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig) throws Exception { Preconditions.checkNotNull(sqlDebugConfig, "sqlDebugConfig should be not null"); Preconditions.checkNotNull(sqlDebugConfig.getSql(), "sql should be not null"); String identifier = UUID.randomUUID().toString(); String debugSql = handleDebugSql(identifier, sqlDebugConfig.getSql(), sqlDebugConfig); log.debug("start sql debug,sql:{}", debugSql); SqlConfig config = SqlConfig.builder().sql(debugSql).jobName("sql_job_debug_test").build(); SqlJob sqlJob = new SqlJob(config); try { CollectionDataWarehouse.registerLock(identifier); sqlJob.start(); return CollectionDataWarehouse.getData(identifier); } finally { CollectionDataWarehouse.remove(identifier); } }  static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig); }
SqlDebugDriver { public static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig) throws Exception { Preconditions.checkNotNull(sqlDebugConfig, "sqlDebugConfig should be not null"); Preconditions.checkNotNull(sqlDebugConfig.getSql(), "sql should be not null"); String identifier = UUID.randomUUID().toString(); String debugSql = handleDebugSql(identifier, sqlDebugConfig.getSql(), sqlDebugConfig); log.debug("start sql debug,sql:{}", debugSql); SqlConfig config = SqlConfig.builder().sql(debugSql).jobName("sql_job_debug_test").build(); SqlJob sqlJob = new SqlJob(config); try { CollectionDataWarehouse.registerLock(identifier); sqlJob.start(); return CollectionDataWarehouse.getData(identifier); } finally { CollectionDataWarehouse.remove(identifier); } }  static Map<String, List<String>> debug(SqlDebugConfig sqlDebugConfig);  }