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