@Test public void windowSqlTrigger() throws Exception { registerOneRowDataStream(); tableEnv.registerTableSink("sink1", new PrintTableSink()); String sql = "insert into sink1 \n" + "SELECT\n" + " user_id,\n" + " cast(TUMBLE_START(proctime, INTERVAL '1' DAY) as varchar) as start_time,\n" + " COUNT(*) as cnt\n" + "FROM tb0\n" + "GROUP BY user_id, TUMBLE(proctime, INTERVAL '1' DAY)\n" + "trigger WITH EVERY 5 SECONDS"; streamSqlBuilder.buildStreamBySql(sql); execEnv.execute(); }
public void buildStreamBySql(String sql) { FlinkSqlParser flinkSqlParser = FlinkSqlParser.builder() .setTableEnv(tableEnv) .setConnectorStore(connectorStore) .build(); Statement statement = sqlParser.createStatement(sql); if (statement instanceof CreateStreamAsSelect) { CreateStreamAsSelect createStreamAsSelect = (CreateStreamAsSelect) statement; Table table = tableEnv.sqlQuery(createStreamAsSelect.getViewSql()); RowTypeInfo rowTypeInfo = (RowTypeInfo) table.getSchema().toRowType(); DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class); stream.getTransformation().setOutputType(rowTypeInfo); registerStreamTable(stream, createStreamAsSelect.getName(), createStreamAsSelect.getWatermark(), ImmutableList.of()); } else if (statement instanceof CreateTable) { if (((CreateTable) statement).getType() == CreateTable.Type.BATCH) { batchTables.add((CreateTable) statement); } else { createStreamTable((CreateTable) statement); } } else if (statement instanceof CreateFunction) { createFunction((CreateFunction) statement); } else if (statement instanceof SelectQuery) { SelectQuery selectQuery = (SelectQuery) statement; String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); try { tableEnv.toAppendStream(table, Row.class).print(); } catch (ValidationException e) { checkState(e.getMessage().equals("Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages."), "sylph and flink versions are not compatible, please feedback to the community"); tableEnv.toRetractStream(table, Row.class).print(); } } else if (statement instanceof InsertInto) { InsertInto insertInto = (InsertInto) statement; SelectQuery selectQuery = insertInto.getSelectQuery(); String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); if (table == null) { throw new ParsingException("table is null"); } try { table.insertInto(insertInto.getTableName()); } catch (TableException e) { checkState(e.getMessage().equals("AppendStreamTableSink requires that Table has only insert changes."), "sylph and flink versions are not compatible, please feedback to the community"); DataStream<Row> retractStream = tableEnv.toRetractStream(table, Row.class) .filter(x -> x.f0).map(x -> x.f1).returns(table.getSchema().toRowType()); tableEnv.fromDataStream(retractStream).insertInto(insertInto.getTableName()); } } else { throw new IllegalArgumentException("this driver class " + statement.getClass() + " have't support!"); } }
StreamSqlBuilder { public void buildStreamBySql(String sql) { FlinkSqlParser flinkSqlParser = FlinkSqlParser.builder() .setTableEnv(tableEnv) .setConnectorStore(connectorStore) .build(); Statement statement = sqlParser.createStatement(sql); if (statement instanceof CreateStreamAsSelect) { CreateStreamAsSelect createStreamAsSelect = (CreateStreamAsSelect) statement; Table table = tableEnv.sqlQuery(createStreamAsSelect.getViewSql()); RowTypeInfo rowTypeInfo = (RowTypeInfo) table.getSchema().toRowType(); DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class); stream.getTransformation().setOutputType(rowTypeInfo); registerStreamTable(stream, createStreamAsSelect.getName(), createStreamAsSelect.getWatermark(), ImmutableList.of()); } else if (statement instanceof CreateTable) { if (((CreateTable) statement).getType() == CreateTable.Type.BATCH) { batchTables.add((CreateTable) statement); } else { createStreamTable((CreateTable) statement); } } else if (statement instanceof CreateFunction) { createFunction((CreateFunction) statement); } else if (statement instanceof SelectQuery) { SelectQuery selectQuery = (SelectQuery) statement; String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); try { tableEnv.toAppendStream(table, Row.class).print(); } catch (ValidationException e) { checkState(e.getMessage().equals("Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages."), "sylph and flink versions are not compatible, please feedback to the community"); tableEnv.toRetractStream(table, Row.class).print(); } } else if (statement instanceof InsertInto) { InsertInto insertInto = (InsertInto) statement; SelectQuery selectQuery = insertInto.getSelectQuery(); String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); if (table == null) { throw new ParsingException("table is null"); } try { table.insertInto(insertInto.getTableName()); } catch (TableException e) { checkState(e.getMessage().equals("AppendStreamTableSink requires that Table has only insert changes."), "sylph and flink versions are not compatible, please feedback to the community"); DataStream<Row> retractStream = tableEnv.toRetractStream(table, Row.class) .filter(x -> x.f0).map(x -> x.f1).returns(table.getSchema().toRowType()); tableEnv.fromDataStream(retractStream).insertInto(insertInto.getTableName()); } } else { throw new IllegalArgumentException("this driver class " + statement.getClass() + " have't support!"); } } }
StreamSqlBuilder { public void buildStreamBySql(String sql) { FlinkSqlParser flinkSqlParser = FlinkSqlParser.builder() .setTableEnv(tableEnv) .setConnectorStore(connectorStore) .build(); Statement statement = sqlParser.createStatement(sql); if (statement instanceof CreateStreamAsSelect) { CreateStreamAsSelect createStreamAsSelect = (CreateStreamAsSelect) statement; Table table = tableEnv.sqlQuery(createStreamAsSelect.getViewSql()); RowTypeInfo rowTypeInfo = (RowTypeInfo) table.getSchema().toRowType(); DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class); stream.getTransformation().setOutputType(rowTypeInfo); registerStreamTable(stream, createStreamAsSelect.getName(), createStreamAsSelect.getWatermark(), ImmutableList.of()); } else if (statement instanceof CreateTable) { if (((CreateTable) statement).getType() == CreateTable.Type.BATCH) { batchTables.add((CreateTable) statement); } else { createStreamTable((CreateTable) statement); } } else if (statement instanceof CreateFunction) { createFunction((CreateFunction) statement); } else if (statement instanceof SelectQuery) { SelectQuery selectQuery = (SelectQuery) statement; String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); try { tableEnv.toAppendStream(table, Row.class).print(); } catch (ValidationException e) { checkState(e.getMessage().equals("Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages."), "sylph and flink versions are not compatible, please feedback to the community"); tableEnv.toRetractStream(table, Row.class).print(); } } else if (statement instanceof InsertInto) { InsertInto insertInto = (InsertInto) statement; SelectQuery selectQuery = insertInto.getSelectQuery(); String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); if (table == null) { throw new ParsingException("table is null"); } try { table.insertInto(insertInto.getTableName()); } catch (TableException e) { checkState(e.getMessage().equals("AppendStreamTableSink requires that Table has only insert changes."), "sylph and flink versions are not compatible, please feedback to the community"); DataStream<Row> retractStream = tableEnv.toRetractStream(table, Row.class) .filter(x -> x.f0).map(x -> x.f1).returns(table.getSchema().toRowType()); tableEnv.fromDataStream(retractStream).insertInto(insertInto.getTableName()); } } else { throw new IllegalArgumentException("this driver class " + statement.getClass() + " have't support!"); } } StreamSqlBuilder(
            StreamTableEnvironment tableEnv,
            ConnectorStore connectorStore,
            AntlrSqlParser sqlParser
    ); }
