pondrs 0.3.0

A pipeline execution library
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

//! Error types for the pipeline framework.

use thiserror::Error;

/// Framework-level errors from dataset I/O and pipeline infrastructure.
///
/// Feature-gated variants are only available when the corresponding feature
/// is enabled. The `DatasetNotLoaded` variant is always available (no_std).
#[derive(Error, Debug)]
pub enum PondError {
    #[cfg(feature = "std")]
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),

    #[cfg(feature = "polars")]
    #[error("Polars error: {0}")]
    Polars(#[from] polars::error::PolarsError),

    #[cfg(feature = "yaml")]
    #[error("YAML parse error: {0}")]
    YamlScan(#[from] yaml_rust2::ScanError),

    #[cfg(feature = "yaml")]
    #[error("YAML emit error: {0}")]
    YamlEmit(#[from] yaml_rust2::EmitError),

    #[cfg(feature = "std")]
    #[error("Serde YAML error: {0}")]
    SerdeYaml(#[from] serde_yaml::Error),

    #[cfg(any(feature = "json", feature = "plotly", feature = "viz"))]
    #[error("JSON error: {0}")]
    Json(#[from] serde_json::Error),

    #[cfg(feature = "image")]
    #[error("Image error: {0}")]
    Image(#[from] ::image::ImageError),

    #[error("Dataset not loaded: no data available")]
    DatasetNotLoaded,

    #[error("Runner not found")]
    RunnerNotFound,

    #[error("Pipeline check failed")]
    CheckFailed,

    #[cfg(feature = "std")]
    #[error("Lock poisoned: {0}")]
    LockPoisoned(std::string::String),

    #[cfg(feature = "std")]
    #[error("{0}")]
    Custom(std::string::String),

    #[cfg(feature = "std")]
    #[error("Key mismatch: expected keys {expected:?}, got {actual:?}")]
    KeyMismatch {
        expected: std::vec::Vec<std::string::String>,
        actual: std::vec::Vec<std::string::String>,
    },

    #[cfg(feature = "std")]
    #[error("Node not found: '{0}'")]
    NodeNotFound(std::string::String),
}

/// Validation error from [`PipelineInfo::check`](crate::pipeline::PipelineInfo::check).
#[derive(Debug)]
pub enum CheckError {
    /// A node reads a dataset that is produced by a later node (wrong order).
    InputNotProduced {
        node_name: &'static str,
        dataset_id: usize,
    },
    /// A dataset is produced by more than one node.
    DuplicateOutput {
        node_name: &'static str,
        dataset_id: usize,
    },
    /// A node writes to a param dataset (params are read-only).
    ParamWritten {
        node_name: &'static str,
        dataset_id: usize,
    },
    /// A pipeline declares an input that none of its children consume.
    UnusedPipelineInput {
        pipeline_name: &'static str,
        dataset_id: usize,
    },
    /// A pipeline declares an output that none of its children produce.
    UnproducedPipelineOutput {
        pipeline_name: &'static str,
        dataset_id: usize,
    },
    /// A child node consumes an external dataset not declared in the pipeline's inputs.
    UndeclaredPipelineInput {
        pipeline_name: &'static str,
        dataset_id: usize,
    },
    /// The fixed-capacity dataset buffer overflowed.
    CapacityExceeded,
}

impl core::fmt::Display for CheckError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            CheckError::InputNotProduced { node_name, dataset_id } => {
                write!(f, "Node '{node_name}' requires dataset {dataset_id:#x}, which is produced by a later node")
            }
            CheckError::DuplicateOutput { node_name, dataset_id } => {
                write!(f, "Node '{node_name}' produces dataset {dataset_id:#x}, which was already produced by an earlier node")
            }
            CheckError::ParamWritten { node_name, dataset_id } => {
                write!(f, "Node '{node_name}' writes to param dataset {dataset_id:#x}, but params are read-only")
            }
            CheckError::UnusedPipelineInput { pipeline_name, dataset_id } => {
                write!(f, "Pipeline '{pipeline_name}' declares input {dataset_id:#x}, but none of its children consume it")
            }
            CheckError::UnproducedPipelineOutput { pipeline_name, dataset_id } => {
                write!(f, "Pipeline '{pipeline_name}' declares output {dataset_id:#x}, but none of its children produce it")
            }
            CheckError::UndeclaredPipelineInput { pipeline_name, dataset_id } => {
                write!(f, "Pipeline '{pipeline_name}' has a child that consumes external dataset {dataset_id:#x}, which is not declared in the pipeline's inputs")
            }
            CheckError::CapacityExceeded => {
                write!(f, "Dataset capacity exceeded; use check_with_capacity::<N>() with a larger N")
            }
        }
    }
}

impl From<core::convert::Infallible> for PondError {
    fn from(x: core::convert::Infallible) -> Self {
        match x {}
    }
}