chronix_engine/segment/
header.rs

1//! Segment file header and footer structures.
2//!
3//! Every `.csx` segment file begins with a [`SegmentHeader`] and ends with a
4//! [`SegmentFooter`]. The footer contains the `CRC32c` checksum of the entire
5//! file (minus the footer itself) for integrity verification.
6//!
7//! ## Format Versioning
8//!
9//! The header contains a `version: u16` field (bytes 4–5, after the 4-byte
10//! magic `CXSG`). The current format version is [`VERSION`].
11//!
12//! On read, [`SegmentHeader::from_bytes`] requires `version == VERSION`
13//! exactly — a file from a newer writer *or* an older one is refused with
14//! [`SegmentError::UnsupportedVersion`].
15//!
16//! Accepting an older version would be silent corruption rather than
17//! compatibility: v2 named its time column `timestamp` and v3 names it
18//! `_time`, so a v2 file opened by a v3 reader yields a batch with a column
19//! nothing looks for — no error, no rows. Chronix is pre-release and there is
20//! no data in the wild, so the reader refuses instead of translating.
21
22use crate::segment::error::{Result, SegmentError};
23use crate::segment::to_array;
24
25/// Segment file magic bytes: `CXSG`.
26pub const MAGIC: [u8; 4] = *b"CXSG";
27
28/// Current segment format version.
29///
30/// - **v1** — original layout.
31/// - **v2** — adds per-block validity bitmaps (D-NULL). Absent values are no
32///   longer encoded as `0`/`""`/`false` sentinels, so `IS NULL` and
33///   aggregates are correct on sparse data. See
34///   [`ColumnBlockMeta::validity_length`].
35/// - **v3** — the time column is stored as `_time`, the one name every layer
36///   uses. It was `timestamp` here, `time` in the schema registry and `_time`
37///   in SQL, so a reader who printed a schema and then wrote a query had to
38///   translate.
39///
40/// Chronix is pre-release, so readers for older versions were removed rather
41/// than kept: there is no data in the wild to migrate, and a reader that
42/// *accepted* v2 would hand back a column called `timestamp` that nothing
43/// looks for any more — zero rows and no error.
44///
45/// [`ColumnBlockMeta::validity_length`]: crate::segment::metadata::ColumnBlockMeta::validity_length
46pub const VERSION: u16 = 3;
47
48/// Size of the serialized header in bytes.
49pub const HEADER_SIZE: usize = 4 + 2 + 2 + 8 + 8 + 8 + 8 + 2 + 4 + 1 + 1;
50// magic(4) + version(2) + flags(2) + created_at(8) + min_ts(8) + max_ts(8)
51// + row_count(8) + column_count(2) + series_count(4) + compression(1)
52// + sort_order(1)
53
54/// Size of the serialized footer in bytes.
55pub const FOOTER_SIZE: usize = 8 + 4 + 4 + 4 + 4;
56// metadata_offset(8) + row_group_count(4) + checksum(4) + magic(4)
57
58/// Segment file header.
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub struct SegmentHeader {
61    /// Format version.
62    pub version: u16,
63    /// Flags (reserved for future use).
64    pub flags: u16,
65    /// Creation timestamp (nanoseconds since epoch).
66    pub created_at: i64,
67    /// Minimum timestamp across all rows in the segment.
68    pub min_timestamp: i64,
69    /// Maximum timestamp across all rows in the segment.
70    pub max_timestamp: i64,
71    /// Total number of rows in the segment.
72    pub row_count: u64,
73    /// Number of columns (including timestamp).
74    pub column_count: u16,
75    /// Number of unique series in the segment.
76    pub series_count: u32,
77    /// Compression codec: 0 = LZ4, 1 = Zstd, 2 = None.
78    pub compression: u8,
79    /// Sort order: 0 = unsorted, 1 = by series+timestamp.
80    pub sort_order: u8,
81}
82
83impl SegmentHeader {
84    /// Serialize the header to bytes (always [`HEADER_SIZE`] bytes).
85    #[must_use]
86    pub fn to_bytes(&self) -> Vec<u8> {
87        let mut buf = Vec::with_capacity(HEADER_SIZE);
88        buf.extend_from_slice(&MAGIC);
89        buf.extend_from_slice(&self.version.to_le_bytes());
90        buf.extend_from_slice(&self.flags.to_le_bytes());
91        buf.extend_from_slice(&self.created_at.to_le_bytes());
92        buf.extend_from_slice(&self.min_timestamp.to_le_bytes());
93        buf.extend_from_slice(&self.max_timestamp.to_le_bytes());
94        buf.extend_from_slice(&self.row_count.to_le_bytes());
95        buf.extend_from_slice(&self.column_count.to_le_bytes());
96        buf.extend_from_slice(&self.series_count.to_le_bytes());
97        buf.push(self.compression);
98        buf.push(self.sort_order);
99        buf
100    }
101
102    /// Deserialize a header from bytes.
103    ///
104    /// # Errors
105    ///
106    /// Returns [`SegmentError::InvalidMagic`] if magic bytes don't match,
107    /// or [`SegmentError::UnsupportedVersion`] if version is not supported.
108    pub fn from_bytes(data: &[u8]) -> Result<Self> {
109        if data.len() < HEADER_SIZE {
110            return Err(SegmentError::CorruptFile {
111                detail: format!(
112                    "header too short: expected {HEADER_SIZE}, got {}",
113                    data.len()
114                ),
115            });
116        }
117
118        if data[..4] != MAGIC {
119            return Err(SegmentError::InvalidMagic);
120        }
121
122        let version = u16::from_le_bytes([data[4], data[5]]);
123        // Equality, not `<=`: an older segment is a *different* format, and
124        // reading one would produce a batch whose time column has the name
125        // this version stopped using.
126        if version != VERSION {
127            return Err(SegmentError::UnsupportedVersion { version });
128        }
129
130        let flags = u16::from_le_bytes([data[6], data[7]]);
131        let created_at = i64::from_le_bytes(to_array!(data[8..16], "header created_at")?);
132        let min_timestamp = i64::from_le_bytes(to_array!(data[16..24], "header min_timestamp")?);
133        let max_timestamp = i64::from_le_bytes(to_array!(data[24..32], "header max_timestamp")?);
134        let row_count = u64::from_le_bytes(to_array!(data[32..40], "header row_count")?);
135        let column_count = u16::from_le_bytes([data[40], data[41]]);
136        let series_count = u32::from_le_bytes(to_array!(data[42..46], "header series_count")?);
137        let compression = data[46];
138        let sort_order = data[47];
139
140        Ok(Self {
141            version,
142            flags,
143            created_at,
144            min_timestamp,
145            max_timestamp,
146            row_count,
147            column_count,
148            series_count,
149            compression,
150            sort_order,
151        })
152    }
153}
154
155/// Segment file footer.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct SegmentFooter {
158    /// Byte offset where column metadata section begins.
159    pub metadata_offset: u64,
160    /// Number of row groups in the segment.
161    pub row_group_count: u32,
162    /// `CRC32c` checksum of the entire file excluding this footer.
163    pub checksum: u32,
164    /// Independent `CRC32c` checksum of the metadata section.
165    /// Allows metadata validation without re-checksumming the entire file.
166    pub metadata_checksum: u32,
167}
168
169impl SegmentFooter {
170    /// Serialize the footer to bytes (always [`FOOTER_SIZE`] bytes).
171    #[must_use]
172    pub fn to_bytes(&self) -> Vec<u8> {
173        let mut buf = Vec::with_capacity(FOOTER_SIZE);
174        buf.extend_from_slice(&self.metadata_offset.to_le_bytes());
175        buf.extend_from_slice(&self.row_group_count.to_le_bytes());
176        buf.extend_from_slice(&self.checksum.to_le_bytes());
177        buf.extend_from_slice(&self.metadata_checksum.to_le_bytes());
178        buf.extend_from_slice(&MAGIC);
179        buf
180    }
181
182    /// Deserialize a footer from bytes.
183    ///
184    /// # Errors
185    ///
186    /// Returns [`SegmentError::InvalidMagic`] if trailing magic bytes don't
187    /// match, or [`SegmentError::CorruptFile`] if the data is too short.
188    pub fn from_bytes(data: &[u8]) -> Result<Self> {
189        if data.len() < FOOTER_SIZE {
190            return Err(SegmentError::CorruptFile {
191                detail: format!(
192                    "footer too short: expected {FOOTER_SIZE}, got {}",
193                    data.len()
194                ),
195            });
196        }
197
198        let metadata_offset = u64::from_le_bytes(to_array!(data[..8], "footer metadata_offset")?);
199        let row_group_count = u32::from_le_bytes(to_array!(data[8..12], "footer row_group_count")?);
200        let checksum = u32::from_le_bytes(to_array!(data[12..16], "footer checksum")?);
201        let metadata_checksum =
202            u32::from_le_bytes(to_array!(data[16..20], "footer metadata_checksum")?);
203
204        if data[20..24] != MAGIC {
205            return Err(SegmentError::InvalidMagic);
206        }
207
208        Ok(Self {
209            metadata_offset,
210            row_group_count,
211            checksum,
212            metadata_checksum,
213        })
214    }
215}
216
217#[cfg(test)]
218mod tests {
219    use super::*;
220
221    #[test]
222    fn header_roundtrip() {
223        let header = SegmentHeader {
224            version: VERSION,
225            flags: 0,
226            created_at: 1_700_000_000_000_000_000,
227            min_timestamp: 1_000_000,
228            max_timestamp: 2_000_000,
229            row_count: 10_000,
230            column_count: 5,
231            series_count: 42,
232            compression: 0,
233            sort_order: 1,
234        };
235        let bytes = header.to_bytes();
236        assert_eq!(bytes.len(), HEADER_SIZE);
237        let recovered = SegmentHeader::from_bytes(&bytes).unwrap();
238        assert_eq!(header, recovered);
239    }
240
241    #[test]
242    fn header_invalid_magic() {
243        let mut bytes = SegmentHeader {
244            version: VERSION,
245            flags: 0,
246            created_at: 0,
247            min_timestamp: 0,
248            max_timestamp: 0,
249            row_count: 0,
250            column_count: 0,
251            series_count: 0,
252            compression: 0,
253            sort_order: 0,
254        }
255        .to_bytes();
256        bytes[0] = b'X';
257        assert!(matches!(
258            SegmentHeader::from_bytes(&bytes),
259            Err(SegmentError::InvalidMagic)
260        ));
261    }
262
263    #[test]
264    fn header_unsupported_version() {
265        let mut bytes = SegmentHeader {
266            version: VERSION,
267            flags: 0,
268            created_at: 0,
269            min_timestamp: 0,
270            max_timestamp: 0,
271            row_count: 0,
272            column_count: 0,
273            series_count: 0,
274            compression: 0,
275            sort_order: 0,
276        }
277        .to_bytes();
278        // Set version to 999
279        bytes[4..6].copy_from_slice(&999u16.to_le_bytes());
280        assert!(matches!(
281            SegmentHeader::from_bytes(&bytes),
282            Err(SegmentError::UnsupportedVersion { version: 999 })
283        ));
284    }
285
286    #[test]
287    fn footer_roundtrip() {
288        let footer = SegmentFooter {
289            metadata_offset: 1024,
290            row_group_count: 3,
291            checksum: 0xDEAD_BEEF,
292            metadata_checksum: 0xCAFE_BABE,
293        };
294        let bytes = footer.to_bytes();
295        assert_eq!(bytes.len(), FOOTER_SIZE);
296        let recovered = SegmentFooter::from_bytes(&bytes).unwrap();
297        assert_eq!(footer, recovered);
298    }
299
300    #[test]
301    fn footer_invalid_magic() {
302        let mut bytes = SegmentFooter {
303            metadata_offset: 0,
304            row_group_count: 0,
305            checksum: 0,
306            metadata_checksum: 0,
307        }
308        .to_bytes();
309        let len = bytes.len();
310        bytes[len - 1] = b'X';
311        assert!(matches!(
312            SegmentFooter::from_bytes(&bytes),
313            Err(SegmentError::InvalidMagic)
314        ));
315    }
316
317    #[test]
318    fn header_too_short() {
319        assert!(SegmentHeader::from_bytes(&[0; 10]).is_err());
320    }
321
322    #[test]
323    fn footer_too_short() {
324        assert!(SegmentFooter::from_bytes(&[0; 5]).is_err());
325    }
326
327    #[test]
328    fn header_version_zero_rejected() {
329        let header = SegmentHeader {
330            version: VERSION,
331            flags: 0,
332            created_at: 0,
333            min_timestamp: 0,
334            max_timestamp: 0,
335            row_count: 0,
336            column_count: 0,
337            series_count: 0,
338            compression: 0,
339            sort_order: 0,
340        };
341        let mut bytes = header.to_bytes();
342        // Set version to 0
343        bytes[4] = 0;
344        bytes[5] = 0;
345        assert!(matches!(
346            SegmentHeader::from_bytes(&bytes),
347            Err(SegmentError::UnsupportedVersion { version: 0 })
348        ));
349    }
350
351    /// An *older* segment is refused too, not read.
352    ///
353    /// v2 called the time column `timestamp`; v3 calls it `_time`. A reader
354    /// that accepted v2 would return a batch whose time column nothing looks
355    /// for — no rows, no error — so the version check is equality.
356    #[test]
357    fn header_older_version_rejected() {
358        let header = SegmentHeader {
359            version: VERSION,
360            flags: 0,
361            created_at: 0,
362            min_timestamp: 0,
363            max_timestamp: 0,
364            row_count: 0,
365            column_count: 0,
366            series_count: 0,
367            compression: 0,
368            sort_order: 0,
369        };
370        let mut bytes = header.to_bytes();
371        let older = super::VERSION - 1;
372        bytes[4] = older as u8;
373        bytes[5] = (older >> 8) as u8;
374        assert!(matches!(
375            SegmentHeader::from_bytes(&bytes),
376            Err(SegmentError::UnsupportedVersion { .. })
377        ));
378    }
379
380    #[test]
381    fn header_future_version_rejected() {
382        let header = SegmentHeader {
383            version: VERSION,
384            flags: 0,
385            created_at: 0,
386            min_timestamp: 0,
387            max_timestamp: 0,
388            row_count: 0,
389            column_count: 0,
390            series_count: 0,
391            compression: 0,
392            sort_order: 0,
393        };
394        let mut bytes = header.to_bytes();
395        // Set version to VERSION + 1
396        let future = super::VERSION + 1;
397        bytes[4] = future as u8;
398        bytes[5] = (future >> 8) as u8;
399        assert!(matches!(
400            SegmentHeader::from_bytes(&bytes),
401            Err(SegmentError::UnsupportedVersion { .. })
402        ));
403    }
404}
chronix_engine/segment/header.rs

chronix_engine/segment/
header.rs
