Line data Source code
1 : import type * as arrow from 'apache-arrow';
2 : import type * as core from '@ankoh/dashql-core';
3 :
4 : import type { QueryExecutor } from '../../connections/query_executor.js';
5 : import type { QueryResultCache } from '../../query/query_result_cache.js';
6 : import { QueryType } from '../../query/query_execution_state.js';
7 : import {
8 : ScriptPlanError,
9 : acquireSessionExecution,
10 : computeScriptResultKeys,
11 : createMountPrefix,
12 : planSessionScripts,
13 : releaseSessionExecution,
14 : runSessionPlan,
15 : type ExecutionMounts,
16 : type RunQueries,
17 : } from './session_execution.js';
18 : import { REGISTER_QUERY, scriptResultOwner, type ScriptData, type SessionScripts } from './session_scripts.js';
19 : import type { ModifySessionScripts } from './session_scripts_registry.js';
20 : import { yieldToBrowser } from '../../utils/yield_to_browser.js';
21 : import type { LoggerLike } from '../../platform/logger/logger.js';
22 :
23 : /// The services of the run of a script
24 : export interface ScriptRunServices {
25 : /// The mounts of the materialized results, null without a compute database
26 : createMounts: () => ExecutionMounts | null;
27 : /// Mount the attachments of a session in the embedded Hyper, absent without a compute database
28 : mountAttachments?: (sessionId: string) => Promise<void>;
29 : /// The result key of a script on a connection, null if the connection has no key
30 : cacheKey: (connectionId: string, signature: string) => Promise<string | null>;
31 : /// Does a connection cache query results? Runs reuse the cached results of its scripts only then.
32 : cachesResults: (connectionId: string) => boolean;
33 : /// The query result cache of the app, null if results aren't cached
34 : cache: QueryResultCache | null;
35 : /// The maximum number of statements a connection runs at once
36 : maxConcurrentQueries: (connectionId: string) => number;
37 : /// The live session, a run fails if its scripts, their order or connections change meanwhile. Without it, a run
38 : /// checks only the text and catalog versions of its scripts.
39 : liveSession?: (sessionId: string) => SessionScripts | null;
40 : }
41 :
42 : /// The services of a run without a session execution provider: no results of other connections are mounted and no
43 : /// results are cached
44 1 : const PLAIN_SERVICES: ScriptRunServices = {
45 1 : createMounts: () => null,
46 1 : cacheKey: async () => null,
47 1 : cachesResults: () => false,
48 1 : cache: null,
49 1 : maxConcurrentQueries: () => 1,
50 1 : };
51 :
52 : /// The scripts of a session in session order and the index of a script
53 1 : function sessionScriptsOf(session: SessionScripts, target: ScriptData) {
54 1 : const files = [...session.scriptOrder];
55 1 : const scripts = files.map(file => session.scripts[session.scriptRefs[file].scriptId]);
56 1 : const targetIndex = scripts.findIndex(script => script.scriptKey === target.scriptKey);
57 1 : if (targetIndex < 0) throw new Error(`${target.fileName} is no script of the session`);
58 1 : return { files, scripts, targetIndex };
59 1 : }
60 :
61 : /// The options of the run of a script
62 : export interface ScriptRunOptions {
63 : /// Read no cached results: the script and the scripts it references run, their results with reusable keys become
64 : /// the newest files of their keys
65 : refresh?: boolean;
66 : /// The caller holds the session already (`waitForSessionExecution`), the run releases it
67 : held?: boolean;
68 : /// Logs the plan, its statements and the cached results it reads
69 : logger?: LoggerLike;
70 : }
71 :
72 : /// Run a script: plan it with the scripts it needs and run the plan (`runSessionPlan`). Resolves with the last output
73 : /// of the script, rejects if the plan is invalid, a statement fails or another run holds the session. A failed
74 : /// statement cancels the other statements of the run.
75 : ///
76 : /// The plan of the script includes the statements before the terminal queries of the scripts it inlines and the
77 : /// scripts it materializes (docs/design/script_references.md). The result of the script is cached if its key is
78 : /// reusable, the key covers the referenced scripts. Like in a run of the session, a script with a reusable cached
79 : /// result doesn't run if nothing that it follows runs.
80 1 : export async function runPlannedScript(
81 1 : services: ScriptRunServices,
82 1 : session: SessionScripts,
83 1 : target: ScriptData,
84 1 : executeQuery: QueryExecutor,
85 1 : modifySessionScripts: ModifySessionScripts,
86 1 : options: ScriptRunOptions = {},
87 1 : ): Promise<arrow.Table | null> {
88 : // The app disables the runs of a held session, a run that starts meanwhile fails
89 1 : if (!options.held && !acquireSessionExecution(session)) throw new Error('Another script of the session is running');
90 1 : const { sessionId } = session;
91 : // Aborted once a statement failed, cancels the other statements
92 1 : const cancellation = new AbortController();
93 1 : const queries: RunQueries = { active: new Map(), cancellations: new Map() };
94 : // Did a statement start? An error before shows where the result of the script would be.
95 1 : let launched = false;
96 : // The query of the last output of the target, its result is the result of the run
97 1 : let result: Promise<arrow.Table | null> | null = null;
98 1 : const executeNode: QueryExecutor = (connectionId, args) => {
99 1 : launched = true;
100 1 : const [queryId, execution] = executeQuery(connectionId, args);
101 1 : if (args.metadata.resultOwner === scriptResultOwner(sessionId, target.scriptKey) && args.analyzeResults) result = execution;
102 1 : return [queryId, execution];
103 1 : };
104 1 : try {
105 1 : const { files, targetIndex } = sessionScriptsOf(session, target);
106 1 : const liveSession = services.liveSession;
107 1 : const outputs = await runSessionPlan({
108 1 : session, getSession: liveSession != null ? () => liveSession(sessionId) : () => session,
109 1 : targets: [targetIndex], run: { kind: 'script', target: files[targetIndex] },
110 1 : executeQuery: executeNode, modifySessionScripts, signal: cancellation.signal, queries,
111 1 : cancelActive: () => cancellation.abort(), cooperate: () => yieldToBrowser(), emit: () => {},
112 1 : mounts: services.createMounts(), mountAttachments: services.mountAttachments,
113 1 : resultKey: services.cacheKey, cachesResults: services.cachesResults,
114 1 : cache: services.cache, refresh: options.refresh ?? false,
115 1 : maxConcurrentQueries: services.maxConcurrentQueries, logger: options.logger,
116 1 : });
117 1 : if (outputs[targetIndex] != null) {
118 1 : await modifySessionScripts({ type: REGISTER_QUERY, value: [target.scriptKey, outputs[targetIndex]!] });
119 : }
120 1 : return await (result ?? Promise.resolve(null));
121 1 : } catch (error) {
122 : // The script shows why it can't be planned or run where it shows the error of a failed statement, e.g. a
123 : // script of the session that can't be analyzed or a plan that needs the compute database
124 1 : if (error instanceof ScriptPlanError || !launched) {
125 1 : await recordPlanError(session, target, executeQuery, modifySessionScripts, error instanceof Error ? error : new Error(String(error)));
126 : }
127 1 : throw error;
128 1 : } finally {
129 1 : releaseSessionExecution(session);
130 1 : }
131 1 : }
132 :
133 : /// Record an error of a run before its statements as a failed query of the script
134 1 : async function recordPlanError(
135 1 : session: SessionScripts,
136 1 : target: ScriptData,
137 1 : executeQuery: QueryExecutor,
138 1 : modifySessionScripts: ModifySessionScripts,
139 1 : error: Error,
140 1 : ): Promise<void> {
141 1 : const [queryId, execution] = executeQuery(target.connection.connectionId, {
142 1 : query: target.scriptSession.getText(),
143 1 : failure: error,
144 1 : metadata: {
145 1 : queryType: QueryType.USER_PROVIDED, title: 'Session Query', description: null, issuer: 'Query Rerun',
146 1 : userProvided: true, sessionId: session.sessionId, resultOwner: scriptResultOwner(session.sessionId, target.scriptKey),
147 1 : },
148 1 : });
149 1 : await modifySessionScripts({ type: REGISTER_QUERY, value: [target.scriptKey, queryId] });
150 1 : await execution.catch(() => null);
151 1 : }
152 :
153 : /// The cache key of the result of a script, the key that a run of the script reads and writes. Null if the result
154 : /// is not reusable or the current analysis of the scripts it needs can't be planned. Plans without analyzing.
155 1 : export async function computeScriptResultCacheKey(
156 1 : services: ScriptRunServices,
157 1 : session: SessionScripts,
158 1 : target: ScriptData,
159 1 : ): Promise<string | null> {
160 1 : const { files, scripts, targetIndex } = sessionScriptsOf(session, target);
161 1 : const { graph } = planSessionScripts({
162 1 : instance: session.instance, scripts, files,
163 1 : mountPrefix: createMountPrefix(), targets: [targetIndex], session,
164 1 : });
165 1 : if (!graph.valid) return null;
166 1 : const connectionOf = (index: number) => scripts[index].connection.connectionId;
167 1 : const keys = await computeScriptResultKeys(graph, (index, signature) => services.cacheKey(connectionOf(index), signature), index => services.cachesResults(connectionOf(index)));
168 1 : const key = keys[targetIndex];
169 1 : return key?.reusable ? key.key : null;
170 1 : }
171 :
172 : /// The services of script runs, one per core instance, registered by the session execution provider
173 1 : const runners = new WeakMap<core.DashQL, ScriptRunServices>();
174 :
175 : /// Register the services of the script runs of a core instance
176 1 : export function registerScriptRunServices(instance: core.DashQL, services: ScriptRunServices | null): void {
177 1 : if (services == null) runners.delete(instance);
178 1 : else runners.set(instance, services);
179 1 : }
180 :
181 : /// The services of the script runs of a core instance. Without a session execution provider, runs neither mount the
182 : /// results of other connections nor cache results.
183 1 : export function scriptRunServices(instance: core.DashQL): ScriptRunServices {
184 1 : return runners.get(instance) ?? PLAIN_SERVICES;
185 1 : }
|
