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 is the first format that exists.** The layout changed several times
31/// before release; those generations were superseded before anyone could
32/// hold one, and their readers were deleted with them, so the count starts
33/// at the version that ships rather than at the number of edits it took to
34/// get there.
35///
36/// The version is checked for **equality**, not `<=`. Two layouts can differ
37/// in a way no tolerant reader bridges — one that named the time column
38/// `timestamp` against one that names it `_time` yields a batch whose time
39/// column nothing looks for, which is zero rows and no error — so a file
40/// this reader did not write is refused rather than guessed at.
41///
42/// [`ColumnBlockMeta::validity_length`]: crate::segment::metadata::ColumnBlockMeta::validity_length
43pub const VERSION: u16 = 1;
44
45/// Size of the serialized header in bytes.
46pub const HEADER_SIZE: usize = 4 + 2 + 2 + 8 + 8 + 8 + 8 + 2 + 4 + 1 + 1;
47// magic(4) + version(2) + flags(2) + created_at(8) + min_ts(8) + max_ts(8)
48// + row_count(8) + column_count(2) + series_count(4) + compression(1)
49// + sort_order(1)
50
51/// Size of the serialized footer in bytes.
52pub const FOOTER_SIZE: usize = 8 + 4 + 4 + 4 + 4;
53// metadata_offset(8) + row_group_count(4) + checksum(4) + magic(4)
54
55/// Segment file header.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct SegmentHeader {
58    /// Format version.
59    pub version: u16,
60    /// Flags (reserved for future use).
61    pub flags: u16,
62    /// Creation timestamp (nanoseconds since epoch).
63    pub created_at: i64,
64    /// Minimum timestamp across all rows in the segment.
65    pub min_timestamp: i64,
66    /// Maximum timestamp across all rows in the segment.
67    pub max_timestamp: i64,
68    /// Total number of rows in the segment.
69    pub row_count: u64,
70    /// Number of columns (including timestamp).
71    pub column_count: u16,
72    /// Number of unique series in the segment.
73    pub series_count: u32,
74    /// Compression codec: 0 = LZ4, 1 = Zstd, 2 = None.
75    pub compression: u8,
76    /// Sort order: 0 = unsorted, 1 = by series+timestamp.
77    pub sort_order: u8,
78}
79
80impl SegmentHeader {
81    /// Serialize the header to bytes (always [`HEADER_SIZE`] bytes).
82    #[must_use]
83    pub fn to_bytes(&self) -> Vec<u8> {
84        let mut buf = Vec::with_capacity(HEADER_SIZE);
85        buf.extend_from_slice(&MAGIC);
86        buf.extend_from_slice(&self.version.to_le_bytes());
87        buf.extend_from_slice(&self.flags.to_le_bytes());
88        buf.extend_from_slice(&self.created_at.to_le_bytes());
89        buf.extend_from_slice(&self.min_timestamp.to_le_bytes());
90        buf.extend_from_slice(&self.max_timestamp.to_le_bytes());
91        buf.extend_from_slice(&self.row_count.to_le_bytes());
92        buf.extend_from_slice(&self.column_count.to_le_bytes());
93        buf.extend_from_slice(&self.series_count.to_le_bytes());
94        buf.push(self.compression);
95        buf.push(self.sort_order);
96        buf
97    }
98
99    /// Deserialize a header from bytes.
100    ///
101    /// # Errors
102    ///
103    /// Returns [`SegmentError::InvalidMagic`] if magic bytes don't match,
104    /// or [`SegmentError::UnsupportedVersion`] if version is not supported.
105    pub fn from_bytes(data: &[u8]) -> Result<Self> {
106        if data.len() < HEADER_SIZE {
107            return Err(SegmentError::CorruptFile {
108                detail: format!(
109                    "header too short: expected {HEADER_SIZE}, got {}",
110                    data.len()
111                ),
112            });
113        }
114
115        if data[..4] != MAGIC {
116            return Err(SegmentError::InvalidMagic);
117        }
118
119        let version = u16::from_le_bytes([data[4], data[5]]);
120        // Equality, not `<=`: an older segment is a *different* format, and
121        // reading one would produce a batch whose time column has the name
122        // this version stopped using.
123        if version != VERSION {
124            return Err(SegmentError::UnsupportedVersion { version });
125        }
126
127        let flags = u16::from_le_bytes([data[6], data[7]]);
128        let created_at = i64::from_le_bytes(to_array!(data[8..16], "header created_at")?);
129        let min_timestamp = i64::from_le_bytes(to_array!(data[16..24], "header min_timestamp")?);
130        let max_timestamp = i64::from_le_bytes(to_array!(data[24..32], "header max_timestamp")?);
131        let row_count = u64::from_le_bytes(to_array!(data[32..40], "header row_count")?);
132        let column_count = u16::from_le_bytes([data[40], data[41]]);
133        let series_count = u32::from_le_bytes(to_array!(data[42..46], "header series_count")?);
134        let compression = data[46];
135        let sort_order = data[47];
136
137        Ok(Self {
138            version,
139            flags,
140            created_at,
141            min_timestamp,
142            max_timestamp,
143            row_count,
144            column_count,
145            series_count,
146            compression,
147            sort_order,
148        })
149    }
150}
151
152/// Segment file footer.
153#[derive(Debug, Clone, PartialEq, Eq)]
154pub struct SegmentFooter {
155    /// Byte offset where column metadata section begins.
156    pub metadata_offset: u64,
157    /// Number of row groups in the segment.
158    pub row_group_count: u32,
159    /// `CRC32c` checksum of the entire file excluding this footer.
160    pub checksum: u32,
161    /// Independent `CRC32c` checksum of the metadata section.
162    /// Allows metadata validation without re-checksumming the entire file.
163    pub metadata_checksum: u32,
164}
165
166impl SegmentFooter {
167    /// Serialize the footer to bytes (always [`FOOTER_SIZE`] bytes).
168    #[must_use]
169    pub fn to_bytes(&self) -> Vec<u8> {
170        let mut buf = Vec::with_capacity(FOOTER_SIZE);
171        buf.extend_from_slice(&self.metadata_offset.to_le_bytes());
172        buf.extend_from_slice(&self.row_group_count.to_le_bytes());
173        buf.extend_from_slice(&self.checksum.to_le_bytes());
174        buf.extend_from_slice(&self.metadata_checksum.to_le_bytes());
175        buf.extend_from_slice(&MAGIC);
176        buf
177    }
178
179    /// Deserialize a footer from bytes.
180    ///
181    /// # Errors
182    ///
183    /// Returns [`SegmentError::InvalidMagic`] if trailing magic bytes don't
184    /// match, or [`SegmentError::CorruptFile`] if the data is too short.
185    pub fn from_bytes(data: &[u8]) -> Result<Self> {
186        if data.len() < FOOTER_SIZE {
187            return Err(SegmentError::CorruptFile {
188                detail: format!(
189                    "footer too short: expected {FOOTER_SIZE}, got {}",
190                    data.len()
191                ),
192            });
193        }
194
195        let metadata_offset = u64::from_le_bytes(to_array!(data[..8], "footer metadata_offset")?);
196        let row_group_count = u32::from_le_bytes(to_array!(data[8..12], "footer row_group_count")?);
197        let checksum = u32::from_le_bytes(to_array!(data[12..16], "footer checksum")?);
198        let metadata_checksum =
199            u32::from_le_bytes(to_array!(data[16..20], "footer metadata_checksum")?);
200
201        if data[20..24] != MAGIC {
202            return Err(SegmentError::InvalidMagic);
203        }
204
205        Ok(Self {
206            metadata_offset,
207            row_group_count,
208            checksum,
209            metadata_checksum,
210        })
211    }
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217
218    #[test]
219    fn header_roundtrip() {
220        let header = SegmentHeader {
221            version: VERSION,
222            flags: 0,
223            created_at: 1_700_000_000_000_000_000,
224            min_timestamp: 1_000_000,
225            max_timestamp: 2_000_000,
226            row_count: 10_000,
227            column_count: 5,
228            series_count: 42,
229            compression: 0,
230            sort_order: 1,
231        };
232        let bytes = header.to_bytes();
233        assert_eq!(bytes.len(), HEADER_SIZE);
234        let recovered = SegmentHeader::from_bytes(&bytes).unwrap();
235        assert_eq!(header, recovered);
236    }
237
238    #[test]
239    fn header_invalid_magic() {
240        let mut bytes = SegmentHeader {
241            version: VERSION,
242            flags: 0,
243            created_at: 0,
244            min_timestamp: 0,
245            max_timestamp: 0,
246            row_count: 0,
247            column_count: 0,
248            series_count: 0,
249            compression: 0,
250            sort_order: 0,
251        }
252        .to_bytes();
253        bytes[0] = b'X';
254        assert!(matches!(
255            SegmentHeader::from_bytes(&bytes),
256            Err(SegmentError::InvalidMagic)
257        ));
258    }
259
260    #[test]
261    fn footer_roundtrip() {
262        let footer = SegmentFooter {
263            metadata_offset: 1024,
264            row_group_count: 3,
265            checksum: 0xDEAD_BEEF,
266            metadata_checksum: 0xCAFE_BABE,
267        };
268        let bytes = footer.to_bytes();
269        assert_eq!(bytes.len(), FOOTER_SIZE);
270        let recovered = SegmentFooter::from_bytes(&bytes).unwrap();
271        assert_eq!(footer, recovered);
272    }
273
274    #[test]
275    fn footer_invalid_magic() {
276        let mut bytes = SegmentFooter {
277            metadata_offset: 0,
278            row_group_count: 0,
279            checksum: 0,
280            metadata_checksum: 0,
281        }
282        .to_bytes();
283        let len = bytes.len();
284        bytes[len - 1] = b'X';
285        assert!(matches!(
286            SegmentFooter::from_bytes(&bytes),
287            Err(SegmentError::InvalidMagic)
288        ));
289    }
290
291    #[test]
292    fn header_too_short() {
293        assert!(SegmentHeader::from_bytes(&[0; 10]).is_err());
294    }
295
296    #[test]
297    fn footer_too_short() {
298        assert!(SegmentFooter::from_bytes(&[0; 5]).is_err());
299    }
300
301    /// The version check is equality, so it refuses in **both** directions.
302    ///
303    /// Two layouts can differ in a way no tolerant reader bridges — one that
304    /// names the time column `timestamp` against one that names it `_time`
305    /// returns a batch whose time column nothing looks for, which is no rows
306    /// and no error. Below `VERSION` that is 0, which is also what a zeroed
307    /// or truncated header reads as; far above it stands for a file from a
308    /// future Chronix.
309    ///
310    /// This replaced four tests that asked the same question with four
311    /// different wrong values.
312    #[test]
313    fn header_version_is_refused_in_both_directions() {
314        let header = SegmentHeader {
315            version: VERSION,
316            flags: 0,
317            created_at: 0,
318            min_timestamp: 0,
319            max_timestamp: 0,
320            row_count: 0,
321            column_count: 0,
322            series_count: 0,
323            compression: 0,
324            sort_order: 0,
325        };
326        for wrong in [VERSION - 1, VERSION + 1, 999] {
327            let mut bytes = header.to_bytes();
328            bytes[4..6].copy_from_slice(&wrong.to_le_bytes());
329            assert!(
330                matches!(
331                    SegmentHeader::from_bytes(&bytes),
332                    Err(SegmentError::UnsupportedVersion { .. })
333                ),
334                "version {wrong} must be refused, not read as {VERSION}"
335            );
336        }
337    }
338}
chronix_engine/segment/header.rs

chronix_engine/segment/
header.rs
