tablestg 0.4.20

use crate::store::{StoreIter, SData};
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)]
pub struct Table {
    /// Next id to be allocated.
    next_id: u64,
    /// Store
    store: Store,
    /// DataType
    datatype: DataType,
}

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

    /// Get the next record id.
    pub fn new_id(&mut self) -> i64 {
        let result = self.next_id;
        self.next_id += 1;
        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(&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 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 mut self, id: i64, ps: & mut PageSet) -> Option<OwnedLazyRow<'a>> {
        let key = IdVKey { id: id as u64 };
        let m = &mut 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,
        }
    }
}

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

// ###################################### 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(100)),
        (LString::from("Bin"), DataType::Binary(50)),
        (LString::from("List"), DataType::IList(20)),
        (LString::from("Address"), dt_slist),
    ])
}

#[cfg(test)]
pub fn test_table(ps: &mut PageSet) {
    use pstd::veca;

    let dt = cust_dt();
    let mut t = Table::new(dt, ps);

    let id = t.new_id();
    let _big: LVec<u8> = veca![b'a'; 120];
    let _big2 = veca![b'g'; 80];

    let v = Value::List(LRc::new(veca![
        Value::Int(id),
        vstr("maz"),
        vstr("maz@gmail.com"),
        vstr(tos(&_big)),
        Value::Binary(LRc::new(_big2)),
        Value::IList(LRc::new(veca![1, 2, 3, 4, 5, 6])),
        Value::List(LRc::new(veca![
            vstr("33 Sandpiper Close"),
            vstr("Quedgeley"),
            vstr("Gloucester"),
        ])),
    ]));

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

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

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