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

Diagram here: here

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 Jim Hughes and Florian Eiden, 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.

Open questions: final mount point/verb, output shape (R3 wants a bare array, R10 wants a typed schema — currently reconciled via envelope-by-default + --raw), timeout behavior (currently stops the statement and exits non-zero; PRD wants graceful degradation).

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.

🤖 Generated with Claude Code

[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 client-side “snapshot query” execution path for Flink SQL, including a top-level confluent query command that submits a statement, blocks until it leaves PENDING, drains all result pages, and prints results (table or serialized JSON/YAML), with best-effort stop behavior on interrupts/timeouts.

Changes:

• Introduces pkg/flink/query with an await+drain loop for synchronous result collection (including boundedness checks, truncation/incomplete signaling, and wrapped fetch errors).
• Adds SQL-type-aware serialization via StatementResultField.ToSerializedValue() and corresponding unit tests.
• Mounts the new confluent query command at the CLI root and tightens the Flink gateway mock to reject malformed stop/update requests consistent with the real gateway.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
test/test-server/flink_gateway_router.go	Makes the mock gateway stricter by rejecting statement updates that omit SQL text, matching real gateway behavior and catching broken stop paths.
pkg/flink/types/statement_traits.go	Adds helpers to read bounded/append-only traits with “known/unknown” signaling.
pkg/flink/types/result_fields.go	Extends StatementResultField with ToSerializedValue() for typed JSON/YAML output.
pkg/flink/types/result_fields_serialized.go	Implements typed serialization rules (e.g., NULL→null, small ints→numbers, BIGINT/DECIMAL→strings).
pkg/flink/types/result_fields_serialized_test.go	Unit tests for typed serialization behavior and JSON round-trip precision expectations.
pkg/flink/types/processed_statement.go	Adds STOPPED/DELETING phases for terminal-phase handling.
pkg/flink/query/README.md	Documents rationale, drain-loop semantics, and known limitations of the synchronous query path.
pkg/flink/query/query.go	Implements query.Run, await, drain, and terminal-phase logic for synchronous snapshot queries.
pkg/flink/query/query_test.go	Unit tests for paging, truncation, incomplete detection, boundedness rejection, and error wrapping.
internal/query/command.go	Adds the confluent query Cobra command, wiring flags, submit/run/stop flow, and output formatting.
internal/command.go	Registers the new root-level query command.

Suppressed comments (1)

internal/query/command.go:539

• The JSON/YAML envelope is constructed without surfacing result.Incomplete, so even if the drain loop flags the result as incomplete, serialized output won't include it.

			Truncated:     result.Truncated,

---

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

[Jim Hughes (jnh5y)]

An automated review pass flagged a few things worth a human glance. These scored moderately-to-highly in initial triage but weren't independently re-verified in a second pass, so please treat them as leads rather than confirmed issues.

Suggestion — flag formatting in error/suggestion messages (internal/query/command.go)

• Line 244: error wraps --max-rows in double quotes instead of backticks
• Line 252: wraps --raw, -o json, -o yaml in double quotes instead of backticks (inconsistent with the correctly-backticked error at line 216 in the same file)
• Line 422: within one suggestion string, `confluent flink statement describe` is backticked correctly but --timeout uses escaped double quotes

This repo's output-formatting convention documents flags as backtick-formatted — these three look like a consistent slip across the new command's error paths.

Suggestion — test coverage (test/fixtures/output/query/help.golden)

• The new command and its 8 flags currently ship with only a --help golden test. No integration test appears to exercise actual query execution, error paths, or flag behavior yet.

Suggestion — root command registration (internal/command.go:43,136)

• This PR adds the import and cmd.AddCommand(query.New(...)) directly. Worth double-checking this was cleared as intended, since root command registration is a file the team usually likes to loop reviewers in on.

Suggestion — stale doc line (pkg/flink/query/README.md:72-74)

• The "Known limitations" section says "no token refresh," but this PR adds an Options.RefreshToken mechanism invoked before every gateway call. Looks like this line wasn't updated after the refresh logic landed.

Suggestion — comment cites the wrong precedent (internal/query/command.go:70-74)

• A comment cites unified-stream-manager as precedent for the Hidden/feature-flag gating pattern, but that package doesn't actually implement any Hidden/flag gating. Might be worth pointing at the actual precedent instead.

Comment generated with the help of an AI agent

[sonarqube-confluent]

Quality Gate failed

Failed conditions
67.2% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube