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        logical_schema::LogicalSchemaError,
16        table_metadata::{IndexKind, IndexValue, IndexValueError},
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/// `transaction_log::segments::SegmentError`).
148#[derive(Debug, Snafu)]
149pub enum SegmentMetaError {
150    /// Persisted segment bounds violate the table's ordered-index domain.
151    #[snafu(display("Invalid ordered-index bounds in segment at {path}: {source}"))]
152    InvalidIndexBounds {
153        /// Segment path containing invalid bounds.
154        path: String,
155        /// Domain or ordering failure.
156        source: IndexValueError,
157    },
158
159    /// The file is too short to be a valid Parquet file.
160    #[snafu(display("Segment file too short to be valid Parquet: {path}"))]
161    TooShort {
162        /// The path to the file that was too short.
163        path: String,
164    },
165
166    /// Parquet reader / metadata failure.
167    #[snafu(display("Error reading Parquet metadata for segment at {path}: {source}"))]
168    ParquetRead {
169        /// The path to the file that caused the Parquet read failure.
170        path: String,
171        /// Underlying parquet error that caused this failure.
172        source: ParquetError,
173        /// Diagnostic backtrace for this error.
174        backtrace: Backtrace,
175    },
176
177    /// The registered ordered-index column is missing or incompatible.
178    #[snafu(transparent)]
179    OrderedIndexColumn {
180        /// Exact registered and observed Parquet column details.
181        source: ParquetIndexColumnError,
182    },
183
184    /// Statistics exist but are not well-shaped (wrong length / unexpected type).
185    #[snafu(display(
186        "Parquet statistics shape invalid for {column} in segment at {path}: {detail}"
187    ))]
188    ParquetStatsShape {
189        /// The path to the file with malformed Parquet statistics.
190        path: String,
191        /// The column whose statistics are malformed.
192        column: String,
193        /// Details about how the statistics are malformed.
194        detail: String,
195    },
196
197    /// The file contains no non-null value for the registered ordered index.
198    #[snafu(display(
199        "No observed {expected_domain} value for ordered-index column {column} in segment at {path}"
200    ))]
201    NoObservedIndexValue {
202        /// Path to the segment file.
203        path: String,
204        /// Registered ordered-index column.
205        column: String,
206        /// Registered ordered-index domain.
207        expected_domain: &'static str,
208    },
209
210    /// Failed to derive a valid LogicalSchema from the Parquet file.
211    #[snafu(display("Invalid logical schema derived from Parquet at {path}: {source}"))]
212    LogicalSchemaInvalid {
213        /// The path to the file without a valid LogicalSchema.
214        path: String,
215        /// Underlying logical schema error that triggered this failure.
216        #[snafu(source)]
217        source: LogicalSchemaError,
218    },
219}
220
221/// Deterministic ordering for segments by ordered-index bounds.
222///
223/// Ordering is by `index_min`, then `index_max`, and finally `path` as a stable
224/// tie-breaker.
225pub(crate) fn cmp_segment_meta_by_index(
226    a: &SegmentMeta,
227    b: &SegmentMeta,
228) -> Result<std::cmp::Ordering, IndexValueError> {
229    let min_order = a.index_min.compare(&b.index_min)?;
230    if !min_order.is_eq() {
231        return Ok(min_order);
232    }
233    let max_order = a.index_max.compare(&b.index_max)?;
234    Ok(max_order.then_with(|| a.path.cmp(&b.path)))
235}
236
237/// Sort segment metadata by typed bounds, rejecting cross-domain values first.
238pub(crate) fn sort_segment_meta_by_index<T>(segments: &mut [T]) -> Result<(), IndexValueError>
239where
240    T: std::borrow::Borrow<SegmentMeta>,
241{
242    let mut domain: Option<&IndexValue> = None;
243    for segment in segments.iter().map(std::borrow::Borrow::borrow) {
244        if segment.index_min.compare(&segment.index_max)?.is_gt() {
245            return Err(IndexValueError::InvalidBounds {
246                min: segment.index_min.clone(),
247                max: segment.index_max.clone(),
248            });
249        }
250        if let Some(domain) = domain {
251            domain.compare(&segment.index_min)?;
252            domain.compare(&segment.index_max)?;
253        } else {
254            domain = Some(&segment.index_min);
255        }
256    }
257
258    let mut sort_error = None;
259    segments.sort_unstable_by(
260        |a, b| match cmp_segment_meta_by_index(a.borrow(), b.borrow()) {
261            Ok(order) => order,
262            Err(error) => {
263                sort_error.get_or_insert(error);
264                std::cmp::Ordering::Equal
265            }
266        },
267    );
268    if let Some(error) = sort_error {
269        return Err(error);
270    }
271    Ok(())
272}
273
274#[cfg(test)]
275mod tests {
276    use std::num::NonZeroU64;
277
278    use super::*;
279    use chrono::{TimeZone, Utc};
280
281    fn seg(id: &str, ts_min: i64, ts_max: i64) -> SegmentMeta {
282        SegmentMeta {
283            path: format!("data/{id}.parquet"),
284            format: FileFormat::Parquet,
285            entity_layout: SegmentEntityLayout::NotApplicable,
286            index_min: IndexValue::Timestamp(Utc.timestamp_opt(ts_min, 0).single().unwrap()),
287            index_max: IndexValue::Timestamp(Utc.timestamp_opt(ts_max, 0).single().unwrap()),
288            row_count: 1,
289            file_size: None,
290            coverage_path: None,
291        }
292    }
293
294    #[test]
295    fn entity_layout_json_roundtrips_are_stable() {
296        let cases = [
297            (SegmentEntityLayout::NotApplicable, "\"NotApplicable\""),
298            (
299                SegmentEntityLayout::Single(
300                    EntityIdentity::try_new(vec!["us".into(), "device-1".into()]).unwrap(),
301                ),
302                r#"{"Single":[{"type":"utf8","value":"us"},{"type":"utf8","value":"device-1"}]}"#,
303            ),
304            (SegmentEntityLayout::Mixed, "\"Mixed\""),
305        ];
306
307        for (layout, expected_json) in cases {
308            let json = serde_json::to_string(&layout).unwrap();
309            assert_eq!(json, expected_json);
310            assert_eq!(
311                serde_json::from_str::<SegmentEntityLayout>(&json).unwrap(),
312                layout
313            );
314        }
315    }
316
317    #[test]
318    fn single_layout_rejects_an_empty_identity() {
319        let error = serde_json::from_str::<SegmentEntityLayout>(r#"{"Single":[]}"#)
320            .expect_err("empty identity must be rejected");
321        assert!(error.to_string().contains("at least one component"));
322    }
323
324    #[test]
325    fn ordering_is_deterministic_with_tie_breakers() {
326        let mut v = vec![
327            seg("c", 10, 20),
328            seg("b", 10, 20),
329            seg("a", 10, 30),
330            seg("d", 5, 7),
331        ];
332
333        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
334
335        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
336        assert_eq!(
337            paths,
338            vec![
339                "data/d.parquet",
340                "data/b.parquet",
341                "data/c.parquet",
342                "data/a.parquet"
343            ]
344        );
345    }
346
347    #[test]
348    fn ordering_is_equal_for_identical_segments() {
349        let a = seg("same", 10, 20);
350        let b = seg("same", 10, 20);
351        assert_eq!(
352            cmp_segment_meta_by_index(&a, &b).unwrap(),
353            std::cmp::Ordering::Equal
354        );
355        assert_eq!(
356            cmp_segment_meta_by_index(&b, &a).unwrap(),
357            std::cmp::Ordering::Equal
358        );
359    }
360
361    #[test]
362    fn ordering_primary_key_ts_min_dominates() {
363        let mut v = vec![seg("z", 20, 30), seg("a", 10, 50), seg("m", 15, 10)];
364
365        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
366
367        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
368        assert_eq!(
369            paths,
370            vec!["data/a.parquet", "data/m.parquet", "data/z.parquet"]
371        );
372    }
373
374    #[test]
375    fn ordering_uses_path_as_final_tie_breaker() {
376        let mut v = vec![seg("b", 10, 20), seg("a", 10, 20), seg("c", 10, 20)];
377
378        v.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).expect("matching domains"));
379
380        let paths: Vec<String> = v.into_iter().map(|s| s.path).collect();
381        assert_eq!(
382            paths,
383            vec!["data/a.parquet", "data/b.parquet", "data/c.parquet"]
384        );
385    }
386
387    #[test]
388    fn segment_bounds_validate_domain_and_native_order() {
389        let signed = IndexKind::Int64 {
390            bucket_width: NonZeroU64::new(1).unwrap(),
391        };
392        let valid = SegmentMeta {
393            path: "data/valid.parquet".to_string(),
394            format: FileFormat::Parquet,
395            entity_layout: SegmentEntityLayout::NotApplicable,
396            index_min: IndexValue::Int64(i64::MIN),
397            index_max: IndexValue::Int64(i64::MAX),
398            row_count: 1,
399            file_size: None,
400            coverage_path: None,
401        };
402        valid.validate_bounds(&signed).unwrap();
403
404        let reversed = SegmentMeta {
405            index_min: IndexValue::Int64(1),
406            index_max: IndexValue::Int64(0),
407            ..valid.clone()
408        };
409        assert!(matches!(
410            reversed.validate_bounds(&signed),
411            Err(SegmentMetaError::InvalidIndexBounds {
412                source: IndexValueError::InvalidBounds { .. },
413                ..
414            })
415        ));
416
417        let wrong_domain = SegmentMeta {
418            index_min: IndexValue::UInt64(0),
419            index_max: IndexValue::UInt64(u64::MAX),
420            ..valid
421        };
422        assert!(matches!(
423            wrong_domain.validate_bounds(&signed),
424            Err(SegmentMetaError::InvalidIndexBounds {
425                source: IndexValueError::KindMismatch { .. },
426                ..
427            })
428        ));
429    }
430
431    #[test]
432    fn integer_segment_ordering_uses_native_bounds_then_path() {
433        let mut segments = vec![
434            SegmentMeta {
435                path: "data/z.parquet".to_string(),
436                format: FileFormat::Parquet,
437                entity_layout: SegmentEntityLayout::NotApplicable,
438                index_min: IndexValue::UInt64(u64::MAX),
439                index_max: IndexValue::UInt64(u64::MAX),
440                row_count: 1,
441                file_size: None,
442                coverage_path: None,
443            },
444            SegmentMeta {
445                path: "data/b.parquet".to_string(),
446                format: FileFormat::Parquet,
447                entity_layout: SegmentEntityLayout::NotApplicable,
448                index_min: IndexValue::UInt64(0),
449                index_max: IndexValue::UInt64(7),
450                row_count: 1,
451                file_size: None,
452                coverage_path: None,
453            },
454            SegmentMeta {
455                path: "data/a.parquet".to_string(),
456                format: FileFormat::Parquet,
457                entity_layout: SegmentEntityLayout::NotApplicable,
458                index_min: IndexValue::UInt64(0),
459                index_max: IndexValue::UInt64(7),
460                row_count: 1,
461                file_size: None,
462                coverage_path: None,
463            },
464        ];
465        segments.sort_unstable_by(|a, b| cmp_segment_meta_by_index(a, b).unwrap());
466        assert_eq!(
467            segments
468                .into_iter()
469                .map(|segment| segment.path)
470                .collect::<Vec<_>>(),
471            vec!["data/a.parquet", "data/b.parquet", "data/z.parquet"]
472        );
473    }
474
475    #[test]
476    fn sorting_rejects_cross_domain_segments() {
477        let mut segments = vec![
478            seg("timestamp", 0, 1),
479            SegmentMeta {
480                path: "data/integer.parquet".to_string(),
481                format: FileFormat::Parquet,
482                entity_layout: SegmentEntityLayout::NotApplicable,
483                index_min: IndexValue::Int64(0),
484                index_max: IndexValue::Int64(1),
485                row_count: 1,
486                file_size: None,
487                coverage_path: None,
488            },
489        ];
490
491        assert!(matches!(
492            sort_segment_meta_by_index(&mut segments),
493            Err(IndexValueError::DomainMismatch { .. })
494        ));
495    }
496
497    #[test]
498    fn segment_json_preserves_integer_bound_extremes() {
499        for (minimum, maximum) in [
500            (IndexValue::Int64(i64::MIN), IndexValue::Int64(i64::MAX)),
501            (IndexValue::UInt64(0), IndexValue::UInt64(u64::MAX)),
502        ] {
503            let segment = SegmentMeta {
504                path: "data/extremes.parquet".to_string(),
505                format: FileFormat::Parquet,
506                entity_layout: SegmentEntityLayout::NotApplicable,
507                index_min: minimum,
508                index_max: maximum,
509                row_count: 2,
510                file_size: Some(42),
511                coverage_path: Some("_coverage/segments/extremes.roar".to_string()),
512            };
513            let json = serde_json::to_string(&segment).unwrap();
514            assert_eq!(serde_json::from_str::<SegmentMeta>(&json).unwrap(), segment);
515        }
516    }
517}
timeseries_table_format/metadata/segments.rs

timeseries_table_format/metadata/
segments.rs
