Line data Source code
1 : //! Reading Hyper plans into the query profile.
2 :
3 : use super::query_profile::{
4 : Fragment, OperatorCrossEdge, OperatorReference, ParsedOperatorNode, PathComponent, QueryProfile,
5 : };
6 : use super::progress::{ProfileInputEvent, Stats};
7 : use dashql_buffers::parser::SymbolSpan;
8 : use dashql_buffers::status::StatusCode;
9 : use dashql_buffers::profile as fb;
10 : use serde_json::{Map, Value};
11 : use std::collections::{HashMap, HashSet};
12 :
13 : /// A JSON node on the DFS stack.
14 : ///
15 : /// The logic is the following:
16 : /// - We do a post-order DFS traversal
17 : /// - Whenever we find an operator, we re-construct the path toward the lowest operator ancestor
18 : struct ParserDfsNode<'a> {
19 : /// The json value
20 : json_value: &'a Value,
21 : /// The DFS visited marker for the post-order traversal
22 : visited: bool,
23 : /// The parent index in the DFS
24 : parent_node_index: Option<usize>,
25 : /// The reference in the parent
26 : parent_path: PathComponent,
27 : /// The operator type (if any)
28 : operator_type: Option<String>,
29 : /// The operator label (if any)
30 : operator_label: Option<String>,
31 : /// The operator id serialized by Hyper (if any)
32 : source_operator_id: Option<u64>,
33 : /// The already emitted children
34 : child_operators: Vec<usize>,
35 : /// The source location
36 : source_location: Option<SymbolSpan>,
37 : }
38 :
39 : impl<'a> ParserDfsNode<'a> {
40 15283 : fn new(json_value: &'a Value, parent_node_index: Option<usize>, parent_path: PathComponent) -> Self {
41 15283 : Self {
42 15283 : json_value,
43 15283 : visited: false,
44 15283 : parent_node_index,
45 15283 : parent_path,
46 15283 : operator_type: None,
47 15283 : operator_label: None,
48 15283 : source_operator_id: None,
49 15283 : child_operators: Vec::new(),
50 15283 : source_location: None,
51 15283 : }
52 15283 : }
53 : }
54 :
55 : /// Find the lowest operator ancestor of a DFS node and the path towards it
56 500 : fn find_ancestor(nodes: &[ParserDfsNode], mut next: usize) -> (Option<usize>, Vec<PathComponent>) {
57 500 : let mut path = Vec::new();
58 :
59 : // Check the current node
60 500 : let node = &nodes[next];
61 500 : match node.parent_node_index {
62 419 : Some(parent) => {
63 419 : path.push(node.parent_path.clone());
64 419 : next = parent;
65 419 : }
66 81 : None => return (None, Vec::new()),
67 : }
68 :
69 : // Walk upwards until we hit the root or an operator
70 : loop {
71 472 : let node = &nodes[next];
72 472 : if node.operator_type.is_some() {
73 415 : path.reverse();
74 415 : return (Some(next), path);
75 57 : }
76 57 : match node.parent_node_index {
77 : None => {
78 4 : path.reverse();
79 4 : return (None, path);
80 : }
81 53 : Some(parent) => {
82 53 : path.push(node.parent_path.clone());
83 53 : next = parent;
84 53 : }
85 : }
86 : }
87 500 : }
88 :
89 : /// Infer the label of a scan from the relation of the attributes it defines
90 19 : fn infer_scan_label(object: &Map<String, Value>) -> Option<String> {
91 19 : let attributes = object.get("attributes")?.as_array()?;
92 1 : let mut relation: Option<&str> = None;
93 2 : for attribute in attributes {
94 2 : let Some(qualified_id) = attribute.get("defines").and_then(|d| d.get("id")).and_then(Value::as_str) else {
95 0 : continue;
96 : };
97 2 : let candidate = match qualified_id.rsplit_once('.') {
98 2 : Some((candidate, _)) if !candidate.is_empty() => candidate,
99 0 : _ => continue,
100 : };
101 2 : if relation.is_some_and(|relation| relation != candidate) {
102 0 : return None;
103 2 : }
104 2 : relation = Some(candidate);
105 : }
106 1 : relation.map(String::from)
107 19 : }
108 :
109 : /// Read the source location of an operator from its `sqlpos` (`[[begin, end], ...]`)
110 199 : fn read_source_location(sql_pos: &[Value]) -> Option<SymbolSpan> {
111 199 : let [begin, end] = sql_pos.first()?.as_array()?.as_slice() else {
112 0 : return None;
113 : };
114 398 : let read = |v: &Value| v.as_u64().and_then(|n| u32::try_from(n).ok()).unwrap_or(0);
115 199 : let (begin, end) = (read(begin), read(end));
116 199 : Some(SymbolSpan::new(begin, end.max(begin) - begin))
117 199 : }
118 :
119 : /// Is the operator type one of the join operators that may reference a magic set?
120 485 : fn is_join_operator(operator_type: &str) -> bool {
121 97 : matches!(
122 485 : operator_type,
123 485 : "join"
124 404 : | "leftouterjoin"
125 401 : | "rightouterjoin"
126 400 : | "fullouterjoin"
127 400 : | "leftantijoin"
128 397 : | "rightantijoin"
129 396 : | "leftsemijoin"
130 393 : | "rightsemijoin"
131 389 : | "leftsinglejoin"
132 389 : | "rightsinglejoin"
133 389 : | "leftmarkjoin"
134 388 : | "rightmarkjoin"
135 : )
136 485 : }
137 :
138 : /// Serialize an attribute value as compact JSON
139 3003 : fn to_json(value: &Value) -> String {
140 3003 : serde_json::to_string(value).expect("a JSON value always serializes")
141 3003 : }
142 :
143 : /// Does a JSON value contain an operator?
144 17408 : fn contains_operator(value: &Value) -> bool {
145 17408 : match value {
146 3274 : Value::Object(members) => {
147 3274 : members.get("operator").is_some_and(Value::is_string) || members.values().any(contains_operator)
148 : }
149 3688 : Value::Array(elements) => elements.iter().any(contains_operator),
150 10446 : _ => false,
151 : }
152 17408 : }
153 :
154 : /// The first of several members that exists
155 2168 : fn find_member<'a>(object: &'a Map<String, Value>, names: &[&str]) -> Option<&'a Value> {
156 4983 : names.iter().find_map(|name| object.get(*name))
157 2168 : }
158 :
159 : /// The first of several members that exists, if it is a number
160 2168 : fn find_number<'a>(object: &'a Map<String, Value>, names: &[&str]) -> Option<&'a serde_json::Number> {
161 2168 : find_member(object, names).and_then(|member| member.as_number())
162 2168 : }
163 :
164 : /// Read the estimated output rows and the runtime counters of an analyzed plan (EXPLAIN ANALYZE)
165 500 : fn read_execution_statistics(source: &Map<String, Value>) -> (Option<f64>, Stats) {
166 500 : let runtime = source.get("statistics").and_then(Value::as_object).unwrap_or(source);
167 :
168 : const ROWS: &[&str] = &["estimated-rows", "estimatedRows", "cardinality"];
169 500 : let estimated_rows = find_number(source, ROWS).or_else(|| find_number(runtime, ROWS)).and_then(|n| n.as_f64());
170 :
171 1500 : let count = |names: &[&str]| find_number(runtime, names).and_then(|n| n.as_u64());
172 500 : let analyzed = Stats {
173 500 : tuples_in: count(&["processed-rows", "processedRows"]),
174 500 : tuples_out: count(&["output-rows", "outputRows"]),
175 500 : memory_bytes: count(&["memory-bytes", "memoryBytes"]),
176 500 : ..Default::default()
177 500 : };
178 500 : (estimated_rows, analyzed)
179 500 : }
180 :
181 : /// The engine-neutral kind of a Hyper operator
182 500 : fn hyper_operator_kind(operator_type: &str) -> fb::ProfileOperatorKind {
183 : use fb::ProfileOperatorKind as K;
184 500 : match operator_type.to_lowercase().replace('-', "").as_str() {
185 500 : "output" => K::OUTPUT,
186 495 : "scan" | "parquetscan" | "tablescan" => K::SCAN,
187 310 : "explicitscan" => K::EXPLICIT_SCAN,
188 298 : "virtualtable" => K::VIRTUAL_TABLE,
189 296 : "tableconstruction" => K::TABLE_CONSTRUCTION,
190 292 : "tablefunction" => K::TABLE_FUNCTION,
191 291 : "select" => K::FILTER,
192 288 : "map" => K::PROJECT,
193 251 : "groupby" => K::AGGREGATE,
194 212 : "window" => K::WINDOW,
195 210 : "sort" => K::SORT,
196 189 : "join" => K::JOIN,
197 108 : "leftouterjoin" => K::LEFT_OUTER_JOIN,
198 105 : "rightouterjoin" => K::RIGHT_OUTER_JOIN,
199 104 : "fullouterjoin" => K::FULL_OUTER_JOIN,
200 104 : "leftsemijoin" => K::LEFT_SEMI_JOIN,
201 101 : "rightsemijoin" => K::RIGHT_SEMI_JOIN,
202 97 : "leftantijoin" => K::LEFT_ANTI_JOIN,
203 94 : "rightantijoin" => K::RIGHT_ANTI_JOIN,
204 93 : "leftmarkjoin" => K::LEFT_MARK_JOIN,
205 92 : "rightmarkjoin" => K::RIGHT_MARK_JOIN,
206 92 : "unionall" => K::UNION_ALL,
207 85 : "union" => K::UNION,
208 85 : "intersectall" => K::INTERSECT_ALL,
209 84 : "intersect" => K::INTERSECT,
210 84 : "exceptall" => K::EXCEPT_ALL,
211 83 : "except" => K::EXCEPT,
212 83 : "iteration" => K::ITERATION,
213 81 : "iterationincrement" => K::ITERATION_INCREMENT,
214 79 : "insert" => K::INSERT,
215 78 : _ => K::OTHER,
216 : }
217 500 : }
218 :
219 : /// The references of an operator to other operators by their Hyper operator ids
220 500 : fn read_operator_references(operator_type: &str, object: &Map<String, Value>) -> Vec<OperatorReference> {
221 500 : let mut references = Vec::new();
222 500 : let mut reference = |name: &str, kind: &'static str| {
223 112 : if let Some(source_operator_id) = object.get(name).and_then(Value::as_u64) {
224 11 : references.push(OperatorReference { source_operator_id, kind, attributes: Vec::new() });
225 101 : }
226 112 : };
227 485 : match operator_type {
228 500 : "explicitscan" | "explicit-scan" => reference("input", "input"),
229 488 : "iterationincrement" => reference("source", "source"),
230 486 : "earlyprobe" => reference("builder", "builder"),
231 485 : t if is_join_operator(t) => reference("magic", "magic"),
232 388 : _ => {}
233 : }
234 :
235 500 : let early_probes = object.get("earlyProbes").or_else(|| object.get("early-probes"));
236 500 : for probe in early_probes.and_then(Value::as_array).into_iter().flatten() {
237 68 : let Some(probe) = probe.as_object() else {
238 0 : continue;
239 : };
240 68 : if let Some(builder) = probe.get("builder").and_then(Value::as_u64) {
241 68 : let attributes = probe
242 68 : .iter()
243 270 : .filter(|(name, _)| *name != "builder" && *name != "kind")
244 202 : .map(|(name, value)| (name.clone(), to_json(value)))
245 68 : .collect();
246 68 : references.push(OperatorReference { source_operator_id: builder, kind: "early-probe", attributes });
247 0 : }
248 : }
249 500 : references
250 500 : }
251 :
252 : /// Read the operator from the DFS node of its JSON object
253 500 : fn read_operator(node: ParserDfsNode, parent_child_path: Vec<PathComponent>) -> ParsedOperatorNode {
254 500 : let object = node.json_value.as_object().expect("operators are objects");
255 500 : let operator_type = node.operator_type.expect("operators have a type");
256 : // The attributes are the members that do not contain other operators
257 500 : let attributes = object
258 500 : .iter()
259 3194 : .filter(|(_, value)| !contains_operator(value))
260 2800 : .map(|(name, value)| (name.clone(), to_json(value)))
261 500 : .collect();
262 500 : let (estimated_rows, analyzed) = read_execution_statistics(object);
263 500 : ParsedOperatorNode {
264 500 : parent_child_path,
265 500 : references: read_operator_references(&operator_type, object),
266 500 : attributes,
267 500 : estimated_rows,
268 500 : analyzed,
269 500 : operator_type: Some(operator_type),
270 500 : operator_label: node.operator_label,
271 500 : source_operator_id: node.source_operator_id,
272 500 : child_operators: node.child_operators,
273 500 : source_location: node.source_location,
274 500 : ..Default::default()
275 500 : }
276 500 : }
277 :
278 : impl QueryProfile {
279 : /// Parse a Hyper plan.
280 : ///
281 : /// The runtime counters of an analyzed plan (EXPLAIN ANALYZE) are applied as one batch of input events, the
282 : /// same path as the progress of a running query.
283 89 : pub fn parse_hyper_plan(&mut self, plan: &[u8]) -> Result<(), StatusCode> {
284 : // Reset the current query profile
285 89 : self.reset();
286 89 : self.engine = fb::ProfileEngine::HYPER;
287 :
288 : // Parse the document
289 84 : let document: Value =
290 89 : serde_json::from_slice(plan).map_err(|_| StatusCode::VIEWMODEL_INPUT_JSON_PARSER_ERROR)?;
291 :
292 : // Run DFS over the json plan.
293 : // Perform a post-order DFS over all json nodes.
294 : // Emit operator nodes on our way up and resolve the lowest ancestor through the DFS stack.
295 84 : let mut parsed_operators: Vec<ParsedOperatorNode> = Vec::new();
296 84 : let mut root_operators: Vec<usize> = Vec::new();
297 84 : let mut pending = vec![ParserDfsNode::new(&document, None, PathComponent::None)];
298 30650 : while !pending.is_empty() {
299 30566 : let current_index = pending.len() - 1;
300 :
301 : // Already visited?
302 30566 : if pending[current_index].visited {
303 : // Is an operator?
304 15283 : if pending[current_index].operator_type.is_some() {
305 : // Build the ancestor path
306 500 : let (ancestor, ancestor_path) = find_ancestor(&pending, current_index);
307 : // Then emit the node
308 500 : let current = pending.pop().unwrap();
309 500 : let op = parsed_operators.len();
310 500 : parsed_operators.push(read_operator(current, ancestor_path));
311 500 : match ancestor {
312 : // Register as child operator in ancestor
313 415 : Some(ancestor) => pending[ancestor].child_operators.push(op),
314 : // No parent operator, register as root
315 85 : None => root_operators.push(op),
316 : }
317 14783 : } else {
318 14783 : // Otherwise do nothing, the attributes are read when emitting the operator
319 14783 : pending.pop();
320 14783 : }
321 15283 : continue;
322 15283 : }
323 : // Mark as visited
324 15283 : pending[current_index].visited = true;
325 :
326 15283 : match pending[current_index].json_value {
327 : // Current node is an object:
328 : // - Add children for DFS
329 : // - Check if it is an operator
330 3244 : Value::Object(members) => {
331 3244 : let pending_begin = pending.len();
332 10054 : for (attribute_name, value) in members {
333 10054 : let current = &mut pending[current_index];
334 10054 : match (attribute_name.as_str(), value) {
335 : // Is the current node an operator?
336 : // Mark as such and skip attribute during DFS
337 10054 : ("operator", Value::String(operator_type)) => {
338 500 : current.operator_type = Some(operator_type.clone());
339 500 : }
340 : // Contains an operator id?
341 9554 : ("operatorId" | "operator-id", _) if value.is_u64() => {
342 407 : current.source_operator_id = value.as_u64();
343 407 : }
344 : // Contains a debug name? Hyper plans use both camelCase and kebab-case spellings.
345 1525 : ("debugName" | "debug-name", Value::Object(debug_name)) => {
346 137 : if let Some(label) = debug_name.get("value").and_then(Value::as_str) {
347 137 : current.operator_label = Some(label.to_string());
348 137 : }
349 : }
350 : // Contains a sqlpos?
351 2790 : ("sqlpos", Value::Array(sql_pos)) => {
352 199 : if let Some(location) = read_source_location(sql_pos) {
353 199 : current.source_location = Some(location);
354 199 : }
355 : }
356 : // Mark pending for DFS traversal
357 8811 : _ => pending.push(ParserDfsNode::new(
358 8811 : value,
359 8811 : Some(current_index),
360 8811 : PathComponent::MemberInObject { attribute: attribute_name.clone() },
361 : )),
362 : }
363 : }
364 3244 : let current = &mut pending[current_index];
365 3244 : if current.operator_label.is_none() && current.operator_type.as_deref() == Some("scan") {
366 19 : current.operator_label = infer_scan_label(members);
367 3225 : }
368 : // Reverse the order of the attribute nodes on the stack
369 3244 : pending[pending_begin..].reverse();
370 : }
371 : // Current node is an array:
372 : // - Add children for DFS
373 3245 : Value::Array(values) => {
374 6388 : for (j, child_value) in values.iter().enumerate().rev() {
375 6388 : pending.push(ParserDfsNode::new(
376 6388 : child_value,
377 6388 : Some(current_index),
378 6388 : PathComponent::EntryInArray { index: j },
379 6388 : ));
380 6388 : }
381 : }
382 : // Skip value nodes during DFS traversal
383 8794 : Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {}
384 : }
385 : }
386 :
387 : // Flatten the Hyper operators
388 84 : self.flatten_operators(parsed_operators, root_operators);
389 500 : for op in &mut self.operators {
390 500 : op.kind = op.operator_type.as_deref().map(hyper_operator_kind).unwrap_or_default();
391 500 : }
392 : // Resolve non-containment references between flattened operators
393 84 : self.identify_operator_cross_edges();
394 : // Identify fragments introduced by federates
395 84 : self.identify_fragments();
396 : // Identify the operator edges
397 84 : self.identify_operator_edges();
398 : // Read pipelines if this plan format provides them. Legacy plans render without pipeline overlays.
399 84 : if let Some(pipelines) = document.get("pipelines").and_then(Value::as_array) {
400 12 : self.parse_hyper_pipelines(pipelines);
401 72 : }
402 84 : self.seal_loaded_structure();
403 :
404 : // Apply the counters of an analyzed plan
405 84 : let analyzed: Vec<ProfileInputEvent> = self
406 84 : .operators
407 84 : .iter_mut()
408 84 : .enumerate()
409 500 : .filter(|(_, op)| !op.analyzed.is_empty())
410 84 : .map(|(operator, op)| ProfileInputEvent::UpdateOperator {
411 : time: 0,
412 7 : operator: operator as u32,
413 7 : status: None,
414 7 : stats: std::mem::take(&mut op.analyzed),
415 7 : })
416 84 : .collect();
417 84 : self.apply_events(&analyzed);
418 84 : Ok(())
419 89 : }
420 :
421 : /// Map the operator ids serialized by Hyper to operators, the first operator wins
422 96 : fn source_operator_ids(&self) -> HashMap<u64, u32> {
423 96 : let mut operator_ids = HashMap::new();
424 546 : for (operator_id, op) in self.operators.iter().enumerate() {
425 546 : if let Some(source_operator_id) = op.source_operator_id {
426 453 : operator_ids.entry(source_operator_id).or_insert(operator_id as u32);
427 453 : }
428 : }
429 96 : operator_ids
430 96 : }
431 :
432 : /// Resolve non-containment operator references
433 84 : fn identify_operator_cross_edges(&mut self) {
434 84 : let operator_ids = self.source_operator_ids();
435 :
436 84 : let mut pending: Vec<(u32, u32, &'static str, Vec<(String, String)>)> = Vec::new();
437 500 : for (operator_id, op) in self.operators.iter_mut().enumerate() {
438 500 : for reference in std::mem::take(&mut op.references) {
439 79 : if let Some(&source) = operator_ids.get(&reference.source_operator_id) {
440 78 : pending.push((source, operator_id as u32, reference.kind, reference.attributes));
441 78 : }
442 : }
443 : }
444 :
445 84 : pending.sort_by_key(|(source_node, ..)| *source_node);
446 84 : self.operator_cross_edges.reserve(pending.len());
447 108 : for (source_node, target_node, kind, attributes) in pending {
448 78 : let source = &mut self.operators[source_node as usize];
449 78 : if source.cross_edge_count == 0 {
450 75 : source.cross_edges_begin = self.operator_cross_edges.len();
451 75 : }
452 78 : source.cross_edge_count += 1;
453 78 : self.operator_cross_edges.push(OperatorCrossEdge {
454 78 : source_node,
455 78 : target_node,
456 78 : kind,
457 78 : attributes,
458 78 : });
459 : }
460 84 : }
461 :
462 : /// Identify fragments introduced by federate operators.
463 : ///
464 : /// An execution target at the root introduces a hidden root fragment, every federate a fragment. Fragments
465 : /// nest: an operator belongs to its innermost fragment, a nested federate starts a child fragment.
466 84 : fn identify_fragments(&mut self) {
467 84 : let operators = &self.operators;
468 84 : let mut boundaries: Vec<u32> = Vec::new();
469 84 : let mut root_boundaries: Vec<u32> = Vec::new();
470 84 : let mut pending: Vec<u32> = Vec::with_capacity(operators.len());
471 86 : for &root in self.root_operators.iter().rev() {
472 85 : if operators[root as usize].operator_type.as_deref() == Some("executiontarget") {
473 56 : root_boundaries.push(root);
474 64 : }
475 85 : pending.push(root);
476 : }
477 924 : let push_children = |pending: &mut Vec<u32>, operator_id: u32| {
478 924 : pending.extend(operators[operator_id as usize].children().rev().map(|c| c as u32));
479 924 : };
480 584 : while let Some(operator_id) = pending.pop() {
481 500 : if operators[operator_id as usize].operator_type.as_deref() == Some("federate") {
482 13 : boundaries.push(operator_id);
483 487 : }
484 500 : push_children(&mut pending, operator_id);
485 : }
486 84 : boundaries.splice(0..0, root_boundaries);
487 85 : let fragment_of: HashMap<u32, u32> = boundaries.iter().enumerate().map(|(f, b)| (*b, f as u32)).collect();
488 :
489 : // Collect the operators of every fragment and stop at nested fragments.
490 : // Enclosing fragments come first, so a nested fragment knows its parent before it is collected.
491 84 : let mut parents: Vec<Option<u32>> = vec![None; boundaries.len()];
492 84 : let mut fragments = Vec::with_capacity(boundaries.len());
493 85 : for (f, &boundary_id) in boundaries.iter().enumerate() {
494 69 : let mut fragment = Fragment {
495 69 : parent_fragment: parents[f],
496 69 : root_operator: Some(boundary_id),
497 69 : operators: vec![boundary_id],
498 69 : hidden: operators[boundary_id as usize].operator_type.as_deref() == Some("executiontarget"),
499 69 : ..Default::default()
500 69 : };
501 69 : pending.clear();
502 69 : push_children(&mut pending, boundary_id);
503 430 : while let Some(operator_id) = pending.pop() {
504 361 : if let Some(&nested) = fragment_of.get(&operator_id) {
505 6 : parents[nested as usize] = Some(f as u32);
506 6 : continue;
507 355 : }
508 355 : fragment.operators.push(operator_id);
509 355 : push_children(&mut pending, operator_id);
510 : }
511 69 : fragments.push(fragment);
512 : }
513 85 : for (f, fragment) in fragments.iter().enumerate() {
514 424 : for &operator_id in &fragment.operators {
515 424 : self.operators[operator_id as usize].fragment = Some(f as u32);
516 424 : }
517 : }
518 84 : self.fragments = fragments;
519 84 : }
520 :
521 : /// Read explicit Hyper pipelines.
522 : ///
523 : /// A member without a member child produces the tuples of the pipeline (source), a member without a member
524 : /// parent consumes them (sink).
525 12 : fn parse_hyper_pipelines(&mut self, source_pipelines: &[Value]) {
526 12 : let operator_ids = self.source_operator_ids();
527 23 : for source_pipeline in source_pipelines {
528 23 : let Some(source_pipeline) = source_pipeline.as_object() else {
529 0 : continue;
530 : };
531 23 : let engine_id = source_pipeline.get("pipelineId").and_then(Value::as_u64).map(|id| id.to_string());
532 23 : let members: Vec<u32> = source_pipeline
533 23 : .get("operators")
534 23 : .and_then(Value::as_array)
535 23 : .into_iter()
536 23 : .flatten()
537 57 : .filter_map(|member| operator_ids.get(&member.as_u64()?).copied())
538 23 : .collect();
539 23 : let membership: HashSet<u32> = members.iter().copied().collect();
540 23 : let slots: Vec<(u32, fb::ProfileSlotRole)> = members
541 23 : .iter()
542 57 : .map(|&member| {
543 57 : let op = &self.operators[member as usize];
544 57 : let consumes = op.children().any(|child| membership.contains(&(child as u32)));
545 57 : let feeds = op.parent.is_some_and(|parent| membership.contains(&(parent as u32)));
546 57 : let role = match (consumes, feeds) {
547 23 : (false, _) => fb::ProfileSlotRole::SOURCE,
548 11 : (true, true) => fb::ProfileSlotRole::THROUGH,
549 23 : (true, false) => fb::ProfileSlotRole::SINK,
550 : };
551 57 : (member, role)
552 57 : })
553 23 : .collect();
554 23 : let fragment = members.first().and_then(|m| self.operators[*m as usize].fragment);
555 23 : let attributes = source_pipeline
556 23 : .iter()
557 47 : .filter(|(name, _)| *name != "pipelineId" && *name != "operators")
558 23 : .map(|(name, value)| (name.clone(), to_json(value)))
559 23 : .collect();
560 23 : self.add_pipeline(engine_id, fragment, &slots, attributes);
561 : }
562 12 : }
563 : }
|
