flink-sql

When to Use SQL vs DataStream

Rule: Start with SQL. Drop to DataStream only when SQL can't express it.

All SQL keywords, functions, and syntax in lowercase.

Changelog Modes (Pick the Sink to Match the Query)

"Table sink doesn't support consuming update changes" is the most common Flink SQL planner error. The rule:

• Append-only queries — window TVF aggregations on event time, interval joins, filters/projections → any sink (kafka, filesystem).
• Updating queries — group aggregations, deduplication, regular joins, CDC sources → upsert sink with a primary key (upsert-kafka, jdbc, iceberg upsert).

Append-only output is a key reason to prefer window TVF aggregations over plain group by.

Connector DDL — Kafka Source

-- kafka source with schema registry (avro)
create table kafka_events (
  event_id string,
  user_id string,
  event_type string,
  payload string,
  event_time timestamp(3),
  proc_time as proctime(),
  watermark for event_time as event_time - interval '5' second
) with (
  'connector' = 'kafka',
  'topic' = 'events.v1',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'flink-sql-consumer',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'avro-confluent',
  'avro-confluent.url' = 'http://schema-registry:8081'
);

Connector DDL — Kafka Sink (Exactly-Once)

-- upsert-kafka sink with exactly-once semantics
create table kafka_sink (
  agg_key string,
  window_start timestamp(3),
  window_end timestamp(3),
  event_count bigint,
  primary key (agg_key, window_start) not enforced
) with (
  'connector' = 'upsert-kafka',
  'topic' = 'aggregates.events.v1',
  'properties.bootstrap.servers' = 'kafka:9092',
  'sink.delivery-guarantee' = 'exactly-once',      -- default is at-least-once
  'sink.transactional-id-prefix' = 'agg-events-',  -- mandatory for exactly-once
  'properties.transaction.timeout.ms' = '900000',  -- must be <= broker transaction.max.timeout.ms
  'key.format' = 'json',
  'value.format' = 'json'
);

Other connectors — JDBC, Iceberg, filesystem, mysql-cdc (direct-from-database CDC): references/connectors.md.

Watermarks

-- bounded out-of-orderness (most common)
create table events (
  event_id string,
  event_time timestamp(3),
  watermark for event_time as event_time - interval '5' second
) with (...);

-- strictly ascending (perfect ordering)
create table events (
  event_id string,
  event_time timestamp(3),
  watermark for event_time as event_time
) with (...);

-- custom expression (deterministic null fallback; never current_timestamp)
create table events (
  event_id string,
  event_time timestamp(3),
  watermark for event_time as
    coalesce(event_time, timestamp '1970-01-01 00:00:00') - interval '10' second
) with (...);

Rule: Always define watermarks on event time columns. No watermark = no event-time windows.

Insert Into (Core Operation)

insert into starts the job — everything before it only declares tables.

insert into kafka_sink
select user_id, window_start, window_end, count(*)
from table(tumble(table kafka_events, descriptor(event_time), interval '1' minute))
group by user_id, window_start, window_end;

Statement Sets (Multi-Sink)

Write multiple sinks from the same source in a single job. Avoids redundant reads.

begin statement set;

-- sink 1: real-time aggregates to kafka
insert into kafka_sink
select user_id, window_start, window_end, count(*)
from table(tumble(table kafka_events, descriptor(event_time), interval '1' minute))
group by user_id, window_start, window_end;

-- sink 2: raw events to iceberg for cold storage
insert into iceberg_catalog.bronze.raw_events
select
  event_id,
  user_id,
  event_type,
  event_time,
  current_timestamp as ingested_at,
  date_format(event_time, 'yyyy-MM-dd') as dt
from kafka_events;

end;

Rule: Use statement sets when a single source feeds multiple sinks. One job, one source read.

Windowed Aggregations

Tumbling Window (Fixed, Non-Overlapping)

select
  user_id,
  window_start,
  window_end,
  count(*) as event_count,
  sum(amount) as total_amount
