Frames-streamly 0.1.0.2 → 0.1.0.3

raw patch · 8 files changed

+610/−94 lines, 8 filesdep +binarydep +bytestringdep +bytestring-strict-builderdep ~Framesdep ~basedep ~primitivePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: binary, bytestring, bytestring-strict-builder, cereal, clock, fast-builder, foldl, mtl, relude, streamly-bytestring, strict, vector

Dependency ranges changed: Frames, base, primitive, streamly, text, vinyl

API changes (from Hackage documentation)

Files

• Changelog.md +3/−0
• Frames-streamly.cabal +47/−4
• examples/Serialize.hs +140/−0
• examples/Strictness.hs +222/−0
• examples/StrictnessPaths.hs +20/−0
• src/Frames/Streamly/CSV.hs +166/−73
• src/Frames/Streamly/InCore.hs +8/−11
• src/Frames/Streamly/Transform.hs +4/−6

@@ -1,3 +1,6 @@+v0.1.0.3+* Bumped some upper bounds and raised the lower bound on relude+ v0.1.0.2 * Relaxed some ```MonadIO``` constraints to ```Monad``` in Frames.Streamly.CSV 

@@ -1,4 +1,4 @@-cabal-version: 1.12+cabal-version: 2.2  -- This file has been generated from package.yaml by hpack version 0.31.2. --@@ -7,7 +7,7 @@ -- hash: f129113aa17f26529351746660f5b3ada35ea09fec7866ca91d25902f08ac8be  name:           Frames-streamly-version:        0.1.0.2+version:        0.1.0.3 synopsis:       A streamly layer for Frames I/O description:    More information is available in the <https://github.com/adamConnerSax/Frames-streamly/blob/master/Readme.md readme>.' category:       Data@@ -16,7 +16,7 @@ author:         Adam Conner-Sax maintainer:     adam_conner_sax@yahoo.com copyright:      2020 Adam Conner-Sax-license:        BSD3+license:        BSD-3-Clause license-file:   LICENSE build-type:     Simple extra-source-files:@@ -36,17 +36,24 @@       Frames.Streamly.CSV       Frames.Streamly.InCore       Frames.Streamly.Transform+  autogen-modules:+     Paths_Frames_streamly   hs-source-dirs:       src   ghc-options: -Wall -O2 -fdicts-strict -fmax-worker-args=16 -fspec-constr-recursive=16   build-depends:       Frames >=0.6 && <0.8-    , base >=4.12.0 && <4.15+    , base >=4.12.0 && <4.17     , exceptions >=0.10.0 && <0.11     , primitive >=0.7 && <0.8+    , relude >=1.0.0 && < 1.1     , streamly >=0.7 && <0.8+    , strict >= 0.4 && < 0.5     , text >=1.2.3 && <1.3     , vinyl >=0.12 && <0.14+  mixins:  base hiding (Prelude)+         , relude (Relude as Prelude+         , Relude.Extra)   default-language: Haskell2010  test-suite Demo@@ -65,4 +72,40 @@     , streamly     , text     , vinyl+  default-language: Haskell2010++test-suite Strictness+  type: exitcode-stdio-1.0+  main-is: Strictness.hs+  other-modules:+      Serialize+      StrictnessPaths+      Paths_Frames_streamly+  hs-source-dirs:+      examples+  ghc-options: -Wall -O2 -fdicts-strict -fmax-worker-args=16 -fspec-constr-recursive=16+--  ghc-options: -Wall -prof -fprof-auto -eventlog -rtsopts -fexternal-interpreter -fdicts-strict -fmax-worker-args=16 -fspec-constr-recursive=16+  build-depends:+      Frames+    , Frames-streamly+    , base+    , bytestring+    , cereal+    , binary+    , fast-builder+    , bytestring-strict-builder+    , mtl+    , foldl+    , primitive+    , relude+    , streamly+    , streamly-bytestring+    , strict+    , clock+    , text+    , vector+    , vinyl+  mixins:  base hiding (Prelude)+         , relude (Relude as Prelude+         , Relude.Extra)   default-language: Haskell2010

