mq_db/storage/
catalog.rs

1use std::collections::HashSet;
2
3use crate::{
4    error::MqdbError,
5    storage::page::{
6        PAGE_BODY_SIZE, PAGE_HEADER_SIZE, PAGE_TYPE_CATALOG, PageFile, make_page, parse_page_header,
7    },
8};
9
10#[derive(Debug, Clone, PartialEq)]
11pub struct CatalogEntry {
12    pub document_id: u32,
13    pub path: Option<String>,
14    pub first_block_page: u32,
15    pub num_blocks: u32,
16    pub zone_map_bytes: Vec<u8>,
17    /// First page of the persisted secondary index chain. 0 = not stored.
18    pub index_start_page: u32,
19}
20
21/// A user-defined table entry stored in the catalog.
22///
23/// Row data is *not* stored inline — it lives in its own page chain
24/// (see [`crate::storage::Storage::write_table_rows`]) so that appending
25/// rows only requires writing the new pages plus this small fixed-size
26/// entry, instead of rewriting every previously-inserted row.
27#[derive(Debug, Clone, PartialEq)]
28pub struct CustomTableEntry {
29    pub name: String,
30    pub columns: Vec<String>,
31    /// First page of the row chain. 0 = no rows persisted yet.
32    pub first_row_page: u32,
33    /// Last page of the row chain — new rows are appended after this page.
34    pub last_row_page: u32,
35    pub num_rows: u32,
36}
37
38fn invalid_data(message: impl Into<String>) -> MqdbError {
39    MqdbError::Storage(message.into())
40}
41
42fn as_u16(value: usize, field: &str) -> u16 {
43    u16::try_from(value).unwrap_or_else(|_| panic!("{field} exceeds u16 range"))
44}
45
46fn as_u32(value: usize, field: &str) -> u32 {
47    u32::try_from(value).unwrap_or_else(|_| panic!("{field} exceeds u32 range"))
48}
49
50struct Decoder<'a> {
51    data: &'a [u8],
52    pos: usize,
53}
54
55impl<'a> Decoder<'a> {
56    fn new(data: &'a [u8]) -> Self {
57        Self { data, pos: 0 }
58    }
59
60    fn read_exact(&mut self, len: usize) -> Result<&'a [u8], MqdbError> {
61        let end = self
62            .pos
63            .checked_add(len)
64            .ok_or_else(|| invalid_data("byte offset overflow"))?;
65        if end > self.data.len() {
66            return Err(invalid_data("unexpected end of catalog data"));
67        }
68        let bytes = &self.data[self.pos..end];
69        self.pos = end;
70        Ok(bytes)
71    }
72
73    fn read_u8(&mut self) -> Result<u8, MqdbError> {
74        Ok(self.read_exact(1)?[0])
75    }
76
77    fn read_u16(&mut self) -> Result<u16, MqdbError> {
78        let bytes: [u8; 2] = self
79            .read_exact(2)?
80            .try_into()
81            .map_err(|_| invalid_data("failed to read u16"))?;
82        Ok(u16::from_le_bytes(bytes))
83    }
84
85    fn read_u32(&mut self) -> Result<u32, MqdbError> {
86        let bytes: [u8; 4] = self
87            .read_exact(4)?
88            .try_into()
89            .map_err(|_| invalid_data("failed to read u32"))?;
90        Ok(u32::from_le_bytes(bytes))
91    }
92
93    fn read_string_u16(&mut self) -> Result<String, MqdbError> {
94        let len = usize::from(self.read_u16()?);
95        let bytes = self.read_exact(len)?;
96        String::from_utf8(bytes.to_vec())
97            .map_err(|e| invalid_data(format!("invalid catalog string UTF-8: {e}")))
98    }
99
100    fn remaining(&self) -> usize {
101        self.data.len() - self.pos
102    }
103}
104
105fn serialize_catalog(entries: &[CatalogEntry], custom_tables: &[CustomTableEntry]) -> Vec<u8> {
106    let mut out = Vec::new();
107    out.extend_from_slice(&as_u32(entries.len(), "catalog entry count").to_le_bytes());
108
109    for entry in entries {
110        out.extend_from_slice(&entry.document_id.to_le_bytes());
111        match &entry.path {
112            Some(path) => {
113                out.push(1);
114                out.extend_from_slice(&as_u16(path.len(), "catalog path length").to_le_bytes());
115                out.extend_from_slice(path.as_bytes());
116            }
117            None => out.push(0),
118        }
119        out.extend_from_slice(&entry.first_block_page.to_le_bytes());
120        out.extend_from_slice(&entry.num_blocks.to_le_bytes());
121        out.extend_from_slice(&as_u32(entry.zone_map_bytes.len(), "zone map length").to_le_bytes());
122        out.extend_from_slice(&entry.zone_map_bytes);
123        out.extend_from_slice(&entry.index_start_page.to_le_bytes());
124    }
125
126    out.extend_from_slice(&as_u32(custom_tables.len(), "custom table count").to_le_bytes());
127    for ct in custom_tables {
128        out.extend_from_slice(&as_u16(ct.name.len(), "table name length").to_le_bytes());
129        out.extend_from_slice(ct.name.as_bytes());
130        out.extend_from_slice(&as_u16(ct.columns.len(), "column count").to_le_bytes());
131        for col in &ct.columns {
132            out.extend_from_slice(&as_u16(col.len(), "column name length").to_le_bytes());
133            out.extend_from_slice(col.as_bytes());
134        }
135        out.extend_from_slice(&ct.first_row_page.to_le_bytes());
136        out.extend_from_slice(&ct.last_row_page.to_le_bytes());
137        out.extend_from_slice(&ct.num_rows.to_le_bytes());
138    }
139
140    out
141}
142
143pub fn write_catalog(
144    pf: &mut PageFile,
145    entries: &[CatalogEntry],
146    custom_tables: &[CustomTableEntry],
147) -> Result<(), MqdbError> {
148    if pf.num_pages < 2 {
149        return Err(invalid_data("catalog start page is missing"));
150    }
151
152    let bytes = serialize_catalog(entries, custom_tables);
153    let chunks: Vec<&[u8]> = if bytes.is_empty() {
154        vec![&[]]
155    } else {
156        bytes.chunks(PAGE_BODY_SIZE).collect()
157    };
158
159    let mut page_ids = Vec::with_capacity(chunks.len());
160    page_ids.push(1);
161
162    for _ in 1..chunks.len() {
163        let placeholder = make_page(PAGE_TYPE_CATALOG, 0, 0, &[]);
164        let page_id = pf.append_page(&placeholder)?;
165        page_ids.push(page_id);
166    }
167
168    for (index, chunk) in chunks.iter().enumerate() {
169        let page_id = page_ids[index];
170        let next_page = page_ids.get(index + 1).copied().unwrap_or(0);
171        let page = make_page(PAGE_TYPE_CATALOG, page_id, next_page, chunk);
172        pf.write_page(page_id, &page)?;
173    }
174
175    Ok(())
176}
177
178pub fn read_catalog(
179    pf: &mut PageFile,
180) -> Result<(Vec<CatalogEntry>, Vec<CustomTableEntry>), MqdbError> {
181    if pf.num_pages < 2 {
182        return Err(invalid_data("catalog start page is missing"));
183    }
184
185    let mut bytes = Vec::new();
186    let mut page_id = 1u32;
187    let mut visited = HashSet::new();
188
189    loop {
190        if !visited.insert(page_id) {
191            return Err(invalid_data("catalog page chain contains a cycle"));
192        }
193
194        let page = pf.read_page(page_id)?;
195        let (page_type, _, stored_page_id, next_page) = parse_page_header(&page);
196        if page_type != PAGE_TYPE_CATALOG {
197            return Err(invalid_data(format!(
198                "page {page_id} is not a catalog page"
199            )));
200        }
201        if stored_page_id != page_id {
202            return Err(invalid_data(format!(
203                "catalog page header mismatch: expected {page_id}, found {stored_page_id}"
204            )));
205        }
206
207        bytes.extend_from_slice(&page[PAGE_HEADER_SIZE..]);
208
209        if next_page == 0 {
210            break;
211        }
212        page_id = next_page;
213    }
214
215    let mut decoder = Decoder::new(&bytes);
216    let entry_count = usize::try_from(decoder.read_u32()?)
217        .map_err(|_| invalid_data("catalog entry count exceeds usize range"))?;
218    let mut entries = Vec::with_capacity(entry_count);
219
220    for _ in 0..entry_count {
221        let document_id = decoder.read_u32()?;
222        let path = match decoder.read_u8()? {
223            0 => None,
224            1 => Some(decoder.read_string_u16()?),
225            value => return Err(invalid_data(format!("invalid path presence tag: {value}"))),
226        };
227        let first_block_page = decoder.read_u32()?;
228        let num_blocks = decoder.read_u32()?;
229        let zone_map_len = usize::try_from(decoder.read_u32()?)
230            .map_err(|_| invalid_data("zone map length exceeds usize range"))?;
231        let zone_map_bytes = decoder.read_exact(zone_map_len)?.to_vec();
232
233        let index_start_page = decoder.read_u32()?;
234        entries.push(CatalogEntry {
235            document_id,
236            path,
237            first_block_page,
238            num_blocks,
239            zone_map_bytes,
240            index_start_page,
241        });
242    }
243
244    let custom_tables = if decoder.remaining() >= 4 {
245        let count = usize::try_from(decoder.read_u32()?)
246            .map_err(|_| invalid_data("custom table count exceeds usize range"))?;
247        let mut tables = Vec::with_capacity(count);
248        for _ in 0..count {
249            let name = decoder.read_string_u16()?;
250            let num_cols = usize::from(decoder.read_u16()?);
251            let mut columns = Vec::with_capacity(num_cols);
252            for _ in 0..num_cols {
253                columns.push(decoder.read_string_u16()?);
254            }
255            let first_row_page = decoder.read_u32()?;
256            let last_row_page = decoder.read_u32()?;
257            let num_rows = decoder.read_u32()?;
258            tables.push(CustomTableEntry {
259                name,
260                columns,
261                first_row_page,
262                last_row_page,
263                num_rows,
264            });
265        }
266        tables
267    } else {
268        vec![]
269    };
270
271    Ok((entries, custom_tables))
272}
mq_db/storage/catalog.rs

mq_db/storage/
catalog.rs
