dpcs 0.3.0

Reference implementation of the Data Pipeline Contract Standard (DPCS)
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

//! Graph validation phase.

use std::collections::{BTreeMap, BTreeSet, VecDeque};

use crate::diagnostics::{categories, Diagnostic, ValidationReport};
use crate::model::PipelineContract;

/// Validate graph edges and detect prohibited cycles.
pub fn validate(contract: &PipelineContract) -> ValidationReport {
    let mut report = ValidationReport::new();
    let step_ids: BTreeSet<&str> = contract.steps.iter().map(|s| s.id.as_str()).collect();

    let mut adjacency: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
    for step in &contract.steps {
        adjacency.entry(step.id.as_str()).or_default();
    }

    for (index, edge) in contract.graph.edges.iter().enumerate() {
        let object_ref = format!("graph.edges[{index}]");

        if edge.from.trim().is_empty() || edge.to.trim().is_empty() {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-001",
                    categories::GRAPH,
                    "graph edge endpoints must not be empty",
                )
                .with_object_ref(object_ref.clone()),
            );
            continue;
        }

        if !step_ids.contains(edge.from.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-002",
                    categories::GRAPH,
                    format!("graph edge references unknown step `{}`", edge.from),
                )
                .with_object_ref(object_ref.clone())
                .with_remediation("Ensure edge endpoints refer to declared step identifiers"),
            );
        }

        if !step_ids.contains(edge.to.as_str()) {
            report.push(
                Diagnostic::error(
                    "DPCS-GRP-003",
                    categories::GRAPH,
                    format!("graph edge references unknown step `{}`", edge.to),
                )
                .with_object_ref(object_ref)
                .with_remediation("Ensure edge endpoints refer to declared step identifiers"),
            );
        }

        adjacency
            .entry(edge.from.as_str())
            .or_default()
            .push(edge.to.as_str());
        adjacency.entry(edge.to.as_str()).or_default();
    }

    if let Some(cycle_node) = find_cycle(&adjacency) {
        report.push(
            Diagnostic::error(
                "DPCS-GRP-004",
                categories::GRAPH,
                format!("pipeline graph contains a prohibited cycle involving `{cycle_node}`"),
            )
            .with_object_ref("graph")
            .with_remediation("Remove cyclic dependencies from graph.edges"),
        );
    }

    report
}

fn find_cycle(adjacency: &BTreeMap<&str, Vec<&str>>) -> Option<String> {
    let mut indegree: BTreeMap<&str, usize> = adjacency.keys().map(|k| (*k, 0usize)).collect();
    for targets in adjacency.values() {
        for target in targets {
            *indegree.entry(target).or_default() += 1;
        }
    }

    let mut queue: VecDeque<&str> = indegree
        .iter()
        .filter_map(|(node, degree)| (*degree == 0).then_some(*node))
        .collect();

    let mut visited = 0usize;
    while let Some(node) = queue.pop_front() {
        visited += 1;
        if let Some(targets) = adjacency.get(node) {
            for target in targets {
                if let Some(degree) = indegree.get_mut(target) {
                    *degree -= 1;
                    if *degree == 0 {
                        queue.push_back(target);
                    }
                }
            }
        }
    }

    if visited == adjacency.len() {
        None
    } else {
        indegree
            .into_iter()
            .find_map(|(node, degree)| (degree > 0).then(|| node.to_string()))
    }
}