flowlog-build 0.2.2

Build-time FlowLog compiler for library mode.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! Node profiling structures.

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

use crate::profiler::addr::Addr;

/// A profiled node in the dataflow plan graph.
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub(super) struct NodeProfile {
    /// Node id.
    id: usize,
    /// Human-friendly node name.
    name: String,
    /// Block label (e.g., input/stratum/inspect) for grouping in UI.
    block: String,
    /// Optional fingerprint used to identify the rule transformation.
    fingerprint: Option<String>,
    /// Tags for presentation and filtering.
    tags: Vec<String>,
    /// Operator addresses associated with this node.
    operators: Vec<Addr>,
    /// Parent node ids (upstream dependencies).
    parents: Vec<usize>,
}

impl NodeProfile {
    /// Update the node id.
    fn update_id(&mut self, new_id: usize) {
        self.id = new_id;
    }

    /// Update the node name.
    fn update_name(&mut self, new_name: String) {
        self.name = new_name;
    }

    /// Update the block label.
    fn update_block(&mut self, new_block: String) {
        self.block = new_block;
    }

    /// Set the fingerprint from a raw u64 value.
    fn update_fingerprint(&mut self, fingerprint: u64) {
        self.fingerprint = Some(format!("0x{:016x}", fingerprint));
    }

    /// Add a tag to the node.
    fn add_tag(&mut self, tag: String) {
        self.tags.push(tag);
    }

    /// Append operator addresses.
    fn add_operators(&mut self, operator_addrs: Vec<Addr>) {
        self.operators.extend(operator_addrs);
    }

    /// Append parent node ids.
    fn add_parents(&mut self, parent_ids: Vec<usize>) {
        self.parents.extend(parent_ids);
    }
}

/// Internal nodes manager that tracks ids, scope addresses, and dependencies.
#[derive(Debug, Clone, Default)]
pub(super) struct NodeManager {
    /// Next node id to allocate.
    id_cnt: usize,
    /// Address counter used to create operator addresses.
    addr_cnt: Addr,
    /// Current block label for node grouping.
    block: String,

    /// Tracks the most recent node id for each variable name.
    node_map: HashMap<String, usize>,
}

impl NodeManager {
    /// Switch to the input block label.
    pub(super) fn update_input_block(&mut self) {
        self.block = "input".to_string();
    }

    /// Switch to a stratum block label.
    pub(super) fn update_stratum_block(&mut self, stratum_id: usize) {
        self.block = format!("stratum {}", stratum_id);
    }

    /// Switch to the inspect block label.
    pub(super) fn update_inspect_block(&mut self) {
        self.block = "inspect".to_string();
    }

    /// Enter a nested scope for address generation.
    pub(super) fn enter_scope(&mut self) {
        self.addr_cnt.enter_scope();
    }

    /// Leave the current scope for address generation.
    pub(super) fn leave_scope(&mut self) {
        self.addr_cnt.leave_scope();
        self.addr_cnt.advance(1);
    }

    /// Build a node with common metadata, optional fingerprint, and parents.
    pub(super) fn build_node(
        &mut self,
        name: String,
        input_variable_names: Vec<String>,
        output_variable_name: Option<String>,
        tag: &str,
        operator_steps: u32,
        fingerprint: Option<u64>,
    ) -> NodeProfile {
        let mut node = NodeProfile::default();

        let parent_ids = input_variable_names
            .iter()
            .filter_map(|variable_name| self.node_map.get(variable_name).copied())
            .collect();

        node.update_id(self.id_cnt);
        node.update_name(name);
        node.update_block(self.block.clone());
        if let Some(fingerprint) = fingerprint {
            node.update_fingerprint(fingerprint);
        }
        node.add_tag(tag.to_string());
        node.add_operators(self.addr_cnt.advance(operator_steps));
        node.add_parents(parent_ids);

        if let Some(output_variable_name) = output_variable_name {
            self.node_map.insert(output_variable_name, self.id_cnt);
        }
        self.id_cnt += 1;

        node
    }
}