# Flink SQL Queries in Confluent Cloud for Apache Flink In Confluent Cloud for Apache Flink®, *queries* are declarative verbs that read and modify data in Apache Flink® tables. Queries that only read data, such as `SELECT`, are Data Query Language (DQL) statements. Queries that modify data, such as `INSERT INTO`, are Data Manipulation Language (DML) statements. Unlike Data Definition Language (DDL) statements, queries modify only data and don’t change metadata. When you want to change metadata, use [DDL statements](../../concepts/statements.md#flink-sql-statements). These are the available queries in Confluent Cloud for Flink SQL. | [Deduplication Queries in Confluent Cloud for Apache Flink](deduplication.md#flink-sql-deduplication) | [Group Aggregation Queries in Confluent Cloud for Apache Flink](group-aggregation.md#flink-sql-group-aggregation) | [INSERT INTO FROM SELECT Statement in Confluent Cloud for Apache Flink](insert-into-from-select.md#flink-sql-insert-into-from-select-statement) | [INSERT VALUES Statement in Confluent Cloud for Apache Flink](insert-values.md#flink-sql-insert-values-statement) | |---------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------| | [Interval joins](joins.md#flink-sql-interval-joins) | [LIMIT Clause in Confluent Cloud for Apache Flink](limit.md#flink-sql-limit) | [EXECUTE STATEMENT SET in Confluent Cloud for Apache Flink](statement-set.md#flink-sql-statement-set) | [ORDER BY Clause in Confluent Cloud for Apache Flink](orderby.md#flink-sql-order-by) | | [Pattern Recognition Queries in Confluent Cloud for Apache Flink](match_recognize.md#flink-sql-pattern-recognition) | [Regular joins](joins.md#flink-sql-regular-joins) | [SELECT Statement in Confluent Cloud for Apache Flink](select.md#flink-sql-select) | [Set Logic in Confluent Cloud for Apache Flink](set-logic.md#flink-sql-set-logic) | | [Temporal joins](joins.md#flink-sql-temporal-joins) | [Top-N Queries in Confluent Cloud for Apache Flink](topn.md#flink-sql-top-n) | [Window Aggregation Queries in Confluent Cloud for Apache Flink](window-aggregation.md#flink-sql-window-aggregation) | [Window Deduplication Queries in Confluent Cloud for Apache Flink](window-deduplication.md#flink-sql-window-deduplication) | | [Window Join Queries in Confluent Cloud for Apache Flink](window-join.md#flink-sql-window-join) | [Window Top-N Queries in Confluent Cloud for Apache Flink](window-topn.md#flink-sql-window-top-n) | [Windowing Table-Valued Functions (Windowing TVFs) in Confluent Cloud for Apache Flink](window-tvf.md#flink-sql-window-tvfs) | [WITH Clause in Confluent Cloud for Apache Flink](with.md#flink-sql-with) | ## Prerequisites You need the following prerequisites to use Flink in Confluent Cloud Console. - Access to Confluent Cloud. You need the following prerequisites to use the Flink SQL shell. - Access to Confluent Cloud. - The organization ID and environment ID for your organization. - The cloud provider and region where you run your Flink SQL statements. Default compute pools are scoped to an environment and a region, so you must specify both unless you name a compute pool. - (Optional) A compute pool ID, if you want to use a specific compute pool instead of the default pool. For more information, see [Compute Pools](../../concepts/compute-pools.md#flink-sql-compute-pools). - The FlinkDeveloper role is granted by default to all users in an environment. To create compute pools manually for workload isolation or cost control, you need the OrganizationAdmin, EnvironmentAdmin, or FlinkAdmin role. If you don’t have the appropriate role, contact your OrganizationAdmin or EnvironmentAdmin. - The Confluent CLI. To use the Flink SQL shell, update to the latest version of the Confluent CLI by running the following command: ```bash confluent update --yes ``` If you used Homebrew to install the Confluent CLI, update the CLI by using the `brew upgrade` command, instead of `confluent update`. For more information, see [Confluent CLI](https://docs.confluent.io/confluent-cli/current/overview.html). ## Use a workspace or the Flink SQL shell You can run queries and statements either in a Confluent Cloud Console workspace or in the Flink SQL shell. ### Confluent Cloud Console To run queries in the Confluent Cloud Console, follow these steps. 1. Log in to the Cloud Console. 2. In the navigation menu, click **SQL workspaces** to open the workspaces page. 3. Click **Create workspace** to open the **New workspace** page. 4. Click **Create new workspace**, and in the dialog, select the cloud provider and region. If you have Kafka topics that you want to run SQL queries on, select the region that has your Kafka cluster. 5. Click **Create workspace**. A new workspace opens with an example query in the code editor, or *cell*. ### Confluent CLI Log in to the Confluent CLI by running the following command: ```bash confluent login --save --organization ${ORG_ID} ``` To run queries in the Flink SQL shell, run the following command: ```bash confluent flink shell \ --environment \ --cloud \ --region ``` You’re ready to run your first Flink SQL query. ## Hello SQL Run the following simple query to print “Hello SQL”. ```sql SELECT 'Hello SQL'; ``` Your output should resemble: ```none EXPR$0 Hello SQL ``` Run the following query to aggregate values in a table. ```sql SELECT Name, COUNT(*) AS Num FROM (VALUES ('Neo'), ('Trinity'), ('Morpheus'), ('Trinity')) AS NameTable(Name) GROUP BY Name; ``` Your output should resemble: ```none Name Num Neo 1 Morpheus 1 Trinity 2 ``` ## Functions Flink supports many built-in functions that help you build sophisticated SQL queries. Run the `SHOW FUNCTIONS` statement to see the full list of built-in functions. ```sql SHOW FUNCTIONS; ``` Your output should resemble: ```none +------------------------+ | function name | +------------------------+ | % | | * | | + | | - | | / | | < | | <= | | <> | | = | | > | | >= | | ABS | | ACOS | | AND | | ARRAY | | ARRAY_CONTAINS | | ... ``` Run the following statement to execute the built-in `CURRENT_TIMESTAMP` function, which returns the local machine’s current system time. ```sql SELECT CURRENT_TIMESTAMP; ``` Your output should resemble: ```none CURRENT_TIMESTAMP 2024-01-17 13:07:43.537 ``` Run the following statement to compute the cosine of 0. ```sql SELECT COS(0) AS cosine; ``` Your output should resemble: ```none cosine 1.0 ``` ## Escape characters The following table shows the C-style escape sequences available in Flink SQL. #### C-style Escape Sequences | Backslash Escape Sequence | Interpretation | |-------------------------------------|--------------------------------------------------| | `\b` | backspace | | `\f` | form feed | | `\n` | newline | | `\r` | carriage return | | `\t` | tab | | `\o, \oo, \ooo` (o = 0–7) | octal byte value | | `\xh, \xhh` (h = 0–9, A–F) | hexadecimal byte value | | `\uxxxx, \Uxxxxxxxx` (x = 0–9, A–F) | 16 or 32-bit hexadecimal Unicode character value | Example : ```sql -- returns 'aaa' SELECT e'\u0061\x61\141' AS c; SELECT E'\u0061\x61\141' AS c; ``` ## Source tables As with all SQL engines, Flink SQL queries operate on rows in tables. But unlike traditional databases, Flink doesn’t manage data-at-rest in a local store. Instead, Flink SQL queries operate continuously over external tables. Flink data processing pipelines begin with source tables. Source tables produce rows operated over during the query’s execution; they are the tables referenced in the `FROM` clause of a query. Tables are created automatically in Confluent Cloud from all the Apache Kafka® topics. Also, you can create tables by using the SQL shell. The Flink SQL shell supports [SQL DDL commands](../../concepts/statements.md#flink-sql-statements) similar to traditional SQL. Standard SQL DDL is used to [create](../statements/create-table.md#flink-sql-create-table) and [alter](../statements/alter-table.md#flink-sql-alter-table) tables. The following statement creates an `employee_information` table. ```sql CREATE TABLE employee_information( emp_id INT, name VARCHAR, dept_id INT); ``` Confluent Cloud creates the corresponding `employee_information` topic automatically. ## Continuous queries You can define a continuous foreground query from the `employee_information` table that reads new rows as they are made available and immediately outputs their results. For example, you can filter for the employees who work in department `1`. ```sql SELECT * from employee_information WHERE dept_id = 1; ``` Although SQL wasn’t designed initially with streaming semantics in mind, it’s a powerful tool for building continuous data pipelines. A Flink query differs from a traditional database query by consuming rows continuously as they arrive and producing updates to the query results. A [continuous query](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-and-continuous-queries) never terminates and produces a *dynamic table* as a result. [Dynamic tables](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables) are the core concept of Flink’s SQL support for streaming data. Aggregations on continuous streams must store aggregated results continuously during the execution of the query. For example, suppose you need to count the number of employees for each department from an incoming data stream. To output timely results as new rows are processed, the query must maintain the most up-to-date count for each department. ```sql SELECT dept_id, COUNT(*) as emp_count FROM employee_information GROUP BY dept_id; ``` Such queries are considered *stateful*. Flink’s advanced fault-tolerance mechanism maintains internal state and consistency, so queries always return the correct result, even in the face of hardware failure. ## Foreground and background queries The core difference between foreground and background queries is how the query runs and where its results go: - **Foreground queries** are interactive and short-lived, with results buffered and pulled by the client. - **Background queries** are long-running infrastructure queries, with results written to Kafka topics. ### Foreground queries A foreground query is a SQL statement without an `INSERT INTO` clause. Results are not written to a Kafka topic but are buffered in Flink until the client fetches them. Foreground queries are designed for interactive and exploratory use cases. Use foreground queries to quickly iterate on SQL in the Confluent Cloud Console or tools, similar to running `SELECT` in a transactional database. #### Execution and lifecycle Foreground queries have these characteristics: - **Tied to user sessions**: When the session or token expires, the foreground query stops. - **Session-scoped and interactive**: Foreground queries are typically run temporarily during an interactive session and are not intended to be persistent infrastructure. - **Limited recovery**: Foreground queries have no retry policy. If errors occur, the statement fails. This is true even for transient errors, like when Kubernetes kills a pod because a node is under pressure. Foreground queries typically run with parallelism set to 1, have minimal recovery semantics, and are not designed to survive failures or restarts. - **No autoscaling**: Autopilot doesn’t scale or restart foreground queries, because scaling risks duplicate or incorrect results. #### Identity and authentication Foreground queries often run with the human user’s identity, so the user experience is frictionless without requiring service account setup. Audit logs show the user action. However, foreground queries can run only while the user’s session is valid. #### Use cases Use foreground queries for these scenarios: - Ad hoc `SELECT` queries on topics or tables for inspection - User-defined function (UDF) or system tests - Quick experiments where correctness matters more than autoscaling and you’re okay with no recovery ### Background queries A background query contains an `INSERT INTO` clause and writes results into a Kafka topic. Results are not buffered for direct retrieval in the Confluent Cloud Console. Background queries are designed as long-running streaming jobs (pipelines) that operate continuously, independent of any user’s session. #### Execution and lifecycle Background queries have these characteristics: - **Run indefinitely**: Background queries process data from one topic and place results into another. They behave like deployed services rather than interactive queries. - **Support recovery and retries**: Background queries are configured to restart on failures and survive upgrades. They can restart and keep running across system failures. - **Autoscaling allowed**: Autopilot can scale background jobs vertically or horizontally. Correctness is handled by normal streaming semantics, not by “no restart” guarantees. #### Identity and authentication Background queries are intended to run with a service account, because they’re long-lived infrastructure and must keep working even when the original user logs out or leaves the company. #### Use cases Use background queries for these scenarios: - Continuous pipelines that enrich, aggregate, or transform streams from input topics to output topics - Production workloads that must be durable, scalable, and recoverable ## Sink tables When running the previous query, the Flink SQL provides output in real-time but in a read-only fashion. Storing results - to power a report or dashboard - requires writing out to another table. You can achieve this by using an `INSERT INTO` statement. The table referenced in this clause is known as a *sink table*. An `INSERT INTO` statement is submitted as a detached query to Flink. ```sql INSERT INTO department_counts SELECT dept_id, COUNT(*) as emp_count FROM employee_information; ``` After it is submitted, this query runs and stores the results in the sink table directly, instead of loading the results into the system memory. ## Syntax Flink parses SQL using [Apache Calcite](https://calcite.apache.org/docs/reference.html), which supports standard ANSI SQL. The following BNF-grammar describes the superset of supported SQL features. ```bnf query: values | WITH withItem [ , withItem ]* query | { select | selectWithoutFrom | query UNION [ ALL ] query | query EXCEPT query | query INTERSECT query } [ ORDER BY orderItem [, orderItem ]* ] [ LIMIT { count | ALL } ] [ OFFSET start { ROW | ROWS } ] [ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY] withItem: name [ '(' column [, column ]* ')' ] AS '(' query ')' orderItem: expression [ ASC | DESC ] select: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } FROM tableExpression [ WHERE booleanExpression ] [ GROUP BY { groupItem [, groupItem ]* } ] [ HAVING booleanExpression ] [ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ] selectWithoutFrom: SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } projectItem: expression [ [ AS ] columnAlias ] | tableAlias . * tableExpression: tableReference [, tableReference ]* | tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ] joinCondition: ON booleanExpression | USING '(' column [, column ]* ')' tableReference: tablePrimary [ matchRecognize ] [ [ AS ] alias [ '(' columnAlias [, columnAlias ]* ')' ] ] tablePrimary: [ TABLE ] tablePath [ dynamicTableOptions ] [systemTimePeriod] [[AS] correlationName] | LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')' | [ LATERAL ] '(' query ')' | UNNEST '(' expression ')' tablePath: [ [ catalogName . ] databaseName . ] tableName systemTimePeriod: FOR SYSTEM_TIME AS OF dateTimeExpression dynamicTableOptions: /*+ OPTIONS(key=val [, key=val]*) */ key: stringLiteral val: stringLiteral values: VALUES expression [, expression ]* groupItem: expression | '(' ')' | '(' expression [, expression ]* ')' | CUBE '(' expression [, expression ]* ')' | ROLLUP '(' expression [, expression ]* ')' | GROUPING SETS '(' groupItem [, groupItem ]* ')' windowRef: windowName | windowSpec windowSpec: [ windowName ] '(' [ ORDER BY orderItem [, orderItem ]* ] [ PARTITION BY expression [, expression ]* ] [ RANGE numericOrIntervalExpression {PRECEDING} | ROWS numericExpression {PRECEDING} ] ')' matchRecognize: MATCH_RECOGNIZE '(' [ PARTITION BY expression [, expression ]* ] [ ORDER BY orderItem [, orderItem ]* ] [ MEASURES measureColumn [, measureColumn ]* ] [ ONE ROW PER MATCH ] [ AFTER MATCH ( SKIP TO NEXT ROW | SKIP PAST LAST ROW | SKIP TO FIRST variable | SKIP TO LAST variable | SKIP TO variable ) ] PATTERN '(' pattern ')' [ WITHIN intervalLiteral ] DEFINE variable AS condition [, variable AS condition ]* ')' measureColumn: expression AS alias pattern: patternTerm [ '|' patternTerm ]* patternTerm: patternFactor [ patternFactor ]* patternFactor: variable [ patternQuantifier ] patternQuantifier: '*' | '*?' | '+' | '+?' | '?' | '??' | '{' { [ minRepeat ], [ maxRepeat ] } '}' ['?'] | '{' repeat '}' statementSet: EXECUTE STATEMENT SET BEGIN { insertStatement ';' }+ END ';' ``` Flink uses a lexical policy for identifiers (table, attribute, and function names) that’s similar to Java. - Flink preserves the case of identifiers regardless of whether they are quoted. - After parsing, Flink matches identifiers case-sensitively. - Unlike Java, back-ticks enable identifiers to contain non-alphanumeric characters, for example: ```sql SELECT a AS `my field` FROM t; ``` String literals must be enclosed in single quotes, for example, `SELECT 'Hello World'`. Duplicate a single quote for escaping, for example, `SELECT 'It''s me'`. ```sql SELECT 'Hello World', 'It''s me'; ``` Your output should resemble: ```none EXPR$0 EXPR$1 Hello World 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 the escaping character (default), for example, `SELECT U&'\263A'`: ```sql SELECT U&'\263A'; ``` Your output should resemble: ```none EXPR$0 ☺ ``` Also, you can use a custom escaping character with UESCAPE, for example, `SELECT U&'#2713' UESCAPE '#'`: ```sql SELECT U&'#2713' UESCAPE '#'; ``` Your output should resemble: ```none EXPR$0 ✓ ``` ## Related content - [Table types by query](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-table-types-by-query) - [DDL Statements](../../concepts/statements.md#flink-sql-statements) - [Stream Processing Concepts](../../concepts/overview.md#flink-sql-stream-processing-concepts) - [Built-in Functions](../functions/overview.md#flink-sql-functions-overview) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).