tablestg 0.4.4

use crate::*;

/// Storage of variable length values (varval).
struct VarValStore {
    t: Table,
}

impl VarValStore {
    pub fn new((next_id, root, len): (u64, u64, u64)) -> Self {
        let rpp = (BMP_SIZE * 8) as u64;
        let record_size = ((PAGE_SIZE - BMP_SIZE as u64) / rpp) as usize;

        let t = Table {
            next_id,
            root,
            len,
            record_size,
            rpp,
        };
        Self { t }
    }

    pub fn save(&self) -> (u64, u64, u64) {
        (self.t.next_id, self.t.root, self.t.len)
    }

    /// Store a new varval, returns id.
    pub fn store(&mut self, user_data: &[u8], ps: &mut PageSet) -> u64 {
        let mut todo: usize = user_data.len();
        let id = self.t.next_id;
        let mut done: usize = 0;

        while todo > 0 {
            let mut amount = self.t.record_size;
            if amount > todo {
                amount = todo
            };
            self.t.append(&user_data[done..done + amount], ps);
            done += amount;
            todo -= amount;
        }
        id
    }

    /// Reads a stored varval with given id and len into user_data.
    ///
    /// Could have version that writes to any Writer ( struct implementing std::io::Write ).
    pub fn get(&self, mut id: u64, len: usize, user_data: &mut [u8], ps: &mut PageSet) {
        assert!(user_data.len() >= len);
        let mut done = 0;
        let mut todo = len;
        while todo > 0 {
            let mut amount = self.t.record_size;
            if amount > todo {
                amount = todo
            };
            self.t.read(id, &mut user_data[done..done + amount], ps);
            done += amount;
            todo -= amount;
            id += 1;
        }
    }

    /// Delete a stored varval with given id and len.
    pub fn delete(&mut self, id: u64, len: usize, ps: &mut PageSet) {
        let count = len.div_ceil(self.t.record_size);
        self.t.delete_range(id, count, ps);
    }

    /// Update a varval, returns new id.
    ///
    /// Will re-use old id if number of chunks needed does not change.
    pub fn update(
        &mut self,
        old_id: u64,
        old_len: usize,
        user_data: &[u8],
        ps: &mut PageSet,
    ) -> u64 {
        let old_count = old_len.div_ceil(self.t.record_size);
        let new_count = user_data.len().div_ceil(self.t.record_size);
        if old_count == new_count {
            let mut done = 0;
            let mut todo = user_data.len();
            let mut id = old_id;
            while todo > 0 {
                let mut amount = self.t.record_size;
                if amount > todo {
                    amount = todo;
                }
                self.t.update(id, &user_data[done..done + amount], ps);
                done += amount;
                todo -= amount;
                id += 1;
            }
            old_id
        } else {
            // println!( "delete old_id={} old_len={} old_count={}", old_id, old_len, old_count );
            self.t.delete_range(old_id, old_count, ps);
            self.store(user_data, ps)
        }
    }
}

// ************** End VarVal

/// Address of stored variable length value ( varval ).
#[derive(Debug, Eq, PartialEq, Copy, Clone, Ord, PartialOrd)] // Comparison should really just be done on id.
pub struct VarValAddr {
    id: u64,
    len: u64,
}

impl VarValAddr {
    pub fn new(dt: &DataType, v: &Value, ps: &mut PageSet) -> Self {
        let bytes = dt.value_to_bytes(v);

        let len = bytes.len() as u64;

        let mut vv = VarValStore::new(ps.vvs);
        let id = vv.store(&bytes, ps);
        ps.vvs = vv.save();

        Self { id, len }
    }

    pub fn get(&self, dt: &DataType, ps: &mut PageSet) -> Value {
        let vv = VarValStore::new(ps.vvs);
        let mut buf = vec![0; self.len as usize];
        vv.get(self.id, self.len as usize, &mut buf, ps);

        let mut ix = 0;
        dt.bytes_to_value(&buf, &mut ix)
    }

    pub fn update(&self, dt: &DataType, v: &Value, ps: &mut PageSet) -> Self {
        let mut vv = VarValStore::new(ps.vvs);

        let bytes = dt.value_to_bytes(v);
        let len = bytes.len() as u64;
        let id = vv.update(self.id, self.len as usize, &bytes, ps);
        ps.vvs = vv.save();

        Self { id, len }
    }

    pub fn delete(&self, ps: &mut PageSet) {
        let mut vv = VarValStore::new(ps.vvs);
        vv.delete(self.id, self.len as usize, ps);
        ps.vvs = vv.save();
    }
}

impl SmallFixed for VarValAddr {
    fn size() -> usize {
        2 * 8
    }
    fn load(bytes: &[u8]) -> Self {
        let id = u64::from_le_bytes(bytes[0..8].try_into().unwrap());
        let len = u64::from_le_bytes(bytes[8..16].try_into().unwrap());
        Self { id, len }
    }
    fn save(&self, bytes: &mut [u8]) {
        bytes[0..8].copy_from_slice(&self.id.to_le_bytes());
        bytes[8..16].copy_from_slice(&self.len.to_le_bytes());
    }
}

// ************** End VarValAddr

#[cfg(test)]
pub fn test_varval(ps: &mut PageSet) {
    let rpp = 64;

    let t = Table {
        next_id: 0,
        root: ps.new_page(),
        len: 0,
        record_size: ((PAGE_SIZE - 8) / rpp) as usize,
        rpp,
    };

    let mut vv = VarValStore { t };

    let mut ids = Vec::new();

    let d1 = b"Hello there George";
    let d2 = b"Tuesday is lovely, so is Wednesday";
    //let d2 = b"Tuesday is lovely! So is Wednesday. .............................................Saturday also.";
    for _ in 0..1_000_000 {
        let id = vv.store(d1, ps);
        ids.push(id);
        // println!("test varval stored {}", id);
        // println!("");
    }

    //println!("ids={:?}", ids);

    let mut buf = vec![0; d1.len()];
    for id in &ids {
        vv.get(*id, d1.len(), &mut buf, ps);
        // println!("buf={}", tos(&buf));
        assert_eq!(buf, d1);
    }

    let mut nids = Vec::new();
    for id in &ids {
        let nid = vv.update(*id, d1.len(), d2, ps);
        nids.push(nid);
    }

    let mut buf = vec![0; d2.len()];
    for id in &nids {
        vv.get(*id, d2.len(), &mut buf, ps);
        assert_eq!(buf, d2);
    }

    println!("t.len (page details)={}", vv.t.len);
}