Queries

SELECT statements and VALUES statements are specified with the sqlQuery() method of the TableEnvironment. The method returns the result of the SELECT statement (or the VALUES statements) as a Table. A Table can be used in subsequent SQL and Table API queries, be converted into a DataStream, or written to a TableSink. SQL and Table API queries can be seamlessly mixed and are holistically optimized and translated into a single program.

In order to access a table in a SQL query, it must be registered in the TableEnvironment. A table can be registered from a TableSource, Table, CREATE TABLE statement, DataStream. Alternatively, users can also register catalogs in a TableEnvironment to specify the location of the data sources.

For convenience, Table.toString() automatically registers the table under a unique name in its TableEnvironment and returns the name. So, Table objects can be directly inlined into SQL queries as shown in the examples below.

Note: Queries that include unsupported SQL features cause a TableException. The supported features of SQL on batch and streaming tables are listed in the following sections.

Specifying a Query

The following examples show how to specify a SQL queries on registered and inlined tables.

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  4. // ingest a DataStream from an external source
  5. DataStream<Tuple3<Long, String, Integer>> ds = env.addSource(...);
  7. // SQL query with an inlined (unregistered) table
  8. Table table = tableEnv.fromDataStream(ds, $("user"), $("product"), $("amount"));
  9. Table result = tableEnv.sqlQuery(
  10.   "SELECT SUM(amount) FROM " + table + " WHERE product LIKE '%Rubber%'");
  12. // SQL query with a registered table
  13. // register the DataStream as view "Orders"
  14. tableEnv.createTemporaryView("Orders", ds, $("user"), $("product"), $("amount"));
  15. // run a SQL query on the Table and retrieve the result as a new Table
  16. Table result2 = tableEnv.sqlQuery(
  17.   "SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");
  19. // create and register a TableSink
  20. final Schema schema = new Schema()
  21.     .field("product", DataTypes.STRING())
  22.     .field("amount", DataTypes.INT());
  24. tableEnv.connect(new FileSystem().path("/path/to/file"))
  25.     .withFormat(...)
  26.     .withSchema(schema)
  27.     .createTemporaryTable("RubberOrders");
  29. // run an INSERT SQL on the Table and emit the result to the TableSink
  30. tableEnv.executeSql(
  31.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment
  2. val tableEnv = StreamTableEnvironment.create(env)
  4. // read a DataStream from an external source
  5. val ds: DataStream[(Long, String, Integer)] = env.addSource(...)
  7. // SQL query with an inlined (unregistered) table
  8. val table = ds.toTable(tableEnv, $"user", $"product", $"amount")
  9. val result = tableEnv.sqlQuery(
  10.   s"SELECT SUM(amount) FROM $table WHERE product LIKE '%Rubber%'")
  12. // SQL query with a registered table
  13. // register the DataStream under the name "Orders"
  14. tableEnv.createTemporaryView("Orders", ds, $"user", $"product", $"amount")
  15. // run a SQL query on the Table and retrieve the result as a new Table
  16. val result2 = tableEnv.sqlQuery(
  17.   "SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")
  19. // create and register a TableSink
  20. val schema = new Schema()
  21.     .field("product", DataTypes.STRING())
  22.     .field("amount", DataTypes.INT())
  24. tableEnv.connect(new FileSystem().path("/path/to/file"))
  25.     .withFormat(...)
  26.     .withSchema(schema)
  27.     .createTemporaryTable("RubberOrders")
  29. // run an INSERT SQL on the Table and emit the result to the TableSink
  30. tableEnv.executeSql(
  31.   "INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")

Python

  1. env = StreamExecutionEnvironment.get_execution_environment()
  2. table_env = StreamTableEnvironment.create(env)
  4. # SQL query with an inlined (unregistered) table
  5. # elements data type: BIGINT, STRING, BIGINT
  6. table = table_env.from_elements(..., ['user', 'product', 'amount'])
  7. result = table_env \
  8.     .sql_query("SELECT SUM(amount) FROM %s WHERE product LIKE '%%Rubber%%'" % table)
  10. # create and register a TableSink
  11. t_env.connect(FileSystem().path("/path/to/file")))
  12.     .with_format(Csv()
  13.                  .field_delimiter(',')
  14.                  .deriveSchema())
  15.     .with_schema(Schema()
  16.                  .field("product", DataTypes.STRING())
  17.                  .field("amount", DataTypes.BIGINT()))
  18.     .create_temporary_table("RubberOrders")
  20. # run an INSERT SQL on the Table and emit the result to the TableSink
  21. table_env \
  22.     .execute_sql("INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'")

