# text2flink

[![CI](https://github.com/clementlemon02/text2flink/actions/workflows/ci.yml/badge.svg)](https://github.com/clementlemon02/text2flink/actions/workflows/ci.yml)

> **Alpha: runnable & tested, APIs may change.**

**Give your AI coding agent the ability to write Apache Flink SQL that provably works.**

AI agents write *plausible* Flink SQL — and streaming SQL is exactly where plausible isn't
correct. A 30-second window where you meant one minute; late records kept when they should be
dropped. It compiles, it runs, it returns believable rows, and it silently under-reports in
production for weeks. The agent has no way to know it got the watermark or window semantics
wrong — a compile check and a golden-file diff both pass.

text2flink closes that loop. Register it once as an **MCP server** and your agent can verify the
SQL *it just wrote* on a **real Flink cluster** and self-correct from precise feedback — before
you ever see it:

```bash
claude mcp add text2flink -- python3 -m text2flink.mcp_server
```

The agent calls **`verify_flink_sql`** with its SQL + a schema + sample data + the intended
semantics, and gets back a machine-readable verdict it can act on:

```jsonc
{ "verified": false,
  "violations": [
    { "kind": "value_mismatch",  "detail": "(00:00:00, u1): got 2.0, expected 3.0" },
    { "kind": "spurious_window", "detail": "(00:00:30, u1) emitted but not expected" } ] }
```

The agent reads the violations ("my window is 30s, should be 60s"), fixes the SQL, and verifies
again — a **~1–2s loop** — until `verified: true`. Not "does it parse", not a golden diff:
**actual streaming correctness, run on real Flink.** (The server also exposes
`generate_flink_job` / `ground_kafka_topic` / `deploy_to_kafka` for the draft-and-ship flow — see
[docs/USE_CASES.md](docs/USE_CASES.md#use-it-from-an-ai-assistant-mcp).)

### Or drive it yourself — CLI + CI

The same engine runs standalone. Write your Flink `SELECT` and a small test-file next to it
(sample data + the semantics you expect); `verify` grades it on Flink and exits non-zero on
failure, so it drops straight into a CI gate:

```bash
text2flink verify orders_per_min.test.json
#  ✘ FAIL  orders_per_min
#    [value_mismatch] (00:00:00, u1): got 2.0, expected 3.0   ← 30s window, not 1 min
```

Fastest taste, no setup: `python3 examples/verify_flink_sql.py` verifies a correct job, then
catches a subtly-broken one.

**New here?** [docs/USE_CASES.md](docs/USE_CASES.md#use-it-from-an-ai-assistant-mcp) for the AI-agent
path, [docs/VERIFY.md](docs/VERIFY.md) for the CLI/CI path, [DESIGN.md](DESIGN.md) for the
architecture.

## Status

Alpha — runnable and execution-verified today; APIs may change. The core loop is proven: take a
Flink SQL job (hand-written or generated), run it on a **real** local Flink cluster, and assert
**streaming-specific properties** (windows fire once, counts match a ground truth, event-time
and watermarks respected) instead of brittle golden diffs — and, for generated jobs, **repair** a
wrong one automatically from structured verification feedback.

### What it does today

- **Verify an AI agent's Flink SQL over MCP** — the `verify_flink_sql` tool runs the SQL the
  agent wrote on real Flink and returns `verified` + machine-readable violations, so the agent
  self-corrects in a ~1–2s loop. This is the headline; see
  [docs/USE_CASES.md](docs/USE_CASES.md#use-it-from-an-ai-assistant-mcp).
- **Verify hand-written Flink SQL** — `text2flink verify job.test.json` runs your own SELECT on
  real Flink and grades it against the streaming semantics you declare, with precise violations
  on failure and a non-zero exit for CI. See [docs/VERIFY.md](docs/VERIFY.md).
- **Draft a job from natural language** — NL → JobSpec extraction via a model-agnostic LLM layer
  that runs on OpenAI (`OPENAI_API_KEY`) or Anthropic (`ANTHROPIC_API_KEY`), with an **offline
  heuristic proposer** fallback so everything runs with no API key. A convenience entry point
  into the same verifier.
- **Windowed aggregates** — tumbling, hopping, and **session** windows (per-key inactivity-gap
  merging via the Flink `SESSION` TVF).
- **Updating operators** — unbounded `GROUP BY` (running totals), **top-N**, and
  **deduplication** (`ROW_NUMBER()`), verified via batch-mode collect and deployed to an
  **upsert-kafka** topic keyed by their partition keys.
- **Joins** — time-bounded **interval joins** of two event-time streams, and **temporal
  (versioned-table) joins** (`FOR SYSTEM_TIME AS OF`). Semantics confirmed against real Flink
  before being encoded in the oracle.
- **Late-data / watermark correctness** as a first-class, tested dimension: the oracle drops
  records whose event-time is at/behind the watermark, and a task verifies the generated job
  drops exactly those records — the thing batch Text2SQL cannot express.
- **Live Kafka grounding** — discover a source schema from a real Kafka topic (sample → infer →
  wire the connector), graded by running on real Flink against the real topic in bounded mode.
  Avro grounding is supported via the Flink `avro` format end-to-end; the Confluent Schema
  Registry (`avro-confluent`) path is verified offline against a mock registry.
- **Deployable Kafka pipelines (topic → topic)** — a spec compiles to a complete
  `INSERT INTO <sink> SELECT …` script; still verified by running to completion and grading the
  sink topic against the oracle.
- **Multiple codegen targets** — Flink SQL (execution-verified), a submittable **PyFlink** Table
  API program (`flink run -py job.py`), and **Apache Spark SQL** (a portability proof graded
  against the same oracle; the product stays Flink-first).

### StreamBench

A declarative, contributable benchmark scored by **execution pass rate**, not string match —
22 tasks as JSON files in [`streambench/tasks/`](streambench/tasks/), grouped into
`core` / `advanced` / `adversarial` tiers. Add a task or submit a model result without writing
engine code (see [CONTRIBUTING.md](CONTRIBUTING.md)). Live results in
**[LEADERBOARD.md](LEADERBOARD.md)**:

| Proposer | core | advanced | adversarial | **total** |
|---|---|---|---|---|
| `openai-gpt-4o` | 6/6 | 9/9 | 7/7 | **22/22 (100%)** |
| `heuristic` (offline) | 6/6 | 9/9 | 0/7 | **15/22 (68%)** |

The **adversarial** tier (misleading phrasing, implicit/negated filters, multi-key grouping,
reworded windows, non-second time units) makes the benchmark *discriminating*, and has surfaced
real bugs no prior task exercised. Every task's gold spec is execution-verified on Flink
(`scripts/check_gold.py`), so a broken task can't hide as a proposer failure.

### Roadmap

Confluent Schema Registry (`avro-confluent`) end-to-end, DataStream codegen, K8s deployment
manifests, and a bigger adversarial tier.

### Requirements

This machine's default Java (26) and Python (3.14) are too new for Flink, so Phase 0 uses:
- **Apache Flink** + **JDK 17**, installed via Homebrew (`brew install apache-flink openjdk@17`).
- Python 3.11+ for the orchestrator (no PyFlink dependency — it drives Flink's SQL client as a
  subprocess, so the system Python is fine).
- **For the live-Kafka example only:** `brew install kafka`, plus the Flink Kafka SQL connector
  jar in Flink's `lib/` (e.g. `flink-sql-connector-kafka-5.0.0-2.2.jar` from Maven Central).
- **For the Avro example:** the Flink Avro format jar in Flink's `lib/` (e.g.
  `flink-sql-avro-2.2.1.jar` from Maven Central).
- **For the (experimental) cross-engine example only:** `brew install apache-spark` (JDK 17).

### Install

```bash
pip install "git+https://github.com/clementlemon02/text2flink.git"   # the library + `text2flink` CLI
```

(A `pip install text2flink` from PyPI is a tag away — see [RELEASING.md](RELEASING.md).)
Contributors work from a checkout instead: `pip install -e ".[dev]"` (editable, with pytest).
No install is even required to try it: `python3 -m text2flink.cli verify <file>` runs from a
checkout, and the example scripts below run directly. Python 3.10+; running a verify needs a
local Flink + JDK 17 (see [Requirements](#requirements)).

### Run it

```bash
text2flink verify examples/verify/orders_per_min.test.json  # verify your Flink SQL (see docs/VERIFY.md)
python3 examples/verify_flink_sql.py   # verify a correct job, then catch a subtly-broken one
python3 examples/tumbling_count.py     # generate a job, verify it, then catch & repair a broken one
python3 examples/run_streambench.py    # run StreamBench, report execution pass rate
```

`run_streambench.py` uses the offline heuristic proposer by default. To use a real model:

```bash
OPENAI_API_KEY=sk-... python3 examples/run_streambench.py       # OpenAI (default gpt-4o)
ANTHROPIC_API_KEY=sk-... python3 examples/run_streambench.py    # Anthropic
```

## Layout

```
text2flink/
  ir.py          # IR: JobSpec (windows/agg) + IntervalJoinSpec + TemporalJoinSpec
  codegen.py     # IR -> Flink SQL: aggregates, interval joins, temporal joins
  data.py        # row-dict -> Flink CSV
  oracle.py      # reference interpreter: ground truth incl. late-drop + session windows
  runtime.py     # local Flink cluster lifecycle + run SQL (sql-client), collect results
  gateway.py     # SQL Gateway harness: submit over REST (~1-2s/case, no per-run JVM boot)
  assertions.py  # property-based streaming correctness checks
  verify.py      # verify(spec) + verify_sql/verify_batch/gateway -> run on Flink -> assert
  testfile.py    # load a verify test-file (SQL + sources + data + expected semantics)
  cli.py         # `text2flink verify` / `text2flink cluster` — the correctness-layer command
  llm.py         # model-agnostic LLM clients: OpenAI + Anthropic (raw HTTPS, no SDK)
  proposer.py    # NL -> JobSpec + repair: LLM / offline-heuristic / scripted proposers
  pipeline.py    # extract -> verify -> repair loop for one task
  kafka.py       # live Kafka grounding: sample a topic, infer schema, build a Source
  schema_registry.py # Avro/Schema-Registry grounding: exact types via avro-confluent
  deploy.py      # deployable INSERT jobs (append + upsert sinks), verify by consuming
  mcp_server.py  # MCP server (stdio) — lets an AI assistant call text2flink as a tool
  mcp_tools.py   # generate-and-run-on-Flink logic behind the MCP tool
  pyflink.py     # third codegen target: same IR -> a submittable PyFlink Table API program
  spark.py       # EXPERIMENTAL portability proof: same IR -> Spark SQL, same oracle
  streambench.py # loader for the StreamBench corpus
streambench/
  tasks/*.json   # the benchmark corpus (declarative, contributable)
examples/
  verify_flink_sql.py  # verify hand-written Flink SQL; catch a subtly-broken job
  verify/              # a verify test-file (.test.json) + the .sql it checks
  tumbling_count.py
  run_streambench.py
  kafka_grounding.py   # discover schema from a live Kafka topic, then verify
  kafka_pipeline.py    # deployable topic -> topic pipeline (INSERT INTO sink), verified
  avro_pipeline.py     # real Avro end-to-end on Flink (no Schema Registry)
  cross_engine.py      # (experimental) one IR compiled + verified on Flink AND Spark
tests/           # fast offline suite (no Flink): corpus, oracle, codegen, parsing
LICENSE          # Apache-2.0
CONTRIBUTING.md  # how to add a StreamBench task
.github/workflows/ci.yml   # runs the offline tests on 3.10–3.12
```

## Development

```bash
pip install -e ".[dev]"
pytest                          # fast, offline — no Flink required
python3 scripts/check_gold.py   # verify every task's gold spec on Flink (task soundness)
python3 scripts/leaderboard.py  # score a proposer, update results/ + LEADERBOARD.md
```

Contributions welcome — the highest-value ones are **a new StreamBench task** (especially
adversarial) and **submitting a model to the leaderboard**. See [CONTRIBUTING.md](CONTRIBUTING.md).