Line data Source code
1 : import * as arrow from 'apache-arrow';
2 : import * as connection from '@ankoh/dashql-jsonschema/connection.js';
3 : import * as buf from '@bufbuild/protobuf';
4 :
5 : import { AsyncConsumer } from '../utils/async_consumer.js';
6 : import { removePrimitiveFromArray } from '../utils/array.js';
7 : import { LoggableException } from '../platform/logger/logger.js';
8 : import { VariantKind } from '../utils/variant.js';
9 : import type { EngineProgressPayload, QueryProfileProgress } from './query_progress_model.js';
10 :
11 1 : export enum QueryExecutionStatus {
12 1 : REQUESTED = 0,
13 1 : PREPARING = 1,
14 1 : SENDING = 2,
15 1 : QUEUED = 3,
16 1 : RUNNING = 4,
17 1 : RECEIVED_FIRST_BATCH = 5,
18 1 : RECEIVED_ALL_BATCHES = 6,
19 1 : PROCESSING_RESULTS = 7,
20 1 : PROCESSED_RESULTS = 8,
21 1 : SUCCEEDED = 9,
22 1 : FAILED = 10,
23 1 : CANCELLED = 11,
24 1 : }
25 :
26 1 : export function queryIsDone(status: QueryExecutionStatus) {
27 1 : switch (status) {
28 1 : case QueryExecutionStatus.SUCCEEDED:
29 1 : case QueryExecutionStatus.FAILED:
30 1 : case QueryExecutionStatus.CANCELLED:
31 1 : return true;
32 1 : default:
33 1 : return false;
34 1 : }
35 1 : }
36 :
37 : export interface QueryExecutionMetrics {
38 : /// The total data bytes
39 : totalDataBytesReceived: number;
40 : /// The total batches received
41 : totalBatchesReceived: number;
42 : /// The total batches received
43 : totalRowsReceived: number;
44 :
45 : /// The query requests that started
46 : totalQueryRequestsStarted: number;
47 : /// The query requests that finished
48 : totalQueryRequestsSucceeded: number;
49 : /// The query requests that failed
50 : totalQueryRequestsFailed: number
51 : /// The total request duration of all requests that failed or succeeded
52 : totalQueryRequestDurationMs: number;
53 :
54 : /// The duration until the first batch
55 : durationUntilFirstBatchMs: number | null;
56 : }
57 :
58 : export interface QueryExecutionProgress {
59 : /// Is queued?
60 : isQueued: boolean | null;
61 : /// The progress in percent, as reported by the server
62 : progressPercent: number | null;
63 : /// The number of result rows, as reported by the server
64 : resultRows: number | null;
65 : /// The number of rows the server processed
66 : rowsProcessed: number | null;
67 : /// The execution time on the server in milliseconds
68 : executionTimeMs: number | null;
69 : /// The query execution metrics
70 : metrics: QueryExecutionMetrics;
71 : /// The phase the engine reports, e.g. the Trino query state (QUEUED, PLANNING, RUNNING, FINISHING)
72 : phase?: string | null;
73 : /// The raw progress payload of the engine, the query executor feeds it into the progress model of the query
74 : engineProgress?: EngineProgressPayload | null;
75 : /// The structured progress (stages, pipelines, the plan), set by the query executor
76 : profileProgress?: QueryProfileProgress | null;
77 : }
78 :
79 : export interface QueryExecutionResponseStream {
80 : /// Ask the backend to stop the query, when the connector supports explicit cancellation.
81 : cancel?(): Promise<void>;
82 : /// Is the profile of the query shown? Connectors fetch the plan and its statistics only then (Trino).
83 : setProfileDetail?(wanted: boolean): void;
84 : /// Get the result metadata (after completion)
85 : getMetadata(): Map<string, string>;
86 : /// Get the stream metrics
87 : getMetrics(): QueryExecutionMetrics;
88 : /// Get the current query status
89 : getStatus(): QueryExecutionStatus;
90 : /// Await the schema message
91 : getSchema(): Promise<arrow.Schema | null>;
92 : /// Await the next record batch
93 : produce(batches: AsyncConsumer<QueryExecutionResponseStream, arrow.RecordBatch>, progress: AsyncConsumer<QueryExecutionResponseStream, QueryExecutionProgress>, abort?: AbortSignal): Promise<void>;
94 : }
95 :
96 : export interface QueryMetrics {
97 : /// The time at which we requested the query (if any)
98 : queryRequestedAt: Date | null;
99 : /// The time at which we prepared the query (if any)
100 : queryPreparingStartedAt: Date | null;
101 : /// The time at which the query was sent (if any)
102 : querySendingStartedAt: Date | null;
103 : /// The time at which the query was queued (if any)
104 : queryQueuedStartedAt: Date | null;
105 : /// The time at which we the query started running (if any)
106 : queryRunningStartedAt: Date | null;
107 : /// Received the frist batch at (if any)
108 : receivedFirstBatchAt: Date | null;
109 : /// Received all batches at (if any)
110 : receivedLastBatchAt: Date | null;
111 : /// Received all batches at (if any)
112 : receivedAllBatchesAt: Date | null;
113 : /// The time at which we started processed the results
114 : processingResultsStartedAt: Date | null;
115 : /// The time at which we processed the results
116 : processedResultsAt: Date | null;
117 : /// The time at which the query execution finished (if any)
118 : querySucceededAt: Date | null;
119 : /// The time at which the query execution failed (if any)
120 : queryFailedAt: Date | null;
121 : /// The time at which the query execution was cancelled (if any)
122 : queryCancelledAt: Date | null;
123 : /// The time at which the query execution was last updated
124 : lastUpdatedAt: Date | null;
125 : /// The number of query_status updates received
126 : progressUpdatesReceived: number;
127 : /// The total query duration
128 : queryDurationMs: number | null;
129 : /// The text length in characters
130 : textLength: number | null;
131 : /// The stream metrics
132 : streamMetrics: QueryExecutionMetrics;
133 : }
134 :
135 : /// A query hint
136 1 : export enum QueryType {
137 1 : USER_PROVIDED,
138 1 : CATALOG_QUERY_INFORMATION_SCHEMA,
139 1 : CATALOG_QUERY_PG_ATTRIBUTE,
140 1 : CATALOG_QUERY_PG_PROC,
141 1 : CATALOG_QUERY_HYPER_CLOUD,
142 1 : HEALTH_CHECK,
143 1 : INTERNAL_SQLFRAME,
144 1 : }
145 :
146 : export interface QueryMetadata {
147 : /// The query type
148 : queryType: QueryType;
149 : /// The title of the query (if any)
150 : title: string | null;
151 : /// The description
152 : description: string | null;
153 : /// The issuer
154 : issuer: string | null;
155 : /// Authored by the user or the app?
156 : userProvided: boolean;
157 : /// The session that issued the query, if any.
158 : /// Several sessions may share a connection, the session scopes the query result cache and the
159 : /// cleanup of the session's queries.
160 : sessionId?: string;
161 : /// The owner of the result, e.g. a script of a session. A succeeded query releases the result table of the
162 : /// earlier queries of its owner, the owner shows only its latest result.
163 : resultOwner?: string;
164 : /// Where a statement of a plan of session scripts comes from
165 : planned?: PlannedStatement;
166 : }
167 :
168 : /// A statement of a plan of session scripts (docs/design/planner_dependencies.md)
169 : export interface PlannedStatement {
170 : /// A run of the whole session or of a script
171 : run: 'session' | 'script';
172 : /// The script that a run of a script runs
173 : target: string | null;
174 : /// The node of the plan
175 : node: number;
176 : /// The script of the statement
177 : script: string;
178 : /// The statement within the statements of the script in the plan, from 1
179 : statement: number;
180 : statements: number;
181 : /// The type of the statement, e.g. SELECT
182 : statementType: string;
183 : /// Is the statement the result of its script?
184 : result: boolean;
185 : /// Is the statement the cached result of its script? The executor reads the file instead of running it.
186 : cached: boolean;
187 : /// The scripts whose queries the statement inlines as common table expressions
188 : inlined: string[];
189 : /// The materialized results of the scripts of other connections that the statement scans, cached if the host
190 : /// provides the file of an earlier run
191 : scans: Array<{ script: string; path: string; cached: boolean }>;
192 : /// The file that the result of the statement is written to, for statements of other connections
193 : materializes: string | null;
194 : /// The nodes the statement runs after and why
195 : after: Array<{ node: number; reason: string }>;
196 : }
197 :
198 : export interface QueryExecutionState {
199 : /// The query id
200 : queryId: number;
201 : /// The trace id for this query execution
202 : traceId: number;
203 : /// The query text
204 : queryText: string;
205 : /// The query metadata
206 : queryMetadata: QueryMetadata;
207 : /// The current status
208 : status: QueryExecutionStatus;
209 : /// The query metric counters
210 : metrics: QueryMetrics;
211 : /// The cancellation signal
212 : cancellation: AbortController;
213 : /// The response stream
214 : resultStream: QueryExecutionResponseStream | null;
215 : /// The loading error (if any)
216 : error: LoggableException | null;
217 : /// The latest update for the query execution
218 : latestProgressUpdate: QueryExecutionProgress | null;
219 : /// One-based active statement position for a core-steered script execution.
220 : statementIndex: number | null;
221 : /// Total statements in the active script execution.
222 : statementCount: number | null;
223 : /// Whether the latest script statement completed successfully.
224 : statementSucceeded: boolean;
225 : /// The number of record batches that are already buffered
226 : resultSchema: arrow.Schema | null;
227 : /// The number of record batches that are already buffered
228 : resultBatches: arrow.RecordBatch[];
229 : /// The result metadata
230 : resultMetadata: Map<string, string> | null;
231 : /// The result query_result iff the query succeeded
232 : resultTable: arrow.Table | null;
233 : /// The file-based query result cache key for this execution, or null when the query is not
234 : /// cacheable (or no key could be derived). Recorded so the UI can offer a "delete cached" action.
235 : cacheKey: string | null;
236 : /// Whether this result was served from the query result cache instead of the backend.
237 : servedFromCache: boolean;
238 : /// Whether the cached `.arrow` entry for this query has been deleted (via the UI). Used to
239 : /// disable the "delete cached" button once there is nothing left to delete.
240 : cacheDeleted: boolean;
241 : /// When the cache entry backing this result was written (`.arrow` file mtime), or null when the
242 : /// result did not come from the cache. Lets the UI show how old a cached result is. Note that a
243 : /// cache hit does not re-touch the file, so this is the write time, not a last-access time.
244 : cachedAt: Date | null;
245 : }
246 :
247 : export interface QueryExecutionTracker {
248 : dispatch(action: QueryExecutionAction): void;
249 : }
250 :
251 1 : export const EXECUTE_QUERY = Symbol('EXECUTE_QUERY');
252 1 : export const QUERY_PREPARING = Symbol('QUERY_PREPARING');
253 1 : export const QUERY_SENDING = Symbol('QUERY_SENDING');
254 1 : export const QUERY_RUNNING = Symbol('QUERY_RUNNING');
255 1 : export const QUERY_PROGRESS_UPDATED = Symbol('QUERY_PROGRESS_UPDATED');
256 1 : export const QUERY_STATEMENT_STARTED = Symbol('QUERY_STATEMENT_STARTED');
257 1 : export const QUERY_STATEMENT_SUCCEEDED = Symbol('QUERY_STATEMENT_SUCCEEDED');
258 1 : export const QUERY_RECEIVED_BATCH = Symbol('QUERY_RECEIVED_BATCH');
259 1 : export const QUERY_RECEIVED_ALL_BATCHES = Symbol('QUERY_RECEIVED_ALL_BATCHES');
260 1 : export const QUERY_PROCESSING_RESULTS = Symbol('QUERY_PROCESSING_RESULTS');
261 1 : export const QUERY_PROCESSED_RESULTS = Symbol('QUERY_PROCESSED_RESULTS');
262 1 : export const QUERY_SUCCEEDED = Symbol('QUERY_SUCCEEDED');
263 1 : export const QUERY_FAILED = Symbol('QUERY_FAILED');
264 1 : export const QUERY_CANCELLED = Symbol('QUERY_CANCELLED');
265 1 : export const QUERY_CACHE_RECORDED = Symbol('QUERY_CACHE_RECORDED');
266 1 : export const QUERY_CACHE_DELETED = Symbol('QUERY_CACHE_DELETED');
267 :
268 : export type QueryExecutionAction =
269 : | VariantKind<typeof EXECUTE_QUERY, [number, QueryExecutionState]>
270 : | VariantKind<typeof QUERY_PREPARING, [number]>
271 : | VariantKind<typeof QUERY_SENDING, [number]>
272 : | VariantKind<typeof QUERY_RUNNING, [number, QueryExecutionResponseStream | null]>
273 : | VariantKind<typeof QUERY_PROGRESS_UPDATED, [number, QueryExecutionProgress]>
274 : | VariantKind<typeof QUERY_STATEMENT_STARTED, [number, number, number]>
275 : | VariantKind<typeof QUERY_STATEMENT_SUCCEEDED, [number, number, number]>
276 : | VariantKind<typeof QUERY_RECEIVED_BATCH, [number, arrow.RecordBatch, QueryExecutionMetrics]>
277 : | VariantKind<typeof QUERY_RECEIVED_ALL_BATCHES, [number, arrow.Table, Map<string, string>, QueryExecutionMetrics]>
278 : | VariantKind<typeof QUERY_PROCESSING_RESULTS, [number]>
279 : | VariantKind<typeof QUERY_PROCESSED_RESULTS, [number]>
280 : | VariantKind<typeof QUERY_SUCCEEDED, [number]>
281 : | VariantKind<typeof QUERY_FAILED, [number, LoggableException, QueryExecutionMetrics | null]>
282 : | VariantKind<typeof QUERY_CANCELLED, [number, LoggableException, QueryExecutionMetrics | null]>
283 : | VariantKind<typeof QUERY_CACHE_RECORDED, [number, string, boolean, number | null]>
284 : | VariantKind<typeof QUERY_CACHE_DELETED, [number]>;
285 :
286 : export interface QueryExecutionHistoryState {
287 : queriesActive: Map<number, QueryExecutionState>;
288 : queriesActiveOrdered: number[];
289 : queriesFinished: Map<number, QueryExecutionState>;
290 : queriesFinishedOrdered: number[];
291 : snapshotQueriesActiveFinished: number;
292 : metrics: connection.ConnectionMetrics;
293 : }
294 :
295 1 : export function createQueryExecutionState(
296 1 : queryId: number,
297 1 : traceId: number,
298 1 : queryText: string,
299 1 : queryMetadata: QueryMetadata,
300 1 : cancellation: AbortController,
301 1 : ): QueryExecutionState {
302 1 : const requestedAt = new Date();
303 1 : return {
304 1 : queryId,
305 1 : traceId,
306 1 : queryText,
307 1 : queryMetadata,
308 1 : status: QueryExecutionStatus.REQUESTED,
309 1 : cancellation,
310 1 : resultStream: null,
311 1 : error: null,
312 1 : metrics: {
313 1 : textLength: queryText.length,
314 1 : queryRequestedAt: requestedAt,
315 1 : queryPreparingStartedAt: null,
316 1 : querySendingStartedAt: null,
317 1 : queryQueuedStartedAt: null,
318 1 : queryRunningStartedAt: null,
319 1 : receivedFirstBatchAt: null,
320 1 : receivedLastBatchAt: null,
321 1 : receivedAllBatchesAt: null,
322 1 : processingResultsStartedAt: null,
323 1 : processedResultsAt: null,
324 1 : querySucceededAt: null,
325 1 : queryFailedAt: null,
326 1 : queryCancelledAt: null,
327 1 : lastUpdatedAt: requestedAt,
328 1 : progressUpdatesReceived: 0,
329 1 : queryDurationMs: null,
330 1 : streamMetrics: createQueryResponseStreamMetrics(),
331 1 : },
332 1 : latestProgressUpdate: null,
333 1 : statementIndex: null,
334 1 : statementCount: null,
335 1 : statementSucceeded: false,
336 1 : resultMetadata: null,
337 1 : resultSchema: null,
338 1 : resultBatches: [],
339 1 : resultTable: null,
340 1 : cacheKey: null,
341 1 : servedFromCache: false,
342 1 : cacheDeleted: false,
343 1 : cachedAt: null,
344 1 : };
345 1 : }
346 :
347 1 : export function reduceQueryAction<T extends QueryExecutionHistoryState>(state: T, action: QueryExecutionAction): T {
348 1 : const now = new Date();
349 1 : const queryId = action.value[0];
350 :
351 : // Initial setup?
352 1 : if (action.type == EXECUTE_QUERY) {
353 1 : state.queriesActive.set(queryId, action.value[1]);
354 1 : state.queriesActiveOrdered.push(queryId);
355 1 : state.snapshotQueriesActiveFinished += 1;
356 1 : return { ...state };
357 1 : }
358 :
359 : // Cache bookkeeping actions can arrive after the query has already moved to `queriesFinished`
360 : // (the write path fires-and-forgets, and the UI delete happens on a finished result), so resolve
361 : // the query from whichever map holds it and write the update back to the same map.
362 1 : if (action.type == QUERY_CACHE_RECORDED || action.type == QUERY_CACHE_DELETED) {
363 1 : const isActive = state.queriesActive.has(queryId);
364 1 : const target = isActive ? state.queriesActive : state.queriesFinished;
365 1 : const q = target.get(queryId);
366 1 : if (!q) {
367 1 : return state;
368 : }
369 1 : let next: QueryExecutionState;
370 1 : if (action.type == QUERY_CACHE_RECORDED) {
371 1 : const [, cacheKey, servedFromCache, cachedAtMs] = action.value;
372 1 : next = {
373 1 : ...q,
374 1 : cacheKey,
375 1 : servedFromCache,
376 1 : cachedAt: cachedAtMs != null ? new Date(cachedAtMs) : q.cachedAt,
377 1 : };
378 1 : } else {
379 1 : next = { ...q, cacheDeleted: true };
380 : }
381 1 : target.set(queryId, next);
382 1 : return { ...state };
383 1 : }
384 :
385 1 : let query = state.queriesActive.get(queryId);
386 1 : if (!query) {
387 1 : return state;
388 : }
389 1 : switch (action.type) {
390 1 : case QUERY_PREPARING: {
391 1 : query = {
392 1 : ...query,
393 1 : status: QueryExecutionStatus.PREPARING,
394 1 : metrics: {
395 1 : ...query.metrics,
396 1 : lastUpdatedAt: now,
397 1 : queryPreparingStartedAt: now,
398 1 : },
399 1 : };
400 1 : state.queriesActive.set(query.queryId, query);
401 1 : return { ...state };
402 : }
403 1 : case QUERY_SENDING: {
404 1 : query = {
405 1 : ...query,
406 1 : status: QueryExecutionStatus.SENDING,
407 1 : metrics: {
408 1 : ...query.metrics,
409 1 : lastUpdatedAt: now,
410 1 : querySendingStartedAt: now,
411 1 : },
412 1 : };
413 1 : state.queriesActive.set(query.queryId, query);
414 1 : return { ...state };
415 : }
416 1 : case QUERY_RUNNING: {
417 1 : query = {
418 1 : ...query,
419 1 : status: QueryExecutionStatus.RUNNING,
420 1 : resultStream: action.value[1],
421 1 : metrics: {
422 1 : ...query.metrics,
423 1 : lastUpdatedAt: now,
424 1 : queryRunningStartedAt: now,
425 1 : },
426 1 : };
427 1 : state.queriesActive.set(query.queryId, query);
428 1 : return { ...state };
429 : }
430 1 : case QUERY_PROGRESS_UPDATED: {
431 : // An update without structured progress keeps the last one, e.g. while the core loads
432 1 : const progress = action.value[1];
433 1 : const profileProgress = progress.profileProgress ?? query.latestProgressUpdate?.profileProgress ?? null;
434 1 : const next = {
435 1 : ...query,
436 1 : latestProgressUpdate: { ...progress, profileProgress },
437 1 : metrics: {
438 1 : ...query.metrics,
439 1 : lastUpdatedAt: now,
440 1 : progressUpdatesReceived: ++query.metrics.progressUpdatesReceived
441 1 : },
442 1 : };
443 : // Is the query queued for the first time?
444 1 : if (action.value[1].isQueued && next.status == QueryExecutionStatus.SENDING) {
445 1 : next.status = QueryExecutionStatus.QUEUED;
446 1 : next.metrics.queryQueuedStartedAt = now;
447 1 : }
448 1 : state.queriesActive.set(query.queryId, next);
449 1 : return { ...state };
450 1 : }
451 1 : case QUERY_STATEMENT_STARTED: {
452 1 : query = {
453 1 : ...query,
454 1 : statementIndex: action.value[1],
455 1 : statementCount: action.value[2],
456 1 : statementSucceeded: false,
457 1 : status: QueryExecutionStatus.RUNNING,
458 1 : metrics: {
459 1 : ...query.metrics,
460 1 : lastUpdatedAt: now,
461 1 : },
462 1 : };
463 1 : state.queriesActive.set(query.queryId, query);
464 1 : return { ...state };
465 : }
466 1 : case QUERY_STATEMENT_SUCCEEDED: {
467 1 : query = {
468 1 : ...query,
469 1 : statementIndex: action.value[1],
470 1 : statementCount: action.value[2],
471 1 : statementSucceeded: true,
472 1 : status: QueryExecutionStatus.RUNNING,
473 1 : metrics: {
474 1 : ...query.metrics,
475 1 : lastUpdatedAt: now,
476 1 : },
477 1 : };
478 1 : state.queriesActive.set(query.queryId, query);
479 1 : return { ...state };
480 : }
481 1 : case QUERY_RECEIVED_BATCH: {
482 1 : const [_queryId, batch, streamMetrics] = action.value;
483 1 : const metrics = { ...query.metrics };
484 1 : if (metrics.receivedFirstBatchAt == null) {
485 1 : metrics.receivedFirstBatchAt = now;
486 : }
487 1 : metrics.receivedLastBatchAt = now;
488 1 : metrics.lastUpdatedAt = now;
489 1 : metrics.streamMetrics = streamMetrics;
490 1 : query.resultBatches.push(batch);
491 1 : query = {
492 1 : ...query,
493 1 : status: QueryExecutionStatus.RECEIVED_FIRST_BATCH,
494 1 : metrics: metrics,
495 1 : };
496 1 : if (query.resultSchema == null) {
497 1 : query.resultSchema = batch.schema;
498 : }
499 1 : state.queriesActive.set(query.queryId, query);
500 1 : return { ...state };
501 1 : }
502 1 : case QUERY_RECEIVED_ALL_BATCHES: {
503 1 : const metrics = { ...query.metrics };
504 1 : metrics.lastUpdatedAt = now;
505 1 : metrics.receivedAllBatchesAt = now;
506 1 : metrics.streamMetrics = action.value[3];
507 1 : query = {
508 1 : ...query,
509 1 : resultMetadata: action.value[2],
510 1 : resultTable: action.value[1],
511 1 : status: QueryExecutionStatus.RECEIVED_ALL_BATCHES,
512 1 : metrics: metrics,
513 1 : };
514 1 : state.queriesActive.set(query.queryId, query);
515 1 : return { ...state, };
516 1 : }
517 1 : case QUERY_PROCESSING_RESULTS: {
518 1 : const metrics = { ...query.metrics };
519 1 : metrics.lastUpdatedAt = now;
520 1 : metrics.processingResultsStartedAt = now;
521 1 : query = {
522 1 : ...query,
523 1 : status: QueryExecutionStatus.PROCESSING_RESULTS,
524 1 : metrics: metrics,
525 1 : };
526 1 : state.queriesActive.set(query.queryId, query);
527 1 : return { ...state, };
528 1 : }
529 1 : case QUERY_PROCESSED_RESULTS: {
530 1 : const metrics = { ...query.metrics };
531 1 : metrics.lastUpdatedAt = now;
532 1 : metrics.processedResultsAt = now;
533 1 : query = {
534 1 : ...query,
535 1 : status: QueryExecutionStatus.PROCESSED_RESULTS,
536 1 : metrics: metrics,
537 1 : };
538 1 : state.queriesActive.set(query.queryId, query);
539 1 : return { ...state, };
540 1 : }
541 1 : case QUERY_SUCCEEDED: {
542 1 : const metrics = { ...query.metrics };
543 1 : const untilNow = now.getTime() - (query.metrics.queryRunningStartedAt ?? query.metrics.queryRequestedAt ?? now).getTime();
544 1 : metrics.lastUpdatedAt = now;
545 1 : metrics.querySucceededAt = now;
546 1 : metrics.queryDurationMs = untilNow;
547 1 : query = {
548 1 : ...query,
549 1 : status: QueryExecutionStatus.SUCCEEDED,
550 1 : metrics: metrics,
551 1 : };
552 1 : finishQuery(state, query);
553 1 : return query.queryMetadata.queryType == QueryType.INTERNAL_SQLFRAME
554 1 : ? { ...state }
555 1 : : {
556 1 : ...state,
557 1 : metrics: {
558 1 : ...state.metrics,
559 1 : successfulQueries: mergeQueryMetrics(state.metrics.successfulQueries!, metrics)
560 1 : },
561 1 : };
562 1 : }
563 1 : case QUERY_CANCELLED: {
564 1 : const untilNow = now.getTime() - (query.metrics.queryRunningStartedAt ?? query.metrics.queryRequestedAt ?? now).getTime();
565 1 : const metrics = { ...query.metrics };
566 1 : metrics.lastUpdatedAt = now;
567 1 : metrics.queryCancelledAt = now;
568 1 : metrics.queryDurationMs = untilNow;
569 1 : metrics.streamMetrics = action.value[2] ?? metrics.streamMetrics;
570 1 : query = {
571 1 : ...query,
572 1 : status: QueryExecutionStatus.CANCELLED,
573 1 : error: action.value[1],
574 1 : metrics
575 1 : };
576 1 : finishQuery(state, query);
577 1 : return query.queryMetadata.queryType == QueryType.INTERNAL_SQLFRAME
578 1 : ? { ...state }
579 1 : : {
580 1 : ...state,
581 1 : metrics: {
582 1 : ...state.metrics,
583 1 : canceledQueries: mergeQueryMetrics(state.metrics.canceledQueries!, metrics)
584 1 : },
585 1 : };
586 1 : }
587 1 : case QUERY_FAILED: {
588 1 : const untilNow = now.getTime() - (query.metrics.queryRunningStartedAt ?? query.metrics.queryRequestedAt ?? now).getTime();
589 1 : const metrics = { ...query.metrics };
590 1 : metrics.lastUpdatedAt = now;
591 1 : metrics.queryFailedAt = now;
592 1 : metrics.queryDurationMs = untilNow;
593 1 : metrics.streamMetrics = action.value[2] ?? metrics.streamMetrics;
594 1 : query = {
595 1 : ...query,
596 1 : status: QueryExecutionStatus.FAILED,
597 1 : error: action.value[1],
598 1 : metrics
599 1 : };
600 1 : finishQuery(state, query);
601 1 : return query.queryMetadata.queryType == QueryType.INTERNAL_SQLFRAME
602 1 : ? { ...state }
603 1 : : {
604 1 : ...state,
605 1 : metrics: {
606 1 : ...state.metrics,
607 1 : failedQueries: mergeQueryMetrics(state.metrics.failedQueries!, metrics)
608 1 : },
609 1 : };
610 1 : }
611 1 : }
612 1 : }
613 :
614 : /// The finished internal queries that the history keeps, internal queries keep no result
615 : export const INTERNAL_QUERY_HISTORY_LIMIT = 100;
616 :
617 : /// Move a terminal query to the finished queries.
618 : ///
619 : /// The finished queries keep their result tables for the views of their results, but not their batches, the
620 : /// table holds them. Internal queries return their results to their callers and keep only their metrics. A
621 : /// succeeded query with a result owner releases the result tables of the other queries of its owner, except for
622 : /// a later one, like the latest result of a script.
623 1 : function finishQuery(state: QueryExecutionHistoryState, query: QueryExecutionState) {
624 1 : const userProvided = query.queryMetadata.queryType == QueryType.USER_PROVIDED;
625 1 : query = {
626 1 : ...query,
627 1 : resultBatches: [],
628 1 : resultTable: userProvided ? query.resultTable : null,
629 1 : };
630 1 : state.queriesActive.delete(query.queryId);
631 1 : removePrimitiveFromArray(state.queriesActiveOrdered, query.queryId);
632 1 : state.queriesFinished.set(query.queryId, query);
633 1 : state.queriesFinishedOrdered.push(query.queryId);
634 1 : state.snapshotQueriesActiveFinished += 1;
635 :
636 1 : const owner = query.queryMetadata.resultOwner;
637 1 : if (userProvided && owner != null && query.status == QueryExecutionStatus.SUCCEEDED && query.resultTable != null) {
638 1 : for (const [otherId, other] of state.queriesFinished) {
639 1 : if (otherId === query.queryId || other.resultTable == null || other.queryMetadata.resultOwner !== owner) continue;
640 : // A later result stays, the earlier query finished after it
641 1 : if (otherId > query.queryId) {
642 1 : state.queriesFinished.set(query.queryId, { ...query, resultTable: null });
643 : } else {
644 1 : state.queriesFinished.set(otherId, { ...other, resultTable: null });
645 : }
646 1 : }
647 : }
648 :
649 : // Internal queries leave only their metrics
650 1 : let internal = 0;
651 1 : for (const queryId of state.queriesFinishedOrdered) {
652 1 : if (state.queriesFinished.get(queryId)?.queryMetadata.queryType != QueryType.USER_PROVIDED) ++internal;
653 : }
654 1 : for (let i = 0; internal > INTERNAL_QUERY_HISTORY_LIMIT && i < state.queriesFinishedOrdered.length;) {
655 1 : const queryId = state.queriesFinishedOrdered[i];
656 1 : if (state.queriesFinished.get(queryId)?.queryMetadata.queryType != QueryType.USER_PROVIDED) {
657 1 : state.queriesFinished.delete(queryId);
658 1 : state.queriesFinishedOrdered.splice(i, 1);
659 1 : --internal;
660 1 : } else {
661 1 : ++i;
662 : }
663 1 : }
664 1 : }
665 :
666 1 : function mergeQueryMetrics(metrics: connection.ConnectionQueryMetrics, query: QueryMetrics): connection.ConnectionQueryMetrics {
667 1 : return {
668 1 : totalQueries: metrics.totalQueries + 1,
669 1 : totalBatchesReceived: metrics.totalBatchesReceived + query.streamMetrics.totalBatchesReceived,
670 1 : totalRowsReceived: metrics.totalRowsReceived + query.streamMetrics.totalRowsReceived,
671 1 : accumulatedTimeUntilFirstBatch: metrics.accumulatedTimeUntilFirstBatch + (query.streamMetrics.durationUntilFirstBatchMs ?? 0),
672 1 : accumulatedQueryDuration: metrics.accumulatedQueryDuration + (query.queryDurationMs ?? 0)
673 1 : };
674 1 : }
675 :
676 1 : export function createQueryResponseStreamMetrics(): QueryExecutionMetrics {
677 1 : return {
678 1 : totalDataBytesReceived: 0,
679 1 : totalBatchesReceived: 0,
680 1 : totalRowsReceived: 0,
681 :
682 1 : totalQueryRequestsStarted: 0,
683 1 : totalQueryRequestsSucceeded: 0,
684 1 : totalQueryRequestsFailed: 0,
685 1 : totalQueryRequestDurationMs: 0,
686 :
687 1 : durationUntilFirstBatchMs: null,
688 1 : };
689 1 : }
690 :
691 1 : export function createQueryExecutionMetrics(): connection.ConnectionMetrics {
692 1 : return {
693 1 : successfulQueries: {
694 1 : totalQueries: 0,
695 1 : totalBatchesReceived: 0,
696 1 : totalRowsReceived: 0,
697 1 : accumulatedTimeUntilFirstBatch: 0,
698 1 : accumulatedQueryDuration: 0,
699 1 : },
700 1 : canceledQueries: {
701 1 : totalQueries: 0,
702 1 : totalBatchesReceived: 0,
703 1 : totalRowsReceived: 0,
704 1 : accumulatedTimeUntilFirstBatch: 0,
705 1 : accumulatedQueryDuration: 0,
706 1 : },
707 1 : failedQueries: {
708 1 : totalQueries: 0,
709 1 : totalBatchesReceived: 0,
710 1 : totalRowsReceived: 0,
711 1 : accumulatedTimeUntilFirstBatch: 0,
712 1 : accumulatedQueryDuration: 0,
713 1 : },
714 1 : };
715 1 : }
|
