2-- Module : Streamly.Internal.Data.Stream.IsStream.Top
3-- Copyright : (c) 2020 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Top level IsStream module that can use all other lower level IsStream
12module Streamly.Internal.Data.Stream.IsStream.Top
16 -- | Value agnostic filtering.
27 -- ** Set like operations
28 -- | These are not exactly set operations because streams are not
29 -- necessarily sets, they may have duplicated elements.
53import Control.Monad.Catch (MonadCatch)
54import Control.Monad.IO.Class (MonadIO(..))
55import Control.Monad.Trans.Class (lift)
56import Control.Monad.Trans.State.Strict (get, put)
57import Data.Function ((&))
58import Data.IORef (newIORef, readIORef, modifyIORef')
59import Data.Kind (Type)
60import Data.Maybe (isJust)
61import Streamly.Internal.Control.Concurrent (MonadAsync)
62import Streamly.Internal.Data.Stream.IsStream.Common (concatM)
63import Streamly.Internal.Data.Stream.IsStream.Type
64 (IsStream(..), adapt, foldl', fromList)
65import Streamly.Internal.Data.Stream.Serial (SerialT)
66import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
68import qualified Data.List as List
69import qualified Data.Map.Strict as Map
70import qualified Streamly.Internal.Data.Array as Array
71 (fromStream, length, toStream)
72import qualified Streamly.Data.Array.Unboxed.Mut as MA
73import qualified Streamly.Internal.Data.Fold as Fold
74 (one, last, toStream, toStreamRev)
75import qualified Streamly.Internal.Data.Parser as Parser
76 (groupByRollingEither)
77import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
78import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream
79import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream
81import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
82import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
83import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
84import qualified Streamly.Internal.Data.Stream.StreamD as StreamD
86import Prelude hiding (filter, zipWith, concatMap, concat)
90-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
91-- >>> import qualified Streamly.Prelude as Stream
92-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
94------------------------------------------------------------------------------
96------------------------------------------------------------------------------
98-- XXX We can implement this using addition instead of "mod" to make it more
101-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
102-- then every element at strides of @stride@.
104-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
109{-# INLINE sampleFromThen #-}
110sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
111 Int -> Int -> t m a -> t m a
112sampleFromThen offset stride =
113 Stream.with Stream.indexed Stream.filter
114 (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
116-- | Continuously evaluate the input stream and sample the last event in time
117-- window of @n@ seconds.
119-- This is also known as @throttle@ in some libraries.
122-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
127{-# INLINE sampleIntervalEnd #-}
128sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
129 Double -> t m a -> t m a
130sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
132-- | Like 'sampleInterval' but samples at the beginning of the time window.
135-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
140{-# INLINE sampleIntervalStart #-}
141sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
142 Double -> t m a -> t m a
143sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
149 | BurstDoneNext !x !t !x
151{-# INLINE sampleBurst #-}
152sampleBurst :: (IsStream t, MonadAsync m, Functor (t m)) =>
153 Bool -> Double -> t m a -> t m a
154sampleBurst sampleAtEnd gap xs =
155 -- XXX Ideally we should schedule a timer event exactly after gap time,
156 -- but the tick stream should work well as long as the timer
157 -- granularity is small enough compared to the gap.
158 Stream.mapMaybe extract
159 $ Stream.scanl' step BurstNone
161 $ Stream.interjectSuffix 0.01 (return Nothing) (Stream.map Just xs)
165 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
168 step BurstNone (t1, Just x1) = BurstWait t1 x1
169 step BurstNone _ = BurstNone
171 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
172 step (BurstDone _) _ = BurstNone
174 step old@(BurstWait t0 x0) (t1, Nothing)
175 | t1 - t0 >= gap1 = BurstDone x0
177 -- This can happen due to scheduling delays, if we received back to
178 -- back events spaced by more than the timeout without an
179 -- intervening timeout event then we emit the old event instead of
180 -- replacing it by the new.
181 step (BurstWait t0 x0) (t1, Just x1)
182 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
183 | sampleAtEnd = BurstWait t1 x1
184 | otherwise = BurstWait t1 x0
186 step (BurstDoneNext _ t0 x0) (t1, Nothing)
187 | t1 - t0 >= gap1 = BurstDone x0
188 | otherwise = BurstWait t0 x0
189 step (BurstDoneNext _ t0 x0) (t1, Just x1)
190 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
191 | sampleAtEnd = BurstWait t1 x1
192 | otherwise = BurstWait t1 x0
194 {-# INLINE extract #-}
195 extract (BurstDoneNext x _ _) = Just x
196 extract (BurstDone x) = Just x
199-- | Sample one event at the end of each burst of events. A burst is a group
200-- of events close together in time, it ends when an event is spaced by more
201-- than the specified time interval (in seconds) from the previous event.
203-- This is known as @debounce@ in some libraries.
205-- The clock granularity is 10 ms.
209{-# INLINE sampleBurstEnd #-}
210sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
211 Double -> t m a -> t m a
212sampleBurstEnd = sampleBurst True
214-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
215-- instead of at the end of it.
219{-# INLINE sampleBurstStart #-}
220sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
221 Double -> t m a -> t m a
222sampleBurstStart = sampleBurst False
224------------------------------------------------------------------------------
226------------------------------------------------------------------------------
228-- We could possibly choose different algorithms depending on whether the
229-- input stream is almost sorted (ascending/descending) or random. We could
230-- serialize the stream to an array and use quicksort.
232-- | Sort the input stream using a supplied comparison function.
236-- Note: this is not the fastest possible implementation as of now.
241sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
242-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
245 Parser.groupByRollingEither
246 (\x -> (< GT) . cmp x)
249 in Stream.concatPairsWith (Stream.mergeBy cmp) id
250 . Stream.parseMany (fmap (either id id) p)
252------------------------------------------------------------------------------
254------------------------------------------------------------------------------
257-- * https://en.wikipedia.org/wiki/Relational_algebra
258-- * https://en.wikipedia.org/wiki/Join_(SQL)
260-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
261-- constraint, the best would be to use an Array with linear search. If the
262-- second stream is sorted we can also use a binary search, using Ord
263-- constraint or an ordering function.
265-- For Storables we can cache the second stream into an unboxed array for
266-- possibly faster access/compact representation?
268-- If we do not want to keep the stream in memory but always read it from the
269-- source (disk/network) every time we iterate through it then we can do that
270-- too by reading the stream every time, the stream must have immutable state
271-- in that case and the user is responsible for the behavior if the stream
272-- source changes during iterations. We can also use an Unfold instead of
273-- stream. We probably need a way to distinguish streams that can be read
274-- mutliple times without any interference (e.g. unfolding a stream using an
275-- immutable handle would work i.e. using pread/pwrite instead of maintianing
276-- an offset in the handle).
278-- XXX We can do this concurrently.
280-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
283-- The second stream is evaluated multiple times. If the second stream is
284-- consume-once stream then it can be cached in an 'Data.Array.Array' before
285-- calling this function. Caching may also improve performance if the stream is
286-- expensive to evaluate.
291{-# INLINE crossJoin #-}
292crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
294 -- XXX use concatMap instead?
299-- XXX We can do this concurrently.
300-- XXX If the second stream is sorted and passed as an Array we could use
301-- binary search if we have an Ord instance or Ordering returning function. The
302-- time complexity would then become (m x log n).
304-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
305-- are equal by the given equality pedicate then return the tuple (a, b).
307-- The second stream is evaluated multiple times. If the stream is a
308-- consume-once stream then the caller should cache it (e.g. in a
309-- 'Data.Array.Array') before calling this function. Caching may also improve
310-- performance if the stream is expensive to evaluate.
312-- For space efficiency use the smaller stream as the second stream.
314-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
315-- is an order of magnitude faster. joinInner may be used when the second
316-- stream is generated from a seed, therefore, need not be stored in memory and
317-- the amount of memory it takes is a concern.
319-- Space: O(n) assuming the second stream is cached in memory.
324{-# INLINE joinInner #-}
326 forall (t :: (Type -> Type) -> Type -> Type) m a b.
327 (IsStream t, Monad m) =>
328 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
329joinInner eq s1 s2 = do
330 -- ConcatMap works faster than bind
331 Stream.concatMap (\a ->
332 Stream.concatMap (\b ->
334 then Stream.fromPure (a, b)
339-- XXX Generate error if a duplicate insertion is attempted?
340toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
341toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
343-- If the second stream is too big it can be partitioned based on hashes and
344-- then we can process one parition at a time.
346-- XXX An IntMap may be faster when the keys are Int.
347-- XXX Use hashmap instead of map?
349-- | Like 'joinInner' but uses a 'Map' for efficiency.
351-- If the input streams have duplicate keys, the behavior is undefined.
353-- For space efficiency use the smaller stream as the second stream.
360{-# INLINE joinInnerMap #-}
361joinInnerMap :: (IsStream t, Monad m, Ord k) =>
362 t m (k, a) -> t m (k, b) -> t m (k, a, b)
365 km <- toMap $ IsStream.adapt s2
366 pure $ Stream.mapMaybe (joinAB km) s1
371 case k `Map.lookup` kvm of
372 Just b -> Just (k, a, b)
375-- | Like 'joinInner' but works only on sorted streams.
382{-# INLINE joinInnerMerge #-}
383joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
384joinInnerMerge = undefined
386-- XXX We can do this concurrently.
387-- XXX If the second stream is sorted and passed as an Array or a seek capable
388-- stream then we could use binary search if we have an Ord instance or
389-- Ordering returning function. The time complexity would then become (m x log
392-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
393-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
394-- m b@ then return @(a, Nothing)@.
396-- The second stream is evaluated multiple times. If the stream is a
397-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
398-- before calling this function. Caching may also improve performance if the
399-- stream is expensive to evaluate.
402-- rightJoin = flip joinLeft
405-- Space: O(n) assuming the second stream is cached in memory.
410{-# INLINE joinLeft #-}
411joinLeft :: Monad m =>
412 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
413joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
414 a <- Stream.liftInner s1
415 -- XXX should we use StreamD monad here?
416 -- XXX Is there a better way to perform some action at the end of a loop
423 else Stream.fromPure Nothing
424 b <- fmap Just (Stream.liftInner s2) <> final
432 Nothing -> return (a, Nothing)
434-- | Like 'joinLeft' but uses a hashmap for efficiency.
441{-# INLINE joinLeftMap #-}
442joinLeftMap :: (IsStream t, Ord k, Monad m) =>
443 t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
446 km <- toMap $ IsStream.adapt s2
447 return $ Stream.map (joinAB km) s1
452 case k `Map.lookup` km of
453 Just b -> (k, a, Just b)
454 Nothing -> (k, a, Nothing)
456-- | Like 'joinLeft' but works only on sorted streams.
463{-# INLINE mergeLeftJoin #-}
464mergeLeftJoin :: -- Monad m =>
465 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
466mergeLeftJoin _eq _s1 _s2 = undefined
468-- XXX We can do this concurrently.
470-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
471-- are equal by the given equality pedicate then return the tuple (Just a, Just
472-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
473-- (Nothing, b) for vice-versa.
475-- For space efficiency use the smaller stream as the second stream.
482{-# INLINE joinOuter #-}
483joinOuter :: MonadIO m =>
487 -> SerialT m (Maybe a, Maybe b)
490 inputArr <- Array.fromStream s
491 let len = Array.length inputArr
495 (Stream.fromList (Prelude.replicate len False))
496 return $ go inputArr foundArr <> leftOver inputArr foundArr
500 leftOver inputArr foundArr =
501 let stream1 = IsStream.fromSerial $ Array.toStream inputArr
502 stream2 = Stream.unfold MA.read foundArr
505 ( Stream.zipWith (\x y ->
508 else Just (Nothing, Just x)
512 go inputArr foundArr = Stream.evalStateT (return False) $ do
513 a <- Stream.liftInner s1
514 -- XXX should we use StreamD monad here?
515 -- XXX Is there a better way to perform some action at the end of a loop
522 else Stream.fromPure Nothing
524 let stream = IsStream.fromSerial $ Array.toStream inputArr
525 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
532 MA.putIndex i True foundArr
533 return (Just a, Just b1)
535 Nothing -> return (Just a, Nothing)
537-- Put the b's that have been paired, in another hash or mutate the hash to set
538-- a flag. At the end go through @t m b@ and find those that are not in that
539-- hash to return (Nothing, b).
541-- | Like 'joinOuter' but uses a 'Map' for efficiency.
548{-# INLINE joinOuterMap #-}
550 (IsStream t, Ord k, MonadIO m) =>
551 t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)
554 km1 <- kvFold $ IsStream.adapt s1
555 km2 <- kvFold $ IsStream.adapt s2
557 -- XXX Not sure if toList/fromList would fuse optimally. We may have to
558 -- create a fused Map.toStream function.
559 let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1
562 case k `Map.lookup` km of
563 Just b -> (k, Just a, Just b)
564 Nothing -> (k, Just a, Nothing)
566 -- XXX We can take advantage of the lookups in the first pass above to
567 -- reduce the number of lookups in this pass. If we keep mutable cells
568 -- in the second Map, we can flag it in the first pass and not do any
569 -- lookup in the second pass if it is flagged.
570 let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2
573 case k `Map.lookup` km of
575 Nothing -> Just (k, Nothing, Just b)
577 return $ Stream.serial res1 res2
581 -- XXX Generate error if a duplicate insertion is attempted?
582 kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
584-- | Like 'joinOuter' but works only on sorted streams.
591{-# INLINE mergeOuterJoin #-}
592mergeOuterJoin :: -- Monad m =>
593 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
594mergeOuterJoin _eq _s1 _s2 = undefined
596------------------------------------------------------------------------------
597-- Set operations (special joins)
598------------------------------------------------------------------------------
600-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
601-- the best would be to use an Array with linear search. If the second stream
602-- is sorted we can also use a binary search, using Ord constraint.
604-- | 'intersectBy' is essentially a filtering operation that retains only those
605-- elements in the first stream that are present in the second stream.
607-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
610-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
613-- 'intersectBy' is similar to but not the same as 'joinInner':
615-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
618-- Space: O(n) where @n@ is the number of elements in the second stream.
620-- Time: O(m x n) where @m@ is the number of elements in the first stream and
621-- @n@ is the number of elements in the second stream.
624{-# INLINE intersectBy #-}
625intersectBy :: (IsStream t, Monad m) =>
626 (a -> a -> Bool) -> t m a -> t m a -> t m a
627intersectBy eq s1 s2 =
630 -- This may work well when s2 is small
631 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
632 return $ Stream.filter (\x -> List.any (eq x) xs) s1
634-- | Like 'intersectBy' but works only on streams sorted in ascending order.
641{-# INLINE intersectBySorted #-}
642intersectBySorted :: (IsStream t, Monad m) =>
643 (a -> a -> Ordering) -> t m a -> t m a -> t m a
644intersectBySorted eq s1 =
646 . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
649-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
651-- | Delete first occurrences of those elements from the first stream that are
652-- present in the second stream. If an element occurs multiple times in the
653-- second stream as many occurrences of it are deleted from the first stream.
655-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
658-- The following laws hold:
661-- (s1 `serial` s2) `differenceBy eq` s1 === s2
662-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
665-- Same as the list 'Data.List.//' operation.
667-- Space: O(m) where @m@ is the number of elements in the first stream.
669-- Time: O(m x n) where @m@ is the number of elements in the first stream and
670-- @n@ is the number of elements in the second stream.
673{-# INLINE differenceBy #-}
674differenceBy :: (IsStream t, Monad m) =>
675 (a -> a -> Bool) -> t m a -> t m a -> t m a
676differenceBy eq s1 s2 =
679 -- This may work well if s1 is small
680 -- If s1 is big we can go through s1, deleting elements from s2 and
681 -- not emitting an element if it was successfully deleted from s2.
682 -- we will need a deleteBy that can return whether the element was
684 xs <- Stream.toList $ adapt s1
685 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
687-- | Like 'differenceBy' but works only on sorted streams.
692{-# INLINE mergeDifferenceBy #-}
693mergeDifferenceBy :: -- (IsStream t, Monad m) =>
694 (a -> a -> Ordering) -> t m a -> t m a -> t m a
695mergeDifferenceBy _eq _s1 _s2 = undefined
697-- | This is essentially an append operation that appends all the extra
698-- occurrences of elements from the second stream that are not already present
699-- in the first stream.
701-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
704-- Equivalent to the following except that @s1@ is evaluated only once:
707-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
710-- Similar to 'joinOuter' but not the same.
717{-# INLINE unionBy #-}
718unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
719 (a -> a -> Bool) -> t m a -> t m a -> t m a
723 xs <- Stream.toList $ adapt s2
724 -- XXX we can use postscanlMAfter' instead of IORef
725 ref <- liftIO $ newIORef $! List.nubBy eq xs
727 liftIO $ modifyIORef' ref (List.deleteBy eq x)
731 xs1 <- liftIO $ readIORef ref
732 return $ fromList xs1
733 return $ Stream.mapM f s1 <> s3
735-- | Like 'unionBy' but works only on sorted streams.
740{-# INLINE mergeUnionBy #-}
741mergeUnionBy :: -- (IsStream t, Monad m) =>
742 (a -> a -> Ordering) -> t m a -> t m a -> t m a
743mergeUnionBy _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