mem-query 0.0.1

use {
    crate::{
        col::Col,
        record::{Record, Record, FromRecord},
        header::{Header,ProjectFrom},
        relation::{RelationImpl, Relation, Queryable, Insert},
        transaction::{UndoLog,Transaction,RevertableOp,Accessor,SubUndoLog},
        query::{
            QueryPlan, QueryPlanImpl,
            request::{QueryRequest,ReplaceFilters,WithCols,ReplaceOrder},
            order::{SortKey},
            fallback::{ProjectionPlan, PostFilterPlan, PostSortPlan},
            filter::{QueryFilter, Exact},
        },
    },
    std::{
        collections::HashMap,
        convert::Infallible,
        hash::Hash,
        marker::PhantomData,
        error::Error,
    },
    tylisp::{
        sexpr, sexpr_val, partial,
        defun_nocalc,
        typenum as tn,
        marker_traits::List,
        ops::{
            list::{DifferP, Any, Find, Missing, CollatedBy, Head, EmptyP},
            logic::{Not, And},
            Cond,
            Partial,
            Is,
        },
    },
    either::Either,
};

#[derive(Default,Clone)]
pub struct HashIndex<K,Rel>(pub HashMap<K,Rel>);

impl<'a, K:'a, Rel> IntoIterator for &'a HashIndex<K, Rel>
where Rel: Relation<'a>,
{
    type IntoIter = impl Iterator<Item=Self::Item>+'a;
    type Item = (&'a K, Rel::Item);

    fn into_iter(self)->Self::IntoIter {
            self.0.iter()
                  .flat_map(|(k,v)| v.iter_all()
                                     .map(move |it| (k, it)))
    }
}

impl<K:Col, R:RelationImpl> RelationImpl for HashIndex<K,R>
where sexpr!{K; R::Cols}: Header {    
    type Cols = sexpr!{K; R::Cols};
    type FastCols = sexpr!{K};
    type Planner = HashIndexPlanner;
}

#[derive(Debug, Default)]
pub struct HashIndexPlanner;
defun_nocalc!{() HashIndexPlanner {
    ('a, K:Col, R, Req:QueryRequest, Cols:Header) { _:&'a HashIndex<K,R>, _:Req }
    where (HashIndex<K,R>: RelationImpl<Cols=Cols>)
    => { Cond, {{DifferP, @Cols, @Req::OutputCols},
                @ProjectionPlan<'a, HashIndex<K,R>, Req>
               },
               {{Any, {Partial, FiltSpansCol, @K}, @Req::Filters},
                @PostFilterPlan<'a, HashIndex<K,R>, Req, partial!{FiltSpansCol, @K}>
               },
               {{Any, {Partial, SortSpansCol, @K}, @Req::OrderBy},
                @PostSortPlan<'a, HashIndex<K,R>, Req>
               },
               {{Any, {Partial, IsExact, @K}, @Req::Filters},
                @HashIndexLookupPlan<'a,K,R,Req>
               },
               {{And, {Not, {EmptyP, @Req::OrderBy}},
                      {SortTermForCol, @K, {Head, @Req::OrderBy}}},
                @HashIndexOrderedScanPlan<'a,K,R,Req>},
               {{Not, {EmptyP, @Req::OrderBy}},
                @PostSortPlan<'a, HashIndex<K,R>, Req>
               },
               {tn::True,
                @HashIndexScanPlan<'a,K,R,Req>
               }
    };
}}

#[derive(Debug, Default)]
pub struct FiltSpansCol;
defun_nocalc!{() FiltSpansCol {
    (C:Col, F:QueryFilter) { _:C, _:F }
    => {And, {DifferP, @F::ReqCols, @{C}},
             {Not, {Is, Missing, {Find, @C, @F::ReqCols}}}};
}}

#[derive(Debug, Default)]
pub struct SortNeedsCol;
defun_nocalc!{() SortNeedsCol {
    (C:Col, K:SortKey) { _:C, _:K }
    => {Not, {Is, Missing, {Find, @C, @K::ReqCols}}};
}}

#[derive(Debug, Default)]
pub struct SortSpansCol;
defun_nocalc!{() SortSpansCol {
    (C:Col, K:SortKey) { _:C, _:K }
    => {And, {Not, {SortTermForCol, @C, @K}}, //{DifferP, @K::ReqCols, @{C}},
             {SortNeedsCol, @C, @K}};
}}

#[derive(Debug, Default)]
pub struct SortTermForCol;
defun_nocalc!{() SortTermForCol {
    (C:Col, K:SortKey) { _:C, _:K }
    => {Not, {DifferP, @K::ReqCols, @{C}}};
}}

#[derive(Debug, Default)]
pub struct IsExact;
defun_nocalc!{() IsExact {
    (C:Col, F:QueryFilter) { _:C, _:F }
    => {And, {FilterForCol, @C, @F},
             {Is, @F, @Exact<C>}};
}}

#[derive(Debug, Default)]
pub struct FilterForCol;
defun_nocalc!{() FilterForCol {
    (C:Col, F:QueryFilter) { _:C, _:F }
    => {Not, {DifferP, @{C}, @F::ReqCols}};
}}

pub struct HashIndexLookupPlan     <'a,K,R,Req>(&'a HashIndex<K,R>, Req);
pub struct HashIndexOrderedScanPlan<'a,K,R,Req>(&'a HashIndex<K,R>, Req);
pub struct HashIndexScanPlan       <'a,K,R,Req>(&'a HashIndex<K,R>, Req);

