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    /// A row group reports an invalid row count or physical byte size.
247    #[snafu(display(
248        "Invalid metadata for row group {row_group_index} in segment at {path}: {detail}"
249    ))]
250    InvalidRowGroupMetadata {
251        /// Path to the segment file.
252        path: String,
253        /// Zero-based row-group position in the Parquet footer.
254        row_group_index: usize,
255        /// Description of the invalid value.
256        detail: String,
257    },
258}
259
260/// Deterministic ordering for segments by ordered-index bounds.
261///
262/// Ordering is by `index_min`, then `index_max`, and finally `path` as a stable
263/// tie-breaker.
264pub(crate) fn cmp_segment_meta_by_index(
265    a: &SegmentMeta,
266    b: &SegmentMeta,
267) -> Result<std::cmp::Ordering, IndexValueError> {
268    let min_order = a.index_min.compare(&b.index_min)?;
269    if !min_order.is_eq() {
270        return Ok(min_order);
271    }
272    let max_order = a.index_max.compare(&b.index_max)?;
273    Ok(max_order.then_with(|| a.path.cmp(&b.path)))
274}
275
276/// Sort segment metadata by typed bounds, rejecting cross-domain values first.
277pub(crate) fn sort_segment_meta_by_index<T>(segments: &mut [T]) -> Result<(), IndexValueError>
278where
279    T: std::borrow::Borrow<SegmentMeta>,
280{
281    let mut domain: Option<&IndexValue> = None;
282    for segment in segments.iter().map(std::borrow::Borrow::borrow) {
283        if segment.index_min.compare(&segment.index_max)?.is_gt() {
284            return Err(IndexValueError::InvalidBounds {
285                min: segment.index_min.clone(),
286                max: segment.index_max.clone(),
287            });
288        }
289        if let Some(domain) = domain {
290            domain.compare(&segment.index_min)?;
291            domain.compare(&segment.index_max)?;
292        } else {
293            domain = Some(&segment.index_min);
294        }
295    }
296
297    let mut sort_error = None;
298    segments.sort_unstable_by(
299        |a, b| match cmp_segment_meta_by_index(a.borrow(), b.borrow()) {
300            Ok(order) => order,
301            Err(error) => {
302                sort_error.get_or_insert(error);
303                std::cmp::Ordering::Equal
304            }
305        },
306    );
307    if let Some(error) = sort_error {
308        return Err(error);
309    }
310    Ok(())
311}
312
313#[cfg(test)]
314mod tests {
315    use std::num::NonZeroU64;
316
317    use super::*;
318    use chrono::{TimeZone, Utc};
319
320    fn seg(id: &str, ts_min: i64, ts_max: i64) -> SegmentMeta {
321        SegmentMeta {
322            path: format!("data/{id}.parquet"),
323            format: FileFormat::Parquet,
324            entity_layout: SegmentEntityLayout::NotApplicable,
325            index_min: IndexValue::Timestamp(Utc.timestamp_opt(ts_min, 0).single().unwrap()),
326            index_max: IndexValue::Timestamp(Utc.timestamp_opt(ts_max, 0).single().unwrap()),
327            row_count: 1,
328            file_size: None,
329            coverage_path: None,
330        }
331    }
332
333    #[test]
334    fn entity_layout_json_roundtrips_are_stable() {
335        let cases = [
336            (SegmentEntityLayout::NotApplicable, "\"NotApplicable\""),
337            (
338                SegmentEntityLayout::Single(
339                    EntityIdentity::try_new(vec!["us".into(), "device-1".into()]).unwrap(),
340                ),
341                r#"{"Single":[{"type":"utf8","value":"us"},{"type":"utf8","value":"device-1"}]}"#,
342            ),
343            (SegmentEntityLayout::Mixed, "\"Mixed\""),
344        ];
345
346        for (layout, expected_json) in cases {
347            let json = serde_json::to_string(&layout).unwrap();
348            assert_eq!(json, expected_json);
349            assert_eq!(
350                serde_json::from_str::<SegmentEntityLayout>(&json).unwrap(),
351                layout
352            );
353        }
354    }
355
356    #[test]
357    fn single_layout_rejects_an_empty_identity() {
358        let error = serde_json::from_str::<SegmentEntityLayout>(r#"{"Single":[]}"#)
359            .expect_err("empty identity must be rejected");
360        assert!(error.to_string().contains("at least one component"));
361    }
362
363    #[test]
364    fn ordering_is_deterministic_with_tie_breakers() {
365        let mut v = vec![
366            seg("c", 10, 20),
367            seg("b", 10, 20),
368            seg("a", 10, 30),
369            seg("d", 5, 7),
370        ];
371
372        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
373
374        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
375        assert_eq!(
376            paths,
377            vec![
378                "data/d.parquet",
379                "data/b.parquet",
380                "data/c.parquet",
381                "data/a.parquet"
382            ]
383        );
384    }
385
386    #[test]
387    fn ordering_is_equal_for_identical_segments() {
388        let a = seg("same", 10, 20);
389        let b = seg("same", 10, 20);
390        assert_eq!(
391            cmp_segment_meta_by_index(&a, &b).unwrap(),
392            std::cmp::Ordering::Equal
393        );
394        assert_eq!(
395            cmp_segment_meta_by_index(&b, &a).unwrap(),
396            std::cmp::Ordering::Equal
397        );
398    }
399
400    #[test]
401    fn ordering_primary_key_ts_min_dominates() {
402        let mut v = vec![seg("z", 20, 30), seg("a", 10, 50), seg("m", 15, 10)];
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/m.parquet", "data/z.parquet"]
410        );
411    }
412
413    #[test]
414    fn ordering_uses_path_as_final_tie_breaker() {
415        let mut v = vec![seg("b", 10, 20), seg("a", 10, 20), seg("c", 10, 20)];
416
417        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
418
419        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
420        assert_eq!(
421            paths,
422            vec!["data/a.parquet", "data/b.parquet", "data/c.parquet"]
423        );
424    }
425
426    #[test]
427    fn segment_bounds_validate_domain_and_native_order() {
428        let signed = IndexKind::Int64 {
429            index_granularity: NonZeroU64::new(1).unwrap(),
430        };
431        let valid = SegmentMeta {
432            path: "data/valid.parquet".to_string(),
433            format: FileFormat::Parquet,
434            entity_layout: SegmentEntityLayout::NotApplicable,
435            index_min: IndexValue::Int64(i64::MIN),
436            index_max: IndexValue::Int64(i64::MAX),
437            row_count: 1,
438            file_size: None,
439            coverage_path: None,
440        };
441        valid.validate_bounds(&signed).unwrap();
442
443        let reversed = SegmentMeta {
444            index_min: IndexValue::Int64(1),
445            index_max: IndexValue::Int64(0),
446            ..valid.clone()
447        };
448        assert!(matches!(
449            reversed.validate_bounds(&signed),
450            Err(SegmentMetaError::InvalidIndexBounds {
451                source: IndexValueError::InvalidBounds { .. },
452                ..
453            })
454        ));
455
456        let wrong_domain = SegmentMeta {
457            index_min: IndexValue::UInt64(0),
458            index_max: IndexValue::UInt64(u64::MAX),
459            ..valid
460        };
461        assert!(matches!(
462            wrong_domain.validate_bounds(&signed),
463            Err(SegmentMetaError::InvalidIndexBounds {
464                source: IndexValueError::KindMismatch { .. },
465                ..
466            })
467        ));
468    }
469
470    #[test]
471    fn integer_segment_ordering_uses_native_bounds_then_path() {
472        let mut segments = vec![
473            SegmentMeta {
474                path: "data/z.parquet".to_string(),
475                format: FileFormat::Parquet,
476                entity_layout: SegmentEntityLayout::NotApplicable,
477                index_min: IndexValue::UInt64(u64::MAX),
478                index_max: IndexValue::UInt64(u64::MAX),
479                row_count: 1,
480                file_size: None,
481                coverage_path: None,
482            },
483            SegmentMeta {
484                path: "data/b.parquet".to_string(),
485                format: FileFormat::Parquet,
486                entity_layout: SegmentEntityLayout::NotApplicable,
487                index_min: IndexValue::UInt64(0),
488                index_max: IndexValue::UInt64(7),
489                row_count: 1,
490                file_size: None,
491                coverage_path: None,
492            },
493            SegmentMeta {
494                path: "data/a.parquet".to_string(),
495                format: FileFormat::Parquet,
496                entity_layout: SegmentEntityLayout::NotApplicable,
497                index_min: IndexValue::UInt64(0),
498                index_max: IndexValue::UInt64(7),
499                row_count: 1,
500                file_size: None,
501                coverage_path: None,
502            },
503        ];
504        segments.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).unwrap());
505        assert_eq!(
506            segments
507                .into_iter()
508                .map(|segment| segment.path)
509                .collect::<Vec<_>>(),
510            vec!["data/a.parquet", "data/b.parquet", "data/z.parquet"]
511        );
512    }
513
514    #[test]
515    fn sorting_rejects_cross_domain_segments() {
516        let mut segments = vec![
517            seg("timestamp", 0, 1),
518            SegmentMeta {
519                path: "data/integer.parquet".to_string(),
520                format: FileFormat::Parquet,
521                entity_layout: SegmentEntityLayout::NotApplicable,
522                index_min: IndexValue::Int64(0),
523                index_max: IndexValue::Int64(1),
524                row_count: 1,
525                file_size: None,
526                coverage_path: None,
527            },
528        ];
529
530        assert!(matches!(
531            sort_segment_meta_by_index(&mut segments),
532            Err(IndexValueError::DomainMismatch { .. })
533        ));
534    }
535
536    #[test]
537    fn segment_json_preserves_integer_bound_extremes() {
538        for (minimum, maximum) in [
539            (IndexValue::Int64(i64::MIN), IndexValue::Int64(i64::MAX)),
540            (IndexValue::UInt64(0), IndexValue::UInt64(u64::MAX)),
541        ] {
542            let segment = SegmentMeta {
543                path: "data/extremes.parquet".to_string(),
544                format: FileFormat::Parquet,
545                entity_layout: SegmentEntityLayout::NotApplicable,
546                index_min: minimum,
547                index_max: maximum,
548                row_count: 2,
549                file_size: Some(42),
550                coverage_path: Some("_coverage/segments/extremes.roar".to_string()),
551            };
552            let json = serde_json::to_string(&segment).unwrap();
553            assert_eq!(serde_json::from_str::<SegmentMeta>(&json).unwrap(), segment);
554        }
555    }
556}
timeseries_table_format/metadata/segments.rs

timeseries_table_format/metadata/
segments.rs
