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)
57-- import Data.Hashable (Hashable)
58import Data.IORef (newIORef, readIORef, modifyIORef')
59import Data.Kind (Type)
60#if !(MIN_VERSION_base(4,11,0))
61import Data.Semigroup (Semigroup(..))
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 Streamly.Internal.Data.Array as Array
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
82import Prelude hiding (filter, zipWith, concatMap, concat)
86-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
87-- >>> import qualified Streamly.Prelude as Stream
88-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
90------------------------------------------------------------------------------
92------------------------------------------------------------------------------
94-- XXX We can implement this using addition instead of "mod" to make it more
97-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
98-- then every element at strides of @stride@.
100-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
105{-# INLINE sampleFromThen #-}
106sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
107 Int -> Int -> t m a -> t m a
108sampleFromThen offset stride =
109 Stream.with Stream.indexed Stream.filter
110 (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
112-- | Continuously evaluate the input stream and sample the last event in time
113-- window of @n@ seconds.
115-- This is also known as @throttle@ in some libraries.
118-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
123{-# INLINE sampleIntervalEnd #-}
124sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
125 Double -> t m a -> t m a
126sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
128-- | Like 'sampleInterval' but samples at the beginning of the time window.
131-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
136{-# INLINE sampleIntervalStart #-}
137sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
138 Double -> t m a -> t m a
139sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
141-- | Sample one event at the end of each burst of events. A burst is a group
142-- of events close together in time, it ends when an event is spaced by more
143-- than the specified time interval from the previous event.
145-- This is known as @debounce@ in some libraries.
147-- The clock granularity is 10 ms.
151{-# INLINE sampleBurstEnd #-}
152sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
153 Double -> t m a -> t m a
155 let f (t1, _) (t2, _) =
156 t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
159 . Stream.groupsByRolling f Fold.last
162-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
163-- instead of at the end of it.
167{-# INLINE sampleBurstStart #-}
168sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
169 Double -> t m a -> t m a
170sampleBurstStart gap =
171 let f (t1, _) (t2, _) =
172 t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
175 . Stream.groupsByRolling f Fold.head
178------------------------------------------------------------------------------
180------------------------------------------------------------------------------
182-- We could possibly choose different algorithms depending on whether the
183-- input stream is almost sorted (ascending/descending) or random. We could
184-- serialize the stream to an array and use quicksort.
186-- | Sort the input stream using a supplied comparison function.
190-- Note: this is not the fastest possible implementation as of now.
195sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a
196-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
199 Parser.groupByRollingEither
200 (\x -> (< GT) . cmp x)
203 in Stream.concatPairsWith (Stream.mergeBy cmp) id
204 . Stream.parseMany (fmap (either id id) p)
206------------------------------------------------------------------------------
208------------------------------------------------------------------------------
211-- * https://en.wikipedia.org/wiki/Relational_algebra
212-- * https://en.wikipedia.org/wiki/Join_(SQL)
214-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
215-- constraint, the best would be to use an Array with linear search. If the
216-- second stream is sorted we can also use a binary search, using Ord
217-- constraint or an ordering function.
219-- For Storables we can cache the second stream into an unboxed array for
220-- possibly faster access/compact representation?
222-- If we do not want to keep the stream in memory but always read it from the
223-- source (disk/network) every time we iterate through it then we can do that
224-- too by reading the stream every time, the stream must have immutable state
225-- in that case and the user is responsible for the behavior if the stream
226-- source changes during iterations. We can also use an Unfold instead of
227-- stream. We probably need a way to distinguish streams that can be read
228-- mutliple times without any interference (e.g. unfolding a stream using an
229-- immutable handle would work i.e. using pread/pwrite instead of maintianing
230-- an offset in the handle).
232-- XXX We can do this concurrently.
234-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
237-- The second stream is evaluated multiple times. If the second stream is
238-- consume-once stream then it can be cached in an 'Data.Array.Array' before
239-- calling this function. Caching may also improve performance if the stream is
240-- expensive to evaluate.
245{-# INLINE crossJoin #-}
246crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
248 -- XXX use concatMap instead?
253-- XXX We can do this concurrently.
254-- XXX If the second stream is sorted and passed as an Array we could use
255-- binary search if we have an Ord instance or Ordering returning function. The
256-- time complexity would then become (m x log n).
258-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
259-- are equal by the given equality pedicate then return the tuple (a, b).
261-- The second stream is evaluated multiple times. If the stream is a
262-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
263-- before calling this function. Caching may also improve performance if the
264-- stream is expensive to evaluate.
266-- For space efficiency use the smaller stream as the second stream.
268-- Space: O(n) assuming the second stream is cached in memory.
273{-# INLINE innerJoin #-}
275 forall (t :: (Type -> Type) -> Type -> Type) m a b.
276 (IsStream t, Monad (t m)) =>
277 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
278innerJoin eq s1 s2 = do
279 -- XXX use concatMap instead?
286-- If the second stream is too big it can be partitioned based on hashes and
287-- then we can process one parition at a time.
289-- | Like 'innerJoin' but uses a hashmap for efficiency.
291-- For space efficiency use the smaller stream as the second stream.
298{-# INLINE hashInnerJoin #-}
299hashInnerJoin :: -- Hashable b =>
300 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
301hashInnerJoin = undefined
303-- | Like 'innerJoin' but works only on sorted streams.
310{-# INLINE mergeInnerJoin #-}
311mergeInnerJoin :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
312mergeInnerJoin = undefined
314-- XXX We can do this concurrently.
315-- XXX If the second stream is sorted and passed as an Array or a seek capable
316-- stream then we could use binary search if we have an Ord instance or
317-- Ordering returning function. The time complexity would then become (m x log
320-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
321-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
322-- m b@ then return @(a, Nothing)@.
324-- The second stream is evaluated multiple times. If the stream is a
325-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
326-- before calling this function. Caching may also improve performance if the
327-- stream is expensive to evaluate.
330-- rightJoin = flip leftJoin
333-- Space: O(n) assuming the second stream is cached in memory.
338{-# INLINE leftJoin #-}
339leftJoin :: Monad m =>
340 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
341leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do
342 a <- Stream.liftInner s1
343 -- XXX should we use StreamD monad here?
344 -- XXX Is there a better way to perform some action at the end of a loop
351 else Stream.fromPure Nothing
352 b <- fmap Just (Stream.liftInner s2) <> final
360 Nothing -> return (a, Nothing)
362-- | Like 'outerJoin' but uses a hashmap for efficiency.
369{-# INLINE hashLeftJoin #-}
370hashLeftJoin :: -- Hashable b =>
371 (a -> b -> Bool) -> t m a -> t m b -> t m (a, Maybe b)
372hashLeftJoin = undefined
374-- | Like 'leftJoin' but works only on sorted streams.
381{-# INLINE mergeLeftJoin #-}
382mergeLeftJoin :: -- Monad m =>
383 (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
384mergeLeftJoin _eq _s1 _s2 = undefined
386-- XXX We can do this concurrently.
388-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
389-- are equal by the given equality pedicate then return the tuple (Just a, Just
390-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
391-- (Nothing, b) for vice-versa.
393-- For space efficiency use the smaller stream as the second stream.
400{-# INLINE outerJoin #-}
401outerJoin :: MonadIO m =>
405 -> SerialT m (Maybe a, Maybe b)
408 arr <- Array.fromStream $ fmap (,False) s
409 return $ go arr <> leftOver arr
414 fmap (\(x, _) -> (Nothing, Just x))
415 . Stream.filter (not . snd)
418 go arr = Stream.evalStateT (return False) $ do
419 a <- Stream.liftInner s1
420 -- XXX should we use StreamD monad here?
421 -- XXX Is there a better way to perform some action at the end of a loop
428 else Stream.fromPure Nothing
430 let stream = IsStream.fromSerial $ Array.toStream arr
431 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
437 -- XXX Need to use a mutable array
438 -- when (not used) $ Array.writeIndex i True
439 return (Just a, Just b1)
441 Nothing -> return (Just a, Nothing)
443-- Put the b's that have been paired, in another hash or mutate the hash to set
444-- a flag. At the end go through @t m b@ and find those that are not in that
445-- hash to return (Nothing, b).
447-- | Like 'outerJoin' but uses a hashmap for efficiency.
449-- For space efficiency use the smaller stream as the second stream.
456{-# INLINE hashOuterJoin #-}
457hashOuterJoin :: -- (Monad m, Hashable b) =>
458 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
459hashOuterJoin _eq _s1 _s2 = undefined
461-- | Like 'outerJoin' but works only on sorted streams.
468{-# INLINE mergeOuterJoin #-}
469mergeOuterJoin :: -- Monad m =>
470 (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
471mergeOuterJoin _eq _s1 _s2 = undefined
473------------------------------------------------------------------------------
474-- Set operations (special joins)
475------------------------------------------------------------------------------
477-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
478-- the best would be to use an Array with linear search. If the second stream
479-- is sorted we can also use a binary search, using Ord constraint.
481-- | 'intersectBy' is essentially a filtering operation that retains only those
482-- elements in the first stream that are present in the second stream.
484-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
487-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
490-- 'intersectBy' is similar to but not the same as 'innerJoin':
492-- >>> Stream.toList $ fmap fst $ Stream.innerJoin (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
495-- Space: O(n) where @n@ is the number of elements in the second stream.
497-- Time: O(m x n) where @m@ is the number of elements in the first stream and
498-- @n@ is the number of elements in the second stream.
501{-# INLINE intersectBy #-}
502intersectBy :: (IsStream t, Monad m) =>
503 (a -> a -> Bool) -> t m a -> t m a -> t m a
504intersectBy eq s1 s2 =
507 -- This may work well when s2 is small
508 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2
509 return $ Stream.filter (\x -> List.any (eq x) xs) s1
511-- | Like 'intersectBy' but works only on sorted streams.
518{-# INLINE mergeIntersectBy #-}
519mergeIntersectBy :: -- (IsStream t, Monad m) =>
520 (a -> a -> Ordering) -> t m a -> t m a -> t m a
521mergeIntersectBy _eq _s1 _s2 = undefined
523-- Roughly leftJoin s1 s2 = s1 `difference` s2 + s1 `intersection` s2
525-- | Delete first occurrences of those elements from the first stream that are
526-- present in the second stream. If an element occurs multiple times in the
527-- second stream as many occurrences of it are deleted from the first stream.
529-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
532-- The following laws hold:
535-- (s1 `serial` s2) `differenceBy eq` s1 === s2
536-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
539-- Same as the list 'Data.List.//' operation.
541-- Space: O(m) where @m@ is the number of elements in the first stream.
543-- Time: O(m x n) where @m@ is the number of elements in the first stream and
544-- @n@ is the number of elements in the second stream.
547{-# INLINE differenceBy #-}
548differenceBy :: (IsStream t, Monad m) =>
549 (a -> a -> Bool) -> t m a -> t m a -> t m a
550differenceBy eq s1 s2 =
553 -- This may work well if s1 is small
554 -- If s1 is big we can go through s1, deleting elements from s2 and
555 -- not emitting an element if it was successfully deleted from s2.
556 -- we will need a deleteBy that can return whether the element was
558 xs <- Stream.toList $ adapt s1
559 fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
561-- | Like 'differenceBy' but works only on sorted streams.
566{-# INLINE mergeDifferenceBy #-}
567mergeDifferenceBy :: -- (IsStream t, Monad m) =>
568 (a -> a -> Ordering) -> t m a -> t m a -> t m a
569mergeDifferenceBy _eq _s1 _s2 = undefined
571-- | This is essentially an append operation that appends all the extra
572-- occurrences of elements from the second stream that are not already present
573-- in the first stream.
575-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
578-- Equivalent to the following except that @s1@ is evaluated only once:
581-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
584-- Similar to 'outerJoin' but not the same.
591{-# INLINE unionBy #-}
592unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
593 (a -> a -> Bool) -> t m a -> t m a -> t m a
597 xs <- Stream.toList $ adapt s2
598 -- XXX we can use postscanlMAfter' instead of IORef
599 ref <- liftIO $ newIORef $! List.nubBy eq xs
601 liftIO $ modifyIORef' ref (List.deleteBy eq x)
605 xs1 <- liftIO $ readIORef ref
606 return $ fromList xs1
607 return $ Stream.mapM f s1 <> s3
609-- | Like 'unionBy' but works only on sorted streams.
614{-# INLINE mergeUnionBy #-}
615mergeUnionBy :: -- (IsStream t, Monad m) =>
616 (a -> a -> Ordering) -> t m a -> t m a -> t m a
617mergeUnionBy _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