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 parquet::errors::ParquetError;
9use serde::{Deserialize, Serialize};
10use snafu::{Backtrace, prelude::*};
11
12use crate::{
13    coverage::EntityIdentity,
14    metadata::{
15        index::{IndexKind, IndexValue, IndexValueError},
16        logical_schema::{ArrowToLogicalSchemaError, LogicalSchemaValidationError},
17    },
18};
19
20/// Supported on-disk file formats for segments.
21///
22/// In v0.1, only `Parquet` is implemented, but the enum keeps the metadata model
23/// open to other formats in future versions.
24///
25/// JSON layout example: `"format": "parquet"`
26#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
27#[serde(rename_all = "lowercase")]
28pub enum FileFormat {
29    /// Apache Parquet columnar format.
30    #[default]
31    Parquet,
32    // Future:
33    // Orc,
34    // Avro,
35    // Csv,
36}
37
38/// Entity distribution within one physical segment.
39#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
40pub enum SegmentEntityLayout {
41    /// The table has no configured entity columns.
42    NotApplicable,
43    /// Every row belongs to one complete entity identity.
44    Single(EntityIdentity),
45    /// Rows belong to more than one complete entity identity.
46    Mixed,
47}
48
49/// Metadata about a single physical segment.
50///
51/// In v0.1, a "segment" corresponds to one stored data object.
52#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
53pub struct SegmentMeta {
54    /// Canonical file path relative to the table root and the segment identity.
55    pub path: String,
56
57    /// File format for this segment.
58    pub format: FileFormat,
59
60    /// Exact entity distribution derived before the segment is committed.
61    pub entity_layout: SegmentEntityLayout,
62
63    /// Minimum observed ordered-index value in this segment (inclusive).
64    pub index_min: IndexValue,
65
66    /// Maximum observed ordered-index value in this segment (inclusive).
67    pub index_max: IndexValue,
68
69    /// Number of rows in this segment.
70    pub row_count: u64,
71
72    /// Optional file size in bytes at the time metadata was captured.
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub file_size: Option<u64>,
75
76    /// Coverage sidecar pointer.
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub coverage_path: Option<String>,
79}
80
81/// Exact Parquet ordered-index column validation failure.
82#[derive(Debug, Snafu, Clone, PartialEq, Eq)]
83#[snafu(display(
84    "Invalid ordered-index column {column} in segment at {path}: expected {expected_domain}, observed {observed_type}"
85))]
86pub struct ParquetIndexColumnError {
87    /// Normalized path to the segment file.
88    pub path: String,
89    /// Registered top-level ordered-index column.
90    pub column: String,
91    /// Registered ordered-index domain.
92    pub expected_domain: &'static str,
93    /// Observed Parquet column shape and annotations.
94    pub observed_type: String,
95}
96
97impl SegmentMeta {
98    /// Set the coverage sidecar path for this segment metadata.
99    pub fn with_coverage_path(mut self, path: impl Into<String>) -> Self {
100        self.coverage_path = Some(path.into());
101        self
102    }
103
104    /// Validate the segment's inclusive ordered-index bounds.
105    ///
106    /// # Errors
107    /// Returns [`SegmentMetaError::InvalidIndexBounds`] when either bound has
108    /// the wrong domain or the minimum is greater than the maximum.
109    pub fn validate_bounds(&self, kind: &IndexKind) -> Result<(), SegmentMetaError> {
110        self.index_min.validate_kind(kind).map_err(|source| {
111            SegmentMetaError::InvalidIndexBounds {
112                path: self.path.clone(),
113                source,
114            }
115        })?;
116        self.index_max.validate_kind(kind).map_err(|source| {
117            SegmentMetaError::InvalidIndexBounds {
118                path: self.path.clone(),
119                source,
120            }
121        })?;
122        if self
123            .index_min
124            .compare(&self.index_max)
125            .map_err(|source| SegmentMetaError::InvalidIndexBounds {
126                path: self.path.clone(),
127                source,
128            })?
129            .is_gt()
130        {
131            return Err(SegmentMetaError::InvalidIndexBounds {
132                path: self.path.clone(),
133                source: IndexValueError::InvalidBounds {
134                    min: self.index_min.clone(),
135                    max: self.index_max.clone(),
136                },
137            });
138        }
139        Ok(())
140    }
141}
142
143/// Errors that can occur while validating or decoding segment metadata.
144///
145/// This enum intentionally contains **no storage backend errors**. IO-related
146/// errors should be wrapped at the IO boundary (for example, in
147/// [`crate::transaction_log::SegmentError`]).
148#[derive(Debug, Snafu)]
149#[non_exhaustive]
150pub enum SegmentMetaError {
151    /// Persisted segment bounds violate the table's ordered-index domain.
152    #[snafu(display("Invalid ordered-index bounds in segment at {path}: {source}"))]
153    InvalidIndexBounds {
154        /// Segment path containing invalid bounds.
155        path: String,
156        /// Domain or ordering failure.
157        source: IndexValueError,
158    },
159
160    /// The file is too short to be a valid Parquet file.
161    #[snafu(display("Segment file too short to be valid Parquet: {path}"))]
162    TooShort {
163        /// The path to the file that was too short.
164        path: String,
165    },
166
167    /// Parquet reader / metadata failure.
168    #[snafu(display("Error reading Parquet metadata for segment at {path}: {source}"))]
169    ParquetRead {
170        /// The path to the file that caused the Parquet read failure.
171        path: String,
172        /// Underlying parquet error that caused this failure.
173        source: ParquetError,
174        /// Diagnostic backtrace for this error.
175        backtrace: Backtrace,
176    },
177
178    /// A parallel row-group inspection task failed before returning its typed result.
179    #[snafu(display("Row-group inspection task failed for segment at {path}: {source}"))]
180    RowGroupTask {
181        /// Segment path being inspected.
182        path: String,
183        /// Tokio task failure, including panic or cancellation details.
184        source: tokio::task::JoinError,
185        /// Backtrace captured while joining the row-group task.
186        backtrace: Backtrace,
187    },
188
189    /// The registered ordered-index column is missing or incompatible.
190    #[snafu(transparent)]
191    OrderedIndexColumn {
192        /// Exact registered and observed Parquet column details.
193        source: ParquetIndexColumnError,
194    },
195
196    /// Statistics exist but are not well-shaped (wrong length / unexpected type).
197    #[snafu(display(
198        "Parquet statistics shape invalid for {column} in segment at {path}: {detail}"
199    ))]
200    ParquetStatsShape {
201        /// The path to the file with malformed Parquet statistics.
202        path: String,
203        /// The column whose statistics are malformed.
204        column: String,
205        /// Details about how the statistics are malformed.
206        detail: String,
207    },
208
209    /// The file contains no non-null value for the registered ordered index.
210    #[snafu(display(
211        "No observed {expected_domain} value for ordered-index column {column} in segment at {path}"
212    ))]
213    NoObservedIndexValue {
214        /// Path to the segment file.
215        path: String,
216        /// Registered ordered-index column.
217        column: String,
218        /// Registered ordered-index domain.
219        expected_domain: &'static str,
220    },
221
222    /// Failed to derive a valid LogicalSchema from the Parquet file.
223    #[snafu(display("Invalid logical schema derived from Parquet at {path}: {source}"))]
224    LogicalSchemaInvalid {
225        /// The path to the file without a valid LogicalSchema.
226        path: String,
227        /// Underlying logical schema error that triggered this failure.
228        #[snafu(source(from(LogicalSchemaValidationError, Box::new)))]
229        source: Box<LogicalSchemaValidationError>,
230        /// Backtrace captured with segment path context.
231        backtrace: Backtrace,
232    },
233
234    /// An embedded Arrow schema cannot be represented by the logical schema model.
235    #[snafu(display("Invalid Arrow schema derived from Parquet at {path}: {source}"))]
236    ArrowToLogicalSchema {
237        /// Segment path containing the embedded Arrow schema.
238        path: String,
239        /// Exact Arrow-to-logical conversion failure.
240        #[snafu(source(from(ArrowToLogicalSchemaError, Box::new)))]
241        source: Box<ArrowToLogicalSchemaError>,
242        /// Backtrace captured with segment path context.
243        backtrace: Backtrace,
244    },
245}
246
247/// Deterministic ordering for segments by ordered-index bounds.
248///
249/// Ordering is by `index_min`, then `index_max`, and finally `path` as a stable
250/// tie-breaker.
251pub(crate) fn cmp_segment_meta_by_index(
252    a: &SegmentMeta,
253    b: &SegmentMeta,
254) -> Result<std::cmp::Ordering, IndexValueError> {
255    let min_order = a.index_min.compare(&b.index_min)?;
256    if !min_order.is_eq() {
257        return Ok(min_order);
258    }
259    let max_order = a.index_max.compare(&b.index_max)?;
260    Ok(max_order.then_with(|| a.path.cmp(&b.path)))
261}
262
263/// Sort segment metadata by typed bounds, rejecting cross-domain values first.
264pub(crate) fn sort_segment_meta_by_index<T>(segments: &mut [T]) -> Result<(), IndexValueError>
265where
266    T: std::borrow::Borrow<SegmentMeta>,
267{
268    let mut domain: Option<&IndexValue> = None;
269    for segment in segments.iter().map(std::borrow::Borrow::borrow) {
270        if segment.index_min.compare(&segment.index_max)?.is_gt() {
271            return Err(IndexValueError::InvalidBounds {
272                min: segment.index_min.clone(),
273                max: segment.index_max.clone(),
274            });
275        }
276        if let Some(domain) = domain {
277            domain.compare(&segment.index_min)?;
278            domain.compare(&segment.index_max)?;
279        } else {
280            domain = Some(&segment.index_min);
281        }
282    }
283
284    let mut sort_error = None;
285    segments.sort_unstable_by(
286        |a, b| match cmp_segment_meta_by_index(a.borrow(), b.borrow()) {
287            Ok(order) => order,
288            Err(error) => {
289                sort_error.get_or_insert(error);
290                std::cmp::Ordering::Equal
291            }
292        },
293    );
294    if let Some(error) = sort_error {
295        return Err(error);
296    }
297    Ok(())
298}
299
300#[cfg(test)]
301mod tests {
302    use std::num::NonZeroU64;
303
304    use super::*;
305    use chrono::{TimeZone, Utc};
306
307    fn seg(id: &str, ts_min: i64, ts_max: i64) -> SegmentMeta {
308        SegmentMeta {
309            path: format!("data/{id}.parquet"),
310            format: FileFormat::Parquet,
311            entity_layout: SegmentEntityLayout::NotApplicable,
312            index_min: IndexValue::Timestamp(Utc.timestamp_opt(ts_min, 0).single().unwrap()),
313            index_max: IndexValue::Timestamp(Utc.timestamp_opt(ts_max, 0).single().unwrap()),
314            row_count: 1,
315            file_size: None,
316            coverage_path: None,
317        }
318    }
319
320    #[test]
321    fn entity_layout_json_roundtrips_are_stable() {
322        let cases = [
323            (SegmentEntityLayout::NotApplicable, "\"NotApplicable\""),
324            (
325                SegmentEntityLayout::Single(
326                    EntityIdentity::try_new(vec!["us".into(), "device-1".into()]).unwrap(),
327                ),
328                r#"{"Single":[{"type":"utf8","value":"us"},{"type":"utf8","value":"device-1"}]}"#,
329            ),
330            (SegmentEntityLayout::Mixed, "\"Mixed\""),
331        ];
332
333        for (layout, expected_json) in cases {
334            let json = serde_json::to_string(&layout).unwrap();
335            assert_eq!(json, expected_json);
336            assert_eq!(
337                serde_json::from_str::<SegmentEntityLayout>(&json).unwrap(),
338                layout
339            );
340        }
341    }
342
343    #[test]
344    fn single_layout_rejects_an_empty_identity() {
345        let error = serde_json::from_str::<SegmentEntityLayout>(r#"{"Single":[]}"#)
346            .expect_err("empty identity must be rejected");
347        assert!(error.to_string().contains("at least one component"));
348    }
349
350    #[test]
351    fn ordering_is_deterministic_with_tie_breakers() {
352        let mut v = vec![
353            seg("c", 10, 20),
354            seg("b", 10, 20),
355            seg("a", 10, 30),
356            seg("d", 5, 7),
357        ];
358
359        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
360
361        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
362        assert_eq!(
363            paths,
364            vec![
365                "data/d.parquet",
366                "data/b.parquet",
367                "data/c.parquet",
368                "data/a.parquet"
369            ]
370        );
371    }
372
373    #[test]
374    fn ordering_is_equal_for_identical_segments() {
375        let a = seg("same", 10, 20);
376        let b = seg("same", 10, 20);
377        assert_eq!(
378            cmp_segment_meta_by_index(&a, &b).unwrap(),
379            std::cmp::Ordering::Equal
380        );
381        assert_eq!(
382            cmp_segment_meta_by_index(&b, &a).unwrap(),
383            std::cmp::Ordering::Equal
384        );
385    }
386
387    #[test]
388    fn ordering_primary_key_ts_min_dominates() {
389        let mut v = vec![seg("z", 20, 30), seg("a", 10, 50), seg("m", 15, 10)];
390
391        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
392
393        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
394        assert_eq!(
395            paths,
396            vec!["data/a.parquet", "data/m.parquet", "data/z.parquet"]
397        );
398    }
399
400    #[test]
401    fn ordering_uses_path_as_final_tie_breaker() {
402        let mut v = vec![seg("b", 10, 20), seg("a", 10, 20), seg("c", 10, 20)];
403
404        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
405
406        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
407        assert_eq!(
408            paths,
409            vec!["data/a.parquet", "data/b.parquet", "data/c.parquet"]
410        );
411    }
412
413    #[test]
414    fn segment_bounds_validate_domain_and_native_order() {
415        let signed = IndexKind::Int64 {
416            index_granularity: NonZeroU64::new(1).unwrap(),
417        };
418        let valid = SegmentMeta {
419            path: "data/valid.parquet".to_string(),
420            format: FileFormat::Parquet,
421            entity_layout: SegmentEntityLayout::NotApplicable,
422            index_min: IndexValue::Int64(i64::MIN),
423            index_max: IndexValue::Int64(i64::MAX),
424            row_count: 1,
425            file_size: None,
426            coverage_path: None,
427        };
428        valid.validate_bounds(&signed).unwrap();
429
430        let reversed = SegmentMeta {
431            index_min: IndexValue::Int64(1),
432            index_max: IndexValue::Int64(0),
433            ..valid.clone()
434        };
435        assert!(matches!(
436            reversed.validate_bounds(&signed),
437            Err(SegmentMetaError::InvalidIndexBounds {
438                source: IndexValueError::InvalidBounds { .. },
439                ..
440            })
441        ));
442
443        let wrong_domain = SegmentMeta {
444            index_min: IndexValue::UInt64(0),
445            index_max: IndexValue::UInt64(u64::MAX),
446            ..valid
447        };
448        assert!(matches!(
449            wrong_domain.validate_bounds(&signed),
450            Err(SegmentMetaError::InvalidIndexBounds {
451                source: IndexValueError::KindMismatch { .. },
452                ..
453            })
454        ));
455    }
456
457    #[test]
458    fn integer_segment_ordering_uses_native_bounds_then_path() {
459        let mut segments = vec![
460            SegmentMeta {
461                path: "data/z.parquet".to_string(),
462                format: FileFormat::Parquet,
463                entity_layout: SegmentEntityLayout::NotApplicable,
464                index_min: IndexValue::UInt64(u64::MAX),
465                index_max: IndexValue::UInt64(u64::MAX),
466                row_count: 1,
467                file_size: None,
468                coverage_path: None,
469            },
470            SegmentMeta {
471                path: "data/b.parquet".to_string(),
472                format: FileFormat::Parquet,
473                entity_layout: SegmentEntityLayout::NotApplicable,
474                index_min: IndexValue::UInt64(0),
475                index_max: IndexValue::UInt64(7),
476                row_count: 1,
477                file_size: None,
478                coverage_path: None,
479            },
480            SegmentMeta {
481                path: "data/a.parquet".to_string(),
482                format: FileFormat::Parquet,
483                entity_layout: SegmentEntityLayout::NotApplicable,
484                index_min: IndexValue::UInt64(0),
485                index_max: IndexValue::UInt64(7),
486                row_count: 1,
487                file_size: None,
488                coverage_path: None,
489            },
490        ];
491        segments.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).unwrap());
492        assert_eq!(
493            segments
494                .into_iter()
495                .map(|segment| segment.path)
496                .collect::<Vec<_>>(),
497            vec!["data/a.parquet", "data/b.parquet", "data/z.parquet"]
498        );
499    }
500
501    #[test]
502    fn sorting_rejects_cross_domain_segments() {
503        let mut segments = vec![
504            seg("timestamp", 0, 1),
505            SegmentMeta {
506                path: "data/integer.parquet".to_string(),
507                format: FileFormat::Parquet,
508                entity_layout: SegmentEntityLayout::NotApplicable,
509                index_min: IndexValue::Int64(0),
510                index_max: IndexValue::Int64(1),
511                row_count: 1,
512                file_size: None,
513                coverage_path: None,
514            },
515        ];
516
517        assert!(matches!(
518            sort_segment_meta_by_index(&mut segments),
519            Err(IndexValueError::DomainMismatch { .. })
520        ));
521    }
522
523    #[test]
524    fn segment_json_preserves_integer_bound_extremes() {
525        for (minimum, maximum) in [
526            (IndexValue::Int64(i64::MIN), IndexValue::Int64(i64::MAX)),
527            (IndexValue::UInt64(0), IndexValue::UInt64(u64::MAX)),
528        ] {
529            let segment = SegmentMeta {
530                path: "data/extremes.parquet".to_string(),
531                format: FileFormat::Parquet,
532                entity_layout: SegmentEntityLayout::NotApplicable,
533                index_min: minimum,
534                index_max: maximum,
535                row_count: 2,
536                file_size: Some(42),
537                coverage_path: Some("_coverage/segments/extremes.roar".to_string()),
538            };
539            let json = serde_json::to_string(&segment).unwrap();
540            assert_eq!(serde_json::from_str::<SegmentMeta>(&json).unwrap(), segment);
541        }
542    }
543}
timeseries_table_format/metadata/segments.rs

timeseries_table_format/metadata/
segments.rs
