tablestg 0.4.27

use crate::*;

/// A Table stores [Value]s which have a specific [DataType].
///
/// The first field (column) of the stored value must be a 64-bit id.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct Table {
    /// Next id to be allocated.
    next_id: u64,
    /// Store
    store: Store,
    /// DataType
    pub datatype: Arc<DataType>,
    /// Changed
    changed: bool,
}

impl Table {
    /// Start a new table.
    pub fn new(datatype: Arc<DataType>, ps: &mut PageSet) -> Self {
        let store = Store::new(ps);
        Self {
            next_id: 1,
            store,
            datatype,
            changed: true,
        }
    }

    pub fn from_bytes(b: &[u8]) -> Self {
        postcard::from_bytes(b).unwrap()
    }

    pub fn to_bytes(&self) -> Vec<u8> {
        postcard::to_stdvec(self).unwrap()
    }

    /// Get the next record id.
    pub fn new_id(&mut self) -> i64 {
        let result = self.next_id;
        self.next_id += 1;
        self.changed = true;
        result as i64
    }

    /// Insert Value ( row, record ) into the table. Returns the id.
    pub fn insert(&mut self, v: &Value, ps: &mut PageSet) -> i64 {
        let m = &mut self.store;
        let x = {
            let mut spx = (m, &mut *ps);
            self.datatype.value_to_bytes(v, &mut spx)
        };

        let id = v.list()[0].int() as u64;

        let key = IdVKey { id };

        self.store.insert(&key, &x, ps);

        id as i64
    }

    /// Fetch the value associated with the specified id.
    pub fn fetch(&self, id: i64, ps: &mut PageSet) -> Option<Value> {
        let key = IdVKey { id: id as u64 };
        let m = &self.store;
        if let Some(sd) = m.get(&key, ps) {
            let mut spx = (m, &mut *ps);
            let result = sd.decode(&self.datatype, &mut spx);
            Some(result)
        } else {
            None
        }
    }

    /// Fetch the value associated with the specified id. [OwnedLazyRow::item] is used to access the columns.
    pub fn lazy_fetch<'a>(&'a self, id: i64, ps: &mut PageSet) -> Option<OwnedLazyRow<'a>> {
        let key = IdVKey { id: id as u64 };
        let m = &self.store;
        if let Some(sdata) = m.get(&key, ps) {
            let items = sdata.lazy_row_items(&self.datatype);
            Some(OwnedLazyRow {
                sdata,
                table: self,
                items,
            })
        } else {
            None
        }
    }

    /// Update the value associated with the specified id.
    pub fn update(&mut self, id: i64, v: &Value, ps: &mut PageSet) {
        let _ = self.remove(id, ps);
        let m = &mut self.store;
        let x = {
            let mut spx = (m, &mut *ps);
            self.datatype.value_to_bytes(v, &mut spx)
        };
        let key = IdVKey { id: id as u64 };
        self.store.insert(&key, &x, ps);
    }

    /// Remove the value (row, record) specified by id from the table.
    pub fn remove(&mut self, id: i64, ps: &mut PageSet) -> Option<Value> {
        let key = IdVKey { id: id as u64 };
        let m = &mut self.store;
        if let Some(sd) = m.get(&key, ps) {
            let result = {
                let mut spx = (&mut *m, &mut *ps);
                sd.decode_del(&self.datatype, &mut spx)
            };
            m.remove(&key, ps);
            Some(result)
        } else {
            None
        }
    }

    /// Get iterator that returns all records (rows).
    pub fn iter(&self, ps: &mut PageSet) -> TableIter<'_> {
        let inner = self.store.iter(ps);
        TableIter { inner, table: self }
    }

    /// Delete everything, table is no longer useable.
    pub fn delete_all(&mut self, ps: &mut PageSet) {
        self.store.delete_all(ps);
    }

    /// Decode the specified item from ref returned by [TableIter::next_ref].
    pub fn select_value(&self, item: usize, buf: &[u8], ps: &mut PageSet) -> Value {
        let mut spx = (&self.store, &mut *ps);
        self.datatype.select_value(item, buf, &mut spx)
    }

    /// Computes offsets of columns from ref returned by [TableIter::next_ref].
    /// [LazyRow::item] is then used to get values.
    pub fn lazy_row<'a>(&'a self, buf: &'a [u8]) -> LazyRow<'a> {
        let mut ix = 0;
        let items = self.datatype.lazy_row_items(buf, &mut ix);
        LazyRow {
            table: self,
            buf,
            items,
        }
    }

    /// Has table changed.
    pub fn changed(&self) -> bool {
        self.changed || self.store.changed()
    }

    /// Save the table to sys_store.
    pub fn save(&mut self, id: i64, ps: &mut PageSet) {
        if self.changed()
        {   
            self.changed = false;
            let ssc = ps.sys_store.clone();
            let mut sys_store = ssc.borrow_mut();
            let data = self.to_bytes();
            let key = IdVKey::new(id as u64);
            sys_store.remove(&key, ps); // Could have method to do remove and insert together?
            sys_store.insert(&key, &data, ps);
        }
    }

    /// Restore table from sys_store, creates new table if it doesn't exist.
    pub fn restore(id: i64, ps: &mut PageSet, dt: Arc<DataType>) -> Self {
        let ssc = ps.sys_store.clone();
        let mut sys_store = ssc.borrow_mut();
        
        let key = IdVKey::new(id as u64);
        let table = if let Some(sdata) = sys_store.get(&key, ps) {
            sdata.decode_table()
        } else {
            // create a new table
            let table = Table::new(dt, ps);
            let data = table.to_bytes();
            sys_store.insert(&key, &data, ps);
            table
        };
        table
    }
}

