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