Execute a Query

A SELECT statement or a VALUES statement can be executed to collect the content to local through the TableEnvironment.executeSql() method. The method returns the result of the SELECT statement (or the VALUES statement) as a TableResult. Similar to a SELECT statement, a Table object can be executed using the Table.execute() method to collect the content of the query to the local client. TableResult.collect() method returns a closeable row iterator. The select job will not be finished unless all result data has been collected. We should actively close the job to avoid resource leak through the CloseableIterator#close() method. We can also print the select result to client console through the TableResult.print() method. The result data in TableResult can be accessed only once. Thus, collect() and print() must not be called after each other.

TableResult.collect() and TableResult.print() have slightly different behaviors under different checkpointing settings (to enable checkpointing for a streaming job, see checkpointing config).

  • For batch jobs or streaming jobs without checkpointing, TableResult.collect() and TableResult.print() have neither exactly-once nor at-least-once guarantee. Query results are immediately accessible by the clients once they’re produced, but exceptions will be thrown when the job fails and restarts.
  • For streaming jobs with exactly-once checkpointing, TableResult.collect() and TableResult.print() guarantee an end-to-end exactly-once record delivery. A result will be accessible by clients only after its corresponding checkpoint completes.

  • For streaming jobs with at-least-once checkpointing, TableResult.collect() and TableResult.print() guarantee an end-to-end at-least-once record delivery. Query results are immediately accessible by the clients once they’re produced, but it is possible for the same result to be delivered multiple times.

    Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2. StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
  4. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)");
  6. // execute SELECT statement
  7. TableResult tableResult1 = tableEnv.executeSql("SELECT * FROM Orders");
  8. // use try-with-resources statement to make sure the iterator will be closed automatically
  9. try (CloseableIterator<Row> it = tableResult1.collect()) {
  10.     while(it.hasNext()) {
  11.         Row row = it.next();
  12.         // handle row
  13.     }
  14. }
  16. // execute Table
  17. TableResult tableResult2 = tableEnv.sqlQuery("SELECT * FROM Orders").execute();
  18. tableResult2.print();

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment()
  2. val tableEnv = StreamTableEnvironment.create(env, settings)
  3. // enable checkpointing
  4. tableEnv.getConfig.getConfiguration.set(
  5.   ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)
  6. tableEnv.getConfig.getConfiguration.set(
  7.   ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10))
  9. tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  11. // execute SELECT statement
  12. val tableResult1 = tableEnv.executeSql("SELECT * FROM Orders")
  13. val it = tableResult1.collect()
  14. try while (it.hasNext) {
  15.   val row = it.next
  16.   // handle row
  17. }
  18. finally it.close() // close the iterator to avoid resource leak
  20. // execute Table
  21. val tableResult2 = tableEnv.sqlQuery("SELECT * FROM Orders").execute()
  22. tableResult2.print()

Python

  1. env = StreamExecutionEnvironment.get_execution_environment()
  2. table_env = StreamTableEnvironment.create(env, settings)
  3. # enable checkpointing
  4. table_env.get_config().get_configuration().set_string("execution.checkpointing.mode", "EXACTLY_ONCE")
  5. table_env.get_config().get_configuration().set_string("execution.checkpointing.interval", "10s")
  7. table_env.execute_sql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")
  9. # execute SELECT statement
  10. table_result1 = table_env.execute_sql("SELECT * FROM Orders")
  11. table_result1.print()
  13. # execute Table
  14. table_result2 = table_env.sql_query("SELECT * FROM Orders").execute()
  15. table_result2.print()

Syntax

Flink parses SQL using Apache Calcite, which supports standard ANSI SQL.

The following BNF-grammar describes the superset of supported SQL features in batch and streaming queries. The Operations section shows examples for the supported features and indicates which features are only supported for batch or streaming queries.

