BeamSqlLine { @SuppressWarnings("argument.type.incompatible") 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, Charsets.UTF_8.name())); } if (errorStream != null) { sqlLine.setErrorStream(new PrintStream(errorStream, false, Charsets.UTF_8.name())); } return sqlLine.begin(modifiedArgs, inputStream, true); }  static void main(String[] args);  }
@Test public void testSqlLine_select() throws Exception { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); String[] args = buildArgs("SELECT 3, 'hello', DATE '2018-05-28';"); BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null); List<List<String>> lines = toLines(byteArrayOutputStream); assertThat( Arrays.asList(Arrays.asList("3", "hello", "2018-05-28")), everyItem(IsIn.isOneOf(lines.toArray()))); }
@Test public void testSqlLine_selectFromTable() throws Exception { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); String[] args = buildArgs( "CREATE EXTERNAL TABLE table_test (col_a VARCHAR, col_b VARCHAR, " + "col_c VARCHAR, col_x TINYINT, col_y INT, col_z BIGINT) TYPE 'test';", "INSERT INTO table_test VALUES ('a', 'b', 'c', 1, 2, 3);", "SELECT * FROM table_test;"); BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null); List<List<String>> lines = toLines(byteArrayOutputStream); assertThat( Arrays.asList( Arrays.asList("col_a", "col_b", "col_c", "col_x", "col_y", "col_z"), Arrays.asList("a", "b", "c", "1", "2", "3")), everyItem(IsIn.isOneOf(lines.toArray()))); }
@Test public void testSqlLine_insertSelect() throws Exception { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); String[] args = buildArgs( "CREATE EXTERNAL TABLE table_test (col_a VARCHAR, col_b VARCHAR) TYPE 'test';", "INSERT INTO table_test SELECT '3', 'hello';", "SELECT * FROM table_test;"); BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null); List<List<String>> lines = toLines(byteArrayOutputStream); assertThat( Arrays.asList(Arrays.asList("3", "hello")), everyItem(IsIn.isOneOf(lines.toArray()))); }
@Test public void testSqlLine_GroupBy() throws Exception { ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); String[] args = buildArgs( "CREATE EXTERNAL TABLE table_test (col_a VARCHAR, col_b VARCHAR) TYPE 'test';", "INSERT INTO table_test SELECT '3', 'foo';", "INSERT INTO table_test SELECT '3', 'bar';", "INSERT INTO table_test SELECT '4', 'foo';", "SELECT col_a, count(*) FROM table_test GROUP BY col_a;"); BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null); List<List<String>> lines = toLines(byteArrayOutputStream); assertThat( Arrays.asList(Arrays.asList("3", "2"), Arrays.asList("4", "1")), everyItem(IsIn.isOneOf(lines.toArray()))); }
@Test public void testSqlLine_fixedWindow() 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 '3', TIMESTAMP '2018-07-01 21:26:07';", "SELECT TUMBLE_START(col_b, INTERVAL '1' SECOND), count(*) FROM table_test " + "GROUP BY TUMBLE(col_b, INTERVAL '1' SECOND);"); BeamSqlLine.runSqlLine(args, null, byteArrayOutputStream, null); List<List<String>> lines = toLines(byteArrayOutputStream); assertThat( Arrays.asList( Arrays.asList("2018-07-01 21:26:06", "1"), Arrays.asList("2018-07-01 21:26:07", "1")), everyItem(IsIn.isOneOf(lines.toArray()))); }
@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(IsIn.isOneOf(lines.toArray()))); }