@@ -0,0 +1,140 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds           #-}+{-# LANGUAGE DerivingVia         #-}+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+module Serialize+  (+    -- * Types+    SElField(..)+  , RecSerialize+  , RecBinary+    -- * Record coercions+  , toS+  , fromS+  -- * Wrapper for serializable frames+  , SFrame (..)+  )+where++import qualified Control.Monad.ST as ST+import qualified Data.Vinyl                    as V+import qualified Data.Vinyl.TypeLevel          as V++import           Data.Binary                   as B+import           Data.Binary.Put               as B+import           Data.Binary.Get               as B+import           Data.Serialize                as S+import qualified Frames as F+import qualified Frames.InCore as FI+import qualified Frames.Streamly.InCore as FS++import           GHC.Generics (Generic,Rep)+import           GHC.TypeLits (KnownSymbol)++import qualified Streamly+import qualified Streamly.Prelude as Streamly+import qualified Streamly.Internal.Data.Fold as Streamly.Fold+++newtype SElField t = SElField { unSElField :: V.ElField t }+deriving via (V.ElField '(s,a)) instance (KnownSymbol s, Show a) => Show (SElField '(s,a))+deriving via (V.ElField '(s,a)) instance (KnownSymbol s) => Generic (SElField '(s,a))+deriving via (V.ElField '(s,a)) instance Eq a => Eq (SElField '(s,a))+deriving via (V.ElField '(s,a)) instance Ord a => Ord (SElField '(s,a))+++toS :: V.RMap rs => V.Rec V.ElField rs -> V.Rec SElField rs+toS = V.rmap coerce+{-# INLINE toS #-}++fromS :: V.RMap rs => V.Rec SElField rs -> V.Rec V.ElField rs+fromS = V.rmap coerce+{-# INLINE fromS #-}++-- those generic instances allow us to derive instances for the serialization libs+-- instance (S.Serialize (V.Snd t), V.KnownField t) => S.Serialize (V.ElField t)+instance (S.Serialize (V.Snd t), V.KnownField t) => S.Serialize (SElField t)+instance (B.Binary (V.Snd t), V.KnownField t) => B.Binary (SElField t)++type RecSerialize rs = (GSerializePut (Rep (V.Rec SElField rs))+                       , GSerializeGet (Rep (V.Rec SElField rs))+                       , Generic (V.Rec SElField rs))++instance RecSerialize rs => S.Serialize (V.Rec SElField rs)++type RecBinary rs = (GBinaryPut (Rep (V.Rec SElField rs))+                       , GBinaryGet (Rep (V.Rec SElField rs))+                       , Generic (V.Rec SElField rs))++instance RecBinary rs => B.Binary (V.Rec SElField rs)++newtype SFrame a = SFrame { unSFrame :: F.Frame a }++type SFrameRec rs = SFrame (F.Record rs)++-- Cereal+instance (V.RMap rs, FI.RecVec rs, RecSerialize rs) => S.Serialize (SFrame (F.Record rs)) where+  put = streamlyPutC . Streamly.map toS . Streamly.fromFoldable . unSFrame+  {-# INLINEABLE put #-}+  get = sframeGetC+  {-# INLINEABLE get #-}++-- we use only one fold to get the length and build the bytestream+streamlyPutC :: S.Serialize a => S.Putter (Streamly.SerialT Identity a)+streamlyPutC s = do+  let lengthF = Streamly.Fold.length+      putF = Streamly.Fold.Fold (\b a -> return $ b <> S.put a) mempty return+      (l, streamPut) = runIdentity $ Streamly.fold (Streamly.Fold.tee lengthF putF) s+  S.putWord64be $ fromIntegral l+  streamPut+{-# INLINEABLE streamlyPutC #-}++-- the ST monad is...tricky!  In some sense, is each invocation of @go@ using a "different" s?+sframeGetC :: forall rs. (FI.RecVec rs, V.RMap rs, RecSerialize rs) => S.Get (SFrameRec rs)+sframeGetC = go Streamly.nil =<< S.getWord64be where+  go :: (forall s.Streamly.SerialT (ST.ST s) (F.Rec SElField rs)) -> Word64 -> S.Get (SFrameRec rs)+  go s nLeft =+    if nLeft == 0+    then return $ SFrame $ ST.runST $ FS.inCoreAoS $ Streamly.map fromS s+    else do+      a <- S.get+      go (Streamly.cons a s) (nLeft - 1)+{-# INLINEABLE sframeGetC #-}++-- Binary+instance (V.RMap rs, FI.RecVec rs, RecBinary rs) => B.Binary (SFrame (F.Record rs)) where+  put = streamlyPutB . Streamly.map toS . Streamly.fromFoldable . unSFrame+  {-# INLINEABLE put #-}+  get = sframeGetB+  {-# INLINEABLE get #-}++-- we use only one fold to get the length and build the bytestream+streamlyPutB :: B.Binary a => Streamly.SerialT Identity a -> B.Put+streamlyPutB s = do+  let lengthF = Streamly.Fold.length+      putF = Streamly.Fold.Fold (\b a -> return $ b <> B.put a) mempty return+      (l, streamPut) = runIdentity $ Streamly.fold (Streamly.Fold.tee lengthF putF) s+  B.putWord64be $ fromIntegral l+  streamPut+{-# INLINEABLE streamlyPutB #-}++sframeGetB :: forall rs. (FI.RecVec rs, V.RMap rs, RecBinary rs) => B.Get (SFrameRec rs)+sframeGetB = go Streamly.nil =<< B.getWord64be where+  go :: (forall s.Streamly.SerialT (ST.ST s) (F.Rec SElField rs)) -> Word64 -> B.Get (SFrameRec rs)+  go s nLeft =+    if nLeft == 0+    then return $ SFrame $ ST.runST $ FS.inCoreAoS $ Streamly.map fromS s+    else do+      a <- B.get+      go (Streamly.cons a s) (nLeft - 1)+{-# INLINEABLE sframeGetB #-}

