Line data Source code
1 : //! The rollup of execution state: slot -> operator / pipeline -> fragment -> query.
2 : //!
3 : //! Every entity takes what the engine reported for it and fills the rest from its children:
4 : //! - an operator from its slots: times, memory and work summed, tuples in summed over all slots, tuples out
5 : //! from the slots that pass tuples on (not the sinks), unless it only has sinks,
6 : //! - a pipeline from its slots: times, memory and work summed, tuples in from the sources, tuples out from the
7 : //! sinks,
8 : //! - a fragment from its pipelines, tuples out from its root operator. What no pipeline knows comes from the
9 : //! operators of the fragment (an analyzed Hyper plan reports operators only),
10 : //! - the query from all fragments, tuples out from the fragments without parent.
11 : //!
12 : //! The state of a fragment is its own work, without its child fragments, like the stage statistics of Trino.
13 : //! Every operator belongs to one fragment only and every pipeline to one fragment, so nothing is counted twice.
14 : //! A status is rolled up over the children that know theirs.
15 :
16 : use super::{Entities, EntityState, Stats, Status};
17 : use crate::query_profile::QueryProfile;
18 : use dashql_buffers::profile as fb;
19 :
20 : /// Add an optional counter
21 17336 : fn add(sum: &mut Option<u64>, value: Option<u64>) {
22 17336 : if let Some(value) = value {
23 3459 : *sum = Some(sum.unwrap_or(0).saturating_add(value));
24 13877 : }
25 17336 : }
26 :
27 : /// The rollup of the children of an entity
28 : #[derive(Default)]
29 : struct Rollup {
30 : /// The rolled up statistics
31 : stats: Stats,
32 : /// The children
33 : children: usize,
34 : /// The children that finished
35 : finished: usize,
36 : /// The latest finish time
37 : finished_at: Option<u64>,
38 : /// The statuses of the children, indexed by `Status as usize`
39 : statuses: [usize; 6],
40 : }
41 :
42 : impl Rollup {
43 : /// Add a child: times, memory and work are summed, the earliest start wins
44 2218 : fn add(&mut self, child: &EntityState) {
45 2218 : let (stats, child_stats) = (&mut self.stats, &child.stats);
46 2218 : self.children += 1;
47 2218 : self.statuses[child.status as usize] += 1;
48 2218 : if let Some(started_at) = child_stats.started_at {
49 55 : stats.started_at = Some(stats.started_at.map_or(started_at, |s| s.min(started_at)));
50 2163 : }
51 2218 : if let Some(finished_at) = child_stats.finished_at {
52 30 : self.finished += 1;
53 30 : self.finished_at = Some(self.finished_at.map_or(finished_at, |f| f.max(finished_at)));
54 2188 : }
55 2218 : add(&mut stats.cpu_time_us, child_stats.cpu_time_us);
56 2218 : add(&mut stats.wall_time_us, child_stats.wall_time_us);
57 2218 : add(&mut stats.blocked_time_us, child_stats.blocked_time_us);
58 2218 : add(&mut stats.memory_bytes, child_stats.memory_bytes);
59 2218 : add(&mut stats.spilled_bytes, child_stats.spilled_bytes);
60 2218 : add(&mut stats.work_done, child_stats.work_done);
61 2218 : add(&mut stats.work_total, child_stats.work_total);
62 2218 : }
63 :
64 : /// Add the input of a child
65 396 : fn add_input(&mut self, child: &Stats) {
66 396 : add(&mut self.stats.tuples_in, child.tuples_in);
67 396 : add(&mut self.stats.bytes_in, child.bytes_in);
68 396 : }
69 :
70 : /// Add the output of a child
71 509 : fn add_output(&mut self, child: &Stats) {
72 509 : add(&mut self.stats.tuples_out, child.tuples_out);
73 509 : add(&mut self.stats.bytes_out, child.bytes_out);
74 509 : }
75 :
76 : /// The rolled up status, over the children that know their status
77 2314 : fn status(&self) -> Status {
78 4089 : let count = |status: Status| self.statuses[status as usize];
79 2314 : let known = self.children - count(Status::Unknown);
80 2314 : if known == 0 {
81 1899 : Status::Unknown
82 415 : } else if count(Status::Failed) > 0 {
83 8 : Status::Failed
84 407 : } else if count(Status::Skipped) == known {
85 0 : Status::Skipped
86 407 : } else if count(Status::Succeeded) + count(Status::Skipped) == known {
87 268 : Status::Succeeded
88 139 : } else if count(Status::Running) > 0 || count(Status::Succeeded) > 0 {
89 139 : Status::Running
90 : } else {
91 0 : Status::Pending
92 : }
93 2314 : }
94 :
95 : /// The rolled up state. An entity finished when all children finished.
96 2314 : fn finish(mut self) -> EntityState {
97 2314 : if self.children > 0 && self.finished == self.children {
98 12 : self.stats.finished_at = self.finished_at;
99 2302 : }
100 2314 : EntityState { status: self.status(), stats: self.stats }
101 2314 : }
102 : }
103 :
104 : /// Merge the report of an entity with the rollup of its children, reported values win
105 2314 : fn merge(reported: &EntityState, rollup: EntityState) -> EntityState {
106 : EntityState {
107 2314 : status: if reported.status != Status::Unknown { reported.status } else { rollup.status },
108 2314 : stats: reported.stats.or(&rollup.stats),
109 : }
110 2314 : }
111 :
112 : /// Compute the effective state of all entities
113 218 : pub(super) fn compute_effective(model: &QueryProfile, reported: &Entities) -> Entities {
114 218 : let mut out = Entities { slots: reported.slots.clone(), ..Default::default() };
115 :
116 : // Operators from their slots
117 218 : out.operators = model
118 218 : .operators
119 218 : .iter()
120 218 : .enumerate()
121 1298 : .map(|(o, op)| {
122 1298 : let mut rollup = Rollup::default();
123 1298 : let passes_on = op.slots.iter().any(|s| model.slots[*s as usize].role != fb::ProfileSlotRole::SINK);
124 1298 : for &s in &op.slots {
125 274 : let slot = &out.slots[s as usize];
126 274 : rollup.add(slot);
127 274 : rollup.add_input(&slot.stats);
128 274 : if !passes_on || model.slots[s as usize].role != fb::ProfileSlotRole::SINK {
129 222 : rollup.add_output(&slot.stats);
130 222 : }
131 : }
132 1298 : merge(&reported.operators[o], rollup.finish())
133 1298 : })
134 218 : .collect();
135 :
136 : // Pipelines from their slots
137 218 : out.pipelines = model
138 218 : .pipelines
139 218 : .iter()
140 218 : .enumerate()
141 218 : .map(|(p, pipeline)| {
142 122 : let mut rollup = Rollup::default();
143 274 : for &s in &pipeline.slots {
144 274 : let slot = &out.slots[s as usize];
145 274 : rollup.add(slot);
146 274 : match model.slots[s as usize].role {
147 122 : fb::ProfileSlotRole::SOURCE => rollup.add_input(&slot.stats),
148 106 : fb::ProfileSlotRole::SINK => rollup.add_output(&slot.stats),
149 46 : _ => {}
150 : }
151 : }
152 122 : merge(&reported.pipelines[p], rollup.finish())
153 122 : })
154 218 : .collect();
155 :
156 : // Fragments from their pipelines, the operators fill what no pipeline knows
157 218 : let mut pipelines_of: Vec<Vec<usize>> = vec![Vec::new(); model.fragments.len()];
158 218 : for (p, pipeline) in model.pipelines.iter().enumerate() {
159 122 : if let Some(f) = pipeline.fragment {
160 122 : pipelines_of[f as usize].push(p);
161 122 : }
162 : }
163 218 : out.fragments = Vec::with_capacity(model.fragments.len());
164 341 : for (f, fragment) in model.fragments.iter().enumerate() {
165 338 : let (mut by_pipelines, mut by_operators) = (Rollup::default(), Rollup::default());
166 338 : for &p in &pipelines_of[f] {
167 122 : by_pipelines.add(&out.pipelines[p]);
168 122 : }
169 1210 : for &o in &fragment.operators {
170 1210 : by_operators.add(&out.operators[o as usize]);
171 1210 : }
172 338 : let mut own = merge(&by_pipelines.finish(), by_operators.finish());
173 338 : if let Some(root) = fragment.root_operator {
174 277 : let root = &out.operators[root as usize].stats;
175 277 : (own.stats.tuples_out, own.stats.bytes_out) = (root.tuples_out, root.bytes_out);
176 277 : }
177 338 : out.fragments.push(merge(&reported.fragments[f], own));
178 : }
179 :
180 : // The query from all fragments, its output is the output of the top-level fragments
181 218 : let mut rollup = Rollup::default();
182 341 : for (f, fragment) in model.fragments.iter().enumerate() {
183 338 : rollup.add(&out.fragments[f]);
184 338 : if fragment.parent_fragment.is_none() {
185 181 : rollup.add_output(&out.fragments[f].stats);
186 187 : }
187 : }
188 218 : out.query = merge(&reported.query, rollup.finish());
189 218 : out
190 218 : }
|
