1{-# OPTIONS_GHC -Wno-deprecations #-}
4-- Module : Streamly.Internal.Data.Stream.IsStream.Top
5-- Copyright : (c) 2020 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
11-- Top level IsStream module that can use all other lower level IsStream
14module Streamly.Internal.Data.Stream.IsStream.Top {-# DEPRECATED "Please use \"Streamly.Data.Stream.*\" instead." #-}
18 -- | Value agnostic filtering.
29 -- ** Set like operations
30 -- | These are not exactly set operations because streams are not
31 -- necessarily sets, they may have duplicated elements.
55import Control.Monad.Catch (MonadCatch)
56import Control.Monad.IO.Class (MonadIO(..))
57-- import Control.Monad.Trans.Class (lift)
58-- import Control.Monad.Trans.State.Strict (get, put)
59-- import Data.Function ((&))
60import Data.IORef (newIORef, readIORef, modifyIORef')
61import Data.Kind (Type)
62-- import Data.Maybe (isJust)
63import Streamly.Internal.Control.Concurrent (MonadAsync)
64import Streamly.Internal.Data.Stream.IsStream.Common (concatM)
65import Streamly.Internal.Data.Stream.IsStream.Type
66 (IsStream(..), adapt, foldl', fromList)
67import Streamly.Internal.Data.Stream.Serial (SerialT)
68import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
70import qualified Data.List as List
71import qualified Data.Map.Strict as Map
72-- import qualified Streamly.Internal.Data.Array.Generic as Array
73-- (fromStream, length, read)
74-- import qualified Streamly.Data.Array.Mut as MA
75import qualified Streamly.Internal.Data.Fold as Fold (one, last)
76import qualified Streamly.Internal.Data.Fold.Type as Fold
77import qualified Streamly.Internal.Data.Parser as Parser
78 (groupByRollingEither)
79-- import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream
81import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream
82import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream
83import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
84import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
85import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
86import qualified Streamly.Internal.Data.Stream.StreamD as StreamD
88import Prelude hiding (filter, zipWith, concatMap, concat)
92-- >>> :set -fno-warn-deprecations
93-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
94-- >>> import qualified Streamly.Prelude as Stream
95-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
97------------------------------------------------------------------------------
99------------------------------------------------------------------------------
101-- XXX We can implement this using addition instead of "mod" to make it more
104-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
105-- then every element at strides of @stride@.
107-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
112{-# INLINE sampleFromThen #-}
113sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
114 Int -> Int -> t m a -> t m a
115sampleFromThen offset stride =
116 Stream.with Stream.indexed Stream.filter
117 (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
119-- | Continuously evaluate the input stream and sample the last event in time
120-- window of @n@ seconds.
122-- This is also known as @throttle@ in some libraries.
125-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
130{-# INLINE sampleIntervalEnd #-}
131sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
132 Double -> t m a -> t m a
133sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
135-- | Like 'sampleInterval' but samples at the beginning of the time window.
138-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
143{-# INLINE sampleIntervalStart #-}
144sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
145 Double -> t m a -> t m a
146sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
152 | BurstDoneNext !x !t !x
154{-# INLINE sampleBurst #-}
155sampleBurst :: (IsStream t, MonadAsync m, Functor (t m)) =>
156 Bool -> Double -> t m a -> t m a
157sampleBurst sampleAtEnd gap xs =
158 -- XXX Ideally we should schedule a timer event exactly after gap time,
159 -- but the tick stream should work well as long as the timer
160 -- granularity is small enough compared to the gap.
161 Stream.mapMaybe extract
162 $ Stream.scanl' step BurstNone
164 $ Stream.interjectSuffix 0.01 (return Nothing) (Stream.map Just xs)
168 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
171 step BurstNone (t1, Just x1) = BurstWait t1 x1
172 step BurstNone _ = BurstNone
174 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
175 step (BurstDone _) _ = BurstNone
177 step old@(BurstWait t0 x0) (t1, Nothing)
178 | t1 - t0 >= gap1 = BurstDone x0
180 -- This can happen due to scheduling delays, if we received back to
181 -- back events spaced by more than the timeout without an
182 -- intervening timeout event then we emit the old event instead of
183 -- replacing it by the new.
184 step (BurstWait t0 x0) (t1, Just x1)
185 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
186 | sampleAtEnd = BurstWait t1 x1
187 | otherwise = BurstWait t1 x0
189 step (BurstDoneNext _ t0 x0) (t1, Nothing)
190 | t1 - t0 >= gap1 = BurstDone x0
191 | otherwise = BurstWait t0 x0
192 step (BurstDoneNext _ t0 x0) (t1, Just x1)
193 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
194 | sampleAtEnd = BurstWait t1 x1
195 | otherwise = BurstWait t1 x0
197 {-# INLINE extract #-}
198 extract (BurstDoneNext x _ _) = Just x
199 extract (BurstDone x) = Just x
202-- | Sample one event at the end of each burst of events. A burst is a group
203-- of events close together in time, it ends when an event is spaced by more
204-- than the specified time interval (in seconds) from the previous event.
206-- This is known as @debounce@ in some libraries.
208-- The clock granularity is 10 ms.
212{-# INLINE sampleBurstEnd #-}
213sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
214 Double -> t m a -> t m a
215sampleBurstEnd = sampleBurst True
217-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
218-- instead of at the end of it.
222{-# INLINE sampleBurstStart #-}
223sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
224 Double -> t m a -> t m a
225sampleBurstStart = sampleBurst False
227------------------------------------------------------------------------------
229------------------------------------------------------------------------------
231-- We could possibly choose different algorithms depending on whether the
232-- input stream is almost sorted (ascending/descending) or random. We could
233-- serialize the stream to an array and use quicksort.
235-- | Sort the input stream using a supplied comparison function.
239-- Note: this is not the fastest possible implementation as of now.
244sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
245-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
248 Parser.groupByRollingEither
249 (\x -> (< GT) . cmp x)
250 (fmap fromStream Fold.toStreamKRev)
251 (fmap fromStream Fold.toStreamK)
252 in Stream.concatPairsWith (Stream.mergeBy cmp) id
253 . Stream.rights . Stream.parseMany (fmap (either id id) p)
255------------------------------------------------------------------------------
257------------------------------------------------------------------------------
260-- * https://en.wikipedia.org/wiki/Relational_algebra
261-- * https://en.wikipedia.org/wiki/Join_(SQL)
263-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
264-- constraint, the best would be to use an Array with linear search. If the
265-- second stream is sorted we can also use a binary search, using Ord
266-- constraint or an ordering function.
268-- For Storables we can cache the second stream into an unboxed array for
269-- possibly faster access/compact representation?
271-- If we do not want to keep the stream in memory but always read it from the
272-- source (disk/network) every time we iterate through it then we can do that
273-- too by reading the stream every time, the stream must have immutable state
274-- in that case and the user is responsible for the behavior if the stream
275-- source changes during iterations. We can also use an Unfold instead of
276-- stream. We probably need a way to distinguish streams that can be read
277-- mutliple times without any interference (e.g. unfolding a stream using an
278-- immutable handle would work i.e. using pread/pwrite instead of maintianing
279-- an offset in the handle).
281-- XXX We can do this concurrently.
283-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
286-- The second stream is evaluated multiple times. If the second stream is
287-- consume-once stream then it can be cached in an 'Data.Array.Array' before
288-- calling this function. Caching may also improve performance if the stream is
289-- expensive to evaluate.
294{-# INLINE crossJoin #-}
295crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
297 -- XXX use concatMap instead?
302-- XXX We can do this concurrently.
303-- XXX If the second stream is sorted and passed as an Array we could use
304-- binary search if we have an Ord instance or Ordering returning function. The
305-- time complexity would then become (m x log n).
307-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
308-- are equal by the given equality pedicate then return the tuple (a, b).
310-- The second stream is evaluated multiple times. If the stream is a
311-- consume-once stream then the caller should cache it (e.g. in a
312-- 'Data.Array.Array') before calling this function. Caching may also improve
313-- performance if the stream is expensive to evaluate.
315-- For space efficiency use the smaller stream as the second stream.
317-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
318-- is an order of magnitude faster. joinInner may be used when the second
319-- stream is generated from a seed, therefore, need not be stored in memory and
320-- the amount of memory it takes is a concern.
322-- Space: O(n) assuming the second stream is cached in memory.
327{-# INLINE joinInner #-}
329 forall (t :: (Type -> Type) -> Type -> Type) m a b.
330 (IsStream t, Monad m) =>
331 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
332joinInner eq s1 s2 = do
333 -- ConcatMap works faster than bind
334 Stream.concatMap (\a ->
335 Stream.concatMap (\b ->
337 then Stream.fromPure (a, b)
342-- XXX Generate error if a duplicate insertion is attempted?
343toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
344toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
346-- If the second stream is too big it can be partitioned based on hashes and
347-- then we can process one parition at a time.
349-- XXX An IntMap may be faster when the keys are Int.
350-- XXX Use hashmap instead of map?
352-- | Like 'joinInner' but uses a 'Map' for efficiency.
354-- If the input streams have duplicate keys, the behavior is undefined.
356-- For space efficiency use the smaller stream as the second stream.
363{-# INLINE joinInnerMap #-}
364joinInnerMap :: (IsStream t, Monad m, Ord k) =>
365 t m (k, a) -> t m (k, b) -> t m (k, a, b)
368 km <- toMap $ IsStream.adapt s2
369 pure $ Stream.mapMaybe (joinAB km) s1
374 case k `Map.lookup` kvm of
375 Just b -> Just (k, a, b)
378-- | Like 'joinInner' but works only on sorted streams.
385{-# INLINE joinInnerMerge #-}
386joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
387joinInnerMerge = undefined
390-- XXX We can do this concurrently.
391-- XXX If the second stream is sorted and passed as an Array or a seek capable
392-- stream then we could use binary search if we have an Ord instance or
393-- Ordering returning function. The time complexity would then become (m x log
396-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
397-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
398-- m b@ then return @(a, Nothing)@.
400-- The second stream is evaluated multiple times. If the stream is a
401-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
402-- before calling this function. Caching may also improve performance if the
403-- stream is expensive to evaluate.
406-- rightJoin = flip joinLeft
409-- Space: O(n) assuming the second stream is cached in memory.
414{-# INLINE joinLeft #-}
415joinLeft :: Monad m =>
416 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
417joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
418 a <- Stream.liftInner s1
419 -- XXX should we use StreamD monad here?
420 -- XXX Is there a better way to perform some action at the end of a loop
427 else Stream.fromPure Nothing
428 b <- fmap Just (Stream.liftInner s2) <> final
436 Nothing -> return (a, Nothing)
439-- | Like 'joinLeft' but uses a hashmap for efficiency.
446{-# INLINE joinLeftMap #-}
447joinLeftMap :: (IsStream t, Ord k, Monad m) =>
448 t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
451 km <- toMap $ IsStream.adapt s2
452 return $ Stream.map (joinAB km) s1
457 case k `Map.lookup` km of
458 Just b -> (k, a, Just b)
459 Nothing -> (k, a, Nothing)
461-- | Like 'joinLeft' but works only on sorted streams.
468{-# INLINE mergeLeftJoin #-}
469mergeLeftJoin :: -- Monad m =>
470 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
471mergeLeftJoin _eq _s1 _s2 = undefined
474-- XXX We can do this concurrently.
476-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
477-- are equal by the given equality pedicate then return the tuple (Just a, Just
478-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
479-- (Nothing, b) for vice-versa.
481-- For space efficiency use the smaller stream as the second stream.
488{-# INLINE joinOuter #-}
489joinOuter :: MonadIO m =>
493 -> SerialT m (Maybe a, Maybe b)
496 inputArr <- Array.fromStream s
497 let len = Array.length inputArr
501 (Stream.fromList (Prelude.replicate len False))
502 return $ go inputArr foundArr <> leftOver inputArr foundArr
506 leftOver inputArr foundArr =
507 let stream1 = IsStream.fromSerial $ Array.read inputArr
508 stream2 = Stream.unfold MA.reader foundArr
511 ( Stream.zipWith (\x y ->
514 else Just (Nothing, Just x)
518 go inputArr foundArr = Stream.evalStateT (return False) $ do
519 a <- Stream.liftInner s1
520 -- XXX should we use StreamD monad here?
521 -- XXX Is there a better way to perform some action at the end of a loop
528 else Stream.fromPure Nothing
530 let stream = IsStream.fromSerial $ Array.read inputArr
531 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
538 MA.putIndex i True foundArr
539 return (Just a, Just b1)
541 Nothing -> return (Just a, Nothing)
544-- Put the b's that have been paired, in another hash or mutate the hash to set
545-- a flag. At the end go through @t m b@ and find those that are not in that
546-- hash to return (Nothing, b).
548-- | Like 'joinOuter' but uses a 'Map' for efficiency.
555{-# INLINE joinOuterMap #-}
557 (IsStream t, Ord k, MonadIO m) =>
558 t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)
561 km1 <- kvFold $ IsStream.adapt s1
562 km2 <- kvFold $ IsStream.adapt s2
564 -- XXX Not sure if toList/fromList would fuse optimally. We may have to
565 -- create a fused Map.toStream function.
566 let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1
569 case k `Map.lookup` km of
570 Just b -> (k, Just a, Just b)
571 Nothing -> (k, Just a, Nothing)
573 -- XXX We can take advantage of the lookups in the first pass above to
574 -- reduce the number of lookups in this pass. If we keep mutable cells
575 -- in the second Map, we can flag it in the first pass and not do any
576 -- lookup in the second pass if it is flagged.
577 let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2
580 case k `Map.lookup` km of
582 Nothing -> Just (k, Nothing, Just b)
584 return $ Stream.serial res1 res2
588 -- XXX Generate error if a duplicate insertion is attempted?
589 kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
591-- | Like 'joinOuter' but works only on sorted streams.
598{-# INLINE mergeOuterJoin #-}
599mergeOuterJoin :: -- Monad m =>
600 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
601mergeOuterJoin _eq _s1 _s2 = undefined
603------------------------------------------------------------------------------
604-- Set operations (special joins)
605------------------------------------------------------------------------------
607-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
608-- the best would be to use an Array with linear search. If the second stream
609-- is sorted we can also use a binary search, using Ord constraint.
611-- | 'intersectBy' is essentially a filtering operation that retains only those
612-- elements in the first stream that are present in the second stream.
614-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
617-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
620-- 'intersectBy' is similar to but not the same as 'joinInner':
622-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
625-- Space: O(n) where @n@ is the number of elements in the second stream.
627-- Time: O(m x n) where @m@ is the number of elements in the first stream and
628-- @n@ is the number of elements in the second stream.
631{-# INLINE intersectBy #-}
632intersectBy :: (IsStream t, Monad m) =>
633 (a -> a -> Bool) -> t m a -> t m a -> t m a
634intersectBy eq s1 s2 =
637 -- This may work well when s2 is small
638 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
639 return $ Stream.filter (\x -> List.any (eq x) xs) s1
641-- | Like 'intersectBy' but works only on streams sorted in ascending order.
648{-# INLINE intersectBySorted #-}
649intersectBySorted :: (IsStream t, Monad m) =>
650 (a -> a -> Ordering) -> t m a -> t m a -> t m a
651intersectBySorted eq s1 =
653 . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
656-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
658-- | Delete first occurrences of those elements from the first stream that are
659-- present in the second stream. If an element occurs multiple times in the
660-- second stream as many occurrences of it are deleted from the first stream.
662-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
665-- The following laws hold:
668-- (s1 `serial` s2) `differenceBy eq` s1 === s2
669-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
672-- Same as the list 'Data.List.//' operation.
674-- Space: O(m) where @m@ is the number of elements in the first stream.
676-- Time: O(m x n) where @m@ is the number of elements in the first stream and
677-- @n@ is the number of elements in the second stream.
680{-# INLINE differenceBy #-}
681differenceBy :: (IsStream t, Monad m) =>
682 (a -> a -> Bool) -> t m a -> t m a -> t m a
683differenceBy eq s1 s2 =
686 -- This may work well if s1 is small
687 -- If s1 is big we can go through s1, deleting elements from s2 and
688 -- not emitting an element if it was successfully deleted from s2.
689 -- we will need a deleteBy that can return whether the element was
691 xs <- Stream.toList $ adapt s1
692 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
694-- | Like 'differenceBy' but works only on sorted streams.
699{-# INLINE mergeDifferenceBy #-}
700mergeDifferenceBy :: -- (IsStream t, Monad m) =>
701 (a -> a -> Ordering) -> t m a -> t m a -> t m a
702mergeDifferenceBy _eq _s1 _s2 = undefined
704-- | This is essentially an append operation that appends all the extra
705-- occurrences of elements from the second stream that are not already present
706-- in the first stream.
708-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
711-- Equivalent to the following except that @s1@ is evaluated only once:
714-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
717-- Similar to 'joinOuter' but not the same.
724{-# INLINE unionBy #-}
725unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
726 (a -> a -> Bool) -> t m a -> t m a -> t m a
730 xs <- Stream.toList $ adapt s2
731 -- XXX we can use postscanlMAfter' instead of IORef
732 ref <- liftIO $ newIORef $! List.nubBy eq xs
734 liftIO $ modifyIORef' ref (List.deleteBy eq x)
738 xs1 <- liftIO $ readIORef ref
739 return $ fromList xs1
740 return $ Stream.mapM f s1 <> s3
742-- | Like 'unionBy' but works only on sorted streams.
747{-# INLINE mergeUnionBy #-}
748mergeUnionBy :: -- (IsStream t, Monad m) =>
749 (a -> a -> Ordering) -> t m a -> t m a -> t m a
750mergeUnionBy _eq _s1 _s2 = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list