tablestg 0.4.15

use crate::*;

/// Table of fixed size records.
///
/// Record size is not limited, although VarValStore currently chooses record size so that page size < PAGE_SIZE.
#[derive(Debug)]
pub struct Table {
    /// Next id to be allocated.
    pub next_id: u64,
    /// Root and Length of underlying TreeVec.
    pub tv: (u64, u64),
    /// Record size
    pub record_size: usize,
    /// Number of records per page.
    pub rpp: u64,
}

impl Table {
    /// Append a new record to the table, returns id of new record.
    pub fn append(&mut self, user_data: &[u8], ps: &mut PageSet) -> u64 {
        assert!(self.record_size >= user_data.len());

        let id = self.next_id;
        self.next_id += 1;

        let pix = id / self.rpp;
        let local_id = (id % self.rpp) as u8;

        let tv = self.get_child_tv(pix, ps);
        let tv = TreeVec::new(tv, ps);
        let mut tp = TablePage::restore(tv, self.record_size);

        tp.append(local_id, user_data);

        self.save_child_tv(pix, tp.save(), ps);

        id
    }

    /// Read a record, identified by supplied id.
    pub fn read(&self, id: u64, user_data: &mut [u8], ps: &mut PageSet) {
        assert!(self.record_size >= user_data.len());
        let pix = id / self.rpp;
        let local_id = (id % self.rpp) as u8;
        let tv = self.get_child_tv(pix, ps);
        let tv = TreeVec::new(tv, ps);
        let mut tp = TablePage::restore(tv, self.record_size);
        tp.read(local_id, user_data);
    }

    /// Update a record, identified by supplied id.
    pub fn update(&mut self, id: u64, user_data: &[u8], ps: &mut PageSet) {
        assert!(self.record_size >= user_data.len());
        let pix = id / self.rpp;
        let local_id = (id % self.rpp) as u8;
        let tv = self.get_child_tv(pix, ps);
        let tv = TreeVec::new(tv, ps);
        let mut tp = TablePage::restore(tv, self.record_size);

        tp.update(local_id, user_data);

        self.save_child_tv(pix, tp.save(), ps);
    }

    /// Delete record identified by supplied id.
    pub fn delete(&mut self, id: u64, ps: &mut PageSet) {
        let pix = id / self.rpp;
        let local_id = (id % self.rpp) as u8;
        let tv = self.get_child_tv(pix, ps);
        let tv = TreeVec::new(tv, ps);
        let mut tp = TablePage::restore(tv, self.record_size);

        tp.delete(local_id);

        self.save_child_tv(pix, tp.save(), ps);
    }

    /// Delete range of records.
    pub fn delete_range(&mut self, mut id: u64, mut count: usize, ps: &mut PageSet) {
        // Should be optimised to use tp.delete_range.
        while count > 0 {
            self.delete(id, ps);
            id += 1;
            count -= 1;
        }
    }

    /// Get root and len of TablePage specified by supplied pix.
    fn get_child_tv(&self, pix: u64, ps: &mut PageSet) -> (u64, u64) {
        let mut tv = TreeVec::new(self.tv, ps);
        let mut buf = [0u8; 16];
        tv.read(pix * 16, &mut buf);

        // Extract root and length from buf.
        let loc = &buf[0..8];
        let mut root = u64::from_le_bytes(loc.try_into().unwrap());

        let loc = &buf[8..16];
        let mut len = u64::from_le_bytes(loc.try_into().unwrap());

        if root == 0 {
            root = tv.ps.new_page();
            len = 0;
        }
        (root, len)
    }

    /// Save root and len of TablePage specified by supplied pix.
    fn save_child_tv(&mut self, pix: u64, (root, len): (u64, u64), ps: &mut PageSet) {
        let mut buf = [0u8; 16];

        let loc = &mut buf[0..8];
        loc.copy_from_slice(&root.to_le_bytes());

        let loc = &mut buf[8..16];
        loc.copy_from_slice(&len.to_le_bytes());

        let mut tv = TreeVec::new(self.tv, ps);
        tv.write(pix * 16, &buf);

        if tv.root_changed || tv.len_changed {
            self.tv = (tv.root, tv.len);
        }
    }
}

////////////

type Bmp = u128;

pub const BMP_SIZE: usize = size_of::<Bmp>();

/// Stores up to 64 or 128 records, depending on type [Bmp].
struct TablePage<'a> {
    pub bmp: Bmp,
    pub v: TreeVec<'a>,
    pub record_size: usize,
}

