[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

[Yiyu Tian (yiyutian1)]

Thanks Jim Hughes (@jnh5y) — addressed most of this in 24468a1:

• backticked --max-rows, --raw, -o json/-o yaml, and --timeout in the error/suggestion strings
• fixed the stale precedent comment — unified-stream-manager doesn't have this pattern, pointed it at the private link ingress endpoint command instead
• corrected the README's "no token refresh" line now that Options.RefreshToken exists, plus a note on why the default timeout makes it rarely matter in practice
• added unit tests for internal/query (command_test.go) to clear the Sonar new-code coverage gate

Root registration in internal/command.go was intentional, same pattern as the other top-level commands.

Still open: integration goldens for actual query execution (only unit tests + help golden so far) — holding off until the mount point/output shape are settled with PM, same as noted in the PR description.

[sonarqube-confluent]

Quality Gate failed

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

See analysis details on SonarQube

[Jim Hughes (jnh5y)]

A few more things from a follow-up pass, including one I reproduced live against a staging compute pool.

Recommendation — intermittent false "Incomplete" on a query that did finish (pkg/flink/query/query.go, drain loop)
The drain loop reads the statement's phase before fetching a page, but doesn't re-check it after a page with no next token comes back. If the statement transitions to a terminal phase in that window, the result gets marked Incomplete (stderr warning, incomplete: true in JSON, and an unnecessary stop attempt) even though the data is actually complete. I reproduced this live: running confluent query --sql "SELECT 1 AS a, 2 AS a" against a staging compute pool, the first invocation hit exactly this — a "gateway stopped returning result pages while statement ... was still in phase RUNNING" warning on a trivial single-row literal query with no source table. Three immediate reruns all completed cleanly, so it looks cold-start/latency-triggered rather than a steady bug, but it's not just theoretical — worth a re-read-phase-after-final-page fix.

Suggestion — stopStatement nil-deref risk (internal/query/command.go:480-491)
stopStatement dereferences statement.Spec right after GetStatement succeeds, with no nil check. If the gateway ever returns a statement with a nil Spec (already-deleted, mid-transition, or an unexpected shape), this panics instead of surfacing the "could not stop" warning the rest of the function is built to produce. Note internal/flink/command_statement_stop.go has the identical pattern already, so this would be consistent with existing code — just flagging in case it's worth hardening both at once.

Suggestion — serialized output silently truncates on a field/header-count mismatch (internal/query/command.go:522-526)
The row-builder loop guards field access with if j < len(headers), which means any row with more fields than headers just drops the extra fields instead of erroring. That's the same silent-short-read failure class the package's README.md says it was written specifically to avoid. Whether it's reachable depends on whether ConvertToInternalResults/the gateway response can ever produce a mismatch — if that's already guaranteed upstream, this is dead code and could just get an explanatory comment; if not, it should probably error loudly instead.

Comment generated with the help of an AI agent

[Jim Hughes (jnh5y)]

Follow-up on the dropped-changelog-operation gap from earlier: I think sql.snapshot.mode needs to be non-overridable, not just a default, given how serialized output handles (or doesn't handle) non-append-only results.

Recommendation — --property can silently defeat the snapshot-mode guarantee that serialized output depends on

buildQueryProperties seeds sql.snapshot.mode=now as a default, but the comment says this is intentional: --property entries are merged in afterward with nothing stopping a value from overriding it. The independent RequireBounded: true check only enforces boundedness, not append-only-ness — a bounded statement can still emit a raw update/delete changelog.

The command already detects this case at runtime: runQuery computes appendOnlyKnown && !isAppendOnly and, when true, prints "Warning: this statement emits updates and deletions. The rows below are the raw changelog, not a materialized table." — but only to stderr. For -o json/-o yaml, that's the only signal; the row data itself carries no operation marker either way (the gap flagged earlier), so a script consuming the JSON has no way to know it just received a changelog instead of a table, and no way to tell an insert from a delete.

Put together: --property sql.snapshot.mode=<something else> (or any other bounded-but-not-append-only path) plus -o json gets you a result that looks like clean row data but is silently ambiguous — and the one warning that would tell a human something's off never reaches a script's stdout.

Given the command's own docs frame it as being for "scripting and one-shot queries," I'd treat this as something to close off now rather than after ships: either make sql.snapshot.mode non-overridable (reject the --property override, or reject/error when the resolved statement isn't append-only), or thread appendOnlyKnown && !isAppendOnly into the serialized envelope itself (a top-level "append_only": false field, say) so a script has a way to detect the case even without per-row operation markers.

Comment generated with the help of an AI agent