@@ -0,0 +1,222 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeOperators #-}+--{-# OPTIONS_GHC -O0 #-}+module Main where++import StrictnessPaths++import qualified Data.Text as T+import qualified Data.Serialize                as S+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Builder as BB+import qualified ByteString.StrictBuilder as BSB+import qualified Control.Foldl                 as FL+import qualified Control.Monad.State           as ST+import qualified Data.Word as Word+import qualified Frames as F+import qualified Frames.Streamly.CSV as FStreamly+import qualified Frames.Streamly.InCore as FStreamly+import qualified Serialize as FS+import qualified Frames.CSV                     as Frames+import qualified Streamly.Data.Fold            as Streamly.Fold+import qualified Streamly.Internal.Data.Fold.Types            as Streamly.Fold+import qualified Streamly.Internal.Data.Fold            as Streamly.Fold+import qualified Streamly.Prelude              as Streamly+import qualified Streamly.Internal.Prelude              as Streamly+import qualified Streamly              as Streamly+import qualified Streamly.Internal.FileSystem.File+                                               as Streamly.File+import qualified Streamly.External.ByteString  as Streamly.ByteString++import qualified Streamly.Internal.Data.Array  as Streamly.Data.Array+import qualified Streamly.Internal.Memory.Array as Streamly.Memory.Array++import qualified System.Clock++import qualified Data.Vector as V+import qualified Data.Vector.Mutable as VM+import qualified Control.Monad.Primitive as Prim++import qualified Data.Strict.Either as Strict+import qualified Data.Strict.Maybe as Strict++F.tableTypes' pumsACS1YrRowGen++main :: IO ()+main= do+  testInIO++encodeOne :: S.Serialize a => a -> BB.Builder+encodeOne !x = S.execPut $ S.put x++bldrToCT  = Streamly.ByteString.toArray . BL.toStrict . BB.toLazyByteString++encodeBSB :: S.Serialize a => a -> BSB.Builder+encodeBSB !x = BSB.bytes $! encodeBS x++encodeBS :: S.Serialize a => a -> BS.ByteString+encodeBS !x = S.runPut $! S.put x+++bsbToCT  = Streamly.ByteString.toArray . BSB.builderBytes++data Accum b = Accum { count :: !Int, bldr :: !b }++streamlySerializeF :: forall c bldr m a ct.(Monad m, Monoid bldr, c a, c Word.Word64)+                   => (forall b. c b => b -> bldr)+                   -> (bldr -> ct)+                   -> Streamly.Fold.Fold m a ct+streamlySerializeF encodeOne bldrToCT = Streamly.Fold.Fold step initial extract where+  step (Accum n b) !a = return $ Accum (n + 1) (b <> encodeOne a)+  initial = return $ Accum 0 mempty+  extract (Accum n b) = return $ bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> b+{-# INLINEABLE streamlySerializeF #-}++toCT :: BSB.Builder -> Int -> Streamly.Memory.Array.Array Word8+toCT bldr n = bsbToCT $ encodeBSB (fromIntegral @Int @Word.Word64 n) <> bldr++streamlySerializeF2 :: forall c bldr m a ct.(Monad m, Monoid bldr, c a, c Word.Word64)+                   => (forall b. c b => b -> bldr)+                   -> (bldr -> ct)+                   -> Streamly.Fold.Fold m a ct+streamlySerializeF2 encodeOne bldrToCT =+  let fBuilder = Streamly.Fold.Fold step initial return where+        step !b !a = return $ b <> encodeOne a+        initial = return mempty+      toCT' bldr n = bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> bldr+  in toCT' <$> fBuilder <*> Streamly.Fold.length+--        extract (Accum n b) = return $ bldrToCT $ encodeOne (fromIntegral @Int @Word.Word64 n) <> b+{-# INLINEABLE streamlySerializeF2 #-}++type SmallRow = [PUMSYEAR, PUMSPERWT, PUMSSTATEFIP, PUMSREGION, PUMSPUMA, PUMSMETRO, PUMSDENSITY, PUMSAGE]++testInIO :: IO ()+testInIO = do+  let pumsCSV = "example_data/acs100k.csv"+  putTextLn "Testing File.toBytes..."+  let rawBytesS =  Streamly.File.toBytes pumsCSV+  rawBytes <-  Streamly.fold Streamly.Fold.length rawBytesS+  putTextLn $ "raw PUMS data has " <> show rawBytes <> " bytes."+  putTextLn "Testing readTable..."+  let sPUMSRawRows :: Streamly.SerialT IO PUMS_Raw+        = FStreamly.readTableOpt Frames.defaultParser pumsCSV+  iRows <-  Streamly.fold Streamly.Fold.length sPUMSRawRows+  putTextLn $ "raw PUMS data has " <> (T.pack $ show iRows) <> " rows."+  putTextLn "Testing Frames.Streamly.inCoreAoS:"+  fPums <- FStreamly.inCoreAoS sPUMSRawRows+  putTextLn $ "raw PUMS frame has " <> show (FL.fold FL.length fPums) <> " rows."++  putTextLn "frame to vec"+  vPostFrame :: V.Vector PUMS_Raw <- FL.foldM FL.vectorM fPums+  putTextLn $ "vPostFrame has " <> show (V.length vPostFrame) <> " rows"++  putTextLn "frame to array"+  aPostFrame :: Streamly.Data.Array.Array PUMS_Raw <- Streamly.fold Streamly.Data.Array.write $ Streamly.fromFoldable fPums+  putTextLn $ "aPostFrame has " <> show (Streamly.Data.Array.length aPostFrame) <> " rows"++{-+  -- Previous goes up to 28MB, looks like via doubling.  Then to 0 (collects fPums after counting?)+  -- This one then climbs to 10MB, rows are smaller.  No large leaks.+  let f :: PUMS_Raw -> F.Record SmallRow+      f !x = F.rcast x+  putTextLn "Testing Frames.Streamly.inCoreAoS with row transform:"+  fPums' :: F.FrameRec SmallRow <- FStreamly.inCoreAoS $ Streamly.map f sPUMSRawRows+  putTextLn $ "transformed PUMS frame has " <> show (FL.fold FL.length fPums') <> " rows."+  putTextLn "v1"+--    sDict  = KS.cerealStreamlyDict+  let countFold = runningCountF "reading..." (\n -> "read " <> show (250000 * n) <> " rows") "finished"+      sPUMSRunningCount = Streamly.map f+                          $ Streamly.tapOffsetEvery 250000 250000 countFold sPUMSRawRows+      sPUMSRCToS = Streamly.map FS.toS sPUMSRunningCount++  serializedBytes :: Streamly.Memory.Array.Array Word.Word8  <- Streamly.fold (streamlySerializeF2 @S.Serialize encodeBSB bsbToCT)  sPUMSRCToS+  print $ Streamly.Memory.Array.length serializedBytes++  putTextLn "v4"+  bldr <- Streamly.foldl' (\acc !x -> let b = S.runPut (S.put x) in b `seq` (acc <> BSB.bytes b)) mempty sPUMSRCToS+  print $ BS.length $ BSB.builderBytes bldr++  putTextLn "In stages"+  putTextLn "Word8: "+  w8Array <- Streamly.Data.Array.fromStream $ Streamly.File.toBytes pumsCSV+  putTextLn $ "w8 array is " <> show (Streamly.Data.Array.length w8Array) <> " long."++  putTextLn "To Vector"+  putTextLn "Word8: "+  vWord8 <- streamToVector $ FStreamly.streamWord8 pumsCSV+  putTextLn $ "w8 vector is " <> show (V.length vWord8) <> " long."++  putTextLn "text lines (via vector)"+  vTextLine <- streamToVector $ FStreamly.streamTextLines pumsCSV+  putTextLn $ "vTextLine has " <> show (V.length vTextLine) <> " elements."++  putTextLn "text lines (via mutable vector)"+  mvTextLine <- streamToVector2 100 $ FStreamly.streamTextLines pumsCSV+  putTextLn $ "mvTextLine has " <> show (V.length mvTextLine) <> " elements."++  putTextLn "tokenized (via vector)"+  vTokenized <- streamToVector $ FStreamly.streamTokenized pumsCSV+  putTextLn $ "vTokenized has " <> show (V.length vTokenized) <> " elements."++  putTextLn "parsed (via vector)"+  vParsed :: V.Vector (F.Rec (Strict.Either Text F.:. F.ElField) (F.RecordColumns PUMS_Raw)) <- streamToVector $ FStreamly.streamParsed pumsCSV+  putTextLn $ "vParsed has " <> show (V.length vParsed) <> " elements."++  putTextLn "parsedMaybe (via vector)"+  vParsedMaybe :: V.Vector (F.Rec (Maybe F.:. F.ElField) (F.RecordColumns PUMS_Raw)) <- streamToVector $ FStreamly.streamParsedMaybe pumsCSV+  putTextLn $ "vParsedMaybe has " <> show (V.length vParsedMaybe) <> " elements."+-}+{-+  putTextLn "Copy to boxed array"+  array <- Streamly.Data.Array.fromStream $ sPUMSRunningCount+  putTextLn $ "array has " <> show (Streamly.Data.Array.length array) <> " elements."+-}++streamToVector :: Monad m => Streamly.SerialT m a -> m (V.Vector a)+streamToVector = V.unfoldrM strictUncons++strictUncons :: Monad m => Streamly.SerialT m a -> m (Maybe (a, Streamly.SerialT m a))+strictUncons s = do+  lu <- Streamly.uncons s+  case lu of+    Nothing -> return Nothing+    Just (!a, s') -> return $ Just (a, s')+{-# INLINE strictUncons #-}++streamToVector2 :: (Monad m, Prim.PrimMonad m) => Int -> Streamly.SerialT m a -> m (V.Vector a)+streamToVector2 initialSize s0 = do+  let go curLength curN s v = do+        mNext <- Streamly.uncons s+        case mNext of+          Nothing -> V.unsafeFreeze (VM.unsafeTake curN v)+          Just (!a, s') -> do+            (curLength', v') <- if curN < curLength+                                then return (curLength, v)+                                else VM.grow v (2 * curLength) >>= \v' -> return (2 * curLength, v')+            VM.write v' curN a+            go curLength' (curN + 1) s' v'+  v0 <- VM.new initialSize+  go initialSize 0 s0 v0+++runningCountF :: ST.MonadIO m => T.Text -> (Int -> T.Text) -> T.Text -> Streamly.Fold.Fold m a ()+runningCountF startMsg countMsg endMsg = Streamly.Fold.Fold step start done where+  start = ST.liftIO (putText startMsg) >> return 0+  step !n _ = ST.liftIO $ do+    t <- System.Clock.getTime System.Clock.ProcessCPUTime+    putStr $ show t ++ ": "+    putTextLn $ countMsg n+    return (n+1)+  done _ = ST.liftIO $ putTextLn endMsg