impl<'a,K,R,Req:'a,ExactFilt:'a,KeyFilt:'a,ValFilt:'a,SubPlan,SubCols:'a,SubSort:'a>
IntoIterator for HashIndexLookupPlan<'a,K,R,Req>
where
    K: Col + Hash + Eq,
    for<'b> &'b K: crate::record::Record<Cols=sexpr!{K}>,
    Req: QueryRequest,
    Req::OutputCols: CollatedBy<partial!{Is, @K}, Failed=SubCols>,
    Req::OrderBy: CollatedBy<partial!{SortTermForCol, @K}, Failed=SubSort>,
    SubCols: Header,
    SubSort: SortKey,
    Req::Filters: CollatedBy<partial!{FilterForCol, @K}, Passed=KeyFilt, Failed=ValFilt>,
    KeyFilt: Clone + QueryFilter + CollatedBy<partial!{IsExact, @K}, Passed = sexpr!{Exact<K>; ExactFilt}>,
    KeyFilt::ReqCols: ProjectFrom<sexpr!{K}>,
    ValFilt: Clone + List,
    R: Relation<'a> + Queryable<'a, WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>, Plan=SubPlan>,
    SubPlan: QueryPlan<'a, R, WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>>,
    ReplaceFilters<Req, ValFilt>: QueryRequest,
    ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>: QueryRequest,
    WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>: QueryRequest,
{
    type IntoIter = impl Iterator<Item=Self::Item>;
    type Item = sexpr!{&'a K, SubPlan::Row};
    fn into_iter(self)->Self::IntoIter {
        let subrequest = self.1.clone();
        let (key_filt, val_filt) = self.1.into_filters().collate();
        // TODO: ref instead of clone here
        let candidate: K = key_filt.clone().collate().0.head.0;
        let subrequest = subrequest.set_filters(val_filt)
                                   .set_order::<SubSort>()
                                   .set_cols::<SubCols>();

            self.0.0.get_key_value::<K>(&candidate)
                    .into_iter()
                    .filter(move |(k,_)| key_filt.test_record(k))
                    .flat_map(move |(k,v)| v.query(subrequest.clone())
                                            .map(move |it| sexpr_val!{k, it}))
    }
}

impl<'a,K,R,Req:'a,KeyFilt:'a,ValFilt:'a,SubPlan,SubCols:'a,SubSort:'a>
IntoIterator for HashIndexOrderedScanPlan<'a,K,R,Req>
where
    K: Col,
    for<'b> &'b K: crate::record::Record<Cols=sexpr!{K}>,
    Req: QueryRequest,
    Req::OutputCols: CollatedBy<partial!{Is, @K}, Failed=SubCols>,
    Req::OrderBy: CollatedBy<partial!{SortTermForCol, @K}, Failed=SubSort>,
    SubCols: Header,
    SubSort: List + SortKey,
    <Req::OrderBy as List>::Head: SortKey<ReqCols = sexpr!{K}>,
    Req::Filters: CollatedBy<partial!{FilterForCol, @K}, Passed=KeyFilt, Failed=ValFilt>,
    KeyFilt: QueryFilter,
    KeyFilt::ReqCols: ProjectFrom<sexpr!{K}>,
    ValFilt: Clone + List,
    R: Relation<'a> + Queryable<'a, WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>, Plan=SubPlan>,
    SubPlan: QueryPlan<'a, R, WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>>,
    ReplaceFilters<Req, ValFilt>: QueryRequest,
    ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>: QueryRequest,
    WithCols<ReplaceOrder<ReplaceFilters<Req, ValFilt>, SubSort>, SubCols>: QueryRequest,
{
    type IntoIter = impl Iterator<Item=Self::Item>;
    type Item = sexpr!{&'a K, SubPlan::Row};
    fn into_iter(self)->Self::IntoIter {
        let subrequest = self.1.clone();
        let (key_filt, val_filt) = self.1.into_filters().collate();
        let subrequest = subrequest.set_filters(val_filt)
                                   .set_order::<SubSort>()
                                   .set_cols::<SubCols>();

        use ::itertools::Itertools;
            self.0.0.iter()
                    .sorted_by(|&(k1,_), &(k2,_)| <Req::OrderBy as List>::Head::cmp_raw(&k1, &k2))
                    .filter(move |(k,_)| key_filt.test_record(k))
                    .flat_map(move |(k,v)| v.query(subrequest.clone())
                                            .map(move |it| sexpr_val!{k, it}))
    }
}

impl<'a,K,R,Req:'a,KeyFilt:'a,ValFilt:'a,SubPlan,SubCols:'a>
IntoIterator for HashIndexScanPlan<'a,K,R,Req>
where
    K: Col,
    for<'b> &'b K: crate::record::Record<Cols=sexpr!{K}>,
    Req: QueryRequest<OrderBy = sexpr!{}>,
    Req::OutputCols: CollatedBy<partial!{Is, @K}, Failed=SubCols>,
    SubCols: Header,
    Req::Filters: CollatedBy<partial!{FilterForCol, @K}, Passed=KeyFilt, Failed=ValFilt>,
    KeyFilt: QueryFilter,
    KeyFilt::ReqCols: ProjectFrom<sexpr!{K}>,
    ValFilt: Clone + List,
    R: Relation<'a> + Queryable<'a, WithCols<ReplaceFilters<Req, ValFilt>, SubCols>, Plan=SubPlan>,
    SubPlan: QueryPlan<'a, R, WithCols<ReplaceFilters<Req, ValFilt>, SubCols>>,
    ReplaceFilters<Req, ValFilt>: QueryRequest,
    WithCols<ReplaceFilters<Req, ValFilt>, SubCols>: QueryRequest,
{
    type IntoIter = impl Iterator<Item=Self::Item>;
    type Item = sexpr!{&'a K, SubPlan::Row};
    fn into_iter(self)->Self::IntoIter {
        let subrequest = self.1.clone();
        let (key_filt, val_filt) = self.1.into_filters().collate();
        let subrequest = subrequest.set_filters(val_filt).set_cols::<SubCols>();

            self.0.0.iter()
                    .filter(move |(k,_)| key_filt.test_record(k))
                    .flat_map(move |(k,v)| v.query(subrequest.clone())
                                            .map(move |it| sexpr_val!{k, it}))
    }
}

impl<'a,K,R,Req> QueryPlanImpl<'a, HashIndex<K,R>, Req> for HashIndexScanPlan<'a,K,R,Req> {
    fn prepare(rel: &'a HashIndex<K,R>, req: Req)->Self {
        HashIndexScanPlan(rel,req)
    }
}

impl<'a,K,R,Req> QueryPlanImpl<'a, HashIndex<K,R>, Req> for HashIndexOrderedScanPlan<'a,K,R,Req> {
    fn prepare(rel: &'a HashIndex<K,R>, req: Req)->Self {
        HashIndexOrderedScanPlan(rel,req)
    }
}

impl<'a,K,R,Req> QueryPlanImpl<'a, HashIndex<K,R>, Req> for HashIndexLookupPlan<'a,K,R,Req> {
    fn prepare(rel: &'a HashIndex<K,R>, req: Req)->Self {
        HashIndexLookupPlan(rel,req)
    }
}

