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    /// Structured committed cluster-subscription failure.
77    Subscription(#[from] crate::subscription::ClusterSubscriptionError),
78
79    /// SQL parse error (from streaming parser)
80    SqlParse(#[from] laminar_sql::parser::ParseError),
81
82    /// Database is shut down
83    Shutdown,
84
85    /// Checkpoint error
86    Checkpoint(String),
87
88    /// Checkpoint store error (preserves structured source error).
89    CheckpointStore(#[from] laminar_core::checkpoint::checkpoint_store::CheckpointStoreError),
90
91    /// Unresolved config variable
92    UnresolvedConfigVar(String),
93
94    /// Connector error
95    Connector(String),
96
97    /// Connector operation error (preserves structured source error).
98    ConnectorOp(#[from] laminar_connectors::error::ConnectorError),
99
100    /// Pipeline error (start/shutdown lifecycle)
101    Pipeline(String),
102
103    /// A deterministic record-path failure that retry or checkpoint recovery cannot repair.
104    PipelineTerminal(String),
105
106    /// `BackpressurePolicy::Fail` tripped; coordinator halts the pipeline.
107    BackpressureFail(String),
108
109    /// A cross-node shuffle target isn't reachable yet (cluster formation).
110    /// Recoverable — `OperatorGraph::execute_single_operator` defers on it.
111    ShuffleNotReady(String),
112
113    /// A permanent structural shuffle-routing failure. Retrying or restoring the same input cannot
114    /// repair it, so the pipeline must halt instead of entering a recovery loop.
115    ShuffleTerminal(String),
116
117    /// A cross-node shuffle send failed after an earlier frame was admitted to the transport and
118    /// may reach its peer. Unlike [`Self::ShuffleNotReady`], this must not be replayed locally: a
119    /// retry could double-fold the admitted rows. Recovery rewinds the complete failure domain to a
120    /// durable cut before replay.
121    ShufflePartialSend(String),
122
123    /// A stateful operator may have changed local state before the attempt failed. This denotes an
124    /// indeterminate apply outcome, not proof that exactly a prefix was applied. The graph must not
125    /// replay that input against the possibly changed state; recovery rewinds the failure domain to
126    /// its last durable cut.
127    StatefulOperatorPartialApply(String),
128
129    /// Managed working state crossed the configured pipeline-wide charged-byte envelope. The
130    /// current graph generation must halt: replaying the same input against the same limit would
131    /// loop, and a record-path detection may follow state mutation but always precedes output.
132    ManagedStateBudgetExceeded {
133        /// Boundary at which the excess was detected.
134        context: String,
135        /// Combined live, prepared, and retired charged bytes.
136        accounted_bytes: usize,
137        /// Configured pipeline-wide maximum.
138        limit_bytes: usize,
139    },
140
141    /// Query pipeline error — wraps a `DataFusion` error with stream context.
142    /// Unlike `Pipeline`, this variant is translated to user-friendly messages.
143    QueryPipeline {
144        /// The stream or query name where the error occurred.
145        context: String,
146        /// The translated error message (already processed through
147        /// `translate_datafusion_error`).
148        translated: String,
149    },
150
151    /// Materialized view error
152    MaterializedView(String),
153
154    /// Storage backend error.
155    Storage(String),
156
157    /// Configuration / profile validation error
158    Config(String),
159
160    /// Operation is not yet implemented.
161    Unsupported(String),
162}
163
164impl DbError {
165    /// Create a `QueryPipeline` error from a `DataFusion` error with stream context.
166    ///
167    /// The `DataFusion` error is translated to a user-friendly message with
168    /// structured error codes. The raw `DataFusion` internals are never exposed.
169    pub fn query_pipeline(
170        context: impl Into<String>,
171        df_error: &datafusion_common::DataFusionError,
172    ) -> Self {
173        let translated = laminar_sql::error::translate_datafusion_error(&df_error.to_string());
174        Self::QueryPipeline {
175            context: context.into(),
176            translated: translated.to_string(),
177        }
178    }
179
180    /// Create a `QueryPipeline` error from a `DataFusion` error with stream
181    /// context and available column names for typo suggestions.
182    pub fn query_pipeline_with_columns(
183        context: impl Into<String>,
184        df_error: &datafusion_common::DataFusionError,
185        available_columns: &[&str],
186    ) -> Self {
187        let translated = laminar_sql::error::translate_datafusion_error_with_context(
188            &df_error.to_string(),
189            Some(available_columns),
190        );
191        Self::QueryPipeline {
192            context: context.into(),
193            translated: translated.to_string(),
194        }
195    }
196
197    /// Create a `QueryPipeline` error from an Arrow error with stream context.
198    pub fn query_pipeline_arrow(
199        context: impl Into<String>,
200        arrow_error: &arrow::error::ArrowError,
201    ) -> Self {
202        let translated = laminar_sql::error::translate_datafusion_error(&arrow_error.to_string());
203        Self::QueryPipeline {
204            context: context.into(),
205            translated: translated.to_string(),
206        }
207    }
208
209    /// Returns the structured `LDB-NNNN` error code for this error.
210    ///
211    /// Every `DbError` variant maps to a stable error code that can be used
212    /// for programmatic handling, log searching, and metrics.
213    #[must_use]
214    pub fn code(&self) -> &'static str {
215        match self {
216            Self::Sql(_) | Self::SqlParse(_) => error_codes::SQL_UNSUPPORTED,
217            Self::Engine(_) | Self::Streaming(_) => error_codes::INTERNAL,
218            Self::DataFusion(_) => error_codes::QUERY_EXECUTION_FAILED,
219            Self::SourceNotFound(_) => error_codes::SOURCE_NOT_FOUND,
220            Self::SinkNotFound(_) => error_codes::SINK_NOT_FOUND,
221            Self::QueryNotFound(_) | Self::StreamNotFound(_) | Self::TableNotFound(_) => {
222                error_codes::SQL_TABLE_NOT_FOUND
223            }
224            Self::SourceAlreadyExists(_)
225            | Self::StreamAlreadyExists(_)
226            | Self::TableAlreadyExists(_) => error_codes::SOURCE_ALREADY_EXISTS,
227            Self::SinkAlreadyExists(_) => error_codes::SINK_ALREADY_EXISTS,
228            Self::InsertError(_) => error_codes::CONNECTOR_WRITE_ERROR,
229            Self::SchemaMismatch(_) => error_codes::SCHEMA_MISMATCH,
230            Self::InvalidOperation(_)
231            | Self::SubscriptionReplayPruned { .. }
232            | Self::SubscriptionEpochNotCommitted { .. }
233            | Self::Unsupported(_) => error_codes::INVALID_OPERATION,
234            Self::Subscription(error) => error.code(),
235            Self::Shutdown => error_codes::SHUTDOWN,
236            Self::Checkpoint(_) | Self::CheckpointStore(_) => error_codes::CHECKPOINT_FAILED,
237            Self::UnresolvedConfigVar(_) => error_codes::UNRESOLVED_CONFIG_VAR,
238            Self::Connector(_) | Self::ConnectorOp(_) => error_codes::CONNECTOR_CONNECTION_FAILED,
239            Self::Pipeline(_)
240            | Self::PipelineTerminal(_)
241            | Self::BackpressureFail(_)
242            | Self::ShuffleNotReady(_)
243            | Self::ShuffleTerminal(_)
244            | Self::ShufflePartialSend(_)
245            | Self::StatefulOperatorPartialApply(_) => error_codes::PIPELINE_ERROR,
246            Self::ManagedStateBudgetExceeded { .. } => error_codes::MANAGED_STATE_BUDGET_EXCEEDED,
247            Self::QueryPipeline { .. } => error_codes::QUERY_PIPELINE_ERROR,
248            Self::MaterializedView(_) => error_codes::MATERIALIZED_VIEW_ERROR,
249            Self::Storage(_) => error_codes::WAL_ERROR,
250            Self::Config(_) => error_codes::INVALID_CONFIG,
251        }
252    }
253
254    /// `true` for [`Self::ShuffleNotReady`] — the operator defers instead of failing the cycle.
255    #[must_use]
256    pub fn is_shuffle_not_ready(&self) -> bool {
257        matches!(self, Self::ShuffleNotReady(_))
258    }
259
260    /// `true` when retry or recovery cannot repair the current input and the pipeline must stop.
261    #[must_use]
262    pub fn requires_pipeline_halt(&self) -> bool {
263        matches!(
264            self,
265            Self::PipelineTerminal(_)
266                | Self::BackpressureFail(_)
267                | Self::ShuffleTerminal(_)
268                | Self::ManagedStateBudgetExceeded { .. }
269        )
270    }
271
272    /// `true` when failure-domain isolation cannot safely keep the current pipeline alive.
273    #[must_use]
274    pub fn requires_pipeline_recovery(&self) -> bool {
275        matches!(
276            self,
277            Self::Checkpoint(_)
278                | Self::ShufflePartialSend(_)
279                | Self::StatefulOperatorPartialApply(_)
280        )
281    }
282
283    /// Whether this error is transient (retryable).
284    #[must_use]
285    pub fn is_transient(&self) -> bool {
286        match self {
287            Self::Streaming(_)
288            | Self::Connector(_)
289            | Self::Checkpoint(_)
290            | Self::CheckpointStore(_) => true,
291            Self::ConnectorOp(e) => e.is_transient(),
292            _ => false,
293        }
294    }
295}
296
297impl std::fmt::Display for DbError {
298    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
299        match self {
300            Self::Sql(e) => write!(f, "SQL error: {e}"),
301            Self::Engine(e) => write!(f, "Engine error: {e}"),
302            Self::Streaming(e) => write!(f, "Streaming error: {e}"),
303            Self::DataFusion(e) => {
304                let translated = laminar_sql::error::translate_datafusion_error(&e.to_string());
305                write!(f, "{translated}")
306            }
307            Self::SourceNotFound(name) => {
308                write!(f, "[{}] Source '{name}' not found", self.code())
309            }
310            Self::SinkNotFound(name) => {
311                write!(f, "[{}] Sink '{name}' not found", self.code())
312            }
313            Self::QueryNotFound(name) => {
314                write!(f, "[{}] Query '{name}' not found", self.code())
315            }
316            Self::SourceAlreadyExists(name) => {
317                write!(f, "[{}] Source '{name}' already exists", self.code())
318            }
319            Self::SinkAlreadyExists(name) => {
320                write!(f, "[{}] Sink '{name}' already exists", self.code())
321            }
322            Self::StreamNotFound(name) => {
323                write!(f, "[{}] Stream '{name}' not found", self.code())
324            }
325            Self::StreamAlreadyExists(name) => {
326                write!(f, "[{}] Stream '{name}' already exists", self.code())
327            }
328            Self::TableNotFound(name) => {
329                write!(f, "[{}] Table '{name}' not found", self.code())
330            }
331            Self::TableAlreadyExists(name) => {
332                write!(f, "[{}] Table '{name}' already exists", self.code())
333            }
334            Self::InsertError(msg) => {
335                write!(f, "[{}] Insert error: {msg}", self.code())
336            }
337            Self::SchemaMismatch(msg) => {
338                write!(f, "[{}] Schema mismatch: {msg}", self.code())
339            }
340            Self::InvalidOperation(msg) => {
341                write!(f, "[{}] Invalid operation: {msg}", self.code())
342            }
343            Self::SubscriptionReplayPruned {
344                name,
345                requested,
346                earliest_retained,
347            } => write!(
348                f,
349                "[{}] Epoch {requested} for stream '{name}' is no longer retained (earliest retained is {earliest_retained})",
350                self.code()
351            ),
352            Self::SubscriptionEpochNotCommitted {
353                name,
354                requested,
355                latest_committed,
356            } => match latest_committed {
357                Some(latest) => write!(
358                    f,
359                    "[{}] Epoch {requested} for stream '{name}' is not committed (latest committed is {latest})",
360                    self.code()
361                ),
362                None => write!(
363                    f,
364                    "[{}] Epoch {requested} for stream '{name}' is not committed (no committed epoch is available)",
365                    self.code()
366                ),
367            },
368            Self::Subscription(error) => write!(f, "[{}] {error}", self.code()),
369            Self::SqlParse(e) => write!(f, "SQL parse error: {e}"),
370            Self::Shutdown => write!(f, "[{}] Database is shut down", self.code()),
371            Self::Checkpoint(msg) => {
372                write!(f, "[{}] Checkpoint error: {msg}", self.code())
373            }
374            Self::CheckpointStore(e) => {
375                write!(f, "[{}] Checkpoint store error: {e}", self.code())
376            }
377            Self::UnresolvedConfigVar(msg) => {
378                write!(f, "[{}] Unresolved config variable: {msg}", self.code())
379            }
380            Self::Connector(msg) => {
381                write!(f, "[{}] Connector error: {msg}", self.code())
382            }
383            Self::ConnectorOp(e) => {
384                write!(f, "[{}] Connector error: {e}", self.code())
385            }
386            Self::Pipeline(msg) => {
387                write!(f, "[{}] Pipeline error: {msg}", self.code())
388            }
389            Self::PipelineTerminal(msg) => {
390                write!(f, "[{}] Terminal pipeline error: {msg}", self.code())
391            }
392            Self::BackpressureFail(msg) => {
393                write!(f, "[{}] Backpressure fail: {msg}", self.code())
394            }
395            Self::ShuffleNotReady(msg) => {
396                write!(f, "[{}] Shuffle target not ready: {msg}", self.code())
397            }
398            Self::ShuffleTerminal(msg) => {
399                write!(f, "[{}] Terminal shuffle routing error: {msg}", self.code())
400            }
401            Self::ShufflePartialSend(msg) => {
402                write!(f, "[{}] Shuffle partial send: {msg}", self.code())
403            }
404            Self::StatefulOperatorPartialApply(msg) => {
405                write!(
406                    f,
407                    "[{}] Stateful operator partial apply: {msg}",
408                    self.code()
409                )
410            }
411            Self::ManagedStateBudgetExceeded {
412                context,
413                accounted_bytes,
414                limit_bytes,
415            } => write!(
416                f,
417                "[{}] Managed state budget exceeded during {context}: accounted={accounted_bytes} bytes, limit={limit_bytes} bytes",
418                self.code()
419            ),
420            Self::QueryPipeline {
421                context,
422                translated,
423            } => write!(f, "Stream '{context}': {translated}"),
424            Self::MaterializedView(msg) => {
425                write!(f, "[{}] Materialized view error: {msg}", self.code())
426            }
427            Self::Storage(msg) => {
428                write!(f, "[{}] Storage error: {msg}", self.code())
429            }
430            Self::Config(msg) => {
431                write!(f, "[{}] Config error: {msg}", self.code())
432            }
433            Self::Unsupported(msg) => {
434                write!(f, "[{}] Unsupported: {msg}", self.code())
435            }
436        }
437    }
438}
laminar_db/error.rs

laminar_db/
error.rs
