{-# LANGUAGE AllowAmbiguousTypes   #-}
{-# LANGUAGE ConstraintKinds       #-}
{-# LANGUAGE DataKinds             #-}
{-# LANGUAGE DeriveGeneric         #-}
{-# LANGUAGE FlexibleContexts      #-}
{-# LANGUAGE FlexibleInstances     #-}
{-# LANGUAGE GADTs                 #-}
{-# LANGUAGE InstanceSigs          #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE OverloadedStrings     #-}
{-# LANGUAGE PolyKinds             #-}
{-# LANGUAGE RankNTypes            #-}
{-# LANGUAGE ScopedTypeVariables   #-}
{-# LANGUAGE TemplateHaskell       #-}
{-# LANGUAGE TypeApplications      #-}
{-# LANGUAGE TypeOperators         #-}
{-# LANGUAGE TypeFamilies          #-}
{-# LANGUAGE UndecidableInstances  #-}
{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}
module Frames.MapReduce.Maybe where

import qualified Control.MapReduce             as MR
import qualified Frames.MapReduce.General      as MG

import qualified Control.Foldl                 as FL
import           Data.Maybe                     ( isJust )

import qualified Frames                        as F
import           Frames                         ( (:.) )
import qualified Frames.Melt                   as F
import qualified Frames.InCore                 as FI
import qualified Data.Vinyl                    as V
import           Data.Vinyl                     ( ElField )
--import qualified Data.Vinyl.Functor            as V
import qualified Data.Vinyl.TypeLevel          as V

-- | Don't do anything 
unpackNoOp
  :: MR.Unpack (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackNoOp :: Unpack
  (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackNoOp = Unpack
  (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
forall k (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *)
       (rs :: k).
Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
MG.unpackNoOp

-- | Filter records using a function on the entire record. 
unpackFilterRow
  :: (record (Maybe :. ElField) rs -> Bool)
  -> MR.Unpack (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackFilterRow :: (record (Maybe :. ElField) rs -> Bool)
-> Unpack
     (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackFilterRow = (record (Maybe :. ElField) rs -> Bool)
-> Unpack
     (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
forall k (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *)
       (rs :: k).
(record (f :. ElField) rs -> Bool)
-> Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
MG.unpackFilterRow

-- | Filter records based on a condition on only one field in the row.  Will usually require a Type Application to indicate which field.
unpackFilterOnField
  :: forall t rs record
   . (V.KnownField t, F.ElemOf rs t, MG.RecGetFieldC t record Maybe rs)
  => (Maybe (V.Snd t) -> Bool)
  -> MR.Unpack
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) rs)
unpackFilterOnField :: (Maybe (Snd t) -> Bool)
-> Unpack
     (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackFilterOnField = forall (rs :: [(Symbol, *)])
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *).
(Functor f, KnownField t, ElemOf rs t,
 RecGetFieldC t record f rs) =>
(f (Snd t) -> Bool)
-> Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
forall (t :: (Symbol, *)) (rs :: [(Symbol, *)])
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *).
(Functor f, KnownField t, ElemOf rs t,
 RecGetFieldC t record f rs) =>
(f (Snd t) -> Bool)
-> Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
MG.unpackFilterOnField @t

-- | An unpack step which specifies a subset of columns, cs,
-- (via a type-application) and then filters a @Rec (Maybe :. Elfield) rs@
-- to only rows which have all good data in that subset.
unpackGoodRows
  :: forall cs rs record
   . (MG.RCastC cs rs record Maybe)
  => (record (Maybe :. ElField) cs -> Bool)
  -> MR.Unpack
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) rs)
unpackGoodRows :: (record (Maybe :. ElField) cs -> Bool)
-> Unpack
     (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
unpackGoodRows = (record (Maybe :. ElField) cs -> Bool)
-> Unpack
     (record (Maybe :. ElField) rs) (record (Maybe :. ElField) rs)
forall k (cs :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *).
RCastC cs rs record f =>
(record (f :. ElField) cs -> Bool)
-> Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
MG.unpackGoodRows  --unpackFilterRow (isJust . F.recMaybe . F.rcast @cs)

unpackGoodRecRows
  :: forall cs rs
   . (MG.RCastC cs rs V.Rec Maybe)
  => MR.Unpack (V.Rec (Maybe :. ElField) rs) (V.Rec (Maybe :. ElField) rs)
unpackGoodRecRows :: Unpack (Rec (Maybe :. ElField) rs) (Rec (Maybe :. ElField) rs)
unpackGoodRecRows = (Rec (Maybe :. ElField) cs -> Bool)
-> Unpack (Rec (Maybe :. ElField) rs) (Rec (Maybe :. ElField) rs)
forall k (cs :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *).
RCastC cs rs record f =>
(record (f :. ElField) cs -> Bool)
-> Unpack (record (f :. ElField) rs) (record (f :. ElField) rs)
MG.unpackGoodRows @cs (Maybe (Record cs) -> Bool
forall a. Maybe a -> Bool
isJust (Maybe (Record cs) -> Bool)
-> (Rec (Maybe :. ElField) cs -> Maybe (Record cs))
-> Rec (Maybe :. ElField) cs
-> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Rec (Maybe :. ElField) cs -> Maybe (Record cs)
forall (cs :: [(Symbol, *)]).
Rec (Maybe :. ElField) cs -> Maybe (Record cs)
F.recMaybe)

-- | Assign both keys and data cols.  Uses type applications to specify them if they cannot be inferred.
-- Keys usually can't. Data sometimes can.
assignKeysAndData
  :: forall ks cs rs record
   . (MG.RCastC ks rs record Maybe, MG.RCastC cs rs record Maybe)
  => MR.Assign
       (record (Maybe :. ElField) ks)
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) cs)
assignKeysAndData :: Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
assignKeysAndData = Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
forall k (ks :: k) (cs :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *).
(RCastC ks rs record f, RCastC cs rs record f) =>
Assign
  (record (f :. ElField) ks)
  (record (f :. ElField) rs)
  (record (f :. ElField) cs)
MG.assignKeysAndData --MR.assign (F.rcast @ks) (F.rcast @cs)
{-# INLINABLE assignKeysAndData #-}

-- | Assign keys and leave the rest of the columns, excluding the keys, in the data passed to reduce.
splitOnKeys
  :: forall ks rs cs record
   . ( MG.RCastC ks rs record Maybe
     , MG.RCastC cs rs record Maybe
     , cs ~ F.RDeleteAll ks rs
     )
  => MR.Assign
       (record (Maybe :. ElField) ks)
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) cs)
splitOnKeys :: Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
splitOnKeys = forall (rs :: [a]) (record :: ((Symbol, *) -> *) -> [a] -> *).
(RCastC ks rs record Maybe, RCastC cs rs record Maybe) =>
Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
forall k (ks :: k) (cs :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *).
(RCastC ks rs record Maybe, RCastC cs rs record Maybe) =>
Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
assignKeysAndData @ks @cs
{-# INLINABLE splitOnKeys #-}

-- | Assign keys and leave the rest of the columns, excluding the keys, in the data passed to reduce.
splitOnData
  :: forall cs rs ks record
   . ( MG.RCastC ks rs record Maybe
     , MG.RCastC cs rs record Maybe
     , ks ~ F.RDeleteAll cs rs
     )
  => MR.Assign
       (record (Maybe :. ElField) ks)
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) cs)
splitOnData :: Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
splitOnData = forall (rs :: [a]) (record :: ((Symbol, *) -> *) -> [a] -> *).
(RCastC ks rs record Maybe, RCastC cs rs record Maybe) =>
Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
forall k (ks :: k) (cs :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *).
(RCastC ks rs record Maybe, RCastC cs rs record Maybe) =>
Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) cs)
assignKeysAndData @ks @cs
{-# INLINABLE splitOnData #-}


-- | Assign keys and leave all columns, including the keys, in the data passed to reduce.
assignKeys
  :: forall ks rs record
   . (MG.RCastC ks rs record Maybe)
  => MR.Assign
       (record (Maybe :. ElField) ks)
       (record (Maybe :. ElField) rs)
       (record (Maybe :. ElField) rs)
assignKeys :: Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) rs)
assignKeys = Assign
  (record (Maybe :. ElField) ks)
  (record (Maybe :. ElField) rs)
  (record (Maybe :. ElField) rs)
forall k (ks :: k) (rs :: k)
       (record :: ((Symbol, *) -> *) -> k -> *) (f :: * -> *).
RCastC ks rs record f =>
Assign
  (record (f :. ElField) ks)
  (record (f :. ElField) rs)
  (record (f :. ElField) rs)
MG.assignKeys
{-# INLINABLE assignKeys #-}

-- | Reduce the data to a single row and then re-attach the key.
reduceAndAddKey
  :: forall ks cs x record
   . ( MG.IsoRec ks record Maybe
     , MG.IsoRec cs record Maybe
     , MG.IsoRec (ks V.++ cs) record Maybe
     , FI.RecVec ((ks V.++ cs))
     )
  => (forall h . Foldable h => h x -> record (Maybe :. ElField) cs) -- ^ reduction step
  -> MR.Reduce
       (record (Maybe :. ElField) ks)
       x
       (record (Maybe :. ElField) (ks V.++ cs))
reduceAndAddKey :: (forall (h :: * -> *).
 Foldable h =>
 h x -> record (Maybe :. ElField) cs)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (record (Maybe :. ElField) (ks ++ cs))
reduceAndAddKey = (forall (h :: * -> *).
 Foldable h =>
 h x -> record (Maybe :. ElField) cs)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (record (Maybe :. ElField) (ks ++ cs))
forall (ks :: [(Symbol, *)]) (cs :: [(Symbol, *)]) x
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *).
(IsoRec ks record f, IsoRec cs record f,
 IsoRec (ks ++ cs) record f) =>
(forall (h :: * -> *).
 Foldable h =>
 h x -> record (f :. ElField) cs)
-> Reduce
     (record (f :. ElField) ks) x (record (f :. ElField) (ks ++ cs))
MG.reduceAndAddKey
{-# INLINABLE reduceAndAddKey #-}

-- | Reduce by folding the data to a single row and then re-attaching the key.
foldAndAddKey
  :: ( MG.IsoRec ks record Maybe
     , MG.IsoRec cs record Maybe
     , MG.IsoRec (ks V.++ cs) record Maybe
     , FI.RecVec ((ks V.++ cs))
     )
  => FL.Fold x (record (Maybe :. ElField) cs) -- ^ reduction fold
  -> MR.Reduce
       (record (Maybe :. ElField) ks)
       x
       (record (Maybe :. ElField) (ks V.++ cs))
foldAndAddKey :: Fold x (record (Maybe :. ElField) cs)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (record (Maybe :. ElField) (ks ++ cs))
foldAndAddKey = Fold x (record (Maybe :. ElField) cs)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (record (Maybe :. ElField) (ks ++ cs))
forall (ks :: [(Symbol, *)])
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *)
       (cs :: [(Symbol, *)]) x.
(IsoRec ks record f, IsoRec cs record f,
 IsoRec (ks ++ cs) record f) =>
Fold x (record (f :. ElField) cs)
-> Reduce
     (record (f :. ElField) ks) x (record (f :. ElField) (ks ++ cs))
MG.foldAndAddKey
{-# INLINABLE foldAndAddKey #-}

-- | Transform a reduce which produces a container of results, with a function from each result to a record,
-- into a reduce which produces a FrameRec of the result records with the key re-attached.
makeRecsWithKey
  :: ( Functor g
     , Foldable g
     , MG.IsoRec ks record Maybe
     , MG.IsoRec as record Maybe
     , MG.IsoRec (ks V.++ as) record Maybe
     , (FI.RecVec (ks V.++ as))
     )
  => (y -> record (Maybe :. ElField) as) -- ^ map a result to a record
  -> MR.Reduce (record (Maybe :. ElField) ks) x (g y) -- ^ original reduce
  -> MR.Reduce
       (record (Maybe :. ElField) ks)
       x
       (g (record (Maybe :. ElField) (ks V.++ as)))
makeRecsWithKey :: (y -> record (Maybe :. ElField) as)
-> Reduce (record (Maybe :. ElField) ks) x (g y)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (g (record (Maybe :. ElField) (ks ++ as)))
makeRecsWithKey = (y -> record (Maybe :. ElField) as)
-> Reduce (record (Maybe :. ElField) ks) x (g y)
-> Reduce
     (record (Maybe :. ElField) ks)
     x
     (g (record (Maybe :. ElField) (ks ++ as)))
forall (g :: * -> *) (ks :: [(Symbol, *)])
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *)
       (as :: [(Symbol, *)]) y x.
(Functor g, Foldable g, IsoRec ks record f, IsoRec as record f,
 IsoRec (ks ++ as) record f) =>
(y -> record (f :. ElField) as)
-> Reduce (record (f :. ElField) ks) x (g y)
-> Reduce
     (record (f :. ElField) ks) x (g (record (f :. ElField) (ks ++ as)))
MG.makeRecsWithKey
{-# INLINABLE makeRecsWithKey #-}

-- | Transform an effectful reduce which produces a container of results, with a function from each result to a record,
-- into a reduce which produces a FrameRec of the result records with the key re-attached.
makeRecsWithKeyM
  :: ( Monad m
     , Functor g
     , Foldable g
     , MG.IsoRec ks record Maybe
     , MG.IsoRec as record Maybe
     , MG.IsoRec (ks V.++ as) record Maybe
     , (FI.RecVec (ks V.++ as))
     )
  => (y -> record (Maybe :. ElField) as) -- ^ map a result to a record
  -> MR.ReduceM m (record (Maybe :. ElField) ks) x (g y) -- ^ original reduce
  -> MR.ReduceM
       m
       (record (Maybe :. ElField) ks)
       x
       (g (record (Maybe :. ElField) (ks V.++ as)))
makeRecsWithKeyM :: (y -> record (Maybe :. ElField) as)
-> ReduceM m (record (Maybe :. ElField) ks) x (g y)
-> ReduceM
     m
     (record (Maybe :. ElField) ks)
     x
     (g (record (Maybe :. ElField) (ks ++ as)))
makeRecsWithKeyM = (y -> record (Maybe :. ElField) as)
-> ReduceM m (record (Maybe :. ElField) ks) x (g y)
-> ReduceM
     m
     (record (Maybe :. ElField) ks)
     x
     (g (record (Maybe :. ElField) (ks ++ as)))
forall (m :: * -> *) (g :: * -> *) (ks :: [(Symbol, *)])
       (record :: ((Symbol, *) -> *) -> [(Symbol, *)] -> *) (f :: * -> *)
       (as :: [(Symbol, *)]) y x.
(Monad m, Functor g, Foldable g, IsoRec ks record f,
 IsoRec as record f, IsoRec (ks ++ as) record f) =>
(y -> record (f :. ElField) as)
-> ReduceM m (record (f :. ElField) ks) x (g y)
-> ReduceM
     m
     (record (f :. ElField) ks)
     x
     (g (record (f :. ElField) (ks ++ as)))
MG.makeRecsWithKeyM
{-# INLINABLE makeRecsWithKeyM #-}