from table(tumble(table events, descriptor(event_time), interval '1' minute))
group by user_id, window_start, window_end;

Hop Window (Sliding, Overlapping)

-- hop(data, descriptor, slide, size)
select
  user_id,
  window_start,
  window_end,
  count(*) as event_count
from table(hop(table events, descriptor(event_time), interval '30' second, interval '1' minute))
group by user_id, window_start, window_end;

Cumulate Window (Expanding Within Period)

-- cumulate(data, descriptor, step, max size)
select
  user_id,
  window_start,
  window_end,
  sum(amount) as cumulative_amount
from table(cumulate(table events, descriptor(event_time), interval '1' minute, interval '1' hour))
group by user_id, window_start, window_end;

Session Window (Gap-Based)

-- session emits the same window_start/window_end columns as the other window tvfs
select
  user_id,
  window_start,
  window_end,
  count(*) as events_in_session
from table(session(table events partition by user_id, descriptor(event_time), interval '30' minute))
group by user_id, window_start, window_end;

Top-N Queries

-- real-time top 10 users by event count (continuous, emits retractions)
-- state: O(distinct user_id) until ttl (aggregate + rank buffer)
select
  user_id,
  event_count,
  row_num
from (
  select
    user_id,
    event_count,
    row_number() over (order by event_count desc) as row_num
  from (
    select user_id, count(*) as event_count
    from events
    group by user_id
  )
)
where row_num <= 10;

Per-window top-n (append-only output): same row_number pattern partitioned by window_start, window_end over a window TVF aggregation — full example in references/match-recognize.md.

Deduplication

-- exact deduplication (keep first occurrence)
-- state: O(distinct event_id) until ttl
select
  event_id,
  user_id,
  event_type,
  event_time
from (
  select
    *,
    row_number() over (
      partition by event_id
      order by event_time asc
    ) as row_num
  from events
)
where row_num = 1;

The STATE_TTL hint is a silent no-op on deduplication. Bound dedup state with the global set 'table.exec.state.ttl', or per-operator via a compiled plan (references/compiled-plans.md).

Temporal Joins

Lookup Join (Dimension Enrichment)

-- dimension table with partial lookup cache
create table dim_users (
  user_id string,
  user_name string,
  country_code string,
  primary key (user_id) not enforced
) with (
  'connector' = 'jdbc',
  'url' = 'jdbc:postgresql://postgres:5432/users',
  'table-name' = 'users',
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '10000',
  'lookup.partial-cache.expire-after-write' = '1 h'
);

-- enrich stream with dimension
select
  e.event_id,
  e.user_id,
  u.user_name,
  u.country_code,
  e.event_time
from events as e
join dim_users for system_time as of e.proc_time as u
  on e.user_id = u.user_id;

Versioned Table Join (Point-in-Time Correct)

-- versioned dimension from CDC
create table products (
  product_id string,
  product_name string,
  price decimal(10, 2),
  updated_at timestamp(3),
  watermark for updated_at as updated_at - interval '5' second,
  primary key (product_id) not enforced
) with (
  'connector' = 'kafka',
  'topic' = 'products.cdc.v1',
  'format' = 'debezium-json'
);

-- join stream with version valid at order time
select
  o.order_id,
  o.product_id,
  p.product_name,
  p.price as price_at_order_time,
  o.quantity,
  o.order_time
from orders as o
join products for system_time as of o.order_time as p
  on o.product_id = p.product_id;

Rule: Use for system_time as of for point-in-time correct joins. Critical for SCD scenarios.

Interval Joins (Stream-to-Stream)

Join two streams within a time-bounded window. Both sides must have event-time watermarks.

-- match orders with payments within 1 hour
-- state: bounded by the interval — rows past the bound are dropped
select
  o.order_id,
  o.user_id,
  o.amount as order_amount,
  p.amount as paid_amount,
  p.payment_time
from orders as o
join payments as p
  on o.order_id = p.order_id
  and p.payment_time between o.order_time and o.order_time + interval '1' hour;

