test_class.declarator: sql
test_class.modifier: public static
test_class.original_string: public static String sql;
test_class.type: String
test_class.var_name: sql
test_class.file: plink-sql/plink-sql-core/src/test/java/com/github/hairless/plink/sql/SqlJobTest.java
test_class.identifier: SqlJobTest
test_case.body: @Test
      public void sqlJobWatermarkTest() {
          List<JSONObject> sourceData = Stream.of(
                  new JSONObject().fluentPut("data_time", "2020-01-01 12:00:01"),
                  new JSONObject().fluentPut("data_time", "2020-01-01 12:00:02"),
                  new JSONObject().fluentPut("data_time", "2020-01-01 12:00:03"),
                  new JSONObject().fluentPut("data_time", "2020-01-01 12:01:01")
          ).collect(Collectors.toList());
          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'='" + JSON.toJSONString(sourceData) + "');" +
                  "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);;";
  
  
          Exception exception = null;
          try {
              SqlConfig config = SqlConfig.builder().sql(sql).jobName("sql_job_test").build();
              SqlJob sqlJob = new SqlJob(config);
              sqlJob.start();
          } catch (Exception e) {
              log.error("sqlJobTest error", e);
              exception = e;
          }
          assert exception == null;
      }
test_case.class_method_signature: SqlJobTest.sqlJobWatermarkTest()
test_case.full_signature: @Test public void sqlJobWatermarkTest()
test_case.identifier: sqlJobWatermarkTest
test_case.invocations: collect of fluentPut fluentPut fluentPut fluentPut toList toJSONString build jobName sql builder start error
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void sqlJobWatermarkTest()
focal_class.declarator: sqlConfig
focal_class.modifier: private final
focal_class.original_string: private final SqlConfig sqlConfig;
focal_class.type: SqlConfig
focal_class.var_name: sqlConfig
focal_class.file: plink-sql/plink-sql-core/src/main/java/com/github/hairless/plink/sql/SqlJob.java
focal_class.identifier: SqlJob
focal_class.class_method_signature: SqlJob.SqlJob(SqlConfig sqlConfig)
focal_class.full_signature: public  SqlJob(SqlConfig sqlConfig)
focal_class.identifier: SqlJob
focal_class.modifiers: public
focal_class.parameters: (SqlConfig sqlConfig)
focal_class.signature:  SqlJob(SqlConfig sqlConfig)
focal_class.class_method_signature: SqlJob.start()
focal_class.full_signature: public JobExecutionResult start()
focal_class.identifier: start
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: JobExecutionResult
focal_class.signature: JobExecutionResult start()
focal_method.body: public JobExecutionResult start() throws Exception {
          log.info("sql job {} staring...", sqlConfig.getJobName());
          StreamExecutionEnvironment env = new LocalStreamEnvironment();
          EnvironmentSettings settings = EnvironmentSettings.newInstance()
                  .useBlinkPlanner().inStreamingMode().build();
          TableEnvironmentImpl tEnv = (TableEnvironmentImpl) StreamTableEnvironment.create(env, settings);
  
          SqlParser sqlParser = SqlParser.create(sqlConfig.getSql(), PlinkSqlParser.sqlParserConfig);
          SqlNodeList sqlNodes = sqlParser.parseStmtList();
          sqlNodes.forEach(sqlNode -> {
              if (sqlNode instanceof SqlSetOption) {
                  String name = ((SqlSetOption) sqlNode).getName().getSimple();
                  String value = ((SqlSetOption) sqlNode).getValue().toString();
                  tEnv.getConfig().getConfiguration().setString(name, value);
              } else {
                  String sql = sqlNode.toSqlString(SkipAnsiCheckSqlDialect.DEFAULT).getSql();
                  tEnv.sqlUpdate(sql);
              }
          });
          return tEnv.execute(sqlConfig.getJobName());
      }
focal_method.class_method_signature: SqlJob.start()
focal_method.full_signature: public JobExecutionResult start()
focal_method.identifier: start
focal_method.invocations: info getJobName build inStreamingMode useBlinkPlanner newInstance create create getSql parseStmtList forEach getSimple getName toString getValue setString getConfiguration getConfig getSql toSqlString sqlUpdate execute getJobName
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: JobExecutionResult
focal_method.signature: JobExecutionResult start()