@SuppressWarnings("argument")
  static Status runSqlLine(
      String[] args,
      @Nullable InputStream inputStream,
      @Nullable OutputStream outputStream,
      @Nullable OutputStream errorStream)
      throws IOException {
    String[] modifiedArgs = checkConnectionArgs(args);
    SqlLine sqlLine = new SqlLine();

    if (outputStream != null) {
      sqlLine.setOutputStream(new PrintStream(outputStream, false, StandardCharsets.UTF_8.name()));
    }

    if (errorStream != null) {
      sqlLine.setErrorStream(new PrintStream(errorStream, false, StandardCharsets.UTF_8.name()));
    }

    return sqlLine.begin(modifiedArgs, inputStream, true);
  }
@Test
  public void testSqlLine_slidingWindow() throws Exception {
    ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
    String[] args =
        buildArgs(
            "CREATE EXTERNAL TABLE table_test (col_a VARCHAR, col_b TIMESTAMP) TYPE 'test';",
            "INSERT INTO table_test SELECT '3', TIMESTAMP '2018-07-01 21:26:06';",
            "INSERT INTO table_test SELECT '4', TIMESTAMP '2018-07-01 21:26:07';",
            "INSERT INTO table_test SELECT '6', TIMESTAMP '2018-07-01 21:26:08';",
            "INSERT INTO table_test SELECT '7', TIMESTAMP '2018-07-01 21:26:09';",
            "SELECT HOP_END(col_b, INTERVAL '1' SECOND, INTERVAL '2' SECOND), count(*) FROM "
                + "table_test GROUP BY HOP(col_b, INTERVAL '1' SECOND, INTERVAL '2' SECOND);");

    BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null);

    List<List<String>> lines = toLines(byteArrayOutputStream);
    assertThat(
        Arrays.asList(
            Arrays.asList("2018-07-01 21:26:07", "1"),
            Arrays.asList("2018-07-01 21:26:08", "2"),
            Arrays.asList("2018-07-01 21:26:09", "2"),
            Arrays.asList("2018-07-01 21:26:10", "2"),
            Arrays.asList("2018-07-01 21:26:11", "1")),
        everyItem(is(oneOf(lines.toArray()))));
  }