flink-sql

name	flink-sql
description	Authoritative guardrails for Confluent Cloud Flink SQL — dialect traps vs Apache Flink, red flags to verify, anti-patterns, and the live `confluent flink` verification loop. TRIGGER when writing or reviewing Flink SQL DDL/DML for Confluent Cloud, working with `confluent flink` CLI, deploying CC Flink UDFs / artifacts / connections, configuring CC compute pools, or writing Terraform that targets the Confluent provider's Flink resources. SKIP for Apache Flink (OSS), Flink DataStream/Table API in Java/Scala, or non-Flink Kafka work.
user-invocable	true
allowed-tools	["Bash(confluent *)","Bash(terraform *)","Read","Write","Edit","WebFetch"]

Confluent Cloud Flink SQL — non-negotiables

This project runs on Confluent Cloud Flink SQL, not Apache Flink. CC Flink is SQL-only, runs on managed compute pools (CFUs), auto-maps Kafka topics to tables, and exposes a constrained dialect. Apache Flink knowledge is not authoritative — training data heavily favors OSS Flink and silently drifts into wrong syntax, unavailable APIs, and nonexistent features.

Authority order

1. Confluent Cloud Flink docs — https://docs.confluent.io/cloud/current/flink/
2. Confluent Cloud Flink SQL reference — https://docs.confluent.io/cloud/current/flink/reference/
3. Confluent Terraform provider — https://registry.terraform.io/providers/confluentinc/confluent/latest/docs
4. Confluent Developer portal - https://developer.confluent.io/tutorials/#flink Follow to Flink subpages. On each subpage, only follow the page section labeled as Confluent Cloud
5. Live confluent flink shell against a real compute pool — minimal reproducer, observed output.

Apache Flink docs, StackOverflow answers tagged apache-flink, and LLM memory of "Flink SQL" are not authoritative. If an answer relies on any of those, verify against sources 1–4 before acting.

Red flags — stop and verify

• Any DataStream / Table API (Java/Scala) reference — CC Flink is SQL-only at runtime.
• CREATE CATALOG ... — catalog is the CC environment, database is the Kafka cluster. Not creatable.
• SET 'execution.checkpointing.interval' or any savepoint / STOP WITH SAVEPOINT — CC manages checkpointing; user-set rejected; savepoints not exposed. Statement deletion = state loss.
• Grouped window syntax: GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE) — CC requires table-valued: TUMBLE(TABLE t, DESCRIPTOR(ts), INTERVAL '1' MINUTE).
• CREATE TABLE ... WITH ('connector' = 'kafka', ...) — tables auto-map from Kafka topics; manual connector DDL is limited.
• CREATE FUNCTION f AS '...' without USING JAR 'confluent-artifact://<id>' — CC UDFs require artifact upload first.
• UDF open(RuntimeContext) with filesystem / socket access — CC sandbox has no filesystem and no outbound sockets without a CONNECTION object + USING CONNECTIONS clause on the function.
• CTE before INSERT — CC requires INSERT INTO <table> WITH cte AS (...) SELECT ..., NOT WITH cte AS (...) INSERT INTO ....
• $rowtime aliased in CTE — silently strips time-attribute property. Never alias it; reference as $rowtime throughout.
• docs.confluent.io returning SPA redirects — if WebFetch/web search gets homepage for deep-linked CC Flink doc URLs, the docs site is having issues. Do NOT fall back to Apache Flink docs. Use confluent flink shell or Terraform examples as ground truth instead.
• RBAC roles not matching docs — DeveloperManage may not grant ALTER TABLE despite docs claiming it does. Verify permissions empirically before assuming doc accuracy.
• CC Flink does NOT support PROCTIME() / processing-time lookup joins — always verify SQL functions against CC Flink docs before recommending
• CC Flink uses unaligned checkpoints by default (not aligned)
• Verify Flink SQL features via Glean search before recommending them

Dialect trap table

