Definition
- Flink SQL – Flink's relational API that lets you express stream and batch processing using standard ANSI SQL.
- Dynamic table – A table that changes over time as records arrive, equivalent to a changelog stream.
- Stream-table duality – A stream can be viewed as a table and a table can be converted back into a stream.
- Continuous query – A query over a dynamic table that never finishes and keeps updating its result as input arrives.
- Calcite optimizer – The cost-based optimizer that parses, optimizes, and compiles SQL into the underlying Flink dataflow operators.
Imagine the National Bank of India wants to ask questions about its live transaction feed without writing low-level stream code. Flink SQL lets an analyst write a familiar SQL query like counting transactions per account, and Flink treats the never-ending transaction feed as a table that keeps growing. As each new transaction arrives, Flink automatically updates the answer, the same way a spreadsheet formula recalculates when you add a row.
Understanding the Concepts
What is a dynamic table?
A normal SQL table is static, but a dynamic table represents data that changes over time. It is equivalent to a changelog stream of inserts, updates, and deletes. This stream-table duality means you can convert a stream into a dynamic table and a dynamic table back into a stream.
-- Define a dynamic table backed by Kafka
CREATE TABLE transactions (
txn_id STRING,
account_id STRING,
amount DECIMAL(15, 2),
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-transactions',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'json',
'scan.startup.mode' = 'latest-offset'
);
How does the Calcite optimizer compile SQL?
When you submit a query, the planner parses it into a logical plan, applies relational optimizations through Apache Calcite such as predicate pushdown, projection pushdown, and join reordering, and translates the optimized plan into physical Flink operators.
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
public class FlinkSqlCompilationDemo {
public static void main(String[] args) {
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.inStreamingMode()
.build();
TableEnvironment tEnv = TableEnvironment.create(settings);
// Calcite optimizes this query:
// 1. Parses SQL into logical plan
// 2. Pushes the amount > 100000 filter to the source
// 3. Compiles into streaming operators
TableResult result = tEnv.executeSql(
"SELECT account_id, SUM(amount) AS total " +
"FROM transactions " +
"WHERE amount > 100000 " +
"GROUP BY account_id"
);
}
}
Logical plan: Scan → Filter(amount > 100000) → GroupBy(account_id) → Project Optimized: Scan(pushdown: amount > 100000) → GroupBy(account_id) → Project Physical: KafkaSource → KeyedAggregate(keyed by account_id) → Sink
What is a continuous query?
Unlike a one-shot batch query, a continuous query never finishes in streaming mode. It keeps running and keeps updating its result dynamic table as input rows arrive. A GROUP BY becomes a keyed stateful operator that maintains the running aggregate per key.
-- This continuous query runs forever
-- Each new transaction updates the affected account's total
SELECT account_id, SUM(amount) AS total_deposits
FROM transactions
WHERE txn_type = 'DEPOSIT'
GROUP BY account_id;
-- As transactions arrive:
-- txn-001 (ACCT-100, 50000) → emits (ACCT-100, 50000)
-- txn-002 (ACCT-100, 30000) → retracts (ACCT-100, 50000)
-- emits (ACCT-100, 80000)
-- txn-003 (ACCT-200, 10000) → emits (ACCT-200, 10000)
What are the result changelog modes?
Because the result is continuously updated, the query produces a changelog. An append-only query emits only inserts. An aggregation emits retractions of old values followed by new values, or upserts keyed by the grouping key.
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class ChangelogModeDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// Append-only: simple filter (no updates)
tEnv.executeSql(
"INSERT INTO high_value_alerts " +
"SELECT * FROM transactions WHERE amount > 1000000"
);
// Retract/Upsert: aggregation (updates existing results)
tEnv.executeSql(
"INSERT INTO account_totals " +
"SELECT account_id, SUM(amount) FROM transactions GROUP BY account_id"
);
// account_totals sink must be upsert-capable
}
}
How does state management work in Flink SQL?
Stateful operators in SQL accumulate state that grows unless bounded. For unbounded key spaces, you set a state time-to-live with idle-state retention to prevent unbounded growth.
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.configuration.Configuration;
public class StateRetentionDemo {
public static void main(String[] args) {
Configuration config = new Configuration();
// Idle state retention: clean up state for keys not seen in 24 hours
config.setString("table.exec.state.ttl", "86400000");
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.newInstance()
.inStreamingMode()
.withConfiguration(config)
.build());
// Without TTL, dormant accounts accumulate state forever
// With 24h TTL, accounts with no transactions in 24h have state cleaned up
tEnv.executeSql(
"SELECT account_id, COUNT(*) as txn_count " +
"FROM transactions GROUP BY account_id"
);
}
}
How does batch vs streaming mode work?
The same SQL runs in both batch and streaming execution modes. In batch mode, the query runs once over bounded data and terminates. In streaming mode, the query is continuous.
package com.bank.flink;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class UnifiedModelDemo {
public static void main(String[] args) {
// Batch mode: query runs once, terminates
TableEnvironment batchEnv = TableEnvironment.create(
EnvironmentSettings.inBatchMode());
// Streaming mode: query runs continuously
TableEnvironment streamEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
// Same SQL works in both modes
String sql = "SELECT branch_id, SUM(amount) FROM transactions GROUP BY branch_id";
// In batch: produces final result and exits
// In streaming: produces continuous changelog updates
}
}
What are connectors and catalogs?
Flink SQL integrates with connectors and catalogs so a table can be backed by Kafka, JDBC, filesystem, or formats like Avro and JSON. You define watermarks in the table DDL for event-time processing.
-- Kafka source with JSON format and watermark
CREATE TABLE loan_payments (
payment_id STRING,
loan_id STRING,
amount DECIMAL(15, 2),
paid_at TIMESTAMP(3),
WATERMARK FOR paid_at AS paid_at - INTERVAL '10' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'nbi-loan-payments',
'properties.bootstrap.servers' = 'kafka.bank.internal:9092',
'format' = 'json'
);
-- JDBC sink for the reporting database
CREATE TABLE loan_summary (
loan_id STRING,
total_paid DECIMAL(15, 2),
payment_count BIGINT,
PRIMARY KEY (loan_id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:postgresql://db.bank.internal:5432/reports',
'table-name' = 'loan_summary'
);
INSERT INTO loan_summary
SELECT loan_id, SUM(amount), COUNT(*)
FROM loan_payments
GROUP BY loan_id;
What should a senior engineer know?
A senior engineer understands that Flink SQL is not a static query engine but a compiler from SQL to a continuously-running stateful streaming graph.
- The result of a continuous query must be written to a sink whose changelog mode matches. An append-only query can go to any sink, but an updating query needs an upsert-capable or retract-capable sink.
- State TTL is critical for production. Without it, per-account state grows forever as dormant accounts accumulate.
- Watermarks declared in the DDL are essential for event-time windows and temporal joins.
- Flink SQL's Calcite optimizer applies predicate pushdown and projection pushdown, reducing the data read from connectors.
- Monitor state size per operator in the Flink dashboard to catch unbounded growth early.
// Monitor state size:
// Flink Web UI → Job → Task → State Size
// Alert if state grows beyond expected bounds
// Set table.exec.state.ttl to bound idle state
Visual Overview
Flink SQL Compilation Pipeline
SQL Query String
│
▼
┌──────────────┐
│ SQL Parser │
└──────┬───────┘
▼
┌──────────────────┐
│ Logical Plan │
│ (relational tree)│
└──────┬───────────┘
▼
┌──────────────────────┐
│ Calcite Optimizer │
│ - predicate pushdown │
│ - projection pruning │
│ - join reordering │
└──────┬───────────────┘
▼
┌──────────────────┐
│ Physical Plan │
│ (Flink operators)│
└──────┬───────────┘
▼
┌──────────────────────────────┐
│ Streaming Dataflow Graph │
│ KafkaSource → KeyedAgg → Sink│
│ (runs continuously) │
└──────────────────────────────┘
Dynamic Table Changelog
Input stream: txn-001(ACCT-100, +50000)
txn-002(ACCT-100, +30000)
txn-003(ACCT-200, +10000)
Continuous query: SELECT account_id, SUM(amount) GROUP BY account_id
Output changelog:
+I (ACCT-100, 50000) ← insert
-U (ACCT-100, 50000) ← retract old
+U (ACCT-100, 80000) ← update new
+I (ACCT-200, 10000) ← insert
Key Takeaway
Flink SQL is Flink's relational API that lets you express stream and batch processing in standard ANSI SQL. The concept that makes it work is the dynamic table, a table that changes over time as records arrive, equivalent to a changelog stream. When you submit a query, the Calcite optimizer parses it, applies optimizations like predicate pushdown, and compiles it into a streaming dataflow graph of operators. In streaming mode the query is continuous: it never finishes and keeps updating its result as new input arrives.
The crucial consequence is that the result is a changelog, not just a final answer. An append-only query emits only inserts. An aggregation emits retractions of old values and new values, or upserts keyed by the grouping key. This means an updating query needs an upsert-capable sink, not a plain append-only sink. Stateful operators accumulate state that must be bounded with a time-to-live for unbounded key spaces.
At the National Bank of India our analysts wrote a plain GROUP BY query to compute live per-account totals from the streaming transaction feed. Flink compiled it into a keyed stateful operator that maintained per-account aggregates and emitted retract changelog updates as transactions arrived. We set idle-state retention to 24 hours so dormant accounts did not grow state forever, and we wrote to an upsert JDBC sink so the dashboard always showed current totals.
Remember one sentence: Flink SQL compiles standard SQL into a continuously-running stateful streaming graph over dynamic tables, where results update as events arrive and must be written to a changelog-compatible sink.
---