oxigdal_distributed/
error.rs

1//! Error types for distributed processing operations.
2//!
3//! This module provides comprehensive error handling for distributed geospatial
4//! processing, including Flight RPC errors, worker failures, and coordination issues.
5
6/// Result type for distributed operations.
7pub type Result<T> = std::result::Result<T, DistributedError>;
8
9/// Errors that can occur during distributed processing.
10#[derive(Debug, thiserror::Error)]
11pub enum DistributedError {
12    /// Error during Flight RPC communication.
13    #[error("Flight RPC error: {0}")]
14    FlightRpc(String),
15
16    /// Error connecting to a worker node.
17    #[error("Worker connection error: {0}")]
18    WorkerConnection(String),
19
20    /// Worker failed to complete a task.
21    #[error("Worker task failure: {0}")]
22    WorkerTaskFailure(String),
23
24    /// Coordinator error.
25    #[error("Coordinator error: {0}")]
26    Coordinator(String),
27
28    /// Data partitioning error.
29    #[error("Partitioning error: {0}")]
30    Partitioning(String),
31
32    /// Data shuffle error.
33    #[error("Shuffle error: {0}")]
34    Shuffle(String),
35
36    /// Task serialization/deserialization error.
37    #[error("Task serialization error: {0}")]
38    TaskSerialization(String),
39
40    /// Network timeout error.
41    #[error("Network timeout: {0}")]
42    Timeout(String),
43
44    /// Authentication error.
45    #[error("Authentication failed: {0}")]
46    Authentication(String),
47
48    /// Resource allocation error.
49    #[error("Resource allocation error: {0}")]
50    ResourceAllocation(String),
51
52    /// Result aggregation error.
53    #[error("Result aggregation error: {0}")]
54    Aggregation(String),
55
56    /// Arrow error during data transfer.
57    #[error("Arrow error: {0}")]
58    Arrow(String),
59
60    /// The requested operation is recognized but not yet implemented on this node.
61    #[error("Operation not implemented: {0}")]
62    OperationNotImplemented(String),
63
64    /// The operation is malformed or references invalid inputs (bad expression,
65    /// unknown column, unsupported column type, out-of-range band index, ...).
66    #[error("Invalid operation: {0}")]
67    InvalidOperation(String),
68
69    /// Core OxiGDAL error.
70    #[error("OxiGDAL core error: {0}")]
71    Core(#[from] oxigdal_core::error::OxiGdalError),
72
73    /// I/O error.
74    #[error("I/O error: {0}")]
75    Io(#[from] std::io::Error),
76
77    /// JSON serialization error.
78    #[error("JSON error: {0}")]
79    Json(#[from] serde_json::Error),
80
81    /// Tonic transport error.
82    #[error("Transport error: {0}")]
83    Transport(#[from] tonic::transport::Error),
84
85    /// Tonic status error.
86    #[error("RPC status error: {0}")]
87    Status(#[from] tonic::Status),
88
89    /// Generic error with custom message.
90    #[error("{0}")]
91    Custom(String),
92}
93
94impl DistributedError {
95    /// Create a Flight RPC error.
96    pub fn flight_rpc<S: Into<String>>(msg: S) -> Self {
97        Self::FlightRpc(msg.into())
98    }
99
100    /// Create a worker connection error.
101    pub fn worker_connection<S: Into<String>>(msg: S) -> Self {
102        Self::WorkerConnection(msg.into())
103    }
104
105    /// Create a worker task failure error.
106    pub fn worker_task_failure<S: Into<String>>(msg: S) -> Self {
107        Self::WorkerTaskFailure(msg.into())
108    }
109
110    /// Create a coordinator error.
111    pub fn coordinator<S: Into<String>>(msg: S) -> Self {
112        Self::Coordinator(msg.into())
113    }
114
115    /// Create a partitioning error.
116    pub fn partitioning<S: Into<String>>(msg: S) -> Self {
117        Self::Partitioning(msg.into())
118    }
119
120    /// Create a shuffle error.
121    pub fn shuffle<S: Into<String>>(msg: S) -> Self {
122        Self::Shuffle(msg.into())
123    }
124
125    /// Create a task serialization error.
126    pub fn task_serialization<S: Into<String>>(msg: S) -> Self {
127        Self::TaskSerialization(msg.into())
128    }
129
130    /// Create a timeout error.
131    pub fn timeout<S: Into<String>>(msg: S) -> Self {
132        Self::Timeout(msg.into())
133    }
134
135    /// Create an authentication error.
136    pub fn authentication<S: Into<String>>(msg: S) -> Self {
137        Self::Authentication(msg.into())
138    }
139
140    /// Create a resource allocation error.
141    pub fn resource_allocation<S: Into<String>>(msg: S) -> Self {
142        Self::ResourceAllocation(msg.into())
143    }
144
145    /// Create an aggregation error.
146    pub fn aggregation<S: Into<String>>(msg: S) -> Self {
147        Self::Aggregation(msg.into())
148    }
149
150    /// Create an Arrow error.
151    pub fn arrow<S: Into<String>>(msg: S) -> Self {
152        Self::Arrow(msg.into())
153    }
154
155    /// Create an "operation not implemented" error.
156    pub fn operation_not_implemented<S: Into<String>>(msg: S) -> Self {
157        Self::OperationNotImplemented(msg.into())
158    }
159
160    /// Create an "invalid operation" error.
161    pub fn invalid_operation<S: Into<String>>(msg: S) -> Self {
162        Self::InvalidOperation(msg.into())
163    }
164
165    /// Create a custom error.
166    pub fn custom<S: Into<String>>(msg: S) -> Self {
167        Self::Custom(msg.into())
168    }
169}
170
171impl From<arrow::error::ArrowError> for DistributedError {
172    fn from(err: arrow::error::ArrowError) -> Self {
173        Self::Arrow(err.to_string())
174    }
175}
176
177impl From<DistributedError> for tonic::Status {
178    fn from(err: DistributedError) -> Self {
179        match err {
180            DistributedError::FlightRpc(msg) => {
181                tonic::Status::internal(format!("Flight RPC error: {}", msg))
182            }
183            DistributedError::WorkerConnection(msg) => {
184                tonic::Status::unavailable(format!("Worker connection error: {}", msg))
185            }
186            DistributedError::WorkerTaskFailure(msg) => {
187                tonic::Status::internal(format!("Worker task failure: {}", msg))
188            }
189            DistributedError::Coordinator(msg) => {
190                tonic::Status::internal(format!("Coordinator error: {}", msg))
191            }
192            DistributedError::Partitioning(msg) => {
193                tonic::Status::invalid_argument(format!("Partitioning error: {}", msg))
194            }
195            DistributedError::Shuffle(msg) => {
196                tonic::Status::internal(format!("Shuffle error: {}", msg))
197            }
198            DistributedError::TaskSerialization(msg) => {
199                tonic::Status::invalid_argument(format!("Task serialization error: {}", msg))
200            }
201            DistributedError::Timeout(msg) => {
202                tonic::Status::deadline_exceeded(format!("Timeout: {}", msg))
203            }
204            DistributedError::Authentication(msg) => {
205                tonic::Status::unauthenticated(format!("Authentication failed: {}", msg))
206            }
207            DistributedError::ResourceAllocation(msg) => {
208                tonic::Status::resource_exhausted(format!("Resource allocation error: {}", msg))
209            }
210            DistributedError::Aggregation(msg) => {
211                tonic::Status::internal(format!("Aggregation error: {}", msg))
212            }
213            DistributedError::Arrow(msg) => {
214                tonic::Status::internal(format!("Arrow error: {}", msg))
215            }
216            DistributedError::OperationNotImplemented(msg) => {
217                tonic::Status::unimplemented(format!("Operation not implemented: {}", msg))
218            }
219            DistributedError::InvalidOperation(msg) => {
220                tonic::Status::invalid_argument(format!("Invalid operation: {}", msg))
221            }
222            DistributedError::Core(err) => tonic::Status::internal(format!("Core error: {}", err)),
223            DistributedError::Io(err) => tonic::Status::internal(format!("I/O error: {}", err)),
224            DistributedError::Json(err) => {
225                tonic::Status::invalid_argument(format!("JSON error: {}", err))
226            }
227            DistributedError::Transport(err) => {
228                tonic::Status::unavailable(format!("Transport error: {}", err))
229            }
230            DistributedError::Status(status) => status,
231            DistributedError::Custom(msg) => tonic::Status::internal(msg),
232        }
233    }
234}
235
236#[cfg(test)]
237mod tests {
238    use super::*;
239
240    #[test]
241    fn test_error_creation() {
242        let err = DistributedError::flight_rpc("test error");
243        assert!(matches!(err, DistributedError::FlightRpc(_)));
244
245        let err = DistributedError::worker_connection("connection failed");
246        assert!(matches!(err, DistributedError::WorkerConnection(_)));
247
248        let err = DistributedError::timeout("operation timed out");
249        assert!(matches!(err, DistributedError::Timeout(_)));
250    }
251
252    #[test]
253    fn test_error_display() {
254        let err = DistributedError::flight_rpc("test error");
255        let msg = format!("{}", err);
256        assert!(msg.contains("Flight RPC error"));
257        assert!(msg.contains("test error"));
258    }
259
260    #[test]
261    fn test_to_tonic_status() {
262        let err = DistributedError::flight_rpc("test");
263        let status: tonic::Status = err.into();
264        assert_eq!(status.code(), tonic::Code::Internal);
265
266        let err = DistributedError::authentication("invalid token");
267        let status: tonic::Status = err.into();
268        assert_eq!(status.code(), tonic::Code::Unauthenticated);
269
270        let err = DistributedError::timeout("too slow");
271        let status: tonic::Status = err.into();
272        assert_eq!(status.code(), tonic::Code::DeadlineExceeded);
273    }
274}
oxigdal_distributed/error.rs

oxigdal_distributed/
error.rs