StreamSqlBuilder { public void buildStreamBySql(String sql) { FlinkSqlParser flinkSqlParser = FlinkSqlParser.builder() .setTableEnv(tableEnv) .setConnectorStore(connectorStore) .build(); Statement statement = sqlParser.createStatement(sql); if (statement instanceof CreateStreamAsSelect) { CreateStreamAsSelect createStreamAsSelect = (CreateStreamAsSelect) statement; Table table = tableEnv.sqlQuery(createStreamAsSelect.getViewSql()); RowTypeInfo rowTypeInfo = (RowTypeInfo) table.getSchema().toRowType(); DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class); stream.getTransformation().setOutputType(rowTypeInfo); registerStreamTable(stream, createStreamAsSelect.getName(), createStreamAsSelect.getWatermark(), ImmutableList.of()); } else if (statement instanceof CreateTable) { if (((CreateTable) statement).getType() == CreateTable.Type.BATCH) { batchTables.add((CreateTable) statement); } else { createStreamTable((CreateTable) statement); } } else if (statement instanceof CreateFunction) { createFunction((CreateFunction) statement); } else if (statement instanceof SelectQuery) { SelectQuery selectQuery = (SelectQuery) statement; String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); try { tableEnv.toAppendStream(table, Row.class).print(); } catch (ValidationException e) { checkState(e.getMessage().equals("Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages."), "sylph and flink versions are not compatible, please feedback to the community"); tableEnv.toRetractStream(table, Row.class).print(); } } else if (statement instanceof InsertInto) { InsertInto insertInto = (InsertInto) statement; SelectQuery selectQuery = insertInto.getSelectQuery(); String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); if (table == null) { throw new ParsingException("table is null"); } try { table.insertInto(insertInto.getTableName()); } catch (TableException e) { checkState(e.getMessage().equals("AppendStreamTableSink requires that Table has only insert changes."), "sylph and flink versions are not compatible, please feedback to the community"); DataStream<Row> retractStream = tableEnv.toRetractStream(table, Row.class) .filter(x -> x.f0).map(x -> x.f1).returns(table.getSchema().toRowType()); tableEnv.fromDataStream(retractStream).insertInto(insertInto.getTableName()); } } else { throw new IllegalArgumentException("this driver class " + statement.getClass() + " have't support!"); } } StreamSqlBuilder(
            StreamTableEnvironment tableEnv,
            ConnectorStore connectorStore,
            AntlrSqlParser sqlParser
    ); void buildStreamBySql(String sql); }