@@ -0,0 +1,20 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+module StrictnessPaths where++import qualified Frames as F+import qualified Frames.TH                     as F+import qualified Data.Text as T++pumsACS1YrCSV :: FilePath+pumsACS1YrCSV = "example_data/acs100k.csv"+++pumsACS1YrRowGen = (F.rowGen pumsACS1YrCSV) { F.tablePrefix = "PUMS"+                                            , F.separator   = ","+                                            , F.rowTypeName = "PUMS_Raw"+                                            }++type PUMSSTATEFIP = "STATEFIP" F.:-> Int

@@ -8,6 +8,7 @@ {-# LANGUAGE KindSignatures #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PolyKinds #-} {-# LANGUAGE RankNTypes        #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StrictData #-}@@ -17,6 +18,7 @@ {-# LANGUAGE TypeOperators #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE UndecidableSuperClasses #-}+--{-# OPTIONS_GHC -O0 #-} {-| Module      : Frames.Streamly.CSV Description : CSV parsing/formatting tools for the Frames library, operating via streamly Streams.@@ -32,7 +34,7 @@ -} module Frames.Streamly.CSV     (-      -- * read from File to Stream of Recs +      -- * read from File to Stream of Recs       readTable     , readTableOpt     , readTableMaybe@@ -42,7 +44,7 @@       -- * convert streaming Text to streaming Records     , streamTable     , streamTableOpt-    , streamTableMaybe  +    , streamTableMaybe     , streamTableMaybeOpt     , streamTableEither     , streamTableEitherOpt@@ -69,35 +71,44 @@     , writeLines     , writeLines'     , word8ToTextLines+    -- * debugging+    , streamWord8+    , streamTextLines+    , streamTokenized+    , streamParsed+    , streamParsedMaybe     ) where +import Prelude hiding(getCompose) import qualified Streamly.Prelude                       as Streamly import qualified Streamly                               as Streamly import           Streamly                                ( IsStream ) import qualified Streamly.Data.Fold                     as Streamly.Fold import qualified Streamly.Data.Unicode.Stream           as Streamly.Unicode+import qualified Streamly.Internal.Memory.Unicode.Array as Streamly.Unicode.Array+import qualified Streamly.Internal.Memory.Array.Types as Streamly.Array import qualified Streamly.Internal.FileSystem.File      as Streamly.File import qualified Streamly.Internal.Data.Unfold          as Streamly.Unfold import           Control.Monad.Catch                     ( MonadCatch )-import           Control.Monad.IO.Class                  ( MonadIO ) -import           Data.Maybe                              (isNothing)+import qualified Data.Strict.Either as Strict import qualified Data.Text                              as T  import qualified Data.Vinyl                             as Vinyl import qualified Data.Vinyl.Functor                     as Vinyl import qualified Data.Vinyl.TypeLevel                   as Vinyl import qualified Data.Vinyl.Class.Method                as Vinyl-import           Data.Word                               ( Word8 )  import qualified Frames                                 as Frames import qualified Frames.CSV                             as Frames-import qualified Frames.ShowCSV                         as Frames +import qualified Frames.ShowCSV                         as Frames+import qualified Frames.ColumnTypeable                  as Frames+import qualified Data.Vinyl as V+import qualified Data.Vinyl.Functor as V (Compose(..), (:.))+import GHC.TypeLits (KnownSymbol) -import Data.Proxy (Proxy (..)) -    -- | Given a stream of @Records@, for which all fields satisfy the `ShowCSV` constraint, -- produce a stream of `Text`, one item (line) per `Record` with the specified separator@@ -105,7 +116,7 @@ streamToSV   :: forall rs m t.      ( Frames.ColumnHeaders rs-     , Monad m+     , MonadIO m      , Vinyl.RecordToList rs      , Vinyl.RecMapMethod Frames.ShowCSV Vinyl.ElField rs      , IsStream t@@ -121,7 +132,7 @@ streamToCSV   :: forall rs m t      . ( Frames.ColumnHeaders rs-       , Monad m+       , MonadIO m        , Vinyl.RecordToList rs        , Vinyl.RecMapMethod Frames.ShowCSV Vinyl.ElField rs        , IsStream t@@ -133,12 +144,12 @@  -- | Given a foldable of @Records@, for which all fields satisfy the `ShowCSV` constraint, -- produce a stream of `Text`, one item (line) per `Record` with the specified separator--- between fields. +-- between fields. streamSV   :: forall f rs m t.      ( Frames.ColumnHeaders rs      , Foldable f-     , Monad m+     , MonadIO m      , Vinyl.RecordToList rs      , Vinyl.RecMapMethod Frames.ShowCSV Vinyl.ElField rs      , IsStream t@@ -146,7 +157,7 @@   => T.Text -- ^ column separator   -> f (Frames.Record rs) -- ^ foldable of Records   -> t m T.Text -- ^ stream of 'Text' rows-streamSV sep = streamToSV sep . Streamly.fromFoldable  +streamSV sep = streamToSV sep . Streamly.fromFoldable {-# INLINEABLE streamSV #-}  -- | Given a foldable of @Records@, for which all fields satisfy the `ShowCSV` constraint,@@ -155,7 +166,7 @@   :: forall f rs m t.      ( Frames.ColumnHeaders rs      , Foldable f-     , Monad m+     , MonadIO m      , Vinyl.RecordToList rs      , Vinyl.RecMapMethod Frames.ShowCSV Vinyl.ElField rs      , IsStream t@@ -172,13 +183,13 @@       , Vinyl.RecordToList rs       , Frames.ColumnHeaders rs       , IsStream t-      , Monad m+      , MonadIO m      )   => (forall a. c a => a -> T.Text) -- ^ @show@-like function for some constraint satisfied by all fields.   -> T.Text -- ^ column separator   -> t m (Frames.Record rs)  -- ^ stream of Records   -> t m T.Text -- ^ stream of 'Text' rows-streamSVClass toText sep s =+streamSVClass toTxt sep s =   (T.intercalate sep . fmap T.pack $ Frames.columnHeaders (Proxy :: Proxy (Frames.Record rs)))   `Streamly.cons`   (Streamly.map (T.intercalate sep . Vinyl.recordToList . Vinyl.rmapMethod @c aux) s)@@ -186,9 +197,9 @@     aux :: (c (Vinyl.PayloadType Vinyl.ElField a))         => Vinyl.ElField a         -> Vinyl.Const T.Text a-    aux (Vinyl.Field x) = Vinyl.Const $ toText x+    aux (Vinyl.Field x) = Vinyl.Const $ toTxt x -    + -- | Given a record of functions to map each field to Text, -- transform a stream of records into a stream of lines of Text, -- headers first, with headers/fields separated by the given separator.@@ -198,27 +209,27 @@      , Vinyl.RApply rs      , Frames.ColumnHeaders rs      , IsStream t-     , Monad m+     , Streamly.MonadAsync m      )   => Vinyl.Rec (Vinyl.Lift (->) f (Vinyl.Const T.Text)) rs -- ^ Vinyl record of formatting functions for the row-type.   -> T.Text  -- ^ column separator   -> t m (Frames.Rec f rs)  -- ^ stream of Records   -> t m T.Text -- ^ stream of 'Text' rows-streamSV' toTextRec sep s = +streamSV' toTextRec sep s =   (T.intercalate sep . fmap T.pack $ Frames.columnHeaders (Proxy :: Proxy (Frames.Record rs)))   `Streamly.cons`   (Streamly.map (T.intercalate sep . Vinyl.recordToList . Vinyl.rapply toTextRec) s) {-# INLINEABLE streamSV' #-}  -- | Convert a streamly stream into a (lazy) list-streamToList :: (IsStream t, Monad m) => t m a -> m [a] +streamToList :: (IsStream t, Monad m) => t m a -> m [a] streamToList = Streamly.toList . Streamly.adapt  -- | lift a field formatting function into the right form to append to a Rec of formatters liftFieldFormatter :: Vinyl.KnownField t                    => (Vinyl.Snd t -> T.Text) -- ^ formatting function for the type in Field @t@                    -> Vinyl.Lift (->) Vinyl.ElField (Vinyl.Const T.Text) t -- ^ formatting function in the form required to use in row-formatters.-liftFieldFormatter toText = Vinyl.Lift $ Vinyl.Const . toText . Vinyl.getField+liftFieldFormatter toTxt = Vinyl.Lift $ Vinyl.Const . toTxt . Vinyl.getField {-# INLINEABLE liftFieldFormatter #-}  -- | lift a composed-field formatting function into the right form to append to a Rec of formatters@@ -226,7 +237,7 @@ liftFieldFormatter1 :: (Functor f, Vinyl.KnownField t)                     => (f (Vinyl.Snd t) -> T.Text) -- ^ formatting function for things like @Maybe a@                     -> Vinyl.Lift (->) (f Vinyl.:. Vinyl.ElField) (Vinyl.Const T.Text) t-liftFieldFormatter1 toText = Vinyl.Lift $ Vinyl.Const . toText . fmap Vinyl.getField . Vinyl.getCompose+liftFieldFormatter1 toTxt = Vinyl.Lift $ Vinyl.Const . toTxt . fmap Vinyl.getField . Vinyl.getCompose {-# INLINEABLE liftFieldFormatter1 #-}  -- | Format a @Text@ field as-is.@@ -234,12 +245,12 @@ formatTextAsIs = liftFieldFormatter id {-# INLINE formatTextAsIs #-} --- | Format a field using the @Show@ instance of the contained type +-- | Format a field using the @Show@ instance of the contained type formatWithShow :: (Vinyl.KnownField t, Show (Vinyl.Snd t)) => Vinyl.Lift (->) Vinyl.ElField (Vinyl.Const T.Text) t formatWithShow = liftFieldFormatter $ T.pack . show {-# INLINE formatWithShow #-} --- | Format a field using the @Frames.ShowCSV@ instance of the contained type +-- | Format a field using the @Frames.ShowCSV@ instance of the contained type formatWithShowCSV :: (Vinyl.KnownField t, Frames.ShowCSV (Vinyl.Snd t)) => Vinyl.Lift (->) Vinyl.ElField (Vinyl.Const T.Text) t formatWithShowCSV = liftFieldFormatter Frames.showCSV {-# INLINE formatWithShowCSV #-}@@ -278,7 +289,7 @@   -> FilePath -- ^ path   -> t m (Frames.Record rs) -- ^ stream of Records   -> m ()-writeStreamSV sep fp = writeLines' fp . streamToSV sep +writeStreamSV sep fp = writeLines' fp . streamToSV sep {-# INLINEABLE writeStreamSV #-}  -- | write a foldable of @Records@ to a file, one line per @Record@.@@ -313,7 +324,7 @@   => FilePath -- ^ file path   -> f (Frames.Record rs) -- ^ 'Foldable' of Records   -> m ()-writeCSV fp = writeSV "," fp +writeCSV fp = writeSV "," fp {-# INLINEABLE writeCSV #-}  -- NB: Uses some internal modules from Streamly.  Will have to change when they become stable@@ -367,7 +378,7 @@   => FilePath -- ^ file path   -> f (Frames.Record rs) -- ^ 'Foldable' of Records   -> m ()-writeCSV_Show fp = writeSV_Show "," fp +writeCSV_Show fp = writeSV_Show "," fp {-# INLINEABLE writeCSV_Show #-}  -- Thanks to Tim Pierson for the functions below!@@ -377,13 +388,13 @@ -- NB:  If the inferred/given rs is different from the actual file row-type, things will go awry. readTableMaybe     :: forall rs t m.-    (MonadIO m+    (Streamly.MonadAsync m+    , MonadCatch m     , IsStream t     , Vinyl.RMap rs-    , Frames.ReadRec rs-    , MonadCatch m)+    , Frames.ReadRec rs)     => FilePath -- ^ file path-    -> t m (Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Maybe :. ElField@ records after parsing.  +    -> t m (Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Maybe :. ElField@ records after parsing. readTableMaybe = readTableMaybeOpt Frames.defaultParser {-# INLINEABLE readTableMaybe #-} @@ -392,14 +403,14 @@ -- NB:  If the inferred/given rs is different from the actual file row-type, things will go awry. readTableMaybeOpt     :: forall rs t m.-    (MonadIO m+    (Streamly.MonadAsync m+    , MonadCatch m     , IsStream t     , Vinyl.RMap rs-    , Frames.ReadRec rs-    , MonadCatch m)+    , Frames.ReadRec rs)     => Frames.ParserOptions -- ^ parsing options     -> FilePath -- ^ file path-    -> t m (Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Maybe :. ElField@ records after parsing. +    -> t m (Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Maybe :. ElField@ records after parsing. readTableMaybeOpt opts = Streamly.map recEitherToMaybe . readTableEitherOpt opts {-# INLINEABLE readTableMaybeOpt #-} @@ -410,13 +421,13 @@ -- NB:  If the inferred/given rs is different from the actual file row-type, things will go awry. readTableEither   :: forall rs t m.-     (MonadIO m+     (Streamly.MonadAsync m+     , MonadCatch m      , IsStream t      , Vinyl.RMap rs-     , Frames.ReadRec rs-     , MonadCatch m)+     , Frames.ReadRec rs)   => FilePath -- ^ file path-  -> t m (Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Either :. ElField@ records after parsing. +  -> t m (Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Either :. ElField@ records after parsing. readTableEither = readTableEitherOpt Frames.defaultParser  -- | Stream a table from a file path.@@ -425,15 +436,15 @@ -- NB:  If the inferred/given rs is different from the actual file row-type, things will go awry. readTableEitherOpt   :: forall rs t m.-     (MonadIO m+     (Streamly.MonadAsync m+     , MonadCatch m      , IsStream t      , Vinyl.RMap rs-     , Frames.ReadRec rs-     , MonadCatch m)+     , Frames.ReadRec rs)   => Frames.ParserOptions -- ^ parsing options   -> FilePath -- ^ file path-  -> t m (Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Either :. ElField@ records after parsing. -readTableEitherOpt opts = streamTableEitherOpt opts . word8ToTextLines . Streamly.File.toBytes +  -> t m (Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs) -- ^ stream of @Either :. ElField@ records after parsing.+readTableEitherOpt opts = streamTableEitherOpt opts . word8ToTextLines . Streamly.File.toBytes {-# INLINEABLE readTableEitherOpt #-}  @@ -442,11 +453,11 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will go awry. readTable   :: forall rs t m.-     (MonadIO m+     (Streamly.MonadAsync m+     , MonadCatch m      , IsStream t      , Vinyl.RMap rs-     , Frames.ReadRec rs-     , MonadCatch m)+     , StrictReadRec rs)   => FilePath -- ^ file path   -> t m (Frames.Record rs) -- ^ stream of Records readTable = readTableOpt Frames.defaultParser@@ -456,15 +467,15 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will go awry. readTableOpt   :: forall rs t m.-     (MonadIO m+     (Streamly.MonadAsync m+     , MonadCatch m      , IsStream t      , Vinyl.RMap rs-     , Frames.ReadRec rs-     , MonadCatch m)+     , StrictReadRec rs)   => Frames.ParserOptions  -- ^ parsing options   -> FilePath -- ^ file path   -> t m (Frames.Record rs)  -- ^ stream of Records-readTableOpt opts = streamTableOpt opts . word8ToTextLines . Streamly.File.toBytes +readTableOpt !opts !fp = streamTableOpt opts $! word8ToTextLines $! Streamly.File.toBytes fp {-# INLINEABLE readTableOpt #-}  -- | Convert a stream of lines of `Text` to a table@@ -474,7 +485,7 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will go awry. streamTableEither     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs     , Frames.ReadRec rs)@@ -490,7 +501,7 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will..go awry. streamTableEitherOpt     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs     , Frames.ReadRec rs)@@ -498,12 +509,12 @@     -> t m T.Text -- ^ stream of 'Text' rows     -> t m (Vinyl.Rec ((Either T.Text) Vinyl.:. Vinyl.ElField) rs)  -- ^ stream of parsed @Either :. ElField@ rows streamTableEitherOpt opts =-    Streamly.map (doParse . Frames.tokenizeRow opts)+    Streamly.map (parse . Frames.tokenizeRow opts)     . handleHeader   where     handleHeader | isNothing (Frames.headerOverride opts) = Streamly.drop 1                  | otherwise                       = id-    doParse = Frames.readRec    +    parse = Frames.readRec {-# INLINEABLE streamTableEitherOpt #-}  -- | Convert a stream of lines of `Text` to a table.@@ -511,13 +522,13 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will..go awry. streamTableMaybe     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs     , Frames.ReadRec rs)-    => t m T.Text -- ^ stream of 'Text' rows +    => t m T.Text -- ^ stream of 'Text' rows     -> t m (Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs) -- ^ stream of parsed @Maybe :. ElField@ rows-streamTableMaybe = streamTableMaybeOpt Frames.defaultParser +streamTableMaybe = streamTableMaybeOpt Frames.defaultParser {-# INLINEABLE streamTableMaybe #-}  -- | Convert a stream of lines of Text to a table .@@ -525,7 +536,7 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will..go awry. streamTableMaybeOpt     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs     , Frames.ReadRec rs)@@ -541,10 +552,10 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will go awry. streamTable     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs-    , Frames.ReadRec rs+    , StrictReadRec rs     )     => t m T.Text -- ^ stream of 'Text' rows     -> t m (Frames.Record rs) -- ^ stream of Records@@ -556,34 +567,97 @@ -- NB:  If the inferred/given @rs@ is different from the actual file row-type, things will go awry. streamTableOpt     :: forall rs t m.-    (Monad m+    (Streamly.MonadAsync m     , IsStream t     , Vinyl.RMap rs-    , Frames.ReadRec rs+    , StrictReadRec rs     )     => Frames.ParserOptions -- ^ parsing options     -> t m T.Text  -- ^ stream of 'Text' rows     -> t m (Frames.Record rs) -- ^ stream of Records-streamTableOpt opts =-    Streamly.mapMaybe (Frames.recMaybe . doParse . Frames.tokenizeRow opts)-    . handleHeader    +streamTableOpt opts = Streamly.mapMaybe (mRec opts) . handleHeader   where     handleHeader | isNothing (Frames.headerOverride opts) = Streamly.drop 1                  | otherwise                       = id-    doParse = recEitherToMaybe . Frames.readRec-{-# INLINE streamTableOpt #-} +--    doParse = recStrictEitherToMaybe . recEitherToStrict . Frames.readRec+--{-# INLINE streamTableOpt #-}++doParse :: (V.RMap rs, StrictReadRec rs) => [Text] -> V.Rec (Maybe V.:. V.ElField) rs+doParse !x = recStrictEitherToMaybe $! strictReadRec x++mRec :: (V.RMap rs, StrictReadRec rs) => Frames.ParserOptions -> Text -> Maybe (V.Rec V.ElField rs)+mRec !opts !x = recMaybe $! doParse $! Frames.tokenizeRow opts x++ recEitherToMaybe :: Vinyl.RMap rs => Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs -> Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs recEitherToMaybe = Vinyl.rmap (either (const (Vinyl.Compose Nothing)) (Vinyl.Compose . Just) . Vinyl.getCompose)-{-# INLINE recEitherToMaybe #-}+--{-# INLINE recEitherToMaybe #-} +recStrictEitherToMaybe :: Vinyl.RMap rs => Vinyl.Rec (Strict.Either T.Text Vinyl.:. Vinyl.ElField) rs -> Vinyl.Rec (Maybe Vinyl.:. Vinyl.ElField) rs+recStrictEitherToMaybe = Vinyl.rmap (Strict.either (const (Vinyl.Compose Nothing)) (Vinyl.Compose . Just) . Vinyl.getCompose)+--{-# INLINE recStrictEitherToMaybe #-}++{-+recEitherToStrict :: Vinyl.RMap rs => Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs -> Vinyl.Rec (Strict.Either T.Text Vinyl.:. Vinyl.ElField) rs+recEitherToStrict = Vinyl.rmap (Vinyl.Compose . either Strict.Left Strict.Right . Vinyl.getCompose)+--{-# INLINE recEitherToStrict #-}++recUnStrictEither :: Vinyl.RMap rs => Vinyl.Rec (Strict.Either T.Text Vinyl.:. Vinyl.ElField) rs -> Vinyl.Rec (Either T.Text Vinyl.:. Vinyl.ElField) rs+recUnStrictEither = Vinyl.rmap (Vinyl.Compose . Strict.either Left Right . Vinyl.getCompose)+--{-# INLINE recUnStrictEither #-}+-} -- | Convert a stream of Word8 to lines of `Text` by decoding as UTF8 and splitting on "\n"-word8ToTextLines :: (IsStream t, Monad m) => t m Word8 -> t m T.Text-word8ToTextLines =  Streamly.splitOnSuffix (== '\n') (fmap T.pack $ Streamly.Fold.toList)+word8ToTextLines :: (IsStream t, MonadIO m) => t m Word8 -> t m T.Text+word8ToTextLines =  Streamly.splitOnSuffix(=='\n') (toText <$> Streamly.Fold.toList)                     . Streamly.Unicode.decodeUtf8-{-# INLINE word8ToTextLines #-}+--{-# INLINE word8ToTextLines #-} +word8ToTextLines2 :: (IsStream t, MonadIO m) => t m Word8 -> t m T.Text+word8ToTextLines2 =  Streamly.map (toText . Streamly.Array.toList)+                     . Streamly.Unicode.Array.lines+                     . Streamly.Unicode.decodeUtf8+--{-# INLINE word8ToTextLines2 #-} +class StrictReadRec rs where+  strictReadRec :: [Text] -> V.Rec (Strict.Either Text V.:. V.ElField) rs++instance StrictReadRec '[] where+  strictReadRec _ = V.RNil++instance (Frames.Parseable t, StrictReadRec ts, KnownSymbol s) => StrictReadRec (s Frames.:-> t ': ts) where+  strictReadRec [] = V.Compose (Strict.Left mempty) V.:& strictReadRec []+  strictReadRec (!h : t) = maybe+                           (V.Compose (Strict.Left (T.copy h)))+                           (V.Compose . Strict.Right . V.Field)+                           (Frames.parse' h)+                           V.:& strictReadRec t+{-+strictCons !a !b = a V.:& b++strictMaybe :: b -> (a -> b) -> Maybe a -> b+strictMaybe !b f ma = case ma of+  Nothing -> b+  Just !a' -> f a'+++-- strictParseCons :: (KnownSymbol s, Frames.Parseable t)+--                => Text -> V.Rec (Strict.Either Text V.:. V.ElField) rs -> V.Rec (Strict.Either Text V.:. V.ElField) (s Frames.:-> t ': rs)+strictParseCons !h !rs = let !parsed = Frames.parse' h in parsed V.:& rs+-}++rtraverse+  :: Applicative h+  => (forall x. f x -> h (g x))+  -> V.Rec f rs+  -> h (V.Rec g rs)+rtraverse _ V.RNil      = pure V.RNil+rtraverse f (x V.:& xs) = (V.:&) <$> (f x)  <*> rtraverse f xs+--{-# INLINABLE rtraverse #-}++recMaybe :: V.Rec (Maybe V.:. V.ElField) cs -> Maybe (V.Rec V.ElField cs)+recMaybe  = rtraverse V.getCompose+--{-# INLINEABLE recMaybe #-} -- tracing fold {- runningCountF :: MonadIO m => T.Text -> (Int -> T.Text) -> T.Text -> Streamly.Fold.Fold m a ()@@ -596,4 +670,23 @@     return (n+1)   done _ = liftIO $ T.putStrLn endMsg -}+-- For debugging+streamWord8 :: (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m Word8+streamWord8 =  Streamly.File.toBytes+{-# INLINE streamWord8 #-} +streamTextLines :: (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m Text+streamTextLines = word8ToTextLines2 . streamWord8+{-# INLINE streamTextLines #-}++streamTokenized :: (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m [Text]+streamTokenized =  Streamly.map (fmap T.copy . Frames.tokenizeRow Frames.defaultParser) . streamTextLines+{-# INLINE streamTokenized #-}++streamParsed :: (V.RMap rs, StrictReadRec rs) => (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m (V.Rec (Strict.Either Text V.:. V.ElField) rs)+streamParsed =  Streamly.map (strictReadRec . Frames.tokenizeRow Frames.defaultParser) . streamTextLines+{-# INLINE streamParsed #-}++streamParsedMaybe :: (V.RMap rs, StrictReadRec rs) => (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m (V.Rec (Maybe V.:. V.ElField) rs)+streamParsedMaybe =  Streamly.map (recStrictEitherToMaybe . strictReadRec . Frames.tokenizeRow Frames.defaultParser) . streamTextLines+{-# INLINE streamParsedMaybe #-}

@@ -34,12 +34,12 @@     , VectorMFor     , VectorMs     , Vectors-    , RecVec(..)    -    +    , RecVec(..)+     ) where -import qualified Streamly                               as Streamly+import qualified Streamly import qualified Streamly.Prelude                       as Streamly import qualified Streamly.Data.Fold                     as Streamly.Fold import qualified Streamly.Internal.Data.Fold            as Streamly.Fold@@ -48,13 +48,11 @@  import qualified Data.Vinyl                             as Vinyl -import qualified Frames                                 as Frames+import qualified Frames import qualified Frames.InCore                          as Frames import           Frames.InCore                           (VectorFor, VectorMFor, VectorMs, Vectors, RecVec(..), toAoS) -import Data.Proxy (Proxy(..)) - -- | Fold a stream of 'Vinyl' records into SoA (Structure-of-Arrays) form. -- Here as a 'streamly' fold, so it may be deployed along with other folds or on only part of a stream. inCoreSoA_F :: forall m rs. (Prim.PrimMonad m, Frames.RecVec rs)@@ -65,11 +63,11 @@                       >>= flip feed row . (i, sz*2,)           | otherwise = do Frames.writeRec (Proxy::Proxy rs) i mvs' row                            return (i+1, sz, mvs')-                         +         initial = do           mvs <- Frames.allocRec (Proxy :: Proxy rs) Frames.initialCapacity           return (0, Frames.initialCapacity, mvs)-          +         fin (n, _, mvs') =           do vs <- Frames.freezeRec (Proxy::Proxy rs) n mvs'              return . (n,) $ Frames.produceRec (Proxy::Proxy rs) vs@@ -102,7 +100,7 @@            -> Streamly.Fold.Fold m (Frames.Record rs) (Frames.FrameRec ss) inCoreAoS'_F f  = fmap (uncurry Frames.toAoS . aux) inCoreSoA_F   where aux (x,y) = (x, f y)-{-# INLINE inCoreAoS'_F #-}  +{-# INLINE inCoreAoS'_F #-}  -- | Perform the more general AoS fold on a stream of records. inCoreAoS' ::  forall ss rs m. (Prim.PrimMonad m, Frames.RecVec rs)@@ -110,5 +108,4 @@            -> Streamly.SerialT m (Frames.Record rs)            -> m (Frames.FrameRec ss) inCoreAoS' f = Streamly.fold (inCoreAoS'_F f)-{-# INLINE inCoreAoS' #-}  -+{-# INLINE inCoreAoS' #-}

@@ -35,7 +35,7 @@ where  import qualified Frames.Streamly.InCore as FS-import Prelude hiding (filter)+import Prelude hiding (filter, mapMaybe)  import qualified Streamly                               as Streamly import qualified Streamly.Prelude                       as Streamly@@ -60,14 +60,14 @@ transform f = FS.inCoreAoS . f . Streamly.fromFoldable {-# INLINE transform #-} --- | Filter using streamly +-- | Filter using streamly filter :: (Frames.RecVec as) => (Frames.Record as -> Bool) -> Frames.FrameRec as -> Frames.FrameRec as filter f frame = runST $ transform (Streamly.serially . Streamly.filter f) frame {-# INLINE filter #-}  -- | map using speculative streams (concurrency that preserves ordering of results). concurrentMapM :: (Prim.PrimMonad m-                  , Streamly.MonadAsync m                  +                  , Streamly.MonadAsync m                   , Frames.RecVec as                   , Frames.RecVec bs                   ) => (Frames.Record as -> m (Frames.Record bs)) -> Frames.FrameRec as -> m (Frames.FrameRec bs)@@ -83,11 +83,9 @@  -- | mapMaybeM using speculative streams (concurrency that preserves ordering of results). concurrentMapMaybeM :: (Prim.PrimMonad m-                       , Streamly.MonadAsync m                  +                       , Streamly.MonadAsync m                        , Frames.RecVec as                        , Frames.RecVec bs                        ) => (Frames.Record as -> m (Maybe (Frames.Record bs))) -> Frames.FrameRec as -> m (Frames.FrameRec bs) concurrentMapMaybeM f = transform (Streamly.aheadly . Streamly.mapMaybeM f) {-# INLINE concurrentMapMaybeM #-}--