Merge pull request #118 from Synicix/pipeline_hash

Add Pipeline hashing, saving, loading, list, and delete

Author: eywalker

.vscode/settings.json

@@ -12,6 +12,7 @@
     "gitlens.showWhatsNewAfterUpgrades": false,
     "lldb.consoleMode": "evaluate",
     "rust-analyzer.check.command": "clippy",
+    "rust-analyzer.checkOnSave": true,
     "rust-analyzer.runnables.extraTestBinaryArgs": [
         "--nocapture"
     ],

src/core/error.rs

@@ -132,6 +132,7 @@ impl fmt::Debug for OrcaError {
             | Kind::InvalidOutputSpecNodeNotInGraph { backtrace, .. }
             | Kind::KeyMissing { backtrace, .. }
             | Kind::MissingInfo { backtrace, .. }
+            | Kind::FailedToGetLabelHashFromFileName { backtrace, .. }
             | Kind::FailedToGetPodJobOutput { backtrace, .. }
             | Kind::PipelineValidationErrorMissingKeys { backtrace, .. }
             | Kind::PodJobProcessingError { backtrace, .. }

src/core/graph.rs

@@ -23,13 +23,16 @@ pub fn make_graph(
 ) -> Result<DiGraph<PipelineNode, ()>> {
     let graph =
         DiGraph::<DotNodeWeight, DotAttrList>::from_dot_graph(DOTGraph::try_from(input_dot)?).map(
-            |node_idx, node| PipelineNode {
-                hash: String::new(),
-                kernel: get(metadata, &node.id)
-                    .unwrap_or_else(|error| panic!("{error}"))
-                    .clone(),
-                label: node.id.clone(),
-                node_idx,
+            |node_idx, node| {
+                let node_id_without_quotes = node.id.replace('"', "");
+                PipelineNode {
+                    hash: String::new(),
+                    kernel: get(metadata, &node_id_without_quotes)
+                        .unwrap_or_else(|error| panic!("{error}"))
+                        .clone(),
+                    label: node_id_without_quotes.clone(),
+                    node_idx,
+                }
             },
             |_, _| (),
         );

src/core/model/mod.rs

@@ -5,12 +5,13 @@ use serde::{Serialize, Serializer};
 use serde_yaml::{self, Value};
 use std::{
     collections::{BTreeMap, HashMap},
+    fmt::Debug,
     hash::BuildHasher,
     result,
 };
 
 /// Trait to handle serialization to yaml for `OrcaPod` models
-pub trait ToYaml: Serialize + Sized {
+pub trait ToYaml: Serialize + Sized + Debug {
     /// Serializes the instance to a YAML string.
     /// # Errors
     /// Will return `Err` if it fail to serialize instance to string

src/core/model/pipeline.rs

@@ -1,10 +1,11 @@
 use std::{
     backtrace::Backtrace,
-    collections::{HashMap, HashSet},
+    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
+    result,
 };
 
 use crate::{
-    core::crypto::hash_buffer,
+    core::{crypto::hash_buffer, model::ToYaml},
     uniffi::{
         error::{Kind, OrcaError, Result, selector},
         model::{
@@ -18,10 +19,10 @@ use petgraph::{
     Direction::Incoming,
     graph::{self, NodeIndex},
 };
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize, ser::SerializeStruct as _};
 use snafu::OptionExt as _;
 
-#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
 pub struct PipelineNode {
     // Hash that represent the node
     pub hash: String,
@@ -259,6 +260,108 @@ impl Pipeline {
 
         &graph[node_idx].hash
     }
+
+    fn to_dot_lex(&self) -> String {
+        // Get all the nodes and their children in lexicographical order
+        let nodes_and_edges = self.graph.node_indices().fold(
+            BTreeMap::<&String, BTreeSet<&String>>::new(),
+            |mut acc, node_idx| {
+                let children = self
+                    .graph
+                    .neighbors_directed(node_idx, petgraph::Direction::Outgoing)
+                    .map(|child_idx| &self.graph[child_idx].hash)
+                    .collect::<BTreeSet<&String>>();
+                acc.insert(&self.graph[node_idx].hash, children);
+                acc
+            },
+        );
+
+        // Build the dot representation string by
+        let mut lines = Vec::new();
+        for (node, node_children) in nodes_and_edges {
+            if node_children.is_empty() {
+                lines.push(format!("  \"{node}\""));
+            } else {
+                for child in node_children {
+                    lines.push(format!("  \"{node}\" -> \"{child}\""));
+                }
+            }
+        }
+
+        // Convert lines into a single string with a proper new line between each entry
+        format!("digraph {{\n{}\n}}", lines.join("\n"))
+    }
+
+    /// Get a `BTreeMap` of <`kernel_hash`, `BTreeSet`<`node_hashes`>> for all nodes in the graph. Mainly use for serialization
+    pub(crate) fn get_kernel_to_node_lut(&self) -> BTreeMap<String, BTreeSet<String>> {
+        self.graph.node_indices().fold(
+            BTreeMap::<String, BTreeSet<String>>::new(),
+            |mut acc, node_idx| {
+                acc.entry(self.graph[node_idx].kernel.get_hash().to_owned())
+                    .or_default()
+                    .insert(self.graph[node_idx].hash.clone());
+                acc
+            },
+        )
+    }
+
+    pub(crate) fn get_kernel_lut(&self) -> HashSet<&Kernel> {
+        self.graph
+            .node_indices()
+            .fold(HashSet::<&Kernel>::new(), |mut acc, node_idx| {
+                acc.insert(&self.graph[node_idx].kernel);
+                acc
+            })
+    }
+
+    /// Get a `HashMap` of <`node_hash`, `node_label`> for all nodes in the graph if label is not empty. Mainly use for serialization
+    pub(crate) fn get_label_lut(&self) -> impl Iterator<Item = (&String, &String)> {
+        self.graph.node_indices().filter_map(|node_idx| {
+            let label = &self.graph[node_idx].label;
+            if label.is_empty() {
+                None
+            } else {
+                Some((&self.graph[node_idx].hash, label))
+            }
+        })
+    }
+}
+
+impl Serialize for Pipeline {
+    fn serialize<S>(&self, serializer: S) -> result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut state = serializer.serialize_struct("Pipeline", 4)?;
+        state.serialize_field("kernel_lut", &self.get_kernel_to_node_lut())?;
+        state.serialize_field("dot", &self.to_dot_lex())?;
+
+        // Input spec needs to be sorted for consistent serialization
+        let input_spec_sorted: BTreeMap<_, Vec<NodeURI>> = self
+            .input_spec
+            .iter()
+            .map(|(k, v)| {
+                let mut sorted_v = v.clone();
+                sorted_v.sort();
+                (k, sorted_v)
+            })
+            .collect();
+        state.serialize_field("input_spec", &input_spec_sorted)?;
+        state.serialize_field("output_spec", &self.output_spec)?;
+        state.end()
+    }
+}
+
+impl ToYaml for Pipeline {
+    fn process_field(
+        field_name: &str,
+        field_value: &serde_yaml::Value,
+    ) -> Option<(String, serde_yaml::Value)> {
+        match field_name {
+            "hash" | "annotation" => None, // Skip annotation field
+            _ => Some((field_name.to_owned(), field_value.clone())),
+        }
+    }
 }
 
 impl PipelineJob {
@@ -316,3 +419,88 @@ impl PipelineJob {
         Ok(node_input_packets)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        core::model::ToYaml as _,
+        uniffi::{
+            error::Result,
+            model::{
+                Annotation,
+                pipeline::{NodeURI, Pipeline},
+            },
+            operator::MapOperator,
+        },
+    };
+    use indoc::indoc;
+    use pretty_assertions::assert_eq;
+    use std::collections::HashMap;
+
+    #[test]
+    fn to_yaml() -> Result<()> {
+        let pipeline = Pipeline::new(
+            indoc! {"
+            digraph {
+                A -> B -> C
+            }
+        "},
+            &HashMap::from([
+                (
+                    "A".into(),
+                    MapOperator::new(HashMap::from([("node_key_1".into(), "node_key_2".into())]))?
+                        .into(),
+                ),
+                (
+                    "B".into(),
+                    MapOperator::new(HashMap::from([("node_key_2".into(), "node_key_1".into())]))?
+                        .into(),
+                ),
+                (
+                    "C".into(),
+                    MapOperator::new(HashMap::from([("node_key_1".into(), "node_key_2".into())]))?
+                        .into(),
+                ),
+            ]),
+            HashMap::from([(
+                "pipeline_key_1".into(),
+                vec![NodeURI {
+                    node_id: "A".into(),
+                    key: "node_key_1".into(),
+                }],
+            )]),
+            HashMap::new(),
+            Some(Annotation {
+                name: "test".into(),
+                version: "0.1".into(),
+                description: "Test pipeline".into(),
+            }),
+        )?;
+
+        assert_eq!(
+            pipeline.to_yaml()?,
+            indoc! {r#"
+            class: pipeline
+            kernel_lut:
+              2980eb39e3702442cc31656d6ec3995f91680ab042a27160a00ffe33b91419af:
+              - 4b498582ed57ca6a10809d7480bd3f159542ad139e402698e3f525fc6b0d4dea
+              c8f036079b69beee914434c1e01be638972ce05cd2e640fc1e9be7bf3d9e76be:
+              - 368c7a517f3fbdd7cab10c90ebc44e44765fc33f66b4f4f5151a6bee322d8217
+              - 6c84111298d0cfe811dff3f10ab444795c0a9d60609ba1b9391c45e642a69afa
+            dot: |-
+              digraph {
+                "368c7a517f3fbdd7cab10c90ebc44e44765fc33f66b4f4f5151a6bee322d8217"
+                "4b498582ed57ca6a10809d7480bd3f159542ad139e402698e3f525fc6b0d4dea" -> "368c7a517f3fbdd7cab10c90ebc44e44765fc33f66b4f4f5151a6bee322d8217"
+                "6c84111298d0cfe811dff3f10ab444795c0a9d60609ba1b9391c45e642a69afa" -> "4b498582ed57ca6a10809d7480bd3f159542ad139e402698e3f525fc6b0d4dea"
+              }
+            input_spec:
+              pipeline_key_1:
+              - node_id: 6c84111298d0cfe811dff3f10ab444795c0a9d60609ba1b9391c45e642a69afa
+                key: node_key_1
+            output_spec: {}
+            "#},
+        );
+
+        Ok(())
+    }
+}