[INIT-6549] Add client-side synchronous snapshot queries to the Flink CLI

[Yiyu Tian (yiyutian1)]

Implements the client-side path for one-shot Flink SQL snapshot queries (INIT-6549 M1, PRD R3): submit → block → drain every page → print → exit non-zero on failure.

example use case:

$ confluent query --sql "SELECT order_id, status FROM orders LIMIT 2;" --compute-pool lfcp-123456 --database my-cluster
+----------+---------+
| order_id | status  |
+----------+---------+
| 1021     | SHIPPED |
| 1044     | PENDING |
+----------+---------+

Mounted at the top level (confluent query), not under flink statement — per discussion with Florian Eiden (@fleidcc), the same one-shot ergonomics should extend to other backends (e.g. Lightning Tables) later without a rename. -o json/-o yaml default to a self-describing envelope (schema + rows); --raw gives a bare row array.

Verified against real staging. Submit → multi-page drain → exit-code behavior confirmed on a real compute pool. That run found and fixed two defects:

1. Stop path was broken. The gateway rejects a body carrying only spec.stopped; every stop (interrupt, timeout, unbounded rejection) left the statement RUNNING. Fixed by reading the statement back and flipping the flag on it, like statement stop does. The mock server was more permissive than the real gateway and missed this — it's now strict enough to catch it.
2. Values serialized as strings only (e.g. an INTEGER came back as "3065"). Now type-aware: numbers as numbers, NULL as null.

Why a separate drain loop instead of the shell's Store/ResultFetcher pipeline: that pipeline is built for a scrolling viewer and degrades silently in ways that become real bugs for a script reading stdout (row-cap eviction, schema-mismatch rows dropped, "done" inferred from a missing page token without checking phase). This drain loop reports each of those conditions instead of hiding them.

Not done: on-prem support, statement cleanup on success, --unsafe-trace still dumps row data, no token-refresh-aware retry beyond a best-effort refresh before each call.

Testing: 14 unit tests over a mocked gateway (multi-page drain, empty-token-while-running, --max-rows boundary, unbounded rejection, schema mismatch, cancellation). make lint-cli clean. Integration goldens deferred — the mount point and flags are still in flux pending PM sign-off.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Pull request overview

Adds a new Cloud-only synchronous, one-shot Flink SQL “snapshot query” path to the CLI (confluent flink statement query), backed by a new client-side drain loop that submits, waits for traits/schema, drains paginated results, and renders machine-readable output.

Changes:

• Introduces pkg/flink/query to await non-PENDING statements and drain paginated results into an in-memory result with truncation/incomplete signaling.
• Adds the flink statement query command with flags (--sql, --timeout, --max-rows, --property, compute pool/db selection), plus human + JSON/YAML rendering.
• Extends the test gateway router with a /results endpoint and adds unit + integration coverage with new golden fixtures.

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
test/test-server/flink_gateway_router.go	Adds a /results route and fixtures for paginated query results + statement traits used by the new query command.
test/flink_test.go	Adds an integration test suite for flink statement query covering table/json/yaml, max-rows, empty, changelog, unbounded, and failed cases.
test/fixtures/output/flink/statement/query.golden	Golden output for the default human/table rendering.
test/fixtures/output/flink/statement/query-yaml.golden	Golden output for YAML envelope rendering.
test/fixtures/output/flink/statement/query-unbounded.golden	Golden output for unbounded-query rejection behavior.
test/fixtures/output/flink/statement/query-negative-max-rows.golden	Golden output for negative --max-rows validation.
test/fixtures/output/flink/statement/query-max-rows.golden	Golden output for truncation warning + truncated table rendering.
test/fixtures/output/flink/statement/query-json.golden	Golden output for JSON envelope rendering.
test/fixtures/output/flink/statement/query-help.golden	Golden output for command help text and flags.
test/fixtures/output/flink/statement/query-failed.golden	Golden output for statement failure messaging/suggestions.
test/fixtures/output/flink/statement/query-empty.golden	Golden output for the “no rows” case.
test/fixtures/output/flink/statement/query-changelog.golden	Golden output for changelog warnings + Operation column in table mode.
pkg/flink/types/statement_traits.go	Adds helpers to read boundedness and append-only traits in a backend-agnostic way.
pkg/flink/types/processed_statement.go	Extends internal phase enum to include STOPPED and DELETING.
pkg/flink/query/README.md	Documents the new query package, limitations, and open questions.
pkg/flink/query/query.go	Implements await + drain loop for synchronous, bounded snapshot querying with truncation/incomplete detection.
pkg/flink/query/query_test.go	Adds unit tests covering multi-page drain, pending wait, empty-page retry, truncation, schema mismatch, cancellation, and gateway errors.
internal/flink/command_statement.go	Registers the new flink statement query subcommand.
internal/flink/command_statement_query.go	Implements command wiring, flag parsing, submit + drain orchestration, stop-on-interrupt, and output rendering.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Yiyu Tian (yiyutian1)]

Superseded by #3466. This PR's branch lives on a personal fork, and Semaphore couldn't check it out (git@github.com: Permission denied (publickey) on clone), so CI never ran here. #3466 has the same two commits, rebased onto current main, on a branch pushed directly to this repo instead.