env {
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    result_table_name = "fake1"
    row.num = 16
    schema = {
      fields {
        name = "string"
        age = "int"
        created_time = "timestamp"
        destroyed_time = "timestamp"
      }
    }
  }
}

transform {


    SQL {
    source_table_name = "fake1"
    result_table_name = "fake2"
    query = "SELECT name, age, created_time, destroyed_time, CURRENT_DATE() today, DATEDIFF(created_time, destroyed_time, 'SECOND') duration FROM fake1 "
  }

  SQL {
    source_table_name = "fake2"
    result_table_name = "fake3"
    query = "SELECT name, age, today, duration FROM fake2 "
  }
}

sink {
  Console {
    source_table_name = "fake3"
  }
}

no exception