timeseries_table_format/metadata/
segments.rs

1//! Segment formats and per-file metadata recorded in table metadata.
2//!
3//! This module contains **pure** data types + non-IO validation/decoding errors.
4//! Any functions that touch storage backends (filesystem, object store, etc.)
5//! must live outside `metadata/` (for example under `transaction_log` or
6//! format-specific helpers).
7
8use arrow::error::ArrowError;
9use chrono::{DateTime, Utc};
10use parquet::errors::ParquetError;
11use serde::{Deserialize, Serialize};
12use snafu::{Backtrace, prelude::*};
13
14use crate::metadata::{logical_schema::LogicalSchemaError, time_column::TimeColumnError};
15
16/// Supported on-disk file formats for segments.
17///
18/// In v0.1, only `Parquet` is implemented, but the enum keeps the metadata model
19/// open to other formats in future versions.
20///
21/// JSON layout example: `"format": "parquet"`
22#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
23#[serde(rename_all = "lowercase")]
24pub enum FileFormat {
25    /// Apache Parquet columnar format.
26    #[default]
27    Parquet,
28    // Future:
29    // Orc,
30    // Avro,
31    // Csv,
32}
33
34/// Metadata about a single physical segment.
35///
36/// In v0.1, a "segment" corresponds to a single data file on disk.
37#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
38pub struct SegmentMeta {
39    /// Canonical file path relative to the table root and the segment identity.
40    pub path: String,
41
42    /// File format for this segment.
43    pub format: FileFormat,
44
45    /// Minimum timestamp contained in this segment (inclusive), in RFC3339 UTC.
46    pub ts_min: DateTime<Utc>,
47
48    /// Maximum timestamp contained in this segment (inclusive), in RFC3339 UTC.
49    pub ts_max: DateTime<Utc>,
50
51    /// Number of rows in this segment.
52    pub row_count: u64,
53
54    /// Optional file size in bytes at the time metadata was captured.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub file_size: Option<u64>,
57
58    /// Coverage sidecar pointer.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub coverage_path: Option<String>,
61}
62
63impl SegmentMeta {
64    /// Set the coverage sidecar path for this segment metadata.
65    pub fn with_coverage_path(mut self, path: impl Into<String>) -> Self {
66        self.coverage_path = Some(path.into());
67        self
68    }
69}
70
71/// Errors that can occur while validating or decoding segment metadata.
72///
73/// This enum intentionally contains **no storage backend errors**. IO-related
74/// errors should be wrapped at the IO boundary (for example, in
75/// `transaction_log::segments::SegmentError`).
76#[derive(Debug, Snafu)]
77pub enum SegmentMetaError {
78    /// The file is too short to be a valid Parquet file.
79    #[snafu(display("Segment file too short to be valid Parquet: {path}"))]
80    TooShort {
81        /// The path to the file that was too short.
82        path: String,
83    },
84
85    /// Parquet reader / metadata failure.
86    #[snafu(display("Error reading Parquet metadata for segment at {path}: {source}"))]
87    ParquetRead {
88        /// The path to the file that caused the Parquet read failure.
89        path: String,
90        /// Underlying parquet error that caused this failure.
91        source: ParquetError,
92        /// Diagnostic backtrace for this error.
93        backtrace: Backtrace,
94    },
95
96    /// Arrow decode failure while reading Parquet data.
97    #[snafu(display("Arrow read error for segment at {path}: {source}"))]
98    ArrowRead {
99        /// The path to the file that caused the Arrow read failure.
100        path: String,
101        /// Underlying Arrow error that caused this failure.
102        source: ArrowError,
103        /// Diagnostic backtrace for this error.
104        backtrace: Backtrace,
105    },
106
107    /// Time column validation or metadata error.
108    #[snafu(display("Time column error in segment at {path}: {source}"))]
109    TimeColumn {
110        /// The path to the segment file with a time column error.
111        path: String,
112        /// The underlying time column error.
113        source: TimeColumnError,
114    },
115
116    /// Statistics exist but are not well-shaped (wrong length / unexpected type).
117    #[snafu(display(
118        "Parquet statistics shape invalid for {column} in segment at {path}: {detail}"
119    ))]
120    ParquetStatsShape {
121        /// The path to the file with malformed Parquet statistics.
122        path: String,
123        /// The column whose statistics are malformed.
124        column: String,
125        /// Details about how the statistics are malformed.
126        detail: String,
127    },
128
129    /// No usable statistics for the time column; v0.1 may fall back to a scan.
130    #[snafu(display("Parquet statistics missing for {column} in segment at {path}"))]
131    ParquetStatsMissing {
132        /// The path to the file missing statistics for the column.
133        path: String,
134        /// The column missing statistics.
135        column: String,
136    },
137
138    /// Failed to derive a valid LogicalSchema from the Parquet file.
139    #[snafu(display("Invalid logical schema derived from Parquet at {path}: {source}"))]
140    LogicalSchemaInvalid {
141        /// The path to the file without a valid LogicalSchema.
142        path: String,
143        /// Underlying logical schema error that triggered this failure.
144        #[snafu(source)]
145        source: LogicalSchemaError,
146    },
147}
148
149/// Deterministic ordering for segments by time.
150///
151/// Ordering is by `ts_min`, then `ts_max`, and finally `path` as a stable
152/// tie-breaker.
153pub(crate) fn cmp_segment_meta_by_time(a: &SegmentMeta, b: &SegmentMeta) -> std::cmp::Ordering {
154    a.ts_min
155        .cmp(&b.ts_min)
156        .then_with(|| a.ts_max.cmp(&b.ts_max))
157        .then_with(|| a.path.cmp(&b.path))
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163    use chrono::{TimeZone, Utc};
164
165    fn seg(id: &str, ts_min: i64, ts_max: i64) -> SegmentMeta {
166        SegmentMeta {
167            path: format!("data/{id}.parquet"),
168            format: FileFormat::Parquet,
169            ts_min: Utc.timestamp_opt(ts_min, 0).single().unwrap(),
170            ts_max: Utc.timestamp_opt(ts_max, 0).single().unwrap(),
171            row_count: 1,
172            file_size: None,
173            coverage_path: None,
174        }
175    }
176
177    #[test]
178    fn ordering_is_deterministic_with_tie_breakers() {
179        let mut v = vec![
180            seg("c", 10, 20),
181            seg("b", 10, 20),
182            seg("a", 10, 30),
183            seg("d", 5, 7),
184        ];
185
186        v.sort_unstable_by(cmp_segment_meta_by_time);
187
188        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
189        assert_eq!(
190            paths,
191            vec![
192                "data/d.parquet",
193                "data/b.parquet",
194                "data/c.parquet",
195                "data/a.parquet"
196            ]
197        );
198    }
199
200    #[test]
201    fn ordering_is_equal_for_identical_segments() {
202        let a = seg("same", 10, 20);
203        let b = seg("same", 10, 20);
204        assert_eq!(cmp_segment_meta_by_time(&a, &b), std::cmp::Ordering::Equal);
205        assert_eq!(cmp_segment_meta_by_time(&b, &a), std::cmp::Ordering::Equal);
206    }
207
208    #[test]
209    fn ordering_primary_key_ts_min_dominates() {
210        let mut v = vec![seg("z", 20, 30), seg("a", 10, 50), seg("m", 15, 10)];
211
212        v.sort_unstable_by(cmp_segment_meta_by_time);
213
214        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
215        assert_eq!(
216            paths,
217            vec!["data/a.parquet", "data/m.parquet", "data/z.parquet"]
218        );
219    }
220
221    #[test]
222    fn ordering_uses_path_as_final_tie_breaker() {
223        let mut v = vec![seg("b", 10, 20), seg("a", 10, 20), seg("c", 10, 20)];
224
225        v.sort_unstable_by(cmp_segment_meta_by_time);
226
227        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
228        assert_eq!(
229            paths,
230            vec!["data/a.parquet", "data/b.parquet", "data/c.parquet"]
231        );
232    }
233}
timeseries_table_format/metadata/segments.rs

timeseries_table_format/metadata/
segments.rs