Rule: Always include a time bound in the on clause. Without it, Flink keeps all state forever.

MATCH_RECOGNIZE (Pattern Detection)

SQL-based CEP — sequences, fraud patterns, sessionization — plus the windowed top-n variant: references/match-recognize.md.

CDC Patterns

Decision rule: data already in Kafka (Debezium topics) → debezium-json format below. Reading straight from the database, no Kafka hop → Flink CDC source connector; mysql-cdc DDL in references/connectors.md.

-- debezium CDC source (captures inserts, updates, deletes)
create table mysql_orders (
  order_id bigint,
  customer_id bigint,
  status string,
  amount decimal(10, 2),
  created_at timestamp(3),
  updated_at timestamp(3),
  watermark for updated_at as updated_at - interval '5' second,
  primary key (order_id) not enforced
) with (
  'connector' = 'kafka',
  'topic' = 'mysql.inventory.orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'debezium-json',
  'debezium-json.schema-include' = 'false'
);

-- materialize to downstream sink (upsert semantics)
insert into order_summary
select
  customer_id,
  count(*) as order_count,
  sum(amount) as total_amount,
  max(updated_at) as last_order_at
from mysql_orders
group by customer_id;

Malformed records: set the format's ignore-parse-errors option deliberately — it drops them silently. A true DLQ requires DataStream (flink-java-skill).

State TTL

-- global TTL (applies to all queries in this session)
set 'table.exec.state.ttl' = '24 h';

-- per-query TTL via hints — regular joins and group aggregations ONLY;
-- silently ignored on dedup, top-n, and every other stateful operator
select /*+ STATE_TTL('orders' = '7 d', 'users' = '30 d') */
  o.order_id,
  u.user_name,
  o.amount
from orders as o
join users as u
  on o.user_id = u.user_id;

-- group aggregation with TTL hint
select /*+ STATE_TTL('events' = '1 d') */
  user_id,
  count(*) as event_count
from events
group by user_id;

Rule: Always configure state TTL. Unbounded state = OOM eventually.

Per-operator TTL on operators the hint cannot reach, and stateful upgrades across Flink versions: references/compiled-plans.md.

Materialized Tables

Declarative pipeline with a freshness target — replaces hand-rolled insert-into refresh jobs for medallion workloads: references/materialized-tables.md.

Table API

Embedding SQL in a Java job, sqlQuery vs Table DSL, DataStream bridge (toChangelogStream): references/table-api.md.

Batch Processing

Bounded sources, idempotent partition overwrite, incremental backfills: references/batch.md.

Configuration

-- state
set 'table.exec.state.ttl' = '24 h';

-- mini-batch (trade latency for throughput)
set 'table.exec.mini-batch.enabled' = 'true';
set 'table.exec.mini-batch.allow-latency' = '5 s';
set 'table.exec.mini-batch.size' = '5000';

-- source idle timeout (for watermark progress)
set 'table.exec.source.idle-timeout' = '60 s';

-- async lookup
set 'table.exec.async-lookup.buffer-capacity' = '100';
set 'table.exec.async-lookup.timeout' = '3 min';

-- parallelism
set 'parallelism.default' = '4';
set 'table.exec.resource.default-parallelism' = '4';

Anti-Patterns

Missing Watermarks on Event Time

Event-time DDL without a watermark for clause: windows never fire, the job looks healthy while emitting nothing. Fix per the Watermarks section.

No State TTL on Unbounded Aggregations

-- broken: state grows forever for inactive users
select user_id, count(*) as total_events
from events
group by user_id;

-- fixed: set TTL
set 'table.exec.state.ttl' = '7 d';
-- or use hint (valid here: group aggregation)
select /*+ STATE_TTL('events' = '7 d') */
  user_id,
  count(*) as total_events
from events
group by user_id;

Regular Join Instead of Temporal Join

-- broken: regular join keeps all versions, state explodes
-- state: O(rows on both sides) until ttl
select e.*, d.*
from events as e
join dim_products as d
  on e.product_id = d.product_id;

