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
142-- | Sample one event at the end of each burst of events. A burst is a group
143-- of events close together in time, it ends when an event is spaced by more
144-- than the specified time interval from the previous event.
146-- This is known as @debounce@ in some libraries.
148-- The clock granularity is 10 ms.
152{-# INLINE sampleBurstEnd #-}
153sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
154 Double -> t m a -> t m a
156 let f (t1, _) (t2, _) =
157 t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
160 . Stream.groupsByRolling f Fold.last
163-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
164-- instead of at the end of it.
168{-# INLINE sampleBurstStart #-}
169sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
170 Double -> t m a -> t m a
171sampleBurstStart gap =
172 let f (t1, _) (t2, _) =
173 t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
176 . Stream.groupsByRolling f Fold.head
179------------------------------------------------------------------------------
181------------------------------------------------------------------------------
183-- We could possibly choose different algorithms depending on whether the
184-- input stream is almost sorted (ascending/descending) or random. We could
185-- serialize the stream to an array and use quicksort.
187-- | Sort the input stream using a supplied comparison function.
191-- Note: this is not the fastest possible implementation as of now.
196sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
197-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
200 Parser.groupByRollingEither
201 (\x -> (< GT) . cmp x)
204 in Stream.concatPairsWith (Stream.mergeBy cmp) id
205 . Stream.parseMany (fmap (either id id) p)
207------------------------------------------------------------------------------
209------------------------------------------------------------------------------
212-- * https://en.wikipedia.org/wiki/Relational_algebra
213-- * https://en.wikipedia.org/wiki/Join_(SQL)
215-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
216-- constraint, the best would be to use an Array with linear search. If the
217-- second stream is sorted we can also use a binary search, using Ord
218-- constraint or an ordering function.
220-- For Storables we can cache the second stream into an unboxed array for
221-- possibly faster access/compact representation?
223-- If we do not want to keep the stream in memory but always read it from the
224-- source (disk/network) every time we iterate through it then we can do that
225-- too by reading the stream every time, the stream must have immutable state
226-- in that case and the user is responsible for the behavior if the stream
227-- source changes during iterations. We can also use an Unfold instead of
228-- stream. We probably need a way to distinguish streams that can be read
229-- mutliple times without any interference (e.g. unfolding a stream using an
230-- immutable handle would work i.e. using pread/pwrite instead of maintianing
231-- an offset in the handle).
233-- XXX We can do this concurrently.
235-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
238-- The second stream is evaluated multiple times. If the second stream is
239-- consume-once stream then it can be cached in an 'Data.Array.Array' before
240-- calling this function. Caching may also improve performance if the stream is
241-- expensive to evaluate.
246{-# INLINE crossJoin #-}
247crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
249 -- XXX use concatMap instead?
254-- XXX We can do this concurrently.
255-- XXX If the second stream is sorted and passed as an Array we could use
256-- binary search if we have an Ord instance or Ordering returning function. The
257-- time complexity would then become (m x log n).
259-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
260-- are equal by the given equality pedicate then return the tuple (a, b).
262-- The second stream is evaluated multiple times. If the stream is a
263-- consume-once stream then the caller should cache it (e.g. in a
264-- 'Data.Array.Array') before calling this function. Caching may also improve
265-- performance if the stream is expensive to evaluate.
267-- For space efficiency use the smaller stream as the second stream.
269-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
270-- is an order of magnitude faster. joinInner may be used when the second
271-- stream is generated from a seed, therefore, need not be stored in memory and
272-- the amount of memory it takes is a concern.
274-- Space: O(n) assuming the second stream is cached in memory.
279{-# INLINE joinInner #-}
281 forall (t :: (Type -> Type) -> Type -> Type) m a b.
282 (IsStream t, Monad m) =>
283 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
284joinInner eq s1 s2 = do
285 -- ConcatMap works faster than bind
286 Stream.concatMap (\a ->
287 Stream.concatMap (\b ->
289 then Stream.fromPure (a, b)
294-- XXX Generate error if a duplicate insertion is attempted?
295toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
296toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
298-- If the second stream is too big it can be partitioned based on hashes and
299-- then we can process one parition at a time.
301-- XXX An IntMap may be faster when the keys are Int.
302-- XXX Use hashmap instead of map?
304-- | Like 'joinInner' but uses a 'Map' for efficiency.
306-- If the input streams have duplicate keys, the behavior is undefined.
308-- For space efficiency use the smaller stream as the second stream.
315{-# INLINE joinInnerMap #-}
316joinInnerMap :: (IsStream t, Monad m, Ord k) =>
317 t m (k, a) -> t m (k, b) -> t m (k, a, b)
320 km <- toMap $ IsStream.adapt s2
321 pure $ Stream.mapMaybe (joinAB km) s1
326 case k `Map.lookup` kvm of
327 Just b -> Just (k, a, b)
330-- | Like 'joinInner' but works only on sorted streams.
337{-# INLINE joinInnerMerge #-}
338joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
339joinInnerMerge = undefined
341-- XXX We can do this concurrently.
342-- XXX If the second stream is sorted and passed as an Array or a seek capable
343-- stream then we could use binary search if we have an Ord instance or
344-- Ordering returning function. The time complexity would then become (m x log
347-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
348-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
349-- m b@ then return @(a, Nothing)@.
351-- The second stream is evaluated multiple times. If the stream is a
352-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
353-- before calling this function. Caching may also improve performance if the
354-- stream is expensive to evaluate.
357-- rightJoin = flip joinLeft
360-- Space: O(n) assuming the second stream is cached in memory.
365{-# INLINE joinLeft #-}
366joinLeft :: Monad m =>
367 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
368joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
369 a <- Stream.liftInner s1
370 -- XXX should we use StreamD monad here?
371 -- XXX Is there a better way to perform some action at the end of a loop
378 else Stream.fromPure Nothing
379 b <- fmap Just (Stream.liftInner s2) <> final
387 Nothing -> return (a, Nothing)
389-- | Like 'joinLeft' but uses a hashmap for efficiency.
396{-# INLINE joinLeftMap #-}
397joinLeftMap :: (IsStream t, Ord k, Monad m) =>
398 t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
401 km <- toMap $ IsStream.adapt s2
402 return $ Stream.map (joinAB km) s1
407 case k `Map.lookup` km of
408 Just b -> (k, a, Just b)
409 Nothing -> (k, a, Nothing)
411-- | Like 'joinLeft' but works only on sorted streams.
418{-# INLINE mergeLeftJoin #-}
419mergeLeftJoin :: -- Monad m =>
420 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
421mergeLeftJoin _eq _s1 _s2 = undefined
423-- XXX We can do this concurrently.
425-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
426-- are equal by the given equality pedicate then return the tuple (Just a, Just
427-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
428-- (Nothing, b) for vice-versa.
430-- For space efficiency use the smaller stream as the second stream.
437{-# INLINE joinOuter #-}
438joinOuter :: MonadIO m =>
442 -> SerialT m (Maybe a, Maybe b)
445 inputArr <- Array.fromStream s
446 let len = length inputArr
450 (Stream.fromList (Prelude.replicate len False))
451 return $ go inputArr foundArr <> leftOver inputArr foundArr
455 leftOver inputArr foundArr =
456 let stream1 = IsStream.fromSerial $ Array.toStream inputArr
457 stream2 = Stream.unfold MA.read foundArr
460 ( Stream.zipWith (\x y ->
463 else Just (Nothing, Just x)
467 go inputArr foundArr = Stream.evalStateT (return False) $ do
468 a <- Stream.liftInner s1
469 -- XXX should we use StreamD monad here?
470 -- XXX Is there a better way to perform some action at the end of a loop
477 else Stream.fromPure Nothing
479 let stream = IsStream.fromSerial $ Array.toStream inputArr
480 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
487 MA.putIndex i True foundArr
488 return (Just a, Just b1)
490 Nothing -> return (Just a, Nothing)
492-- Put the b's that have been paired, in another hash or mutate the hash to set
493-- a flag. At the end go through @t m b@ and find those that are not in that
494-- hash to return (Nothing, b).
496-- | Like 'joinOuter' but uses a 'Map' for efficiency.
503{-# INLINE joinOuterMap #-}
505 (IsStream t, Ord k, MonadIO m) =>
506 t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)
509 km1 <- kvFold $ IsStream.adapt s1
510 km2 <- kvFold $ IsStream.adapt s2
512 -- XXX Not sure if toList/fromList would fuse optimally. We may have to
513 -- create a fused Map.toStream function.
514 let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1
517 case k `Map.lookup` km of
518 Just b -> (k, Just a, Just b)
519 Nothing -> (k, Just a, Nothing)
521 -- XXX We can take advantage of the lookups in the first pass above to
522 -- reduce the number of lookups in this pass. If we keep mutable cells
523 -- in the second Map, we can flag it in the first pass and not do any
524 -- lookup in the second pass if it is flagged.
525 let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2
528 case k `Map.lookup` km of
530 Nothing -> Just (k, Nothing, Just b)
532 return $ Stream.serial res1 res2
536 -- XXX Generate error if a duplicate insertion is attempted?
537 kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
539-- | Like 'joinOuter' but works only on sorted streams.
546{-# INLINE mergeOuterJoin #-}
547mergeOuterJoin :: -- Monad m =>
548 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
549mergeOuterJoin _eq _s1 _s2 = undefined
551------------------------------------------------------------------------------
552-- Set operations (special joins)
553------------------------------------------------------------------------------
555-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
556-- the best would be to use an Array with linear search. If the second stream
557-- is sorted we can also use a binary search, using Ord constraint.
559-- | 'intersectBy' is essentially a filtering operation that retains only those
560-- elements in the first stream that are present in the second stream.
562-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
565-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
568-- 'intersectBy' is similar to but not the same as 'joinInner':
570-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
573-- Space: O(n) where @n@ is the number of elements in the second stream.
575-- Time: O(m x n) where @m@ is the number of elements in the first stream and
576-- @n@ is the number of elements in the second stream.
579{-# INLINE intersectBy #-}
580intersectBy :: (IsStream t, Monad m) =>
581 (a -> a -> Bool) -> t m a -> t m a -> t m a
582intersectBy eq s1 s2 =
585 -- This may work well when s2 is small
586 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
587 return $ Stream.filter (\x -> List.any (eq x) xs) s1
589-- | Like 'intersectBy' but works only on streams sorted in ascending order.
596{-# INLINE intersectBySorted #-}
597intersectBySorted :: (IsStream t, Monad m) =>
598 (a -> a -> Ordering) -> t m a -> t m a -> t m a
599intersectBySorted eq s1 =
601 . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
604-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
606-- | Delete first occurrences of those elements from the first stream that are
607-- present in the second stream. If an element occurs multiple times in the
608-- second stream as many occurrences of it are deleted from the first stream.
610-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
613-- The following laws hold:
616-- (s1 `serial` s2) `differenceBy eq` s1 === s2
617-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
620-- Same as the list 'Data.List.//' operation.
622-- Space: O(m) where @m@ is the number of elements in the first stream.
624-- Time: O(m x n) where @m@ is the number of elements in the first stream and
625-- @n@ is the number of elements in the second stream.
628{-# INLINE differenceBy #-}
629differenceBy :: (IsStream t, Monad m) =>
630 (a -> a -> Bool) -> t m a -> t m a -> t m a
631differenceBy eq s1 s2 =
634 -- This may work well if s1 is small
635 -- If s1 is big we can go through s1, deleting elements from s2 and
636 -- not emitting an element if it was successfully deleted from s2.
637 -- we will need a deleteBy that can return whether the element was
639 xs <- Stream.toList $ adapt s1
640 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
642-- | Like 'differenceBy' but works only on sorted streams.
647{-# INLINE mergeDifferenceBy #-}
648mergeDifferenceBy :: -- (IsStream t, Monad m) =>
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
650mergeDifferenceBy _eq _s1 _s2 = undefined
652-- | This is essentially an append operation that appends all the extra
653-- occurrences of elements from the second stream that are not already present
654-- in the first stream.
656-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
659-- Equivalent to the following except that @s1@ is evaluated only once:
662-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
665-- Similar to 'joinOuter' but not the same.
672{-# INLINE unionBy #-}
673unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
674 (a -> a -> Bool) -> t m a -> t m a -> t m a
678 xs <- Stream.toList $ adapt s2
679 -- XXX we can use postscanlMAfter' instead of IORef
680 ref <- liftIO $ newIORef $! List.nubBy eq xs
682 liftIO $ modifyIORef' ref (List.deleteBy eq x)
686 xs1 <- liftIO $ readIORef ref
687 return $ fromList xs1
688 return $ Stream.mapM f s1 <> s3
690-- | Like 'unionBy' but works only on sorted streams.
695{-# INLINE mergeUnionBy #-}
696mergeUnionBy :: -- (IsStream t, Monad m) =>
697 (a -> a -> Ordering) -> t m a -> t m a -> t m a
698mergeUnionBy _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