Phase 7: Table API, Streaming SQL Engine & Complex Event Processing (CEP)¶

Audience: Principal Distributed Systems Architect & Real-Time Analytics Engineer
Source References: flink-table/, flink-cep/

1. Dynamic Tables & Continuous Query Duality¶

Flink SQL treats streaming data as Dynamic Tables governed by continuous mathematical duality:

The Stream-Table Duality:
Stream ────────► Continuous Query ────────► Dynamic Table
  ▲                                               │
  └──────────────── Changelog Stream ─────────────┘

Dynamic Table: A table that changes continuously over time as new events arrive.
Continuous Query: A declarative SQL query that never terminates, continuously updating its dynamic result table.
Changelog Stream: The physical stream of delta mutations emitted by the dynamic table to downstream operators.

2. Retract Streams & The 4 `RowKind` Markers¶

When an aggregation is executed (e.g., COUNT(*) or SUM()), an incoming event can modify a previously emitted calculation. Flink handles this through Retract Streams using four internal RowKind byte markers:

┌─────────┬───────────────────┬──────────────────────────────────────────────────┐
│ Marker  │ RowKind Name      │ Semantic Meaning                                 │
├─────────┼───────────────────┼──────────────────────────────────────────────────┤
│ +I      │ INSERT            │ Insertion of a brand new record                  │
│ -U      │ UPDATE_BEFORE     │ Retraction / cancellation of the previous record │
│ +U      │ UPDATE_AFTER      │ New replacement value following the update       │
│ -D      │ DELETE            │ Explicit deletion of a previously emitted record │
└─────────┴───────────────────┴──────────────────────────────────────────────────┘

2.1 Retract Stream Execution Walkthrough¶

Consider:

SELECT user_id, COUNT(*) as order_count FROM orders GROUP BY user_id;

Event Stream Progression:
1. User "Alice" places Order 1:
   - Flink emits: [+I, "Alice", 1]

2. User "Alice" places Order 2:
   - Flink must update the count from 1 to 2.
   - Flink emits: [-U, "Alice", 1]  <--- Retracts previous answer!
   - Flink emits: [+U, "Alice", 2]  <--- Emits correct updated answer!

The Sink Requirement: Upsert Connectors¶

If a downstream sink does not support retractions (e.g., standard append-only file sinks), the query fails compilation! For retract streams, you must use Upsert Sinks (e.g., upsert-kafka, JDBC sink, or Elasticsearch) that apply updates based on primary keys.

3. Apache Calcite Query Optimization Engine¶

Flink Table/SQL leverages Apache Calcite as its core relational compiler:

Flink SQL Compilation Architecture:
[User SQL Query]
       │
       ▼
[SQL Parser: JavaCC / Calcite] ──► Validated Relational AST
       │
       ▼
[Logical Optimization: HepPlanner] ──► Pushdown Predicates ($filter pushdown)
       │
       ▼
[Cost-Based Optimizer: VolcanoPlanner] ──► Evaluates candidate physical execution DAGs
       │                                  (Selects join algorithms: Hash vs Two-Way Sort-Merge)
       ▼
[Bytecode Generation (Janino JIT)] ──► Compiles directly into high-speed Java bytecode!

4. Complex Event Processing (CEP) Internals¶

While standard SQL filters records individually, Flink CEP detects complex temporal patterns across sequences of multiple events:

// Financial Fraud Pattern: 3 consecutive login failures within 2 minutes followed by a withdrawal
Pattern<Event, ?> fraudPattern = Pattern.<Event>begin("failed_login")
    .where(SimpleCondition.of(e -> e.getType().equals("LOGIN_FAILURE")))
    .times(3)
    .consecutive()
    .followedBy("withdrawal")
    .where(SimpleCondition.of(e -> e.getType().equals("WITHDRAWAL")))
    .within(Time.minutes(2));

PatternStream<Event> patternStream = CEP.pattern(keyedStream, fraudPattern);

4.1 The Non-Deterministic Finite Automaton (NFA)¶

Under the hood, Flink CEP compiles the pattern into an in-memory NFA (Non-Deterministic Finite Automaton). - Each incoming event transitions the NFA between states (Start -> State 1 -> State 2 -> Matched). - The NFA state paths are stored in Flink's Keyed State Backend (RocksDB), guaranteeing that even if a TaskManager crashes mid-pattern, pattern detection resumes without dropping state!