Grammar ↕

  1. query:
  2.   values
  3.   | {
  4.       select
  5.       | selectWithoutFrom
  6.       | query UNION [ ALL ] query
  7.       | query EXCEPT query
  8.       | query INTERSECT query
  9.     }
  10.     [ ORDER BY orderItem [, orderItem ]* ]
  11.     [ LIMIT { count | ALL } ]
  12.     [ OFFSET start { ROW | ROWS } ]
  13.     [ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY]
  15. orderItem:
  16.   expression [ ASC | DESC ]
  18. select:
  19.   SELECT [ ALL | DISTINCT ]
  20.   { * | projectItem [, projectItem ]* }
  21.   FROM tableExpression
  22.   [ WHERE booleanExpression ]
  23.   [ GROUP BY { groupItem [, groupItem ]* } ]
  24.   [ HAVING booleanExpression ]
  25.   [ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ]
  27. selectWithoutFrom:
  28.   SELECT [ ALL | DISTINCT ]
  29.   { * | projectItem [, projectItem ]* }
  31. projectItem:
  32.   expression [ [ AS ] columnAlias ]
  33.   | tableAlias . *
  35. tableExpression:
  36.   tableReference [, tableReference ]*
  37.   | tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ]
  39. joinCondition:
  40.   ON booleanExpression
  41.   | USING '(' column [, column ]* ')'
  43. tableReference:
  44.   tablePrimary
  45.   [ matchRecognize ]
  46.   [ [ AS ] alias [ '(' columnAlias [, columnAlias ]* ')' ] ]
  48. tablePrimary:
  49.   [ TABLE ] tablePath [ dynamicTableOptions ] [systemTimePeriod] [[AS] correlationName]
  50.   | LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')'
  51.   | UNNEST '(' expression ')'
  53. tablePath:
  54.   [ [ catalogName . ] schemaName . ] tableName
  56. systemTimePeriod:
  57.   FOR SYSTEM_TIME AS OF dateTimeExpression
  59. dynamicTableOptions:
  60.   /*+ OPTIONS(key=val [, key=val]*) */
  62. key:
  63.   stringLiteral
  65. val:
  66.   stringLiteral
  68. values:
  69.   VALUES expression [, expression ]*
  71. groupItem:
  72.   expression
  73.   | '(' ')'
  74.   | '(' expression [, expression ]* ')'
  75.   | CUBE '(' expression [, expression ]* ')'
  76.   | ROLLUP '(' expression [, expression ]* ')'
  77.   | GROUPING SETS '(' groupItem [, groupItem ]* ')'
  79. windowRef:
  80.     windowName
  81.   | windowSpec
  83. windowSpec:
  84.     [ windowName ]
  85.     '('
  86.     [ ORDER BY orderItem [, orderItem ]* ]
  87.     [ PARTITION BY expression [, expression ]* ]
  88.     [
  89.         RANGE numericOrIntervalExpression {PRECEDING}
  90.       | ROWS numericExpression {PRECEDING}
  91.     ]
  92.     ')'
  94. matchRecognize:
  95.       MATCH_RECOGNIZE '('
  96.       [ PARTITION BY expression [, expression ]* ]
  97.       [ ORDER BY orderItem [, orderItem ]* ]
  98.       [ MEASURES measureColumn [, measureColumn ]* ]
  99.       [ ONE ROW PER MATCH ]
  100.       [ AFTER MATCH
  101.             ( SKIP TO NEXT ROW
  102.             | SKIP PAST LAST ROW
  103.             | SKIP TO FIRST variable
  104.             | SKIP TO LAST variable
  105.             | SKIP TO variable )
  106.       ]
  107.       PATTERN '(' pattern ')'
  108.       [ WITHIN intervalLiteral ]
  109.       DEFINE variable AS condition [, variable AS condition ]*
  110.       ')'
  112. measureColumn:
  113.       expression AS alias
  115. pattern:
  116.       patternTerm [ '|' patternTerm ]*
  118. patternTerm:
  119.       patternFactor [ patternFactor ]*
  121. patternFactor:
  122.       variable [ patternQuantifier ]
  124. patternQuantifier:
  125.       '*'
  126.   |   '*?'
  127.   |   '+'
  128.   |   '+?'
  129.   |   '?'
  130.   |   '??'
  131.   |   '{' { [ minRepeat ], [ maxRepeat ] } '}' ['?']
  132.   |   '{' repeat '}'

Flink SQL uses a lexical policy for identifier (table, attribute, function names) similar to Java:

  • The case of identifiers is preserved whether or not they are quoted.
  • After which, identifiers are matched case-sensitively.
  • Unlike Java, back-ticks allow identifiers to contain non-alphanumeric characters (e.g. “SELECT a AS `my field` FROM t”).

String literals must be enclosed in single quotes (e.g., SELECT 'Hello World'). Duplicate a single quote for escaping (e.g., SELECT 'It''s me.'). Unicode characters are supported in string literals. If explicit unicode code points are required, use the following syntax:

  • Use the backslash (\) as escaping character (default): SELECT U&'\263A'
  • Use a custom escaping character: SELECT U&'#263A' UESCAPE '#'

Operations