timeseries_table_format/table/
error.rs

1//! Top-level error facade for table operations.
2//!
3//! [`TableError`] adds only operation context. Each variant wraps the complete
4//! error owned by that operation and delegates its source and backtrace. Add
5//! detailed failures to the owning operation error rather than copying
6//! subsystem-specific variants into this facade.
7
8use snafu::prelude::*;
9
10use super::operations::{
11    AppendError, CoverageQueryError, CreateTableError, OpenTableError, OptimizeError, ScanError,
12    TableStateAccessError,
13};
14
15/// Errors from high-level time-series table operations.
16///
17/// Each variant carries enough context for callers to surface actionable
18/// messages to users or implement retries where appropriate (for example,
19/// conflicts on optimistic concurrency control).
20#[derive(Debug, Snafu)]
21#[snafu(visibility(pub(crate)))]
22#[non_exhaustive]
23pub enum TableError {
24    /// A table creation operation failed.
25    #[snafu(display("Table creation failed: {source}"))]
26    Create {
27        /// Complete table creation failure.
28        #[snafu(source, backtrace)]
29        source: CreateTableError,
30    },
31
32    /// A table open operation failed.
33    #[snafu(display("Table open failed: {source}"))]
34    Open {
35        /// Complete table open failure.
36        #[snafu(source, backtrace)]
37        source: OpenTableError,
38    },
39
40    /// A table state access or refresh operation failed.
41    #[snafu(display("Table state access failed: {source}"))]
42    StateAccess {
43        /// Complete state-access failure.
44        #[snafu(source, backtrace)]
45        source: TableStateAccessError,
46    },
47
48    /// An append operation failed.
49    #[snafu(display("Append failed: {source}"))]
50    Append {
51        /// Complete append-owned failure.
52        #[snafu(source, backtrace)]
53        source: AppendError,
54    },
55
56    /// A table scan failed during planning or lazy execution.
57    #[snafu(display("Table scan failed: {source}"))]
58    Scan {
59        /// Complete scan operation error.
60        #[snafu(source, backtrace)]
61        source: ScanError,
62    },
63
64    /// A table coverage query failed.
65    #[snafu(display("Table coverage query failed: {source}"))]
66    CoverageQuery {
67        /// Complete coverage query operation error.
68        #[snafu(source, backtrace)]
69        source: CoverageQueryError,
70    },
71
72    /// An entity-layout optimization operation failed.
73    #[snafu(display("Entity-layout optimization failed: {source}"))]
74    Optimize {
75        /// Complete optimization-owned failure.
76        #[snafu(source, backtrace)]
77        source: OptimizeError,
78    },
79}
80
81#[cfg(test)]
82mod tests {
83    use std::{error::Error as _, io};
84
85    use arrow::error::ArrowError;
86    use parquet::errors::ParquetError;
87    use snafu::{Backtrace, ErrorCompat, IntoError};
88
89    use crate::coverage::{CoverageCodecError, CoverageSidecarError};
90    use crate::formats::parquet::EntityRewriteError;
91    use crate::metadata::{
92        index::IndexSpecError, logical_schema::LogicalToArrowSchemaError,
93        schema_compat::SchemaCompatibilityError,
94    };
95    use crate::storage::StorageError;
96    use crate::transaction_log::CommitError;
97
98    use super::*;
99
100    #[test]
101    fn append_facade_preserves_arrow_source_and_backtrace() {
102        let append_error = AppendError::ArrowInput {
103            source: ArrowError::ComputeError("input failed".to_string()),
104            backtrace: Backtrace::capture(),
105        };
106        let error = AppendSnafu.into_error(append_error);
107
108        let append_source = error
109            .source()
110            .and_then(|source| source.downcast_ref::<AppendError>())
111            .expect("append source");
112        let arrow_source = append_source
113            .source()
114            .and_then(|source| source.downcast_ref::<ArrowError>())
115            .expect("Arrow source");
116        let append_backtrace = ErrorCompat::backtrace(append_source).expect("append backtrace");
117        let table_backtrace = ErrorCompat::backtrace(&error).expect("table backtrace");
118
119        assert!(
120            matches!(arrow_source, ArrowError::ComputeError(message) if message == "input failed")
121        );
122        assert!(std::ptr::eq(table_backtrace, append_backtrace));
123    }
124
125    #[test]
126    fn append_facade_preserves_parquet_source_and_backtrace() {
127        let error = AppendSnafu.into_error(AppendError::ParquetWrite {
128            source: ParquetError::General("write failed".to_string()),
129            backtrace: Backtrace::capture(),
130        });
131
132        let append_source = error
133            .source()
134            .and_then(|source| source.downcast_ref::<AppendError>())
135            .expect("append source");
136        let parquet_source = append_source
137            .source()
138            .and_then(|source| source.downcast_ref::<ParquetError>())
139            .expect("Parquet source");
140        let append_backtrace = ErrorCompat::backtrace(append_source).expect("append backtrace");
141        let table_backtrace = ErrorCompat::backtrace(&error).expect("table backtrace");
142
143        assert!(
144            matches!(parquet_source, ParquetError::General(message) if message == "write failed")
145        );
146        assert!(std::ptr::eq(table_backtrace, append_backtrace));
147    }
148
149    #[test]
150    fn append_schema_validation_leaf_does_not_manufacture_a_backtrace() {
151        let error = AppendError::from(SchemaCompatibilityError::MissingTableSchema);
152
153        assert!(error.to_string().starts_with("Schema validation failed:"));
154        assert!(ErrorCompat::backtrace(&error).is_none());
155    }
156
157    #[test]
158    fn schema_wrappers_delegate_the_originating_backtrace() {
159        let schema = SchemaCompatibilityError::RegisteredSchemaConversion {
160            source: Box::new(LogicalToArrowSchemaError::Int96Unsupported {
161                column: "time".to_string(),
162                backtrace: Backtrace::capture(),
163            }),
164        };
165        let error = AppendSnafu.into_error(AppendError::from(schema));
166
167        let append = error
168            .source()
169            .and_then(|source| source.downcast_ref::<AppendError>())
170            .expect("append source");
171        let schema = append
172            .source()
173            .and_then(|source| source.downcast_ref::<Box<SchemaCompatibilityError>>())
174            .map(Box::as_ref)
175            .expect("schema compatibility source");
176        let conversion = schema
177            .source()
178            .and_then(|source| source.downcast_ref::<Box<LogicalToArrowSchemaError>>())
179            .map(Box::as_ref)
180            .expect("logical-to-Arrow source");
181        let originating_backtrace = ErrorCompat::backtrace(conversion).expect("source backtrace");
182
183        assert!(std::ptr::eq(
184            ErrorCompat::backtrace(schema).expect("schema backtrace"),
185            originating_backtrace
186        ));
187        assert!(std::ptr::eq(
188            ErrorCompat::backtrace(append).expect("append backtrace"),
189            originating_backtrace
190        ));
191        assert!(std::ptr::eq(
192            ErrorCompat::backtrace(&error).expect("table backtrace"),
193            originating_backtrace
194        ));
195    }
196
197    #[test]
198    fn append_rollback_preserves_primary_chain_cleanup_errors_and_backtrace() {
199        let commit_error = CommitError::Conflict {
200            expected: 1,
201            found: 2,
202            backtrace: Backtrace::capture(),
203        };
204        let cleanup_error = StorageError::OtherIo {
205            path: "data/segment.parquet".to_string(),
206            source: io::Error::other("cleanup failed").into(),
207            backtrace: Backtrace::capture(),
208        };
209        let error = AppendSnafu.into_error(AppendError::Rollback {
210            source: Box::new(AppendError::Commit {
211                source: commit_error,
212            }),
213            cleanup_errors: vec![cleanup_error],
214        });
215
216        let rollback = error
217            .source()
218            .and_then(|source| source.downcast_ref::<AppendError>())
219            .expect("rollback source");
220        let cleanup_errors = match rollback {
221            AppendError::Rollback { cleanup_errors, .. } => cleanup_errors,
222            other => panic!("unexpected append error: {other:?}"),
223        };
224        let primary = rollback
225            .source()
226            .and_then(|source| source.downcast_ref::<Box<AppendError>>())
227            .map(Box::as_ref)
228            .expect("primary append source");
229        let commit = primary
230            .source()
231            .and_then(|source| source.downcast_ref::<CommitError>())
232            .expect("commit source");
233        let commit_backtrace = ErrorCompat::backtrace(commit).expect("commit backtrace");
234
235        assert!(matches!(
236            commit,
237            CommitError::Conflict {
238                expected: 1,
239                found: 2,
240                ..
241            }
242        ));
243        assert!(matches!(
244            cleanup_errors.as_slice(),
245            [StorageError::OtherIo { path, .. }] if path == "data/segment.parquet"
246        ));
247        let message = error.to_string();
248        assert!(message.contains("cleanup failed"));
249        assert!(!message.contains("Backtrace"));
250        assert!(std::ptr::eq(
251            ErrorCompat::backtrace(&error).expect("table backtrace"),
252            commit_backtrace
253        ));
254        assert!(std::ptr::eq(
255            ErrorCompat::backtrace(rollback).expect("rollback backtrace"),
256            commit_backtrace
257        ));
258        assert!(std::ptr::eq(
259            ErrorCompat::backtrace(primary).expect("primary backtrace"),
260            commit_backtrace
261        ));
262    }
263
264    #[test]
265    fn create_facade_preserves_index_source_and_operation_backtrace() {
266        let error = CreateSnafu.into_error(CreateTableError::IndexSpecValidation {
267            source: IndexSpecError::EmptyColumn,
268            backtrace: Backtrace::capture(),
269        });
270
271        let create = error
272            .source()
273            .and_then(|source| source.downcast_ref::<CreateTableError>())
274            .expect("create source");
275        let index = create
276            .source()
277            .and_then(|source| source.downcast_ref::<IndexSpecError>())
278            .expect("index specification source");
279
280        assert!(matches!(index, IndexSpecError::EmptyColumn));
281        assert!(std::ptr::eq(
282            ErrorCompat::backtrace(&error).expect("table backtrace"),
283            ErrorCompat::backtrace(create).expect("create backtrace")
284        ));
285    }
286
287    #[test]
288    fn open_facade_preserves_commit_source_and_backtrace() {
289        let error = OpenSnafu.into_error(OpenTableError::Commit {
290            source: CommitError::Conflict {
291                expected: 8,
292                found: 9,
293                backtrace: Backtrace::capture(),
294            },
295        });
296
297        let open = error
298            .source()
299            .and_then(|source| source.downcast_ref::<OpenTableError>())
300            .expect("open source");
301        let commit = open
302            .source()
303            .and_then(|source| source.downcast_ref::<CommitError>())
304            .expect("commit source");
305        let commit_backtrace = ErrorCompat::backtrace(commit).expect("commit backtrace");
306
307        assert!(matches!(commit, CommitError::Conflict { .. }));
308        assert!(std::ptr::eq(
309            ErrorCompat::backtrace(open).expect("open backtrace"),
310            commit_backtrace
311        ));
312        assert!(std::ptr::eq(
313            ErrorCompat::backtrace(&error).expect("table backtrace"),
314            commit_backtrace
315        ));
316    }
317
318    #[test]
319    fn state_access_facade_preserves_commit_source_and_backtrace() {
320        let error = StateAccessSnafu.into_error(TableStateAccessError::Commit {
321            source: CommitError::MissingTableMetadata {
322                current_version: 7,
323                backtrace: Backtrace::capture(),
324            },
325        });
326
327        let state_access = error
328            .source()
329            .and_then(|source| source.downcast_ref::<TableStateAccessError>())
330            .expect("state access source");
331        let commit = state_access
332            .source()
333            .and_then(|source| source.downcast_ref::<CommitError>())
334            .expect("commit source");
335        let commit_backtrace = ErrorCompat::backtrace(commit).expect("commit backtrace");
336
337        assert!(matches!(commit, CommitError::MissingTableMetadata { .. }));
338        assert!(std::ptr::eq(
339            ErrorCompat::backtrace(state_access).expect("state access backtrace"),
340            commit_backtrace
341        ));
342        assert!(std::ptr::eq(
343            ErrorCompat::backtrace(&error).expect("table backtrace"),
344            commit_backtrace
345        ));
346    }
347
348    #[test]
349    fn scan_facade_preserves_parquet_source_and_backtrace() {
350        let error = ScanSnafu.into_error(ScanError::Parquet {
351            path: "data/segment.parquet".to_string(),
352            operation: "reading metadata",
353            source: Box::new(ParquetError::General("corrupt footer".to_string())),
354            backtrace: Box::new(Backtrace::capture()),
355        });
356
357        let scan = error
358            .source()
359            .and_then(|source| source.downcast_ref::<ScanError>())
360            .expect("scan source");
361        let parquet = scan
362            .source()
363            .and_then(|source| source.downcast_ref::<Box<ParquetError>>())
364            .map(Box::as_ref)
365            .expect("Parquet source");
366        let scan_backtrace = ErrorCompat::backtrace(scan).expect("scan backtrace");
367
368        assert!(matches!(parquet, ParquetError::General(message) if message == "corrupt footer"));
369        assert!(std::ptr::eq(
370            ErrorCompat::backtrace(&error).expect("table backtrace"),
371            scan_backtrace
372        ));
373    }
374
375    #[test]
376    fn coverage_facade_preserves_codec_source_and_backtrace() {
377        let error = CoverageQuerySnafu.into_error(CoverageQueryError::CoverageSnapshotRead {
378            coverage_path: "coverage/table.coverage".to_string(),
379            source: Box::new(CoverageSidecarError::Codec {
380                source: CoverageCodecError::InvalidEntityCoverageMagic {
381                    backtrace: Backtrace::capture(),
382                },
383            }),
384        });
385
386        let coverage_query = error
387            .source()
388            .and_then(|source| source.downcast_ref::<CoverageQueryError>())
389            .expect("coverage query source");
390        let sidecar = coverage_query
391            .source()
392            .and_then(|source| source.downcast_ref::<Box<CoverageSidecarError>>())
393            .map(Box::as_ref)
394            .expect("coverage sidecar source");
395        let codec = sidecar
396            .source()
397            .and_then(|source| source.downcast_ref::<CoverageCodecError>())
398            .expect("coverage codec source");
399        let codec_backtrace = ErrorCompat::backtrace(codec).expect("codec backtrace");
400
401        assert!(matches!(
402            codec,
403            CoverageCodecError::InvalidEntityCoverageMagic { .. }
404        ));
405        assert!(std::ptr::eq(
406            ErrorCompat::backtrace(coverage_query).expect("coverage query backtrace"),
407            codec_backtrace
408        ));
409        assert!(std::ptr::eq(
410            ErrorCompat::backtrace(&error).expect("table backtrace"),
411            codec_backtrace
412        ));
413    }
414
415    #[test]
416    fn optimize_facade_preserves_commit_source_and_backtrace() {
417        let error = OptimizeSnafu.into_error(OptimizeError::Commit {
418            source: CommitError::Conflict {
419                expected: 3,
420                found: 4,
421                backtrace: Backtrace::capture(),
422            },
423        });
424
425        let optimize = error
426            .source()
427            .and_then(|source| source.downcast_ref::<OptimizeError>())
428            .expect("optimize source");
429        let commit = optimize
430            .source()
431            .and_then(|source| source.downcast_ref::<CommitError>())
432            .expect("commit source");
433        let commit_backtrace = ErrorCompat::backtrace(commit).expect("commit backtrace");
434
435        assert!(matches!(
436            commit,
437            CommitError::Conflict {
438                expected: 3,
439                found: 4,
440                ..
441            }
442        ));
443        assert!(std::ptr::eq(
444            ErrorCompat::backtrace(optimize).expect("optimize backtrace"),
445            commit_backtrace
446        ));
447        assert!(std::ptr::eq(
448            ErrorCompat::backtrace(&error).expect("table backtrace"),
449            commit_backtrace
450        ));
451    }
452
453    #[test]
454    fn optimize_facade_preserves_rewrite_storage_source_and_backtrace() {
455        let error = OptimizeSnafu.into_error(OptimizeError::from(EntityRewriteError::Storage {
456            source: StorageError::OtherIo {
457                path: "data/mixed.parquet".to_string(),
458                source: io::Error::other("read failed").into(),
459                backtrace: Backtrace::capture(),
460            },
461        }));
462
463        let optimize = error
464            .source()
465            .and_then(|source| source.downcast_ref::<OptimizeError>())
466            .expect("optimize source");
467        let rewrite = optimize
468            .source()
469            .and_then(|source| source.downcast_ref::<Box<EntityRewriteError>>())
470            .map(Box::as_ref)
471            .expect("rewrite source");
472        let storage = rewrite
473            .source()
474            .and_then(|source| source.downcast_ref::<StorageError>())
475            .expect("storage source");
476        let storage_backtrace = ErrorCompat::backtrace(storage).expect("storage backtrace");
477
478        assert!(matches!(
479            storage,
480            StorageError::OtherIo { path, .. } if path == "data/mixed.parquet"
481        ));
482        assert!(std::ptr::eq(
483            ErrorCompat::backtrace(rewrite).expect("rewrite backtrace"),
484            storage_backtrace
485        ));
486        assert!(std::ptr::eq(
487            ErrorCompat::backtrace(optimize).expect("optimize backtrace"),
488            storage_backtrace
489        ));
490        assert!(std::ptr::eq(
491            ErrorCompat::backtrace(&error).expect("table backtrace"),
492            storage_backtrace
493        ));
494    }
495
496    #[test]
497    fn optimize_rollback_preserves_primary_chain_and_typed_cleanup_errors() {
498        let error = OptimizeSnafu.into_error(OptimizeError::Rollback {
499            source: Box::new(OptimizeError::Commit {
500                source: CommitError::Conflict {
501                    expected: 5,
502                    found: 6,
503                    backtrace: Backtrace::capture(),
504                },
505            }),
506            cleanup_errors: vec![StorageError::OtherIo {
507                path: "data/_staged/segment.parquet".to_string(),
508                source: io::Error::other("cleanup failed").into(),
509                backtrace: Backtrace::capture(),
510            }],
511        });
512
513        let rollback = error
514            .source()
515            .and_then(|source| source.downcast_ref::<OptimizeError>())
516            .expect("optimize source");
517        let cleanup_errors = match rollback {
518            OptimizeError::Rollback { cleanup_errors, .. } => cleanup_errors,
519            other => panic!("unexpected optimize error: {other:?}"),
520        };
521        let primary = rollback
522            .source()
523            .and_then(|source| source.downcast_ref::<Box<OptimizeError>>())
524            .map(Box::as_ref)
525            .expect("primary optimize source");
526        let commit = primary
527            .source()
528            .and_then(|source| source.downcast_ref::<CommitError>())
529            .expect("commit source");
530        let commit_backtrace = ErrorCompat::backtrace(commit).expect("commit backtrace");
531
532        assert!(matches!(
533            cleanup_errors.as_slice(),
534            [StorageError::OtherIo { path, .. }] if path == "data/_staged/segment.parquet"
535        ));
536        assert!(std::ptr::eq(
537            ErrorCompat::backtrace(rollback).expect("rollback backtrace"),
538            commit_backtrace
539        ));
540        assert!(std::ptr::eq(
541            ErrorCompat::backtrace(&error).expect("table backtrace"),
542            commit_backtrace
543        ));
544    }
545}
timeseries_table_format/table/error.rs

timeseries_table_format/table/
error.rs
