diff --git a/controlplane/admin/trino_client.go b/controlplane/admin/trino_client.go index d6b8c72d..f17dde40 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 da395b4e..97958749 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 a45a00d2..8143c0d9 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 32535b1c..acbee359 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 f024d694..0ce96f54 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 b2aad505..0bfc5f6e 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 f4c13cef..caab053f 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 55411c9c..4b16c08c 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 1bc94cf5..ec3fb67c 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 929acfbc..182241b8 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 3c54d934..9a391795 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)"