# Flink SQL Statements in Confluent Cloud for Apache Flink In Confluent Cloud for Apache Flink®, a *statement* represents a high-level resource that’s created when you enter a SQL query. Each statement has a property that holds the SQL query that you entered. Based on the SQL query, the statement can be one of these kinds: - A metadata operation, or [DDL statement](../reference/statements/overview.md#flink-sql-statements-overview). - A *background statement*, which writes data back to a table/topic while running in the background. - A *foreground statement*, which writes data back to the UI or a client. In all of these cases, the statement represents any SQL statement for [Data Definition Language (DDL)](../reference/statements/overview.md#flink-sql-statements-overview) or [queries](../reference/queries/overview.md#flink-sql-queries), which include Data Manipulation Language (DML) and Data Query Language (DQL) statements. When you submit a SQL query, Confluent Cloud creates a statement resource. You can create a statement resource from any Confluent-supported interface, including the SQL shell, Confluent CLI, Cloud Console, the [REST API](../operate-and-deploy/flink-rest-api.md#flink-rest-api), and [Terraform](../../clusters/terraform-provider.md#confluent-terraform-provider-resources-flink). The SQL query within a statement is immutable, which means that you can’t make changes to the SQL query after you submit it. If you need to edit a statement, stop the running statement and create a new statement. You can change the [security principal](../operate-and-deploy/flink-rbac.md#flink-rbac) for the statement. If a statement is running under a user account, you can change it to run under a service account by using the Confluent Cloud Console, Confluent CLI, the [REST API](../operate-and-deploy/flink-rest-api.md#flink-rest-api-update-statement), or the [Terraform provider](../../clusters/terraform-provider.md#confluent-terraform-provider). Running a statement under a service account provides better security and stability, ensuring that changes in user status or authorization don’t affect your statements. Also, you can change the compute pool that runs a statement. This can be useful if you’re close to maxing out the resources in one pool. You must stop the statement before changing the principal or compute pool, then restart the statement after the change. Confluent Cloud for Apache Flink enforces a 30-day retention for statements in terminal states. For example, after a statement transitions to the STOPPED state, it no longer consumes compute, and Confluent Cloud for Apache Flink deletes it after 30 days. If there is no consumer for the results of a foreground statement for five minutes or longer, Confluent Cloud moves the statement to the STOPPED state. ## Statements and materialized tables Confluent Cloud for Apache Flink offers two approaches for running continuous queries: - **Materialized tables**: the preferred approach for long-running streaming queries that act as incremental materialized views or production pipelines. Materialized tables are persistent objects that you can evolve in place by using [CREATE OR ALTER MATERIALIZED TABLE](../reference/statements/create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table). For more information, see [Materialized Tables](materialized-tables.md#flink-sql-materialized-tables). - **Statements**: for exploratory queries, snapshot (batch) processing, one-time data analysis, and interactive SQL in the workspace. Statements are the direct SQL execution model described on this page. If you are building a streaming pipeline that runs continuously and might need updates over time, use a materialized table. If you are running ad hoc queries, exploring data, or performing batch analysis, use statements directly. ## Limit on query text size Confluent Cloud for Apache Flink has a limit of **4 MB** on the size of query text. This limit includes string and binary literals that are part of the query. The maximum length of a statement name is 72 characters. If you combine multiple SQL statements into a single semicolon-separated string, the length limit applies to the entire string. If the query size is greater than the 4 MB limit, you receive the following error. ```none This query is too large to process (exceeds 4194304 bytes). This can happen due to: * Complex query structure. * Too many columns selected or expanded due to * usage. * Multiple table joins. * Large number of conditions. Try simplifying your query or breaking it into smaller parts. ``` ## Limits on state size The amount of state that a statement stores when it performs stateful calculations, such as [windowed aggregations](../reference/queries/window-aggregation.md#flink-sql-window-aggregation), has the potential to grow indefinitely. Confluent Cloud for Apache Flink implements a two-step approach to help you manage Flink applications that have extremely large state: - *Soft limit*: sets an upper boundary of **500 GB per statement** on the amount of state that Confluent Cloud for Apache Flink can support. - *Hard limit*: sets an absolute upper limit of **1000 GB (1 TB) per statement** on the amount of state that a Flink application can accumulate. Confluent Cloud for Apache Flink warns you proactively when your Flink applications are within 80% of reaching their soft or hard limits. If a Flink application hits the soft limit, Confluent Cloud for Apache Flink stops the statement, and you can decide whether to allow the application to keep running or resubmit the application with [State TTL](../how-to-guides/resolve-common-query-problems.md#flink-sql-warning-high-state-no-ttl) to reduce the amount of state. These limits are absolute, not dynamic. They are fixed global values for all individual statements. They don’t scale up or down based on the compute pool size (CFUs). Confluent Cloud for Apache Flink does not stop a statement running in a small pool, for example, 5 CFUs, if it exceeds the “per-CFU” theoretical capacity of 50 GB. Instead, it triggers a POOL_EXHAUSTED status, allowing the pool to autoscale or you to increase CFUs. The “Statement Stopped” mechanism triggers only when a statement reaches the absolute global limit of 500 GB, regardless of the compute pool configuration. ## Lifecycle operations for statements These are the supported lifecycle operations for a statement. Statements have a lifecycle that includes the following states: - **Pending**: The statement has been submitted and Flink is preparing to start running the statement. - **Running**: Flink is actively running the statement. - **Completed**: The statement has completed all of its work. - **Deleting**: The statement is being deleted. - **Failing**: The statement has encountered an error and is transitioning to the failed state. - **Failed**: The statement has encountered an error and is no longer running. - **Degraded**: The statement appears unhealthy, for example, no transactions have been committed for a long time, or the statement has been restarting frequently. - **Stopping**: The statement is about to be stopped. - **Stopped**: The statement has been stopped and is no longer running. ### Submit a statement - [SQL shell](../get-started/quick-start-shell.md#flink-sql-quick-start-shell) - [Cloud Console](../get-started/quick-start-cloud-console.md#flink-sql-quick-start-run-sql-statement) - [REST API statements endpoint](../operate-and-deploy/flink-rest-api.md#flink-rest-api-submit-statement) ### List running statements - [SQL shell SHOW JOBS statement](../get-started/quick-start-shell.md#flink-sql-quick-start-shell) - [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli-list-statements) - [Cloud Console](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console) - [REST API statements endpoint](../operate-and-deploy/flink-rest-api.md#flink-rest-api-list-statements) ### Describe a statement - [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli-describe-statement) - [Cloud Console](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console) - [REST API statement endpoint](../operate-and-deploy/flink-rest-api.md#flink-rest-api-get-statement) ### Delete a statement - [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli-delete-statement) - [Cloud Console](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console) - [REST API DELETE request](../operate-and-deploy/flink-rest-api.md#flink-rest-api-delete-statement) ### List statement exceptions - [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli-list-exceptions) - [Cloud Console](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console) ### Stop and resume a statement - [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli-update-statement) - [REST API UPDATE request](../operate-and-deploy/flink-rest-api.md#flink-rest-api-update-statement) - [Cloud Console](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console) To move a statement to a different compute pool, see [Move a Statement to a Different Compute Pool](../how-to-guides/move-statement-compute-pool.md#flink-sql-move-statement-compute-pool). ### Set a scaling floor Set a [Baseline CFU](autopilot.md#flink-sql-baseline-cfu), a best-effort minimum scaling target that [Autopilot](autopilot.md#flink-sql-autopilot) keeps the statement at or above. - [REST API PATCH request](../operate-and-deploy/flink-rest-api.md#flink-rest-api-update-statement) For step-by-step instructions, see [Set a Baseline CFU for a Flink SQL Statement](../how-to-guides/manage-baseline-cfu.md#flink-sql-manage-baseline-cfu). ## Queries in Flink Flink enables you to issue queries with ANSI-standard SQL on data at rest (batch) and data in motion (streams). These are the queries that are possible with Flink SQL. Metadata queries : CRUD on catalogs, databases, tables, and so on. Because Flink implements ANSI-Standard SQL, Flink uses a database analogy, and similar to a database, it uses the concepts of catalogs, databases, and tables. In Apache Kafka®, these concepts map to environments, Kafka clusters, and topics, respectively. Ad hoc / exploratory queries : You can issue queries on a topic and see the results immediately. A query can be a batch query (“show me what happened up to now”), or a transient streaming query (“show me what happened up to now and give me updates for the near future”). In this case, when the query or the session ends, no more compute is needed. Streaming queries : These queries run continuously and read data from one or more tables/topics and write results of the queries to one table/topic. In general, Flink supports both batch and stream processing, but the exact subset of allowed operations differs slightly depending on the type of query. For more information, see [Flink SQL Queries](../reference/queries/overview.md#flink-sql-queries). Flink executes all queries in streaming execution mode, whether the sources are bounded or unbounded. ## Data lifecycle Broadly speaking, the Flink SQL lifecycle is: - Flink reads data into a Flink table from Kafka through the Flink connector for Kafka. - Flink processes data by using SQL statements. - Flink processes data by using Flink task managers (managed by Confluent and not exposed to users), which are part of the Flink runtime. Flink can store some data temporarily as state while it processes the data. - Flink returns data to the user as a result-set. - The result-set can be bounded, in which case the query terminates. - The result-set can be unbounded, in which case the query runs until canceled manually. OR - Flink writes data back out to one or more tables. - Flink stores data in Kafka topics. - Flink stores the schema for the table in the Flink Metastore and synchronizes it to Schema Registry. ## Flink SQL Data Definition Language (DDL) statements Data Definition Language (DDL) statements are imperative verbs that define metadata in Flink SQL by adding, changing, or deleting tables. Data Definition Language statements modify metadata only and don’t operate on data. Use these statements with declarative [Flink SQL Queries](../reference/queries/overview.md#flink-sql-queries) to create your Flink SQL applications. Flink SQL makes it simple to develop streaming applications using standard SQL. It’s easy to learn Flink SQL if you’ve ever worked with a database or SQL-like system that’s ANSI-SQL 2011 compliant. ## Available DDL statements These are the available DDL statements in Confluent Cloud for Flink SQL. ALTER : - [ALTER MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/alter-materialized-table.md#flink-sql-alter-materialized-table) - [ALTER MODEL Statement in Confluent Cloud for Apache Flink](../reference/statements/alter-model.md#flink-sql-alter-model) - [ALTER TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/alter-table.md#flink-sql-alter-table) - [ALTER VIEW Statement in Confluent Cloud for Apache Flink](../reference/statements/alter-view.md#flink-sql-alter-view) CREATE : - [CREATE FUNCTION Statement](../reference/statements/create-function.md#flink-sql-create-function) - [CREATE MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/create-materialized-table.md#flink-sql-create-materialized-table) - [CREATE MODEL Statement in Confluent Cloud for Apache Flink](../reference/statements/create-model.md#flink-sql-create-model) - [CREATE OR ALTER MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/create-or-alter-materialized-table.md#flink-sql-create-or-alter-materialized-table) - [CREATE TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/create-table.md#flink-sql-create-table) - [CREATE VIEW Statement in Confluent Cloud for Apache Flink](../reference/statements/create-view.md#flink-sql-create-view) DESCRIBE : - [DESCRIBE Statement in Confluent Cloud for Apache Flink](../reference/statements/describe.md#flink-sql-describe) DROP : - [DROP MATERIALIZED TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/drop-materialized-table.md#flink-sql-drop-materialized-table) - [DROP MODEL Statement in Confluent Cloud for Apache Flink](../reference/statements/drop-model.md#flink-sql-drop-model) - [DROP TABLE Statement in Confluent Cloud for Apache Flink](../reference/statements/drop-table.md#flink-sql-drop-table) - [DROP VIEW Statement in Confluent Cloud for Apache Flink](../reference/statements/drop-view.md#flink-sql-drop-view) EXPLAIN : - [EXPLAIN Statement in Confluent Cloud for Apache Flink](../reference/statements/explain.md#flink-sql-explain) RESET : - [RESET Statement in Confluent Cloud for Apache Flink](../reference/statements/reset.md#flink-sql-reset-statement) SET : - [SET Statement in Confluent Cloud for Apache Flink](../reference/statements/set.md#flink-sql-set-statement) SHOW : - [SHOW Statements in Confluent Cloud for Apache Flink](../reference/statements/show.md#flink-sql-show) USE : - [USE CATALOG Statement in Confluent Cloud for Apache Flink](../reference/statements/use-catalog.md#flink-sql-use-catalog-statement) - [USE Statement in Confluent Cloud for Apache Flink](../reference/statements/use-database.md#flink-sql-use-database-statement) ## Related content - [Move a Statement to a Different Compute Pool](../how-to-guides/move-statement-compute-pool.md#flink-sql-move-statement-compute-pool) - [Flink SQL Queries](../reference/queries/overview.md#flink-sql-queries) - [Stream Processing Concepts](overview.md#flink-sql-stream-processing-concepts) - [Built-in Functions](../reference/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).