timeseries_table_format/transaction_log/
segments.rs

1//! Segment access error types.
2//!
3//! The canonical segment metadata model lives in [`crate::metadata::segments`]
4//! and contains **no storage IO**.
5//!
6//! This module maps storage failures into the segment errors returned by
7//! format-specific readers.
8
9use snafu::prelude::*;
10
11use crate::storage::StorageError;
12
13// Expose the pure segment types alongside their access errors.
14pub use crate::metadata::segments::{
15    FileFormat, SegmentEntityLayout, SegmentMeta, SegmentMetaError,
16};
17
18/// Errors while reading or validating one segment.
19#[derive(Debug, Snafu)]
20#[non_exhaustive]
21pub enum SegmentError {
22    /// The segment file is missing.
23    #[snafu(display("Segment file not found: {path}"))]
24    MissingFile {
25        /// Path to the missing file.
26        path: String,
27        /// Storage failure that identified the missing file.
28        #[snafu(source, backtrace)]
29        source: StorageError,
30    },
31
32    /// Another storage failure occurred while accessing the segment.
33    #[snafu(display("Storage error while accessing segment at {path}: {source}"))]
34    Storage {
35        /// Path to the segment that could not be accessed.
36        path: String,
37        /// Underlying storage failure.
38        #[snafu(source, backtrace)]
39        source: StorageError,
40    },
41
42    /// Pure metadata/decoding/validation error.
43    #[snafu(context(false), display("{source}"))]
44    Metadata {
45        /// The underlying pure metadata error.
46        #[snafu(source, backtrace)]
47        source: SegmentMetaError,
48    },
49}
50
51/// Convenience alias for results returned by segment access operations.
52#[allow(clippy::result_large_err)]
53pub type SegmentResult<T> = Result<T, SegmentError>;
54
55impl From<StorageError> for SegmentError {
56    fn from(source: StorageError) -> Self {
57        let is_missing = matches!(&source, StorageError::NotFound { .. });
58        let path = match &source {
59            StorageError::InvalidRelativePath { path, .. }
60            | StorageError::NotFound { path, .. }
61            | StorageError::AlreadyExists { path, .. }
62            | StorageError::OtherIo { path, .. }
63            | StorageError::CleanupFailed { path, .. } => path.clone(),
64        };
65
66        if is_missing {
67            Self::MissingFile { path, source }
68        } else {
69            Self::Storage { path, source }
70        }
71    }
72}
73
74#[cfg(test)]
75mod tests {
76    use std::{error::Error as _, io};
77
78    use super::*;
79    use chrono::Utc;
80    use chrono::{DateTime, TimeZone};
81    use snafu::{Backtrace, ErrorCompat};
82
83    use crate::storage::StorageBackendError;
84
85    fn utc_datetime(
86        year: i32,
87        month: u32,
88        day: u32,
89        hour: u32,
90        minute: u32,
91        second: u32,
92    ) -> DateTime<Utc> {
93        Utc.with_ymd_and_hms(year, month, day, hour, minute, second)
94            .single()
95            .expect("valid UTC timestamp")
96    }
97
98    fn sample_segment_meta() -> SegmentMeta {
99        SegmentMeta {
100            path: "data/seg-001.parquet".to_string(),
101            format: FileFormat::Parquet,
102            entity_layout: SegmentEntityLayout::NotApplicable,
103            index_min: (utc_datetime(2025, 1, 1, 0, 0, 0)).into(),
104            index_max: (utc_datetime(2025, 1, 1, 1, 0, 0)).into(),
105            row_count: 123,
106            file_size: None,
107            coverage_path: None,
108        }
109    }
110
111    #[test]
112    fn segment_meta_json_roundtrip_with_and_without_coverage_path() {
113        // Without coverage_path
114        let seg = sample_segment_meta();
115        let json = serde_json::to_string(&seg).unwrap();
116        let back: SegmentMeta = serde_json::from_str(&json).unwrap();
117        assert_eq!(back.coverage_path, None);
118        assert_eq!(back.file_size, None);
119
120        // With coverage_path
121        let mut seg2 = sample_segment_meta().with_coverage_path("_coverage/segments/a.roar");
122        seg2.file_size = Some(42);
123        let json2 = serde_json::to_string(&seg2).unwrap();
124        let back2: SegmentMeta = serde_json::from_str(&json2).unwrap();
125        assert_eq!(
126            back2.coverage_path.as_deref(),
127            Some("_coverage/segments/a.roar")
128        );
129        assert_eq!(back2.file_size, Some(42));
130    }
131
132    #[test]
133    fn segment_meta_json_requires_entity_layout() {
134        let mut value = serde_json::to_value(sample_segment_meta()).unwrap();
135        value
136            .as_object_mut()
137            .expect("segment metadata is an object")
138            .remove("entity_layout");
139
140        let error = serde_json::from_value::<SegmentMeta>(value)
141            .expect_err("segment metadata must include entity_layout");
142        assert!(error.to_string().contains("entity_layout"));
143    }
144
145    #[test]
146    fn missing_segment_preserves_storage_source_and_backtrace() {
147        let storage = StorageError::NotFound {
148            path: "data/missing.parquet".to_string(),
149            source: io::Error::new(io::ErrorKind::NotFound, "missing").into(),
150            backtrace: Backtrace::capture(),
151        };
152        let error = SegmentError::from(storage);
153
154        let segment_backtrace = ErrorCompat::backtrace(&error).expect("segment backtrace");
155        let storage = error
156            .source()
157            .and_then(|source| source.downcast_ref::<StorageError>())
158            .expect("storage source");
159        let storage_backtrace = ErrorCompat::backtrace(storage).expect("storage backtrace");
160        let backend = storage
161            .source()
162            .and_then(|source| source.downcast_ref::<StorageBackendError>())
163            .expect("storage backend source");
164        let io_source = backend
165            .source()
166            .and_then(|source| source.downcast_ref::<io::Error>())
167            .expect("io source");
168
169        assert!(matches!(&error, SegmentError::MissingFile { .. }));
170        assert_eq!(io_source.kind(), io::ErrorKind::NotFound);
171        assert!(std::ptr::eq(segment_backtrace, storage_backtrace));
172    }
173
174    #[test]
175    fn non_missing_storage_failure_remains_a_storage_error() {
176        let storage = StorageError::OtherIo {
177            path: "data/unreadable.parquet".to_string(),
178            source: io::Error::from(io::ErrorKind::PermissionDenied).into(),
179            backtrace: Backtrace::capture(),
180        };
181
182        assert!(matches!(
183            SegmentError::from(storage),
184            SegmentError::Storage {
185                source: StorageError::OtherIo { .. },
186                ..
187            }
188        ));
189    }
190}
timeseries_table_format/transaction_log/segments.rs

timeseries_table_format/transaction_log/
segments.rs
