Line data Source code
1 : //! The query profile and its FlatBuffers encoding.
2 : //!
3 : //! The structure (operators, edges, fragments, pipelines, slots) carries no execution state. The execution state
4 : //! lives in `progress` and leaves the model only as events.
5 :
6 : use super::progress::ExecutionState;
7 : use dashql_buffers::parser::SymbolSpan;
8 : use dashql_buffers::profile as fb;
9 : use flatbuffers::{FlatBufferBuilder, WIPOffset};
10 : use std::collections::HashMap;
11 :
12 : /// A component of the path from an operator to the JSON value of its child operator
13 : #[derive(Clone, Debug, PartialEq)]
14 : pub(super) enum PathComponent {
15 : /// A member in an object
16 : MemberInObject { attribute: String },
17 : /// An entry in an array
18 : EntryInArray { index: usize },
19 : /// No path (the document root)
20 : None,
21 : }
22 :
23 : /// A reference of an operator to another operator through the operator id serialized by Hyper
24 : #[derive(Debug)]
25 : pub(super) struct OperatorReference {
26 : /// The operator id serialized by Hyper of the referenced operator
27 : pub(super) source_operator_id: u64,
28 : /// The relationship kind
29 : pub(super) kind: &'static str,
30 : /// Additional reference attributes, values serialized as JSON
31 : pub(super) attributes: Vec<(String, String)>,
32 : }
33 :
34 : /// An operator found in the JSON plan, before flattening
35 : #[derive(Debug, Default)]
36 : pub(super) struct ParsedOperatorNode {
37 : /// The path from the parent operator (or the document root)
38 : pub(super) parent_child_path: Vec<PathComponent>,
39 : /// The members of the operator object that contain no operators, values serialized as JSON
40 : pub(super) attributes: Vec<(String, String)>,
41 : /// The estimated output rows
42 : pub(super) estimated_rows: Option<f64>,
43 : /// The runtime counters of an analyzed plan
44 : pub(super) analyzed: super::progress::Stats,
45 : /// The references to other operators
46 : pub(super) references: Vec<OperatorReference>,
47 : /// The operator type
48 : pub(super) operator_type: Option<String>,
49 : /// The operator label
50 : pub(super) operator_label: Option<String>,
51 : /// The operator id serialized by Hyper
52 : pub(super) source_operator_id: Option<u64>,
53 : /// The operator id of an engine with string ids (Trino plan node ids), else the Hyper id is the engine id
54 : pub(super) engine_id: Option<String>,
55 : /// The innermost fragment, if the parser knows it while reading (Trino)
56 : pub(super) fragment: Option<u32>,
57 : /// The child operators (indexes of parsed operators)
58 : pub(super) child_operators: Vec<usize>,
59 : /// The source location
60 : pub(super) source_location: Option<SymbolSpan>,
61 : }
62 :
63 : /// A fragment: a part of the plan that the engine schedules as a unit (a Hyper federate, a Trino stage).
64 : ///
65 : /// Fragments form a hierarchy. Every operator belongs to its innermost fragment only, so rollups over the
66 : /// hierarchy count every operator once.
67 : #[derive(Debug, Default)]
68 : pub(super) struct Fragment {
69 : /// The fragment id of the engine
70 : pub(super) engine_id: Option<String>,
71 : /// The enclosing fragment
72 : pub(super) parent_fragment: Option<u32>,
73 : /// The root operator
74 : pub(super) root_operator: Option<u32>,
75 : /// The operators owned by the fragment
76 : pub(super) operators: Vec<u32>,
77 : /// The attributes, values serialized as JSON
78 : pub(super) attributes: Vec<(String, String)>,
79 : /// Draw no contour?
80 : pub(super) hidden: bool,
81 : }
82 :
83 : /// A pipeline.
84 : /// Note that a pipeline does not need to to be linear.
85 : /// Hyper implements a Fork operator that will effectively result in two pipeline targets.
86 : #[derive(Debug, Default)]
87 : pub(super) struct Pipeline {
88 : /// The pipeline id of the engine
89 : pub(super) engine_id: Option<String>,
90 : /// The fragment that runs the pipeline
91 : pub(super) fragment: Option<u32>,
92 : /// The slots of the pipeline (indexes into `QueryProfile::slots`)
93 : pub(super) slots: Vec<u32>,
94 : /// The attributes, values serialized as JSON
95 : pub(super) attributes: Vec<(String, String)>,
96 : }
97 :
98 : /// A slot: an operator in a pipeline, the finest grain of execution statistics
99 : #[derive(Clone, Copy, Debug)]
100 : pub(super) struct Slot {
101 : /// The pipeline
102 : pub(super) pipeline: u32,
103 : /// The operator
104 : pub(super) operator: u32,
105 : /// The role of the operator in the pipeline
106 : pub(super) role: fb::ProfileSlotRole,
107 : }
108 :
109 : /// A non-containment relationship between operators
110 : #[derive(Debug)]
111 : pub(super) struct OperatorCrossEdge {
112 : /// The referenced producer operator
113 : pub(super) source_node: u32,
114 : /// The operator containing the reference
115 : pub(super) target_node: u32,
116 : /// The relationship kind
117 : pub(super) kind: &'static str,
118 : /// Additional reference attributes, values serialized as JSON
119 : pub(super) attributes: Vec<(String, String)>,
120 : }
121 :
122 : /// An operator edge
123 : #[derive(Debug)]
124 : pub(super) struct OperatorEdge {
125 : /// The parent operator
126 : pub(super) parent_operator: u32,
127 : /// The child operator
128 : pub(super) child_operator: u32,
129 : /// The port count of the parent
130 : pub(super) parent_port_count: usize,
131 : /// The port id in the parent where this edge ends
132 : pub(super) parent_port_index: usize,
133 : /// The slot of the child in the pipeline that carries the edge
134 : pub(super) slot: Option<u32>,
135 : }
136 :
137 : /// A sealed operator node
138 : #[derive(Debug, Default)]
139 : pub(super) struct OperatorNode {
140 : /// The operator type
141 : pub(super) operator_type: Option<String>,
142 : /// The operator label
143 : pub(super) operator_label: Option<String>,
144 : /// The engine-neutral operator kind
145 : pub(super) kind: fb::ProfileOperatorKind,
146 : /// The operator id of the engine
147 : pub(super) engine_id: Option<String>,
148 : /// The operator id serialized by Hyper, until the cross edges and pipelines are resolved
149 : pub(super) source_operator_id: Option<u64>,
150 : /// The members of the operator object that contain no operators, values serialized as JSON
151 : pub(super) attributes: Vec<(String, String)>,
152 : /// The estimated output rows
153 : pub(super) estimated_rows: Option<f64>,
154 : /// The runtime counters of an analyzed plan, until they are applied as events
155 : pub(super) analyzed: super::progress::Stats,
156 : /// The references to other operators, until the cross edges are resolved
157 : pub(super) references: Vec<OperatorReference>,
158 : /// The parent operator
159 : pub(super) parent: Option<usize>,
160 : /// The parent path
161 : pub(super) parent_path: Vec<PathComponent>,
162 : /// The source location
163 : pub(super) source_location: Option<SymbolSpan>,
164 : /// The child operators
165 : pub(super) children_begin: usize,
166 : /// The number of child operators
167 : pub(super) children_count: usize,
168 : /// The begin of outgoing cross edges
169 : pub(super) cross_edges_begin: usize,
170 : /// The number of outgoing cross edges
171 : pub(super) cross_edge_count: usize,
172 : /// The slots of the operator
173 : pub(super) slots: Vec<u32>,
174 : /// The innermost fragment
175 : pub(super) fragment: Option<u32>,
176 : /// The layout info
177 : pub(super) layout_rect: Option<fb::ProfileLayoutRect>,
178 : }
179 :
180 : impl OperatorNode {
181 : /// Construct from a parsed node
182 611 : pub(super) fn from_parsed(parsed: ParsedOperatorNode) -> Self {
183 : Self {
184 611 : operator_type: parsed.operator_type,
185 611 : operator_label: parsed.operator_label,
186 611 : engine_id: parsed.engine_id.or_else(|| parsed.source_operator_id.map(|id| id.to_string())),
187 611 : fragment: parsed.fragment,
188 611 : source_operator_id: parsed.source_operator_id,
189 611 : attributes: parsed.attributes,
190 611 : estimated_rows: parsed.estimated_rows,
191 611 : analyzed: parsed.analyzed,
192 611 : references: parsed.references,
193 611 : parent_path: parsed.parent_child_path,
194 611 : source_location: parsed.source_location,
195 611 : ..Default::default()
196 : }
197 611 : }
198 :
199 : /// The child operators
200 2203 : pub(super) fn children(&self) -> std::ops::Range<usize> {
201 2203 : self.children_begin..self.children_begin + self.children_count
202 2203 : }
203 :
204 : /// Serialize the parent child path
205 525 : pub(super) fn serialize_parent_path(&self) -> String {
206 525 : let mut out = String::new();
207 576 : for (i, component) in self.parent_path.iter().enumerate() {
208 576 : match component {
209 449 : PathComponent::MemberInObject { attribute } => {
210 449 : if i > 0 {
211 0 : out.push('.');
212 449 : }
213 449 : out.push_str(attribute);
214 : }
215 127 : PathComponent::EntryInArray { index } => {
216 127 : out.push('[');
217 127 : out.push_str(&index.to_string());
218 127 : out.push(']');
219 127 : }
220 0 : PathComponent::None => {}
221 : }
222 : }
223 525 : out
224 525 : }
225 : }
226 :
227 : /// A string dictionary
228 : #[derive(Default)]
229 : struct StringDictionary {
230 : /// The allocated strings
231 : strings: Vec<String>,
232 : /// The string ids
233 : string_ids: HashMap<String, u32>,
234 : }
235 :
236 : impl StringDictionary {
237 : /// Allocate a string in the string dictionary
238 8314 : fn allocate(&mut self, s: &str) -> u32 {
239 8314 : if let Some(id) = self.string_ids.get(s) {
240 4246 : return *id;
241 4068 : }
242 4068 : let id = self.strings.len() as u32;
243 4068 : self.strings.push(s.to_string());
244 4068 : self.string_ids.insert(s.to_string(), id);
245 4068 : id
246 8314 : }
247 :
248 : /// Allocate an optional string, u32::MAX if there is none
249 1709 : fn allocate_opt(&mut self, s: Option<&str>) -> u32 {
250 1709 : s.map(|s| self.allocate(s)).unwrap_or(u32::MAX)
251 1709 : }
252 : }
253 :
254 : /// Pack attributes, returns the begin and the count
255 807 : fn pack_attributes(
256 807 : source: &[(String, String)],
257 807 : strings: &mut StringDictionary,
258 807 : attributes: &mut Vec<fb::ProfileAttribute>,
259 807 : ) -> (u32, u32) {
260 807 : let begin = attributes.len();
261 3293 : for (name, value_json) in source {
262 3293 : attributes.push(fb::ProfileAttribute::new(strings.allocate(name), strings.allocate(value_json)));
263 3293 : }
264 807 : (begin as u32, (attributes.len() - begin) as u32)
265 807 : }
266 :
267 : /// An optional index as u32, u32::MAX if there is none
268 1201 : pub(super) fn index_or_max(index: Option<u32>) -> u32 {
269 1201 : index.unwrap_or(u32::MAX)
270 1201 : }
271 :
272 : /// The profile of a query: the structure of its plan and its execution state.
273 : ///
274 : /// It holds the structure of a plan (Hyper or Trino) and the execution state of a running or analyzed query.
275 : pub struct QueryProfile {
276 : /// The engine that produced the plan
277 : pub(super) engine: fb::ProfileEngine,
278 : /// The structure version
279 : pub(super) structure_version: u32,
280 : /// The operators
281 : pub(super) operators: Vec<OperatorNode>,
282 : /// The operator edges
283 : pub(super) operator_edges: Vec<OperatorEdge>,
284 : /// The operator cross edges
285 : pub(super) operator_cross_edges: Vec<OperatorCrossEdge>,
286 : /// The root operators
287 : pub(super) root_operators: Vec<u32>,
288 : /// The fragments
289 : pub(super) fragments: Vec<Fragment>,
290 : /// The pipelines
291 : pub(super) pipelines: Vec<Pipeline>,
292 : /// The slots
293 : pub(super) slots: Vec<Slot>,
294 : /// The number of fragments, pipelines and slots of the loaded plan, the rest was registered at runtime
295 : pub(super) loaded_counts: (usize, usize, usize),
296 : /// The execution state
297 : pub(super) execution: ExecutionState,
298 : /// The layout config
299 : pub(super) layout_config: fb::DerivedProfileLayoutConfig,
300 : /// The layout info of the entire plan
301 : pub(super) layout_rect: Option<fb::ProfileLayoutRect>,
302 : }
303 :
304 : impl Default for QueryProfile {
305 0 : fn default() -> Self {
306 0 : Self::new()
307 0 : }
308 : }
309 :
310 : impl QueryProfile {
311 : /// Constructor
312 111 : pub fn new() -> Self {
313 111 : Self {
314 111 : engine: fb::ProfileEngine::UNKNOWN,
315 111 : structure_version: 0,
316 111 : operators: Vec::new(),
317 111 : operator_edges: Vec::new(),
318 111 : operator_cross_edges: Vec::new(),
319 111 : root_operators: Vec::new(),
320 111 : fragments: Vec::new(),
321 111 : pipelines: Vec::new(),
322 111 : slots: Vec::new(),
323 111 : loaded_counts: (0, 0, 0),
324 111 : execution: ExecutionState::default(),
325 111 : layout_config: fb::DerivedProfileLayoutConfig::default(),
326 111 : layout_rect: None,
327 111 : }
328 111 : }
329 :
330 : /// Reset the entire profile. The layout config is kept, the structure version grows.
331 111 : pub fn reset(&mut self) {
332 111 : self.engine = fb::ProfileEngine::UNKNOWN;
333 111 : self.structure_version = self.structure_version.wrapping_add(1);
334 111 : self.fragments.clear();
335 111 : self.pipelines.clear();
336 111 : self.slots.clear();
337 111 : self.root_operators.clear();
338 111 : self.operators.clear();
339 111 : self.operator_edges.clear();
340 111 : self.operator_cross_edges.clear();
341 111 : self.loaded_counts = (0, 0, 0);
342 111 : self.execution = ExecutionState::default();
343 111 : self.layout_rect = None;
344 111 : }
345 :
346 : /// Start an empty model for a query of an engine. The structure appears with the progress of the query
347 : /// (Trino stages are registered as fragments), a plan can be loaded later.
348 6 : pub fn start_query(&mut self, engine: fb::ProfileEngine) {
349 6 : self.reset();
350 6 : self.engine = engine;
351 6 : self.seal_loaded_structure();
352 6 : }
353 :
354 : /// The structure version
355 14 : pub fn structure_version(&self) -> u32 {
356 14 : self.structure_version
357 14 : }
358 :
359 : /// Seal the structure of a loaded plan: everything registered later is runtime structure
360 102 : pub(super) fn seal_loaded_structure(&mut self) {
361 102 : self.loaded_counts = (self.fragments.len(), self.pipelines.len(), self.slots.len());
362 102 : self.execution.resize(self.fragments.len(), self.pipelines.len(), self.slots.len(), self.operators.len());
363 102 : }
364 :
365 : /// Add a pipeline with its slots, returns the pipeline index.
366 : ///
367 : /// An operator appears at most once per pipeline, the first role wins. Edges between two operators of the
368 : /// pipeline that have no slot yet are carried by the slot of their child in this pipeline.
369 64 : pub(super) fn add_pipeline(
370 64 : &mut self,
371 64 : engine_id: Option<String>,
372 64 : fragment: Option<u32>,
373 64 : slots: &[(u32, fb::ProfileSlotRole)],
374 64 : attributes: Vec<(String, String)>,
375 64 : ) -> u32 {
376 64 : let pipeline = self.pipelines.len() as u32;
377 64 : let mut slot_of: HashMap<u32, u32> = HashMap::with_capacity(slots.len());
378 64 : let mut pipeline_slots = Vec::with_capacity(slots.len());
379 150 : for &(operator, role) in slots {
380 150 : if slot_of.contains_key(&operator) {
381 0 : continue;
382 150 : }
383 150 : let slot = self.slots.len() as u32;
384 150 : self.slots.push(Slot { pipeline, operator, role });
385 150 : self.operators[operator as usize].slots.push(slot);
386 150 : slot_of.insert(operator, slot);
387 150 : pipeline_slots.push(slot);
388 : }
389 558 : for edge in &mut self.operator_edges {
390 558 : if edge.slot.is_none() && slot_of.contains_key(&edge.parent_operator) {
391 111 : edge.slot = slot_of.get(&edge.child_operator).copied();
392 459 : }
393 : }
394 64 : self.pipelines.push(Pipeline { engine_id, fragment, slots: pipeline_slots, attributes });
395 64 : pipeline
396 64 : }
397 :
398 : /// Flatten the operators.
399 : ///
400 : /// Runs a post-order DFS over the parsed operator tree and writes the children of every operator as one block.
401 96 : pub(super) fn flatten_operators(&mut self, parsed_ops: Vec<ParsedOperatorNode>, parsed_roots: Vec<usize>) {
402 96 : let mut parsed_ops: Vec<Option<ParsedOperatorNode>> = parsed_ops.into_iter().map(Some).collect();
403 96 : let children_of: Vec<Vec<usize>> =
404 611 : parsed_ops.iter().map(|op| op.as_ref().unwrap().child_operators.clone()).collect();
405 :
406 : // Prepare the operators: (parsed operator, visited)
407 98 : let mut pending: Vec<(usize, bool)> = parsed_roots.iter().rev().map(|r| (*r, false)).collect();
408 96 : self.operators.reserve(parsed_ops.len());
409 :
410 : // The flattened operators that are not yet placed
411 96 : let mut mapped: Vec<Option<OperatorNode>> = (0..parsed_ops.len()).map(|_| None).collect();
412 1318 : while let Some(&(op, visited)) = pending.last() {
413 1222 : if visited {
414 : // Add child operators
415 611 : let children_begin = self.operators.len();
416 611 : for &child in &children_of[op] {
417 514 : let node = mapped[child].take().expect("child operator must be flattened");
418 514 : self.place_operator(node);
419 514 : }
420 611 : let child_count = self.operators.len() - children_begin;
421 611 : let mut flat = OperatorNode::from_parsed(parsed_ops[op].take().unwrap());
422 611 : flat.children_begin = children_begin;
423 611 : flat.children_count = child_count;
424 :
425 : // Register flat operator
426 611 : mapped[op] = Some(flat);
427 611 : pending.pop();
428 : } else {
429 611 : pending.last_mut().unwrap().1 = true;
430 : // Add the children, reversed since we're using a DFS stack
431 611 : pending.extend(children_of[op].iter().rev().map(|c| (*c, false)));
432 : }
433 : }
434 :
435 : // Now only the root operators are left. They are placed in reverse plan order
436 : // (see docs/design/core_quirks.md).
437 96 : self.root_operators.reserve(parsed_roots.len());
438 98 : for root in parsed_roots.into_iter().rev() {
439 97 : let node = mapped[root].take().expect("root operator must be flattened");
440 97 : let oid = self.place_operator(node);
441 97 : self.root_operators.push(oid as u32);
442 97 : }
443 96 : }
444 :
445 : /// Derive the operator edges (child to parent) from the flattened operators
446 96 : pub(super) fn identify_operator_edges(&mut self) {
447 96 : let mut child_edges = Vec::with_capacity(self.operators.len());
448 611 : for (parent_id, parent) in self.operators.iter().enumerate() {
449 611 : for (port, child_id) in parent.children().enumerate() {
450 514 : child_edges.push(OperatorEdge {
451 514 : parent_operator: parent_id as u32,
452 514 : child_operator: child_id as u32,
453 514 : parent_port_count: parent.children_count,
454 514 : parent_port_index: port,
455 514 : slot: None,
456 514 : });
457 514 : }
458 : }
459 96 : self.operator_edges = child_edges;
460 96 : }
461 :
462 : /// Append a flattened operator and point its children at it
463 611 : fn place_operator(&mut self, node: OperatorNode) -> usize {
464 611 : let operator_id = self.operators.len();
465 611 : let children = node.children();
466 611 : self.operators.push(node);
467 611 : for child in &mut self.operators[children] {
468 514 : child.parent = Some(operator_id);
469 514 : }
470 611 : operator_id
471 611 : }
472 :
473 : /// Pack a plan operator
474 525 : fn pack_operator(
475 525 : &self,
476 525 : op: &OperatorNode,
477 525 : slots_begin: usize,
478 525 : strings: &mut StringDictionary,
479 525 : attributes: &mut Vec<fb::ProfileAttribute>,
480 525 : ) -> fb::ProfileOperator {
481 525 : let mut out = fb::ProfileOperator::default();
482 525 : out.set_engine_id(strings.allocate_opt(op.engine_id.as_deref()));
483 525 : out.set_kind(op.kind);
484 525 : out.set_type_name(strings.allocate_opt(op.operator_type.as_deref()));
485 525 : out.set_label(strings.allocate_opt(op.operator_label.as_deref()));
486 525 : out.set_parent(op.parent.map(|p| p as u32).unwrap_or(u32::MAX));
487 525 : out.set_parent_path(strings.allocate(&op.serialize_parent_path()));
488 525 : if let Some(loc) = &op.source_location {
489 184 : out.set_source_location(&SymbolSpan::new(loc.offset(), loc.length()));
490 341 : }
491 525 : out.set_children_begin(op.children_begin as u32);
492 525 : out.set_children_count(op.children_count as u32);
493 525 : out.set_cross_edges_begin(op.cross_edges_begin as u32);
494 525 : out.set_cross_edge_count(op.cross_edge_count as u32);
495 525 : let (attributes_begin, attribute_count) = pack_attributes(&op.attributes, strings, attributes);
496 525 : out.set_attributes_begin(attributes_begin);
497 525 : out.set_attribute_count(attribute_count);
498 525 : out.set_slots_begin(slots_begin as u32);
499 525 : out.set_slot_count(op.slots.len() as u32);
500 525 : out.set_fragment(index_or_max(op.fragment));
501 525 : out.set_estimated_rows(op.estimated_rows.unwrap_or(f64::NAN));
502 525 : if let Some(rect) = &op.layout_rect {
503 511 : out.set_layout_rect(rect);
504 511 : }
505 525 : out
506 525 : }
507 :
508 : /// Pack the query profile as flatbuffer
509 80 : pub fn pack<'a>(&self, builder: &mut FlatBufferBuilder<'a>) -> WIPOffset<fb::QueryProfile<'a>> {
510 : // Track strings in a dictionary for flatbuffer
511 80 : let mut dictionary = StringDictionary::default();
512 80 : let mut flat_attributes: Vec<fb::ProfileAttribute> = Vec::new();
513 :
514 : // Pack plan operators and their slots
515 80 : let mut flat_ops = Vec::with_capacity(self.operators.len());
516 80 : let mut flat_operator_slots: Vec<u32> = Vec::with_capacity(self.slots.len());
517 525 : for op in &self.operators {
518 525 : let slots_begin = flat_operator_slots.len();
519 525 : flat_operator_slots.extend_from_slice(&op.slots);
520 525 : flat_ops.push(self.pack_operator(op, slots_begin, &mut dictionary, &mut flat_attributes));
521 525 : }
522 :
523 : // Pack the plan edges
524 80 : let flat_op_edges: Vec<fb::ProfileOperatorEdge> = self
525 80 : .operator_edges
526 80 : .iter()
527 449 : .map(|edge| {
528 449 : fb::ProfileOperatorEdge::new(
529 449 : edge.child_operator,
530 449 : edge.parent_operator,
531 449 : edge.parent_port_index as u32,
532 449 : edge.parent_port_count as u32,
533 449 : index_or_max(edge.slot),
534 : )
535 449 : })
536 80 : .collect();
537 :
538 : // Pack the plan cross edges and their relationship metadata
539 80 : let mut flat_cross_edges = Vec::with_capacity(self.operator_cross_edges.len());
540 98 : for edge in &self.operator_cross_edges {
541 74 : let attributes_begin = flat_attributes.len();
542 74 : let kind = (String::from("kind"), format!("\"{}\"", edge.kind));
543 74 : pack_attributes(std::slice::from_ref(&kind), &mut dictionary, &mut flat_attributes);
544 74 : pack_attributes(&edge.attributes, &mut dictionary, &mut flat_attributes);
545 74 : flat_cross_edges.push(fb::ProfileOperatorCrossEdge::new(
546 74 : edge.source_node,
547 74 : edge.target_node,
548 74 : attributes_begin as u32,
549 74 : (flat_attributes.len() - attributes_begin) as u32,
550 74 : ));
551 74 : }
552 :
553 : // Pack plan fragments
554 80 : let mut flat_fragments = Vec::with_capacity(self.fragments.len());
555 80 : let mut flat_fragment_operators: Vec<u32> = Vec::new();
556 98 : for f in &self.fragments {
557 93 : let operators_begin = flat_fragment_operators.len();
558 93 : flat_fragment_operators.extend_from_slice(&f.operators);
559 93 : let (attributes_begin, attribute_count) =
560 93 : pack_attributes(&f.attributes, &mut dictionary, &mut flat_attributes);
561 93 : flat_fragments.push(fb::ProfileFragment::new(
562 93 : dictionary.allocate_opt(f.engine_id.as_deref()),
563 93 : index_or_max(f.parent_fragment),
564 93 : index_or_max(f.root_operator),
565 93 : operators_begin as u32,
566 93 : (flat_fragment_operators.len() - operators_begin) as u32,
567 93 : attributes_begin,
568 93 : attribute_count,
569 93 : f.hidden as u8,
570 93 : ));
571 93 : }
572 :
573 : // Pack plan pipelines and their slots.
574 : // Slots are created pipeline by pipeline (`add_pipeline`), so the slots of a pipeline are contiguous.
575 80 : let flat_slots: Vec<fb::ProfileSlot> =
576 112 : self.slots.iter().map(|s| fb::ProfileSlot::new(s.pipeline, s.operator, s.role)).collect();
577 80 : let mut flat_pipelines = Vec::with_capacity(self.pipelines.len());
578 80 : for p in &self.pipelines {
579 41 : let (attributes_begin, attribute_count) =
580 41 : pack_attributes(&p.attributes, &mut dictionary, &mut flat_attributes);
581 41 : flat_pipelines.push(fb::ProfilePipeline::new(
582 41 : dictionary.allocate_opt(p.engine_id.as_deref()),
583 41 : index_or_max(p.fragment),
584 41 : p.slots.first().copied().unwrap_or(0),
585 41 : p.slots.len() as u32,
586 41 : attributes_begin,
587 41 : attribute_count,
588 41 : ));
589 41 : }
590 :
591 80 : let flat_fragments_ofs = builder.create_vector(&flat_fragments);
592 80 : let flat_fragment_operators_ofs = builder.create_vector(&flat_fragment_operators);
593 80 : let flat_pipelines_ofs = builder.create_vector(&flat_pipelines);
594 80 : let flat_slots_ofs = builder.create_vector(&flat_slots);
595 80 : let flat_operator_slots_ofs = builder.create_vector(&flat_operator_slots);
596 80 : let flat_ops_ofs = builder.create_vector(&flat_ops);
597 80 : let flat_attributes_ofs = builder.create_vector(&flat_attributes);
598 80 : let flat_edges_ofs = builder.create_vector(&flat_op_edges);
599 80 : let flat_cross_edges_ofs = builder.create_vector(&flat_cross_edges);
600 80 : let flat_roots_ofs = builder.create_vector(&self.root_operators);
601 80 : let string_offsets: Vec<WIPOffset<&str>> =
602 4068 : dictionary.strings.iter().map(|s| builder.create_string(s)).collect();
603 80 : let string_dictionary_ofs = builder.create_vector(&string_offsets);
604 :
605 80 : fb::QueryProfile::create(
606 80 : builder,
607 80 : &fb::QueryProfileArgs {
608 80 : engine: self.engine,
609 80 : structure_version: self.structure_version,
610 80 : layout_config: Some(&self.layout_config),
611 80 : layout_rect: self.layout_rect.as_ref(),
612 80 : string_dictionary: Some(string_dictionary_ofs),
613 80 : attributes: Some(flat_attributes_ofs),
614 80 : operators: Some(flat_ops_ofs),
615 80 : root_operators: Some(flat_roots_ofs),
616 80 : operator_edges: Some(flat_edges_ofs),
617 80 : operator_cross_edges: Some(flat_cross_edges_ofs),
618 80 : fragments: Some(flat_fragments_ofs),
619 80 : fragment_operators: Some(flat_fragment_operators_ofs),
620 80 : pipelines: Some(flat_pipelines_ofs),
621 80 : slots: Some(flat_slots_ofs),
622 80 : operator_slots: Some(flat_operator_slots_ofs),
623 80 : },
624 : )
625 80 : }
626 :
627 : /// Pack the query profile into a finished flatbuffer
628 80 : pub fn pack_buffer(&self) -> Vec<u8> {
629 80 : let mut builder = FlatBufferBuilder::new();
630 80 : let vm = self.pack(&mut builder);
631 80 : builder.finish(vm, None);
632 80 : builder.finished_data().to_vec()
633 80 : }
634 : }
|