/// Result of [Table::lazy_fetch].
pub struct OwnedLazyRow<'a> {
    table: &'a Table,
    sdata: SData,
    items: LVec<LazyItem>,
}

impl<'a> OwnedLazyRow<'a> {
    /// Get specified item from row.
    /// A copy is kept, which is cloned if the same item is fetched again.
    pub fn item(&mut self, item: usize, ps: &mut PageSet) -> Value {
        let x = &mut self.items[item];
        match x {
            LazyItem::Value(v) => v.clone(),
            LazyItem::Offset(off) => {
                let mut spx = (&self.table.store, ps);
                let dt = &self.table.datatype.dt_struct(item);
                let v = self.sdata.decode_at(dt, *off, &mut spx);
                *x = LazyItem::Value(v.clone());
                v
            }
        }
    }
}

/// LazyRow allows a subset of columns to be fetched, see [Table::lazy_row].
pub struct LazyRow<'a> {
    table: &'a Table,
    buf: &'a [u8],
    items: LVec<LazyItem>,
}

impl<'a> LazyRow<'a> {
    /// Get specified item from row.
    /// A copy is kept, which is cloned if the same item is fetched again.
    pub fn item(&mut self, item: usize, ps: &mut PageSet) -> Value {
        let x = &mut self.items[item];
        match x {
            LazyItem::Value(v) => v.clone(),
            LazyItem::Offset(off) => {
                let b = &self.buf[*off..];
                let mut spx = (&self.table.store, ps);
                let dt = &self.table.datatype.dt_struct(item);
                let v = dt.bytes_to_value(b, &mut spx);
                *x = LazyItem::Value(v.clone());
                v
            }
        }
    }

    /// Get offset for item ( which must not have been fetched by item ).
    pub fn item_ref(&mut self, item: usize) -> &[u8] {
        let x = &mut self.items[item];
        match x {
            LazyItem::Value(_v) => panic!(),
            LazyItem::Offset(off) => &self.buf[*off..],
        }
    }

    /// Get byte slice for item with length ( e.g. string or binary ). Returns None if value is stored indirectly.
    pub fn item_bytes(&mut self, item: usize) -> Option<&[u8]> {
        let dt = &self.table.datatype.dt_struct(item);
        match dt {
            DataType::String(_) => {}
            DataType::Binary(_) => (),
            _ => panic!(),
        }
        let b = self.item_ref(item);
        DataType::bytes(b)
    }
}

/// Result of [Table::iter].
pub struct TableIter<'a> {
    inner: StoreIter<'a>,
    table: &'a Table,
}

impl<'a> TableIter<'a> {
    /// Get next value (row, record).
    pub fn next_value(&mut self, ps: &mut PageSet) -> Option<Value> {
        if let Some(data) = self.inner.next(ps) {
            let mut spx = (&self.table.store, &mut *ps);
            let result = self.table.datatype.bytes_to_value(data, &mut spx);
            Some(result)
        } else {
            None
        }
    }

    /// Get ref to data for next record ( row ) from the table.
    ///
    /// This can be more efficient than decoding the whole row.
    /// Use [Table::select_value] or [Table::lazy_row] to select individual values using the ref.
    pub fn next_ref(&mut self, ps: &mut PageSet) -> Option<&[u8]> {
        self.inner.next(ps)
    }
}

use std::hash::{Hash, Hasher};

