Line data Source code
1 : //! Script references.
2 : //!
3 : //! A table parameter `:foo` references the script `foo.sql` and reads the result of its terminal query. The
4 : //! connections of the two scripts decide what a reference becomes:
5 : //!
6 : //! A referenced script is a single SELECT, a reference never runs commands.
7 : //!
8 : //! - Within one connection the query is inlined into the statement as a common table expression,
9 : //! `with foo as (<query of foo>) select * from foo`.
10 : //! - A statement of the LOCAL connection (the embedded Hyper) scans the result of a script of another
11 : //! connection, which the host writes to a Parquet file and mounts:
12 : //! `with foo as (select * from external('<path>', format => 'parquet')) select * from foo`.
13 : //! - A statement of a REMOTE connection cannot read a script of another connection.
14 : //!
15 : //! A table parameter `:sales` can also read the attachment `sales.parquet` of the session. Only a statement of the
16 : //! LOCAL connection reads attachments, it scans the file below the attachment prefix that the host mounts:
17 : //! `with sales as (select * from external('<prefix>/sales.parquet', format => 'parquet')) select * from sales`.
18 : //!
19 : //! The common table expressions of a statement are flattened: the references of an inlined query are resolved
20 : //! in the context of the statement, every script appears once, after the scripts it references. They go into
21 : //! the outermost SELECT around the references, before the common table expressions the statement already has.
22 :
23 : use crate::compiler::read_executable_text_with;
24 : use crate::planner::{runs_on_request, script_label, ConnectionKind, Prepared, SessionPlanRequest};
25 : use dashql_buffers::execution::ExecutionGraphMaterializationT;
26 : use dashql_buffers::parser::{AttributeKey as Key, NodeType, StatementType};
27 : use dashql_core::analyzer::ParameterKind;
28 : use dashql_core::table_parameters::{colliding_names, find_table_parameter, TableParameterLookup, TableParameterSource};
29 : use dashql_core::utils::quote_identifier;
30 : use dashql_parser::attrs;
31 : use dashql_parser::keywords::Keyword;
32 : use dashql_parser::parser::context::resolve_text_span;
33 : use dashql_parser::span::TextSpan;
34 : use std::collections::{BTreeMap, BTreeSet};
35 :
36 : /// Receives the errors of script references
37 : pub(crate) trait ReferenceErrors {
38 : /// Report an error at the text of a script
39 : fn error(&mut self, script: u32, text_span: Option<TextSpan>, message: String);
40 : }
41 :
42 : /// What a script reference becomes
43 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
44 : pub enum ReferenceKind {
45 : /// The terminal query of the referenced script is inlined into the statement
46 : Inline,
47 : /// The statement scans the materialized result of the referenced script
48 : Scan,
49 : /// The reference has errors, it cannot be planned
50 : Invalid,
51 : /// The statement scans an attachment of the session
52 : Attachment,
53 : }
54 :
55 : /// A table parameter that references a script or an attachment
56 : #[derive(Clone, Debug)]
57 : pub(crate) struct Reference {
58 : /// The statement of the parameter
59 : pub statement_id: u32,
60 : /// The parameter node
61 : pub ast_node_id: u32,
62 : /// The parameter in the text, with the prefix
63 : pub text_span: TextSpan,
64 : /// The parameter name
65 : pub name: String,
66 : /// The referenced script
67 : pub producer: Option<u32>,
68 : /// The referenced attachment
69 : pub attachment: Option<u32>,
70 : /// What the reference becomes
71 : pub kind: ReferenceKind,
72 : }
73 :
74 : impl Reference {
75 : /// What the reference reads
76 929 : pub(crate) fn source(&self) -> Option<TableParameterSource> {
77 929 : match (self.producer, self.attachment) {
78 837 : (Some(producer), _) => Some(TableParameterSource::Script(producer)),
79 76 : (None, Some(attachment)) => Some(TableParameterSource::Attachment(attachment)),
80 16 : (None, None) => None,
81 : }
82 929 : }
83 : }
84 :
85 : /// The label of an attachment in messages, its file name
86 11 : pub(crate) fn attachment_label(request: &SessionPlanRequest, attachment: u32) -> String {
87 11 : match request.attachments.get(attachment as usize) {
88 11 : Some(a) => a.file.to_string(),
89 0 : None => format!("attachment {attachment}"),
90 : }
91 11 : }
92 :
93 : /// The label of what a table parameter reads in messages
94 20 : pub(crate) fn source_label(request: &SessionPlanRequest, source: TableParameterSource) -> String {
95 20 : match source {
96 15 : TableParameterSource::Script(script) => script_label(request, script),
97 5 : TableParameterSource::Attachment(attachment) => attachment_label(request, attachment),
98 : }
99 20 : }
100 :
101 : /// The source of the name `index` among the names of the scripts followed by the names of the attachments
102 516 : fn source_of(request: &SessionPlanRequest, index: usize) -> TableParameterSource {
103 516 : match index.checked_sub(request.scripts.len()) {
104 473 : None => TableParameterSource::Script(index as u32),
105 43 : Some(attachment) => TableParameterSource::Attachment(attachment as u32),
106 : }
107 516 : }
108 :
109 : /// The names of the table parameters of a request: the scripts in session order, then the attachments
110 785 : fn table_parameter_names<'a>(request: &'a SessionPlanRequest) -> impl Iterator<Item = &'a str> + Clone {
111 785 : request.scripts.iter().map(|s| s.name).chain(request.attachments.iter().map(|a| a.name))
112 785 : }
113 :
114 : /// The script references of the scripts of a plan
115 : pub(crate) struct ScriptReferences {
116 : /// The references of every script in text order
117 : by_script: Vec<Vec<Reference>>,
118 : }
119 :
120 : impl ScriptReferences {
121 755 : pub(crate) fn new(script_count: usize) -> Self {
122 755 : Self { by_script: vec![Vec::new(); script_count] }
123 755 : }
124 :
125 : /// The references of a script
126 9230 : pub(crate) fn of(&self, script: u32) -> &[Reference] {
127 9230 : &self.by_script[script as usize]
128 9230 : }
129 :
130 : /// The scripts that a script references, in the order of their first reference
131 914 : pub(crate) fn producers_of(&self, script: u32) -> Vec<u32> {
132 914 : let mut out: Vec<u32> = Vec::new();
133 914 : for producer in self.of(script).iter().filter_map(|r| r.producer) {
134 277 : if !out.contains(&producer) {
135 267 : out.push(producer);
136 267 : }
137 : }
138 914 : out
139 914 : }
140 :
141 : /// Invalidate the references of scripts that were not prepared. A tolerant plan keeps going after their errors and
142 : /// splices no invalid reference.
143 48 : pub(crate) fn invalidate_unprepared(&mut self, prepared: &[Option<Prepared>]) {
144 50 : for r in self.by_script.iter_mut().flatten() {
145 32 : if r.producer.is_some_and(|p| prepared[p as usize].is_none()) {
146 3 : r.kind = ReferenceKind::Invalid;
147 29 : }
148 : }
149 48 : }
150 :
151 : /// Resolve the table parameters of a script to the scripts and attachments they reference
152 1448 : pub(crate) fn resolve(&mut self, reporter: &mut dyn ReferenceErrors, request: &SessionPlanRequest, script: u32, prepared: &Prepared) {
153 1448 : let (scanned, parsed, analyzed) = (prepared.scanned, prepared.parsed, prepared.analyzed);
154 1448 : let consumer = &request.scripts[script as usize];
155 1448 : let mut out = Vec::new();
156 1448 : for parameter in analyzed.parameters.iter().filter(|p| p.kind == ParameterKind::Table) {
157 533 : let text_span = resolve_text_span(&scanned.text, &scanned.symbols, parameter.location);
158 533 : let named = scanned.text.as_bytes().get(text_span.offset as usize) == Some(&b':');
159 533 : let Some(statement_id) = parameter.ast_statement_id else { continue };
160 533 : let label = &scanned.text[text_span.offset as usize..text_span.end() as usize];
161 533 : let name = match (¶meter.name, named) {
162 523 : (Some(name), true) => name.to_string(),
163 : _ => {
164 10 : reporter.error(script, Some(text_span), format!("the positional table parameter {label} cannot reference a script or an attachment"));
165 10 : continue;
166 : }
167 : };
168 523 : let mut reference = Reference {
169 523 : statement_id,
170 523 : ast_node_id: parameter.ast_node_id,
171 523 : text_span,
172 523 : name: name.clone(),
173 523 : producer: None,
174 523 : attachment: None,
175 523 : kind: ReferenceKind::Invalid,
176 523 : };
177 : // The table reference of the parameter, a table parameter cannot be the target of a write
178 523 : let tableref = &parsed.nodes[parsed.nodes[parameter.ast_node_id as usize].parent() as usize];
179 523 : if matches!(tableref.attribute_key(), Key::SQL_INSERT_TARGET | Key::SQL_DELETE_TARGET) {
180 15 : reporter.error(script, Some(text_span), format!("the table parameter {label} cannot be written"));
181 15 : out.push(reference);
182 15 : continue;
183 508 : }
184 : // An exact name, or a single name that differs in case, among the scripts and the attachments
185 508 : let producer = match find_table_parameter(table_parameter_names(request), &name) {
186 488 : TableParameterLookup::Found(i) => match source_of(request, i) {
187 452 : TableParameterSource::Script(producer) => producer as usize,
188 36 : TableParameterSource::Attachment(attachment) => {
189 36 : reference.attachment = Some(attachment);
190 36 : reference.kind = if request.connections[consumer.connection as usize].kind == ConnectionKind::Local {
191 30 : ReferenceKind::Attachment
192 : } else {
193 6 : reporter.error(
194 6 : script,
195 6 : Some(text_span),
196 6 : format!(
197 : "{} runs on the connection {} and cannot read the attachment {}, only the embedded Hyper reads attachments",
198 6 : script_label(request, script),
199 6 : request.connections[consumer.connection as usize].id,
200 6 : attachment_label(request, attachment)
201 : ),
202 : );
203 6 : ReferenceKind::Invalid
204 : };
205 36 : out.push(reference);
206 36 : continue;
207 : }
208 : },
209 : TableParameterLookup::Unknown => {
210 10 : reporter.error(script, Some(text_span), format!("unknown script or attachment {label}"));
211 10 : out.push(reference);
212 10 : continue;
213 : }
214 10 : TableParameterLookup::Ambiguous(matches) => {
215 20 : let labels: Vec<String> = matches.iter().map(|&i| source_label(request, source_of(request, i))).collect();
216 10 : reporter.error(script, Some(text_span), format!("the table parameter {label} matches {}", labels.join(" and ")));
217 10 : out.push(reference);
218 10 : continue;
219 : }
220 : };
221 452 : reference.producer = Some(producer as u32);
222 452 : let producer_connection = request.scripts[producer].connection;
223 452 : let connection = |c: u32| request.connections[c as usize].id;
224 452 : reference.kind = if producer_connection == consumer.connection {
225 356 : ReferenceKind::Inline
226 96 : } else if runs_on_request(request.scripts[producer].script) {
227 : // Reading the result on another connection would run the script in full
228 5 : reporter.error(
229 5 : script,
230 5 : Some(text_span),
231 5 : format!(
232 : "{} runs only on request (@manual), the scripts of the connection {} inline it but {} of the connection {} cannot read it",
233 5 : script_label(request, producer as u32),
234 5 : connection(producer_connection),
235 5 : script_label(request, script),
236 5 : connection(consumer.connection)
237 : ),
238 : );
239 5 : ReferenceKind::Invalid
240 91 : } else if request.connections[consumer.connection as usize].kind == ConnectionKind::Local {
241 80 : ReferenceKind::Scan
242 : } else {
243 11 : reporter.error(
244 11 : script,
245 11 : Some(text_span),
246 11 : format!(
247 : "{} runs on the connection {} and cannot read {} of the connection {}",
248 11 : script_label(request, script),
249 11 : connection(consumer.connection),
250 11 : script_label(request, producer as u32),
251 11 : connection(producer_connection)
252 : ),
253 : );
254 11 : ReferenceKind::Invalid
255 : };
256 452 : out.push(reference);
257 : }
258 1448 : self.by_script[script as usize] = out;
259 1448 : }
260 :
261 : /// The scripts that references read, every script after the scripts it references, with what reads them: the
262 : /// producer of a reference and, through the terminal query of an inlined script, the producers of its references
263 628 : fn closure<'r>(&self, prepared: &[Option<Prepared>], roots: impl IntoIterator<Item = &'r Reference>) -> Vec<(u32, ReferenceKind)> {
264 773 : fn visit(this: &ScriptReferences, prepared: &[Option<Prepared>], r: &Reference, visited: &mut BTreeSet<u32>, out: &mut Vec<(u32, ReferenceKind)>) {
265 773 : let Some(producer) = r.producer else { return };
266 742 : if r.kind == ReferenceKind::Invalid || !visited.insert(producer) {
267 53 : return;
268 689 : }
269 689 : if r.kind == ReferenceKind::Inline {
270 555 : let terminal = prepared[producer as usize].as_ref().and_then(Prepared::terminal);
271 555 : for nested in this.of(producer).iter().filter(|n| Some(n.statement_id) == terminal) {
272 131 : visit(this, prepared, nested, visited, out);
273 131 : }
274 134 : }
275 689 : out.push((producer, r.kind));
276 773 : }
277 628 : let (mut visited, mut out) = (BTreeSet::new(), Vec::new());
278 642 : for r in roots {
279 642 : visit(self, prepared, r, &mut visited, &mut out);
280 642 : }
281 628 : out
282 628 : }
283 :
284 : /// Check the referenced scripts: they are a single query, references form no cycle, and a view does not scan a
285 : /// materialized result, neither directly nor through the queries it inlines. A reference with errors becomes
286 : /// invalid.
287 698 : pub(crate) fn check(&mut self, reporter: &mut dyn ReferenceErrors, request: &SessionPlanRequest, prepared: &[Option<Prepared>]) {
288 : // The references with errors: (script, index)
289 698 : let mut failed: BTreeSet<(u32, usize)> = BTreeSet::new();
290 698 : let mut reporter = FailedReferences { reporter, failed: &mut failed, current: (0, 0) };
291 698 : let reporter = &mut reporter;
292 1397 : for (script, references) in self.by_script.iter().enumerate() {
293 1397 : let Some(consumer) = prepared[script].as_ref() else { continue };
294 1369 : for (index, r) in references.iter().enumerate() {
295 485 : reporter.current = (script as u32, index);
296 485 : let producer = match r.source() {
297 436 : Some(TableParameterSource::Script(producer)) => {
298 436 : let Some(p) = prepared[producer as usize].as_ref() else { continue };
299 429 : let query = p.parses() && matches!(&p.parsed.statements[..], [statement] if statement.statement_type == StatementType::SELECT);
300 429 : if !query {
301 33 : reporter.error(
302 33 : script as u32,
303 33 : Some(r.text_span),
304 33 : format!("{} is no single SELECT that {} could read", script_label(request, producer), script_label(request, script as u32)),
305 33 : );
306 396 : }
307 429 : Some(producer)
308 : }
309 : // An attachment is a file, it is read like a table
310 33 : Some(TableParameterSource::Attachment(_)) => None,
311 16 : None => continue,
312 : };
313 462 : let anchor = anchor_of(consumer.parsed, r.ast_node_id);
314 462 : if anchor.is_none() {
315 5 : reporter.error(script as u32, Some(r.text_span), "a table parameter must be part of a query".into());
316 457 : }
317 462 : let Some(producer) = producer else { continue };
318 : // The scripts that the reference reads, the nearest first
319 429 : let closure: Vec<(u32, ReferenceKind)> = self.closure(prepared, [r]).into_iter().rev().collect();
320 : // A relation of an inlined query must not resolve to a common table expression of the statement
321 429 : if let (Some(anchor), ReferenceKind::Inline) = (anchor, r.kind) {
322 337 : let visible = visible_ctes(consumer, r.statement_id, anchor);
323 407 : let conflict = closure.iter().filter(|(_, kind)| *kind == ReferenceKind::Inline).find_map(|&(inlined, _)| {
324 397 : let p = prepared[inlined as usize].as_ref()?;
325 397 : let terminal = p.terminal()?;
326 : // The query reads its own common table expressions
327 390 : let (mut own, mut relations) = (BTreeSet::new(), BTreeSet::new());
328 390 : cte_names(p, terminal, &mut own);
329 390 : relation_names(p, terminal, &mut relations);
330 390 : relations.difference(&own).find(|name| visible.contains(*name)).map(|name| (inlined, name.clone()))
331 397 : });
332 337 : if let Some((inlined, relation)) = conflict {
333 10 : reporter.error(
334 10 : script as u32,
335 10 : Some(r.text_span),
336 10 : format!(
337 10 : "the query of {} reads {relation}, which {} defines as a common table expression there",
338 10 : script_label(request, inlined),
339 10 : script_label(request, script as u32)
340 10 : ),
341 10 : );
342 327 : }
343 92 : }
344 429 : let statement_type = consumer.parsed.statements[r.statement_id as usize].statement_type;
345 429 : if statement_type == StatementType::CREATE_VIEW {
346 : // The view scans what the queries it inlines scan
347 10 : let message = |materialized: u32| format!("a view cannot read {} of another connection", script_label(request, materialized));
348 10 : let lifetime = "its result exists only while the statement runs";
349 15 : let scanned = closure.iter().find(|(_, kind)| *kind == ReferenceKind::Scan).map(|&(materialized, _)| materialized);
350 10 : if r.kind == ReferenceKind::Scan {
351 5 : reporter.error(script as u32, Some(r.text_span), format!("{}, {lifetime}", message(producer)));
352 5 : } else if let Some(materialized) = scanned {
353 5 : reporter.error(
354 5 : script as u32,
355 5 : Some(r.text_span),
356 5 : format!("{} through {}, {lifetime}", message(materialized), script_label(request, producer)),
357 5 : );
358 5 : }
359 419 : }
360 : }
361 : }
362 : // Cycles: a reference back to a script on the path
363 : #[derive(Clone, Copy, PartialEq)]
364 : enum Color {
365 : White,
366 : Gray,
367 : Black,
368 : }
369 698 : let mut colors = vec![Color::White; self.by_script.len()];
370 698 : let mut path: Vec<u32> = Vec::new();
371 732 : fn visit(this: &ScriptReferences, reporter: &mut FailedReferences, request: &SessionPlanRequest, script: u32, colors: &mut [Color], path: &mut Vec<u32>) {
372 732 : colors[script as usize] = Color::Gray;
373 732 : path.push(script);
374 732 : for (index, r) in this.of(script).iter().enumerate() {
375 485 : let Some(producer) = r.producer else { continue };
376 436 : match colors[producer as usize] {
377 : Color::Gray => {
378 17 : let begin = path.iter().position(|&s| s == producer).unwrap();
379 51 : let cycle: Vec<String> = path[begin..].iter().chain(std::iter::once(&producer)).map(|&s| script_label(request, s)).collect();
380 17 : reporter.current = (script, index);
381 17 : reporter.error(script, Some(r.text_span), format!("cyclic script reference: {}", cycle.join(" -> ")));
382 : }
383 314 : Color::White => visit(this, reporter, request, producer, colors, path),
384 105 : Color::Black => {}
385 : }
386 : }
387 732 : path.pop();
388 732 : colors[script as usize] = Color::Black;
389 732 : }
390 1397 : for script in 0..self.by_script.len() as u32 {
391 1397 : if colors[script as usize] == Color::White && !self.of(script).is_empty() {
392 418 : visit(self, reporter, request, script, &mut colors, &mut path);
393 979 : }
394 : }
395 698 : for (script, index) in failed {
396 75 : self.by_script[script as usize][index].kind = ReferenceKind::Invalid;
397 75 : }
398 698 : }
399 : }
400 :
401 : /// Passes the errors of the checked reference on and remembers that it failed
402 : struct FailedReferences<'r> {
403 : reporter: &'r mut dyn ReferenceErrors,
404 : failed: &'r mut BTreeSet<(u32, usize)>,
405 : /// The checked reference: (script, index)
406 : current: (u32, usize),
407 : }
408 :
409 : impl ReferenceErrors for FailedReferences<'_> {
410 75 : fn error(&mut self, script: u32, text_span: Option<TextSpan>, message: String) {
411 75 : self.failed.insert(self.current);
412 75 : self.reporter.error(script, text_span, message);
413 75 : }
414 : }
415 :
416 : /// A checked table parameter that references a script or an attachment
417 : #[derive(Clone, Debug, PartialEq, Eq)]
418 : pub struct CheckedReference {
419 : /// The statement of the parameter, absent for the errors of parameters that reference no script
420 : pub statement_id: Option<u32>,
421 : /// The parameter in the text of the script, with the prefix, in UTF-8 bytes
422 : pub text_span: TextSpan,
423 : /// The parameter name, empty for a positional parameter
424 : pub parameter: String,
425 : /// The referenced script, if the name finds one
426 : pub script: Option<u32>,
427 : /// The referenced attachment, if the name finds one
428 : pub attachment: Option<u32>,
429 : /// What the reference reads
430 : pub kind: ReferenceKind,
431 : /// The errors of the reference, the messages of the planner
432 : pub errors: Vec<String>,
433 : }
434 :
435 : /// The checked script references of a script
436 : #[derive(Clone, Debug, Default, PartialEq, Eq)]
437 : pub struct CheckedScript {
438 : /// Was the script checked? A script whose analysis is not current with its text is not.
439 : pub checked: bool,
440 : /// The text version of the analysis that was checked
441 : pub text_version: u64,
442 : /// The table parameters of the script in text order
443 : pub references: Vec<CheckedReference>,
444 : }
445 :
446 : /// Collects the errors of script references by script and text
447 : #[derive(Default)]
448 : struct CollectedErrors {
449 : errors: BTreeMap<(u32, u32, u32), Vec<String>>,
450 : }
451 :
452 : impl ReferenceErrors for CollectedErrors {
453 56 : fn error(&mut self, script: u32, text_span: Option<TextSpan>, message: String) {
454 56 : let span = text_span.unwrap_or_default();
455 56 : let messages = self.errors.entry((script, span.offset, span.length)).or_default();
456 56 : if !messages.contains(&message) {
457 56 : messages.push(message);
458 56 : }
459 56 : }
460 : }
461 :
462 : /// Check the script references of every script of a session whose analysis is current with its text, like a plan
463 : /// of the session would. Other scripts may be broken or stale meanwhile: a referenced script without a current
464 : /// analysis is read as a single SELECT. Ignores the scope, the order of the scripts and the catalogs.
465 277 : pub fn check_script_references(request: &SessionPlanRequest) -> Vec<CheckedScript> {
466 277 : let count = request.scripts.len();
467 542 : let prepared: Vec<Option<Prepared>> = request.scripts.iter().map(|s| Prepared::lenient(s.script)).collect();
468 277 : let mut references = ScriptReferences::new(count);
469 277 : let mut errors = CollectedErrors::default();
470 542 : for (i, p) in prepared.iter().enumerate() {
471 542 : if let Some(p) = p {
472 534 : references.resolve(&mut errors, request, i as u32, p);
473 534 : }
474 : }
475 277 : references.check(&mut errors, request, &prepared);
476 277 : let mut out = Vec::with_capacity(count);
477 542 : for (i, p) in prepared.iter().enumerate() {
478 542 : let Some(p) = p else {
479 8 : out.push(CheckedScript::default());
480 8 : continue;
481 : };
482 534 : let mut checked: Vec<CheckedReference> = references
483 534 : .of(i as u32)
484 534 : .iter()
485 534 : .map(|r| {
486 202 : CheckedReference {
487 202 : statement_id: Some(r.statement_id),
488 202 : text_span: r.text_span,
489 202 : parameter: r.name.clone(),
490 202 : script: r.producer,
491 202 : attachment: r.attachment,
492 202 : kind: r.kind,
493 202 : errors: errors.errors.remove(&(i as u32, r.text_span.offset, r.text_span.length)).unwrap_or_default(),
494 202 : }
495 202 : })
496 534 : .collect();
497 : // Errors of parameters that reference no script, e.g. positional ones
498 534 : let rest: Vec<_> = errors.errors.keys().filter(|(s, _, _)| *s == i as u32).copied().collect();
499 534 : for key in rest {
500 4 : let (_, offset, length) = key;
501 4 : checked.push(CheckedReference {
502 4 : statement_id: None,
503 4 : text_span: TextSpan { offset, length },
504 4 : parameter: String::new(),
505 4 : script: None,
506 4 : attachment: None,
507 4 : kind: ReferenceKind::Invalid,
508 4 : errors: errors.errors.remove(&key).unwrap_or_default(),
509 4 : });
510 4 : }
511 534 : checked.sort_by_key(|r| (r.text_span.offset, r.text_span.length));
512 534 : out.push(CheckedScript { checked: true, text_version: p.scanned.text_version, references: checked });
513 : }
514 277 : out
515 277 : }
516 :
517 : /// Scripts and attachments whose names are equal in case
518 : #[derive(Clone, Debug, PartialEq, Eq)]
519 : pub struct TableParameterCollision {
520 : /// The name of the first of them
521 : pub name: String,
522 : /// The scripts with the name
523 : pub scripts: Vec<u32>,
524 : /// The attachments with the name
525 : pub attachments: Vec<u32>,
526 : }
527 :
528 : /// The names of the scripts and attachments of a session that collide. A table parameter with such a name is
529 : /// ambiguous, at least in case.
530 277 : pub fn table_parameter_collisions(request: &SessionPlanRequest) -> Vec<TableParameterCollision> {
531 277 : let names: Vec<&str> = table_parameter_names(request).collect();
532 277 : colliding_names(names.iter().copied())
533 277 : .into_iter()
534 277 : .map(|group| {
535 4 : let mut collision = TableParameterCollision { name: names[group[0]].to_string(), scripts: Vec::new(), attachments: Vec::new() };
536 8 : for i in group {
537 8 : match source_of(request, i) {
538 6 : TableParameterSource::Script(s) => collision.scripts.push(s),
539 2 : TableParameterSource::Attachment(a) => collision.attachments.push(a),
540 : }
541 : }
542 4 : collision
543 4 : })
544 277 : .collect()
545 277 : }
546 :
547 : /// A reference of a spliced statement
548 : pub(crate) struct SplicedReference {
549 : /// The reference
550 : pub reference: Reference,
551 : /// The name of the common table expression
552 : pub cte_name: String,
553 : /// The materialization that the reference scans
554 : pub materialization: Option<u32>,
555 : }
556 :
557 : /// A statement with its script references spliced in
558 : #[derive(Default)]
559 : pub(crate) struct SplicedStatement {
560 : /// The edits of the statement text
561 : pub edits: Vec<(TextSpan, String)>,
562 : /// The references of the statement
563 : pub references: Vec<SplicedReference>,
564 : /// The queries of the inlined scripts: (script, statement)
565 : pub inlined: Vec<(u32, u32)>,
566 : /// The materializations that the statement scans
567 : pub inputs: Vec<u32>,
568 : /// The attachments that the statement scans, also those that the inlined queries read
569 : pub attachments: Vec<u32>,
570 : /// The values of inputs that the inlined queries don't read: (script, text span, message)
571 : pub input_errors: Vec<(u32, TextSpan, String)>,
572 : }
573 :
574 : /// The path of an attachment that a statement scans
575 57 : pub(crate) fn attachment_path(request: &SessionPlanRequest, attachment: u32) -> String {
576 57 : format!("{}/{}", request.attachment_prefix, request.attachments[attachment as usize].file)
577 57 : }
578 :
579 : /// The text of a CTE name or a script name as identifier
580 451 : fn identifier(name: &str) -> String {
581 451 : let quoted = quote_identifier(name);
582 451 : if quoted == name && Keyword::find(name.as_bytes()).is_some() {
583 6 : return format!("\"{name}\"");
584 445 : }
585 445 : quoted
586 451 : }
587 :
588 : /// The relation names that a statement uses unqualified, lower-cased
589 1094 : fn relation_names(prepared: &Prepared, statement_id: u32, out: &mut BTreeSet<String>) {
590 1094 : for table_ref in prepared.analyzed.table_references.iter().filter(|t| t.ast_statement_id == Some(statement_id) && !t.reads_table_parameter()) {
591 563 : let Some(relation) = &table_ref.inner else { continue };
592 563 : let name = &relation.table_name;
593 563 : if name.database_name.text.is_empty() && name.schema_name.text.is_empty() {
594 563 : out.insert(name.table_name.text.to_lowercase());
595 563 : }
596 : }
597 1094 : }
598 :
599 : /// The common table expressions of a statement: the node, its name lower-cased
600 1079 : fn ctes<'a>(prepared: &'a Prepared, statement_id: u32) -> impl Iterator<Item = (u32, String)> + 'a {
601 1079 : let (scanned, parsed) = (prepared.scanned, prepared.parsed);
602 1079 : let statement = &parsed.statements[statement_id as usize];
603 17422 : (statement.nodes_begin..statement.nodes_begin + statement.node_count).filter_map(move |ni| {
604 17422 : let node = &parsed.nodes[ni as usize];
605 17422 : if node.node_type() != NodeType::OBJECT_SQL_CTE {
606 17375 : return None;
607 47 : }
608 47 : let [name] = attrs!(parsed.children(node), [Key::SQL_CTE_NAME]);
609 47 : let span = resolve_text_span(&scanned.text, &scanned.symbols, *name?.symbol_span());
610 47 : let text = &scanned.text[span.offset as usize..span.end() as usize];
611 47 : Some((ni, text.trim_matches('"').to_lowercase()))
612 17422 : })
613 1079 : }
614 :
615 : /// The names of the common table expressions of a statement, lower-cased
616 742 : fn cte_names(prepared: &Prepared, statement_id: u32, out: &mut BTreeSet<String>) {
617 742 : out.extend(ctes(prepared, statement_id).map(|(_, name)| name));
618 742 : }
619 :
620 : /// The relation names that a statement uses unqualified and the names of its common table expressions,
621 : /// lower-cased. A common table expression of a script must not shadow them.
622 352 : fn used_names(prepared: &Prepared, statement_id: u32, out: &mut BTreeSet<String>) {
623 352 : relation_names(prepared, statement_id, out);
624 352 : cte_names(prepared, statement_id, out);
625 352 : }
626 :
627 : /// The names of the common table expressions of a statement that the common table expressions of the inlined queries
628 : /// see at an anchor, lower-cased: those of an enclosing WITH, e.g. the one of an INSERT, and those of the WITH of the
629 : /// anchor if it is recursive. The common table expressions go before the ones of a WITH that is not recursive.
630 337 : fn visible_ctes(prepared: &Prepared, statement_id: u32, anchor: u32) -> BTreeSet<String> {
631 337 : let nodes = &prepared.parsed.nodes;
632 337 : let encloses = |owner: u32| {
633 5 : let mut node_id = anchor;
634 : loop {
635 5 : let parent = nodes[node_id as usize].parent();
636 5 : if parent == node_id {
637 0 : return false;
638 5 : }
639 5 : if parent == owner {
640 5 : return true;
641 0 : }
642 0 : node_id = parent;
643 : }
644 5 : };
645 337 : let mut out = BTreeSet::new();
646 337 : for (cte, name) in ctes(prepared, statement_id) {
647 : // The common table expression is an element of the WITH array of its owner
648 30 : let owner = nodes[nodes[cte as usize].parent() as usize].parent();
649 30 : let visible = if owner == anchor {
650 25 : let [recursive] = attrs!(prepared.parsed.children(&nodes[owner as usize]), [Key::SQL_SELECT_WITH_RECURSIVE]);
651 25 : recursive.is_some()
652 : } else {
653 5 : encloses(owner)
654 : };
655 30 : if visible {
656 15 : out.insert(name);
657 15 : }
658 : }
659 337 : out
660 337 : }
661 :
662 : /// The offset of a query within the parentheses that enclose all of it, `(select ...)` is the query of a subquery.
663 : /// `(select 1) union (select 2)` is enclosed by no parentheses.
664 200 : fn inside_parentheses(text: &[u8], span: TextSpan) -> u32 {
665 200 : let (mut begin, mut end) = (span.offset as usize, (span.end() as usize).min(text.len()));
666 : loop {
667 214 : while begin < end && text[begin].is_ascii_whitespace() {
668 3 : begin += 1;
669 3 : }
670 214 : while end > begin && text[end - 1].is_ascii_whitespace() {
671 3 : end -= 1;
672 3 : }
673 211 : if begin >= end || text[begin] != b'(' || text[end - 1] != b')' {
674 198 : return begin as u32;
675 13 : }
676 : // Does the opening parenthesis close at the end? Quoted text and comments do not count.
677 13 : let mut depth = 0usize;
678 13 : let mut i = begin;
679 13 : let mut closes_at_end = false;
680 177 : while i < end {
681 177 : match text[i] {
682 : b'\'' | b'"' => {
683 2 : let quote = text[i];
684 2 : i += 1;
685 4 : while i < end && text[i] != quote {
686 2 : i += 1;
687 2 : }
688 : }
689 1 : b'-' if text.get(i + 1) == Some(&b'-') => {
690 5 : while i < end && text[i] != b'\n' {
691 4 : i += 1;
692 4 : }
693 : }
694 2 : b'/' if text.get(i + 1) == Some(&b'*') => {
695 2 : i += 2;
696 8 : while i + 1 < end && !(text[i] == b'*' && text[i + 1] == b'/') {
697 6 : i += 1;
698 6 : }
699 2 : i += 1;
700 : }
701 14 : b'(' => depth += 1,
702 : b')' => {
703 14 : depth -= 1;
704 14 : if depth == 0 {
705 13 : closes_at_end = i == end - 1;
706 13 : break;
707 1 : }
708 : }
709 144 : _ => {}
710 : }
711 164 : i += 1;
712 : }
713 13 : if !closes_at_end {
714 2 : return begin as u32;
715 11 : }
716 11 : begin += 1;
717 11 : end -= 1;
718 : }
719 200 : }
720 :
721 : /// The outermost SELECT around a node within its statement
722 675 : fn anchor_of(parsed: &dashql_parser::parser::ParsedScript, ast_node_id: u32) -> Option<u32> {
723 675 : let mut anchor = None;
724 675 : let mut node_id = ast_node_id;
725 : loop {
726 3120 : let node = &parsed.nodes[node_id as usize];
727 3120 : if node.node_type() == NodeType::OBJECT_SQL_SELECT {
728 734 : anchor = Some(node_id);
729 2386 : }
730 3120 : let parent = node.parent();
731 3120 : if parent == node_id {
732 675 : break;
733 2445 : }
734 2445 : node_id = parent;
735 : }
736 675 : anchor
737 675 : }
738 :
739 : /// Splice the script references of a statement into its text
740 1054 : pub(crate) fn splice_statement(
741 1054 : request: &SessionPlanRequest,
742 1054 : prepared: &[Option<Prepared>],
743 1054 : references: &ScriptReferences,
744 1054 : script: u32,
745 1054 : statement_id: u32,
746 1054 : materialization_of: &BTreeMap<u32, u32>,
747 1054 : materializations: &[ExecutionGraphMaterializationT],
748 1054 : inputs: &crate::session_inputs::SessionInputs,
749 1054 : ) -> SplicedStatement {
750 1054 : let mut out = SplicedStatement::default();
751 1054 : let direct: Vec<&Reference> = references.of(script).iter().filter(|r| r.statement_id == statement_id && r.kind != ReferenceKind::Invalid).collect();
752 1054 : if direct.is_empty() {
753 855 : return out;
754 199 : }
755 199 : let consumer = prepared[script as usize].as_ref().unwrap();
756 199 : let (scanned, parsed) = (consumer.scanned, consumer.parsed);
757 :
758 : // The scripts of the statement, every script after the scripts it references
759 199 : let order = references.closure(prepared, direct.iter().copied());
760 :
761 : // The names of the common table expressions, distinct from the names the statement and the inlined queries use
762 199 : let mut used = BTreeSet::new();
763 199 : used_names(consumer, statement_id, &mut used);
764 199 : for &(producer, kind) in &order {
765 197 : if kind == ReferenceKind::Inline {
766 153 : let p = prepared[producer as usize].as_ref().unwrap();
767 153 : if let Some(terminal) = p.terminal() {
768 153 : used_names(p, terminal, &mut used);
769 153 : }
770 44 : }
771 : }
772 : // The attachments that the statement and the inlined queries scan, in the order of their first reference
773 199 : let mut attachments: Vec<u32> = Vec::new();
774 : {
775 233 : let mut read = |r: &Reference| {
776 233 : if let (ReferenceKind::Attachment, Some(a)) = (r.kind, r.attachment) {
777 24 : if !attachments.contains(&a) {
778 23 : attachments.push(a);
779 23 : }
780 209 : }
781 233 : };
782 213 : direct.iter().for_each(|r| read(r));
783 199 : for &(producer, kind) in &order {
784 197 : let terminal = prepared[producer as usize].as_ref().and_then(Prepared::terminal);
785 197 : if kind == ReferenceKind::Inline && terminal.is_some() {
786 153 : references.of(producer).iter().filter(|r| Some(r.statement_id) == terminal).for_each(&mut read);
787 44 : }
788 : }
789 : }
790 : // What the common table expressions stand for: the attachments first, the inlined queries may read them, then the
791 : // scripts, every script after the scripts it references
792 199 : let sources: Vec<TableParameterSource> =
793 199 : attachments.iter().map(|&a| TableParameterSource::Attachment(a)).chain(order.iter().map(|&(p, _)| TableParameterSource::Script(p))).collect();
794 451 : let name_of = |source: TableParameterSource| match source {
795 404 : TableParameterSource::Script(script) => request.scripts[script as usize].name,
796 47 : TableParameterSource::Attachment(attachment) => request.attachments[attachment as usize].name,
797 451 : };
798 :
799 : // A table parameter is no relation of the statement, a relation of the same name as a script would be shadowed
800 199 : let mut cte_names: BTreeMap<TableParameterSource, String> = BTreeMap::new();
801 199 : let mut taken = used;
802 220 : for &source in &sources {
803 220 : let name = name_of(source);
804 229 : let shadows = |candidate: &str| taken.contains(&candidate.to_lowercase());
805 220 : let mut candidate = name.to_string();
806 220 : if shadows(&candidate) {
807 9 : let mut n = 2;
808 : loop {
809 9 : candidate = format!("{name}_{n}");
810 9 : if !shadows(&candidate) {
811 9 : break;
812 0 : }
813 0 : n += 1;
814 : }
815 211 : }
816 220 : taken.insert(candidate.to_lowercase());
817 220 : cte_names.insert(source, candidate);
818 : }
819 :
820 : // The edits of the references of a statement text: the parameter becomes the name of the common table expression,
821 : // a renamed one keeps the name of the script as alias unless the statement reads a relation of that name
822 352 : let reference_edits = |p: &Prepared, statement: u32, refs: &[Reference]| -> Vec<(TextSpan, String)> {
823 352 : let mut edits = Vec::new();
824 352 : let mut relations = BTreeSet::new();
825 352 : relation_names(p, statement, &mut relations);
826 352 : for r in refs.iter().filter(|r| r.statement_id == statement && r.kind != ReferenceKind::Invalid) {
827 231 : let Some(source) = r.source() else { continue };
828 231 : let cte = &cte_names[&source];
829 231 : let tableref = &p.parsed.nodes[p.parsed.nodes[r.ast_node_id as usize].parent() as usize];
830 231 : let [alias] = attrs!(p.parsed.children(tableref), [Key::SQL_TABLEREF_ALIAS]);
831 231 : let has_alias = alias.is_some_and(|a| a.node_type() != NodeType::NONE);
832 231 : let script_name = name_of(source);
833 231 : let keeps_name = has_alias || cte == script_name || relations.contains(&script_name.to_lowercase());
834 231 : let text = if keeps_name { identifier(cte) } else { format!("{} as {}", identifier(cte), identifier(script_name)) };
835 231 : edits.push((r.text_span, text));
836 : }
837 352 : edits
838 352 : };
839 :
840 : // The common table expressions in order
841 199 : let mut ctes: Vec<String> = Vec::new();
842 199 : for &attachment in &attachments {
843 23 : let cte = identifier(&cte_names[&TableParameterSource::Attachment(attachment)]);
844 23 : let path = attachment_path(request, attachment);
845 23 : ctes.push(format!("{cte} as (select * from external('{}', format => 'parquet'))", path.replace('\'', "''")));
846 23 : }
847 199 : out.attachments = attachments;
848 199 : for &(producer, kind) in &order {
849 197 : let cte = identifier(&cte_names[&TableParameterSource::Script(producer)]);
850 197 : match kind {
851 : ReferenceKind::Inline => {
852 153 : let p = prepared[producer as usize].as_ref().unwrap();
853 153 : let Some(terminal) = p.terminal() else { continue };
854 153 : let span = p.parsed.associate_descriptions()[terminal as usize].statement_span;
855 153 : let mut edits = reference_edits(p, terminal, references.of(producer));
856 : // The inlined query binds the inputs of its script
857 : // Within the connection of the statement
858 153 : let dialect = request.connections[request.scripts[script as usize].connection as usize].dialect;
859 153 : let bindings = crate::session_inputs::bind_inputs(inputs, p, Some(terminal), dialect);
860 153 : edits.extend(bindings.edits);
861 153 : out.input_errors.extend(bindings.errors.into_iter().map(|(span, message)| (producer, span, message)));
862 153 : let body = read_executable_text_with(p.scanned, span, &edits);
863 153 : ctes.push(format!("{cte} as (\n{}\n)", body.trim_end()));
864 153 : out.inlined.push((producer, terminal));
865 : }
866 : _ => {
867 44 : let Some(&m) = materialization_of.get(&producer) else { continue };
868 44 : let path = materializations[m as usize].path.as_deref().unwrap_or("");
869 44 : ctes.push(format!("{cte} as (select * from external('{}', format => 'parquet'))", path.replace('\'', "''")));
870 44 : if !out.inputs.contains(&m) {
871 44 : out.inputs.push(m);
872 44 : }
873 : }
874 : }
875 : }
876 :
877 : // The common table expressions go into the outermost SELECT around the references, before the ones it has
878 199 : let mut anchors: BTreeMap<u32, Vec<&Reference>> = BTreeMap::new();
879 213 : for r in &direct {
880 213 : if let Some(anchor) = anchor_of(parsed, r.ast_node_id) {
881 213 : anchors.entry(anchor).or_default().push(r);
882 213 : }
883 : }
884 199 : for &anchor in anchors.keys() {
885 199 : let node = &parsed.nodes[anchor as usize];
886 199 : let [with] = attrs!(parsed.children(node), [Key::SQL_SELECT_WITH_CTES]);
887 199 : let first_cte = with.filter(|w| w.node_type() == NodeType::ARRAY && w.children_count() > 0).map(|w| parsed.children(w)[0]);
888 199 : match first_cte {
889 9 : Some(first) => {
890 9 : let at = resolve_text_span(&scanned.text, &scanned.symbols, *first.symbol_span()).offset;
891 9 : out.edits.push((TextSpan::new(at, 0), format!("{},\n", ctes.join(",\n"))));
892 9 : }
893 190 : None => {
894 190 : let span = resolve_text_span(&scanned.text, &scanned.symbols, *node.symbol_span());
895 190 : let at = inside_parentheses(scanned.text.as_bytes(), span);
896 190 : out.edits.push((TextSpan::new(at, 0), format!("with {}\n", ctes.join(",\n"))));
897 190 : }
898 : }
899 : }
900 199 : out.edits.extend(reference_edits(consumer, statement_id, references.of(script)));
901 426 : out.edits.sort_by_key(|(s, _)| (s.offset, s.length));
902 213 : for r in direct {
903 213 : let Some(source) = r.source() else { continue };
904 213 : out.references.push(SplicedReference {
905 213 : reference: r.clone(),
906 213 : cte_name: cte_names[&source].clone(),
907 213 : materialization: match (r.kind, r.producer) {
908 41 : (ReferenceKind::Scan, Some(producer)) => materialization_of.get(&producer).copied(),
909 172 : _ => None,
910 : },
911 : });
912 : }
913 199 : out
914 1054 : }
915 :
916 : #[cfg(test)]
917 : mod tests {
918 : use super::*;
919 :
920 : /// The text from the offset of the query inside the parentheses that enclose all of it
921 10 : fn inside(text: &str) -> &str {
922 10 : let span = TextSpan { offset: 0, length: text.len() as u32 };
923 10 : &text[inside_parentheses(text.as_bytes(), span) as usize..]
924 10 : }
925 :
926 : #[test]
927 1 : fn skips_the_parentheses_that_enclose_all_of_a_query() {
928 1 : assert_eq!(inside("select 1"), "select 1");
929 1 : assert_eq!(inside(" (select 1) "), "select 1) ");
930 1 : assert_eq!(inside("(( select 1 ))"), "select 1 ))");
931 : // The first parenthesis closes before the end
932 1 : assert_eq!(inside("(select 1) union (select 2)"), "(select 1) union (select 2)");
933 1 : assert_eq!(inside("()"), ")");
934 1 : }
935 :
936 : #[test]
937 1 : fn parentheses_in_quotes_and_comments_do_not_count() {
938 1 : assert_eq!(inside("(select ')' as a)"), "select ')' as a)");
939 1 : assert_eq!(inside(r#"(select 1 as ")")"#), r#"select 1 as ")")"#);
940 1 : assert_eq!(inside("(select 1 -- )\n)"), "select 1 -- )\n)");
941 1 : assert_eq!(inside("(select /* ) */ 1)"), "select /* ) */ 1)");
942 1 : assert_eq!(inside("(select 1 /* ) */) union (select 2)"), "(select 1 /* ) */) union (select 2)");
943 1 : }
944 : }
|