#[test]
fn test_hashmap_rel() {
    col!{pub A: u32};
    col!{pub B: u32};
    col!{pub C: u32};

    use tylisp::sexpr;
    use crate::relation::Queryable;

    use crate::record::Record;

    assert_trait!{ ( A,  (B,C)): Record };
    assert_trait!{ (&A, &(B,C)): Record };

    assert_trait!{ HashIndex<A, Vec<(B,C)>>: RelationImpl };
    assert_trait!{ HashIndex<A, Vec<(B,C)>>: for<'a> Relation<'a> };

    assert_trait!{ HashIndex<A, Vec<B>>: RelationImpl };
    assert_trait!{ HashIndex<A, Vec<B>>: for<'a> Relation<'a> };
    
    let mut rel: HashIndex<A, Vec<B>> = Default::default();
    rel.insert_multi(vec![
        (A(3), B(11)),
        (A(7), B(5)),
        (A(3), B(7)),
    ]).unwrap();

    use crate::query::request::{BlankRequest, QueryRequest};

    assert_trait!{ BlankRequest: QueryRequest };


    let all:Vec<_> = rel.query(BlankRequest.set_cols::<sexpr!{A,B}>()).collect();
    assert_eq!(3, all.len());
    use crate::record::{Record};

    trait Sorted {
        fn sorted(self)->Self;
    }

    impl Sorted for Vec<u32> {
        fn sorted(mut self)->Self {
            self.sort();    
            self
        }
    }

    assert_eq!(vec![5,7,11], rel.query(BlankRequest.set_cols::<sexpr!{B}>())
                                .map(|rec| **(rec.col_ref::<B>()))
                                .collect::<Vec<_>>()
                                .sorted()
    );

    assert_eq!(
        vec![true,true,true],
        rel.query(BlankRequest.set_cols::<sexpr!{B}>())
                              .map(|rec| rec.col_opt::<A>().is_none())
                              .collect::<Vec<_>>()
    );


    use crate::query::filter::Exact;
    use crate::query::order::{Asc, Desc};


    assert_eq!(vec![11,7],
               rel.query(BlankRequest.set_cols::<sexpr!{B,A}>()
                                     .add_filter(Exact(A(3))))
                  .map(|rec| **rec.col_ref::<B>() )
                  .collect::<Vec<_>>()
    );

    assert_eq!(vec![7,3,3],
               rel.query(BlankRequest.set_cols::<sexpr!{B,A}>()
                                     .set_order::<sexpr!{Asc<B>}>()                   
                  )
                  .map(|rec| **rec.col_ref::<A>() )
                  .collect::<Vec<_>>()
    );

    assert_eq!(vec![3,3,7],
               rel.query(BlankRequest.set_cols::<sexpr!{B,A}>()
                                     .set_order::<sexpr!{Desc<B>}>()                   
                  )
                  .map(|rec| **rec.col_ref::<A>() )
                  .collect::<Vec<_>>()
    );

    assert_eq!(vec![7,11,5],
               rel.query(BlankRequest.set_cols::<sexpr!{A,B}>()
                                     .set_order::<sexpr!{Asc<A>, Asc<B>}>()
                  )
                  .map(|rec| **rec.col_ref::<B>() )
                  .collect::<Vec<_>>()
    );

    assert_eq!(vec![7,11],
               rel.query(BlankRequest.set_cols::<sexpr!{B,A}>()
                                     .add_filter(Exact(A(3)))
                                     .set_order::<sexpr!{Asc<A>, Asc<B>}>())
                  .map(|rec| **rec.col_ref::<B>() )
                  .collect::<Vec<_>>()
    );

    assert_eq!(vec![11, 7],
               rel.query(BlankRequest.set_cols::<sexpr!{B,A}>()
                                     .set_order::<sexpr!{Desc<B>, Asc<A>}>()
                                     .add_filter(Exact(A(3))))
                  .map(|rec| **rec.col_ref::<B>() )
                  .collect::<Vec<_>>()
    );
}


