laminar_db/
error.rs

1//! Error types for the `LaminarDB` facade.
2
3use laminar_core::error_codes;
4
5/// Errors from database operations.
6#[derive(Debug, thiserror::Error)]
7pub enum DbError {
8    /// SQL parse error
9    Sql(#[from] laminar_sql::Error),
10
11    /// Core engine error
12    Engine(#[from] laminar_core::Error),
13
14    /// Streaming API error
15    Streaming(#[from] laminar_core::streaming::StreamingError),
16
17    /// `DataFusion` error (translated to user-friendly messages on display)
18    DataFusion(#[from] datafusion_common::DataFusionError),
19
20    /// Source not found
21    SourceNotFound(String),
22
23    /// Sink not found
24    SinkNotFound(String),
25
26    /// Query not found
27    QueryNotFound(String),
28
29    /// Source already exists
30    SourceAlreadyExists(String),
31
32    /// Sink already exists
33    SinkAlreadyExists(String),
34
35    /// Stream not found
36    StreamNotFound(String),
37
38    /// Stream already exists
39    StreamAlreadyExists(String),
40
41    /// Table not found
42    TableNotFound(String),
43
44    /// Table already exists
45    TableAlreadyExists(String),
46
47    /// Insert error
48    InsertError(String),
49
50    /// Schema mismatch between Rust type and SQL definition
51    SchemaMismatch(String),
52
53    /// Invalid SQL statement for the operation
54    InvalidOperation(String),
55
56    /// Requested subscription epoch has committed but its replay suffix was pruned.
57    SubscriptionReplayPruned {
58        /// Subscription object name.
59        name: String,
60        /// Checkpoint epoch requested by the subscriber.
61        requested: u64,
62        /// Earliest checkpoint epoch whose cut remains replayable.
63        earliest_retained: u64,
64    },
65
66    /// Requested subscription epoch has not committed.
67    SubscriptionEpochNotCommitted {
68        /// Subscription object name.
69        name: String,
70        /// Checkpoint epoch requested by the subscriber.
71        requested: u64,
72        /// Newest durably committed checkpoint epoch, if one exists.
73        latest_committed: Option<u64>,
74    },
75
76    /// SQL parse error (from streaming parser)
77    SqlParse(#[from] laminar_sql::parser::ParseError),
78
79    /// Database is shut down
80    Shutdown,
81
82    /// Checkpoint error
83    Checkpoint(String),
84
85    /// Checkpoint store error (preserves structured source error).
86    CheckpointStore(#[from] laminar_core::storage::checkpoint_store::CheckpointStoreError),
87
88    /// Unresolved config variable
89    UnresolvedConfigVar(String),
90
91    /// Connector error
92    Connector(String),
93
94    /// Connector operation error (preserves structured source error).
95    ConnectorOp(#[from] laminar_connectors::error::ConnectorError),
96
97    /// Pipeline error (start/shutdown lifecycle)
98    Pipeline(String),
99
100    /// `BackpressurePolicy::Fail` tripped; coordinator halts the pipeline.
101    BackpressureFail(String),
102
103    /// A cross-node shuffle target isn't reachable yet (cluster formation).
104    /// Recoverable — `OperatorGraph::execute_single_operator` defers on it.
105    ShuffleNotReady(String),
106
107    /// A permanent structural shuffle-routing failure. Retrying or restoring the same input cannot
108    /// repair it, so the pipeline must halt instead of entering a recovery loop.
109    ShuffleTerminal(String),
110
111    /// A cross-node shuffle send failed after an earlier frame was admitted to the transport and
112    /// may reach its peer. Unlike [`Self::ShuffleNotReady`], this must not be replayed locally: a
113    /// retry could double-fold the admitted rows. Recovery rewinds the complete failure domain to a
114    /// durable cut before replay.
115    ShufflePartialSend(String),
116
117    /// A stateful operator admitted input before a later step in the same cycle failed. The graph
118    /// must not replay that input against the mutated local state; recovery rewinds the failure
119    /// domain to its last durable cut.
120    StatefulOperatorPartialApply(String),
121
122    /// Query pipeline error — wraps a `DataFusion` error with stream context.
123    /// Unlike `Pipeline`, this variant is translated to user-friendly messages.
124    QueryPipeline {
125        /// The stream or query name where the error occurred.
126        context: String,
127        /// The translated error message (already processed through
128        /// `translate_datafusion_error`).
129        translated: String,
130    },
131
132    /// Materialized view error
133    MaterializedView(String),
134
135    /// Storage backend error.
136    Storage(String),
137
138    /// Configuration / profile validation error
139    Config(String),
140
141    /// Operation is not yet implemented.
142    Unsupported(String),
143}
144
145impl DbError {
146    /// Create a `QueryPipeline` error from a `DataFusion` error with stream context.
147    ///
148    /// The `DataFusion` error is translated to a user-friendly message with
149    /// structured error codes. The raw `DataFusion` internals are never exposed.
150    pub fn query_pipeline(
151        context: impl Into<String>,
152        df_error: &datafusion_common::DataFusionError,
153    ) -> Self {
154        let translated = laminar_sql::error::translate_datafusion_error(&df_error.to_string());
155        Self::QueryPipeline {
156            context: context.into(),
157            translated: translated.to_string(),
158        }
159    }
160
161    /// Create a `QueryPipeline` error from a `DataFusion` error with stream
162    /// context and available column names for typo suggestions.
163    pub fn query_pipeline_with_columns(
164        context: impl Into<String>,
165        df_error: &datafusion_common::DataFusionError,
166        available_columns: &[&str],
167    ) -> Self {
168        let translated = laminar_sql::error::translate_datafusion_error_with_context(
169            &df_error.to_string(),
170            Some(available_columns),
171        );
172        Self::QueryPipeline {
173            context: context.into(),
174            translated: translated.to_string(),
175        }
176    }
177
178    /// Create a `QueryPipeline` error from an Arrow error with stream context.
179    pub fn query_pipeline_arrow(
180        context: impl Into<String>,
181        arrow_error: &arrow::error::ArrowError,
182    ) -> Self {
183        let translated = laminar_sql::error::translate_datafusion_error(&arrow_error.to_string());
184        Self::QueryPipeline {
185            context: context.into(),
186            translated: translated.to_string(),
187        }
188    }
189
190    /// Returns the structured `LDB-NNNN` error code for this error.
191    ///
192    /// Every `DbError` variant maps to a stable error code that can be used
193    /// for programmatic handling, log searching, and metrics.
194    #[must_use]
195    pub fn code(&self) -> &'static str {
196        match self {
197            Self::Sql(_) | Self::SqlParse(_) => error_codes::SQL_UNSUPPORTED,
198            Self::Engine(_) | Self::Streaming(_) => error_codes::INTERNAL,
199            Self::DataFusion(_) => error_codes::QUERY_EXECUTION_FAILED,
200            Self::SourceNotFound(_) => error_codes::SOURCE_NOT_FOUND,
201            Self::SinkNotFound(_) => error_codes::SINK_NOT_FOUND,
202            Self::QueryNotFound(_) | Self::StreamNotFound(_) | Self::TableNotFound(_) => {
203                error_codes::SQL_TABLE_NOT_FOUND
204            }
205            Self::SourceAlreadyExists(_)
206            | Self::StreamAlreadyExists(_)
207            | Self::TableAlreadyExists(_) => error_codes::SOURCE_ALREADY_EXISTS,
208            Self::SinkAlreadyExists(_) => error_codes::SINK_ALREADY_EXISTS,
209            Self::InsertError(_) => error_codes::CONNECTOR_WRITE_ERROR,
210            Self::SchemaMismatch(_) => error_codes::SCHEMA_MISMATCH,
211            Self::InvalidOperation(_)
212            | Self::SubscriptionReplayPruned { .. }
213            | Self::SubscriptionEpochNotCommitted { .. }
214            | Self::Unsupported(_) => error_codes::INVALID_OPERATION,
215            Self::Shutdown => error_codes::SHUTDOWN,
216            Self::Checkpoint(_) | Self::CheckpointStore(_) => error_codes::CHECKPOINT_FAILED,
217            Self::UnresolvedConfigVar(_) => error_codes::UNRESOLVED_CONFIG_VAR,
218            Self::Connector(_) | Self::ConnectorOp(_) => error_codes::CONNECTOR_CONNECTION_FAILED,
219            Self::Pipeline(_)
220            | Self::BackpressureFail(_)
221            | Self::ShuffleNotReady(_)
222            | Self::ShuffleTerminal(_)
223            | Self::ShufflePartialSend(_)
224            | Self::StatefulOperatorPartialApply(_) => error_codes::PIPELINE_ERROR,
225            Self::QueryPipeline { .. } => error_codes::QUERY_PIPELINE_ERROR,
226            Self::MaterializedView(_) => error_codes::MATERIALIZED_VIEW_ERROR,
227            Self::Storage(_) => error_codes::WAL_ERROR,
228            Self::Config(_) => error_codes::INVALID_CONFIG,
229        }
230    }
231
232    /// `true` for [`Self::ShuffleNotReady`] — the operator defers instead of failing the cycle.
233    #[must_use]
234    pub fn is_shuffle_not_ready(&self) -> bool {
235        matches!(self, Self::ShuffleNotReady(_))
236    }
237
238    /// `true` when retry or recovery cannot repair the current input and the pipeline must stop.
239    #[must_use]
240    pub fn requires_pipeline_halt(&self) -> bool {
241        matches!(self, Self::BackpressureFail(_) | Self::ShuffleTerminal(_))
242    }
243
244    /// `true` when failure-domain isolation cannot safely keep the current pipeline alive.
245    #[must_use]
246    pub fn requires_pipeline_recovery(&self) -> bool {
247        matches!(
248            self,
249            Self::Checkpoint(_)
250                | Self::ShufflePartialSend(_)
251                | Self::StatefulOperatorPartialApply(_)
252        )
253    }
254
255    /// Whether this error is transient (retryable).
256    #[must_use]
257    pub fn is_transient(&self) -> bool {
258        match self {
259            Self::Streaming(_)
260            | Self::Connector(_)
261            | Self::Checkpoint(_)
262            | Self::CheckpointStore(_) => true,
263            Self::ConnectorOp(e) => e.is_transient(),
264            _ => false,
265        }
266    }
267}
268
269impl std::fmt::Display for DbError {
270    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
271        match self {
272            Self::Sql(e) => write!(f, "SQL error: {e}"),
273            Self::Engine(e) => write!(f, "Engine error: {e}"),
274            Self::Streaming(e) => write!(f, "Streaming error: {e}"),
275            Self::DataFusion(e) => {
276                let translated = laminar_sql::error::translate_datafusion_error(&e.to_string());
277                write!(f, "{translated}")
278            }
279            Self::SourceNotFound(name) => {
280                write!(f, "[{}] Source '{name}' not found", self.code())
281            }
282            Self::SinkNotFound(name) => {
283                write!(f, "[{}] Sink '{name}' not found", self.code())
284            }
285            Self::QueryNotFound(name) => {
286                write!(f, "[{}] Query '{name}' not found", self.code())
287            }
288            Self::SourceAlreadyExists(name) => {
289                write!(f, "[{}] Source '{name}' already exists", self.code())
290            }
291            Self::SinkAlreadyExists(name) => {
292                write!(f, "[{}] Sink '{name}' already exists", self.code())
293            }
294            Self::StreamNotFound(name) => {
295                write!(f, "[{}] Stream '{name}' not found", self.code())
296            }
297            Self::StreamAlreadyExists(name) => {
298                write!(f, "[{}] Stream '{name}' already exists", self.code())
299            }
300            Self::TableNotFound(name) => {
301                write!(f, "[{}] Table '{name}' not found", self.code())
302            }
303            Self::TableAlreadyExists(name) => {
304                write!(f, "[{}] Table '{name}' already exists", self.code())
305            }
306            Self::InsertError(msg) => {
307                write!(f, "[{}] Insert error: {msg}", self.code())
308            }
309            Self::SchemaMismatch(msg) => {
310                write!(f, "[{}] Schema mismatch: {msg}", self.code())
311            }
312            Self::InvalidOperation(msg) => {
313                write!(f, "[{}] Invalid operation: {msg}", self.code())
314            }
315            Self::SubscriptionReplayPruned {
316                name,
317                requested,
318                earliest_retained,
319            } => write!(
320                f,
321                "[{}] Epoch {requested} for stream '{name}' is no longer retained (earliest retained is {earliest_retained})",
322                self.code()
323            ),
324            Self::SubscriptionEpochNotCommitted {
325                name,
326                requested,
327                latest_committed,
328            } => match latest_committed {
329                Some(latest) => write!(
330                    f,
331                    "[{}] Epoch {requested} for stream '{name}' is not committed (latest committed is {latest})",
332                    self.code()
333                ),
334                None => write!(
335                    f,
336                    "[{}] Epoch {requested} for stream '{name}' is not committed (no committed epoch is available)",
337                    self.code()
338                ),
339            },
340            Self::SqlParse(e) => write!(f, "SQL parse error: {e}"),
341            Self::Shutdown => {
342                write!(f, "[{}] Database is shut down", self.code())
343            }
344            Self::Checkpoint(msg) => {
345                write!(f, "[{}] Checkpoint error: {msg}", self.code())
346            }
347            Self::CheckpointStore(e) => {
348                write!(f, "[{}] Checkpoint store error: {e}", self.code())
349            }
350            Self::UnresolvedConfigVar(msg) => {
351                write!(f, "[{}] Unresolved config variable: {msg}", self.code())
352            }
353            Self::Connector(msg) => {
354                write!(f, "[{}] Connector error: {msg}", self.code())
355            }
356            Self::ConnectorOp(e) => {
357                write!(f, "[{}] Connector error: {e}", self.code())
358            }
359            Self::Pipeline(msg) => {
360                write!(f, "[{}] Pipeline error: {msg}", self.code())
361            }
362            Self::BackpressureFail(msg) => {
363                write!(f, "[{}] Backpressure fail: {msg}", self.code())
364            }
365            Self::ShuffleNotReady(msg) => {
366                write!(f, "[{}] Shuffle target not ready: {msg}", self.code())
367            }
368            Self::ShuffleTerminal(msg) => {
369                write!(f, "[{}] Terminal shuffle routing error: {msg}", self.code())
370            }
371            Self::ShufflePartialSend(msg) => {
372                write!(f, "[{}] Shuffle partial send: {msg}", self.code())
373            }
374            Self::StatefulOperatorPartialApply(msg) => {
375                write!(
376                    f,
377                    "[{}] Stateful operator partial apply: {msg}",
378                    self.code()
379                )
380            }
381            Self::QueryPipeline {
382                context,
383                translated,
384            } => write!(f, "Stream '{context}': {translated}"),
385            Self::MaterializedView(msg) => {
386                write!(f, "[{}] Materialized view error: {msg}", self.code())
387            }
388            Self::Storage(msg) => {
389                write!(f, "[{}] Storage error: {msg}", self.code())
390            }
391            Self::Config(msg) => {
392                write!(f, "[{}] Config error: {msg}", self.code())
393            }
394            Self::Unsupported(msg) => {
395                write!(f, "[{}] Unsupported: {msg}", self.code())
396            }
397        }
398    }
399}
laminar_db/error.rs

laminar_db/
error.rs