impl<'a> TablePage<'a> {
    fn restore(mut v: TreeVec<'a>, record_size: usize) -> Self {
        Self {
            bmp: Self::read_bmp(&mut v),
            v,
            record_size,
        }
    }

    fn save(&mut self) -> (u64, u64) {
        self.save_bmp();
        (self.v.root, self.v.len)
    }

    fn read_bmp(v: &mut TreeVec<'a>) -> Bmp {
        let mut buf = [0; BMP_SIZE];
        v.read(0, &mut buf);
        Bmp::from_le_bytes(buf)
    }

    fn save_bmp(&mut self) {
        let mut buf = [0; BMP_SIZE];
        buf.copy_from_slice(&self.bmp.to_le_bytes());
        self.v.write(0, &buf);
    }

    pub fn valid(&self) -> usize {
        self.bmp.count_ones() as usize
    }

    fn index(&self, id: u8) -> usize {
        let mask: Bmp = (1 << id) - 1;
        let upto: Bmp = mask & self.bmp;

        upto.count_ones() as usize
    }

    fn test_bit(&self, id: u8) -> bool {
        let mask: Bmp = 1 << id;
        let test: Bmp = mask & self.bmp;
        test != 0
    }

    fn set_bit(&mut self, id: u8) {
        let mask: Bmp = 1 << id;
        self.bmp |= mask;
    }

    fn clear_bit(&mut self, id: u8) {
        let mask: Bmp = 1 << id;
        let mask = Bmp::MAX - mask;
        self.bmp &= mask;
    }

    /*
        fn last_id(&self) -> u8
        {
           if self.bmp == 0 { return 0; }
           1 + self.bmp.ilog2() as u8
        }
    */

    fn offset(&self, ix: usize) -> u64 {
        (BMP_SIZE as u64) + (ix as u64) * self.record_size as u64
    }

    pub fn append(&mut self, id: u8, user_data: &[u8]) {
        assert!(user_data.len() <= self.record_size);
        assert!(!self.test_bit(id));

        let ix = self.index(id);
        let off = self.offset(ix);

        self.v.write(off, user_data);
        self.set_bit(id);
    }

    pub fn read(&mut self, id: u8, user_data: &mut [u8]) {
        assert!(self.test_bit(id));

        let ix = self.index(id);
        let off = self.offset(ix);

        self.v.read(off, user_data);
    }

    pub fn update(&mut self, id: u8, user_data: &[u8]) {
        assert!(self.test_bit(id));

        let ix = self.index(id);
        let off = self.offset(ix);
        self.v.write(off, user_data);
    }

    pub fn delete(&mut self, id: u8) {
        assert!(self.test_bit(id));

        // Shuffle remaining records down.
        let ix = self.index(id);

        let mut buf = vec![0; self.record_size];
        for i in ix..self.valid() - 1 {
            let off = self.offset(i);
            self.v.read(off + self.record_size as u64, &mut buf);
            self.v.write(off, &buf);
        }
        // ToDo : truncate self.v, size = 8 + self.valid() * self.record_size
        self.clear_bit(id);
    }
}

#[cfg(test)]
pub fn test_table(ps: &mut PageSet) {
    let mut t = Box::new(Table {
        next_id: 0,
        tv: (ps.new_page(), 0),
        record_size: 6,
        rpp: 64,
    });

    let d1 = b"George";
    let id1 = t.append(d1, ps);
    let mut buf = [0u8; 6];
    t.read(id1, &mut buf, ps);
    println!("buf={}", tos(&buf));
    assert_eq!(d1, &buf);

    let d2 = b"Barwoo";
    let id2 = t.append(d2, ps);
    let mut buf = [0u8; 6];
    t.read(id2, &mut buf, ps);
    println!("buf={}", tos(&buf));
    assert_eq!(d2, &buf);

    let d3 = b"Marily";
    let id3 = t.append(d3, ps);

    println!("table info = {:?}", t);

    t.delete(id2, ps);

    println!("table info = {:?}", t);

    let d1x = b"Cather";
    t.update(id1, d1x, ps);

    let mut buf = [0u8; 6];
    t.read(id1, &mut buf, ps);
    println!("buf={}", tos(&buf));
    assert_eq!(d1x, &buf);

    t.delete(id1, ps);

    // Check 3rd record still ok after deletes.
    let mut buf = [0u8; 6];
    t.read(id3, &mut buf, ps);
    println!("buf={}", tos(&buf));
    assert_eq!(d3, &buf);
}