Definition
- Dynamic table – A logical table whose contents evolve over time as new rows are inserted and existing rows are updated or deleted.
- Static batch table – A fixed, bounded snapshot of data at a single point in time where queries run once and terminate.
- Stream-table duality – A stream can be interpreted as a continuously-changing table, and a dynamic table can be converted back into a stream.
- Append mode – Each incoming record is a new row appended to the table, fitting event logs.
- Upsert mode – The stream carries a key, and each record inserts or updates the row for that key, fitting current-state views.
A static batch table is like a printed end-of-day account statement at the National Bank of India: it is a fixed snapshot that never changes. A dynamic table is like the live account balance screen in the banking app: it keeps changing as new transactions post, reversals happen, and corrections are applied. If you ask "what is Rajesh's balance" of the printed statement, you get one fixed answer forever, but if you ask the live screen, the answer keeps updating.
Understanding the Concepts
What is stream-table duality?
Stream-table duality is the key idea behind dynamic tables. A stream of records can be converted into a dynamic table, and a dynamic table can be converted back into a stream. This lets you write SQL over streams as if they were tables.
-- A Kafka topic is a stream.
-- CREATE TABLE makes it a dynamic table.
CREATE TABLE account_events (
account_id STRING,
event_type STRING,
balance DECIMAL(15, 2),
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-account-events',
'format' = 'json'
);
-- This dynamic table grows as events arrive on the Kafka topic.
-- A query over it produces another dynamic table.
What is append mode?
In append mode, each incoming record is a new row appended to the table. Rows are never updated or removed. This fits event logs like a raw transaction feed where every event is immutable.
-- Raw transaction feed: append-only
CREATE TABLE raw_transactions (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
txn_time TIMESTAMP(3),
WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-raw-transactions',
'format' = 'json'
);
-- Every transaction is a new row. No row is ever changed.
-- Time 10:00 → (TXN-001, ACCT-100, 50000, 10:00)
-- Time 10:01 → (TXN-002, ACCT-100, 30000, 10:01) ← new row, not an update
What is upsert mode?
In upsert or changelog mode, the stream carries a key. Each record either inserts a new row for a new key or updates the existing row for a known key. Deletes remove a key. This fits views that represent the current state of entities, like account balances.
-- Account balance view: upsert mode keyed by account_id
CREATE TABLE account_balances (
account_id STRING,
balance DECIMAL(15, 2),
updated_at TIMESTAMP(3),
PRIMARY KEY (account_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'nbi-account-balances',
'key.format' = 'json',
'value.format' = 'json'
);
-- Each message replaces the previous balance for that account.
-- ACCT-100: 50000 → 80000 → 75000 (only the latest is the "current" row)
How does a dynamic table convert back to a stream?
Converting a dynamic table back into a stream also has modes. An append-only table becomes a simple insert stream. A table that experiences updates becomes either a retract stream or an upsert stream.
Append-only table → Insert stream: +I (TXN-001, ACCT-100, 50000) +I (TXN-002, ACCT-100, 30000) Updating table → Retract stream: +I (ACCT-100, 50000) ← insert -U (ACCT-100, 50000) ← retract old value +U (ACCT-100, 80000) ← insert new value Updating table → Upsert stream (keyed by account_id): +U (ACCT-100, 50000) ← upsert +U (ACCT-100, 80000) ← upsert replaces previous
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class DynamicTableToStreamDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// This aggregation produces a retract/upsert dynamic table
// because the SUM changes as new transactions arrive
tEnv.executeSql(
"INSERT INTO account_totals " +
"SELECT account_id, SUM(amount) FROM raw_transactions GROUP BY account_id"
);
// The sink must handle retractions or upserts
}
}
How does a dynamic table differ from a static batch table?
A static table is bounded and complete. A query over it runs once, sees all the data, and terminates with a final result. A dynamic table in streaming mode is conceptually unbounded: input keeps arriving, so a continuous query never terminates and its result keeps updating.
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class BatchVsStreamingDemo {
public static void main(String[] args) {
// Batch: query runs once, terminates
TableEnvironment batchEnv = TableEnvironment.create(
EnvironmentSettings.inBatchMode());
batchEnv.executeSql(
"SELECT branch_id, SUM(amount) FROM eod_transactions GROUP BY branch_id"
);
// Produces one final result and exits
// Streaming: query runs forever, updates continuously
TableEnvironment streamEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
streamEnv.executeSql(
"SELECT branch_id, SUM(amount) FROM live_transactions GROUP BY branch_id"
);
// Never terminates; result updates with every transaction
}
}
What about event time and watermarks?
Event time and watermarks apply to dynamic tables so that time-based operations like windows know when input is complete. Without a watermark on the source, event-time windows can never close.
-- Watermark enables event-time windowing
CREATE TABLE transactions (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
txn_time TIMESTAMP(3),
WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'nbi-txns', 'format' = 'json');
-- Tumbling window uses the watermark to know when each hour is complete
SELECT
account_id,
TUMBLE_START(txn_time, INTERVAL '1' HOUR) AS window_start,
SUM(amount) AS hourly_total
FROM transactions
GROUP BY account_id, TUMBLE(txn_time, INTERVAL '1' HOUR);
What about state growth?
Because a continuous query maintains state for every key, you must think about state growth. For unbounded key spaces, set a state time-to-live so that keys not seen within the retention period have their state cleaned up.
// Set idle-state retention to 24 hours
// Keys not seen in 24h will have state cleaned up
// This prevents dormant accounts from accumulating state forever
Configuration config = new Configuration();
config.setString("table.exec.state.ttl", "86400000");
What should a senior engineer know?
The conceptual leap is that a dynamic table is not a materialized thing sitting in memory but a logical, evolving relation defined by its changelog.
- Understand which mode your source produces: append or upsert.
- Match the output changelog mode to the sink's capabilities.
- The unified batch-stream model means the same SQL works on bounded and unbounded data.
- State growth is the most common production issue with continuous queries over dynamic tables.
- Watermarks must be declared on the source table where event time enters the pipeline.
// Senior tip: use EXPLAIN to see the physical plan
// and verify changelog mode
TableResult explain = tEnv.executeSql(
"EXPLAIN SELECT account_id, SUM(amount) FROM transactions GROUP BY account_id"
);
// Shows: changelog mode, optimizer choices, state operators
Visual Overview
Stream-Table Duality
Stream of events Dynamic Table
┌──────────────┐ ┌──────────────────────┐
│ +I ACCT-100 │ ──────▶ │ ACCT-100 | 50,000 │
│ 50,000 │ │ │
│ +I ACCT-200 │ ──────▶ │ ACCT-100 | 50,000 │
│ 10,000 │ │ ACCT-200 | 10,000 │
│ +U ACCT-100 │ ──────▶ │ ACCT-100 | 80,000 │
│ 80,000 │ │ ACCT-200 | 10,000 │
└──────────────┘ └──────────────────────┘
Stream ◄──────────────▶ Table
Batch vs Streaming Query
Batch (bounded): Streaming (unbounded):
┌─────────────┐ ┌─────────────────────┐
│ All data │ │ Data keeps arriving │
│ available │ │ ──────────────────▶ │
└──────┬──────┘ └──────┬──────────────┘
▼ ▼
┌─────────────┐ ┌─────────────────────┐
│ Query runs │ │ Continuous query │
│ ONCE │ │ runs FOREVER │
└──────┬──────┘ └──────┬──────────────┘
▼ ▼
┌─────────────┐ ┌─────────────────────┐
│ Final result │ │ Updating changelog │
│ (terminates) │ │ (never terminates) │
└─────────────┘ └─────────────────────┘
Key Takeaway
A dynamic table in Flink SQL is a logical table whose contents evolve over time as rows are inserted, updated, and deleted. The key idea is stream-table duality: a stream can be viewed as a continuously-changing table, and a dynamic table can be converted back into a stream. In append mode every record is a new row, fitting event logs. In upsert mode the stream carries a key and each record replaces the current value for that key, fitting views like account balances.
The difference from a static batch table is fundamental. A static table is bounded and complete, so a query runs once and terminates. A dynamic table in streaming mode is unbounded, so a continuous query never terminates and keeps emitting a changelog that updates its result as new input arrives. The same SQL works in both modes, which is the unified model.
At the National Bank of India, our raw transaction feed was an append-mode dynamic table that only grew, used for audit. Our live account-balance view was an upsert-mode dynamic table keyed by account, where each transaction updated the single current row. When finance needed a frozen end-of-day figure, that was a batch query over a bounded snapshot. We set state retention to 24 hours so dormant accounts did not accumulate state forever.
Remember one sentence: a dynamic table evolves over time through stream-table duality, driving continuous queries that never terminate, unlike a static batch table that produces one final result.
---