# Tables and Topics in Confluent Cloud for Apache Flink A dynamic table is an abstraction for working with both batch and streaming data in a unified manner. Apache Flink® and the Table API use dynamic tables to define, modify, and query structured data as it changes over time. In contrast to the static tables that represent batch data, dynamic tables change over time. But like static batch tables, systems can execute queries over dynamic tables. Confluent Cloud for Apache Flink implements American National Standards Institute (ANSI) standard SQL and has the familiar concepts of catalogs, databases, and tables. Confluent Cloud maps a Flink catalog to an environment and *vice versa*. Similarly, Flink databases and tables are mapped to Apache Kafka® clusters and topics. For more information, see [Metadata mapping between Kafka cluster, topics, schemas, and Flink](../overview.md#ccloud-flink-overview-metadata-mapping). ## Dynamic tables and continuous queries Every table in Flink is equivalent to a stream of events that describe the changes to that table. A stream of changes like this is a *changelog stream*. Essentially, a stream is the changelog of a table, and a stream backs every table. This is also the case for regular database tables. Querying a dynamic table yields a *continuous query*. A continuous query never ends and produces dynamic results, which are themselves a dynamic table. The query continuously updates its dynamic result table to reflect changes on its dynamic input tables. Essentially, a continuous query on a dynamic table is similar to a query that defines a materialized view. The output of a continuous query is always equivalent to the result of the same query executed in batch mode on a snapshot of the input tables. ### Append-only table ![Animation of new INSERT rows being appended to the Orders table and the matching INSERT events appended to its changelog stream](flink/images/flink-dynamic-tables-append-only.gif) In an append-only table, the only changes are new rows appended to the end of the table, as shown in this animation for the `Orders` table. The corresponding changelog stream is a stream of INSERT events. Adding another order to the table is the same as adding another INSERT statement to the stream, as shown by the added INSERT event in the changelog stream. This is an example of an append-only or insert-only table. ### Updating table Not all tables are append-only tables. Tables can also contain events that modify or delete existing rows. The changelog stream used by Flink SQL contains three additional event types to accommodate different ways that tables can be updated. Besides the regular *Insertion* event, *Update Before* and *Update After* are a pair of events that work together to update an earlier result. The *Delete* event has the effect you would expect, removing a record from the table. ![Animation of an UPDATE_BEFORE and UPDATE_AFTER event pair updating the Bestsellers table after an order is canceled](flink/images/flink-dynamic-tables-updating.gif) This animation has the same starting point as the previous example that showed the append-only table. But this time, the user canceled an order, and the item in that order didn’t sell. As a result, the `Bestsellers` table is *updated*, rather than receiving another insert. The update starts with another order appended to the append-only or insert-only `Orders` table, which the changelog stream registers as an INSERT event. Because the SQL statement groups, the result is an updating table instead of an append-only or insert-only table. In this example, the user cancels an order for 15 hats. To process the event with the 15-hat order cancellation, the query produces two update events: - The first is an UPDATE_BEFORE event that retracts the current result that showed 50 hats as the bestselling item. - The second is an UPDATE_AFTER event that replaces the old entry with a new one that shows 35 hats. Conceptually, Confluent Cloud for Apache Flink processes the UPDATE_BEFORE event first, which removes the old entry from the `Bestsellers` table. Then, the sink processes the UPDATE_AFTER event, which inserts the updated results. The following figure visualizes the relationship of streams, dynamic tables, and continuous queries: ![Relationship between streams, dynamic tables, and continuous queries](flink/images/flink-stream-query-stream.png) 1. Confluent Cloud for Apache Flink converts a stream into a dynamic table. 2. Confluent Cloud for Apache Flink evaluates a continuous query on the dynamic table, yielding a new dynamic table. 3. Confluent Cloud for Apache Flink converts the resulting dynamic table back into a stream. Dynamic tables are a logical concept. The Flink SQL runtime materializes only the state strictly necessary to produce correct results for the specific query that runs. For example, the previous diagram shows a query that executes a simple filter. This requires no state, so the runtime materializes nothing. ### Changelog entries Flink has four types of changelog entries: | Short name | Long name | Semantics | |--------------|---------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | +I | Insertion | Records only the insertions that occur. | | -U | Update Before | Retracts a previously emitted result.
Update Before is an update operation with the previous content of the
updated row. This kind occurs together with Update After (+U) for modeling
an update that must retract the previous row first. It is useful in
cases of a non-idempotent update, which is an update of a row that
is not uniquely identifiable by a key. | | +U | Update After | Updates a previously emitted result.
Update After is an update operation with new content for the updated row.
This kind can occur together with Update Before (-U) for modeling an
update that must retract the previous row first, or it can describe an
idempotent update, which is an update of a row that is uniquely identifiable
by a key. | | -D | Delete | Deletes the last result. | The `-` character always indicates row removal. If the downstream system upserts, you should use a primary key in Confluent Cloud for Apache Flink to avoid the need to use Update Before. Depending on the combination of source, sink, and business logic applied, you can end up with the following types of changelog streams. | Changelog stream types | Stream category | Changelog entry types | |--------------------------|-------------------|-------------------------------------------------------------| | Appending stream | Append stream | Contains only +I | | Upserting streams | Update stream | +I, +U, -D (never contains -U but can contain +U and/or -D) | | Retracting stream | Update stream | +I, +U, -U, -D (contains +I and can contain -U and/or -D) | - All streams can have +I / inserts. - Both retract and upsert streams can have -D / deletes and +U / upserts (upsert afters). - Only retract streams can have -U. ### The internal changelog versus a changelog you own Flink’s *internal* changelog is its own representation of how a table changes over time, expressed as the `+I`, `-U`, `+U`, and `-D` row kinds. Flink produces and consumes it automatically. When Flink writes a table to a Kafka topic, the table’s [changelog mode](../reference/serialization.md#flink-sql-serialization-changelog-formats) (append, upsert, or retract) decides how each row kind is encoded on that topic so that Flink can read it back later. This is different from a changelog that another system owns, which this documentation calls a *custom changelog*. In a custom changelog, the operation is stated explicitly in a field that you define, for example a change data capture (CDC) record with an `op` field set to `c`, `u`, or `d`. Flink doesn’t interpret that field natively, and a non-Flink consumer doesn’t interpret Flink’s internal row kinds either. To read a custom changelog into Flink, or to produce one for a downstream consumer, use the [FROM_CHANGELOG](../reference/functions/changelog-conversion.md#flink-ptfs-from-changelog) and [TO_CHANGELOG](../reference/functions/changelog-conversion.md#flink-ptfs-to-changelog) built-in functions. For a step-by-step guide, see [Read and write custom changelog formats](../how-to-guides/read-write-custom-changelog.md#flink-read-write-changelog). Some CDC formats are an exception. Flink natively understands a set of standard changelog formats, including Debezium, so even though a Debezium record carries an explicit operation, Flink decodes it for you and you don’t need `FROM_CHANGELOG`. A changelog is *custom* in the sense used here only when Flink has no built-in format for it. For the formats Flink recognizes, see [changelog formats](../reference/serialization.md#flink-sql-serialization-changelog-formats). ## Table types by query Whether a query accepts an [append-only table](#flink-sql-dynamic-tables-append-only-table) or an [updating table](#flink-sql-dynamic-tables-updating-table) as input, and which of the two it produces as output depends on the query type. The following table summarizes these requirements for each query type. For the full explanation and any special cases, see the “Table types” section on each query’s own page. | Query type | Required input | Produced output | Notes | |----------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------|--------------------------------|------------------------------------------------------------------------------------------------------------------------| | [Regular joins](../reference/queries/joins.md#flink-sql-regular-joins) | Append-only or updating | Updating | | | [Interval joins](../reference/queries/joins.md#flink-sql-interval-joins) | Append-only | Append-only | | | [Temporal joins](../reference/queries/joins.md#flink-sql-temporal-joins) | Left: append-only or updating.
Right: updating, with a primary key | Matches the left input | | | [Lookup joins](../reference/queries/joins.md#flink-sql-lookup-joins) | Append-only or updating | Matches the input | The lookup side is an external table, not a Flink SQL changelog | | [OVER aggregation](../reference/queries/over-aggregation.md#flink-sql-over-aggregation) | Append-only | Append-only | | | [Group aggregation](../reference/queries/group-aggregation.md#flink-sql-group-aggregation) | Append-only or updating | Updating | | | [Deduplication](../reference/queries/deduplication.md#flink-sql-deduplication) | Append-only or updating | Usually updating | Can produce append-only output in a narrow special case; see the page | | [Top-N](../reference/queries/topn.md#flink-sql-top-n) | Append-only or updating | Updating | | | `SELECT DISTINCT` | Append-only or updating | Updating | See [SELECT Statement in Confluent Cloud for Apache Flink](../reference/queries/select.md#flink-sql-select) | | [UNION / UNION ALL](../reference/queries/set-logic.md#flink-sql-set-logic-union) | Append-only or updating | Matches the inputs | | | [EXCEPT, INTERSECT, IN, EXISTS](../reference/queries/set-logic.md#flink-sql-set-logic) | Append-only or updating | Updating | The optimizer rewrites these into a join and a group aggregation | | [LIMIT](../reference/queries/limit.md#flink-sql-limit) | Append-only or updating | Matches the input, or updating | Stays append-only only if the input is already append-only | | [ORDER BY](../reference/queries/orderby.md#flink-sql-order-by) | Append-only | Append-only | | | [Windowing TVFs](../reference/queries/window-tvf.md#flink-sql-window-tvfs) | Append-only or updating | Matches the input | Assigning a window doesn’t itself require append-only input; the
operator built on top of the windowed table might | | [Window aggregation](../reference/queries/window-aggregation.md#flink-sql-window-aggregation) | Append-only or updating | Append-only | | | [Window Top-N](../reference/queries/window-topn.md#flink-sql-window-top-n) | Append-only | Append-only | | | [Window deduplication](../reference/queries/window-deduplication.md#flink-sql-window-deduplication) | Append-only | Append-only | | | [Window join](../reference/queries/window-join.md#flink-sql-window-join) | Append-only | Append-only | | | [Pattern recognition (MATCH_RECOGNIZE)](../reference/queries/match_recognize.md#flink-sql-pattern-recognition) | Append-only | Append-only | | ## Related content - [Read and write custom changelog formats](../how-to-guides/read-write-custom-changelog.md#flink-read-write-changelog) - [Changelog Conversion](../reference/functions/changelog-conversion.md#flink-sql-changelog-conversion-functions) - [Flink SQL Queries](../reference/queries/overview.md#flink-sql-queries) - [Process Table Functions](process-table-functions.md#flink-ptfs) - [Inferred tables](../reference/statements/create-table.md#flink-sql-create-table-inferred-tables), for concrete examples of the tables that Flink derives from existing Kafka topics and Schema Registry schemas #### 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).