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
76import qualified Streamly.Internal.Data.Parser as Parser
77 (groupByRollingEither)
78-- import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
79import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream
81import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream
82import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
83import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
84import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
85import qualified Streamly.Internal.Data.Stream as StreamD
87import Prelude hiding (Foldable(..), filter, zipWith, concatMap, concat)
91-- >>> :set -fno-warn-deprecations
92-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
93-- >>> import qualified Streamly.Prelude as Stream
94-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
96------------------------------------------------------------------------------
98------------------------------------------------------------------------------
100-- XXX We can implement this using addition instead of "mod" to make it more
103-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
104-- then every element at strides of @stride@.
106-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
111{-# INLINE sampleFromThen #-}
112sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
113 Int -> Int -> t m a -> t m a
114sampleFromThen offset stride =
115 Stream.with Stream.indexed Stream.filter
116 (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
118-- | Continuously evaluate the input stream and sample the last event in time
119-- window of @n@ seconds.
121-- This is also known as @throttle@ in some libraries.
124-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
129{-# INLINE sampleIntervalEnd #-}
130sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
131 Double -> t m a -> t m a
132sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
134-- | Like 'sampleInterval' but samples at the beginning of the time window.
137-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
142{-# INLINE sampleIntervalStart #-}
143sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
144 Double -> t m a -> t m a
145sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
151 | BurstDoneNext !x !t !x
153{-# INLINE sampleBurst #-}
154sampleBurst :: (IsStream t, MonadAsync m, Functor (t m)) =>
155 Bool -> Double -> t m a -> t m a
156sampleBurst sampleAtEnd gap xs =
157 -- XXX Ideally we should schedule a timer event exactly after gap time,
158 -- but the tick stream should work well as long as the timer
159 -- granularity is small enough compared to the gap.
160 Stream.mapMaybe extract
161 $ Stream.scanl' step BurstNone
163 $ Stream.interjectSuffix 0.01 (return Nothing) (Stream.map Just xs)
167 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
170 step BurstNone (t1, Just x1) = BurstWait t1 x1
171 step BurstNone _ = BurstNone
173 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
174 step (BurstDone _) _ = BurstNone
176 step old@(BurstWait t0 x0) (t1, Nothing)
177 | t1 - t0 >= gap1 = BurstDone x0
179 -- This can happen due to scheduling delays, if we received back to
180 -- back events spaced by more than the timeout without an
181 -- intervening timeout event then we emit the old event instead of
182 -- replacing it by the new.
183 step (BurstWait t0 x0) (t1, Just x1)
184 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
185 | sampleAtEnd = BurstWait t1 x1
186 | otherwise = BurstWait t1 x0
188 step (BurstDoneNext _ t0 x0) (t1, Nothing)
189 | t1 - t0 >= gap1 = BurstDone x0
190 | otherwise = BurstWait t0 x0
191 step (BurstDoneNext _ t0 x0) (t1, Just x1)
192 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
193 | sampleAtEnd = BurstWait t1 x1
194 | otherwise = BurstWait t1 x0
196 {-# INLINE extract #-}
197 extract (BurstDoneNext x _ _) = Just x
198 extract (BurstDone x) = Just x
201-- | Sample one event at the end of each burst of events. A burst is a group
202-- of events close together in time, it ends when an event is spaced by more
203-- than the specified time interval (in seconds) from the previous event.
205-- This is known as @debounce@ in some libraries.
207-- The clock granularity is 10 ms.
211{-# INLINE sampleBurstEnd #-}
212sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
213 Double -> t m a -> t m a
214sampleBurstEnd = sampleBurst True
216-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
217-- instead of at the end of it.
221{-# INLINE sampleBurstStart #-}
222sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
223 Double -> t m a -> t m a
224sampleBurstStart = sampleBurst False
226------------------------------------------------------------------------------
228------------------------------------------------------------------------------
230-- We could possibly choose different algorithms depending on whether the
231-- input stream is almost sorted (ascending/descending) or random. We could
232-- serialize the stream to an array and use quicksort.
234-- | Sort the input stream using a supplied comparison function.
238-- Note: this is not the fastest possible implementation as of now.
243sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
244-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
247 Parser.groupByRollingEither
248 (\x -> (< GT) . cmp x)
249 (fmap fromStream Fold.toStreamKRev)
250 (fmap fromStream Fold.toStreamK)
251 in Stream.concatPairsWith (Stream.mergeBy cmp) id
252 . Stream.rights . Stream.parseMany (fmap (either id id) p)
254------------------------------------------------------------------------------
256------------------------------------------------------------------------------
259-- * https://en.wikipedia.org/wiki/Relational_algebra
260-- * https://en.wikipedia.org/wiki/Join_(SQL)
262-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
263-- constraint, the best would be to use an Array with linear search. If the
264-- second stream is sorted we can also use a binary search, using Ord
265-- constraint or an ordering function.
267-- For Storables we can cache the second stream into an unboxed array for
268-- possibly faster access/compact representation?
270-- If we do not want to keep the stream in memory but always read it from the
271-- source (disk/network) every time we iterate through it then we can do that
272-- too by reading the stream every time, the stream must have immutable state
273-- in that case and the user is responsible for the behavior if the stream
274-- source changes during iterations. We can also use an Unfold instead of
275-- stream. We probably need a way to distinguish streams that can be read
276-- mutliple times without any interference (e.g. unfolding a stream using an
277-- immutable handle would work i.e. using pread/pwrite instead of maintianing
278-- an offset in the handle).
280-- XXX We can do this concurrently.
282-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
285-- The second stream is evaluated multiple times. If the second stream is
286-- consume-once stream then it can be cached in an 'Data.Array.Array' before
287-- calling this function. Caching may also improve performance if the stream is
288-- expensive to evaluate.
293{-# INLINE crossJoin #-}
294crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
296 -- XXX use concatMap instead?
301-- XXX We can do this concurrently.
302-- XXX If the second stream is sorted and passed as an Array we could use
303-- binary search if we have an Ord instance or Ordering returning function. The
304-- time complexity would then become (m x log n).
306-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
307-- are equal by the given equality pedicate then return the tuple (a, b).
309-- The second stream is evaluated multiple times. If the stream is a
310-- consume-once stream then the caller should cache it (e.g. in a
311-- 'Data.Array.Array') before calling this function. Caching may also improve
312-- performance if the stream is expensive to evaluate.
314-- For space efficiency use the smaller stream as the second stream.
316-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
317-- is an order of magnitude faster. joinInner may be used when the second
318-- stream is generated from a seed, therefore, need not be stored in memory and
319-- the amount of memory it takes is a concern.
321-- Space: O(n) assuming the second stream is cached in memory.
326{-# INLINE joinInner #-}
328 forall (t :: (Type -> Type) -> Type -> Type) m a b.
329 (IsStream t, Monad m) =>
330 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
331joinInner eq s1 s2 = do
332 -- ConcatMap works faster than bind
333 Stream.concatMap (\a ->
334 Stream.concatMap (\b ->
336 then Stream.fromPure (a, b)
341-- XXX Generate error if a duplicate insertion is attempted?
342toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
343toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
345-- If the second stream is too big it can be partitioned based on hashes and
346-- then we can process one parition at a time.
348-- XXX An IntMap may be faster when the keys are Int.
349-- XXX Use hashmap instead of map?
351-- | Like 'joinInner' but uses a 'Map' for efficiency.
353-- If the input streams have duplicate keys, the behavior is undefined.
355-- For space efficiency use the smaller stream as the second stream.
362{-# INLINE joinInnerMap #-}
363joinInnerMap :: (IsStream t, Monad m, Ord k) =>
364 t m (k, a) -> t m (k, b) -> t m (k, a, b)
367 km <- toMap $ IsStream.adapt s2
368 pure $ Stream.mapMaybe (joinAB km) s1
373 case k `Map.lookup` kvm of
374 Just b -> Just (k, a, b)
377-- | Like 'joinInner' but works only on sorted streams.
384{-# INLINE joinInnerMerge #-}
385joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
386joinInnerMerge = undefined
389-- XXX We can do this concurrently.
390-- XXX If the second stream is sorted and passed as an Array or a seek capable
391-- stream then we could use binary search if we have an Ord instance or
392-- Ordering returning function. The time complexity would then become (m x log
395-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
396-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
397-- m b@ then return @(a, Nothing)@.
399-- The second stream is evaluated multiple times. If the stream is a
400-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
401-- before calling this function. Caching may also improve performance if the
402-- stream is expensive to evaluate.
405-- rightJoin = flip joinLeft
408-- Space: O(n) assuming the second stream is cached in memory.
413{-# INLINE joinLeft #-}
414joinLeft :: Monad m =>
415 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
416joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
417 a <- Stream.liftInner s1
418 -- XXX should we use StreamD monad here?
419 -- XXX Is there a better way to perform some action at the end of a loop
426 else Stream.fromPure Nothing
427 b <- fmap Just (Stream.liftInner s2) <> final
435 Nothing -> return (a, Nothing)
438-- | Like 'joinLeft' but uses a hashmap for efficiency.
445{-# INLINE joinLeftMap #-}
446joinLeftMap :: (IsStream t, Ord k, Monad m) =>
447 t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
450 km <- toMap $ IsStream.adapt s2
451 return $ Stream.map (joinAB km) s1
456 case k `Map.lookup` km of
457 Just b -> (k, a, Just b)
458 Nothing -> (k, a, Nothing)
460-- | Like 'joinLeft' but works only on sorted streams.
467{-# INLINE mergeLeftJoin #-}
468mergeLeftJoin :: -- Monad m =>
469 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
470mergeLeftJoin _eq _s1 _s2 = undefined
473-- XXX We can do this concurrently.
475-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
476-- are equal by the given equality pedicate then return the tuple (Just a, Just
477-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
478-- (Nothing, b) for vice-versa.
480-- For space efficiency use the smaller stream as the second stream.
487{-# INLINE joinOuter #-}
488joinOuter :: MonadIO m =>
492 -> SerialT m (Maybe a, Maybe b)
495 inputArr <- Array.fromStream s
496 let len = Array.length inputArr
500 (Stream.fromList (Prelude.replicate len False))
501 return $ go inputArr foundArr <> leftOver inputArr foundArr
505 leftOver inputArr foundArr =
506 let stream1 = IsStream.fromSerial $ Array.read inputArr
507 stream2 = Stream.unfold MA.reader foundArr
510 ( Stream.zipWith (\x y ->
513 else Just (Nothing, Just x)
517 go inputArr foundArr = Stream.evalStateT (return False) $ do
518 a <- Stream.liftInner s1
519 -- XXX should we use StreamD monad here?
520 -- XXX Is there a better way to perform some action at the end of a loop
527 else Stream.fromPure Nothing
529 let stream = IsStream.fromSerial $ Array.read inputArr
530 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
537 MA.putIndex i True foundArr
538 return (Just a, Just b1)
540 Nothing -> return (Just a, Nothing)
543-- Put the b's that have been paired, in another hash or mutate the hash to set
544-- a flag. At the end go through @t m b@ and find those that are not in that
545-- hash to return (Nothing, b).
547-- | Like 'joinOuter' but uses a 'Map' for efficiency.
554{-# INLINE joinOuterMap #-}
556 (IsStream t, Ord k, MonadIO m) =>
557 t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)
560 km1 <- kvFold $ IsStream.adapt s1
561 km2 <- kvFold $ IsStream.adapt s2
563 -- XXX Not sure if toList/fromList would fuse optimally. We may have to
564 -- create a fused Map.toStream function.
565 let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1
568 case k `Map.lookup` km of
569 Just b -> (k, Just a, Just b)
570 Nothing -> (k, Just a, Nothing)
572 -- XXX We can take advantage of the lookups in the first pass above to
573 -- reduce the number of lookups in this pass. If we keep mutable cells
574 -- in the second Map, we can flag it in the first pass and not do any
575 -- lookup in the second pass if it is flagged.
576 let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2
579 case k `Map.lookup` km of
581 Nothing -> Just (k, Nothing, Just b)
583 return $ Stream.serial res1 res2
587 -- XXX Generate error if a duplicate insertion is attempted?
588 kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
590-- | Like 'joinOuter' but works only on sorted streams.
597{-# INLINE mergeOuterJoin #-}
598mergeOuterJoin :: -- Monad m =>
599 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
600mergeOuterJoin _eq _s1 _s2 = undefined
602------------------------------------------------------------------------------
603-- Set operations (special joins)
604------------------------------------------------------------------------------
606-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
607-- the best would be to use an Array with linear search. If the second stream
608-- is sorted we can also use a binary search, using Ord constraint.
610-- | 'intersectBy' is essentially a filtering operation that retains only those
611-- elements in the first stream that are present in the second stream.
613-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
616-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
619-- 'intersectBy' is similar to but not the same as 'joinInner':
621-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
624-- Space: O(n) where @n@ is the number of elements in the second stream.
626-- Time: O(m x n) where @m@ is the number of elements in the first stream and
627-- @n@ is the number of elements in the second stream.
630{-# INLINE intersectBy #-}
631intersectBy :: (IsStream t, Monad m) =>
632 (a -> a -> Bool) -> t m a -> t m a -> t m a
633intersectBy eq s1 s2 =
636 -- This may work well when s2 is small
637 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
638 return $ Stream.filter (\x -> List.any (eq x) xs) s1
640-- | Like 'intersectBy' but works only on streams sorted in ascending order.
647{-# INLINE intersectBySorted #-}
648intersectBySorted :: (IsStream t, Monad m) =>
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
650intersectBySorted eq s1 =
652 . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
655-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
657-- | Delete first occurrences of those elements from the first stream that are
658-- present in the second stream. If an element occurs multiple times in the
659-- second stream as many occurrences of it are deleted from the first stream.
661-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
664-- The following laws hold:
667-- (s1 `serial` s2) `differenceBy eq` s1 === s2
668-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
671-- Same as the list 'Data.List.//' operation.
673-- Space: O(m) where @m@ is the number of elements in the first stream.
675-- Time: O(m x n) where @m@ is the number of elements in the first stream and
676-- @n@ is the number of elements in the second stream.
679{-# INLINE differenceBy #-}
680differenceBy :: (IsStream t, Monad m) =>
681 (a -> a -> Bool) -> t m a -> t m a -> t m a
682differenceBy eq s1 s2 =
685 -- This may work well if s1 is small
686 -- If s1 is big we can go through s1, deleting elements from s2 and
687 -- not emitting an element if it was successfully deleted from s2.
688 -- we will need a deleteBy that can return whether the element was
690 xs <- Stream.toList $ adapt s1
691 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
693-- | Like 'differenceBy' but works only on sorted streams.
698{-# INLINE mergeDifferenceBy #-}
699mergeDifferenceBy :: -- (IsStream t, Monad m) =>
700 (a -> a -> Ordering) -> t m a -> t m a -> t m a
701mergeDifferenceBy _eq _s1 _s2 = undefined
703-- | This is essentially an append operation that appends all the extra
704-- occurrences of elements from the second stream that are not already present
705-- in the first stream.
707-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
710-- Equivalent to the following except that @s1@ is evaluated only once:
713-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
716-- Similar to 'joinOuter' but not the same.
723{-# INLINE unionBy #-}
724unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
725 (a -> a -> Bool) -> t m a -> t m a -> t m a
729 xs <- Stream.toList $ adapt s2
730 -- XXX we can use postscanlMAfter' instead of IORef
731 ref <- liftIO $ newIORef $! List.nubBy eq xs
733 liftIO $ modifyIORef' ref (List.deleteBy eq x)
737 xs1 <- liftIO $ readIORef ref
738 return $ fromList xs1
739 return $ Stream.mapM f s1 <> s3
741-- | Like 'unionBy' but works only on sorted streams.
746{-# INLINE mergeUnionBy #-}
747mergeUnionBy :: -- (IsStream t, Monad m) =>
748 (a -> a -> Ordering) -> t m a -> t m a -> t m a
749mergeUnionBy _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