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
71import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
72import qualified Streamly.Internal.Data.Fold as Fold
73import qualified Streamly.Internal.Data.Parser as Parser
74import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
75import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream
76import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream
77import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream
78import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
79import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
81import qualified Streamly.Internal.Data.Stream.StreamD as StreamD
83import Prelude hiding (filter, zipWith, concatMap, concat)
87-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
88-- >>> import qualified Streamly.Prelude as Stream
89-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
91------------------------------------------------------------------------------
93------------------------------------------------------------------------------
95-- XXX We can implement this using addition instead of "mod" to make it more
98-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
99-- then every element at strides of @stride@.
101-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
106{-# INLINE sampleFromThen #-}
107sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
108 Int -> Int -> t m a -> t m a
109sampleFromThen offset stride =
110 Stream.with Stream.indexed Stream.filter
111 (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
113-- | Continuously evaluate the input stream and sample the last event in time
114-- window of @n@ seconds.
116-- This is also known as @throttle@ in some libraries.
119-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
124{-# INLINE sampleIntervalEnd #-}
125sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
126 Double -> t m a -> t m a
127sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
129-- | Like 'sampleInterval' but samples at the beginning of the time window.
132-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
137{-# INLINE sampleIntervalStart #-}
138sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
139 Double -> t m a -> t m a
140sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
146 | BurstDoneNext !x !t !x
148{-# INLINE sampleBurst #-}
149sampleBurst :: (IsStream t, MonadAsync m, Functor (t m)) =>
150 Bool -> Double -> t m a -> t m a
151sampleBurst sampleAtEnd gap xs =
152 -- XXX Ideally we should schedule a timer event exactly after gap time,
153 -- but the tick stream should work well as long as the timer
154 -- granularity is small enough compared to the gap.
155 Stream.mapMaybe extract
156 $ Stream.scanl' step BurstNone
158 $ Stream.interjectSuffix 0.01 (return Nothing) (Stream.map Just xs)
162 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
165 step BurstNone (t1, Just x1) = BurstWait t1 x1
166 step BurstNone _ = BurstNone
168 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
169 step (BurstDone _) _ = BurstNone
171 step old@(BurstWait t0 x0) (t1, Nothing)
172 | t1 - t0 >= gap1 = BurstDone x0
174 -- This can happen due to scheduling delays, if we received back to
175 -- back events spaced by more than the timeout without an
176 -- intervening timeout event then we emit the old event instead of
177 -- replacing it by the new.
178 step (BurstWait t0 x0) (t1, Just x1)
179 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
180 | sampleAtEnd = BurstWait t1 x1
181 | otherwise = BurstWait t1 x0
183 step (BurstDoneNext _ t0 x0) (t1, Nothing)
184 | t1 - t0 >= gap1 = BurstDone x0
185 | otherwise = BurstWait t0 x0
186 step (BurstDoneNext _ t0 x0) (t1, Just x1)
187 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
188 | sampleAtEnd = BurstWait t1 x1
189 | otherwise = BurstWait t1 x0
191 {-# INLINE extract #-}
192 extract (BurstDoneNext x _ _) = Just x
193 extract (BurstDone x) = Just x
196-- | Sample one event at the end of each burst of events. A burst is a group
197-- of events close together in time, it ends when an event is spaced by more
198-- than the specified time interval (in seconds) from the previous event.
200-- This is known as @debounce@ in some libraries.
202-- The clock granularity is 10 ms.
206{-# INLINE sampleBurstEnd #-}
207sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
208 Double -> t m a -> t m a
209sampleBurstEnd = sampleBurst True
211-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
212-- instead of at the end of it.
216{-# INLINE sampleBurstStart #-}
217sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
218 Double -> t m a -> t m a
219sampleBurstStart = sampleBurst False
221------------------------------------------------------------------------------
223------------------------------------------------------------------------------
225-- We could possibly choose different algorithms depending on whether the
226-- input stream is almost sorted (ascending/descending) or random. We could
227-- serialize the stream to an array and use quicksort.
229-- | Sort the input stream using a supplied comparison function.
233-- Note: this is not the fastest possible implementation as of now.
238sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
239-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
242 Parser.groupByRollingEither
243 (\x -> (< GT) . cmp x)
246 in Stream.concatPairsWith (Stream.mergeBy cmp) id
247 . Stream.parseMany (fmap (either id id) p)
249------------------------------------------------------------------------------
251------------------------------------------------------------------------------
254-- * https://en.wikipedia.org/wiki/Relational_algebra
255-- * https://en.wikipedia.org/wiki/Join_(SQL)
257-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
258-- constraint, the best would be to use an Array with linear search. If the
259-- second stream is sorted we can also use a binary search, using Ord
260-- constraint or an ordering function.
262-- For Storables we can cache the second stream into an unboxed array for
263-- possibly faster access/compact representation?
265-- If we do not want to keep the stream in memory but always read it from the
266-- source (disk/network) every time we iterate through it then we can do that
267-- too by reading the stream every time, the stream must have immutable state
268-- in that case and the user is responsible for the behavior if the stream
269-- source changes during iterations. We can also use an Unfold instead of
270-- stream. We probably need a way to distinguish streams that can be read
271-- mutliple times without any interference (e.g. unfolding a stream using an
272-- immutable handle would work i.e. using pread/pwrite instead of maintianing
273-- an offset in the handle).
275-- XXX We can do this concurrently.
277-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
280-- The second stream is evaluated multiple times. If the second stream is
281-- consume-once stream then it can be cached in an 'Data.Array.Array' before
282-- calling this function. Caching may also improve performance if the stream is
283-- expensive to evaluate.
288{-# INLINE crossJoin #-}
289crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
291 -- XXX use concatMap instead?
296-- XXX We can do this concurrently.
297-- XXX If the second stream is sorted and passed as an Array we could use
298-- binary search if we have an Ord instance or Ordering returning function. The
299-- time complexity would then become (m x log n).
301-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
302-- are equal by the given equality pedicate then return the tuple (a, b).
304-- The second stream is evaluated multiple times. If the stream is a
305-- consume-once stream then the caller should cache it (e.g. in a
306-- 'Data.Array.Array') before calling this function. Caching may also improve
307-- performance if the stream is expensive to evaluate.
309-- For space efficiency use the smaller stream as the second stream.
311-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
312-- is an order of magnitude faster. joinInner may be used when the second
313-- stream is generated from a seed, therefore, need not be stored in memory and
314-- the amount of memory it takes is a concern.
316-- Space: O(n) assuming the second stream is cached in memory.
321{-# INLINE joinInner #-}
323 forall (t :: (Type -> Type) -> Type -> Type) m a b.
324 (IsStream t, Monad m) =>
325 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
326joinInner eq s1 s2 = do
327 -- ConcatMap works faster than bind
328 Stream.concatMap (\a ->
329 Stream.concatMap (\b ->
331 then Stream.fromPure (a, b)
336-- XXX Generate error if a duplicate insertion is attempted?
337toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
338toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
340-- If the second stream is too big it can be partitioned based on hashes and
341-- then we can process one parition at a time.
343-- XXX An IntMap may be faster when the keys are Int.
344-- XXX Use hashmap instead of map?
346-- | Like 'joinInner' but uses a 'Map' for efficiency.
348-- If the input streams have duplicate keys, the behavior is undefined.
350-- For space efficiency use the smaller stream as the second stream.
357{-# INLINE joinInnerMap #-}
358joinInnerMap :: (IsStream t, Monad m, Ord k) =>
359 t m (k, a) -> t m (k, b) -> t m (k, a, b)
362 km <- toMap $ IsStream.adapt s2
363 pure $ Stream.mapMaybe (joinAB km) s1
368 case k `Map.lookup` kvm of
369 Just b -> Just (k, a, b)
372-- | Like 'joinInner' but works only on sorted streams.
379{-# INLINE joinInnerMerge #-}
380joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
381joinInnerMerge = undefined
383-- XXX We can do this concurrently.
384-- XXX If the second stream is sorted and passed as an Array or a seek capable
385-- stream then we could use binary search if we have an Ord instance or
386-- Ordering returning function. The time complexity would then become (m x log
389-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
390-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
391-- m b@ then return @(a, Nothing)@.
393-- The second stream is evaluated multiple times. If the stream is a
394-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
395-- before calling this function. Caching may also improve performance if the
396-- stream is expensive to evaluate.
399-- rightJoin = flip joinLeft
402-- Space: O(n) assuming the second stream is cached in memory.
407{-# INLINE joinLeft #-}
408joinLeft :: Monad m =>
409 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
410joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
411 a <- Stream.liftInner s1
412 -- XXX should we use StreamD monad here?
413 -- XXX Is there a better way to perform some action at the end of a loop
420 else Stream.fromPure Nothing
421 b <- fmap Just (Stream.liftInner s2) <> final
429 Nothing -> return (a, Nothing)
431-- | Like 'joinLeft' but uses a hashmap for efficiency.
438{-# INLINE joinLeftMap #-}
439joinLeftMap :: (IsStream t, Ord k, Monad m) =>
440 t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
443 km <- toMap $ IsStream.adapt s2
444 return $ Stream.map (joinAB km) s1
449 case k `Map.lookup` km of
450 Just b -> (k, a, Just b)
451 Nothing -> (k, a, Nothing)
453-- | Like 'joinLeft' but works only on sorted streams.
460{-# INLINE mergeLeftJoin #-}
461mergeLeftJoin :: -- Monad m =>
462 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
463mergeLeftJoin _eq _s1 _s2 = undefined
465-- XXX We can do this concurrently.
467-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
468-- are equal by the given equality pedicate then return the tuple (Just a, Just
469-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
470-- (Nothing, b) for vice-versa.
472-- For space efficiency use the smaller stream as the second stream.
479{-# INLINE joinOuter #-}
480joinOuter :: MonadIO m =>
484 -> SerialT m (Maybe a, Maybe b)
487 inputArr <- Array.fromStream s
488 let len = length inputArr
492 (Stream.fromList (Prelude.replicate len False))
493 return $ go inputArr foundArr <> leftOver inputArr foundArr
497 leftOver inputArr foundArr =
498 let stream1 = IsStream.fromSerial $ Array.toStream inputArr
499 stream2 = Stream.unfold MA.read foundArr
502 ( Stream.zipWith (\x y ->
505 else Just (Nothing, Just x)
509 go inputArr foundArr = Stream.evalStateT (return False) $ do
510 a <- Stream.liftInner s1
511 -- XXX should we use StreamD monad here?
512 -- XXX Is there a better way to perform some action at the end of a loop
519 else Stream.fromPure Nothing
521 let stream = IsStream.fromSerial $ Array.toStream inputArr
522 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
529 MA.putIndex i True foundArr
530 return (Just a, Just b1)
532 Nothing -> return (Just a, Nothing)
534-- Put the b's that have been paired, in another hash or mutate the hash to set
535-- a flag. At the end go through @t m b@ and find those that are not in that
536-- hash to return (Nothing, b).
538-- | Like 'joinOuter' but uses a 'Map' for efficiency.
545{-# INLINE joinOuterMap #-}
547 (IsStream t, Ord k, MonadIO m) =>
548 t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)
551 km1 <- kvFold $ IsStream.adapt s1
552 km2 <- kvFold $ IsStream.adapt s2
554 -- XXX Not sure if toList/fromList would fuse optimally. We may have to
555 -- create a fused Map.toStream function.
556 let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1
559 case k `Map.lookup` km of
560 Just b -> (k, Just a, Just b)
561 Nothing -> (k, Just a, Nothing)
563 -- XXX We can take advantage of the lookups in the first pass above to
564 -- reduce the number of lookups in this pass. If we keep mutable cells
565 -- in the second Map, we can flag it in the first pass and not do any
566 -- lookup in the second pass if it is flagged.
567 let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2
570 case k `Map.lookup` km of
572 Nothing -> Just (k, Nothing, Just b)
574 return $ Stream.serial res1 res2
578 -- XXX Generate error if a duplicate insertion is attempted?
579 kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
581-- | Like 'joinOuter' but works only on sorted streams.
588{-# INLINE mergeOuterJoin #-}
589mergeOuterJoin :: -- Monad m =>
590 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
591mergeOuterJoin _eq _s1 _s2 = undefined
593------------------------------------------------------------------------------
594-- Set operations (special joins)
595------------------------------------------------------------------------------
597-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
598-- the best would be to use an Array with linear search. If the second stream
599-- is sorted we can also use a binary search, using Ord constraint.
601-- | 'intersectBy' is essentially a filtering operation that retains only those
602-- elements in the first stream that are present in the second stream.
604-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
607-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
610-- 'intersectBy' is similar to but not the same as 'joinInner':
612-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
615-- Space: O(n) where @n@ is the number of elements in the second stream.
617-- Time: O(m x n) where @m@ is the number of elements in the first stream and
618-- @n@ is the number of elements in the second stream.
621{-# INLINE intersectBy #-}
622intersectBy :: (IsStream t, Monad m) =>
623 (a -> a -> Bool) -> t m a -> t m a -> t m a
624intersectBy eq s1 s2 =
627 -- This may work well when s2 is small
628 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
629 return $ Stream.filter (\x -> List.any (eq x) xs) s1
631-- | Like 'intersectBy' but works only on streams sorted in ascending order.
638{-# INLINE intersectBySorted #-}
639intersectBySorted :: (IsStream t, Monad m) =>
640 (a -> a -> Ordering) -> t m a -> t m a -> t m a
641intersectBySorted eq s1 =
643 . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
646-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
648-- | Delete first occurrences of those elements from the first stream that are
649-- present in the second stream. If an element occurs multiple times in the
650-- second stream as many occurrences of it are deleted from the first stream.
652-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
655-- The following laws hold:
658-- (s1 `serial` s2) `differenceBy eq` s1 === s2
659-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
662-- Same as the list 'Data.List.//' operation.
664-- Space: O(m) where @m@ is the number of elements in the first stream.
666-- Time: O(m x n) where @m@ is the number of elements in the first stream and
667-- @n@ is the number of elements in the second stream.
670{-# INLINE differenceBy #-}
671differenceBy :: (IsStream t, Monad m) =>
672 (a -> a -> Bool) -> t m a -> t m a -> t m a
673differenceBy eq s1 s2 =
676 -- This may work well if s1 is small
677 -- If s1 is big we can go through s1, deleting elements from s2 and
678 -- not emitting an element if it was successfully deleted from s2.
679 -- we will need a deleteBy that can return whether the element was
681 xs <- Stream.toList $ adapt s1
682 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
684-- | Like 'differenceBy' but works only on sorted streams.
689{-# INLINE mergeDifferenceBy #-}
690mergeDifferenceBy :: -- (IsStream t, Monad m) =>
691 (a -> a -> Ordering) -> t m a -> t m a -> t m a
692mergeDifferenceBy _eq _s1 _s2 = undefined
694-- | This is essentially an append operation that appends all the extra
695-- occurrences of elements from the second stream that are not already present
696-- in the first stream.
698-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
701-- Equivalent to the following except that @s1@ is evaluated only once:
704-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
707-- Similar to 'joinOuter' but not the same.
714{-# INLINE unionBy #-}
715unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
716 (a -> a -> Bool) -> t m a -> t m a -> t m a
720 xs <- Stream.toList $ adapt s2
721 -- XXX we can use postscanlMAfter' instead of IORef
722 ref <- liftIO $ newIORef $! List.nubBy eq xs
724 liftIO $ modifyIORef' ref (List.deleteBy eq x)
728 xs1 <- liftIO $ readIORef ref
729 return $ fromList xs1
730 return $ Stream.mapM f s1 <> s3
732-- | Like 'unionBy' but works only on sorted streams.
737{-# INLINE mergeUnionBy #-}
738mergeUnionBy :: -- (IsStream t, Monad m) =>
739 (a -> a -> Ordering) -> t m a -> t m a -> t m a
740mergeUnionBy _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