quill_sql/storage/page/
table_page.rs

1use crate::buffer::{PageId, INVALID_PAGE_ID, PAGE_SIZE};
2use crate::catalog::SchemaRef;
3use crate::error::{QuillSQLError, QuillSQLResult};
4use crate::recovery::Lsn;
5use crate::storage::codec::{TablePageHeaderCodec, TablePageHeaderTupleInfoCodec, TupleCodec};
6use crate::storage::tuple::Tuple;
7use crate::transaction::{CommandId, TransactionId, INVALID_COMMAND_ID};
8use std::fmt::{Display, Formatter};
9use std::sync::LazyLock;
10
11pub static EMPTY_TUPLE_META: TupleMeta = TupleMeta {
12    insert_txn_id: 0,
13    insert_cid: 0,
14    delete_txn_id: 0,
15    delete_cid: INVALID_COMMAND_ID,
16    is_deleted: false,
17    next_version: None,
18    prev_version: None,
19};
20
21pub static EMPTY_TUPLE_INFO: LazyLock<TupleInfo> = LazyLock::new(|| TupleInfo {
22    offset: 0,
23    size: 0,
24    meta: EMPTY_TUPLE_META,
25});
26
27/**
28 * Slotted page format:
29 * ```text
30 *  ---------------------------------------------------------
31 *  | HEADER | ... FREE SPACE ... | ... INSERTED TUPLES ... |
32 *  ---------------------------------------------------------
33 *                                ^
34 *                                free space pointer
35 * ```
36 *
37 * Header format (size in bytes):
38 * ```text
39 *  --------------------------------------------------------------------------------
40 *  | LSN (8) | NextPageId (4) | NumTuples(2) | NumDeletedTuples(2) |
41 *  --------------------------------------------------------------------------------
42 *  ----------------------------------------------------------------
43 *  | Tuple_1 offset+size + TupleMeta | Tuple_2 offset+size + TupleMeta | ... |
44 *  ----------------------------------------------------------------
45 * ```
46 */
47#[derive(Debug, Clone, Eq, PartialEq)]
48pub struct TablePage {
49    pub schema: SchemaRef,
50    pub header: TablePageHeader,
51    // 整个页原始数据
52    pub data: [u8; PAGE_SIZE],
53}
54
55#[derive(Debug, Clone, Eq, PartialEq)]
56pub struct TablePageHeader {
57    pub next_page_id: PageId,
58    pub num_tuples: u16,
59    pub num_deleted_tuples: u16,
60    pub tuple_infos: Vec<TupleInfo>,
61    pub lsn: Lsn,
62}
63
64#[derive(Debug, Clone, Eq, PartialEq)]
65pub struct TupleInfo {
66    pub offset: u16,
67    pub size: u16,
68    pub meta: TupleMeta,
69}
70
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub struct TupleMeta {
73    pub insert_txn_id: TransactionId,
74    pub insert_cid: CommandId,
75    pub delete_txn_id: TransactionId,
76    pub delete_cid: CommandId,
77    pub is_deleted: bool,
78    pub next_version: Option<RecordId>,
79    pub prev_version: Option<RecordId>,
80}
81
82impl TupleMeta {
83    pub fn new(insert_txn_id: TransactionId, insert_cid: CommandId) -> Self {
84        Self {
85            insert_txn_id,
86            insert_cid,
87            delete_txn_id: 0,
88            delete_cid: INVALID_COMMAND_ID,
89            is_deleted: false,
90            next_version: None,
91            prev_version: None,
92        }
93    }
94
95    pub fn mark_deleted(&mut self, txn_id: TransactionId, delete_cid: CommandId) {
96        self.is_deleted = true;
97        self.delete_txn_id = txn_id;
98        self.delete_cid = delete_cid;
99    }
100
101    pub fn clear_delete(&mut self) {
102        self.is_deleted = false;
103        self.delete_txn_id = 0;
104        self.delete_cid = INVALID_COMMAND_ID;
105    }
106
107    pub fn set_next_version(&mut self, next: Option<RecordId>) {
108        self.next_version = next;
109    }
110
111    pub fn set_prev_version(&mut self, prev: Option<RecordId>) {
112        self.prev_version = prev;
113    }
114
115    pub fn clear_chain(&mut self) {
116        self.next_version = None;
117        self.prev_version = None;
118    }
119}
120
121pub const INVALID_RID: RecordId = RecordId {
122    page_id: INVALID_PAGE_ID,
123    slot_num: 0,
124};
125
126#[derive(derive_new::new, Debug, Clone, Copy, PartialEq, Eq, Hash)]
127pub struct RecordId {
128    pub page_id: PageId,
129    pub slot_num: u32,
130}
131
132impl Display for RecordId {
133    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
134        write!(f, "{}-{}", self.page_id, self.slot_num)
135    }
136}
137
138impl TablePage {
139    pub fn new(schema: SchemaRef, next_page_id: PageId) -> Self {
140        Self {
141            schema,
142            header: TablePageHeader {
143                next_page_id,
144                num_tuples: 0,
145                num_deleted_tuples: 0,
146                tuple_infos: Vec::new(),
147                lsn: 0,
148            },
149            data: [0; PAGE_SIZE],
150        }
151    }
152
153    // Get the offset for the next tuple insertion.
154    pub fn next_tuple_offset(&self, tuple: &Tuple) -> QuillSQLResult<usize> {
155        let tuple_bytes = TupleCodec::encode(tuple);
156        self.next_tuple_offset_with_len(tuple_bytes.len())
157    }
158
159    pub fn next_tuple_offset_with_len(&self, tuple_len: usize) -> QuillSQLResult<usize> {
160        // Get the ending offset of the current slot. If there are inserted tuples,
161        // get the offset of the previous inserted tuple; otherwise, set it to the size of the page.
162        let slot_end_offset = if self.header.num_tuples > 0 {
163            self.header.tuple_infos[self.header.num_tuples as usize - 1].offset as usize
164        } else {
165            PAGE_SIZE
166        };
167
168        // Check if the current slot has enough space for the new tuple. Return None if not.
169        if slot_end_offset < tuple_len {
170            return Err(QuillSQLError::Storage(
171                "No enough space to store tuple".to_string(),
172            ));
173        }
174
175        // Calculate the insertion offset for the new tuple by subtracting its data length
176        // from the ending offset of the current slot.
177        let tuple_offset = slot_end_offset - tuple_len;
178
179        // Calculate the minimum valid tuple insertion offset, including the table page header size,
180        // the total size of each tuple info (existing tuple infos and newly added tuple info).
181        let min_tuple_offset = TablePageHeaderCodec::encode(&self.header).len()
182            + TablePageHeaderTupleInfoCodec::encode(&EMPTY_TUPLE_INFO).len();
183        if tuple_offset < min_tuple_offset {
184            return Err(QuillSQLError::Storage(
185                "No enough space to store tuple".to_string(),
186            ));
187        }
188
189        // Return the calculated insertion offset for the new tuple.
190        Ok(tuple_offset)
191    }
192
193    pub fn insert_tuple(&mut self, meta: &TupleMeta, tuple: &Tuple) -> QuillSQLResult<u16> {
194        let tuple_bytes = TupleCodec::encode(tuple);
195        self.insert_tuple_bytes(meta, &tuple_bytes)
196    }
197
198    pub fn insert_tuple_bytes(
199        &mut self,
200        meta: &TupleMeta,
201        tuple_bytes: &[u8],
202    ) -> QuillSQLResult<u16> {
203        // Get the offset for the next tuple insertion.
204        let tuple_offset = self.next_tuple_offset_with_len(tuple_bytes.len())?;
205        let tuple_id = self.header.num_tuples;
206        debug_assert!(tuple_bytes.len() < u16::MAX as usize);
207
208        // Store tuple information including offset, length, and metadata.
209        self.header.tuple_infos.push(TupleInfo {
210            offset: tuple_offset as u16,
211            size: tuple_bytes.len() as u16,
212            meta: *meta,
213        });
214
215        // only check
216        assert_eq!(tuple_id, self.header.tuple_infos.len() as u16 - 1);
217
218        self.header.num_tuples += 1;
219        if meta.is_deleted {
220            self.header.num_deleted_tuples += 1;
221        }
222
223        // Copy the tuple's data into the appropriate position within the page's data buffer.
224        self.data[tuple_offset..tuple_offset + tuple_bytes.len()].copy_from_slice(&tuple_bytes);
225        Ok(tuple_id)
226    }
227
228    pub fn update_tuple_meta(&mut self, meta: TupleMeta, slot_num: u16) -> QuillSQLResult<()> {
229        if slot_num >= self.header.num_tuples {
230            return Err(QuillSQLError::Storage(format!(
231                "tuple_id {} out of range",
232                slot_num
233            )));
234        }
235        if meta.is_deleted && !self.header.tuple_infos[slot_num as usize].meta.is_deleted {
236            self.header.num_deleted_tuples += 1;
237        }
238
239        self.header.tuple_infos[slot_num as usize].meta = meta;
240        Ok(())
241    }
242
243    pub fn update_tuple(&mut self, tuple: Tuple, slot_num: u16) -> QuillSQLResult<()> {
244        if slot_num >= self.header.num_tuples {
245            return Err(QuillSQLError::Storage(format!(
246                "tuple_id {} out of range",
247                slot_num
248            )));
249        }
250        let offset = self.header.tuple_infos[slot_num as usize].offset as usize;
251        let size = self.header.tuple_infos[slot_num as usize].size as usize;
252        let tuple_bytes = TupleCodec::encode(&tuple);
253        if tuple_bytes.len() == size {
254            self.data[offset..(offset + size)].copy_from_slice(&tuple_bytes);
255        } else {
256            // need move other tuples
257            let mut full_tuples = vec![];
258            for info in self.header.tuple_infos.iter() {
259                full_tuples.push((
260                    info.meta,
261                    TupleCodec::decode(
262                        &self.data[info.offset as usize..(info.offset + info.size) as usize],
263                        self.schema.clone(),
264                    )?
265                    .0,
266                ));
267            }
268            full_tuples[slot_num as usize].1 = tuple;
269
270            let mut new_page = TablePage::new(self.schema.clone(), self.header.next_page_id);
271            for (meta, tuple) in full_tuples.iter() {
272                new_page.insert_tuple(meta, tuple)?;
273            }
274            self.header = new_page.header;
275            self.data = new_page.data;
276        }
277        Ok(())
278    }
279
280    /// Remove the tuple at `slot_num` and compact remaining tuples.
281    pub fn reclaim_tuple(&mut self, slot_num: u16) -> QuillSQLResult<()> {
282        if slot_num >= self.header.num_tuples {
283            return Err(QuillSQLError::Storage(format!(
284                "tuple_id {} out of range",
285                slot_num
286            )));
287        }
288
289        let snapshot_infos = self.header.tuple_infos.clone();
290        let next_page_id = self.header.next_page_id;
291        let lsn = self.header.lsn;
292
293        let mut rebuilt = TablePage::new(self.schema.clone(), next_page_id);
294        rebuilt.set_lsn(lsn);
295
296        for (idx, info) in snapshot_infos.into_iter().enumerate() {
297            if idx == slot_num as usize {
298                continue;
299            }
300            let (tuple, _) = TupleCodec::decode(
301                &self.data[info.offset as usize..(info.offset + info.size) as usize],
302                self.schema.clone(),
303            )?;
304            rebuilt.insert_tuple(&info.meta, &tuple)?;
305        }
306
307        // Preserve LSN and next pointer while replacing storage.
308        rebuilt.header.lsn = lsn;
309        self.header = rebuilt.header;
310        self.data = rebuilt.data;
311        Ok(())
312    }
313
314    pub fn tuple(&self, slot_num: u16) -> QuillSQLResult<(TupleMeta, Tuple)> {
315        if slot_num >= self.header.num_tuples {
316            return Err(QuillSQLError::Storage(format!(
317                "tuple_id {} out of range",
318                slot_num
319            )));
320        }
321
322        let offset = self.header.tuple_infos[slot_num as usize].offset;
323        let size = self.header.tuple_infos[slot_num as usize].size;
324        let meta = self.header.tuple_infos[slot_num as usize].meta;
325        let (tuple, _) = TupleCodec::decode(
326            &self.data[offset as usize..(offset + size) as usize],
327            self.schema.clone(),
328        )?;
329
330        Ok((meta, tuple))
331    }
332
333    pub fn tuple_meta(&self, slot_num: u16) -> QuillSQLResult<TupleMeta> {
334        if slot_num >= self.header.num_tuples {
335            return Err(QuillSQLError::Storage(format!(
336                "tuple_id {} out of range",
337                slot_num
338            )));
339        }
340
341        Ok(self.header.tuple_infos[slot_num as usize].meta)
342    }
343
344    pub fn get_next_rid(&self, rid: &RecordId) -> Option<RecordId> {
345        let next_slot = rid.slot_num + 1;
346        if next_slot < self.header.num_tuples as u32 {
347            Some(RecordId::new(rid.page_id, next_slot))
348        } else {
349            None
350        }
351    }
352}
353
354impl TablePage {
355    pub fn set_lsn(&mut self, lsn: Lsn) {
356        self.header.lsn = lsn;
357    }
358
359    pub fn lsn(&self) -> Lsn {
360        self.header.lsn
361    }
362}
363
364#[cfg(test)]
365mod tests {
366    use crate::catalog::{Column, DataType, Schema};
367    use crate::storage::page::EMPTY_TUPLE_META;
368    use crate::storage::tuple::Tuple;
369    use std::sync::Arc;
370
371    #[test]
372    pub fn test_table_page_get_tuple() {
373        let schema = Arc::new(Schema::new(vec![
374            Column::new("a", DataType::Int8, false),
375            Column::new("b", DataType::Int16, false),
376        ]));
377        let mut table_page = super::TablePage::new(schema.clone(), 0);
378        let tuple_id = table_page
379            .insert_tuple(
380                &EMPTY_TUPLE_META,
381                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
382            )
383            .unwrap();
384        assert_eq!(tuple_id, 0);
385        let tuple_id = table_page
386            .insert_tuple(
387                &EMPTY_TUPLE_META,
388                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
389            )
390            .unwrap();
391        assert_eq!(tuple_id, 1);
392        let tuple_id = table_page
393            .insert_tuple(
394                &EMPTY_TUPLE_META,
395                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
396            )
397            .unwrap();
398        assert_eq!(tuple_id, 2);
399
400        let (tuple_meta, tuple) = table_page.tuple(0).unwrap();
401        assert_eq!(tuple_meta, EMPTY_TUPLE_META);
402        assert_eq!(tuple.data, vec![1i8.into(), 1i16.into()]);
403        let (_tuple_meta, tuple) = table_page.tuple(1).unwrap();
404        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
405        let (_tuple_meta, tuple) = table_page.tuple(2).unwrap();
406        assert_eq!(tuple.data, vec![3i8.into(), 3i16.into()]);
407    }
408
409    #[test]
410    pub fn test_table_page_update_tuple_meta() {
411        let schema = Arc::new(Schema::new(vec![
412            Column::new("a", DataType::Int8, false),
413            Column::new("b", DataType::Int16, false),
414        ]));
415        let mut table_page = super::TablePage::new(schema.clone(), 0);
416        let _tuple_id = table_page
417            .insert_tuple(
418                &EMPTY_TUPLE_META,
419                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
420            )
421            .unwrap();
422        let _tuple_id = table_page
423            .insert_tuple(
424                &EMPTY_TUPLE_META,
425                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
426            )
427            .unwrap();
428        let _tuple_id = table_page
429            .insert_tuple(
430                &EMPTY_TUPLE_META,
431                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
432            )
433            .unwrap();
434
435        let mut tuple_meta = table_page.tuple_meta(0).unwrap();
436        tuple_meta.mark_deleted(1, 0);
437        tuple_meta.insert_txn_id = 2;
438        tuple_meta.insert_cid = 1;
439
440        table_page.update_tuple_meta(tuple_meta, 0).unwrap();
441        let tuple_meta = table_page.tuple_meta(0).unwrap();
442        assert!(tuple_meta.is_deleted);
443        assert_eq!(tuple_meta.delete_txn_id, 1);
444        assert_eq!(tuple_meta.delete_cid, 0);
445        assert_eq!(tuple_meta.insert_txn_id, 2);
446        assert_eq!(tuple_meta.insert_cid, 1);
447    }
448
449    #[test]
450    fn reclaim_tuple_removes_slot_and_compacts() {
451        let schema = Arc::new(Schema::new(vec![Column::new("id", DataType::Int32, false)]));
452        let mut table_page = super::TablePage::new(schema.clone(), 0);
453        let tuple1 = Tuple::new(schema.clone(), vec![1i32.into()]);
454        let tuple2 = Tuple::new(schema.clone(), vec![2i32.into()]);
455        let tuple3 = Tuple::new(schema.clone(), vec![3i32.into()]);
456
457        table_page
458            .insert_tuple(&super::TupleMeta::new(1, 0), &tuple1)
459            .unwrap();
460        table_page
461            .insert_tuple(&super::TupleMeta::new(2, 0), &tuple2)
462            .unwrap();
463        table_page
464            .insert_tuple(&super::TupleMeta::new(3, 0), &tuple3)
465            .unwrap();
466
467        table_page.reclaim_tuple(1).unwrap();
468        assert_eq!(table_page.header.num_tuples, 2);
469
470        let (meta0, t0) = table_page.tuple(0).unwrap();
471        assert_eq!(meta0.insert_txn_id, 1);
472        assert_eq!(t0, tuple1);
473
474        let (meta1, t1) = table_page.tuple(1).unwrap();
475        assert_eq!(meta1.insert_txn_id, 3);
476        assert_eq!(t1, tuple3);
477    }
478}
quill_sql/storage/page/table_page.rs

quill_sql/storage/page/
table_page.rs
