Frames-streamly 0.1.1.1 → 0.1.2.0

raw patch · 6 files changed

+82/−17 lines, 6 filesdep ~streamlyPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: streamly

API changes (from Hackage documentation)

Files

• Changelog.md +3/−0
• Frames-streamly.cabal +2/−2
• Readme.md +1/−1
• src/Frames/Streamly/CSV.hs +25/−8
• src/Frames/Streamly/InCore.hs +23/−0
• src/Frames/Streamly/Transform.hs +28/−6

@@ -1,3 +1,6 @@+v0.1.2.0+* Made compatible with streamly-0.8.0+ v0.1.1.0 * Added ```StrictReadRec``` class 

@@ -7,7 +7,7 @@ -- hash: f129113aa17f26529351746660f5b3ada35ea09fec7866ca91d25902f08ac8be  name:           Frames-streamly-version:        0.1.1.1+version:        0.1.2.0 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@@ -47,7 +47,7 @@     , exceptions >=0.10.0 && <0.11     , primitive >=0.7 && <0.8     , relude >=1.0.0 && < 1.1-    , streamly >=0.7 && <0.8+    , streamly >=0.7 && <0.9     , strict >= 0.4 && < 0.5     , text >=1.2.3 && <1.3     , vinyl >=0.12 && <0.14

@@ -1,4 +1,4 @@-# Frames-streamly- v 0.1.1.1+# Frames-streamly- v 0.1.2.0  [![Build Status][travis-badge]][travis] [![Hackage][hackage-badge]][hackage]