Apache Flink SQL	Confluent Cloud Flink SQL
CREATE CATALOG ...	Catalog = env, database = cluster; not creatable
SET 'execution.checkpointing.*'	CC manages checkpointing; not user-settable
CREATE TABLE ... WITH ('connector'='kafka')	Tables auto-mapped from topics; manual DDL limited
WITH cte AS (...) INSERT INTO ...	INSERT INTO ... WITH cte AS (...) — CTE comes AFTER INSERT INTO
CREATE TEMPORARY VIEW v AS (...)	Not supported — must use CTE (WITH v AS (...))
INSERT INTO ... VALUES	Materializes; test only, not production
Custom event_time column for rowtime	Use $rowtime system column; never qualify as t.$rowtime
$rowtime AS alias in CTE	Silently strips time-attribute — downstream watermark breaks
ALTER TABLE t ADD WATERMARK ...	Uses MODIFY not ADD for watermarks
SOURCE_WATERMARK()	Adaptive histogram, 95th percentile, requires 250+ events before emitting
Multiple OVER windows in one query	Not supported
MATCH_RECOGNIZE with PREV()/NEXT()	No physical offsets, flat columns only, no greedy quantifiers as last pattern
CAST('2024-01-01T00:00:00Z' AS TIMESTAMP)	ISO-8601 fails — use TO_TIMESTAMP(REPLACE(...))
UNIX_TIMESTAMP(ts_column) on TIMESTAMP	Only accepts STRING — cast first
SESSION_START(ts) with one arg	Requires two arguments
current_watermark() on updating table	Not supported — non-deterministic function with update messages error
UDF open(RuntimeContext)	Sandbox — no FS, no sockets without CONNECTION + USING CONNECTIONS
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE)	TUMBLE(TABLE t, DESCRIPTOR(ts), INTERVAL '1' MINUTE)
CREATE FUNCTION f AS '...'	... USING JAR 'confluent-artifact://<id>' + USING CONNECTIONS for egress
'value.format' = 'json'	CC may require 'json-registry' with Schema Registry
Savepoints, STOP WITH SAVEPOINT	Not exposed; statement deletion = state loss; use prevent_destroy in TF
DataStream / Table API	SQL only

Maintain a running docs/flink-dialect-traps.md — append every new CC-vs-OSS gotcha hit.

Verification — no mocks for the SQL path

Any claim that a CC Flink feature or SQL construct works must be backed by a real command against the live environment. Canonical loop:

1. Write minimal reproducer in repro/<slug>.sql.
2. Run via confluent flink shell or confluent flink statement create.
3. Consume downstream topic to confirm emitted records.
4. Paste command + output into docs/verification-<slug>.md.

Canonical commands:

confluent flink shell --compute-pool <pool-id> --environment <env-id>
confluent flink statement create <name> --sql-file repro/<slug>.sql --compute-pool <id>
confluent flink statement list --compute-pool <pool-id>
confluent flink statement describe <name>
confluent flink statement exception list <name>
confluent flink statement delete <name>
confluent flink artifact create --artifact-file <jar>
confluent flink artifact list
confluent kafka topic consume <topic> --from-beginning --value-format avro
confluent schema-registry schema describe --subject <subject>

Stuck states — escalate, do not silently work around:

• Statement stuck in PENDING > 60s after submit → confluent flink statement exception list <name>.
• UDF deploy fails with "jar not found" → confluent flink artifact list, verify the artifact ID used in USING JAR.
• Schema inference mismatch → DESCRIBE <table>, diff against producer schema.
• Egress denied from UDF → verify CREATE CONNECTION exists and function declares USING CONNECTIONS ('<name>').

Anti-patterns (hard no)

• Treating Apache Flink docs or apache-flink-tagged answers as authoritative for CC.
• Mocking confluent CLI or the Kafka / Schema Registry side in anything claiming end-to-end verification.
• terraform apply -auto-approve on first run of a root module.
• Committing .tfvars, .tfstate*, .terraform/, *.secret*.
• Hardcoded secrets in CREATE CONNECTION or UDF source. TF-inject connection parameters; never commit.
• Swallowing Flink statement exceptions to make a test pass. Fail loud; read statement exception list.

CLI session expectation

• confluent CLI v4.53+ logged in at host level — confluent organization list succeeds without prompting.
• Cloud-side CLI (az / aws / gcloud) logged in at host level.
• Terraform providers inherit from CLI sessions where possible.
• Environment context — CC_ORG_ID, CC_ENV_ID, CC_CLUSTER_ID, CC_COMPUTE_POOL_ID, Kafka bootstrap, SR endpoint — captured once in docs/env.md.