micromegas_analytics/lakehouse/partition.rs
1use super::view::{ScanSortColumn, ViewMetadata};
2use crate::time::TimeRange;
3use chrono::{DateTime, Utc};
4
5/// Partition metadata (without embedded file_metadata for performance)
6/// Use load_partition_metadata() to load metadata on-demand when needed
7#[derive(Clone, Debug)]
8pub struct Partition {
9 /// Metadata about the view this partition belongs to.
10 pub view_metadata: ViewMetadata,
11 /// The insert time range for this partition.
12 pub insert_time_range: TimeRange,
13 /// The event time range for this partition. None for empty partitions.
14 pub event_time_range: Option<TimeRange>,
15 /// The last time this partition was updated.
16 pub updated: DateTime<Utc>,
17 /// The path to the Parquet file for this partition. None for empty partitions.
18 pub file_path: Option<String>,
19 /// The size of the Parquet file in bytes. 0 for empty partitions.
20 pub file_size: i64,
21 /// A hash of the source data that generated this partition.
22 pub source_data_hash: Vec<u8>,
23 /// The number of rows in this partition. 0 for empty partitions.
24 pub num_rows: i64,
25 /// The sort guarantee this partition's rows carry, if any (e.g. `Some(["insert_time"])`).
26 /// `None` means no ordering guarantee is recorded -- true for every partition written before
27 /// this field existed, and for any view/merge that doesn't declare one. See
28 /// `View::get_merged_partition_sort_order` and the `blocks_view_ordered_merges_plan.md` plan.
29 pub sort_order: Option<Vec<String>>,
30 /// The true maximum value of the view's declared `Concatenated` leading sort column across
31 /// this partition's rows (`begin`, for `thread_spans`) -- as opposed to `max_event_time`,
32 /// which for `thread_spans` is the max span *end* and only a loose stand-in for it. `None`
33 /// means "not recorded": every partition written before this field existed, and every
34 /// partition from a view that doesn't declare a `Concatenated` event-time ordering (only
35 /// `ThreadSpansView` populates it today). See
36 /// `tasks/completed/thread_spans_segment_boundary_overlap_plan.md`, Part B.
37 pub max_sort_key_time: Option<DateTime<Utc>>,
38}
39
40impl Partition {
41 /// Returns true if this partition has no data (num_rows = 0).
42 pub fn is_empty(&self) -> bool {
43 self.num_rows == 0
44 }
45
46 /// Returns the min event time, if this partition has data.
47 pub fn min_event_time(&self) -> Option<DateTime<Utc>> {
48 self.event_time_range.as_ref().map(|r| r.begin)
49 }
50
51 /// Returns the max event time, if this partition has data.
52 pub fn max_event_time(&self) -> Option<DateTime<Utc>> {
53 self.event_time_range.as_ref().map(|r| r.end)
54 }
55
56 /// Returns the recorded true maximum leading-sort-column value, if any. See the field's doc
57 /// for what "recorded" means and why it can be `None` even for a non-empty partition.
58 pub fn max_sort_key_time(&self) -> Option<DateTime<Utc>> {
59 self.max_sort_key_time
60 }
61
62 /// Returns the beginning of the insert time range.
63 pub fn begin_insert_time(&self) -> DateTime<Utc> {
64 self.insert_time_range.begin
65 }
66
67 /// Returns the end of the insert time range.
68 pub fn end_insert_time(&self) -> DateTime<Utc> {
69 self.insert_time_range.end
70 }
71
72 /// True when this partition's recorded `sort_order` certifies `columns` as an ordering its
73 /// rows already satisfy: `columns` (ascending-only) must name a prefix of the recorded,
74 /// ascending-implied sort order. Empty partitions certify vacuously -- there are no rows to
75 /// violate the ordering. An empty (non-empty-partition) `columns` never certifies: it names no
76 /// ordering to satisfy, so declaring it certified would let a `ScanOrdering::PerFile { columns:
77 /// vec![] }` degrade to `Unordered` instead of silently planning and recording a vacuous
78 /// ordering (see `ScanOrdering::PerFile`).
79 pub fn certifies_sort_order(&self, columns: &[ScanSortColumn]) -> bool {
80 if self.is_empty() {
81 return true;
82 }
83 if columns.is_empty() {
84 return false;
85 }
86 let Some(recorded) = &self.sort_order else {
87 return false;
88 };
89 columns.len() <= recorded.len()
90 && columns
91 .iter()
92 .zip(recorded.iter())
93 .all(|(declared, recorded_col)| {
94 !declared.descending && *declared.column == *recorded_col
95 })
96 }
97
98 /// Validates partition invariants. Returns error if partition is inconsistent.
99 ///
100 /// Invariants:
101 /// - Non-empty partitions (num_rows > 0) MUST have both event_time_range and file_path
102 /// - Empty partitions (num_rows = 0) MUST NOT have event_time_range or file_path
103 /// - num_rows must not be negative
104 /// - A partition with no event_time_range MUST NOT carry a max_sort_key_time (an empty
105 /// partition has no sort-key bound to record)
106 ///
107 /// Deliberately **not** checked here: any ordering relationship between `max_sort_key_time`
108 /// and `min_event_time`/`max_event_time` (e.g. `min_event_time <= max_sort_key_time <=
109 /// max_event_time`). That holds by construction for `thread_spans` today, but two properties
110 /// of this function rule out asserting it as an invariant:
111 ///
112 /// 1. `validate` runs only on the read path (its callers are all in `partition_cache.rs`,
113 /// propagating with `?`); nothing on the write path calls it. A violating partition would
114 /// therefore be written and committed silently, then hard-fail *reads* -- and since the
115 /// fetch this feeds is not view-scoped, one bad row would fail materialization for every
116 /// view, not just the offending one. The invariant must assert only what the writer
117 /// guarantees by construction, never a hopeful expectation.
118 /// 2. The bound would depend on data the writer doesn't control: `min_event_time`/
119 /// `max_event_time` derive straight from client-supplied block ticks with no validation
120 /// anywhere in ingestion or analytics, so a malformed client (e.g. `begin_ticks >
121 /// end_ticks`) could make an ordering clause unsatisfiable -- turning one bad row into a
122 /// lakehouse-wide read outage. And the invariant would apply to every future view
123 /// populating `max_sort_key_time`, not just `thread_spans`: `BlocksView`, for instance, has
124 /// a `Concatenated` sort key of `insert_time` while its `event_time_range` is `[min
125 /// begin_time, max insert_time]`, so a client clock skewed ahead of the server could put a
126 /// block's `begin_time` above its `insert_time`.
127 ///
128 /// The row-level property this plan actually relies on (the partition's true max `begin`
129 /// really is bounded by its `event_time_range`) is asserted where it is cheap and safe
130 /// instead: at write time (`ensure_begin_non_decreasing`) and pinned by a no-DB test
131 /// (`call_tree_tests.rs`).
132 pub fn validate(&self) -> anyhow::Result<()> {
133 if self.num_rows > 0 {
134 // Non-empty partition must have event_time_range and file_path
135 if self.event_time_range.is_none() {
136 anyhow::bail!(
137 "non-empty partition (num_rows={}) has no event_time_range",
138 self.num_rows
139 );
140 }
141 if self.file_path.is_none() {
142 anyhow::bail!(
143 "non-empty partition (num_rows={}) has no file_path",
144 self.num_rows
145 );
146 }
147 } else if self.num_rows == 0 {
148 // Empty partition must NOT have event_time_range or file_path
149 if self.event_time_range.is_some() {
150 anyhow::bail!("empty partition has event_time_range");
151 }
152 if self.file_path.is_some() {
153 anyhow::bail!("empty partition has file_path");
154 }
155 } else {
156 anyhow::bail!("partition has negative num_rows: {}", self.num_rows);
157 }
158 if self.event_time_range.is_none() && self.max_sort_key_time.is_some() {
159 anyhow::bail!("partition has no event_time_range but carries a max_sort_key_time");
160 }
161 Ok(())
162 }
163}