@@ -1,6 +1,7 @@ {-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE BangPatterns #-} {-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE FlexibleContexts #-}@@ -84,12 +85,23 @@  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+#if MIN_VERSION_streamly(0,8,0)+import Streamly.Prelude                       (IsStream)+import qualified Streamly.Internal.Unicode.Array.Char as Streamly.Unicode.Array+import qualified Streamly.Data.Array.Foreign as Streamly.Array+import qualified Streamly.Unicode.Stream           as Streamly.Unicode+#else 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.Data.Unicode.Stream           as Streamly.Unicode+#endif+ import qualified Streamly.Internal.FileSystem.File      as Streamly.File import qualified Streamly.Internal.Data.Unfold          as Streamly.Unfold import           Control.Monad.Catch                     ( MonadCatch )@@ -261,10 +273,15 @@ -- | write a stream of @Text@ to a file, one line per stream item. writeLines' :: (Streamly.MonadAsync m, MonadCatch m, Streamly.IsStream t) => FilePath -> t m T.Text -> m () writeLines' fp s = do+#if MIN_VERSION_streamly(0,8,0)+  let unfoldMany = Streamly.unfoldMany+#else+  let unfoldMany = Streamly.concatUnfold+#endif   Streamly.fold (Streamly.File.write fp)     $ Streamly.Unicode.encodeUtf8     $ Streamly.adapt-    $ Streamly.concatUnfold Streamly.Unfold.fromList+    $ unfoldMany Streamly.Unfold.fromList     $ Streamly.map T.unpack     $ Streamly.intersperse "\n" s {-# INLINEABLE writeLines' #-}@@ -673,22 +690,22 @@   done _ = liftIO $ T.putStrLn endMsg -} -- For debugging-streamWord8 :: (Streamly.IsStream t, MonadIO m, MonadCatch m) => FilePath -> t m Word8+streamWord8 :: (Streamly.IsStream t, Streamly.MonadAsync 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 :: (Streamly.IsStream t, Streamly.MonadAsync 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.IsStream t, Streamly.MonadAsync 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 :: (V.RMap rs, StrictReadRec rs) => (Streamly.IsStream t, Streamly.MonadAsync 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 :: (V.RMap rs, StrictReadRec rs) => (Streamly.IsStream t, Streamly.MonadAsync 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 #-}

@@ -1,4 +1,5 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE FlexibleContexts #-}@@ -39,7 +40,10 @@     ) where +#if MIN_VERSION_streamly(0,8,0)+#else import qualified Streamly+#endif import qualified Streamly.Prelude                       as Streamly import qualified Streamly.Data.Fold                     as Streamly.Fold import qualified Streamly.Internal.Data.Fold            as Streamly.Fold@@ -55,8 +59,26 @@  -- | 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.+#if MIN_VERSION_streamly(0,8,0) inCoreSoA_F :: forall m rs. (Prim.PrimMonad m, Frames.RecVec rs)           => Streamly.Fold.Fold m (Frames.Record rs) (Int, Vinyl.Rec (((->) Int) Frames.:. Frames.ElField) rs)+inCoreSoA_F = Streamly.Fold.mkFoldM feed initial fin+  where feed (!i, !sz, !mvs') row+          | i == sz = Frames.growRec (Proxy::Proxy rs) mvs'+                      >>= flip feed row . (i, sz*2,)+          | otherwise = do Frames.writeRec (Proxy::Proxy rs) i mvs' row+                           return $ Streamly.Fold.Partial (i+1, sz, mvs')++        initial = do+          mvs <- Frames.allocRec (Proxy :: Proxy rs) Frames.initialCapacity+          return $ Streamly.Fold.Partial (0, Frames.initialCapacity, mvs)++        fin (n, _, mvs') =+          do vs <- Frames.freezeRec (Proxy::Proxy rs) n mvs'+             return . (n,) $ Frames.produceRec (Proxy::Proxy rs) vs+#else+inCoreSoA_F :: forall m rs. (Prim.PrimMonad m, Frames.RecVec rs)+          => Streamly.Fold.Fold m (Frames.Record rs) (Int, Vinyl.Rec (((->) Int) Frames.:. Frames.ElField) rs) inCoreSoA_F = Streamly.Fold.mkFold feed initial fin   where feed (!i, !sz, !mvs') row           | i == sz = Frames.growRec (Proxy::Proxy rs) mvs'@@ -71,6 +93,7 @@         fin (n, _, mvs') =           do vs <- Frames.freezeRec (Proxy::Proxy rs) n mvs'              return . (n,) $ Frames.produceRec (Proxy::Proxy rs) vs+#endif {-# INLINE inCoreSoA_F #-}  -- | Perform the 'inCoreSoA_F' fold on a stream of records.

@@ -1,4 +1,5 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE FlexibleContexts #-}@@ -37,17 +38,38 @@ import qualified Frames.Streamly.InCore as FS import Prelude hiding (filter, mapMaybe) +#if MIN_VERSION_streamly(0,8,0)+import Streamly.Prelude                       (IsStream)+#else import qualified Streamly                               as Streamly-import qualified Streamly.Prelude                       as Streamly import           Streamly                                ( IsStream )+#endif++import qualified Streamly.Prelude                       as Streamly import qualified Control.Monad.Primitive                as Prim import Control.Monad.ST (runST)-import qualified Frames                                 as Frames+import qualified Frames import qualified Frames.InCore                          as Frames +#if MIN_VERSION_streamly(0,8,0)+fromSerial :: IsStream t => Streamly.SerialT m a -> t m a+fromSerial = Streamly.fromSerial+{-# INLINE fromSerial #-} +fromAhead :: IsStream t => Streamly.AheadT m a -> t m a+fromAhead = Streamly.fromAhead+{-# INLINE fromAhead #-}+#else+fromSerial :: IsStream t => Streamly.SerialT m a -> t m a+fromSerial = Streamly.serially+{-# INLINE fromSerial #-} +fromAhead :: IsStream t => Streamly.AheadT m a -> t m a+fromAhead = Streamly.aheadly+{-# INLINE fromAhead #-}+#endif + -- | Use streamly to transform a frame. transform ::   forall t as bs m.@@ -62,7 +84,7 @@  -- | 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+filter f frame = runST $ transform (fromSerial . Streamly.filter f) frame {-# INLINE filter #-}  -- | map using speculative streams (concurrency that preserves ordering of results).@@ -71,14 +93,14 @@                   , Frames.RecVec as                   , Frames.RecVec bs                   ) => (Frames.Record as -> m (Frames.Record bs)) -> Frames.FrameRec as -> m (Frames.FrameRec bs)-concurrentMapM f = transform (Streamly.aheadly . Streamly.mapM f)+concurrentMapM f = transform (fromAhead . Streamly.mapM f) {-# INLINE concurrentMapM #-}  -- | mapMaybe using streamly mapMaybe :: (Frames.RecVec as                       , Frames.RecVec bs                       ) => (Frames.Record as -> Maybe (Frames.Record bs)) -> Frames.FrameRec as -> Frames.FrameRec bs-mapMaybe f frame = runST $ transform (Streamly.aheadly . Streamly.mapMaybe f) frame+mapMaybe f frame = runST $ transform (fromAhead . Streamly.mapMaybe f) frame {-# INLINE mapMaybe #-}  -- | mapMaybeM using speculative streams (concurrency that preserves ordering of results).@@ -87,5 +109,5 @@                        , 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)+concurrentMapMaybeM f = transform (fromAhead . Streamly.mapMaybeM f) {-# INLINE concurrentMapMaybeM #-}