StreamSqlBuilder { public void buildStreamBySql(String sql) { FlinkSqlParser flinkSqlParser = FlinkSqlParser.builder() .setTableEnv(tableEnv) .setConnectorStore(connectorStore) .build(); Statement statement = sqlParser.createStatement(sql); if (statement instanceof CreateStreamAsSelect) { CreateStreamAsSelect createStreamAsSelect = (CreateStreamAsSelect) statement; Table table = tableEnv.sqlQuery(createStreamAsSelect.getViewSql()); RowTypeInfo rowTypeInfo = (RowTypeInfo) table.getSchema().toRowType(); DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class); stream.getTransformation().setOutputType(rowTypeInfo); registerStreamTable(stream, createStreamAsSelect.getName(), createStreamAsSelect.getWatermark(), ImmutableList.of()); } else if (statement instanceof CreateTable) { if (((CreateTable) statement).getType() == CreateTable.Type.BATCH) { batchTables.add((CreateTable) statement); } else { createStreamTable((CreateTable) statement); } } else if (statement instanceof CreateFunction) { createFunction((CreateFunction) statement); } else if (statement instanceof SelectQuery) { SelectQuery selectQuery = (SelectQuery) statement; String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); try { tableEnv.toAppendStream(table, Row.class).print(); } catch (ValidationException e) { checkState(e.getMessage().equals("Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages."), "sylph and flink versions are not compatible, please feedback to the community"); tableEnv.toRetractStream(table, Row.class).print(); } } else if (statement instanceof InsertInto) { InsertInto insertInto = (InsertInto) statement; SelectQuery selectQuery = insertInto.getSelectQuery(); String pushDownQuery = selectQuery.getQuery(); Table table = flinkSqlParser.parser(pushDownQuery, ImmutableList.copyOf(batchTables)); if (table == null) { throw new ParsingException("table is null"); } try { table.insertInto(insertInto.getTableName()); } catch (TableException e) { checkState(e.getMessage().equals("AppendStreamTableSink requires that Table has only insert changes."), "sylph and flink versions are not compatible, please feedback to the community"); DataStream<Row> retractStream = tableEnv.toRetractStream(table, Row.class) .filter(x -> x.f0).map(x -> x.f1).returns(table.getSchema().toRowType()); tableEnv.fromDataStream(retractStream).insertInto(insertInto.getTableName()); } } else { throw new IllegalArgumentException("this driver class " + statement.getClass() + " have't support!"); } } StreamSqlBuilder(
            StreamTableEnvironment tableEnv,
            ConnectorStore connectorStore,
            AntlrSqlParser sqlParser
    ); void buildStreamBySql(String sql);  }