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        // Get the ending offset of the current slot. If there are inserted tuples,
156        // get the offset of the previous inserted tuple; otherwise, set it to the size of the page.
157        let slot_end_offset = if self.header.num_tuples > 0 {
158            self.header.tuple_infos[self.header.num_tuples as usize - 1].offset as usize
159        } else {
160            PAGE_SIZE
161        };
162
163        // Check if the current slot has enough space for the new tuple. Return None if not.
164        if slot_end_offset < TupleCodec::encode(tuple).len() {
165            return Err(QuillSQLError::Storage(
166                "No enough space to store tuple".to_string(),
167            ));
168        }
169
170        // Calculate the insertion offset for the new tuple by subtracting its data length
171        // from the ending offset of the current slot.
172        let tuple_offset = slot_end_offset - TupleCodec::encode(tuple).len();
173
174        // Calculate the minimum valid tuple insertion offset, including the table page header size,
175        // the total size of each tuple info (existing tuple infos and newly added tuple info).
176        let min_tuple_offset = TablePageHeaderCodec::encode(&self.header).len()
177            + TablePageHeaderTupleInfoCodec::encode(&EMPTY_TUPLE_INFO).len();
178        if tuple_offset < min_tuple_offset {
179            return Err(QuillSQLError::Storage(
180                "No enough space to store tuple".to_string(),
181            ));
182        }
183
184        // Return the calculated insertion offset for the new tuple.
185        Ok(tuple_offset)
186    }
187
188    pub fn insert_tuple(&mut self, meta: &TupleMeta, tuple: &Tuple) -> QuillSQLResult<u16> {
189        // Get the offset for the next tuple insertion.
190        let tuple_offset = self.next_tuple_offset(tuple)?;
191        let tuple_id = self.header.num_tuples;
192        let tuple_bytes = TupleCodec::encode(tuple);
193        debug_assert!(tuple_bytes.len() < u16::MAX as usize);
194
195        // Store tuple information including offset, length, and metadata.
196        self.header.tuple_infos.push(TupleInfo {
197            offset: tuple_offset as u16,
198            size: tuple_bytes.len() as u16,
199            meta: *meta,
200        });
201
202        // only check
203        assert_eq!(tuple_id, self.header.tuple_infos.len() as u16 - 1);
204
205        self.header.num_tuples += 1;
206        if meta.is_deleted {
207            self.header.num_deleted_tuples += 1;
208        }
209
210        // Copy the tuple's data into the appropriate position within the page's data buffer.
211        self.data[tuple_offset..tuple_offset + tuple_bytes.len()].copy_from_slice(&tuple_bytes);
212        Ok(tuple_id)
213    }
214
215    pub fn update_tuple_meta(&mut self, meta: TupleMeta, slot_num: u16) -> QuillSQLResult<()> {
216        if slot_num >= self.header.num_tuples {
217            return Err(QuillSQLError::Storage(format!(
218                "tuple_id {} out of range",
219                slot_num
220            )));
221        }
222        if meta.is_deleted && !self.header.tuple_infos[slot_num as usize].meta.is_deleted {
223            self.header.num_deleted_tuples += 1;
224        }
225
226        self.header.tuple_infos[slot_num as usize].meta = meta;
227        Ok(())
228    }
229
230    pub fn update_tuple(&mut self, tuple: Tuple, slot_num: u16) -> QuillSQLResult<()> {
231        if slot_num >= self.header.num_tuples {
232            return Err(QuillSQLError::Storage(format!(
233                "tuple_id {} out of range",
234                slot_num
235            )));
236        }
237        let offset = self.header.tuple_infos[slot_num as usize].offset as usize;
238        let size = self.header.tuple_infos[slot_num as usize].size as usize;
239        let tuple_bytes = TupleCodec::encode(&tuple);
240        if tuple_bytes.len() == size {
241            self.data[offset..(offset + size)].copy_from_slice(&tuple_bytes);
242        } else {
243            // need move other tuples
244            let mut full_tuples = vec![];
245            for info in self.header.tuple_infos.iter() {
246                full_tuples.push((
247                    info.meta,
248                    TupleCodec::decode(
249                        &self.data[info.offset as usize..(info.offset + info.size) as usize],
250                        self.schema.clone(),
251                    )?
252                    .0,
253                ));
254            }
255            full_tuples[slot_num as usize].1 = tuple;
256
257            let mut new_page = TablePage::new(self.schema.clone(), self.header.next_page_id);
258            for (meta, tuple) in full_tuples.iter() {
259                new_page.insert_tuple(meta, tuple)?;
260            }
261            self.header = new_page.header;
262            self.data = new_page.data;
263        }
264        Ok(())
265    }
266
267    /// Remove the tuple at `slot_num` and compact remaining tuples.
268    pub fn reclaim_tuple(&mut self, slot_num: u16) -> QuillSQLResult<()> {
269        if slot_num >= self.header.num_tuples {
270            return Err(QuillSQLError::Storage(format!(
271                "tuple_id {} out of range",
272                slot_num
273            )));
274        }
275
276        let snapshot_infos = self.header.tuple_infos.clone();
277        let next_page_id = self.header.next_page_id;
278        let lsn = self.header.lsn;
279
280        let mut rebuilt = TablePage::new(self.schema.clone(), next_page_id);
281        rebuilt.set_lsn(lsn);
282
283        for (idx, info) in snapshot_infos.into_iter().enumerate() {
284            if idx == slot_num as usize {
285                continue;
286            }
287            let (tuple, _) = TupleCodec::decode(
288                &self.data[info.offset as usize..(info.offset + info.size) as usize],
289                self.schema.clone(),
290            )?;
291            rebuilt.insert_tuple(&info.meta, &tuple)?;
292        }
293
294        // Preserve LSN and next pointer while replacing storage.
295        rebuilt.header.lsn = lsn;
296        self.header = rebuilt.header;
297        self.data = rebuilt.data;
298        Ok(())
299    }
300
301    pub fn tuple(&self, slot_num: u16) -> QuillSQLResult<(TupleMeta, Tuple)> {
302        if slot_num >= self.header.num_tuples {
303            return Err(QuillSQLError::Storage(format!(
304                "tuple_id {} out of range",
305                slot_num
306            )));
307        }
308
309        let offset = self.header.tuple_infos[slot_num as usize].offset;
310        let size = self.header.tuple_infos[slot_num as usize].size;
311        let meta = self.header.tuple_infos[slot_num as usize].meta;
312        let (tuple, _) = TupleCodec::decode(
313            &self.data[offset as usize..(offset + size) as usize],
314            self.schema.clone(),
315        )?;
316
317        Ok((meta, tuple))
318    }
319
320    pub fn tuple_meta(&self, slot_num: u16) -> QuillSQLResult<TupleMeta> {
321        if slot_num >= self.header.num_tuples {
322            return Err(QuillSQLError::Storage(format!(
323                "tuple_id {} out of range",
324                slot_num
325            )));
326        }
327
328        Ok(self.header.tuple_infos[slot_num as usize].meta)
329    }
330
331    pub fn get_next_rid(&self, rid: &RecordId) -> Option<RecordId> {
332        let next_slot = rid.slot_num + 1;
333        if next_slot < self.header.num_tuples as u32 {
334            Some(RecordId::new(rid.page_id, next_slot))
335        } else {
336            None
337        }
338    }
339}
340
341impl TablePage {
342    pub fn set_lsn(&mut self, lsn: Lsn) {
343        self.header.lsn = lsn;
344    }
345
346    pub fn lsn(&self) -> Lsn {
347        self.header.lsn
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use crate::catalog::{Column, DataType, Schema};
354    use crate::storage::page::EMPTY_TUPLE_META;
355    use crate::storage::tuple::Tuple;
356    use std::sync::Arc;
357
358    #[test]
359    pub fn test_table_page_get_tuple() {
360        let schema = Arc::new(Schema::new(vec![
361            Column::new("a", DataType::Int8, false),
362            Column::new("b", DataType::Int16, false),
363        ]));
364        let mut table_page = super::TablePage::new(schema.clone(), 0);
365        let tuple_id = table_page
366            .insert_tuple(
367                &EMPTY_TUPLE_META,
368                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
369            )
370            .unwrap();
371        assert_eq!(tuple_id, 0);
372        let tuple_id = table_page
373            .insert_tuple(
374                &EMPTY_TUPLE_META,
375                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
376            )
377            .unwrap();
378        assert_eq!(tuple_id, 1);
379        let tuple_id = table_page
380            .insert_tuple(
381                &EMPTY_TUPLE_META,
382                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
383            )
384            .unwrap();
385        assert_eq!(tuple_id, 2);
386
387        let (tuple_meta, tuple) = table_page.tuple(0).unwrap();
388        assert_eq!(tuple_meta, EMPTY_TUPLE_META);
389        assert_eq!(tuple.data, vec![1i8.into(), 1i16.into()]);
390        let (_tuple_meta, tuple) = table_page.tuple(1).unwrap();
391        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
392        let (_tuple_meta, tuple) = table_page.tuple(2).unwrap();
393        assert_eq!(tuple.data, vec![3i8.into(), 3i16.into()]);
394    }
395
396    #[test]
397    pub fn test_table_page_update_tuple_meta() {
398        let schema = Arc::new(Schema::new(vec![
399            Column::new("a", DataType::Int8, false),
400            Column::new("b", DataType::Int16, false),
401        ]));
402        let mut table_page = super::TablePage::new(schema.clone(), 0);
403        let _tuple_id = table_page
404            .insert_tuple(
405                &EMPTY_TUPLE_META,
406                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
407            )
408            .unwrap();
409        let _tuple_id = table_page
410            .insert_tuple(
411                &EMPTY_TUPLE_META,
412                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
413            )
414            .unwrap();
415        let _tuple_id = table_page
416            .insert_tuple(
417                &EMPTY_TUPLE_META,
418                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
419            )
420            .unwrap();
421
422        let mut tuple_meta = table_page.tuple_meta(0).unwrap();
423        tuple_meta.mark_deleted(1, 0);
424        tuple_meta.insert_txn_id = 2;
425        tuple_meta.insert_cid = 1;
426
427        table_page.update_tuple_meta(tuple_meta, 0).unwrap();
428        let tuple_meta = table_page.tuple_meta(0).unwrap();
429        assert!(tuple_meta.is_deleted);
430        assert_eq!(tuple_meta.delete_txn_id, 1);
431        assert_eq!(tuple_meta.delete_cid, 0);
432        assert_eq!(tuple_meta.insert_txn_id, 2);
433        assert_eq!(tuple_meta.insert_cid, 1);
434    }
435
436    #[test]
437    fn reclaim_tuple_removes_slot_and_compacts() {
438        let schema = Arc::new(Schema::new(vec![Column::new("id", DataType::Int32, false)]));
439        let mut table_page = super::TablePage::new(schema.clone(), 0);
440        let tuple1 = Tuple::new(schema.clone(), vec![1i32.into()]);
441        let tuple2 = Tuple::new(schema.clone(), vec![2i32.into()]);
442        let tuple3 = Tuple::new(schema.clone(), vec![3i32.into()]);
443
444        table_page
445            .insert_tuple(&super::TupleMeta::new(1, 0), &tuple1)
446            .unwrap();
447        table_page
448            .insert_tuple(&super::TupleMeta::new(2, 0), &tuple2)
449            .unwrap();
450        table_page
451            .insert_tuple(&super::TupleMeta::new(3, 0), &tuple3)
452            .unwrap();
453
454        table_page.reclaim_tuple(1).unwrap();
455        assert_eq!(table_page.header.num_tuples, 2);
456
457        let (meta0, t0) = table_page.tuple(0).unwrap();
458        assert_eq!(meta0.insert_txn_id, 1);
459        assert_eq!(t0, tuple1);
460
461        let (meta1, t1) = table_page.tuple(1).unwrap();
462        assert_eq!(meta1.insert_txn_id, 3);
463        assert_eq!(t1, tuple3);
464    }
465}
quill_sql/storage/page/table_page.rs

quill_sql/storage/page/
table_page.rs
