Line data Source code
1 : import * as core from '@ankoh/dashql-core';
2 :
3 : import { scriptDisplayName } from '../scripts/script_types.js';
4 : import { orderedSessionScripts, sessionAnalysisSignature, type SessionScripts } from '../scripts/session_scripts.js';
5 : import { readSessionInputType, sessionPlanRequest, type SessionInputType } from '../inputs/session_inputs.js';
6 :
7 1 : const StatementType = core.buffers.parser.StatementType;
8 :
9 : /// What carries the data of a dataflow edge
10 1 : export enum DataflowEdgeKind {
11 : /// A table parameter inlines the terminal query of the referenced script
12 1 : Inline = 'inline',
13 : /// A table parameter scans the materialized result of a script of another connection
14 1 : Scan = 'scan',
15 : /// A table parameter with errors
16 1 : Invalid = 'invalid',
17 : /// A statement reads or writes a relation that another statement created or wrote
18 1 : Table = 'table',
19 : /// A statement uses an input, a named scalar parameter
20 1 : Input = 'input',
21 1 : }
22 :
23 : /// A statement of a script on the canvas
24 : export interface DataflowStatement {
25 : statementId: number;
26 : statementType: core.buffers.parser.StatementType;
27 : /// The relation that the statement creates, writes or drops
28 : target: string | null;
29 : /// The text of the statement, in UTF-16 code units of the analyzed text
30 : from: number;
31 : to: number;
32 : /// The label of the statement row: the target, or the first line of the statement
33 : label: string;
34 : }
35 :
36 : /// A script on the canvas
37 : export interface DataflowScript {
38 : scriptKey: number;
39 : fileName: string;
40 : connectionId: string;
41 : statements: DataflowStatement[];
42 : /// Are the statements those of the current text? A script that is being edited keeps its last statements.
43 : current: boolean;
44 : }
45 :
46 : /// An input of the session on the canvas, a named scalar parameter
47 : export interface DataflowInput {
48 : name: string;
49 : type: SessionInputType;
50 : /// The value of the session, null if no script has an `@param` tag of the input
51 : value: string | null;
52 : }
53 :
54 : /// The script key of the source port of an input edge, the input of the edge names its source
55 : export const INPUT_PORT_SCRIPT_KEY = -1;
56 :
57 : /// An end of an edge
58 : export interface DataflowPort {
59 : scriptKey: number;
60 : statementId: number;
61 : }
62 :
63 : /// The flow of data from one statement to another
64 : export interface DataflowEdge {
65 : /// A key that is unique within the graph
66 : key: string;
67 : from: DataflowPort;
68 : to: DataflowPort;
69 : kind: DataflowEdgeKind;
70 : /// The parameter name of a reference, the relation name of a table edge, the name of the input of an input edge
71 : name: string;
72 : /// The input that an input edge starts at, null for the other edges
73 : input: string | null;
74 : }
75 :
76 : /// The dataflow of a session
77 : export interface DataflowGraph {
78 : /// The scripts in session order
79 : scripts: DataflowScript[];
80 : /// The inputs in the order of their first use
81 : inputs: DataflowInput[];
82 : edges: DataflowEdge[];
83 : }
84 :
85 1 : const EMPTY_GRAPH: DataflowGraph = { scripts: [], inputs: [], edges: [] };
86 1 : const NO_STATEMENTS: DataflowStatement[] = [];
87 :
88 : /// The short name of a statement type, e.g. `CREATE` of `CREATE_TABLE_AS`
89 1 : export function statementTypeLabel(type: core.buffers.parser.StatementType): string {
90 1 : switch (type) {
91 1 : case StatementType.SELECT: return 'SELECT';
92 1 : case StatementType.INSERT: return 'INSERT';
93 0 : case StatementType.DELETE: return 'DELETE';
94 1 : case StatementType.CREATE_TABLE:
95 1 : case StatementType.CREATE_TABLE_AS:
96 1 : case StatementType.CREATE_VIEW:
97 1 : case StatementType.SELECT_INTO:
98 1 : return 'CREATE';
99 0 : case StatementType.DROP_TABLE:
100 1 : case StatementType.DROP_VIEW:
101 1 : return 'DROP';
102 1 : default: {
103 1 : const name = StatementType[type] ?? 'SQL';
104 1 : return name.split('_')[0];
105 : }
106 1 : }
107 1 : }
108 :
109 : /// The label of a statement row
110 1 : function statementLabel(text: string, target: string | null): string {
111 1 : if (target != null && target.length > 0) return target;
112 1 : const line = text.split('\n').map(l => l.trim()).find(l => l.length > 0) ?? '';
113 1 : return line;
114 1 : }
115 :
116 : /// The edge kind of a core edge
117 1 : function edgeKind(kind: core.buffers.execution.DataflowEdgeKind): DataflowEdgeKind {
118 1 : switch (kind) {
119 1 : case core.buffers.execution.DataflowEdgeKind.INLINE_REFERENCE: return DataflowEdgeKind.Inline;
120 0 : case core.buffers.execution.DataflowEdgeKind.SCAN_REFERENCE: return DataflowEdgeKind.Scan;
121 0 : case core.buffers.execution.DataflowEdgeKind.INVALID_REFERENCE: return DataflowEdgeKind.Invalid;
122 0 : case core.buffers.execution.DataflowEdgeKind.INPUT: return DataflowEdgeKind.Input;
123 1 : default: return DataflowEdgeKind.Table;
124 1 : }
125 1 : }
126 :
127 1 : function edgeKey(from: DataflowPort, to: DataflowPort, kind: DataflowEdgeKind, name: string): string {
128 1 : if (kind === DataflowEdgeKind.Input) return `:${name}>${to.scriptKey}.${to.statementId}:${kind}`;
129 1 : return `${from.scriptKey}.${from.statementId}>${to.scriptKey}.${to.statementId}:${kind}:${name}`;
130 1 : }
131 :
132 : /// Derive the dataflow of the scripts of a session with the core. A script without a current analysis has no
133 : /// statements and no edges.
134 1 : function computeDataflowGraph(session: SessionScripts): DataflowGraph {
135 1 : const scripts = orderedSessionScripts(session);
136 1 : if (scripts.length === 0) return EMPTY_GRAPH;
137 1 : const dataflow: core.buffers.execution.SessionDataflowT = session.instance.deriveSessionDataflow(
138 1 : sessionPlanRequest(scripts, scripts.map(script => script.fileName), session),
139 1 : ).unpack();
140 1 : const out: DataflowScript[] = scripts.map((script, index) => {
141 1 : const derived = dataflow.scripts[index];
142 1 : const analyzed = derived?.analyzed ?? false;
143 1 : const text = analyzed ? script.scriptSession.getText() : '';
144 1 : const statements = !analyzed ? NO_STATEMENTS : derived.statements.map((statement, statementId) => {
145 1 : const from = statement.textSpan?.offset ?? 0;
146 1 : const to = from + (statement.textSpan?.length ?? 0);
147 1 : const target = statement.target != null ? String(statement.target) : null;
148 1 : return {
149 1 : statementId,
150 1 : statementType: statement.statementType,
151 1 : target,
152 1 : from,
153 1 : to,
154 1 : label: statementLabel(text.slice(from, to), target),
155 1 : };
156 1 : });
157 1 : return {
158 1 : scriptKey: script.scriptKey,
159 1 : fileName: script.fileName,
160 1 : connectionId: script.connection.connectionId,
161 1 : statements,
162 1 : current: analyzed,
163 1 : };
164 1 : });
165 1 : const inputs: DataflowInput[] = dataflow.inputs.map(input => ({
166 0 : name: String(input.name ?? ''),
167 0 : type: readSessionInputType(input.sqlType, input.typeSource),
168 0 : value: input.hasValue ? String(input.value ?? '') : null,
169 1 : }));
170 1 : const edges: DataflowEdge[] = dataflow.edges.flatMap(edge => {
171 : // The canvas draws no attachments yet
172 1 : if (edge.kind === core.buffers.execution.DataflowEdgeKind.ATTACHMENT) return [];
173 1 : const kind = edgeKind(edge.kind);
174 1 : const toScript = scripts[edge.toScript];
175 1 : const source = kind === DataflowEdgeKind.Input ? inputs[edge.fromScript] : null;
176 1 : const fromScript = kind === DataflowEdgeKind.Input ? null : scripts[edge.fromScript];
177 1 : if (toScript == null || (source == null && fromScript == null)) return [];
178 1 : const from = fromScript != null
179 1 : ? { scriptKey: fromScript.scriptKey, statementId: edge.fromStatement }
180 0 : : { scriptKey: INPUT_PORT_SCRIPT_KEY, statementId: 0 };
181 1 : const to = { scriptKey: toScript.scriptKey, statementId: edge.toStatement };
182 1 : const name = String(edge.name ?? '');
183 1 : return [{ key: edgeKey(from, to, kind, name), from, to, kind, name, input: source?.name ?? null }];
184 1 : });
185 1 : return { scripts: out, inputs, edges };
186 1 : }
187 :
188 1 : function sameStatements(l: DataflowStatement[], r: DataflowStatement[]): boolean {
189 1 : return l.length === r.length && l.every((s, i) => {
190 1 : const o = r[i];
191 1 : return s.statementType === o.statementType && s.target === o.target && s.from === o.from && s.to === o.to && s.label === o.label;
192 1 : });
193 1 : }
194 :
195 1 : function sameScript(l: DataflowScript, r: DataflowScript): boolean {
196 1 : return l.scriptKey === r.scriptKey && l.fileName === r.fileName && l.connectionId === r.connectionId
197 1 : && l.current === r.current && sameStatements(l.statements, r.statements);
198 1 : }
199 :
200 : /// Merge a new graph into the previous one. A script without a current analysis keeps its last statements and the
201 : /// edges that start or end at it, as long as the other end still exists, until it is analyzed again. Scripts and
202 : /// edges that did not change keep their identity, as does the graph if nothing changed.
203 1 : export function mergeDataflowGraph(previous: DataflowGraph | null, next: DataflowGraph): DataflowGraph {
204 1 : if (previous == null) return next;
205 1 : const before = new Map(previous.scripts.map(s => [s.scriptKey, s]));
206 1 : const scripts = next.scripts.map(script => {
207 1 : const last = before.get(script.scriptKey);
208 1 : if (last == null) return script;
209 1 : const merged = !script.current && last.statements.length > 0
210 1 : ? { ...script, statements: last.statements, current: false }
211 1 : : script;
212 1 : return sameScript(merged, last) ? last : merged;
213 1 : });
214 1 : const stale = new Set(scripts.filter(s => !s.current && s.statements.length > 0).map(s => s.scriptKey));
215 1 : const present = new Map(scripts.map(s => [s.scriptKey, s]));
216 1 : const exists = (port: DataflowPort) => (present.get(port.scriptKey)?.statements.length ?? 0) > port.statementId;
217 1 : const edgesBefore = new Map(previous.edges.map(e => [e.key, e]));
218 1 : const edges: DataflowEdge[] = next.edges.map(e => edgesBefore.get(e.key) ?? e);
219 1 : const keys = new Set(edges.map(e => e.key));
220 1 : for (const edge of previous.edges) {
221 1 : if (keys.has(edge.key)) continue;
222 1 : if ((stale.has(edge.from.scriptKey) || stale.has(edge.to.scriptKey)) && (edge.input != null || exists(edge.from)) && exists(edge.to)) {
223 1 : edges.push(edge);
224 1 : keys.add(edge.key);
225 1 : }
226 1 : }
227 : // The inputs of a script that is being edited stay with its edges
228 1 : const inputsBefore = new Map(previous.inputs.map(i => [i.name, i]));
229 1 : const inputs: DataflowInput[] = next.inputs.map(input => {
230 0 : const last = inputsBefore.get(input.name);
231 0 : return last != null && JSON.stringify(last) === JSON.stringify(input) ? last : input;
232 1 : });
233 1 : const names = new Set(inputs.map(i => i.name));
234 1 : for (const edge of edges) {
235 1 : if (edge.input != null && !names.has(edge.input)) {
236 0 : const last = inputsBefore.get(edge.input);
237 0 : if (last != null) {
238 0 : inputs.push(last);
239 0 : names.add(last.name);
240 0 : }
241 0 : }
242 : }
243 1 : const unchanged = scripts.length === previous.scripts.length && scripts.every((s, i) => s === previous.scripts[i])
244 1 : && inputs.length === previous.inputs.length && inputs.every((input, i) => input === previous.inputs[i])
245 1 : && edges.length === previous.edges.length && edges.every(e => edgesBefore.get(e.key) === e);
246 1 : return unchanged ? previous : { scripts, inputs, edges };
247 1 : }
248 :
249 : interface CachedGraph {
250 : signature: string;
251 : graph: DataflowGraph;
252 : }
253 :
254 : /// The last dataflow of every session, by session id
255 1 : const GRAPHS = new Map<string, CachedGraph>();
256 :
257 : /// The dataflow of the scripts of a session. Derives it again only when a name, a connection or an analysis of a
258 : /// script changed, see `mergeDataflowGraph` for what stays.
259 1 : export function deriveDataflowGraph(session: SessionScripts): DataflowGraph {
260 : // Sessions without the core, e.g. of tests, have no dataflow
261 1 : if (session.scriptOrder == null || session.scriptOrder.length === 0 || typeof session.instance?.deriveSessionDataflow !== 'function') {
262 0 : return EMPTY_GRAPH;
263 : }
264 1 : const signature = sessionAnalysisSignature(session);
265 1 : const cached = GRAPHS.get(session.sessionId);
266 1 : if (cached?.signature === signature) return cached.graph;
267 1 : let graph: DataflowGraph;
268 1 : try {
269 1 : graph = mergeDataflowGraph(cached?.graph ?? null, computeDataflowGraph(session));
270 1 : } catch (e) {
271 0 : console.warn('cannot derive the dataflow of the session', e);
272 0 : return cached?.graph ?? EMPTY_GRAPH;
273 0 : }
274 1 : GRAPHS.set(session.sessionId, { signature, graph });
275 1 : return graph;
276 1 : }
277 :
278 : /// Forget the dataflow of a session
279 1 : export function forgetDataflowGraph(sessionId: string): void {
280 1 : GRAPHS.delete(sessionId);
281 1 : }
|