/*
pub struct StringKey<'a> {
    s: &'a str,
    table: &'a Table,
    col: usize, // More generally this could be any expression on the table record that yields a value of the correct type.
}

impl<'a> StringKey<'a> {
    pub fn new(s: &'a str, table: &'a Table, col: usize) -> Self {
        Self { s, table, col }
    }

    /// Get key column from self.table.
    fn lookup_key(&self, bytes: &[u8], ps: &mut PageSet) -> Value {
        // Get first id in list, get table record and compare with [col] string in that.
        let ix_dt = DataType::IList(50);
        let mut spx = (&self.table.store, &mut *ps);
        let first_id = ix_dt.bytes_to_value(bytes, &mut spx).ilist()[0];
        let mut lz = self.table.lazy_fetch(first_id, ps).unwrap();
        lz.item(self.col, ps)
    }
}

impl<'a> VKey for StringKey<'a> {
    fn ok(&self, bytes: &[u8], ps: &mut PageSet) -> bool {
        let s = self.lookup_key(bytes, ps);
        let s = s.string();
        s == self.s
    }
    fn rehash<H: Hasher>(&self, bytes: &[u8], h: &mut H, ps: &mut PageSet) {
        let s = self.lookup_key(bytes, ps);
        s.string().hash( h );
    }
}

impl<'a> Hash for StringKey<'a> {
    fn hash<H>(&self, h: &mut H)
    where
        H: Hasher,
    {
        self.s.hash(h);
    }
}
*/

/// Key for an index (that allows duplicates) on a Table column.
pub struct IndexKey<'a> {
    val: &'a Value,
    table: &'a Table,
    col: usize,
}

impl<'a> IndexKey<'a> {
    pub fn new(val: &'a Value, table: &'a Table, col: usize) -> Self {
        Self { val, table, col }
    }

    /// Get key column from self.table.
    fn lookup_key(&self, bytes: &[u8], ps: &mut PageSet) -> Value {
        // Get first id in dup list, get col from table record.
        let ix_dt = DataType::IList(0);
        let first_id = ix_dt.bytes_to_value0(bytes).ilist()[0];
        let mut lz = self.table.lazy_fetch(first_id, ps).unwrap();
        lz.item(self.col, ps)
    }
}

impl<'a> VKey for IndexKey<'a> {
    fn ok(&self, bytes: &[u8], ps: &mut PageSet) -> bool {
        let v = self.lookup_key(bytes, ps);
        &v == self.val
    }
    fn rehash<H: Hasher>(&self, bytes: &[u8], h: &mut H, ps: &mut PageSet) {
        let tsv = self.lookup_key(bytes, ps);
        tsv.hash(h);
    }
}

impl<'a> Hash for IndexKey<'a> {
    fn hash<H: Hasher>(&self, h: &mut H) {
        self.val.hash(h);
    }
}

/// Insert id into dup index.
pub fn index_insert(index: &mut Store, id: i64, key: &IndexKey, ps: &mut PageSet) {
    use pstd::veca;

    let ix_dt = DataType::IList(0);
    let sdata = index.get(key, ps);

    let ixv = Value::IList(if let Some(sdata) = sdata {
        let v = sdata.decode0(&ix_dt);
        let mut list = v.ilist().clone();
        let mlist = LRc::make_mut(&mut list);
        mlist.push(id);
        index.remove(key, ps);
        list
    } else {
        LRc::new(veca![id])
    });
    {
        let enc = ix_dt.value_to_bytes0(&ixv);
        index.insert(key, &enc, ps);
    }
}

// ###################################### test test test test ################################

#[cfg(test)]
fn tos(s: &[u8]) -> &str {
    str::from_utf8(s).unwrap()
}

#[cfg(test)]
fn vstr(s: &str) -> Value {
    Value::String(LRc::new(LString::from(s)))
}

#[cfg(test)]
fn cust_dt() -> DataType {
    use pstd::veca;

    let dt_slist = DataType::List(LBox::new(DataType::String(50)), 20);

    DataType::Struct(veca![
        (LString::from("Id"), DataType::Int),
        (LString::from("Name"), DataType::String(50)),
        (LString::from("Email"), DataType::String(50)),
        (LString::from("Postal"), DataType::String(5)),
        (LString::from("Bin"), DataType::Binary(5)),
        (LString::from("List"), DataType::IList(10)),
        (LString::from("Address"), dt_slist),
    ])
}

#[cfg(test)]
fn insert_cust(t: &mut Table, index: &mut Store, name: &str, email: &str, ps: &mut PageSet) -> i64 {
    use pstd::veca;

    let id = t.new_id();

    println!("inserting id={} name={:?} email={:?}", id, name, email);

    let _big: LVec<u8> = veca![b'a'; 10];
    let _big2 = veca![b'g'; 10];

    let v = Value::List(LRc::new(veca![
        Value::Int(id),
        vstr(name),
        vstr(email), // col 2
        vstr(tos(&_big)),
        Value::Binary(LRc::new(_big2)),
        Value::IList(LRc::new(veca![1, 2, 3, 4])),
        Value::List(LRc::new(veca![
            vstr("33 Sandpiper Close"),
            vstr("Quedgeley"),
            vstr("Gloucester"),
        ])),
    ]));

    let id = t.insert(&v, ps);

    {
        let kval = &vstr(email);
        let ikey = IndexKey::new(&kval, t, 2);
        index_insert(index, id, &ikey, ps);
    }

    id
}

