tablestg 0.4.22

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
    datatype: DataType,
    /// Changed
    changed: bool,
}

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

    /// 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()
    }
}

/// Result of [Table::lazy_fetch].
pub struct OwnedLazyRow<'a> {
    pub table: &'a Table,
    pub sdata: SData,
    pub 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> {
    pub table: &'a Table,
    pub buf: &'a [u8],
    pub 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 tsv = self.lookup_key(bytes, ps);
        let ts = tsv.string();

        println!(
            "ok .... ts={:?} self.s={:?} equal={}",
            ts,
            self.s,
            ts == self.s
        );

        ts == self.s
    }
    fn rehash<H: Hasher>(&self, _bytes: &[u8], _h: &mut H, _ps: &mut PageSet) {
        let _ix_dt = DataType::IList(50);
        todo!() // similar to ok above
    }
}

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

// ###################################### 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 do_insert(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),
        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);

    {
        // Index cust on email address.
        let ix_dt = DataType::IList(50);
        let ekey = StringKey::new(email, &t, 2);

        let sdata = index.get(&ekey, ps);
        let ixv = if let Some(sdata) = sdata {
            index.remove(&ekey, ps); // Maybe could use remove instead of get above.

            let mut spx = (&*index, &mut *ps);
            let v = sdata.decode(&ix_dt, &mut spx);
            let mut list = v.ilist().clone();
            let mlist = LRc::make_mut(&mut list);
            mlist.push(id);
            println!("insert index mlist = {:?}", mlist);
            Value::IList(list)
        } else {
            Value::IList(LRc::new(veca![id as i64]))
        };
        {
            let mut spx = (&mut *index, &mut *ps);
            let enc = ix_dt.value_to_bytes(&ixv, &mut spx);
            index.insert(&ekey, &enc, ps);
        }
    }
    id
}

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

    // let idt = string_index_dt();
    let mut index = Store::new(ps);

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

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

    println!();
    println!("table ={:?}", &t);
    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);
    }

    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");
}