From 3152573c39479b25ccd9017bc4baf88c279a42f6 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Wed, 7 Oct 2026 14:02:02 +0000 Subject: [PATCH] feat(admin): make the Trino queries view readable at pool scale - Count every pre-execution state (QUEUED, WAITING_FOR_RESOURCES, DISPATCHING, PLANNING, STARTING) as waiting. Only QUEUED was counted, so a saturated cell reported 0 queued. - Lead each row with the statement verb and target table, with leading client comments (SQLMesh plan markers) stripped. Rows expand to the full SQL and execution detail (drivers, rows, network, spill, tag). - Per-org and per-pool-instance breakdowns; click one to filter. - State-bucket filter chips, sortable columns, a progress bar, and search over SQL text and query id as well as org and principal. - The pool observer stamps each query (list and detail) with the instance whose coordinator holds it. - The cell picker defaults to the first registered cell when the URL names none, instead of rendering an empty page. An explicit cell in the URL is never replaced. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Nic9bR3LPTRLk56ntJ4M5Z --- controlplane/admin/trino_client.go | 4 + .../src/components/TrinoCellPicker.test.tsx | 25 +- .../ui/src/components/TrinoCellPicker.tsx | 24 +- controlplane/admin/ui/src/lib/trino.test.ts | 95 +++- controlplane/admin/ui/src/lib/trino.ts | 192 +++++++- .../admin/ui/src/pages/TrinoQueries.test.tsx | 117 ++++- .../admin/ui/src/pages/TrinoQueries.tsx | 449 +++++++++++++++--- controlplane/admin/ui/src/types/api.ts | 3 + controlplane/trino_pool_observer.go | 6 + controlplane/trino_pool_observer_test.go | 26 + tests/mw-dev/e2e/trino.sh | 9 + 11 files changed, 855 insertions(+), 95 deletions(-) diff --git a/controlplane/admin/trino_client.go b/controlplane/admin/trino_client.go index d6b8c72d5..f17dde404 100644 --- a/controlplane/admin/trino_client.go +++ b/controlplane/admin/trino_client.go @@ -104,6 +104,10 @@ type TrinoQuery struct { ResourceGroup string `json:"resource_group"` // Query is the redacted SQL text. Query string `json:"query"` + // Instance is the shared-pool instance whose coordinator runs the query, + // set by the pool observer. Empty for a cell with a single coordinator. + // It is what shows one instance saturated while the others are idle. + Instance string `json:"instance,omitempty"` Created time.Time `json:"created,omitempty"` ElapsedMS int64 `json:"elapsed_ms"` diff --git a/controlplane/admin/ui/src/components/TrinoCellPicker.test.tsx b/controlplane/admin/ui/src/components/TrinoCellPicker.test.tsx index da395b4e8..97958749d 100644 --- a/controlplane/admin/ui/src/components/TrinoCellPicker.test.tsx +++ b/controlplane/admin/ui/src/components/TrinoCellPicker.test.tsx @@ -11,17 +11,28 @@ function mount(path = "/trino/queries?active=true") { render(); } beforeEach(() => hooks.useTrinoCells.mockReturnValue({ data: { cells: [{ id: "cell-test" }] } })); -it("requires explicit selection even when only one registered cell exists", () => { +it("defaults to the only registered cell when the URL names none", () => { mount(); - expect(screen.getByLabelText("Trino cell")).toHaveValue(""); - expect(screen.getByLabelText("Location")).toHaveTextContent("?active=true"); - fireEvent.change(screen.getByLabelText("Trino cell"), { target: { value: "cell-test" } }); + expect(screen.getByLabelText("Trino cell")).toHaveValue("cell-test"); expect(screen.getByLabelText("Location")).toHaveTextContent("active=true&cell=cell-test"); }); -it("does not infer a default from a cell name", () => { - hooks.useTrinoCells.mockReturnValue({ data: { cells: [{ id: "legacy" }, { id: "cell-test" }] } }); +it("defaults to the first registered cell when several exist", () => { + hooks.useTrinoCells.mockReturnValue({ data: { cells: [{ id: "cell-001" }, { id: "cell-002" }] } }); mount(); - expect(screen.getByLabelText("Trino cell")).toHaveValue(""); + expect(screen.getByLabelText("Location")).toHaveTextContent("cell=cell-001"); + fireEvent.change(screen.getByLabelText("Trino cell"), { target: { value: "cell-002" } }); + expect(screen.getByLabelText("Location")).toHaveTextContent("cell=cell-002"); +}); +it("keeps an explicit cell", () => { + hooks.useTrinoCells.mockReturnValue({ data: { cells: [{ id: "cell-001" }, { id: "cell-002" }] } }); + mount("/trino/queries?cell=cell-002"); + expect(screen.getByLabelText("Trino cell")).toHaveValue("cell-002"); +}); +it("does not navigate before the cell list loads", () => { + hooks.useTrinoCells.mockReturnValue({ data: undefined }); + mount(); + expect(screen.getByLabelText("Location")).toHaveTextContent("?active=true"); + expect(screen.getByLabelText("Location")).not.toHaveTextContent("cell="); }); it("does not silently replace an unknown explicit cell", () => { mount("/trino/queries?cell=removed"); diff --git a/controlplane/admin/ui/src/components/TrinoCellPicker.tsx b/controlplane/admin/ui/src/components/TrinoCellPicker.tsx index a45a00d2e..8143c0d93 100644 --- a/controlplane/admin/ui/src/components/TrinoCellPicker.tsx +++ b/controlplane/admin/ui/src/components/TrinoCellPicker.tsx @@ -1,3 +1,4 @@ +import { useEffect } from "react"; import { useSearchParams } from "react-router-dom"; import { useTrinoCells } from "@/hooks/useApi"; @@ -6,19 +7,38 @@ export function TrinoCellPicker() { const [params, setParams] = useSearchParams(); const entries = cells.data?.cells ?? []; const selected = params.get("cell") ?? ""; + const fallback = entries[0]?.id ?? ""; + + // Landing on a Trino page with no cell used to render an empty view until + // the operator picked one. Default to the first registered cell instead. + // This only fills an ABSENT parameter: an explicit cell in the URL, even + // one the server no longer lists, is never replaced, so a shared link keeps + // pointing where it was sent. replace keeps the defaulting out of history. + useEffect(() => { + if (selected || !fallback) return; + setParams( + (previous) => { + if (previous.get("cell")) return previous; + const next = new URLSearchParams(previous); + next.set("cell", fallback); + return next; + }, + { replace: true }, + ); + }, [selected, fallback, setParams]); + if (!entries.length) return null; return ( diff --git a/controlplane/admin/ui/src/lib/trino.test.ts b/controlplane/admin/ui/src/lib/trino.test.ts index 32535b1cd..acbee359c 100644 --- a/controlplane/admin/ui/src/lib/trino.test.ts +++ b/controlplane/admin/ui/src/lib/trino.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from "vitest"; import { + groupTrinoQueries, + isWaitingTrinoQuery, + sortTrinoQueries, + trinoSqlSummary, + trinoStateBucket, TRINO_DEGRADED_FAILURE_RATIO, TRINO_LONG_RUNNING_MS, isActiveTrinoQuery, @@ -330,7 +335,8 @@ describe("trinoStateVariant", () => { }); it("falls back rather than throwing on a state it does not know", () => { - expect(trinoStateVariant("WAITING_FOR_RESOURCES")).toBe("outline"); + expect(trinoStateVariant("SOME_FUTURE_STATE")).toBe("outline"); + expect(trinoStateVariant("WAITING_FOR_RESOURCES")).toBe("warning"); }); }); @@ -433,3 +439,90 @@ describe("summarizeTrinoNodes from system.runtime.nodes", () => { expect(h.degraded).toBe(0); }); }); + +describe("waiting states", () => { + it("counts every pre-execution state as queued, not only QUEUED", () => { + const states = ["QUEUED", "WAITING_FOR_RESOURCES", "DISPATCHING", "PLANNING", "STARTING", "RUNNING", "FINISHING"]; + const s = summarizeTrinoQueries(states.map((state, i) => query({ query_id: `q${i}`, state }))); + expect(s.queued).toBe(5); + expect(s.running).toBe(2); + }); + it("flags a query waiting for resources", () => { + expect(isWaitingTrinoQuery(query({ state: "WAITING_FOR_RESOURCES" }))).toBe(true); + expect(trinoQueryFlag(query({ state: "WAITING_FOR_RESOURCES" }))).toBe("queued"); + expect(trinoStateBucket(query({ state: "PLANNING" }))).toBe("waiting"); + expect(trinoStateBucket(query({ state: "FINISHING" }))).toBe("running"); + expect(trinoStateBucket(query({ state: "FAILED" }))).toBe("failed"); + }); +}); + +describe("trinoSqlSummary", () => { + it("strips the SQLMesh plan comment and names the target", () => { + const s = trinoSqlSummary( + '/* SQLMESH_PLAN: 4b1d */ INSERT INTO "org_portola"."sqlmesh_trino"."events__123" ("a") SELECT 1', + ); + expect(s).toMatchObject({ + verb: "INSERT INTO", + target: "org_portola.sqlmesh_trino.events__123", + tag: "SQLMESH_PLAN: 4b1d", + }); + expect(s.text.startsWith("INSERT INTO")).toBe(true); + }); + it("reports DDL kind and skips IF NOT EXISTS", () => { + expect(trinoSqlSummary("CREATE TABLE IF NOT EXISTS a.b AS SELECT 1")).toMatchObject({ + verb: "CREATE TABLE", + target: "a.b", + }); + expect(trinoSqlSummary("create or replace view x.y as select 1")).toMatchObject({ + verb: "CREATE VIEW", + target: "x.y", + }); + expect(trinoSqlSummary("DROP TABLE IF EXISTS s.t")).toMatchObject({ verb: "DROP TABLE", target: "s.t" }); + }); + it("treats WITH as a read and finds its first relation", () => { + expect(trinoSqlSummary("-- note\nWITH c AS (SELECT * FROM s.events) SELECT * FROM c")).toMatchObject({ + verb: "SELECT", + target: "s.events", + tag: "note", + }); + }); + it("handles empty and unknown text", () => { + expect(trinoSqlSummary("")).toMatchObject({ verb: "", target: "", tag: "" }); + expect(trinoSqlSummary("SHOW CATALOGS")).toMatchObject({ verb: "SHOW", target: "" }); + }); +}); + +describe("groupTrinoQueries", () => { + it("rolls up per key, busiest first", () => { + const groups = groupTrinoQueries( + [ + query({ query_id: "1", org: "a", cpu_ms: 10 }), + query({ query_id: "2", org: "b", state: "QUEUED" }), + query({ query_id: "3", org: "b", cpu_ms: 5, physical_input_bytes: 7 }), + query({ query_id: "4", org: "b", state: "FINISHED", elapsed_ms: 99_999 }), + ], + (q) => q.org, + ); + expect(groups.map((g) => g.key)).toEqual(["b", "a"]); + expect(groups[0]).toMatchObject({ total: 3, running: 1, queued: 1, cpuMs: 5, scannedBytes: 7, longestMs: 1_000 }); + }); +}); + +describe("sortTrinoQueries", () => { + const rows = [ + query({ query_id: "a", elapsed_ms: 5, cpu_ms: 3, state: "QUEUED" }), + query({ query_id: "b", elapsed_ms: 9, cpu_ms: 1 }), + query({ query_id: "c", elapsed_ms: 1, cpu_ms: 2, progress_percentage: 50 }), + ]; + it("sorts by a numeric key in both directions without mutating", () => { + expect(sortTrinoQueries(rows, "cpu", "desc").map((q) => q.query_id)).toEqual(["a", "c", "b"]); + expect(sortTrinoQueries(rows, "cpu", "asc").map((q) => q.query_id)).toEqual(["b", "c", "a"]); + expect(rows.map((q) => q.query_id)).toEqual(["a", "b", "c"]); + }); + it("puts running before waiting and breaks ties by elapsed", () => { + expect(sortTrinoQueries(rows, "state", "asc").map((q) => q.query_id)).toEqual(["b", "c", "a"]); + }); + it("sorts unknown progress below zero", () => { + expect(sortTrinoQueries(rows, "progress", "desc")[0].query_id).toBe("c"); + }); +}); diff --git a/controlplane/admin/ui/src/lib/trino.ts b/controlplane/admin/ui/src/lib/trino.ts index f024d6949..0ce96f54f 100644 --- a/controlplane/admin/ui/src/lib/trino.ts +++ b/controlplane/admin/ui/src/lib/trino.ts @@ -24,10 +24,29 @@ export function isActiveTrinoQuery(q: TrinoQuery): boolean { return !(TRINO_TERMINAL_STATES as readonly string[]).includes(q.state); } +// States in which a query holds a slot in the list but is not yet executing. +// QUEUED is only the resource-group queue; a query then passes through +// WAITING_FOR_RESOURCES (cluster minimum workers / memory), DISPATCHING, +// PLANNING and STARTING before it runs. Counting only QUEUED reported "0 +// queued" while dozens of queries sat in WAITING_FOR_RESOURCES. +export const TRINO_WAITING_STATES = [ + "QUEUED", + "WAITING_FOR_RESOURCES", + "DISPATCHING", + "PLANNING", + "STARTING", +] as const; + +export function isWaitingTrinoQuery(q: Pick): boolean { + return (TRINO_WAITING_STATES as readonly string[]).includes(q.state.toUpperCase()); +} + export function trinoStateVariant(state: string): BadgeProps["variant"] { switch (state.toUpperCase()) { case "RUNNING": + case "FINISHING": return "success"; + case "WAITING_FOR_RESOURCES": case "QUEUED": // Queued is not an error, but it IS the state an operator is looking // for when a tenant reports slowness — resource-group saturation. @@ -59,7 +78,7 @@ export function trinoQueryFlag(q: TrinoQuery): TrinoQueryFlag | null { // waiting on the metadata store or on S3, which is a cell-level problem // wearing one query's clothes. if (q.state === "RUNNING" && q.fully_blocked) return "blocked"; - if (q.state === "QUEUED") return "queued"; + if (isWaitingTrinoQuery(q)) return "queued"; if (isActiveTrinoQuery(q) && q.elapsed_ms >= TRINO_LONG_RUNNING_MS) return "long_running"; return null; } @@ -67,6 +86,7 @@ export function trinoQueryFlag(q: TrinoQuery): TrinoQueryFlag | null { export interface TrinoQuerySummary { total: number; running: number; + // queued counts every waiting state (TRINO_WAITING_STATES), not only QUEUED. queued: number; blocked: number; failed: number; @@ -90,8 +110,8 @@ export function summarizeTrinoQueries(queries: TrinoQuery[]): TrinoQuerySummary longestMs: 0, }; for (const q of queries) { - if (q.state === "RUNNING") s.running += 1; - if (q.state === "QUEUED") s.queued += 1; + if (q.state === "RUNNING" || q.state === "FINISHING") s.running += 1; + if (isWaitingTrinoQuery(q)) s.queued += 1; if (q.state === "FAILED") s.failed += 1; if (q.state === "RUNNING" && q.fully_blocked) s.blocked += 1; s.scannedBytes += q.physical_input_bytes ?? 0; @@ -229,3 +249,169 @@ export function trinoScanEfficiency(q: TrinoQuery): number | null { if (!q.processed_input_rows || q.processed_input_rows <= 0) return null; return q.physical_input_bytes / q.processed_input_rows; } + +// ---- Making a list of hundreds of SQLMesh statements readable ---- + +export interface TrinoSqlSummary { + // verb is the leading statement keyword(s): SELECT, INSERT INTO, CREATE + // TABLE, MERGE INTO, DROP TABLE, ... ("" when the text is empty). + verb: string; + // target is the relation the statement writes (or, for a SELECT, the first + // relation it reads). "" when none could be found. + target: string; + // tag is the client's leading comment when it is a short label, e.g. the + // SQLMesh `/* SQLMESH_PLAN: ... */` marker. "" when there was none. + tag: string; + // text is the statement with leading comments and whitespace removed. + text: string; +} + +const LEADING_COMMENT = /^\s*(?:\/\*([\s\S]*?)\*\/|--([^\n]*)(?:\n|$))/; +const IDENT = String.raw`(?:"(?:[^"]|"")+"|[A-Za-z_][\w$]*)`; +const QUALIFIED = String.raw`${IDENT}(?:\s*\.\s*${IDENT}){0,2}`; +const WRITE_PATTERNS: Array<[RegExp, string]> = [ + [new RegExp(String.raw`^insert\s+(?:overwrite\s+)?into\s+(${QUALIFIED})`, "i"), "INSERT INTO"], + [new RegExp(String.raw`^merge\s+into\s+(${QUALIFIED})`, "i"), "MERGE INTO"], + [new RegExp(String.raw`^delete\s+from\s+(${QUALIFIED})`, "i"), "DELETE FROM"], + [new RegExp(String.raw`^update\s+(${QUALIFIED})`, "i"), "UPDATE"], + [new RegExp(String.raw`^create\s+(?:or\s+replace\s+)?(materialized\s+view|view|table|schema)\s+(?:if\s+not\s+exists\s+)?(${QUALIFIED})`, "i"), "CREATE"], + [new RegExp(String.raw`^drop\s+(materialized\s+view|view|table|schema)\s+(?:if\s+exists\s+)?(${QUALIFIED})`, "i"), "DROP"], + [new RegExp(String.raw`^alter\s+(table|view|schema)\s+(?:if\s+exists\s+)?(${QUALIFIED})`, "i"), "ALTER"], +]; +const FIRST_FROM = new RegExp(String.raw`\bfrom\s+(${QUALIFIED})`, "i"); + +function unquote(name: string): string { + return (name.match(new RegExp(IDENT, "g")) ?? []) + .map((part) => (part.startsWith('"') ? part.slice(1, -1).replace(/""/g, '"') : part)) + .join("."); +} + +export function trinoSqlSummary(sql: string): TrinoSqlSummary { + let text = sql ?? ""; + let tag = ""; + for (let m = LEADING_COMMENT.exec(text); m; m = LEADING_COMMENT.exec(text)) { + const body = (m[1] ?? m[2] ?? "").trim().replace(/\s+/g, " "); + if (!tag && body && body.length <= 120) tag = body; + text = text.slice(m[0].length); + } + text = text.trim(); + for (const [pattern, verb] of WRITE_PATTERNS) { + const m = pattern.exec(text); + if (!m) continue; + if (verb === "CREATE" || verb === "DROP" || verb === "ALTER") { + return { verb: `${verb} ${m[1].toUpperCase().replace(/\s+/g, " ")}`, target: unquote(m[2]), tag, text }; + } + return { verb, target: unquote(m[1]), tag, text }; + } + const word = /^[A-Za-z]+/.exec(text)?.[0]?.toUpperCase() ?? ""; + // A WITH ... SELECT is a read; report it as one rather than as "WITH". + const verb = word === "WITH" ? "SELECT" : word; + const from = verb === "SELECT" ? FIRST_FROM.exec(text) : null; + return { verb, target: from ? unquote(from[1]) : "", tag, text }; +} + +export interface TrinoQueryGroup { + key: string; + total: number; + running: number; + queued: number; + blocked: number; + cpuMs: number; + scannedBytes: number; + peakMemoryBytes: number; + longestMs: number; +} + +// Group the listed queries by a key (org, instance, resource group), largest +// first. This is what answers "who is using the cell right now" without +// reading every row. +export function groupTrinoQueries( + queries: TrinoQuery[], + keyOf: (q: TrinoQuery) => string, +): TrinoQueryGroup[] { + const groups = new Map(); + for (const q of queries) { + const key = keyOf(q); + let g = groups.get(key); + if (!g) { + g = { key, total: 0, running: 0, queued: 0, blocked: 0, cpuMs: 0, scannedBytes: 0, peakMemoryBytes: 0, longestMs: 0 }; + groups.set(key, g); + } + g.total += 1; + if (q.state === "RUNNING" || q.state === "FINISHING") g.running += 1; + if (isWaitingTrinoQuery(q)) g.queued += 1; + if (q.state === "RUNNING" && q.fully_blocked) g.blocked += 1; + g.cpuMs += q.cpu_ms ?? 0; + g.scannedBytes += q.physical_input_bytes ?? 0; + g.peakMemoryBytes += q.peak_memory_bytes ?? 0; + if (isActiveTrinoQuery(q)) g.longestMs = Math.max(g.longestMs, q.elapsed_ms ?? 0); + } + return [...groups.values()].sort( + (a, b) => b.running + b.queued - (a.running + a.queued) || b.total - a.total || a.key.localeCompare(b.key), + ); +} + +// Coarse lifecycle bucket for the state filter: what an operator asks for is +// "show me what is waiting", not one of Trino's nine states. +export type TrinoStateBucket = "running" | "waiting" | "finished" | "failed"; + +export function trinoStateBucket(q: Pick): TrinoStateBucket { + const state = q.state.toUpperCase(); + if (state === "FAILED") return "failed"; + if (state === "FINISHED") return "finished"; + if (isWaitingTrinoQuery(q)) return "waiting"; + return "running"; +} + +export type TrinoQuerySortKey = + | "elapsed" + | "cpu" + | "scanned" + | "memory" + | "progress" + | "state" + | "org"; + +// Running before waiting before finished, so the default "state" order puts +// live work on top. +const BUCKET_ORDER: Record = { running: 0, waiting: 1, failed: 2, finished: 3 }; + +function sortValue(q: TrinoQuery, key: TrinoQuerySortKey): number | string { + switch (key) { + case "elapsed": + return q.elapsed_ms ?? 0; + case "cpu": + return q.cpu_ms ?? 0; + case "scanned": + return q.physical_input_bytes ?? 0; + case "memory": + return q.peak_memory_bytes ?? 0; + case "progress": + // Unknown progress sorts below 0%. + return q.progress_percentage ?? -1; + case "state": + return BUCKET_ORDER[trinoStateBucket(q)]; + case "org": + return q.org || q.principal || ""; + } +} + +// Stable sort; ties fall back to elapsed time, longest first, then query id +// so rows do not jump around between 2s polls. +export function sortTrinoQueries( + queries: TrinoQuery[], + key: TrinoQuerySortKey, + direction: "asc" | "desc", +): TrinoQuery[] { + const sign = direction === "asc" ? 1 : -1; + return [...queries].sort((a, b) => { + const va = sortValue(a, key); + const vb = sortValue(b, key); + const primary = + typeof va === "string" || typeof vb === "string" + ? String(va).localeCompare(String(vb)) + : (va as number) - (vb as number); + if (primary !== 0) return sign * primary; + return (b.elapsed_ms ?? 0) - (a.elapsed_ms ?? 0) || a.query_id.localeCompare(b.query_id); + }); +} diff --git a/controlplane/admin/ui/src/pages/TrinoQueries.test.tsx b/controlplane/admin/ui/src/pages/TrinoQueries.test.tsx index b2aad5052..0bfc5f6e2 100644 --- a/controlplane/admin/ui/src/pages/TrinoQueries.test.tsx +++ b/controlplane/admin/ui/src/pages/TrinoQueries.test.tsx @@ -1,5 +1,5 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; -import { render, screen, within } from "@testing-library/react"; +import { fireEvent, render, screen, within } from "@testing-library/react"; import { MemoryRouter } from "react-router-dom"; import { TooltipProvider } from "@/components/ui/tooltip"; import type { TrinoQuery, TrinoStatus } from "@/types/api"; @@ -103,13 +103,16 @@ describe("TrinoQueries page", () => { query({ query_id: "a", state: "RUNNING", elapsed_ms: 30_000, physical_input_bytes: 1024 }), query({ query_id: "b", state: "RUNNING", fully_blocked: true, elapsed_ms: 90_000 }), query({ query_id: "c", state: "QUEUED", elapsed_ms: 5_000 }), + // Past the resource-group queue but not yet executing: this is the + // state a saturated cell parks work in, and it used to count as 0. + query({ query_id: "d", state: "WAITING_FOR_RESOURCES", elapsed_ms: 5_000 }), ], }), ); renderPage(); expect(statValue("Running")).toBe("2"); - expect(statValue("Queued")).toBe("1"); + expect(statValue("Waiting")).toBe("2"); // Blocked is counted separately from running: it means every driver is // waiting on the metadata store or S3, which is a cell problem. expect(statValue("Blocked")).toBe("1"); @@ -160,7 +163,7 @@ describe("TrinoQueries page", () => { }), ); renderPage(); - expect(screen.getByText("duckgres-provisioner")).toBeInTheDocument(); + expect(within(screen.getByTestId("query-q1")).getByText("duckgres-provisioner")).toBeInTheDocument(); }); it("explains an unconfigured deployment differently from an outage", () => { @@ -203,4 +206,112 @@ describe("TrinoQueries page", () => { renderPage(); expect(screen.getByRole("button", { name: "Kill" })).toBeDisabled(); }); + + function withQueries(queries: TrinoQuery[]) { + hooks.useTrinoQueries.mockReturnValue(ok({ cell: status().cell, available: true, queries })); + } + const rowIds = () => + screen.getAllByTestId(/^query-/).map((row) => row.getAttribute("data-testid")!.slice("query-".length)); + + it("leads each row with the statement and its target, not the client's comment", () => { + withQueries([ + query({ query: '/* SQLMESH_PLAN: abc */ INSERT INTO "org_p"."sqlmesh_trino"."events__1" SELECT 1' }), + ]); + renderPage(); + const row = screen.getByTestId("query-q1"); + expect(within(row).getByText("INSERT INTO")).toBeInTheDocument(); + expect(within(row).getByText("org_p.sqlmesh_trino.events__1")).toBeInTheDocument(); + }); + + it("expands a row to the full SQL and its execution detail", () => { + withQueries([query({ query: "SELECT * FROM events", total_drivers: 8, completed_drivers: 3 })]); + renderPage(); + expect(screen.queryByText("Query id")).not.toBeInTheDocument(); + fireEvent.click(screen.getByTestId("query-q1")); + expect(screen.getByText("Query id")).toBeInTheDocument(); + expect(screen.getByText(/3\/8 done/)).toBeInTheDocument(); + fireEvent.click(screen.getByTestId("query-q1")); + expect(screen.queryByText("Query id")).not.toBeInTheDocument(); + }); + + it("sorts by elapsed by default and re-sorts by a clicked column", () => { + withQueries([ + query({ query_id: "short", elapsed_ms: 1_000, cpu_ms: 900 }), + query({ query_id: "long", elapsed_ms: 60_000, cpu_ms: 10 }), + ]); + renderPage(); + expect(rowIds()).toEqual(["long", "short"]); + fireEvent.click(screen.getByRole("button", { name: "CPU" })); + expect(rowIds()).toEqual(["short", "long"]); + fireEvent.click(screen.getByRole("button", { name: "CPU" })); + expect(rowIds()).toEqual(["long", "short"]); + }); + + it("filters by state bucket while the headline keeps describing the cell", () => { + withQueries([ + query({ query_id: "run", state: "RUNNING" }), + query({ query_id: "wait", state: "WAITING_FOR_RESOURCES" }), + query({ query_id: "plan", state: "PLANNING" }), + ]); + renderPage(); + const chips = screen.getByRole("group", { name: "State filter" }); + fireEvent.click(within(chips).getByRole("button", { name: /Waiting/ })); + expect(rowIds().sort()).toEqual(["plan", "wait"]); + expect(statValue("Running")).toBe("1"); + }); + + it("breaks load down by org and narrows the list to a clicked org", () => { + hooks.useOrgLabels.mockReturnValue(new Map([["org-a-id", "alpha"], ["org-b-id", "beta"]])); + withQueries([ + query({ query_id: "a1", org: "org-a-id" }), + query({ query_id: "b1", org: "org-b-id" }), + query({ query_id: "b2", org: "org-b-id", state: "QUEUED" }), + ]); + renderPage(); + // Busiest org first. + const groups = screen.getAllByTestId(/^group-/).map((g) => g.getAttribute("data-testid")); + expect(groups).toEqual(["group-org-b-id", "group-org-a-id"]); + fireEvent.click(screen.getByTestId("group-org-b-id")); + expect(rowIds().sort()).toEqual(["b1", "b2"]); + fireEvent.click(screen.getByRole("button", { name: /Clear beta/ })); + expect(rowIds()).toHaveLength(3); + }); + + it("shows which pool instance holds each query when the cell is pooled", () => { + withQueries([ + query({ query_id: "x", instance: "cell-0a1b2c3d" }), + query({ query_id: "y", instance: "cell-0a1b2c3d" }), + query({ query_id: "z", instance: "cell-ffee0011" }), + ]); + renderPage(); + expect(screen.getByText("By pool instance")).toBeInTheDocument(); + expect(within(screen.getByTestId("group-cell-0a1b2c3d")).getByText("2")).toBeInTheDocument(); + expect(within(screen.getByTestId("query-z")).getByText("cell-ffee0011")).toBeInTheDocument(); + }); + + it("omits the instance breakdown for a single-coordinator cell", () => { + withQueries([query()]); + renderPage(); + expect(screen.queryByText("By pool instance")).not.toBeInTheDocument(); + }); + + it("searches SQL text and query ids, not only orgs", () => { + withQueries([ + query({ query_id: "20261007_a", query: "SELECT * FROM persons" }), + query({ query_id: "20261007_b", query: "SELECT * FROM events" }), + ]); + renderPage(); + const search = screen.getByPlaceholderText(/Search org/); + fireEvent.change(search, { target: { value: "persons" } }); + expect(rowIds()).toEqual(["20261007_a"]); + fireEvent.change(search, { target: { value: "_b" } }); + expect(rowIds()).toEqual(["20261007_b"]); + }); + + it("does not expand the row when Kill is clicked", () => { + withQueries([query()]); + renderPage(); + fireEvent.click(screen.getByRole("button", { name: "Kill" })); + expect(screen.queryByText("Query id")).not.toBeInTheDocument(); + }); }); diff --git a/controlplane/admin/ui/src/pages/TrinoQueries.tsx b/controlplane/admin/ui/src/pages/TrinoQueries.tsx index f4c13cef7..caab053f1 100644 --- a/controlplane/admin/ui/src/pages/TrinoQueries.tsx +++ b/controlplane/admin/ui/src/pages/TrinoQueries.tsx @@ -1,6 +1,6 @@ -import { useMemo, useState } from "react"; +import { Fragment, useMemo, useState, type ReactNode } from "react"; import { useSearchParams } from "react-router-dom"; -import { AlertTriangle, Ban, Database, Gauge, Hourglass, Timer } from "lucide-react"; +import { AlertTriangle, ArrowDown, ArrowUp, Ban, ChevronDown, ChevronRight, Database, Gauge, Hourglass, Timer } from "lucide-react"; import { PageBody, PageHeader } from "@/components/AppShell"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { Badge } from "@/components/ui/badge"; @@ -23,15 +23,23 @@ import { import { useKillTrinoQuery, useOrgLabels, useTrinoQueries, useTrinoStatus } from "@/hooks/useApi"; import { fmtBytes, fmtCompact, fmtDurationMs } from "@/lib/format"; import { + groupTrinoQueries, isActiveTrinoQuery, + sortTrinoQueries, summarizeTrinoQueries, + trinoSqlSummary, + trinoStateBucket, trinoQueryFlag, trinoScanEfficiency, trinoStateVariant, trinoUnavailableMessage, trinoUnavailableReason, type TrinoQueryFlag, + type TrinoQueryGroup, + type TrinoQuerySortKey, + type TrinoStateBucket, } from "@/lib/trino"; +import { cn } from "@/lib/utils"; import type { TrinoQuery } from "@/types/api"; const FLAG_LABEL: Record = { @@ -110,12 +118,147 @@ function KillDialog({ ); } +// SortHeader is a clickable column header. Clicking the active column flips +// direction; clicking another column starts descending, because "biggest +// first" is the question every numeric column here answers. +function SortHeader({ + label, + column, + sort, + onSort, + align = "left", +}: { + label: string; + column: TrinoQuerySortKey; + sort: { key: TrinoQuerySortKey; dir: "asc" | "desc" }; + onSort: (key: TrinoQuerySortKey) => void; + align?: "left" | "right"; +}) { + const active = sort.key === column; + return ( + + + + ); +} + +function ProgressBar({ value }: { value: number | null }) { + if (value === null) return —; + const pct = Math.max(0, Math.min(100, value)); + return ( +
+
+
+
+ {pct.toFixed(0)}% +
+ ); +} + +function Detail({ label, children }: { label: string; children: ReactNode }) { + return ( +
+
{label}
+
{children}
+
+ ); +} + +// Breakdown answers "who is using the cell right now" in one glance, and +// clicking a row narrows the list to it. +function Breakdown({ + title, + groups, + label, + selected, + onSelect, +}: { + title: string; + groups: TrinoQueryGroup[]; + label: (key: string) => ReactNode; + selected: string | null; + onSelect: (key: string | null) => void; +}) { + return ( + + + {title} + + + + + + + Running + Waiting + CPU + Scanned + Peak mem + Longest + + + + {groups.map((g) => ( + onSelect(selected === g.key ? null : g.key)} + > + {label(g.key)} + + {g.running} + {g.blocked > 0 && ({g.blocked} blocked)} + + 0 && "text-warning")}>{g.queued} + {fmtDurationMs(g.cpuMs)} + {fmtBytes(g.scannedBytes)} + {fmtBytes(g.peakMemoryBytes)} + {fmtDurationMs(g.longestMs)} + + ))} + +
+ + + ); +} + +const BUCKETS: Array<{ key: TrinoStateBucket | "all"; label: string; activeOnly?: boolean }> = [ + { key: "all", label: "All" }, + { key: "running", label: "Running" }, + { key: "waiting", label: "Waiting" }, + { key: "failed", label: "Failed", activeOnly: false }, + { key: "finished", label: "Finished", activeOnly: false }, +]; + +// orgKey groups a query by tenant; a control-plane query has no org and is +// grouped by its X-Trino-Source instead, as the Org column shows it. +const orgKey = (q: TrinoQuery) => q.org || q.source || q.principal || "—"; + export function TrinoQueries() { // active=true is the live view: the states an operator can still act on. // Unchecking it brings in recently-finished queries, which is the closest // thing to history the coordinator holds in memory. const [activeOnly, setActiveOnly] = useState(true); - const [orgFilter, setOrgFilter] = useState(""); + const [search, setSearch] = useState(""); + const [bucket, setBucket] = useState("all"); + const [orgPick, setOrgPick] = useState(null); + const [instancePick, setInstancePick] = useState(null); + const [sort, setSort] = useState<{ key: TrinoQuerySortKey; dir: "asc" | "desc" }>({ key: "elapsed", dir: "desc" }); + const [expanded, setExpanded] = useState>(() => new Set()); const [killing, setKilling] = useState(null); const [params] = useSearchParams(); @@ -124,20 +267,61 @@ export function TrinoQueries() { const queries = useTrinoQueries({ active: activeOnly, ...(cell ? { cell } : {}) }); const orgLabels = useOrgLabels(); - const rows = useMemo(() => { - const all = queries.data?.queries ?? []; - if (orgFilter === "") return all; - const needle = orgFilter.toLowerCase(); + const all = useMemo(() => queries.data?.queries ?? [], [queries.data]); + const orgName = (key: string) => orgLabels.get(key) ?? key; + + // Text search covers who (org label/id, principal) and what (SQL, query id), + // so "portola", "events__" and "20261007_" all find what an operator means. + const searched = useMemo(() => { + if (search === "") return all; + const needle = search.toLowerCase(); return all.filter( (q) => q.org.toLowerCase().includes(needle) || q.principal.toLowerCase().includes(needle) || - (orgLabels.get(q.org) ?? "").toLowerCase().includes(needle), + (orgLabels.get(q.org) ?? "").toLowerCase().includes(needle) || + q.query_id.toLowerCase().includes(needle) || + q.query.toLowerCase().includes(needle), ); - }, [queries.data, orgFilter, orgLabels]); + }, [all, search, orgLabels]); + + // The headline numbers and breakdowns follow the search, not the state + // chips or a breakdown pick — those narrow the list below, and summary + // cards that changed with them would stop describing the cell. + const summary = useMemo(() => summarizeTrinoQueries(searched), [searched]); + const byOrg = useMemo(() => groupTrinoQueries(searched, orgKey), [searched]); + const hasInstances = searched.some((q) => q.instance); + const byInstance = useMemo( + () => (hasInstances ? groupTrinoQueries(searched, (q) => q.instance || "—") : []), + [searched, hasInstances], + ); + const bucketCounts = useMemo(() => { + const counts: Record = { running: 0, waiting: 0, failed: 0, finished: 0 }; + for (const q of searched) counts[trinoStateBucket(q)] += 1; + return counts; + }, [searched]); + + const rows = useMemo(() => { + const filtered = searched.filter( + (q) => + (bucket === "all" || trinoStateBucket(q) === bucket) && + (orgPick === null || orgKey(q) === orgPick) && + (instancePick === null || (q.instance || "—") === instancePick), + ); + return sortTrinoQueries(filtered, sort.key, sort.dir); + }, [searched, bucket, orgPick, instancePick, sort]); - const summary = useMemo(() => summarizeTrinoQueries(rows), [rows]); const reason = trinoUnavailableReason(status.data); + const onSort = (key: TrinoQuerySortKey) => + setSort((prev) => (prev.key === key ? { key, dir: prev.dir === "asc" ? "desc" : "asc" } : { key, dir: key === "org" ? "asc" : "desc" })); + const toggle = (id: string) => + setExpanded((prev) => { + const next = new Set(prev); + if (next.has(id)) next.delete(id); + else next.add(id); + return next; + }); + const colCount = hasInstances ? 10 : 9; return ( <> @@ -152,15 +336,18 @@ export function TrinoQueries() {
setOrgFilter(e.target.value)} + className="h-8 w-64" + placeholder="Search org, principal, SQL or query id…" + value={search} + onChange={(e) => setSearch(e.target.value)} /> @@ -180,10 +367,10 @@ export function TrinoQueries() {
} /> 0 ? "warning" : "default"} - hint="waiting on a resource group" + hint="queued, waiting for resources, planning or starting" icon={} /> } />
+ {byOrg.length > 0 && ( +
+ (all.some((q) => q.org === key) ? : {key})} + selected={orgPick} + onSelect={setOrgPick} + /> + {hasInstances && ( + {key}} + selected={instancePick} + onSelect={setInstancePick} + /> + )} +
+ )} + - + {activeOnly ? "Active queries" : "Queries held by the coordinator"} +
+ {BUCKETS.filter((b) => !(activeOnly && b.activeOnly === false)).map((b) => ( + + ))} + {(orgPick !== null || instancePick !== null) && ( + + )} +
{queries.isLoading ? ( - + ) : queries.isError ? ( void queries.refetch()} /> ) : rows.length === 0 ? ( @@ -225,13 +462,16 @@ export function TrinoQueries() { - State - Org + + + Query - Elapsed - CPU - Scanned - Peak mem + {hasInstances && Instance} + + + + + @@ -239,62 +479,113 @@ export function TrinoQueries() { {rows.map((q) => { const flag = trinoQueryFlag(q); const perRow = trinoScanEfficiency(q); + const sql = trinoSqlSummary(q.query); + const open = expanded.has(q.query_id); return ( - - - {q.state} - {flag && ( - - {FLAG_LABEL[flag]} - - )} - - - {q.org ? ( - - ) : ( - // No org means a control-plane principal: the - // reconcile loop's DDL or this console's own - // reads, both tagged by X-Trino-Source. - - {q.source || q.principal || "—"} + + toggle(q.query_id)} + > + + {open ? : } + + + {q.state} + {flag && ( + + {FLAG_LABEL[flag]} + + )} + + + {q.org ? ( + + ) : ( + // No org means a control-plane principal: the + // reconcile loop's DDL or this console's own + // reads, both tagged by X-Trino-Source. + + {q.source || q.principal || "—"} + + )} + {q.org && q.principal && ( + {q.principal} + )} + + + {sql.verb && ( + + {sql.verb} + {sql.target && {sql.target}} + + )} + + {sql.text || q.query} + + {hasInstances && ( + {q.instance || "—"} )} - - - - {q.query} - - - {q.resource_group || "unassigned"} - {q.progress_percentage !== null - ? ` · ${q.progress_percentage.toFixed(0)}%` - : ""} - {perRow !== null ? ` · ${fmtBytes(perRow)}/row` : ""} - - - - {fmtDurationMs(q.elapsed_ms)} - - - {fmtDurationMs(q.cpu_ms)} - - - {fmtBytes(q.physical_input_bytes)} - - - {fmtBytes(q.peak_memory_bytes)} - - - {isActiveTrinoQuery(q) && ( - - - - )} - - + + {fmtDurationMs(q.elapsed_ms)} + {q.queued_ms > 1000 && ( + {fmtDurationMs(q.queued_ms)} queued + )} + + + + + {fmtDurationMs(q.cpu_ms)} + + {fmtBytes(q.physical_input_bytes)} + {perRow !== null && ( + {fmtBytes(perRow)}/row + )} + + {fmtBytes(q.peak_memory_bytes)} + e.stopPropagation()}> + {isActiveTrinoQuery(q) && ( + + + + )} + + + {open && ( + + + +
+ {q.query_id} + {q.principal || "—"} + {q.source || "—"} + {q.resource_group || "unassigned"} + + {q.running_drivers} running · {q.queued_drivers} queued · {q.completed_drivers}/{q.total_drivers} done + + {fmtCompact(q.processed_input_rows)} + {fmtBytes(q.internal_network_bytes)} + {fmtBytes(q.spilled_bytes)} + {fmtDurationMs(q.queued_ms)} + {sql.tag && {sql.tag}} + {q.error_type && ( + + {q.error_type} + {q.error_code ? ` · ${q.error_code}` : ""} + + )} +
+ 
+                                {q.query}
+
+ + + )} + ); })} diff --git a/controlplane/admin/ui/src/types/api.ts b/controlplane/admin/ui/src/types/api.ts index 55411c9ce..4b16c08c7 100644 --- a/controlplane/admin/ui/src/types/api.ts +++ b/controlplane/admin/ui/src/types/api.ts @@ -774,6 +774,9 @@ export interface TrinoQuery { source: string; resource_group: string; query: string; + // instance is the shared-pool instance whose coordinator holds the query. + // Absent on a cell with a single coordinator. + instance?: string; created?: string; elapsed_ms: number; queued_ms: number; diff --git a/controlplane/trino_pool_observer.go b/controlplane/trino_pool_observer.go index 1bc94cf58..ec3fb67c7 100644 --- a/controlplane/trino_pool_observer.go +++ b/controlplane/trino_pool_observer.go @@ -139,6 +139,9 @@ func (o *trinoPoolObserver) Queries(ctx context.Context) ([]admin.TrinoQuery, er continue } succeeded++ + for i := range queries { + queries[i].Instance = member.instanceID + } all = append(all, queries...) } if succeeded == 0 { @@ -158,6 +161,9 @@ func (o *trinoPoolObserver) Query(ctx context.Context, queryID string) (*admin.T for _, member := range members { query, err := member.client.Query(ctx, queryID) if err == nil { + if query != nil { + query.Instance = member.instanceID + } return query, nil } lastErr = err diff --git a/controlplane/trino_pool_observer_test.go b/controlplane/trino_pool_observer_test.go index 929acfbc0..182241b81 100644 --- a/controlplane/trino_pool_observer_test.go +++ b/controlplane/trino_pool_observer_test.go @@ -115,6 +115,14 @@ func TestTrinoPoolObserverUnionsLiveMembers(t *testing.T) { if lister.poolIDs[0] != "registered:cell-001" { t.Fatalf("listed pool %q, want the stored pool id", lister.poolIDs[0]) } + // Each query names the instance whose coordinator holds it, so the + // console can show one instance saturated while the others idle. + for _, q := range queries { + want := map[string]string{"qa": "cell-a", "qb": "cell-b"}[q.QueryID] + if q.Instance != want { + t.Fatalf("query %s instance = %q, want %q", q.QueryID, q.Instance, want) + } + } } // One unreachable member must not hide the rest; only a pool where every @@ -175,6 +183,24 @@ func TestTrinoPoolObserverKillsOnOwningMember(t *testing.T) { } } +func TestTrinoPoolObserverQueryDetailNamesItsInstance(t *testing.T) { + members := map[string]*fakePoolMember{ + "cell-a": {queries: []admin.TrinoQuery{{QueryID: "qa"}}}, + "cell-b": {queries: []admin.TrinoQuery{{QueryID: "qb"}}}, + } + o, _, _ := poolObserverFixture([]configstore.TrinoPoolInstance{ + instance("cell-a", trinopool.PhaseServing), + instance("cell-b", trinopool.PhaseServing), + }, members) + q, err := o.Query(context.Background(), "qb") + if err != nil { + t.Fatalf("Query: %v", err) + } + if q.Instance != "cell-b" { + t.Fatalf("Query(qb).Instance = %q, want cell-b", q.Instance) + } +} + // A retired instance's client is dropped, so the cache cannot grow without // bound as the pool replaces members. func TestTrinoPoolObserverDropsRetiredClients(t *testing.T) { diff --git a/tests/mw-dev/e2e/trino.sh b/tests/mw-dev/e2e/trino.sh index 3c54d9344..9a391795c 100644 --- a/tests/mw-dev/e2e/trino.sh +++ b/tests/mw-dev/e2e/trino.sh @@ -410,6 +410,15 @@ done api "$API/api/v1/trino/queries/$query_id?cell=pool-test" | jq -e --arg q "$query_id" --arg org "$ORG_A" \ '.query_id == $q and .org == $org' >/dev/null \ || fail "admin Trino query detail did not identify tenant A query $query_id" +# A pooled cell names the instance whose coordinator holds each query (the +# console's per-instance breakdown). It must be one of the pool's own +# instances, in both the list and the detail. +pool_instances="$(api "$API/api/v1/trino/instances?cell=pool-test" | jq -c '[.instances[].instance_id]')" +for view in "$(api "$API/api/v1/trino/queries?cell=pool-test&org=$ORG_A&active=1" | jq -c --arg q "$query_id" '.queries[] | select(.query_id == $q)')" \ + "$(api "$API/api/v1/trino/queries/$query_id?cell=pool-test")"; do + printf %s "$view" | jq -e --argjson ids "$pool_instances" '(.instance // "") as $i | $i != "" and ($ids | index($i)) != null' >/dev/null \ + || fail "pooled Trino query $query_id does not name a pool instance (instances $pool_instances): $view" +done code="$(curl --cacert "$CA" -sS -o /tmp/trino-cross-query -w '%{http_code}' \ --user "$DB_B:$pw_b" -H "X-Trino-User: $DB_B" "$TRINO/v1/query/$query_id")" [ "$code" = 403 ] || fail "tenant B query detail for tenant A returned HTTP $code, want 403: $(cat /tmp/trino-cross-query)"