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 = crate::format::SEGMENT_FORMAT_VERSION;
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 {
125                version,
126                detail: crate::format::version_mismatch(
127                    "this `.csx` segment",
128                    u64::from(version),
129                    u64::from(VERSION),
130                ),
131            });
132        }
133
134        let flags = u16::from_le_bytes([data[6], data[7]]);
135        let created_at = i64::from_le_bytes(to_array!(data[8..16], "header created_at")?);
136        let min_timestamp = i64::from_le_bytes(to_array!(data[16..24], "header min_timestamp")?);
137        let max_timestamp = i64::from_le_bytes(to_array!(data[24..32], "header max_timestamp")?);
138        let row_count = u64::from_le_bytes(to_array!(data[32..40], "header row_count")?);
139        let column_count = u16::from_le_bytes([data[40], data[41]]);
140        let series_count = u32::from_le_bytes(to_array!(data[42..46], "header series_count")?);
141        let compression = data[46];
142        let sort_order = data[47];
143
144        Ok(Self {
145            version,
146            flags,
147            created_at,
148            min_timestamp,
149            max_timestamp,
150            row_count,
151            column_count,
152            series_count,
153            compression,
154            sort_order,
155        })
156    }
157}
158
159/// Segment file footer.
160#[derive(Debug, Clone, PartialEq, Eq)]
161pub struct SegmentFooter {
162    /// Byte offset where column metadata section begins.
163    pub metadata_offset: u64,
164    /// Number of row groups in the segment.
165    pub row_group_count: u32,
166    /// `CRC32c` checksum of the entire file excluding this footer.
167    pub checksum: u32,
168    /// Independent `CRC32c` checksum of the metadata section.
169    /// Allows metadata validation without re-checksumming the entire file.
170    pub metadata_checksum: u32,
171}
172
173impl SegmentFooter {
174    /// Serialize the footer to bytes (always [`FOOTER_SIZE`] bytes).
175    #[must_use]
176    pub fn to_bytes(&self) -> Vec<u8> {
177        let mut buf = Vec::with_capacity(FOOTER_SIZE);
178        buf.extend_from_slice(&self.metadata_offset.to_le_bytes());
179        buf.extend_from_slice(&self.row_group_count.to_le_bytes());
180        buf.extend_from_slice(&self.checksum.to_le_bytes());
181        buf.extend_from_slice(&self.metadata_checksum.to_le_bytes());
182        buf.extend_from_slice(&MAGIC);
183        buf
184    }
185
186    /// Deserialize a footer from bytes.
187    ///
188    /// # Errors
189    ///
190    /// Returns [`SegmentError::InvalidMagic`] if trailing magic bytes don't
191    /// match, or [`SegmentError::CorruptFile`] if the data is too short.
192    pub fn from_bytes(data: &[u8]) -> Result<Self> {
193        if data.len() < FOOTER_SIZE {
194            return Err(SegmentError::CorruptFile {
195                detail: format!(
196                    "footer too short: expected {FOOTER_SIZE}, got {}",
197                    data.len()
198                ),
199            });
200        }
201
202        let metadata_offset = u64::from_le_bytes(to_array!(data[..8], "footer metadata_offset")?);
203        let row_group_count = u32::from_le_bytes(to_array!(data[8..12], "footer row_group_count")?);
204        let checksum = u32::from_le_bytes(to_array!(data[12..16], "footer checksum")?);
205        let metadata_checksum =
206            u32::from_le_bytes(to_array!(data[16..20], "footer metadata_checksum")?);
207
208        if data[20..24] != MAGIC {
209            return Err(SegmentError::InvalidMagic);
210        }
211
212        Ok(Self {
213            metadata_offset,
214            row_group_count,
215            checksum,
216            metadata_checksum,
217        })
218    }
219}
220
221#[cfg(test)]
222mod tests {
223    use super::*;
224
225    #[test]
226    fn header_roundtrip() {
227        let header = SegmentHeader {
228            version: VERSION,
229            flags: 0,
230            created_at: 1_700_000_000_000_000_000,
231            min_timestamp: 1_000_000,
232            max_timestamp: 2_000_000,
233            row_count: 10_000,
234            column_count: 5,
235            series_count: 42,
236            compression: 0,
237            sort_order: 1,
238        };
239        let bytes = header.to_bytes();
240        assert_eq!(bytes.len(), HEADER_SIZE);
241        let recovered = SegmentHeader::from_bytes(&bytes).unwrap();
242        assert_eq!(header, recovered);
243    }
244
245    #[test]
246    fn header_invalid_magic() {
247        let mut bytes = SegmentHeader {
248            version: VERSION,
249            flags: 0,
250            created_at: 0,
251            min_timestamp: 0,
252            max_timestamp: 0,
253            row_count: 0,
254            column_count: 0,
255            series_count: 0,
256            compression: 0,
257            sort_order: 0,
258        }
259        .to_bytes();
260        bytes[0] = b'X';
261        assert!(matches!(
262            SegmentHeader::from_bytes(&bytes),
263            Err(SegmentError::InvalidMagic)
264        ));
265    }
266
267    #[test]
268    fn footer_roundtrip() {
269        let footer = SegmentFooter {
270            metadata_offset: 1024,
271            row_group_count: 3,
272            checksum: 0xDEAD_BEEF,
273            metadata_checksum: 0xCAFE_BABE,
274        };
275        let bytes = footer.to_bytes();
276        assert_eq!(bytes.len(), FOOTER_SIZE);
277        let recovered = SegmentFooter::from_bytes(&bytes).unwrap();
278        assert_eq!(footer, recovered);
279    }
280
281    #[test]
282    fn footer_invalid_magic() {
283        let mut bytes = SegmentFooter {
284            metadata_offset: 0,
285            row_group_count: 0,
286            checksum: 0,
287            metadata_checksum: 0,
288        }
289        .to_bytes();
290        let len = bytes.len();
291        bytes[len - 1] = b'X';
292        assert!(matches!(
293            SegmentFooter::from_bytes(&bytes),
294            Err(SegmentError::InvalidMagic)
295        ));
296    }
297
298    #[test]
299    fn header_too_short() {
300        assert!(SegmentHeader::from_bytes(&[0; 10]).is_err());
301    }
302
303    #[test]
304    fn footer_too_short() {
305        assert!(SegmentFooter::from_bytes(&[0; 5]).is_err());
306    }
307
308    /// The version check is equality, so it refuses in **both** directions.
309    ///
310    /// Two layouts can differ in a way no tolerant reader bridges — one that
311    /// names the time column `timestamp` against one that names it `_time`
312    /// returns a batch whose time column nothing looks for, which is no rows
313    /// and no error. Below `VERSION` that is 0, which is also what a zeroed
314    /// or truncated header reads as; far above it stands for a file from a
315    /// future Chronix.
316    ///
317    /// This replaced four tests that asked the same question with four
318    /// different wrong values.
319    #[test]
320    fn header_version_is_refused_in_both_directions() {
321        let header = SegmentHeader {
322            version: VERSION,
323            flags: 0,
324            created_at: 0,
325            min_timestamp: 0,
326            max_timestamp: 0,
327            row_count: 0,
328            column_count: 0,
329            series_count: 0,
330            compression: 0,
331            sort_order: 0,
332        };
333        for wrong in [VERSION - 1, VERSION + 1, 999] {
334            let mut bytes = header.to_bytes();
335            bytes[4..6].copy_from_slice(&wrong.to_le_bytes());
336            assert!(
337                matches!(
338                    SegmentHeader::from_bytes(&bytes),
339                    Err(SegmentError::UnsupportedVersion { .. })
340                ),
341                "version {wrong} must be refused, not read as {VERSION}"
342            );
343        }
344    }
345}
chronix_engine/segment/header.rs

chronix_engine/segment/
header.rs
