Line data Source code
1 : //! The execution state of a plan, driven by events.
2 : //!
3 : //! Engines report what they observe as input events (`ProfileInputEvent`): statistics of the query, of fragments,
4 : //! pipelines, slots (an operator in a pipeline) or operators, on whatever level the engine has. The model rolls
5 : //! them up (slot -> operator / pipeline -> fragment -> query, see `rollup`) and emits output events
6 : //! (`ProfileUpdate`, packed as `QueryProfileEvents`) for every entity whose effective state changed. Reported values
7 : //! win over rolled up values, and values that nobody reported stay unknown.
8 : //!
9 : //! Counters are cumulative, so applying an event twice changes nothing and a dropped event is repaired by the
10 : //! next one. The input events have a JSON form (serde), a recorded event log replays an execution.
11 :
12 : mod rollup;
13 : mod trino;
14 :
15 : use super::query_profile::QueryProfile;
16 : use dashql_buffers::profile as fb;
17 : use flatbuffers::FlatBufferBuilder;
18 : use serde::{Deserialize, Serialize};
19 :
20 : /// The execution status of an entity
21 : #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
22 : #[serde(rename_all = "lowercase")]
23 : pub enum Status {
24 : /// Not reported
25 : #[default]
26 : Unknown,
27 : /// Waiting to run
28 : Pending,
29 : /// Running
30 : Running,
31 : /// Finished successfully
32 : Succeeded,
33 : /// Failed
34 : Failed,
35 : /// Skipped
36 : Skipped,
37 : }
38 :
39 : impl Status {
40 : /// The FlatBuffers status
41 415 : pub fn to_fb(self) -> fb::ProfileExecutionStatus {
42 415 : match self {
43 7 : Status::Unknown => fb::ProfileExecutionStatus::UNKNOWN,
44 13 : Status::Pending => fb::ProfileExecutionStatus::PENDING,
45 135 : Status::Running => fb::ProfileExecutionStatus::RUNNING,
46 249 : Status::Succeeded => fb::ProfileExecutionStatus::SUCCEEDED,
47 11 : Status::Failed => fb::ProfileExecutionStatus::FAILED,
48 0 : Status::Skipped => fb::ProfileExecutionStatus::SKIPPED,
49 : }
50 415 : }
51 : }
52 :
53 : /// The execution statistics of an entity, `None` is unknown (not 0).
54 : /// The counters are cumulative.
55 : #[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
56 : #[serde(rename_all = "camelCase", default, deny_unknown_fields)]
57 : pub struct Stats {
58 : /// The time when the entity started in milliseconds since epoch
59 : #[serde(skip_serializing_if = "Option::is_none")]
60 : pub started_at: Option<u64>,
61 : /// The time when the entity finished in milliseconds since epoch
62 : #[serde(skip_serializing_if = "Option::is_none")]
63 : pub finished_at: Option<u64>,
64 : /// The tuples consumed
65 : #[serde(skip_serializing_if = "Option::is_none")]
66 : pub tuples_in: Option<u64>,
67 : /// The tuples produced
68 : #[serde(skip_serializing_if = "Option::is_none")]
69 : pub tuples_out: Option<u64>,
70 : /// The bytes consumed
71 : #[serde(skip_serializing_if = "Option::is_none")]
72 : pub bytes_in: Option<u64>,
73 : /// The bytes produced
74 : #[serde(skip_serializing_if = "Option::is_none")]
75 : pub bytes_out: Option<u64>,
76 : /// The CPU time in microseconds
77 : #[serde(skip_serializing_if = "Option::is_none")]
78 : pub cpu_time_us: Option<u64>,
79 : /// The wall time in microseconds
80 : #[serde(skip_serializing_if = "Option::is_none")]
81 : pub wall_time_us: Option<u64>,
82 : /// The blocked time in microseconds
83 : #[serde(skip_serializing_if = "Option::is_none")]
84 : pub blocked_time_us: Option<u64>,
85 : /// The memory currently attributed to the entity in bytes
86 : #[serde(skip_serializing_if = "Option::is_none")]
87 : pub memory_bytes: Option<u64>,
88 : /// The bytes spilled to disk
89 : #[serde(skip_serializing_if = "Option::is_none")]
90 : pub spilled_bytes: Option<u64>,
91 : /// The completed work units (splits, drivers, morsels)
92 : #[serde(skip_serializing_if = "Option::is_none")]
93 : pub work_done: Option<u64>,
94 : /// The known work units
95 : #[serde(skip_serializing_if = "Option::is_none")]
96 : pub work_total: Option<u64>,
97 : }
98 :
99 : /// The `ProfileStatsField` bit of every field, in the order of `Stats::fields`
100 : const FIELD_BITS: [fb::ProfileStatsField; 13] = [
101 : fb::ProfileStatsField::STARTED_AT,
102 : fb::ProfileStatsField::FINISHED_AT,
103 : fb::ProfileStatsField::TUPLES_IN,
104 : fb::ProfileStatsField::TUPLES_OUT,
105 : fb::ProfileStatsField::BYTES_IN,
106 : fb::ProfileStatsField::BYTES_OUT,
107 : fb::ProfileStatsField::CPU_TIME,
108 : fb::ProfileStatsField::WALL_TIME,
109 : fb::ProfileStatsField::BLOCKED_TIME,
110 : fb::ProfileStatsField::MEMORY_BYTES,
111 : fb::ProfileStatsField::SPILLED_BYTES,
112 : fb::ProfileStatsField::WORK_DONE,
113 : fb::ProfileStatsField::WORK_TOTAL,
114 : ];
115 :
116 : impl Stats {
117 : /// The fields
118 4108 : fn fields(&self) -> [Option<u64>; 13] {
119 4108 : [
120 4108 : self.started_at,
121 4108 : self.finished_at,
122 4108 : self.tuples_in,
123 4108 : self.tuples_out,
124 4108 : self.bytes_in,
125 4108 : self.bytes_out,
126 4108 : self.cpu_time_us,
127 4108 : self.wall_time_us,
128 4108 : self.blocked_time_us,
129 4108 : self.memory_bytes,
130 4108 : self.spilled_bytes,
131 4108 : self.work_done,
132 4108 : self.work_total,
133 4108 : ]
134 4108 : }
135 :
136 : /// The fields, mutable
137 2702 : fn fields_mut(&mut self) -> [&mut Option<u64>; 13] {
138 2702 : [
139 2702 : &mut self.started_at,
140 2702 : &mut self.finished_at,
141 2702 : &mut self.tuples_in,
142 2702 : &mut self.tuples_out,
143 2702 : &mut self.bytes_in,
144 2702 : &mut self.bytes_out,
145 2702 : &mut self.cpu_time_us,
146 2702 : &mut self.wall_time_us,
147 2702 : &mut self.blocked_time_us,
148 2702 : &mut self.memory_bytes,
149 2702 : &mut self.spilled_bytes,
150 2702 : &mut self.work_done,
151 2702 : &mut self.work_total,
152 2702 : ]
153 2702 : }
154 :
155 : /// The reported fields (`ProfileStatsField` bits)
156 1406 : pub fn reported(&self) -> u16 {
157 18278 : self.fields().iter().zip(FIELD_BITS).filter(|(v, _)| v.is_some()).fold(0, |bits, (_, bit)| bits | bit.0)
158 1406 : }
159 :
160 : /// Does the entity report nothing?
161 989 : pub fn is_empty(&self) -> bool {
162 989 : self.reported() == 0
163 989 : }
164 :
165 : /// Overwrite the fields that another report knows
166 388 : pub fn overwrite(&mut self, report: &Stats) {
167 5044 : for (field, value) in self.fields_mut().into_iter().zip(report.fields()) {
168 5044 : if value.is_some() {
169 2445 : *field = value;
170 2924 : }
171 : }
172 388 : }
173 :
174 : /// Fill the unknown fields from a fallback
175 2314 : pub fn or(mut self, fallback: &Stats) -> Stats {
176 30082 : for (field, value) in self.fields_mut().into_iter().zip(fallback.fields()) {
177 30082 : if field.is_none() {
178 28078 : *field = value;
179 28078 : }
180 : }
181 2314 : self
182 2314 : }
183 :
184 : /// Pack as FlatBuffers struct
185 415 : pub fn to_fb(&self) -> fb::ProfileStats {
186 5395 : let v = |field: Option<u64>| field.unwrap_or(0);
187 415 : fb::ProfileStats::new(
188 415 : self.reported(),
189 415 : v(self.started_at),
190 415 : v(self.finished_at),
191 415 : v(self.tuples_in),
192 415 : v(self.tuples_out),
193 415 : v(self.bytes_in),
194 415 : v(self.bytes_out),
195 415 : v(self.cpu_time_us),
196 415 : v(self.wall_time_us),
197 415 : v(self.blocked_time_us),
198 415 : v(self.memory_bytes),
199 415 : v(self.spilled_bytes),
200 415 : v(self.work_done),
201 415 : v(self.work_total),
202 : )
203 415 : }
204 : }
205 :
206 : /// An operator of a pipeline that is registered at runtime
207 : #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
208 : #[serde(rename_all = "camelCase", deny_unknown_fields)]
209 : pub struct SlotSpec {
210 : /// The operator
211 : pub operator: u32,
212 : /// The role of the operator in the pipeline
213 : pub role: SlotRole,
214 : }
215 :
216 : /// The role of an operator in a pipeline
217 : #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
218 : #[serde(rename_all = "lowercase")]
219 : pub enum SlotRole {
220 : /// Produces the tuples of the pipeline
221 : Source,
222 : /// Passes the tuples on
223 : Through,
224 : /// Consumes the tuples of the pipeline
225 : Sink,
226 : }
227 :
228 : impl SlotRole {
229 : /// The FlatBuffers role
230 93 : pub fn to_fb(self) -> fb::ProfileSlotRole {
231 93 : match self {
232 41 : SlotRole::Source => fb::ProfileSlotRole::SOURCE,
233 17 : SlotRole::Through => fb::ProfileSlotRole::THROUGH,
234 35 : SlotRole::Sink => fb::ProfileSlotRole::SINK,
235 : }
236 93 : }
237 : }
238 :
239 : /// An input event: what the engine reported. Entities are addressed by their index in the profile, entities
240 : /// registered at runtime get the next free index.
241 : #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
242 : #[serde(tag = "type", rename_all = "camelCase", rename_all_fields = "camelCase", deny_unknown_fields)]
243 : pub enum ProfileInputEvent {
244 : /// A fragment appeared at runtime (a Trino stage without a plan yet)
245 : RegisterFragment {
246 : /// The event time in milliseconds since epoch
247 : time: u64,
248 : /// The fragment id of the engine
249 : #[serde(default, skip_serializing_if = "Option::is_none")]
250 : engine_id: Option<String>,
251 : /// The enclosing fragment, must exist
252 : #[serde(default, skip_serializing_if = "Option::is_none")]
253 : parent_fragment: Option<u32>,
254 : },
255 : /// A pipeline appeared at runtime (Trino pipelines are known from the operator summaries only)
256 : RegisterPipeline {
257 : /// The event time in milliseconds since epoch
258 : time: u64,
259 : /// The pipeline id of the engine
260 : #[serde(default, skip_serializing_if = "Option::is_none")]
261 : engine_id: Option<String>,
262 : /// The fragment that runs the pipeline
263 : #[serde(default, skip_serializing_if = "Option::is_none")]
264 : fragment: Option<u32>,
265 : /// The operators of the pipeline, they get the next slot indexes in this order
266 : slots: Vec<SlotSpec>,
267 : },
268 : /// The query state
269 : UpdateQuery {
270 : /// The event time in milliseconds since epoch
271 : time: u64,
272 : /// The status, unchanged if missing
273 : #[serde(default, skip_serializing_if = "Option::is_none")]
274 : status: Option<Status>,
275 : /// The reported statistics, missing fields are unchanged
276 : #[serde(default, skip_serializing_if = "Stats::is_empty")]
277 : stats: Stats,
278 : },
279 : /// The state of a fragment
280 : UpdateFragment {
281 : /// The event time in milliseconds since epoch
282 : time: u64,
283 : /// The fragment
284 : fragment: u32,
285 : /// The status, unchanged if missing
286 : #[serde(default, skip_serializing_if = "Option::is_none")]
287 : status: Option<Status>,
288 : /// The reported statistics, missing fields are unchanged
289 : #[serde(default, skip_serializing_if = "Stats::is_empty")]
290 : stats: Stats,
291 : },
292 : /// The state of a pipeline
293 : UpdatePipeline {
294 : /// The event time in milliseconds since epoch
295 : time: u64,
296 : /// The pipeline
297 : pipeline: u32,
298 : /// The status, unchanged if missing
299 : #[serde(default, skip_serializing_if = "Option::is_none")]
300 : status: Option<Status>,
301 : /// The reported statistics, missing fields are unchanged
302 : #[serde(default, skip_serializing_if = "Stats::is_empty")]
303 : stats: Stats,
304 : },
305 : /// The state of a slot (an operator in a pipeline)
306 : UpdateSlot {
307 : /// The event time in milliseconds since epoch
308 : time: u64,
309 : /// The slot
310 : slot: u32,
311 : /// The status, unchanged if missing
312 : #[serde(default, skip_serializing_if = "Option::is_none")]
313 : status: Option<Status>,
314 : /// The reported statistics, missing fields are unchanged
315 : #[serde(default, skip_serializing_if = "Stats::is_empty")]
316 : stats: Stats,
317 : },
318 : /// The state of an operator across its pipelines
319 : UpdateOperator {
320 : /// The event time in milliseconds since epoch
321 : time: u64,
322 : /// The operator
323 : operator: u32,
324 : /// The status, unchanged if missing
325 : #[serde(default, skip_serializing_if = "Option::is_none")]
326 : status: Option<Status>,
327 : /// The reported statistics, missing fields are unchanged
328 : #[serde(default, skip_serializing_if = "Stats::is_empty")]
329 : stats: Stats,
330 : },
331 : }
332 :
333 : impl ProfileInputEvent {
334 : /// The event time
335 509 : pub fn time(&self) -> u64 {
336 509 : match self {
337 23 : ProfileInputEvent::RegisterFragment { time, .. }
338 46 : | ProfileInputEvent::RegisterPipeline { time, .. }
339 101 : | ProfileInputEvent::UpdateQuery { time, .. }
340 146 : | ProfileInputEvent::UpdateFragment { time, .. }
341 0 : | ProfileInputEvent::UpdatePipeline { time, .. }
342 157 : | ProfileInputEvent::UpdateSlot { time, .. }
343 509 : | ProfileInputEvent::UpdateOperator { time, .. } => *time,
344 : }
345 509 : }
346 : }
347 :
348 : /// The state of an entity
349 : #[derive(Clone, Copy, Debug, Default, PartialEq)]
350 : pub struct EntityState {
351 : /// The status
352 : pub status: Status,
353 : /// The statistics
354 : pub stats: Stats,
355 : }
356 :
357 : impl EntityState {
358 : /// Apply a report
359 388 : fn apply(&mut self, status: Option<Status>, stats: &Stats) {
360 388 : if let Some(status) = status {
361 372 : self.status = status;
362 372 : }
363 388 : self.stats.overwrite(stats);
364 388 : }
365 :
366 : /// Is nothing known?
367 560 : fn is_unknown(&self) -> bool {
368 560 : self.status == Status::Unknown && self.stats.is_empty()
369 560 : }
370 : }
371 :
372 : /// An output event: the effective state of an entity changed
373 : #[derive(Clone, Copy, Debug, PartialEq)]
374 : pub struct ProfileUpdate {
375 : /// The event time in milliseconds since epoch
376 : pub time: u64,
377 : /// The entity kind
378 : pub entity: fb::ProfileEntity,
379 : /// The entity index, 0 for the query
380 : pub index: u32,
381 : /// The effective state
382 : pub state: EntityState,
383 : }
384 :
385 : /// The states of all entities
386 : #[derive(Clone, Debug, Default)]
387 : pub(crate) struct Entities {
388 : /// The query
389 : pub(crate) query: EntityState,
390 : /// The fragments
391 : pub(crate) fragments: Vec<EntityState>,
392 : /// The pipelines
393 : pub(crate) pipelines: Vec<EntityState>,
394 : /// The slots
395 : pub(crate) slots: Vec<EntityState>,
396 : /// The operators
397 : pub(crate) operators: Vec<EntityState>,
398 : }
399 :
400 : impl Entities {
401 : /// Resize to the entity counts
402 322 : fn resize(&mut self, fragments: usize, pipelines: usize, slots: usize, operators: usize) {
403 322 : self.fragments.resize(fragments, EntityState::default());
404 322 : self.pipelines.resize(pipelines, EntityState::default());
405 322 : self.slots.resize(slots, EntityState::default());
406 322 : self.operators.resize(operators, EntityState::default());
407 322 : }
408 :
409 : /// The state of an entity
410 2250 : fn get(&self, entity: fb::ProfileEntity, index: u32) -> Option<&EntityState> {
411 2250 : let index = index as usize;
412 2250 : match entity {
413 218 : fb::ProfileEntity::QUERY => Some(&self.query),
414 338 : fb::ProfileEntity::FRAGMENT => self.fragments.get(index),
415 122 : fb::ProfileEntity::PIPELINE => self.pipelines.get(index),
416 274 : fb::ProfileEntity::SLOT => self.slots.get(index),
417 1298 : _ => self.operators.get(index),
418 : }
419 2250 : }
420 :
421 : /// Visit every entity with its kind and index
422 273 : fn each(&self) -> impl Iterator<Item = (fb::ProfileEntity, u32, &EntityState)> {
423 1092 : fn all(
424 1092 : entity: fb::ProfileEntity,
425 1092 : states: &[EntityState],
426 1092 : ) -> impl Iterator<Item = (fb::ProfileEntity, u32, &EntityState)> {
427 2537 : states.iter().enumerate().map(move |(i, s)| (entity, i as u32, s))
428 1092 : }
429 273 : std::iter::once((fb::ProfileEntity::QUERY, 0, &self.query))
430 273 : .chain(all(fb::ProfileEntity::FRAGMENT, &self.fragments))
431 273 : .chain(all(fb::ProfileEntity::PIPELINE, &self.pipelines))
432 273 : .chain(all(fb::ProfileEntity::SLOT, &self.slots))
433 273 : .chain(all(fb::ProfileEntity::OPERATOR, &self.operators))
434 273 : }
435 : }
436 :
437 : /// The execution state of a plan
438 : #[derive(Clone, Debug, Default)]
439 : pub(crate) struct ExecutionState {
440 : /// What the engine reported
441 : pub(crate) reported: Entities,
442 : /// The effective states after rollups
443 : pub(crate) effective: Entities,
444 : /// The time when the effective state of an entity last changed, in the order of `Entities::each`
445 : pub(crate) changed_at: Vec<u64>,
446 : }
447 :
448 : impl ExecutionState {
449 : /// Resize to the entity counts
450 161 : pub(crate) fn resize(&mut self, fragments: usize, pipelines: usize, slots: usize, operators: usize) {
451 161 : self.reported.resize(fragments, pipelines, slots, operators);
452 161 : self.effective.resize(fragments, pipelines, slots, operators);
453 161 : self.changed_at.resize(1 + fragments + pipelines + slots + operators, 0);
454 161 : }
455 : }
456 :
457 : impl QueryProfile {
458 : /// Apply input events and return the updates of the effective state
459 218 : pub fn apply_events(&mut self, events: &[ProfileInputEvent]) -> Vec<ProfileUpdate> {
460 218 : let mut time = 0;
461 452 : for event in events {
462 449 : time = time.max(event.time());
463 449 : self.apply_event(event);
464 449 : }
465 218 : self.recompute(time)
466 218 : }
467 :
468 : /// Replay an event log: consecutive events with the same time form one batch (one poll of the engine).
469 : /// Returns the updates of all batches in order.
470 6 : pub fn replay(&mut self, events: &[ProfileInputEvent]) -> Vec<ProfileUpdate> {
471 30 : events.chunk_by(|a, b| a.time() == b.time()).flat_map(|batch| self.apply_events(batch)).collect()
472 6 : }
473 :
474 : /// Apply one input event, events that reference unknown entities are ignored
475 449 : fn apply_event(&mut self, event: &ProfileInputEvent) {
476 449 : let reported = &mut self.execution.reported;
477 449 : let update = |states: &mut Vec<EntityState>, index: u32, status: &Option<Status>, stats: &Stats| {
478 293 : if let Some(state) = states.get_mut(index as usize) {
479 292 : state.apply(*status, stats);
480 292 : }
481 293 : };
482 449 : match event {
483 18 : ProfileInputEvent::RegisterFragment { engine_id, parent_fragment, .. } => {
484 18 : if parent_fragment.is_some_and(|p| p as usize >= self.fragments.len()) {
485 1 : return;
486 17 : }
487 17 : self.fragments.push(super::query_profile::Fragment {
488 17 : engine_id: engine_id.clone(),
489 17 : parent_fragment: *parent_fragment,
490 17 : ..Default::default()
491 17 : });
492 17 : self.structure_changed();
493 : }
494 42 : ProfileInputEvent::RegisterPipeline { engine_id, fragment, slots, .. } => {
495 42 : if fragment.is_some_and(|f| f as usize >= self.fragments.len())
496 94 : || slots.iter().any(|s| s.operator as usize >= self.operators.len())
497 : {
498 1 : return;
499 41 : }
500 93 : let slots: Vec<_> = slots.iter().map(|s| (s.operator, s.role.to_fb())).collect();
501 41 : self.add_pipeline(engine_id.clone(), *fragment, &slots, Vec::new());
502 41 : self.structure_changed();
503 : }
504 96 : ProfileInputEvent::UpdateQuery { status, stats, .. } => reported.query.apply(*status, stats),
505 138 : ProfileInputEvent::UpdateFragment { fragment, status, stats, .. } => {
506 138 : update(&mut reported.fragments, *fragment, status, stats)
507 : }
508 0 : ProfileInputEvent::UpdatePipeline { pipeline, status, stats, .. } => {
509 0 : update(&mut reported.pipelines, *pipeline, status, stats)
510 : }
511 133 : ProfileInputEvent::UpdateSlot { slot, status, stats, .. } => {
512 133 : update(&mut reported.slots, *slot, status, stats)
513 : }
514 22 : ProfileInputEvent::UpdateOperator { operator, status, stats, .. } => {
515 22 : update(&mut reported.operators, *operator, status, stats)
516 : }
517 : }
518 449 : }
519 :
520 : /// The structure changed at runtime
521 58 : fn structure_changed(&mut self) {
522 58 : self.structure_version = self.structure_version.wrapping_add(1);
523 58 : self.execution.resize(self.fragments.len(), self.pipelines.len(), self.slots.len(), self.operators.len());
524 58 : }
525 :
526 : /// Recompute the effective state and return the entities that changed
527 218 : fn recompute(&mut self, time: u64) -> Vec<ProfileUpdate> {
528 218 : let effective = rollup::compute_effective(self, &self.execution.reported);
529 218 : let previous = std::mem::replace(&mut self.execution.effective, effective);
530 218 : let mut updates = Vec::new();
531 218 : let changed_at = &mut self.execution.changed_at;
532 2250 : for (i, (entity, index, state)) in self.execution.effective.each().enumerate() {
533 : // An entity registered in this batch starts unknown
534 2250 : let before = previous.get(entity, index).copied().unwrap_or_default();
535 2250 : if before == *state {
536 1733 : continue;
537 517 : }
538 517 : changed_at[i] = time;
539 517 : updates.push(ProfileUpdate { time, entity, index, state: *state });
540 : }
541 218 : updates
542 218 : }
543 :
544 : /// The current effective state as updates: one update for every entity that has a state.
545 : /// A renderer that mounts late applies them like any other updates.
546 55 : pub fn snapshot_events(&self) -> Vec<ProfileUpdate> {
547 55 : self.execution
548 55 : .effective
549 55 : .each()
550 55 : .zip(&self.execution.changed_at)
551 560 : .filter(|((_, _, state), _)| !state.is_unknown())
552 122 : .map(|((entity, index, state), time)| ProfileUpdate { time: *time, entity, index, state: *state })
553 55 : .collect()
554 55 : }
555 :
556 : /// Reset the execution state: drop all reported state and the structure that was registered at runtime.
557 : /// Replaying the events of an execution up to a point in time after a reset seeks backwards.
558 1 : pub fn reset_execution(&mut self) {
559 1 : let (fragments, pipelines, slots) = self.loaded_counts;
560 1 : let registered = self.fragments.len() > fragments || self.pipelines.len() > pipelines;
561 1 : self.fragments.truncate(fragments);
562 1 : self.pipelines.truncate(pipelines);
563 1 : self.slots.truncate(slots);
564 4 : for op in &mut self.operators {
565 6 : op.slots.retain(|s| (*s as usize) < slots);
566 : }
567 3 : for edge in &mut self.operator_edges {
568 3 : edge.slot = edge.slot.filter(|s| (*s as usize) < slots);
569 : }
570 1 : if registered {
571 1 : self.structure_version = self.structure_version.wrapping_add(1);
572 1 : }
573 1 : self.execution = ExecutionState::default();
574 1 : self.execution.resize(self.fragments.len(), self.pipelines.len(), self.slots.len(), self.operators.len());
575 1 : }
576 :
577 : /// Apply input events in their JSON form (an array of events)
578 1 : pub fn apply_events_json(&mut self, events: &[u8]) -> Result<Vec<ProfileUpdate>, serde_json::Error> {
579 1 : let events: Vec<ProfileInputEvent> = serde_json::from_slice(events)?;
580 1 : Ok(self.apply_events(&events))
581 1 : }
582 :
583 : /// Pack updates as `QueryProfileEvents`
584 19 : pub fn pack_events(&self, updates: &[ProfileUpdate]) -> Vec<u8> {
585 19 : let mut builder = FlatBufferBuilder::new();
586 19 : let events: Vec<fb::ProfileUpdateEvent> = updates
587 19 : .iter()
588 415 : .map(|u| {
589 415 : let stats = u.state.stats.to_fb();
590 415 : fb::ProfileUpdateEvent::new(u.time, u.entity, u.index, u.state.status.to_fb(), &stats)
591 415 : })
592 19 : .collect();
593 19 : let events = builder.create_vector(&events);
594 19 : let root = fb::QueryProfileEvents::create(
595 19 : &mut builder,
596 19 : &fb::QueryProfileEventsArgs { structure_version: self.structure_version, events: Some(events) },
597 : );
598 19 : builder.finish(root, None);
599 19 : builder.finished_data().to_vec()
600 19 : }
601 : }
|
