text2flink

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:

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:

{ "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.)

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:

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 for the AI-agent path, docs/VERIFY.md for the CLI/CI path, 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.

• 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.

• 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/, grouped into core / advanced / adversarial tiers. Add a task or submit a model result without writing engine code (see CONTRIBUTING.md). Live results in LEADERBOARD.md:

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

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.) 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).

Run it

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:

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

Related MCP server: Kafka MCP Server

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

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.