Line data Source code
1 : import * as dashql from '@ankoh/dashql-core';
2 : import { ProfileExecutionState } from './profile_execution_state.js';
3 :
4 : /// A raw progress payload of an engine, the progress model of the query turns it into events
5 : export type EngineProgressPayload =
6 : /// The `stats` of a Trino `/v1/statement` response
7 : | { kind: 'trino_statement_stats'; stats: object; receivedAt: number }
8 : /// The QueryInfo of a Trino query (`/v1/query/{id}`) with the plans of its stages
9 : | { kind: 'trino_query_info'; info: object; receivedAt: number }
10 : /// A later QueryInfo of a Trino query: the operator summaries of the stages whose tasks are done
11 : | { kind: 'trino_query_info_stats'; info: object; receivedAt: number }
12 : /// The status of a Hyper query (web HTTP API): the progress and the processed rows of the whole query
13 : | { kind: 'hyper_query_status'; status: HyperQueryStatus; receivedAt: number };
14 :
15 : /// The status of a Hyper query, the fields of the Hyper HTTP API that describe its progress
16 : export interface HyperQueryStatus {
17 : completionStatus: string;
18 : rowCount: number | null;
19 : progress: number | null;
20 : executionStats: { wallClockTime: number; rowsProcessed: number } | null;
21 : }
22 :
23 : /// The work units of a progress fraction, Hyper reports a fraction instead of work units
24 1 : const HYPER_PROGRESS_UNITS = 1000;
25 :
26 : /// The input events of the status of a Hyper query: the query only, Hyper reports no stages or pipelines yet
27 1 : function hyperQueryEvents(status: HyperQueryStatus, time: number): object[] {
28 1 : const stats: Record<string, number> = {};
29 1 : if (status.executionStats != null) {
30 1 : stats.tuplesIn = Math.max(0, Math.round(status.executionStats.rowsProcessed));
31 1 : stats.wallTimeUs = Math.max(0, Math.round(status.executionStats.wallClockTime * 1000));
32 1 : }
33 1 : if (status.rowCount != null) stats.tuplesOut = Math.max(0, Math.round(status.rowCount));
34 1 : if (status.progress != null) {
35 : // A fraction, values above 1 are percentages
36 1 : const fraction = Math.min(1, Math.max(0, status.progress > 1 ? status.progress / 100 : status.progress));
37 1 : stats.workDone = Math.round(fraction * HYPER_PROGRESS_UNITS);
38 1 : stats.workTotal = HYPER_PROGRESS_UNITS;
39 1 : }
40 1 : const finished = status.completionStatus === 'FINISHED';
41 1 : return [{ type: 'updateQuery', time: Math.max(0, Math.round(time)), status: finished ? 'succeeded' : 'running', stats }];
42 1 : }
43 :
44 : /// The structured progress of a query: the profile structure (stages, the plan operators once the plan is known)
45 : /// and its execution state
46 : export interface QueryProfileProgress {
47 : structure: dashql.FlatBufferPtr<dashql.buffers.profile.QueryProfile>;
48 : execution: ProfileExecutionState;
49 : /// The recording of the payloads so far, replaying it reproduces the profile
50 : recording: () => ProfileRecording;
51 : }
52 :
53 : /// A recording of the progress of a query: the raw payloads of the engine and when they arrived. Replaying the
54 : /// payloads through the adapters of the core reproduces the profile at any point in time.
55 : export interface ProfileRecording {
56 : version: 1;
57 : payloads: RecordedPayload[];
58 : }
59 :
60 : /// A payload of a recording
61 : export interface RecordedPayload {
62 : /// The milliseconds since the first payload
63 : elapsedMs: number;
64 : payload: EngineProgressPayload;
65 : }
66 :
67 : /// Replay a recording up to a point in time, returns null if nothing happened until then
68 1 : export function replayRecording(
69 1 : core: dashql.DashQL,
70 1 : recording: ProfileRecording,
71 1 : untilMs: number,
72 1 : layout?: dashql.buffers.profile.ProfileLayoutConfigT,
73 1 : ): QueryProfileProgress | null {
74 1 : const payloads = recording.payloads.filter(p => p.elapsedMs <= untilMs);
75 1 : if (payloads.length === 0) return null;
76 1 : const model = new QueryProgressModel(core, engineOf(payloads[0].payload), layout);
77 1 : try {
78 1 : let progress: QueryProfileProgress | null = null;
79 1 : for (const { payload } of payloads) {
80 1 : progress = model.apply(payload);
81 : }
82 1 : return progress;
83 1 : } finally {
84 1 : model.destroy();
85 1 : }
86 1 : }
87 :
88 : /// A progress model that follows the recording of the progress of a query with a layout of its own, e.g. the
89 : /// other orientation of the plan. The progress model of the query keeps its layout, the follower applies the
90 : /// recorded payloads again, the ones that are recorded later as they come.
91 1 : export class QueryProgressFollower {
92 1 : private model: QueryProgressModel | null = null;
93 1 : private source: (() => ProfileRecording) | null = null;
94 1 : private applied = 0;
95 1 : private progress: QueryProfileProgress | null = null;
96 :
97 1 : public constructor(private core: dashql.DashQL, private layout: dashql.buffers.profile.ProfileLayoutConfigT) {}
98 :
99 : /// Follow the progress of a query, null before its first payload
100 1 : public follow(progress: QueryProfileProgress): QueryProfileProgress | null {
101 1 : const payloads = progress.recording().payloads;
102 : // The progress of another query or another replay starts over
103 1 : if (progress.recording !== this.source || payloads.length < this.applied) {
104 1 : this.reset();
105 1 : this.source = progress.recording;
106 1 : }
107 1 : if (payloads.length === 0) return null;
108 1 : this.model ??= new QueryProgressModel(this.core, engineOf(payloads[0].payload), this.layout);
109 1 : for (let i = this.applied; i < payloads.length; ++i) {
110 1 : try {
111 1 : this.progress = this.model.apply(payloads[i].payload);
112 1 : } catch (e) {
113 : // A malformed payload is skipped, like by the progress model of the query
114 0 : console.warn(e);
115 0 : }
116 : }
117 1 : this.applied = payloads.length;
118 1 : return this.progress;
119 1 : }
120 :
121 1 : private reset() {
122 1 : this.model?.destroy();
123 1 : this.model = null;
124 1 : this.applied = 0;
125 1 : this.progress = null;
126 1 : }
127 :
128 1 : public destroy() {
129 1 : this.reset();
130 1 : this.source = null;
131 1 : }
132 1 : }
133 :
134 : /// Read a recording: a profile recording of the app, or a recording of the Trino test harness
135 : /// (snapshots/profiles/trino/recordings/, written by //packages/trino-testing:trino_recorder)
136 1 : export function readProfileRecording(text: string): ProfileRecording {
137 1 : const document = JSON.parse(text);
138 1 : if (document?.version === 1 && Array.isArray(document.payloads)) return document as ProfileRecording;
139 1 : if (Array.isArray(document?.statementResponses)) return convertTrinoHarnessRecording(document);
140 1 : throw new Error('The file is no profile recording');
141 1 : }
142 :
143 : /// A recording of the Trino test harness
144 : interface TrinoHarnessRecording {
145 : statementResponses: { elapsedMs: number; body: { stats?: { rootStage?: unknown } } }[];
146 : queryInfos: { elapsedMs: number; body: { stages?: { stages?: { stageStats?: object }[] } } }[];
147 : }
148 :
149 : /// Convert a recording of the Trino test harness. Its QueryInfos are the ones the app fetches when the tasks of
150 : /// stages are done and when the query finished. The last one carries the plans, the app requests them when the
151 : /// stages appear, so the plans move there, without the operator summaries that arrive with the later QueryInfos.
152 1 : function convertTrinoHarnessRecording(recording: TrinoHarnessRecording): ProfileRecording {
153 1 : const payloads: RecordedPayload[] = [];
154 1 : for (const response of recording.statementResponses) {
155 1 : if (response.body.stats == null) continue;
156 1 : payloads.push({ elapsedMs: response.elapsedMs, payload: { kind: 'trino_statement_stats', stats: response.body.stats, receivedAt: response.elapsedMs } });
157 1 : }
158 1 : for (const sample of recording.queryInfos) {
159 1 : payloads.push({ elapsedMs: sample.elapsedMs, payload: { kind: 'trino_query_info_stats', info: sample.body, receivedAt: sample.elapsedMs } });
160 : }
161 1 : const final = recording.queryInfos[recording.queryInfos.length - 1]?.body;
162 1 : const stagesAt = recording.statementResponses.find(r => r.body.stats?.rootStage != null)?.elapsedMs;
163 1 : if (final != null && stagesAt != null) {
164 1 : const stages = (final.stages?.stages ?? []).map(stage => ({
165 1 : ...stage,
166 1 : stageStats: { ...stage.stageStats, operatorSummaries: [] },
167 1 : }));
168 1 : const plans = { ...final, stages: { ...final.stages, stages } };
169 1 : payloads.push({ elapsedMs: stagesAt, payload: { kind: 'trino_query_info', info: plans, receivedAt: stagesAt } });
170 1 : }
171 1 : payloads.sort((a, b) => a.elapsedMs - b.elapsedMs);
172 1 : return { version: 1, payloads };
173 1 : }
174 :
175 : /// The engine of a payload
176 1 : function engineOf(payload: EngineProgressPayload): dashql.buffers.profile.ProfileEngine {
177 1 : switch (payload.kind) {
178 1 : case 'trino_statement_stats':
179 1 : case 'trino_query_info':
180 1 : case 'trino_query_info_stats':
181 1 : return dashql.buffers.profile.ProfileEngine.TRINO;
182 1 : case 'hyper_query_status':
183 1 : return dashql.buffers.profile.ProfileEngine.HYPER;
184 1 : }
185 1 : }
186 :
187 : /// The progress model of one query: a core query profile that the payloads of the engine are applied to.
188 : /// The model lives in the query executor; the query state only holds the immutable `QueryProfileProgress`.
189 1 : export class QueryProgressModel {
190 1 : private profile: dashql.DashQLQueryProfile;
191 1 : private engine: dashql.buffers.profile.ProfileEngine;
192 1 : private progress: QueryProfileProgress;
193 : /// The latest statement stats, applied again on top of a plan that arrives later
194 1 : private latestStats: Extract<EngineProgressPayload, { kind: 'trino_statement_stats' }> | null = null;
195 : /// The payloads so far, the recording of the query
196 1 : private payloads: RecordedPayload[] = [];
197 1 : private readonly recording = (): ProfileRecording => ({ version: 1, payloads: [...this.payloads] });
198 :
199 1 : public constructor(core: dashql.DashQL, engine: dashql.buffers.profile.ProfileEngine, layout = new dashql.buffers.profile.ProfileLayoutConfigT()) {
200 1 : this.profile = core.createQueryProfile(layout);
201 1 : this.engine = engine;
202 1 : this.progress = { structure: this.profile.startQuery(engine), execution: ProfileExecutionState.EMPTY, recording: this.recording };
203 1 : }
204 :
205 : /// Apply a payload of the engine, a payload of another engine changes nothing
206 1 : public apply(payload: EngineProgressPayload): QueryProfileProgress {
207 1 : if (engineOf(payload) !== this.engine) return this.progress;
208 1 : const first = this.payloads[0]?.payload.receivedAt ?? payload.receivedAt;
209 1 : this.payloads.push({ elapsedMs: Math.max(0, payload.receivedAt - first), payload });
210 1 : let execution: ProfileExecutionState;
211 1 : switch (payload.kind) {
212 1 : case 'trino_statement_stats': {
213 1 : this.latestStats = payload;
214 1 : const events = this.profile.applyTrinoStatementStats(JSON.stringify(payload.stats), payload.receivedAt);
215 1 : execution = this.progress.execution.withEvents(events.read());
216 1 : break;
217 1 : }
218 1 : case 'trino_query_info': {
219 : // The plan replaces the stages that the stats registered. The stats are cumulative, applying the
220 : // latest ones again restores the state on the stages of the plan.
221 1 : try {
222 1 : this.profile.loadTrinoPlan(JSON.stringify(payload.info));
223 1 : } catch (e) {
224 : // Without a plan the stages stay a list
225 1 : console.warn(e);
226 1 : this.profile.startQuery(this.engine);
227 1 : }
228 1 : if (this.latestStats != null) {
229 1 : this.profile.applyTrinoStatementStats(JSON.stringify(this.latestStats.stats), this.latestStats.receivedAt);
230 : }
231 : // The operator summaries of the stages that finished already
232 1 : this.profile.applyTrinoQueryInfo(JSON.stringify(payload.info), payload.receivedAt);
233 1 : execution = ProfileExecutionState.EMPTY.withEvents(this.profile.snapshotEvents().read());
234 1 : break;
235 : }
236 1 : case 'hyper_query_status': {
237 1 : const events = this.profile.applyEvents(JSON.stringify(hyperQueryEvents(payload.status, payload.receivedAt)));
238 1 : execution = this.progress.execution.withEvents(events.read());
239 1 : break;
240 1 : }
241 1 : case 'trino_query_info_stats': {
242 1 : const events = this.profile.applyTrinoQueryInfo(JSON.stringify(payload.info), payload.receivedAt);
243 1 : execution = this.progress.execution.withEvents(events.read());
244 1 : break;
245 1 : }
246 1 : }
247 1 : const structure = this.profile.buffer ?? this.progress.structure;
248 1 : if (execution !== this.progress.execution || structure !== this.progress.structure) {
249 1 : this.progress = { structure, execution, recording: this.recording };
250 : }
251 1 : return this.progress;
252 1 : }
253 :
254 1 : public destroy() {
255 1 : this.profile.destroy();
256 1 : }
257 1 : }
258 :
259 : /// The progress models of running queries, created with the first payload of a query
260 1 : export class QueryProgressModels {
261 1 : private models = new Map<number, QueryProgressModel>();
262 :
263 : /// The layout of the profiles of the queries
264 1 : public constructor(private layout = new dashql.buffers.profile.ProfileLayoutConfigT()) {}
265 :
266 : /// Apply a payload to the model of a query. Without core (still loading) the payload is dropped, the counters
267 : /// are cumulative, so the next payload catches up.
268 1 : public apply(core: dashql.DashQL | null, queryId: number, payload: EngineProgressPayload): QueryProfileProgress | null {
269 1 : let model = this.models.get(queryId);
270 1 : if (model == null) {
271 1 : if (core == null) return null;
272 1 : model = new QueryProgressModel(core, engineOf(payload), this.layout);
273 1 : this.models.set(queryId, model);
274 1 : }
275 1 : try {
276 1 : return model.apply(payload);
277 1 : } catch (e) {
278 : // A malformed payload must not interrupt the query
279 0 : console.warn(e);
280 0 : return null;
281 0 : }
282 1 : }
283 :
284 : /// Release the model of a query
285 1 : public release(queryId: number) {
286 1 : this.models.get(queryId)?.destroy();
287 1 : this.models.delete(queryId);
288 1 : }
289 1 : }
|