-- fixed: temporal join for dimension lookup
select e.*, d.*
from events as e
join dim_products for system_time as of e.proc_time as d
  on e.product_id = d.product_id;

Processing Time When Event Time Available

-- broken: non-reproducible results, uses deprecated grouped window syntax
select
  tumble_start(proc_time, interval '1' minute) as window_start,
  count(*) as cnt
from events
group by tumble(proc_time, interval '1' minute);

-- fixed: event time with TVF syntax
select
  window_start,
  count(*) as cnt
from table(
  tumble(table events, descriptor(event_time), interval '1' minute)
)
group by window_start, window_end;

Lookup Join Without Caching

-- broken: every record hits database
create table dim_users (...) with (
  'connector' = 'jdbc'
  -- no cache config!
);

-- fixed: enable partial lookup cache
create table dim_users (...) with (
  'connector' = 'jdbc',
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '10000',
  'lookup.partial-cache.expire-after-write' = '1 h',
  'lookup.max-retries' = '3'
);

Interval Join Without Time Bound

A stream-to-stream join whose on clause carries no time bound keeps both sides in state forever. Add and b.t between a.t and a.t + interval '...' per the Interval Joins section.

Non-Deterministic Functions in Streaming

-- broken: different results on replay
select
  event_id,
  current_timestamp as processed_at,
  uuid() as trace_id
from events;

-- fixed: derive from input
select
  event_id,
  event_time as event_at,
  md5(concat(event_id, cast(event_time as string))) as trace_id
from events;

Comment Doctrine

• Comments are for the person reading and reviewing the SQL. Say why a filter, join, or CTE exists, call out anything that is not obvious, and write the reason as a plain sentence. No WHY: label.
• Keep the work plan out of comments. No phases, plan steps, ticket numbers, or words like "for now", "temporary", or "later" (migrations attract this rot the most). Those go stale and someone has to find and delete them. Unfinished work belongs in plan.md.
• Non-trivial queries, views, migrations, and SQL functions get a header comment above them: one line saying what it is for, plus the cost driver. Do not put a compute or time big-O on a query — the planner picks the plan, so an O(...) time bound would be a lie. State size is the exception: O(distinct user_id) or O(rows on both sides) for the state a query retains is good, because that state is memory you own and must bound. Name the cost driver: streaming state size and growth, state TTL, watermark and retention (here memory means managed state). This header is the SQL analogue of the big-O header in the code skills.
• Editing counts too. When you change a query, migration, or function, keep the comments that are there, fix any your change made wrong, and add the missing header, cost note, or reason. Do this only for SQL you add or change, not for nearby SQL you did not touch.

-- per-user running event count, materialized to the summary sink
-- cost driver: one state entry per distinct user_id, growth unbounded without ttl; bound it with table.exec.state.ttl
select /*+ STATE_TTL('events' = '7 d') */
  user_id,
  count(*) as event_count
from events
group by user_id;

Code Review Checklist

• Watermarks: Event time columns have watermark for clause
• State TTL: Global or per-query TTL for unbounded operations (hint covers joins/group aggs only)
• Explicit columns: No select * in production pipelines — upstream schema drift breaks a running job (sql-skill rule, doubly so here)
• Join predicates: All joins have explicit on clause
• Temporal joins: Dimension lookups use for system_time as of
• Interval joins: Stream-to-stream joins have time bounds
• Lookup caching: JDBC/HBase lookups have lookup.cache = PARTIAL configured
• Statement sets: Multi-sink jobs use begin statement set
• Deterministic functions: No current_timestamp, uuid() in streaming
• Connector configs: Exactly-once needs sink.delivery-guarantee + sink.transactional-id-prefix
• Malformed records: Format ignore-parse-errors set deliberately (drops silently); true DLQ needs DataStream (flink-java-skill)
• Naming: snake_case tables/columns, lowercase keywords, trailing commas
• Comments: Non-trivial queries/views/migrations have a header (what it is for + cost driver), no work-plan notes — see Comment Doctrine