Line data Source code
1 : //! The dataflow of a session: which statements read the data that other statements produce.
2 : //!
3 : //! The dataflow view draws every script as a box of its statements and the flow of data between the
4 : //! statements as edges. The dataflow is a projection of a tolerant plan of the session (`PlanScope::Tolerant`),
5 : //! so relations and script references are resolved exactly like a run resolves them:
6 : //!
7 : //! - A table parameter `:foo` reads the result of the terminal query of `foo.sql`, inlined within one connection
8 : //! or scanned as materialized result on the LOCAL connection. Invalid references are edges too, as long as their
9 : //! name finds a script, the editor reports their errors.
10 : //! - A statement reads or writes a relation after another statement created or wrote it: the planner edges
11 : //! `READ_AFTER_WRITE`, `WRITE_AFTER_WRITE` and `DDL` of a relation whose source statement wrote it. A drop
12 : //! reads no data. A statement that reads a relation only through the query of an inlined script reads it through
13 : //! that script, the edge into the statement is left out.
14 : //!
15 : //! - A statement uses an input, a named scalar parameter `:min` (`session_inputs`). The inputs are nodes of their own,
16 : //! an input edge starts at the input and ends at every statement that uses it.
17 : //! - A statement scans an attachment of the session with a table parameter `:sales`. The attachments are nodes of
18 : //! their own, an attachment edge starts at the attachment and ends at the statement of the parameter.
19 : //!
20 : //! The planner edges that only order statements (`WRITE_AFTER_READ`, `BARRIER`, `SEQUENCE`) and the edges of
21 : //! materializations, which the references stand for, are no dataflow.
22 :
23 : use crate::planner::{plan, runs_on_request, PlanScope, Prepared, SessionPlanRequest};
24 : use crate::script_references::ReferenceKind;
25 : use crate::session_inputs::SessionInput;
26 : use dashql_buffers::execution::{ExecutionGraphAccessMode as Mode, ExecutionGraphEdgeReason as Reason, ExecutionGraphNodeT, ExecutionGraphT};
27 : use dashql_buffers::parser::StatementType;
28 : use dashql_parser::span::TextSpan;
29 : use std::collections::{BTreeMap, BTreeSet};
30 :
31 : const NONE: u32 = u32::MAX;
32 :
33 : /// What carries the data of an edge
34 : #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
35 : pub enum DataflowEdgeKind {
36 : /// A table parameter inlines the terminal query of the referenced script
37 : InlineReference,
38 : /// A table parameter scans the materialized result of the referenced script
39 : ScanReference,
40 : /// A table parameter with errors
41 : InvalidReference,
42 : /// A statement reads or writes a relation that another statement created or wrote
43 : Table,
44 : /// A statement uses an input, the script of the source port is the index of the input
45 : Input,
46 : /// A statement scans an attachment, the script of the source port is the index of the attachment
47 : Attachment,
48 : }
49 :
50 : /// An attachment of the session
51 : #[derive(Clone, Debug, PartialEq, Eq)]
52 : pub struct DataflowAttachment {
53 : /// The name of the table parameter, `sales` for `sales.parquet`
54 : pub name: String,
55 : /// The file name
56 : pub file: String,
57 : }
58 :
59 : /// A statement of a script
60 : #[derive(Clone, Debug, PartialEq, Eq)]
61 : pub struct DataflowStatement {
62 : /// The statement type
63 : pub statement_type: StatementType,
64 : /// The relation that the statement creates, writes or drops
65 : pub target: Option<String>,
66 : /// The text of the statement, in UTF-8 bytes
67 : pub text_span: TextSpan,
68 : }
69 :
70 : /// A script of the session
71 : #[derive(Clone, Debug, Default, PartialEq, Eq)]
72 : pub struct DataflowScript {
73 : /// Was the script planned? A script without a current analysis is not, its statements and the edges that start
74 : /// or end at it are unknown.
75 : pub analyzed: bool,
76 : /// The text version of the analysis
77 : pub text_version: u64,
78 : /// The statements in script order
79 : pub statements: Vec<DataflowStatement>,
80 : /// Does the script run only on request, its `@manual` tag?
81 : pub manual: bool,
82 : }
83 :
84 : /// An end of an edge: a statement of a script
85 : #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
86 : pub struct DataflowPort {
87 : /// The script index
88 : pub script: u32,
89 : /// The statement id
90 : pub statement: u32,
91 : }
92 :
93 : /// The flow of data from one statement to another
94 : #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
95 : pub struct DataflowEdge {
96 : /// The statement that produces the data
97 : pub from: DataflowPort,
98 : /// The statement that reads the data
99 : pub to: DataflowPort,
100 : /// What carries the data
101 : pub kind: DataflowEdgeKind,
102 : /// The parameter name of a reference, the relation name of a table edge
103 : pub name: String,
104 : }
105 :
106 : /// The dataflow of a session
107 : #[derive(Clone, Debug, Default, PartialEq)]
108 : pub struct Dataflow {
109 : /// The scripts in session order
110 : pub scripts: Vec<DataflowScript>,
111 : /// The inputs in the order of their first use
112 : pub inputs: Vec<SessionInput>,
113 : /// The attachments of the request, the sources of the attachment edges
114 : pub attachments: Vec<DataflowAttachment>,
115 : /// The edges, sorted and without duplicates
116 : pub edges: Vec<DataflowEdge>,
117 : }
118 :
119 : /// A relation that a node accesses: an entity or an unresolved reference
120 : type Relation = (u32, u32);
121 :
122 159 : fn relation_of(entity: u32, unresolved: u32) -> Option<Relation> {
123 159 : (entity != NONE || unresolved != NONE).then_some((entity, unresolved))
124 159 : }
125 :
126 : /// The name of a relation of the graph
127 95 : fn relation_name(graph: &ExecutionGraphT, (entity, unresolved): Relation) -> String {
128 95 : let name = if entity != NONE {
129 95 : graph.entities.as_ref().and_then(|e| e.get(entity as usize)).and_then(|e| e.name.clone())
130 : } else {
131 0 : graph.unresolved_references.as_ref().and_then(|r| r.get(unresolved as usize)).and_then(|r| r.name.clone())
132 : };
133 95 : name.unwrap_or_default()
134 95 : }
135 :
136 : /// The relations that a node writes or creates
137 186 : fn writes(node: &ExecutionGraphNodeT) -> impl Iterator<Item = Relation> + '_ {
138 186 : node.accesses
139 186 : .iter()
140 186 : .flatten()
141 186 : .filter(|a| a.mode != Mode::READ)
142 186 : .filter_map(|a| relation_of(a.entity, a.unresolved_reference))
143 186 : }
144 :
145 : /// Derive the dataflow of the scripts of a session. Ignores the scope and the materializations of the request.
146 41 : pub fn derive_session_dataflow(request: &SessionPlanRequest) -> Dataflow {
147 41 : let tolerant = SessionPlanRequest { scope: PlanScope::Tolerant, materialized: &[], ..*request };
148 41 : let plan = plan(&tolerant);
149 41 : let graph = &plan.graph;
150 41 : let empty = Vec::new();
151 41 : let nodes = graph.nodes.as_ref().unwrap_or(&empty);
152 132 : let port = |node: &ExecutionGraphNodeT| DataflowPort { script: node.script, statement: node.statement_id };
153 130 : let node_of: BTreeMap<(u32, u32), u32> = nodes.iter().enumerate().map(|(i, n)| ((n.script, n.statement_id), i as u32)).collect();
154 :
155 : // The statements, from the nodes of the plan
156 41 : let mut scripts: Vec<DataflowScript> = vec![DataflowScript::default(); request.scripts.len()];
157 130 : for (node, &text_span) in nodes.iter().zip(&plan.spans) {
158 130 : let Some(script) = scripts.get_mut(node.script as usize) else { continue };
159 130 : script.analyzed = true;
160 130 : script.statements.push(DataflowStatement {
161 130 : statement_type: node.statement_type,
162 130 : target: writes(node).next().map(|r| relation_name(graph, r)),
163 130 : text_span,
164 : });
165 : }
166 113 : for (i, script) in scripts.iter_mut().enumerate() {
167 113 : let input = request.scripts[i].script;
168 : // A script without statements has no nodes, it is planned if its analysis is current
169 113 : if !script.analyzed {
170 3 : script.analyzed = Prepared::lenient(input).is_some_and(|p| p.parsed.statements.is_empty());
171 110 : }
172 113 : if script.analyzed {
173 110 : script.text_version = input.analyzed.as_ref().map_or(0, |a| a.parsed.scanned.text_version);
174 110 : script.manual = runs_on_request(input);
175 3 : }
176 : }
177 :
178 41 : let mut edges: BTreeSet<DataflowEdge> = BTreeSet::new();
179 : // The terminal node of a script
180 41 : let terminal_of = |script: u32| -> Option<u32> {
181 28 : let meta = graph.scripts.as_ref()?.get(script as usize)?;
182 28 : node_of.get(&(script, meta.terminal_statement_id)).copied()
183 28 : };
184 :
185 : // Script references and attachments
186 41 : for reference in graph.references.iter().flatten() {
187 18 : if reference.attachment != NONE {
188 2 : let Some(to) = nodes.get(reference.node as usize) else { continue };
189 2 : edges.insert(DataflowEdge {
190 2 : from: DataflowPort { script: reference.attachment, statement: 0 },
191 2 : to: port(to),
192 2 : kind: DataflowEdgeKind::Attachment,
193 2 : name: reference.parameter.clone().unwrap_or_default(),
194 2 : });
195 2 : continue;
196 16 : }
197 16 : let (Some(from), Some(to)) = (terminal_of(reference.script), nodes.get(reference.node as usize)) else { continue };
198 16 : edges.insert(DataflowEdge {
199 16 : from: port(&nodes[from as usize]),
200 16 : to: port(to),
201 16 : kind: if reference.materialization != NONE { DataflowEdgeKind::ScanReference } else { DataflowEdgeKind::InlineReference },
202 16 : name: reference.parameter.clone().unwrap_or_default(),
203 : });
204 : }
205 113 : for consumer in 0..request.scripts.len() as u32 {
206 113 : for reference in plan.references.of(consumer).iter().filter(|r| r.kind == ReferenceKind::Invalid) {
207 6 : let Some(from) = reference.producer.and_then(terminal_of) else { continue };
208 4 : if !node_of.contains_key(&(consumer, reference.statement_id)) {
209 0 : continue;
210 4 : }
211 4 : edges.insert(DataflowEdge {
212 4 : from: port(&nodes[from as usize]),
213 4 : to: DataflowPort { script: consumer, statement: reference.statement_id },
214 4 : kind: DataflowEdgeKind::InvalidReference,
215 4 : name: reference.name.clone(),
216 4 : });
217 : }
218 : }
219 :
220 : // Relations: the edges whose source wrote the relation
221 41 : let mut table_edges: BTreeMap<(u32, Relation), BTreeSet<u32>> = BTreeMap::new();
222 87 : for e in graph.edges.iter().flatten() {
223 87 : if !matches!(e.reason, Reason::READ_AFTER_WRITE | Reason::WRITE_AFTER_WRITE | Reason::DDL) {
224 31 : continue;
225 56 : }
226 56 : let Some(relation) = relation_of(e.entity, e.unresolved_reference) else { continue };
227 56 : let (from, to) = (&nodes[e.from as usize], &nodes[e.to as usize]);
228 : // A drop neither produces nor reads data
229 107 : let drop = |node: &ExecutionGraphNodeT| matches!(node.statement_type, StatementType::DROP_TABLE | StatementType::DROP_VIEW);
230 56 : if !writes(from).any(|r| r == relation) || drop(from) || drop(to) {
231 5 : continue;
232 51 : }
233 51 : table_edges.entry((e.from, relation)).or_default().insert(e.to);
234 : }
235 41 : for (&(from, relation), targets) in &table_edges {
236 51 : for &to in targets {
237 : // A read only through the query of an inlined script flows through that script
238 51 : let node = &nodes[to as usize];
239 51 : let through_inlined = node
240 51 : .inlined_scripts
241 51 : .iter()
242 51 : .flatten()
243 51 : .filter_map(|&script| terminal_of(script))
244 51 : .any(|terminal| terminal != to && targets.contains(&terminal));
245 51 : if through_inlined && plan.inherited.contains(&(to, relation.0, relation.1)) {
246 4 : continue;
247 47 : }
248 47 : edges.insert(DataflowEdge {
249 47 : from: port(&nodes[from as usize]),
250 47 : to: port(&nodes[to as usize]),
251 47 : kind: DataflowEdgeKind::Table,
252 47 : name: relation_name(graph, relation),
253 47 : });
254 : }
255 : }
256 :
257 : // Inputs
258 41 : for (i, input) in plan.inputs.inputs.iter().enumerate() {
259 10 : for consumer in input.consumers.iter().filter(|c| node_of.contains_key(&(c.script, c.statement))) {
260 10 : edges.insert(DataflowEdge {
261 10 : from: DataflowPort { script: i as u32, statement: 0 },
262 10 : to: DataflowPort { script: consumer.script, statement: consumer.statement },
263 10 : kind: DataflowEdgeKind::Input,
264 10 : name: input.name.to_string(),
265 10 : });
266 10 : }
267 : }
268 :
269 41 : let attachments = request.attachments.iter().map(|a| DataflowAttachment { name: a.name.to_string(), file: a.file.to_string() }).collect();
270 41 : Dataflow { scripts, inputs: plan.inputs.inputs, attachments, edges: edges.into_iter().collect() }
271 41 : }
|