#[cfg(test)]
fn policy_dt() -> DataType {
    use pstd::veca;

    DataType::Struct(veca![
        (LString::from("Id"), DataType::Int),
        (LString::from("Cust"), DataType::Int),
        (LString::from("Value"), DataType::Int),
    ])
}

#[cfg(test)]
fn insert_policy(t: &mut Table, index: &mut Store, cust: i64, value: i64, ps: &mut PageSet) -> i64 {
    use pstd::veca;

    let id = t.new_id();

    println!(
        "insert policy id={} cust={:?} value={} index={:?}",
        id, cust, value, index
    );

    let v = Value::List(LRc::new(veca![
        Value::Int(id),
        Value::Int(cust),
        Value::Int(value),
    ]));

    let id = t.insert(&v, ps);

    {
        let kval = Value::Int(cust);
        let ikey = IndexKey::new(&kval, t, 1);
        index_insert(index, id, &ikey, ps);
    }

    id
}

#[cfg(test)]
pub fn test_table(ps: &mut PageSet) {
    let dt = cust_dt();
    let mut t = Table::new(dt, ps);
    let mut index = Store::new(ps);

    let mut pt = Table::new(policy_dt(), ps);
    let mut pindex = Store::new(ps);

    println!(
        "t={:?} pt={:?} index={:?} pindex={:?}",
        &t, &pt, &index, &pindex
    );

    let _id = insert_cust(&mut t, &mut index, "maz1", "maz@gmail.com", ps);
    let _id = insert_cust(&mut t, &mut index, "george", "george@gmail.com", ps);
    let _id = insert_cust(&mut t, &mut index, "maz2", "maz@gmail.com", ps);
    let id = insert_cust(&mut t, &mut index, "maz3", "maz@gmail.com", ps);

    let _pid = insert_policy(&mut pt, &mut pindex, 1, 90, ps);
    let _pid = insert_policy(&mut pt, &mut pindex, 2, 91, ps);

    for value in 93..100 {
        let _pid = insert_policy(&mut pt, &mut pindex, 1, value, ps);
        let _pid = insert_policy(&mut pt, &mut pindex, 2, value + 100, ps);
    }

    let v2 = t.fetch(id, ps).unwrap();

    println!();
    println!("cust table ={:?}", &t);
    println!("policy table ={:?}", &pt);
    println!();

    let mut lr = t.lazy_fetch(id, ps).unwrap();
    assert_eq!(lr.item(2, ps), v2.list()[2]);

    let mut iter = t.iter(ps);
    while let Some(v) = iter.next_value(ps) {
        println!("iter test v={:?}", v);
    }

    {
        println!("index test");
        let email = vstr("maz@gmail.com");
        let ikey = IndexKey::new(&email, &t, 2);
        let sdata = index.get(&ikey, ps).unwrap();
        let mut spx = (&index, &mut *ps);
        let list = sdata.decode(&DataType::IList(0), &mut spx);
        let list = list.ilist();
        for id in &**list {
            println!("id = {}", id);
            let _v = t.fetch((*id) as i64, ps).unwrap();
        }
    }

    {
        println!("policy index test");
        for cust in 1..=5 {
            let custv = Value::Int(cust);
            let ikey = IndexKey::new(&custv, &pt, 1);
            if let Some(sdata) = pindex.get(&ikey, ps) {
                let mut spx = (&pindex, &mut *ps);
                let list = sdata.decode(&DataType::IList(0), &mut spx);
                let list = list.ilist();
                println!("policy ids for cust {}={:?}", cust, list);
                for id in &**list {
                    // println!("id = {}", id);
                    let v = pt.fetch((*id) as i64, ps).unwrap();
                    println!("policy={:?}", v);
                }
            } else {
                println!("cust {} has no policies", cust);
            }
        }
    }

    let mut iter = t.iter(ps);
    while let Some(d) = iter.next_ref(ps) {
        // println!("iter test d={:?}", d);
        let v = t.select_value(2, d, ps);
        println!("Item 2={:?}", v);

        let mut lazy_row = t.lazy_row(d);

        // Example of accessing data directly.
        if let Some(vb) = lazy_row.item_bytes(2) {
            println!("vb(2)={:?}", tos(vb));
        }

        let v1 = lazy_row.item(1, ps);
        let v2 = lazy_row.item(2, ps);
        println!("v1={:?} v2={:?}", v1, v2);
    }

    t.remove(id, ps);
    println!();
    println!("table after remove ={:?}", &t);
    println!();

    println!("table test finished");
}