#[derive(thiserror::Error, Debug)]
#[error("Error in subtransaction: {0}")]
pub struct SubtransactionError ( #[from] Box<dyn Error> );

impl<K,R,H:Header> Insert<H> for HashIndex<K,R>
where K: Col + FromRecord<H> + Hash + Eq,
      K::Remainder: Record,
      R: Insert<<K::Remainder as Record>::Cols> + Default,
      Self: RelationImpl {
    type Remainder = R::Remainder;
    type Op = InsertRequest<K,R,<K::Remainder as Record>::Cols>;

    fn insert_op<FromRec:Record<Cols=H>>(rec:FromRec)
    -> (Self::Op, Self::Remainder) {
        let (key, rest) = K::from_rec(rec);
        let (op, remainder) = R::insert_op(rest);
        (InsertRequest { key, op, header: PhantomData }, remainder)
    }
}

pub struct InsertRequest<K,R:Insert<H>,H:Header> {
    key: K,
    op: R::Op,
    header: PhantomData<H>
}

impl<K,R,H:Header> RevertableOp<HashIndex<K,R>> for InsertRequest<K,R,H>
where R:Insert<H> + Default, K:Clone+Hash+Eq {
    type Err = Either<Either<<R::Op as RevertableOp<R>>::Err,
                             Infallible>,
                      Either<Infallible,
                             Infallible>>;
    type Log = (SubUndoLog<InEntry<K>, (<R::Op as RevertableOp<R>>::Log, ()), R>,
                (Option<DelEntry<K>>, ())
               );
    fn apply(self, idx: &mut HashIndex<K,R>)->Result<Self::Log, Self::Err> {
        let InsertRequest { key, op, header:_ } = self;
        Transaction::start(idx)
            .apply(CreateEntry(key.clone()))
            .subtransaction(InEntry(key.clone()), move |xact|
                xact.apply(op)
            )
            .into_undo_log()
    }
}

struct CreateEntry<K>(K);
impl<K:Hash+Eq+Clone,R> RevertableOp<HashIndex<K,R>> for CreateEntry<K>
where R:Default {
    type Err = std::convert::Infallible;
    type Log = Option<DelEntry<K>>;
    fn apply(self, idx: &mut HashIndex<K,R>)->Result<Self::Log, Self::Err> {
        if idx.0.contains_key(&self.0) {
            Ok(None)
        } else {
            idx.0.insert(self.0.clone(), Default::default());
            Ok(Some(DelEntry(self.0)))
        }
    }
}

pub struct DelEntry<K>(K);
impl<K:Hash+Eq,R> UndoLog<HashIndex<K,R>> for DelEntry<K> {
    fn revert(self, idx: &mut HashIndex<K,R>) {
        idx.0.remove(&self.0);
    }
}

pub struct InEntry<K>(K);
impl<K:Hash+Eq,R> Accessor<HashIndex<K,R>> for InEntry<K> {
    type Dest = R;
    fn access<F,O>(&self, idx: &mut HashIndex<K,R>, op:F)->O
        where F: FnOnce(&mut Self::Dest)->O
    {
        op(idx.0.get_mut(&self.0).unwrap())
    }
}