Line data Source code
1 : //! The planner.
2 : //!
3 : //! Plans relation-level hazards between the statements of several scripts: every statement becomes
4 : //! a node with the relations it reads, writes or defines, and edges order conflicting accesses.
5 : //! Opaque SELECT functions may have side effects; barriers order statements but do not promise
6 : //! backend session affinity.
7 : //!
8 : //! Scripts run on connections, every connection has its own catalog. Relations of different connections
9 : //! never conflict. A script references the result of another script with a table parameter, `:foo` reads
10 : //! the terminal query of `foo.sql` (see `script_references`). A reference within one connection inlines the
11 : //! query into the statement as a common table expression, a reference of the LOCAL connection to another
12 : //! connection scans the result as Parquet file that the host materializes, a reference of a REMOTE
13 : //! connection to another connection is an error.
14 : //!
15 : //! A plan covers the whole session or the target scripts and what they need: the statements that an inlined
16 : //! reference needs before it, and the scripts that a scan reference scans unless the host provides
17 : //! their results. A plan of the session leaves out the scripts whose results the host provides. A plan of targets treats statements like a run of a single script: statements of every
18 : //! kind keep their SQL, and the statements of a script run in their order.
19 :
20 : use crate::catalog_changes::{derive_catalog_change, CatalogChangeOptions};
21 : use crate::compiler::{compile_visualization, produces_output, read_executable_text, read_executable_text_with};
22 : use crate::script_references::{self, ReferenceErrors, ReferenceKind, ScriptReferences};
23 : use crate::session_inputs::{bind_inputs, collect_session_inputs, SessionInputs};
24 : use dashql_buffers::analyzer::AnalyzerErrorSeverity;
25 : use dashql_buffers::formatting::FormattingDialect;
26 : use dashql_buffers::execution::{
27 : ExecutionGraphAccessMode as Mode, ExecutionGraphAccessT, ExecutionGraphCatalogObjectIDT, ExecutionGraphConnectionKind,
28 : ExecutionGraphConnectionT, ExecutionGraphDiagnosticT, ExecutionGraphEdgeReason as Reason, ExecutionGraphEdgeT, ExecutionGraphEntityT,
29 : ExecutionGraphAttachmentT, ExecutionGraphEntityType, ExecutionGraphMaterializationT, ExecutionGraphNodeT, ExecutionGraphScriptReferenceT, ExecutionGraphScriptT,
30 : ExecutionGraphT, ExecutionGraphUnresolvedReferenceT,
31 : };
32 : use dashql_buffers::parser::{AttributeKey as Key, Node, NodeType, StatementType as Type, TempType, TextSpanT};
33 : use dashql_core::analyzer::{AnalyzedScript, ReferencedTable, RelationExpression, TableRole};
34 : use dashql_core::catalog::{Catalog, CatalogEntryId, CatalogEntryRef, QualifiedCatalogObjectId, QualifiedTableName, TableDeclaration};
35 : use dashql_core::pack;
36 : use dashql_core::script::Script;
37 : use dashql_parser::attrs;
38 : use dashql_parser::parser::context::resolve_text_span;
39 : use dashql_parser::parser::ParsedScript;
40 : use dashql_parser::scanner::ScannedScript;
41 : use dashql_parser::span::TextSpan;
42 : use std::collections::{BTreeMap, BTreeSet};
43 : use std::sync::Arc;
44 :
45 : /// A relation name: (database, schema, table)
46 : type Name = (String, String, String);
47 :
48 : const NONE: u32 = u32::MAX;
49 :
50 2571 : fn resource_name(name: &QualifiedTableName) -> Name {
51 2571 : (name.database_name.text.to_string(), name.schema_name.text.to_string(), name.table_name.text.to_string())
52 2571 : }
53 :
54 : /// The table declarations of an entry with their AST statement ids
55 1066 : fn tables(entry: &CatalogEntryRef) -> &[TableDeclaration] {
56 1066 : &entry.data().table_declarations
57 1066 : }
58 :
59 : /// The kind of a connection
60 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
61 : pub enum ConnectionKind {
62 : /// A database that only reads its own relations
63 : Remote,
64 : /// The embedded Hyper, which scans the materialized results of scripts of other connections
65 : Local,
66 : }
67 :
68 : /// A connection of the plan
69 : #[derive(Clone, Copy)]
70 : pub struct PlannerConnection<'a> {
71 : /// The connection id of the host
72 : pub id: &'a str,
73 : /// The kind of the connection
74 : pub kind: ConnectionKind,
75 : /// The catalog of the connection, its scripts were analyzed with it
76 : pub catalog: &'a Catalog,
77 : /// The SQL dialect of the connection, the planner prints the values of inputs in it
78 : pub dialect: FormattingDialect,
79 : }
80 :
81 : /// A script of the plan
82 : #[derive(Clone, Copy)]
83 : pub struct PlannerScript<'a> {
84 : /// The script
85 : pub script: &'a Script,
86 : /// The name of the script, `foo` for `foo.sql`, that table parameters reference
87 : pub name: &'a str,
88 : /// The connection index
89 : pub connection: u32,
90 : /// The name of the Parquet file of the result of the script below the mount prefix, if a statement of another
91 : /// connection scans it. `s<index>.parquet` if absent.
92 : pub materialization_file: Option<&'a str>,
93 : }
94 :
95 : /// An attachment of the session, a Parquet file that the table parameter `:name` reads
96 : #[derive(Clone, Copy)]
97 : pub struct PlannerAttachment<'a> {
98 : /// The name, `foo` for `foo.parquet`
99 : pub name: &'a str,
100 : /// The file name below the attachment prefix
101 : pub file: &'a str,
102 : /// Changes when the file changes, e.g. its size and modification time. The signature of a script that reads the
103 : /// attachment covers it.
104 : pub stamp: &'a str,
105 : }
106 :
107 : /// What a plan covers
108 : #[derive(Clone, Copy)]
109 : pub enum PlanScope<'a> {
110 : /// Every statement of every script whose result the host doesn't provide, in session order
111 : Session,
112 : /// The target scripts and what they need
113 : Targets(&'a [u32]),
114 : /// Every statement of every script whose analysis is current with its text, for the dataflow of a session. Like a
115 : /// plan of targets, but problems are diagnostics that don't stop the plan: a script without a current analysis is
116 : /// left out, as are script references with errors or to scripts that were left out. An analysis of an earlier
117 : /// catalog revision is planned as it is. A plan that left something out is not valid.
118 : Tolerant,
119 : }
120 :
121 : /// A plan request
122 : #[derive(Clone, Copy)]
123 : pub struct SessionPlanRequest<'a> {
124 : /// The connections
125 : pub connections: &'a [PlannerConnection<'a>],
126 : /// The scripts in session order
127 : pub scripts: &'a [PlannerScript<'a>],
128 : /// What the plan covers
129 : pub scope: PlanScope<'a>,
130 : /// The scripts whose materialized results the host provides. A plan computes them only for its targets, a scan
131 : /// reads the file of the result.
132 : pub materialized: &'a [u32],
133 : /// The directory that the host mounts materialized results in, e.g. the query result cache at `/mnt/external/cache`
134 : pub mount_prefix: &'a str,
135 : /// The attachments of the session
136 : pub attachments: &'a [PlannerAttachment<'a>],
137 : /// The directory that the host mounts the attachments in, e.g. `/mnt/external/attachments/<session>`
138 : pub attachment_prefix: &'a str,
139 : }
140 :
141 : /// A script with a current analysis
142 : pub(crate) struct Prepared<'a> {
143 : pub(crate) scanned: &'a ScannedScript,
144 : pub(crate) parsed: &'a ParsedScript,
145 : pub(crate) analyzed: &'a Arc<AnalyzedScript>,
146 : }
147 :
148 : impl<'a> Prepared<'a> {
149 : /// The analysis of a script if it is current with the text, whatever catalog it was analyzed with. Script
150 : /// references depend on the text only.
151 3375 : pub(crate) fn lenient(script: &'a Script) -> Option<Self> {
152 3375 : script.current_analysis().map(|(scanned, parsed, analyzed)| Prepared { scanned, parsed, analyzed })
153 3375 : }
154 :
155 : /// Does the script parse without errors?
156 1519 : pub(crate) fn parses(&self) -> bool {
157 1519 : self.scanned.errors.is_empty() && self.parsed.errors.is_empty()
158 1519 : }
159 :
160 : /// The last statement
161 3163 : pub(crate) fn terminal(&self) -> Option<u32> {
162 3163 : self.parsed.statements.len().checked_sub(1).map(|s| s as u32)
163 3163 : }
164 : }
165 :
166 : /// How much of a script a plan includes
167 : #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
168 : enum Include {
169 : /// Nothing
170 : None,
171 : /// No statement, the single query of the script is inlined into its consumers, its references are planned
172 : Inlined,
173 : /// Every statement
174 : All,
175 : }
176 :
177 : /// The state of the plan
178 : struct PlanState {
179 : /// The execution graph, completed with the lists below when finished
180 : graph: ExecutionGraphT,
181 : /// The entity indexes by connection and table object id
182 : entities: BTreeMap<(u32, QualifiedCatalogObjectId), u32>,
183 : /// The unresolved reference indexes by connection and name
184 : unresolved_references: BTreeMap<(u32, Name), u32>,
185 : /// The interned table entities, in interning order
186 : entity_list: Vec<ExecutionGraphEntityT>,
187 : /// The interned unresolved references, in interning order
188 : unresolved_list: Vec<ExecutionGraphUnresolvedReferenceT>,
189 : /// The diagnostics of the plan
190 : diagnostics: Vec<ExecutionGraphDiagnosticT>,
191 : /// Do errors leave something out of the plan instead of stopping it? See `PlanScope::Tolerant`.
192 : tolerant: bool,
193 : /// Did a tolerant plan leave something out?
194 : partial: bool,
195 : /// The inputs of the session, the scalar parameters that the statements bind
196 : inputs: SessionInputs,
197 : /// The text span of the statement of every node
198 : spans: Vec<TextSpan>,
199 : /// The relations that a node reads only through the queries of inlined scripts: (node, entity, unresolved)
200 : inherited: BTreeSet<(u32, u32, u32)>,
201 : }
202 :
203 : /// A plan and what the planner derived on the way, the dataflow of a session projects it
204 : pub(crate) struct Plan {
205 : /// The execution graph
206 : pub(crate) graph: ExecutionGraphT,
207 : /// The script references, an invalid one keeps the script that its name finds
208 : pub(crate) references: ScriptReferences,
209 : /// The inputs of the session
210 : pub(crate) inputs: SessionInputs,
211 : /// The text span of the statement of every node
212 : pub(crate) spans: Vec<TextSpan>,
213 : /// The relations that a node reads only through the queries of inlined scripts: (node, entity, unresolved)
214 : pub(crate) inherited: BTreeSet<(u32, u32, u32)>,
215 : }
216 :
217 : impl PlanState {
218 : /// Report a diagnostic, an error invalidates the plan
219 274 : fn report(&mut self, node: u32, script: u32, text_span: Option<TextSpan>, message: String, error: bool) {
220 274 : let mut d = ExecutionGraphDiagnosticT::default();
221 274 : d.node = node;
222 274 : d.script = script;
223 274 : d.message = opt(Some(message));
224 274 : d.text_span = text_span.map(|s| {
225 98 : let mut t = TextSpanT::default();
226 98 : t.offset = s.offset;
227 98 : t.length = s.length;
228 98 : t
229 98 : });
230 274 : d.error = error;
231 274 : self.diagnostics.push(d);
232 274 : if error && self.tolerant {
233 13 : self.partial = true;
234 265 : } else if error {
235 132 : self.graph.valid = false;
236 138 : }
237 274 : }
238 :
239 : /// Report a diagnostic of a statement. It is a warning in every scope: the statement runs as written, a barrier
240 : /// orders it against the other statements of its connection.
241 129 : fn diagnose(&mut self, node: u32, script: u32, message: String) {
242 129 : self.report(node, script, None, message, false);
243 129 : }
244 :
245 917 : fn intern_entity(&mut self, connection: u32, table: &TableDeclaration) -> u32 {
246 917 : if let Some(&i) = self.entities.get(&(connection, table.object_id)) {
247 466 : return i;
248 451 : }
249 451 : let i = self.entity_list.len() as u32;
250 451 : self.entities.insert((connection, table.object_id), i);
251 451 : let mut entity = ExecutionGraphEntityT::default();
252 451 : let mut id = ExecutionGraphCatalogObjectIDT::default();
253 451 : id.object_type = ExecutionGraphEntityType::TABLE;
254 451 : id.object_id = table.table_id().pack();
255 451 : id.sub_object_id = 0;
256 451 : entity.catalog_object_id = id;
257 451 : (entity.catalog_database_id, entity.catalog_schema_id) = table.catalog_schema_id.unpack_schema_id();
258 451 : let (db, schema, name) = resource_name(&table.table_name);
259 451 : entity.database_name = opt(Some(db));
260 451 : entity.schema_name = opt(Some(schema));
261 451 : entity.name = opt(Some(name));
262 451 : entity.connection = connection;
263 451 : self.entity_list.push(entity);
264 451 : i
265 917 : }
266 :
267 42 : fn intern_unresolved(&mut self, connection: u32, name: &Name) -> u32 {
268 42 : if let Some(&i) = self.unresolved_references.get(&(connection, name.clone())) {
269 9 : return i;
270 33 : }
271 33 : let i = self.unresolved_list.len() as u32;
272 33 : self.unresolved_references.insert((connection, name.clone()), i);
273 33 : let mut r = ExecutionGraphUnresolvedReferenceT::default();
274 33 : r.database_name = opt(Some(name.0.clone()));
275 33 : r.schema_name = opt(Some(name.1.clone()));
276 33 : r.name = opt(Some(name.2.clone()));
277 33 : r.connection = connection;
278 33 : self.unresolved_list.push(r);
279 33 : i
280 42 : }
281 : }
282 :
283 : /// The relations of a connection: the declarations of its catalog and of the planned scripts, and the live
284 : /// namespace while the statements are planned in order
285 : #[derive(Default)]
286 : struct ConnectionState<'a> {
287 : /// The node of the statement that declares a table, by packed table id
288 : producers: BTreeMap<u64, u32>,
289 : /// The declarations by object id
290 : declarations: BTreeMap<QualifiedCatalogObjectId, &'a TableDeclaration>,
291 : /// The kinds of the published declarations
292 : published_views: BTreeMap<QualifiedCatalogObjectId, bool>,
293 : /// Declarations that are a view in one place and a table in another
294 : incompatible_kinds: BTreeSet<QualifiedCatalogObjectId>,
295 : /// The views by packed table id
296 : views: BTreeSet<u64>,
297 : /// The relations that a planned statement created
298 : live_producers: BTreeSet<Name>,
299 : /// The relations that a planned statement dropped
300 : dropped_resources: BTreeSet<Name>,
301 : /// The views that a planned statement created
302 : live_views: BTreeSet<Name>,
303 : /// The declarations by name, as the planned statements see them
304 : live_namespace: BTreeMap<Name, &'a TableDeclaration>,
305 : }
306 :
307 : impl<'a> ConnectionState<'a> {
308 : /// The name of a relation and the declaration that it reads, None if it is unresolved or its declaration is
309 : /// incompatible or unavailable. Analysis can bind an earlier access to a later declaration of the same name, the
310 : /// live lifetime still owns that access.
311 766 : fn resolve(&self, relation: &RelationExpression) -> (Name, Option<&'a TableDeclaration>) {
312 766 : let name = resource_name(relation.resolved_table.as_ref().map(|t| &t.table_name).unwrap_or(&relation.table_name));
313 766 : let Some(resolved) = &relation.resolved_table else { return (name, None) };
314 731 : let declaration = self.declarations.get(&resolved.catalog_table_id).copied().filter(|d| {
315 731 : d.catalog_schema_id == resolved.catalog_schema_id && resource_name(&d.table_name) == name && !self.incompatible_kinds.contains(&resolved.catalog_table_id)
316 731 : });
317 731 : let owned = self.live_producers.contains(&name) || self.producers.contains_key(&resolved.catalog_table_id.unpack_table_id().pack());
318 731 : let table = declaration.map(|d| self.live_namespace.get(&name).copied().filter(|_| owned).unwrap_or(d));
319 731 : (name, table)
320 766 : }
321 : }
322 :
323 : /// Validate the analysis of a script against the catalog of its connection
324 928 : fn prepare<'a>(state: &mut PlanState, index: u32, script: &'a Script, catalog: &Catalog) -> Option<Prepared<'a>> {
325 928 : let Some(prepared) = Prepared::lenient(script) else {
326 12 : state.report(NONE, index, None, "missing or stale script analysis".into(), true);
327 12 : return None;
328 : };
329 916 : let analyzed = prepared.analyzed;
330 916 : if analyzed.snapshot.id() != catalog.id() {
331 1 : state.report(NONE, index, None, "script was analyzed with another catalog than the one of its connection".into(), true);
332 1 : return None;
333 915 : }
334 : // A tolerant plan takes the analysis of an earlier catalog revision, its declarations keep their ids
335 915 : if !catalog.is_current(&analyzed.snapshot) && !state.tolerant {
336 1 : state.report(NONE, index, None, "stale catalog analysis".into(), true);
337 1 : return None;
338 914 : }
339 914 : Some(prepared)
340 928 : }
341 :
342 : /// Report the syntax errors of a script, they invalidate a plan of the session
343 710 : fn diagnose_syntax(state: &mut PlanState, index: u32, prepared: &Prepared) {
344 710 : let (scanned, parsed) = (prepared.scanned, prepared.parsed);
345 710 : for e in &scanned.errors {
346 3 : state.report(NONE, index, None, format!("scanner: {}", e.message), true);
347 3 : }
348 710 : for e in &parsed.errors {
349 9 : state.report(NONE, index, None, format!("parser: {}", e.message), true);
350 9 : }
351 710 : if parsed.statements.is_empty() {
352 18 : let text = scanned.text.as_bytes();
353 18 : let mut offset = 0;
354 30 : while offset < text.len() {
355 24 : let comment = scanned.comments.iter().find(|c| c.offset as usize <= offset && offset < (c.offset + c.length) as usize);
356 24 : if let Some(c) = comment {
357 9 : offset = (c.offset + c.length) as usize;
358 9 : continue;
359 15 : }
360 15 : if !b" \t\r\n\x0c;".contains(&text[offset]) {
361 12 : state.report(NONE, index, None, "unsupported SQL: parser produced no statements".into(), true);
362 12 : break;
363 3 : }
364 3 : offset += 1;
365 : }
366 692 : }
367 710 : }
368 :
369 : /// Does a script run only on request? Its current analysis has a `@manual` tag.
370 1640 : pub(crate) fn runs_on_request(script: &Script) -> bool {
371 1640 : script.current_analysis().is_some_and(|(_, _, analyzed)| analyzed.manual)
372 1640 : }
373 :
374 : /// The scripts whose statements a plan includes in full before their references: the targets, or the scripts of the
375 : /// session whose results the host doesn't provide and that don't run only on request
376 956 : fn seeds(request: &SessionPlanRequest) -> Vec<u32> {
377 956 : match request.scope {
378 764 : PlanScope::Session => (0..request.scripts.len() as u32)
379 1446 : .filter(|&i| !request.materialized.contains(&i) && !runs_on_request(request.scripts[i as usize].script))
380 764 : .collect(),
381 96 : PlanScope::Targets(targets) => targets.to_vec(),
382 96 : PlanScope::Tolerant => (0..request.scripts.len() as u32).collect(),
383 : }
384 956 : }
385 :
386 : /// Plan the statements of the scripts of a session. Consumes the current analysis without analyzing, publishing or
387 : /// changing scripts.
388 437 : pub fn plan_session(request: &SessionPlanRequest) -> ExecutionGraphT {
389 437 : plan(request).graph
390 437 : }
391 :
392 : /// Plan the statements of the scripts of a session, see `plan_session`
393 478 : pub(crate) fn plan(request: &SessionPlanRequest) -> Plan {
394 : // A plan of targets, and a tolerant plan like one, prepares what its targets need, runs the statements of a script
395 : // in order and passes a script that doesn't parse through as written
396 478 : let targeted = matches!(request.scope, PlanScope::Targets(_) | PlanScope::Tolerant);
397 478 : let tolerant = matches!(request.scope, PlanScope::Tolerant);
398 478 : let mut state = PlanState {
399 478 : graph: ExecutionGraphT::default(),
400 478 : entities: BTreeMap::new(),
401 478 : unresolved_references: BTreeMap::new(),
402 478 : entity_list: Vec::new(),
403 478 : unresolved_list: Vec::new(),
404 478 : diagnostics: Vec::new(),
405 478 : tolerant,
406 478 : partial: false,
407 478 : inputs: collect_session_inputs(request),
408 478 : spans: Vec::new(),
409 478 : inherited: BTreeSet::new(),
410 478 : };
411 478 : state.graph.valid = true;
412 478 : let script_count = request.scripts.len();
413 :
414 : // The request itself
415 478 : let mut catalogs: BTreeSet<u64> = BTreeSet::new();
416 549 : for connection in request.connections {
417 549 : if !catalogs.insert(connection.catalog.id()) {
418 1 : state.report(NONE, NONE, None, format!("connections share a catalog: {}", connection.id), true);
419 548 : }
420 : }
421 : // Catalog entry ids are unique within a catalog
422 478 : let mut script_ids: BTreeSet<(u32, CatalogEntryId)> = BTreeSet::new();
423 478 : let mut names: BTreeSet<&str> = BTreeSet::new();
424 951 : for (i, s) in request.scripts.iter().enumerate() {
425 951 : if s.connection as usize >= request.connections.len() {
426 1 : state.report(NONE, i as u32, None, "unknown planner connection".into(), true);
427 950 : }
428 951 : if !script_ids.insert((s.connection, s.script.catalog_entry_id())) {
429 1 : state.report(NONE, i as u32, None, "duplicate planner script".into(), true);
430 950 : }
431 951 : if !s.name.is_empty() && !names.insert(s.name) {
432 1 : state.report(NONE, i as u32, None, format!("duplicate script name: {}", s.name), true);
433 950 : }
434 : }
435 478 : let mut files: BTreeSet<&str> = BTreeSet::new();
436 478 : for a in request.attachments {
437 34 : if !files.insert(a.file) {
438 1 : state.report(NONE, NONE, None, format!("duplicate attachment: {}", a.file), true);
439 33 : }
440 : }
441 478 : let targets = seeds(request);
442 910 : if targets.iter().chain(request.materialized).any(|&t| t as usize >= script_count) {
443 1 : state.report(NONE, NONE, None, "unknown planner script".into(), true);
444 477 : }
445 : // An invalid request stops every plan, also a tolerant one
446 478 : if !state.graph.valid || state.partial {
447 6 : state.graph.valid = false;
448 6 : return finish(state, request, &[], &[], ScriptReferences::new(script_count));
449 472 : }
450 :
451 : // Prepare the scripts: all of them in a plan of the session, also those whose results the host provides, the
452 : // targets and the scripts they reference in a plan of targets
453 472 : let mut prepared: Vec<Option<Prepared>> = (0..script_count).map(|_| None).collect();
454 472 : let mut planned = vec![false; script_count];
455 472 : let mut references = ScriptReferences::new(script_count);
456 472 : let mut pending: Vec<u32> = if targeted { targets.clone() } else { (0..script_count as u32).collect() };
457 1649 : while let Some(i) = pending.pop() {
458 1177 : if planned[i as usize] {
459 249 : continue;
460 928 : }
461 928 : planned[i as usize] = true;
462 928 : let input = &request.scripts[i as usize];
463 928 : let catalog = request.connections[input.connection as usize].catalog;
464 928 : let Some(p) = prepare(&mut state, i, input.script, catalog) else { continue };
465 914 : if !targeted {
466 710 : diagnose_syntax(&mut state, i, &p);
467 825 : }
468 914 : references.resolve(&mut state_reporter(&mut state), request, i, &p);
469 914 : pending.extend(references.of(i).iter().filter_map(|r| r.producer));
470 914 : prepared[i as usize] = Some(p);
471 : }
472 : // Invalid input snapshots must never yield executable partial plans
473 472 : if !state.graph.valid {
474 51 : return finish(state, request, &prepared, &planned, references);
475 421 : }
476 421 : references.check(&mut state_reporter(&mut state), request, &prepared);
477 421 : if tolerant {
478 48 : references.invalidate_unprepared(&prepared);
479 414 : }
480 421 : if !state.graph.valid {
481 24 : return finish(state, request, &prepared, &planned, references);
482 397 : }
483 :
484 : // The statements of the plan
485 397 : let mut include = vec![Include::None; script_count];
486 : {
487 733 : let mut pending: Vec<(u32, Include)> = targets.iter().map(|&t| (t, Include::All)).collect();
488 1328 : while let Some((i, level)) = pending.pop() {
489 931 : if include[i as usize] >= level {
490 70 : continue;
491 861 : }
492 861 : include[i as usize] = level;
493 : // The references of the included statements and of the terminal query, which an inline reference inlines
494 861 : for r in references.of(i) {
495 240 : let Some(producer) = r.producer else { continue };
496 44 : match r.kind {
497 163 : ReferenceKind::Inline => pending.push((producer, Include::Inlined)),
498 44 : ReferenceKind::Scan if request.materialized.contains(&producer) => {}
499 35 : ReferenceKind::Scan => pending.push((producer, Include::All)),
500 7 : ReferenceKind::Invalid | ReferenceKind::Attachment => {}
501 : }
502 : }
503 : }
504 : }
505 : // A plan includes every statement of a script or none
506 1839 : let included = |i: usize| include[i] == Include::All;
507 : // Scripts that a plan of targets passes through as one statement, they don't parse
508 780 : let opaque: Vec<bool> = (0..script_count).map(|i| targeted && include[i] == Include::All && prepared[i].as_ref().is_some_and(|p| !p.parses())).collect();
509 :
510 : // The node ids of the included statements
511 397 : let mut node_of: BTreeMap<(u32, u32), u32> = BTreeMap::new();
512 397 : let mut node_count = 0u32;
513 780 : for i in 0..script_count {
514 780 : let Some(p) = prepared[i].as_ref() else { continue };
515 760 : if opaque[i] {
516 4 : node_of.insert((i as u32, 0), node_count);
517 4 : node_count += 1;
518 4 : continue;
519 756 : }
520 756 : if !included(i) {
521 29 : continue;
522 727 : }
523 1054 : for statement_id in 0..p.parsed.statements.len() as u32 {
524 1054 : node_of.insert((i as u32, statement_id), node_count);
525 1054 : node_count += 1;
526 1054 : }
527 : }
528 780 : let participates = |i: usize| node_of.range((i as u32, 0)..=(i as u32, u32::MAX)).next().is_some();
529 :
530 : // The relations of every connection
531 452 : let mut connections: Vec<ConnectionState> = (0..request.connections.len()).map(|_| ConnectionState::default()).collect();
532 : // Catalog entry ids are unique within the catalog of a connection
533 397 : let participating_ids: BTreeSet<(u32, CatalogEntryId)> = (0..script_count)
534 780 : .filter(|&i| participates(i))
535 728 : .map(|i| (request.scripts[i].connection, request.scripts[i].script.catalog_entry_id()))
536 397 : .collect();
537 452 : for (ci, connection) in request.connections.iter().enumerate() {
538 452 : let c = &mut connections[ci];
539 453 : for (_, entry) in connection.catalog.entries_ranked() {
540 400 : c.views.extend(tables(entry).iter().filter(|t| t.is_view).map(|t| t.table_id().pack()));
541 474 : for table in tables(entry) {
542 473 : c.declarations.insert(table.object_id, table);
543 473 : c.published_views.insert(table.object_id, table.is_view);
544 473 : }
545 : }
546 924 : for (i, s) in request.scripts.iter().enumerate() {
547 924 : let Some(p) = prepared[i].as_ref().filter(|_| s.connection as usize == ci) else { continue };
548 760 : let analyzed = p.analyzed;
549 760 : c.views.extend(analyzed.entry.table_declarations.iter().filter(|t| t.is_view).map(|t| t.table_id().pack()));
550 760 : for table in &analyzed.entry.table_declarations {
551 147 : if c.published_views.get(&table.object_id).is_some_and(|&v| v != table.is_view) {
552 0 : c.incompatible_kinds.insert(table.object_id);
553 147 : }
554 147 : c.declarations.insert(table.object_id, table);
555 147 : let node = table.ast_statement_id.and_then(|statement| node_of.get(&(i as u32, statement)));
556 147 : if let Some(&node) = node {
557 147 : c.producers.insert(table.table_id().pack(), node);
558 147 : }
559 : }
560 : }
561 453 : for (_, entry) in connection.catalog.entries_ranked() {
562 : // Participating declarations enter the namespace only when their CREATE is reached
563 400 : if participating_ids.contains(&(ci as u32, entry.data().catalog_entry_id)) {
564 134 : continue;
565 266 : }
566 329 : for table in tables(entry) {
567 329 : c.live_namespace.entry(resource_name(&table.table_name)).or_insert(table);
568 329 : }
569 : }
570 : }
571 :
572 : // The materializations: the scripts that a reference of the LOCAL connection scans
573 397 : let mut materializations: Vec<ExecutionGraphMaterializationT> = Vec::new();
574 397 : let mut materialization_of: BTreeMap<u32, u32> = BTreeMap::new();
575 780 : for i in 0..script_count as u32 {
576 780 : for r in references.of(i) {
577 226 : if include[i as usize] == Include::None {
578 3 : continue;
579 223 : }
580 223 : let (ReferenceKind::Scan, Some(producer)) = (r.kind, r.producer) else { continue };
581 41 : materialization_of.entry(producer).or_insert_with(|| {
582 38 : let mut m = ExecutionGraphMaterializationT::default();
583 38 : m.script = producer;
584 38 : let terminal = prepared[producer as usize].as_ref().and_then(Prepared::terminal);
585 38 : m.node = terminal.and_then(|t| node_of.get(&(producer, t)).copied()).unwrap_or(NONE);
586 38 : m.path = Some(match request.scripts[producer as usize].materialization_file {
587 1 : Some(file) => format!("{}/{}", request.mount_prefix, file),
588 37 : None => format!("{}/s{}.parquet", request.mount_prefix, producer),
589 : });
590 38 : materializations.push(m);
591 38 : (materializations.len() - 1) as u32
592 38 : });
593 : }
594 : }
595 :
596 397 : let mut nodes: Vec<ExecutionGraphNodeT> = Vec::new();
597 397 : let mut graph_references: Vec<ExecutionGraphScriptReferenceT> = Vec::new();
598 : // Edges of script references: (from, to, reason)
599 397 : let mut reference_edges: Vec<(u32, u32, Reason, u32)> = Vec::new();
600 : // Temporary relations do not outlive the database session
601 397 : let change_options = CatalogChangeOptions::default();
602 780 : for si in 0..script_count {
603 780 : let Some(p) = prepared[si].as_ref() else { continue };
604 760 : let connection = request.scripts[si].connection;
605 760 : let (scanned, parsed, analyzed) = (p.scanned, p.parsed, p.analyzed);
606 760 : if opaque[si] {
607 : // A script that does not parse runs as written, as one statement
608 4 : let node_id = nodes.len() as u32;
609 4 : let mut node = ExecutionGraphNodeT::default();
610 4 : node.script = si as u32;
611 4 : node.statement_id = 0;
612 4 : node.statement_type = parsed.statements.first().map(|s| s.statement_type).unwrap_or(Type::NONE);
613 4 : node.sql = opt(Some(read_executable_text(scanned, TextSpan::new(0, scanned.text.len() as u32)).trim().to_string()));
614 4 : node.produces_output = true;
615 4 : node.script_result = true;
616 4 : node.barrier = true;
617 4 : node.connection = connection;
618 4 : state.diagnose(node_id, si as u32, "the script does not parse, it runs as written".into());
619 4 : state.spans.push(parsed.associate_descriptions().first().map(|d| d.statement_span).unwrap_or_default());
620 4 : nodes.push(node);
621 4 : continue;
622 756 : }
623 756 : let text_at = |offset: u32, length: u32| -> String {
624 36 : let text = scanned.text.as_bytes();
625 36 : let begin = (offset as usize).min(text.len());
626 36 : let end = (begin + length as usize).min(text.len());
627 36 : String::from_utf8_lossy(&text[begin..end]).into_owned()
628 36 : };
629 756 : let descriptions = parsed.associate_descriptions();
630 756 : let terminal = p.terminal();
631 1083 : for (statement_id, statement) in parsed.statements.iter().enumerate() {
632 1083 : if !included(si) {
633 29 : continue;
634 1054 : }
635 1054 : let node_id = nodes.len() as u32;
636 1054 : debug_assert_eq!(node_of.get(&(si as u32, statement_id as u32)), Some(&node_id));
637 1054 : let c = &mut connections[connection as usize];
638 1054 : let mut node = ExecutionGraphNodeT::default();
639 1054 : node.script = si as u32;
640 1054 : node.statement_id = statement_id as u32;
641 1054 : node.statement_type = statement.statement_type;
642 1054 : node.connection = connection;
643 1054 : let ty = statement.statement_type;
644 1054 : let span = descriptions[statement_id].statement_span;
645 1054 : state.spans.push(span);
646 1054 : node.produces_output = produces_output(parsed, statement);
647 1054 : node.script_result = terminal == Some(statement_id as u32) && node.produces_output;
648 1054 : if node.script_result {
649 574 : if let Some(&m) = materialization_of.get(&(si as u32)) {
650 29 : node.materialization = m;
651 545 : }
652 480 : }
653 : // The SQL of the statement as written, with the script references spliced in
654 1054 : let mut spliced = script_references::splice_statement(
655 1054 : request,
656 1054 : &prepared,
657 1054 : &references,
658 1054 : si as u32,
659 1054 : statement_id as u32,
660 1054 : &materialization_of,
661 1054 : &materializations,
662 1054 : &state.inputs,
663 : );
664 : // The inputs of the statement become literals, an input without a value fails the plan
665 1054 : let dialect = request.connections[connection as usize].dialect;
666 1054 : let bindings = bind_inputs(&state.inputs, p, Some(statement_id as u32), dialect);
667 1054 : spliced.edits.extend(bindings.edits);
668 1054 : for (span, message) in bindings.errors {
669 12 : state.report(node_id, si as u32, Some(span), message, true);
670 12 : }
671 1054 : for (script, span, message) in std::mem::take(&mut spliced.input_errors) {
672 0 : state.report(node_id, script, Some(span), message, true);
673 0 : }
674 1054 : let mut sql = if spliced.edits.is_empty() { read_executable_text(scanned, span) } else { read_executable_text_with(scanned, span, &spliced.edits) };
675 1054 : for r in &spliced.references {
676 213 : let mut reference = ExecutionGraphScriptReferenceT::default();
677 213 : reference.node = node_id;
678 213 : reference.consumer_script = si as u32;
679 213 : reference.text_span = Some(text_span_t(r.reference.text_span));
680 213 : reference.script = r.reference.producer.unwrap_or(NONE);
681 213 : reference.attachment = r.reference.attachment.unwrap_or(NONE);
682 213 : reference.parameter = Some(r.reference.name.clone());
683 213 : reference.cte_name = Some(r.cte_name.clone());
684 213 : reference.materialization = r.materialization.unwrap_or(NONE);
685 213 : graph_references.push(reference);
686 213 : }
687 1054 : let root = &parsed.nodes[statement.root as usize];
688 1054 : let mut explain_executes = false;
689 1054 : let mut accesses: BTreeMap<(u32, u32), Mode> = BTreeMap::new();
690 1054 : let mut barrier = false;
691 : // A statement that the planner can't classify is a barrier, it runs in session order
692 : macro_rules! barrier {
693 : ($message:expr) => {{
694 : barrier = true;
695 : state.diagnose(node_id, si as u32, $message.to_string());
696 : }};
697 : }
698 1054 : let access = |state: &mut PlanState, accesses: &mut BTreeMap<(u32, u32), Mode>, name: &Name, table: Option<&TableDeclaration>, mode: Mode| {
699 959 : let target = match table {
700 917 : Some(t) => (state.intern_entity(connection, t), NONE),
701 42 : None => (NONE, state.intern_unresolved(connection, name)),
702 : };
703 959 : let entry = accesses.entry(target).or_insert(mode);
704 959 : if mode > *entry {
705 7 : *entry = mode;
706 952 : }
707 959 : target
708 959 : };
709 1054 : match ty {
710 : Type::SELECT
711 : | Type::INSERT
712 : | Type::DELETE
713 : | Type::EXPLAIN
714 : | Type::CREATE_TABLE
715 : | Type::CREATE_TABLE_AS
716 : | Type::CREATE_VIEW
717 : | Type::SELECT_INTO
718 : | Type::DROP_TABLE
719 1047 : | Type::DROP_VIEW => {}
720 7 : _ => {
721 7 : barrier!("unsupported statement effects; backend session semantics are not guaranteed");
722 7 : }
723 : }
724 : // A query with `@vis` tags carries its visualization
725 1054 : let visualization = analyzed.visualization_specs.iter().find(|s| s.ast_statement_id == Some(statement_id as u32));
726 1054 : if let Some(spec) = visualization.filter(|_| ty == Type::SELECT) {
727 31 : match compile_visualization(spec, analyzed) {
728 31 : Ok(mut compiled) => {
729 31 : compiled.renderer = opt(compiled.renderer.take());
730 31 : compiled.vegalite_spec = opt(compiled.vegalite_spec.take());
731 31 : node.visualization = Some(Box::new(compiled));
732 31 : }
733 0 : Err(e) => {
734 0 : barrier = true;
735 0 : state.report(node_id, si as u32, None, format!("visualization: {e}"), true);
736 0 : sql.clear();
737 0 : }
738 : }
739 1023 : }
740 1054 : if ty == Type::EXPLAIN {
741 70 : let [body, options] = attrs!(parsed.children(root), [Key::EXT_EXPLAIN_STATEMENT, Key::EXT_EXPLAIN_OPTIONS]);
742 70 : let option_text = |n: Option<&Node>| -> String {
743 36 : let Some(n) = n.filter(|n| n.node_type() != NodeType::NONE) else { return String::new() };
744 36 : let ts = resolve_text_span(&scanned.text, &scanned.symbols, *n.symbol_span());
745 36 : let mut text = text_at(ts.offset, ts.length).to_ascii_lowercase();
746 36 : let b = text.as_bytes();
747 36 : if b.len() >= 2 && (b[0] == b'\'' || b[0] == b'"') && b[b.len() - 1] == b[0] {
748 3 : text = text[1..text.len() - 1].to_string();
749 33 : }
750 36 : text
751 36 : };
752 70 : if let Some(options) = options.filter(|o| o.node_type() == NodeType::ARRAY) {
753 33 : for option in parsed.children(options) {
754 33 : if option.node_type() != NodeType::OBJECT_SQL_GENERIC_OPTION {
755 0 : continue;
756 33 : }
757 33 : let [key, value] = attrs!(parsed.children(option), [Key::SQL_GENERIC_OPTION_KEY, Key::SQL_GENERIC_OPTION_VALUE]);
758 33 : let name = option_text(key);
759 33 : if name == "analyze" || name == "analyse" {
760 : // Unknown arguments are execution-enabled, not proof of read-only behavior
761 24 : explain_executes |= value.is_none_or(|v| v.node_type() != NodeType::BOOL || v.children_begin_or_value() != 0);
762 9 : } else if name == "format" && matches!(option_text(value).as_str(), "analyze" | "analyse") {
763 3 : explain_executes = true;
764 6 : }
765 : }
766 40 : }
767 70 : if body.is_none_or(|b| b.node_type() == NodeType::NONE) {
768 0 : barrier!("EXPLAIN requires a statement or expressions body");
769 70 : }
770 984 : }
771 1054 : for e in &analyzed.errors {
772 14 : let Some(error_node) = e.ast_node_id else { continue };
773 9 : if error_node < statement.nodes_begin || error_node >= statement.nodes_begin + statement.node_count {
774 6 : continue;
775 3 : }
776 3 : state.diagnose(node_id, si as u32, format!("analyzer: {}", e.message));
777 3 : if e.severity == AnalyzerErrorSeverity::ERROR {
778 3 : barrier = true;
779 3 : }
780 : }
781 : // AST effects that are not represented by ordinary relation reads, reported once per statement
782 1054 : let mut effects = false;
783 12854 : for ni in statement.nodes_begin..statement.nodes_begin + statement.node_count {
784 12854 : let ast = &parsed.nodes[ni as usize];
785 12854 : if ast.node_type() == NodeType::NONE {
786 0 : continue;
787 12854 : }
788 12854 : let mutating = matches!(
789 12854 : ast.node_type(),
790 : NodeType::OBJECT_SQL_INSERT
791 : | NodeType::OBJECT_SQL_DELETE
792 : | NodeType::OBJECT_SQL_CREATE
793 : | NodeType::OBJECT_SQL_CREATE_AS
794 : | NodeType::OBJECT_SQL_VIEW
795 : | NodeType::OBJECT_SQL_CREATE_FUNCTION
796 12571 : ) || ast.attribute_key() == Key::SQL_SELECT_INTO;
797 12854 : if ty == Type::EXPLAIN && explain_executes && mutating {
798 12 : barrier!("execution-enabled EXPLAIN with a mutating body cannot be safely classified");
799 12842 : }
800 12854 : if ast.node_type() == NodeType::ENUM_SQL_TEMP_TYPE && ast.children_begin_or_value() == TempType::NONE.0 as u32 {
801 8 : continue;
802 12846 : }
803 12842 : if matches!(
804 12846 : ast.attribute_key(),
805 : Key::SQL_CREATE_TABLE_TEMP
806 : | Key::SQL_CREATE_AS_TEMP
807 : | Key::SQL_VIEW_TEMP
808 : | Key::SQL_TEMP_TYPE
809 : | Key::SQL_SELECT_ROW_LOCKING
810 : | Key::SQL_TABLE_CONSTRAINT_REFERENCES_NAME
811 12842 : ) || matches!(ast.node_type(), NodeType::OBJECT_SQL_GRAPH_TABLE | NodeType::OBJECT_SQL_PROPERTY_GRAPH)
812 : {
813 10 : if !effects {
814 7 : barrier!("unsupported session, locking, constraint, or graph effects");
815 7 : }
816 10 : effects = true;
817 12836 : }
818 : }
819 1709 : for table_ref in &analyzed.table_references {
820 : // A table parameter reads its script through the script references
821 1646 : if table_ref.ast_statement_id != Some(statement_id as u32) || table_ref.reads_table_parameter() {
822 965 : continue;
823 681 : }
824 681 : let Some(relation) = &table_ref.inner else {
825 0 : barrier!("unsupported table source");
826 0 : continue;
827 : };
828 681 : if !relation.resolved_alternatives.is_empty() {
829 7 : barrier!("ambiguous relation reference");
830 674 : }
831 : // References to CTEs are no relation accesses
832 681 : if is_cte_reference(analyzed, table_ref) {
833 36 : continue;
834 645 : }
835 645 : let (name, table) = c.resolve(relation);
836 645 : if relation.resolved_table.is_some() && table.is_none() {
837 0 : barrier!("resolved relation declaration is incompatible or unavailable");
838 645 : }
839 645 : let write = table_ref.role == TableRole::Write && (ty != Type::EXPLAIN || explain_executes);
840 645 : access(&mut state, &mut accesses, &name, table, if write { Mode::WRITE } else { Mode::READ });
841 645 : if c.dropped_resources.contains(&name) {
842 4 : barrier!("relation was dropped without an earlier lifetime producer; later producer cannot be reordered");
843 641 : }
844 645 : match &relation.resolved_table {
845 35 : None => barrier!(format!("unresolved relation: {}", relation.table_name.debug_string())),
846 610 : Some(resolved) => {
847 610 : let id = match table {
848 610 : Some(t) => t.table_id().pack(),
849 0 : None => resolved.catalog_table_id.unpack_table_id().pack(),
850 : };
851 610 : if c.producers.get(&id).is_some_and(|&p| p >= node_id) && !c.live_producers.contains(&name) {
852 7 : barrier!("relation resolves to a later producer in input order");
853 603 : }
854 610 : if c.views.contains(&id) || c.live_views.contains(&name) {
855 24 : barrier!("view dependencies are not expanded; conservative access barrier");
856 586 : }
857 : }
858 : }
859 : }
860 : // An inline script reference reads the relations of the terminal query of the referenced script
861 1054 : let direct: BTreeSet<(u32, u32)> = accesses.keys().copied().collect();
862 1054 : let mut inherited: BTreeSet<(u32, u32)> = BTreeSet::new();
863 1054 : for &(producer, producer_statement) in &spliced.inlined {
864 153 : let producer_analyzed = prepared[producer as usize].as_ref().unwrap().analyzed;
865 153 : for table_ref in &producer_analyzed.table_references {
866 124 : if table_ref.ast_statement_id != Some(producer_statement)
867 124 : || is_cte_reference(producer_analyzed, table_ref)
868 121 : || table_ref.reads_table_parameter()
869 : {
870 3 : continue;
871 121 : }
872 121 : let Some(relation) = &table_ref.inner else { continue };
873 121 : let (name, table) = c.resolve(relation);
874 121 : let target = access(&mut state, &mut accesses, &name, table, Mode::READ);
875 121 : if !direct.contains(&target) {
876 113 : inherited.insert(target);
877 113 : }
878 121 : if relation.resolved_table.is_none() {
879 0 : barrier!(format!("unresolved relation: {}", relation.table_name.debug_string()));
880 121 : }
881 : }
882 : }
883 1054 : for &m in &spliced.inputs {
884 44 : let producer_node = materializations[m as usize].node;
885 44 : if producer_node != NONE {
886 35 : reference_edges.push((producer_node, node_id, Reason::MATERIALIZATION, materializations[m as usize].script));
887 35 : }
888 : }
889 1054 : node.inputs = (!spliced.inputs.is_empty()).then(|| spliced.inputs.clone());
890 1054 : node.inlined_scripts = (!spliced.inlined.is_empty()).then(|| spliced.inlined.iter().map(|&(producer, _)| producer).collect());
891 1054 : node.attachments = (!spliced.attachments.is_empty()).then(|| spliced.attachments.clone());
892 1054 : let drop = matches!(ty, Type::DROP_TABLE | Type::DROP_VIEW);
893 1054 : let ddl = drop || matches!(ty, Type::CREATE_TABLE | Type::CREATE_TABLE_AS | Type::CREATE_VIEW | Type::SELECT_INTO);
894 1054 : if ddl {
895 193 : let packed = pack::pack_parsed_statement(parsed, statement_id);
896 193 : match packed.target {
897 0 : None => barrier!("missing DDL target"),
898 193 : Some(target) => {
899 193 : let mut name: Name = (
900 193 : target.database_name.clone().unwrap_or_default(),
901 193 : target.schema_name.clone().unwrap_or_default(),
902 193 : target.relation_name.clone().unwrap_or_default(),
903 193 : );
904 193 : let mut table: Option<&TableDeclaration> = None;
905 193 : if drop {
906 : // Prefer an exact live name, then native suffix matches. Never let a future
907 : // declaration in an analyzed input redirect this DROP.
908 46 : if !c.live_namespace.contains_key(&name) {
909 13 : let matches: BTreeSet<Name> = c
910 13 : .live_namespace
911 13 : .keys()
912 13 : .filter(|n| n.2 == name.2 && (name.1.is_empty() || n.1 == name.1) && (name.0.is_empty() || n.0 == name.0))
913 13 : .cloned()
914 13 : .collect();
915 13 : if matches.len() == 1 {
916 6 : name = matches.into_iter().next().unwrap();
917 7 : } else if matches.len() > 1 {
918 3 : barrier!("ambiguous DROP target");
919 4 : }
920 33 : }
921 46 : match c.live_namespace.get(&name) {
922 39 : Some(&live) => table = Some(live),
923 7 : None => barrier!("unresolved DROP target; conservative access barrier"),
924 : }
925 46 : if let Some(t) = table {
926 39 : if c.views.contains(&t.table_id().pack()) || c.live_views.contains(&name) {
927 9 : barrier!("view dependencies are not expanded; conservative access barrier");
928 30 : }
929 7 : }
930 : } else {
931 175 : for declaration in &analyzed.entry.table_declarations {
932 175 : if declaration.ast_statement_id == Some(statement_id as u32) && resource_name(&declaration.table_name) == name {
933 147 : table = Some(declaration);
934 147 : }
935 : }
936 147 : match table {
937 147 : Some(t) => name = resource_name(&t.table_name),
938 0 : None => barrier!("missing DDL target declaration"),
939 : }
940 : }
941 193 : let target = access(&mut state, &mut accesses, &name, table, Mode::DDL);
942 193 : inherited.remove(&target);
943 193 : if drop {
944 46 : c.live_producers.remove(&name);
945 46 : c.live_namespace.remove(&name);
946 46 : c.live_views.remove(&name);
947 46 : c.dropped_resources.insert(name);
948 46 : } else {
949 147 : c.live_producers.insert(name.clone());
950 147 : if let Some(t) = table {
951 147 : c.live_namespace.insert(name.clone(), t);
952 147 : }
953 147 : c.dropped_resources.remove(&name);
954 147 : if ty == Type::CREATE_VIEW {
955 17 : c.live_views.insert(name);
956 130 : } else {
957 130 : c.live_views.remove(&name);
958 130 : }
959 : }
960 : }
961 : }
962 861 : }
963 1054 : state.inherited.extend(inherited.into_iter().map(|(entity, unresolved)| (node_id, entity, unresolved)));
964 1054 : node.accesses = opt_vec(
965 1054 : accesses
966 1054 : .into_iter()
967 1054 : .map(|((entity, unresolved), mode)| {
968 938 : let mut a = ExecutionGraphAccessT::default();
969 938 : a.entity = entity;
970 938 : a.unresolved_reference = unresolved;
971 938 : a.mode = mode;
972 938 : a
973 938 : })
974 1054 : .collect(),
975 : );
976 : // The catalog changes of the statement once it succeeded. A script with syntax errors changes
977 : // nothing we could follow.
978 1054 : if !sql.is_empty() && parsed.errors.is_empty() && scanned.errors.is_empty() {
979 1054 : node.catalog_changes = derive_catalog_change(analyzed, statement_id, &change_options).map(|c| vec![c.pack()]);
980 0 : }
981 1054 : node.sql = opt(Some(sql));
982 1054 : node.barrier = barrier;
983 1054 : nodes.push(node);
984 : }
985 : }
986 :
987 : // Order conflicting accesses along the timeline of every relation
988 : /// The accesses of a relation that later accesses must be ordered after
989 : #[derive(Default)]
990 : struct Timeline {
991 : /// The node that wrote the relation last
992 : writer: Option<u32>,
993 : /// The access mode of the last writer
994 : mode: Option<Mode>,
995 : /// The nodes that read the relation since the last write
996 : readers: Vec<u32>,
997 : }
998 397 : let mut timelines: BTreeMap<(u32, Name), Timeline> = BTreeMap::new();
999 397 : let mut edges: BTreeSet<(u32, u32, u32, u32, u8)> = BTreeSet::new();
1000 2389 : let mut edge = |from: Option<u32>, to: u32, entity: u32, unresolved: u32, reason: Reason| {
1001 2389 : if let Some(from) = from {
1002 568 : if from != to {
1003 568 : edges.insert((from, to, entity, unresolved, reason.0));
1004 568 : }
1005 1821 : }
1006 2389 : };
1007 : // Barriers order the statements of their connection
1008 397 : let mut last_barrier: BTreeMap<u32, u32> = BTreeMap::new();
1009 397 : let mut since_barrier: BTreeMap<u32, Vec<u32>> = BTreeMap::new();
1010 397 : let mut previous_of_script: BTreeMap<u32, u32> = BTreeMap::new();
1011 1058 : for (i, node) in nodes.iter().enumerate() {
1012 1058 : let i = i as u32;
1013 1058 : let connection = node.connection;
1014 : // A plan of targets runs the statements of a script in order
1015 1058 : if targeted {
1016 212 : edge(previous_of_script.insert(node.script, i), i, NONE, NONE, Reason::SEQUENCE);
1017 986 : }
1018 1058 : edge(last_barrier.get(&connection).copied(), i, NONE, NONE, Reason::BARRIER);
1019 1058 : let since = since_barrier.entry(connection).or_default();
1020 1058 : if node.barrier {
1021 122 : for &earlier in since.iter() {
1022 71 : edge(Some(earlier), i, NONE, NONE, Reason::BARRIER);
1023 71 : }
1024 120 : since.clear();
1025 120 : last_barrier.insert(connection, i);
1026 938 : } else {
1027 938 : since.push(i);
1028 938 : }
1029 : // Merge by name only for conflict tracking
1030 1058 : let mut merged: BTreeMap<Name, &ExecutionGraphAccessT> = BTreeMap::new();
1031 1058 : for a in node.accesses.iter().flatten() {
1032 938 : let name = if a.entity != NONE {
1033 896 : let e = &state.entity_list[a.entity as usize];
1034 896 : (e.database_name.clone().unwrap_or_default(), e.schema_name.clone().unwrap_or_default(), e.name.clone().unwrap_or_default())
1035 : } else {
1036 42 : let r = &state.unresolved_list[a.unresolved_reference as usize];
1037 42 : (r.database_name.clone().unwrap_or_default(), r.schema_name.clone().unwrap_or_default(), r.name.clone().unwrap_or_default())
1038 : };
1039 938 : match merged.get(&name) {
1040 0 : Some(existing) if a.mode <= existing.mode => {}
1041 938 : _ => {
1042 938 : merged.insert(name, a);
1043 938 : }
1044 : }
1045 : }
1046 1058 : for (name, a) in merged {
1047 938 : let timeline = timelines.entry((connection, name)).or_default();
1048 938 : let ddl = a.mode == Mode::DDL || timeline.mode == Some(Mode::DDL);
1049 938 : if a.mode == Mode::READ {
1050 631 : edge(timeline.writer, i, a.entity, a.unresolved_reference, if ddl { Reason::DDL } else { Reason::READ_AFTER_WRITE });
1051 631 : timeline.readers.push(i);
1052 : } else {
1053 307 : edge(timeline.writer, i, a.entity, a.unresolved_reference, if ddl { Reason::DDL } else { Reason::WRITE_AFTER_WRITE });
1054 307 : for &reader in &timeline.readers {
1055 78 : edge(Some(reader), i, a.entity, a.unresolved_reference, if a.mode == Mode::DDL { Reason::DDL } else { Reason::WRITE_AFTER_READ });
1056 : }
1057 307 : timeline.readers.clear();
1058 307 : timeline.writer = Some(i);
1059 307 : timeline.mode = Some(a.mode);
1060 : }
1061 : }
1062 : }
1063 : // A scan reference needs the result of the referenced script before it in the session
1064 397 : for (from, to, reason, producer) in reference_edges {
1065 35 : if from < to {
1066 32 : edge(Some(from), to, NONE, NONE, reason);
1067 32 : } else {
1068 3 : let consumer = nodes[to as usize].script;
1069 3 : let message = format!(
1070 3 : "{} references {}, which runs later; move {} above {}",
1071 3 : script_label(request, consumer),
1072 3 : script_label(request, producer),
1073 3 : script_label(request, producer),
1074 3 : script_label(request, consumer)
1075 3 : );
1076 3 : state.report(to, consumer, None, message, true);
1077 3 : }
1078 : }
1079 397 : let edges = edges
1080 397 : .into_iter()
1081 572 : .map(|(from, to, entity, unresolved, reason)| {
1082 568 : let mut e = ExecutionGraphEdgeT::default();
1083 568 : e.from = from;
1084 568 : e.to = to;
1085 568 : e.entity = entity;
1086 568 : e.unresolved_reference = unresolved;
1087 568 : e.reason = Reason(reason);
1088 568 : e
1089 568 : })
1090 397 : .collect();
1091 397 : state.graph.nodes = opt_vec(nodes);
1092 397 : state.graph.edges = opt_vec(edges);
1093 397 : state.graph.materializations = opt_vec(materializations);
1094 397 : state.graph.references = opt_vec(graph_references);
1095 397 : finish(state, request, &prepared, &planned, references)
1096 478 : }
1097 :
1098 : /// The label of a script in diagnostics, `foo.sql`
1099 217 : pub(crate) fn script_label(request: &SessionPlanRequest, script: u32) -> String {
1100 217 : match request.scripts.get(script as usize).map(|s| s.name).filter(|n| !n.is_empty()) {
1101 217 : Some(name) => format!("{name}.sql"),
1102 0 : None => format!("script {script}"),
1103 : }
1104 217 : }
1105 :
1106 : /// Does a table reference read a common table expression?
1107 805 : fn is_cte_reference(analyzed: &AnalyzedScript, table_ref: &dashql_core::analyzer::TableReference) -> bool {
1108 805 : let Some(relation) = &table_ref.inner else { return false };
1109 805 : if relation.resolved_table.is_some()
1110 74 : || table_ref.role != TableRole::Read
1111 74 : || !relation.table_name.database_name.text.is_empty()
1112 62 : || !relation.table_name.schema_name.text.is_empty()
1113 : {
1114 746 : return false;
1115 59 : }
1116 59 : let Some(&scope) = table_ref.ast_scope_root.and_then(|r| analyzed.name_scopes_by_root_node.get(&r)) else { return false };
1117 59 : let alias = match &table_ref.alias {
1118 6 : Some((a, _)) => a.text.clone(),
1119 53 : None => relation.table_name.table_name.text.clone(),
1120 : };
1121 59 : matches!(analyzed.name_scopes[scope as usize].source(&alias).map(|s| &s.table), Some(ReferencedTable::Cte(_)))
1122 805 : }
1123 :
1124 213 : fn text_span_t(span: TextSpan) -> TextSpanT {
1125 213 : let mut t = TextSpanT::default();
1126 213 : t.offset = span.offset;
1127 213 : t.length = span.length;
1128 213 : t
1129 213 : }
1130 :
1131 : /// A reporter for the diagnostics of script references
1132 : struct Reporter<'s> {
1133 : state: &'s mut PlanState,
1134 : }
1135 :
1136 : impl ReferenceErrors for Reporter<'_> {
1137 86 : fn error(&mut self, script: u32, text_span: Option<TextSpan>, message: String) {
1138 86 : self.state.report(NONE, script, text_span, message, true);
1139 86 : }
1140 : }
1141 :
1142 1335 : fn state_reporter(state: &mut PlanState) -> Reporter<'_> {
1143 1335 : Reporter { state }
1144 1335 : }
1145 :
1146 : /// A string that is left out of the packed buffer when empty
1147 5362 : fn opt(s: Option<String>) -> Option<String> {
1148 5362 : s.filter(|s| !s.is_empty())
1149 5362 : }
1150 :
1151 : /// A vector that is left out of the packed buffer when empty
1152 6424 : fn opt_vec<T>(v: Vec<T>) -> Option<Vec<T>> {
1153 6424 : (!v.is_empty()).then_some(v)
1154 6424 : }
1155 :
1156 : /// Complete the graph of a plan with its connections, scripts, entities and diagnostics
1157 478 : fn finish(state: PlanState, request: &SessionPlanRequest, prepared: &[Option<Prepared>], planned: &[bool], references: ScriptReferences) -> Plan {
1158 478 : let targets: BTreeSet<u32> = seeds(request).into_iter().collect();
1159 478 : let mut graph = state.graph;
1160 478 : if state.partial {
1161 11 : graph.valid = false;
1162 467 : }
1163 478 : graph.connections = opt_vec(
1164 478 : request
1165 478 : .connections
1166 478 : .iter()
1167 549 : .map(|c| {
1168 549 : let mut out = ExecutionGraphConnectionT::default();
1169 549 : out.connection_id = opt(Some(c.id.to_string()));
1170 549 : out.kind = match c.kind {
1171 440 : ConnectionKind::Remote => ExecutionGraphConnectionKind::REMOTE,
1172 109 : ConnectionKind::Local => ExecutionGraphConnectionKind::LOCAL,
1173 : };
1174 549 : out
1175 549 : })
1176 478 : .collect(),
1177 : );
1178 478 : graph.attachments = opt_vec(
1179 478 : request
1180 478 : .attachments
1181 478 : .iter()
1182 478 : .enumerate()
1183 478 : .map(|(i, a)| {
1184 34 : let mut out = ExecutionGraphAttachmentT::default();
1185 34 : out.name = opt(Some(a.name.to_string()));
1186 34 : out.file = opt(Some(a.file.to_string()));
1187 34 : out.path = opt(Some(script_references::attachment_path(request, i as u32)));
1188 34 : out
1189 34 : })
1190 478 : .collect(),
1191 : );
1192 478 : graph.scripts = opt_vec(
1193 478 : request
1194 478 : .scripts
1195 478 : .iter()
1196 478 : .enumerate()
1197 951 : .map(|(i, s)| {
1198 951 : let mut meta = ExecutionGraphScriptT::default();
1199 951 : let script = s.script;
1200 951 : meta.script_id = script.catalog_entry_id();
1201 951 : meta.text_version = script.text_version();
1202 951 : if let Some(analyzed) = script.analyzed.as_ref() {
1203 951 : meta.analyzed_text_version = analyzed.parsed.scanned.text_version;
1204 951 : meta.analyzed_catalog_revision = analyzed.snapshot.version() as u64;
1205 951 : }
1206 951 : meta.name = opt(Some(s.name.to_string()));
1207 951 : meta.connection = s.connection;
1208 951 : meta.planned = planned.get(i).copied().unwrap_or(false);
1209 951 : meta.target = targets.contains(&(i as u32));
1210 951 : if let Some(p) = prepared.get(i).and_then(Option::as_ref) {
1211 : // The signature of the AST, the values that the script binds, `:min=10` per line, and the stamps of the
1212 : // attachments it reads, `attachment sales.parquet=<stamp>` per line. The keys of the scripts it
1213 : // references cover what they read.
1214 914 : meta.signature = opt(dashql_core::signature::compute_signature(p.parsed).ok().map(|signature| {
1215 : // The values in the Hyper dialect, the key covers the connection
1216 898 : let values = bind_inputs(&state.inputs, p, None, FormattingDialect::HYPER).values;
1217 898 : let attachments: BTreeSet<(&str, &str)> = references
1218 898 : .of(i as u32)
1219 898 : .iter()
1220 898 : .filter_map(|r| request.attachments.get(r.attachment? as usize))
1221 898 : .map(|a| (a.file, a.stamp))
1222 898 : .collect();
1223 898 : std::iter::once(signature)
1224 898 : .chain(values.iter().map(|(name, value)| format!(":{name}={value}")))
1225 898 : .chain(attachments.iter().map(|(file, stamp)| format!("attachment {file}={stamp}")))
1226 898 : .collect::<Vec<_>>()
1227 898 : .join("\n")
1228 898 : }));
1229 : // Queries only. The scripts that a script references are single queries, or the plan is invalid.
1230 1063 : meta.cacheable = p.parses() && !p.parsed.statements.is_empty() && p.parsed.statements.iter().all(|s| matches!(s.statement_type, Type::SELECT | Type::EXPLAIN));
1231 914 : meta.terminal_statement_id = p.terminal().unwrap_or(NONE);
1232 914 : meta.referenced_scripts = opt_vec(references.producers_of(i as u32));
1233 37 : } else {
1234 37 : meta.terminal_statement_id = NONE;
1235 37 : }
1236 951 : meta
1237 951 : })
1238 478 : .collect(),
1239 : );
1240 478 : graph.entities = opt_vec(state.entity_list);
1241 478 : graph.unresolved_references = opt_vec(state.unresolved_list);
1242 478 : graph.diagnostics = opt_vec(state.diagnostics);
1243 478 : Plan { graph, references, inputs: state.inputs, spans: state.spans, inherited: state.inherited }
1244 478 : }
1245 :
1246 : #[cfg(test)]
1247 : mod tests {
1248 : use super::*;
1249 :
1250 12 : fn analyzed(catalog: &Catalog, text: &str) -> Script {
1251 12 : let mut script = Script::new(catalog);
1252 12 : script.replace_text(text);
1253 12 : script.scan();
1254 12 : script.parse();
1255 12 : script.analyze(catalog, false).unwrap();
1256 12 : script
1257 12 : }
1258 :
1259 9 : fn messages(graph: &ExecutionGraphT) -> Vec<String> {
1260 9 : graph.diagnostics.iter().flatten().filter_map(|d| d.message.clone()).collect()
1261 9 : }
1262 :
1263 18 : fn request<'a>(connections: &'a [PlannerConnection<'a>], scripts: &'a [PlannerScript<'a>]) -> SessionPlanRequest<'a> {
1264 18 : SessionPlanRequest { connections, scripts, scope: PlanScope::Session, materialized: &[], mount_prefix: "/mnt/external/test", attachments: &[], attachment_prefix: "" }
1265 18 : }
1266 :
1267 : #[test]
1268 1 : fn rejects_duplicate_script_names() {
1269 1 : let catalog = Catalog::new();
1270 1 : let (a, b) = (analyzed(&catalog, "select 1"), analyzed(&catalog, "select 2"));
1271 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1272 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }, PlannerScript { script: &b, name: "foo", connection: 0, materialization_file: None }];
1273 1 : let graph = plan_session(&request(&connections, &scripts));
1274 1 : assert!(!graph.valid);
1275 1 : assert!(graph.nodes.is_none());
1276 1 : assert_eq!(messages(&graph), vec!["duplicate script name: foo"]);
1277 1 : }
1278 :
1279 : #[test]
1280 1 : fn rejects_connections_that_share_a_catalog() {
1281 1 : let catalog = Catalog::new();
1282 1 : let a = analyzed(&catalog, "select 1");
1283 1 : let connections = [
1284 1 : PlannerConnection { id: "first", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER },
1285 1 : PlannerConnection { id: "second", kind: ConnectionKind::Remote, catalog: &catalog, dialect: FormattingDialect::HYPER },
1286 1 : ];
1287 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }];
1288 1 : let graph = plan_session(&request(&connections, &scripts));
1289 1 : assert!(!graph.valid);
1290 1 : assert_eq!(messages(&graph), vec!["connections share a catalog: second"]);
1291 1 : }
1292 :
1293 : #[test]
1294 1 : fn rejects_unknown_connections_and_scripts() {
1295 1 : let catalog = Catalog::new();
1296 1 : let a = analyzed(&catalog, "select 1");
1297 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1298 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 1, materialization_file: None }];
1299 1 : assert_eq!(messages(&plan_session(&request(&connections, &scripts))), vec!["unknown planner connection"]);
1300 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }];
1301 1 : let mut targets = request(&connections, &scripts);
1302 1 : targets.scope = PlanScope::Targets(&[3]);
1303 1 : assert_eq!(messages(&plan_session(&targets)), vec!["unknown planner script"]);
1304 1 : }
1305 :
1306 : #[test]
1307 1 : fn names_materializations_after_the_files_of_the_host() {
1308 1 : let (trino, local) = (Catalog::new(), Catalog::new());
1309 1 : let (foo, bar) = (analyzed(&trino, "select 1 as a"), analyzed(&local, "select * from :foo"));
1310 1 : let connections = [
1311 1 : PlannerConnection { id: "trino", kind: ConnectionKind::Remote, catalog: &trino, dialect: FormattingDialect::HYPER },
1312 1 : PlannerConnection { id: "local", kind: ConnectionKind::Local, catalog: &local, dialect: FormattingDialect::HYPER },
1313 1 : ];
1314 1 : let scripts = [
1315 1 : PlannerScript { script: &foo, name: "foo", connection: 0, materialization_file: Some("abc.0123.parquet") },
1316 1 : PlannerScript { script: &bar, name: "bar", connection: 1, materialization_file: None },
1317 1 : ];
1318 1 : let graph = plan_session(&request(&connections, &scripts));
1319 1 : assert!(graph.valid, "{:?}", messages(&graph));
1320 1 : let path = graph.materializations.as_ref().unwrap()[0].path.clone().unwrap();
1321 1 : assert_eq!(path, "/mnt/external/test/abc.0123.parquet");
1322 1 : assert!(graph.nodes.as_ref().unwrap()[1].sql.as_ref().unwrap().contains("external('/mnt/external/test/abc.0123.parquet'"));
1323 1 : }
1324 :
1325 : #[test]
1326 1 : fn rejects_duplicate_scripts_and_attachments() {
1327 1 : let catalog = Catalog::new();
1328 1 : let a = analyzed(&catalog, "select 1");
1329 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1330 : // The same script under two names
1331 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }, PlannerScript { script: &a, name: "bar", connection: 0, materialization_file: None }];
1332 1 : let graph = plan_session(&request(&connections, &scripts));
1333 1 : assert!(!graph.valid);
1334 1 : assert_eq!(messages(&graph), vec!["duplicate planner script"]);
1335 : // Two attachments of one file
1336 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }];
1337 1 : let attachments = [PlannerAttachment { name: "x", file: "x.parquet", stamp: "1" }, PlannerAttachment { name: "y", file: "x.parquet", stamp: "1" }];
1338 1 : let graph = plan_session(&SessionPlanRequest { attachments: &attachments, ..request(&connections, &scripts) });
1339 1 : assert!(!graph.valid);
1340 1 : assert_eq!(messages(&graph), vec!["duplicate attachment: x.parquet"]);
1341 1 : }
1342 :
1343 : #[test]
1344 1 : fn rejects_analyses_of_an_older_catalog_revision() {
1345 1 : let mut catalog = Catalog::new();
1346 1 : let a = analyzed(&catalog, "select * from t");
1347 : // Publishing another script makes the analysis stale
1348 1 : let schema = analyzed(&catalog, "create table t (x int)");
1349 1 : catalog.load_script(schema.analyzed.clone().unwrap(), dashql_core::catalog::CatalogLayer::SESSION).unwrap();
1350 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1351 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }];
1352 1 : let graph = plan_session(&request(&connections, &scripts));
1353 1 : assert!(!graph.valid);
1354 1 : assert_eq!(messages(&graph), vec!["stale catalog analysis"]);
1355 : // A tolerant plan takes the older analysis
1356 1 : let tolerant = plan_session(&SessionPlanRequest { scope: PlanScope::Tolerant, ..request(&connections, &scripts) });
1357 1 : assert!(!messages(&tolerant).contains(&"stale catalog analysis".to_string()));
1358 1 : }
1359 :
1360 : #[test]
1361 1 : fn rejects_scripts_of_another_catalog() {
1362 1 : let (catalog, other) = (Catalog::new(), Catalog::new());
1363 1 : let a = analyzed(&other, "select 1");
1364 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Local, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1365 1 : let scripts = [PlannerScript { script: &a, name: "foo", connection: 0, materialization_file: None }];
1366 1 : let graph = plan_session(&request(&connections, &scripts));
1367 1 : assert!(!graph.valid);
1368 1 : assert_eq!(messages(&graph), vec!["script was analyzed with another catalog than the one of its connection"]);
1369 1 : }
1370 :
1371 : #[test]
1372 1 : fn caches_the_results_of_scripts_of_queries() {
1373 : // A cache hit skips the whole script, so every statement must be a query
1374 7 : for (text, cacheable) in [
1375 1 : ("select 1", true),
1376 1 : ("explain select 1", true),
1377 1 : ("select 1; select 2", true),
1378 1 : ("create table t as select 1 as a; select a from t", false),
1379 1 : ("set search_path = x; select 1", false),
1380 1 : ("create table t (a int)", false),
1381 1 : ("select 1 +;", false),
1382 1 : ] {
1383 7 : let catalog = Catalog::new();
1384 7 : let mut script = Script::new(&catalog);
1385 7 : script.replace_text(text);
1386 7 : script.scan();
1387 7 : script.parse();
1388 7 : script.analyze(&catalog, false).unwrap();
1389 7 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Remote, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1390 7 : let scripts = [PlannerScript { script: &script, name: "foo", connection: 0, materialization_file: None }];
1391 7 : let mut targets = request(&connections, &scripts);
1392 7 : targets.scope = PlanScope::Targets(&[0]);
1393 7 : let graph = plan_session(&targets);
1394 7 : assert!(graph.valid, "{text}: {:?}", messages(&graph));
1395 7 : assert_eq!(graph.scripts.as_ref().unwrap()[0].cacheable, cacheable, "{text}");
1396 : }
1397 1 : }
1398 :
1399 : #[test]
1400 1 : fn plans_scripts_without_names() {
1401 1 : let catalog = Catalog::new();
1402 1 : let (a, b) = (analyzed(&catalog, "select 1"), analyzed(&catalog, "select 2"));
1403 1 : let connections = [PlannerConnection { id: "c", kind: ConnectionKind::Remote, catalog: &catalog, dialect: FormattingDialect::HYPER }];
1404 1 : let scripts = [PlannerScript { script: &a, name: "", connection: 0, materialization_file: None }, PlannerScript { script: &b, name: "", connection: 0, materialization_file: None }];
1405 1 : let graph = plan_session(&request(&connections, &scripts));
1406 1 : assert!(graph.valid);
1407 1 : assert_eq!(graph.nodes.as_ref().map(Vec::len), Some(2));
1408 1 : assert!(graph.connections.as_ref().is_some_and(|c| c.len() == 1 && c[0].kind == ExecutionGraphConnectionKind::REMOTE));
1409 1 : }
1410 : }
|
