@@ -0,0 +1,578 @@
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
12+module Streamly.Internal.Data.Stream.IsStream.Top
16+ -- | Value agnostic filtering.
18+ , sampleIntervalStart
24+ -- ** Set like operations
25+ -- | These are not exactly set operations because streams are not
26+ -- necessarily sets, they may have duplicated elements.
34+ -- ** Join operations
50+import Control.Monad.IO.Class (MonadIO(..))
51+import Control.Monad.Trans.Class (lift)
52+import Control.Monad.Trans.State.Strict (get, put)
53+-- import Data.Hashable (Hashable)
54+import Data.IORef (newIORef, readIORef, modifyIORef')
55+import Data.Kind (Type)
56+#if !(MIN_VERSION_base(4,11,0))
57+import Data.Semigroup (Semigroup(..))
59+import Streamly.Internal.Data.SVar (MonadAsync)
60+import Streamly.Internal.Data.Stream.IsStream.Common (concatM)
61+import Streamly.Internal.Data.Stream.Prelude (foldl', fromList)
62+import Streamly.Internal.Data.Stream.Serial (SerialT)
63+import Streamly.Internal.Data.Stream.StreamK (IsStream)
64+import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
66+import qualified Data.List as List
67+import qualified Streamly.Internal.Data.Array as Array
68+import qualified Streamly.Internal.Data.Fold as Fold
69+import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
70+import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream
71+import qualified Streamly.Internal.Data.Stream.IsStream.Nesting as Stream
72+import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
73+import qualified Streamly.Internal.Data.Stream.StreamK as StreamK
75+import Prelude hiding (filter, zipWith, concatMap, concat)
77+------------------------------------------------------------------------------
79+------------------------------------------------------------------------------
81+-- XXX We can implement this using addition instead of "mod" to make it more
84+-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
85+-- then every element at strides of @stride@.
88+-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
94+{-# INLINE sampleFromThen #-}
95+sampleFromThen :: (IsStream t, Monad m, Functor (t m)) =>
96+ Int -> Int -> t m a -> t m a
97+sampleFromThen offset stride =
98+ Stream.with Stream.indexed Stream.filter
99+ (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
101+-- | Continuously evaluate the input stream and sample the last event in time
102+-- window of @n@ seconds.
104+-- This is also known as @throttle@ in some libraries.
107+-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
112+{-# INLINE sampleIntervalEnd #-}
113+sampleIntervalEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
114+ Double -> t m a -> t m a
115+sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
117+-- | Like 'sampleInterval' but samples at the beginning of the time window.
120+-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
125+{-# INLINE sampleIntervalStart #-}
126+sampleIntervalStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
127+ Double -> t m a -> t m a
128+sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.head
130+-- | Sample one event at the end of each burst of events. A burst is a group
131+-- of events close together in time, it ends when an event is spaced by more
132+-- than the specified time interval from the previous event.
134+-- This is known as @debounce@ in some libraries.
136+-- The clock granularity is 10 ms.
140+{-# INLINE sampleBurstEnd #-}
141+sampleBurstEnd :: (IsStream t, MonadAsync m, Functor (t m)) =>
142+ Double -> t m a -> t m a
144+ let f (t1, _) (t2, _) =
145+ t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
148+ . Stream.groupsByRolling f Fold.last
149+ . Stream.timeIndexed
151+-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
152+-- instead of at the end of it.
156+{-# INLINE sampleBurstStart #-}
157+sampleBurstStart :: (IsStream t, MonadAsync m, Functor (t m)) =>
158+ Double -> t m a -> t m a
159+sampleBurstStart gap =
160+ let f (t1, _) (t2, _) =
161+ t2 - t1 >= toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
164+ . Stream.groupsByRolling f Fold.head
165+ . Stream.timeIndexed
167+------------------------------------------------------------------------------
169+------------------------------------------------------------------------------
172+-- * https://en.wikipedia.org/wiki/Relational_algebra
173+-- * https://en.wikipedia.org/wiki/Join_(SQL)
175+-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
176+-- constraint, the best would be to use an Array with linear search. If the
177+-- second stream is sorted we can also use a binary search, using Ord
178+-- constraint or an ordering function.
180+-- For Storables we can cache the second stream into an unboxed array for
181+-- possibly faster access/compact representation?
183+-- If we do not want to keep the stream in memory but always read it from the
184+-- source (disk/network) every time we iterate through it then we can do that
185+-- too by reading the stream every time, the stream must have immutable state
186+-- in that case and the user is responsible for the behavior if the stream
187+-- source changes during iterations. We can also use an Unfold instead of
188+-- stream. We probably need a way to distinguish streams that can be read
189+-- mutliple times without any interference (e.g. unfolding a stream using an
190+-- immutable handle would work i.e. using pread/pwrite instead of maintianing
191+-- an offset in the handle).
193+-- XXX We can do this concurrently.
195+-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
198+-- The second stream is evaluated multiple times. If the second stream is
199+-- consume-once stream then it can be cached in an 'Data.Array.Array' before
200+-- calling this function. Caching may also improve performance if the stream is
201+-- expensive to evaluate.
206+{-# INLINE crossJoin #-}
207+crossJoin :: Monad (t m) => t m a -> t m b -> t m (a, b)
209+ -- XXX use concatMap instead?
214+-- XXX We can do this concurrently.
215+-- XXX If the second stream is sorted and passed as an Array we could use
216+-- binary search if we have an Ord instance or Ordering returning function. The
217+-- time complexity would then become (m x log n).
219+-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
220+-- are equal by the given equality pedicate then return the tuple (a, b).
222+-- The second stream is evaluated multiple times. If the stream is a
223+-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
224+-- before calling this function. Caching may also improve performance if the
225+-- stream is expensive to evaluate.
227+-- For space efficiency use the smaller stream as the second stream.
229+-- Space: O(n) assuming the second stream is cached in memory.
234+{-# INLINE innerJoin #-}
236+ forall (t :: (Type -> Type) -> Type -> Type) m a b.
237+ (IsStream t, Monad (t m)) =>
238+ (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
239+innerJoin eq s1 s2 = do
240+ -- XXX use concatMap instead?
247+-- If the second stream is too big it can be partitioned based on hashes and
248+-- then we can process one parition at a time.
250+-- | Like 'innerJoin' but uses a hashmap for efficiency.
252+-- For space efficiency use the smaller stream as the second stream.
259+{-# INLINE hashInnerJoin #-}
260+hashInnerJoin :: -- Hashable b =>
261+ (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
262+hashInnerJoin = undefined
264+-- | Like 'innerJoin' but works only on sorted streams.
271+{-# INLINE mergeInnerJoin #-}
272+mergeInnerJoin :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
273+mergeInnerJoin = undefined
275+-- XXX We can do this concurrently.
276+-- XXX If the second stream is sorted and passed as an Array or a seek capable
277+-- stream then we could use binary search if we have an Ord instance or
278+-- Ordering returning function. The time complexity would then become (m x log
281+-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
282+-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
283+-- m b@ then return @(a, Nothing)@.
285+-- The second stream is evaluated multiple times. If the stream is a
286+-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
287+-- before calling this function. Caching may also improve performance if the
288+-- stream is expensive to evaluate.
291+-- rightJoin = flip leftJoin
294+-- Space: O(n) assuming the second stream is cached in memory.
299+{-# INLINE leftJoin #-}
300+leftJoin :: Monad m =>
301+ (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
302+leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do
303+ a <- Stream.liftInner s1
304+ -- XXX should we use StreamD monad here?
305+ -- XXX Is there a better way to perform some action at the end of a loop
312+ else StreamK.yield Nothing
313+ b <- fmap Just (Stream.liftInner s2) <> final
321+ Nothing -> return (a, Nothing)
323+-- | Like 'outerJoin' but uses a hashmap for efficiency.
330+{-# INLINE hashLeftJoin #-}
331+hashLeftJoin :: -- Hashable b =>
332+ (a -> b -> Bool) -> t m a -> t m b -> t m (a, Maybe b)
333+hashLeftJoin = undefined
335+-- | Like 'leftJoin' but works only on sorted streams.
342+{-# INLINE mergeLeftJoin #-}
343+mergeLeftJoin :: -- Monad m =>
344+ (a -> b -> Ordering) -> t m a -> t m b -> t m (a, Maybe b)
345+mergeLeftJoin _eq _s1 _s2 = undefined
347+-- XXX We can do this concurrently.
349+-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
350+-- are equal by the given equality pedicate then return the tuple (Just a, Just
351+-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
352+-- (Nothing, b) for vice-versa.
354+-- For space efficiency use the smaller stream as the second stream.
361+{-# INLINE outerJoin #-}
362+outerJoin :: MonadIO m =>
366+ -> SerialT m (Maybe a, Maybe b)
369+ arr <- Array.fromStream $ fmap (,False) s
370+ return $ go arr <> leftOver arr
375+ fmap (\(x, _) -> (Nothing, Just x))
376+ . Stream.filter (not . snd)
379+ go arr = Stream.evalStateT (return False) $ do
380+ a <- Stream.liftInner s1
381+ -- XXX should we use StreamD monad here?
382+ -- XXX Is there a better way to perform some action at the end of a loop
389+ else StreamK.yield Nothing
392+ $ fmap Just (Stream.liftInner (Array.toStream arr)) <> final
398+ -- XXX Need to use a mutable array
399+ -- when (not used) $ Array.writeIndex i True
400+ return (Just a, Just b1)
402+ Nothing -> return (Just a, Nothing)
404+-- Put the b's that have been paired, in another hash or mutate the hash to set
405+-- a flag. At the end go through @t m b@ and find those that are not in that
406+-- hash to return (Nothing, b).
408+-- | Like 'outerJoin' but uses a hashmap for efficiency.
410+-- For space efficiency use the smaller stream as the second stream.
417+{-# INLINE hashOuterJoin #-}
418+hashOuterJoin :: -- (Monad m, Hashable b) =>
419+ (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
420+hashOuterJoin _eq _s1 _s2 = undefined
422+-- | Like 'outerJoin' but works only on sorted streams.
429+{-# INLINE mergeOuterJoin #-}
430+mergeOuterJoin :: -- Monad m =>
431+ (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)
432+mergeOuterJoin _eq _s1 _s2 = undefined
434+------------------------------------------------------------------------------
435+-- Set operations (special joins)
436+------------------------------------------------------------------------------
438+-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
439+-- the best would be to use an Array with linear search. If the second stream
440+-- is sorted we can also use a binary search, using Ord constraint.
442+-- | 'intersectBy' is essentially a filtering operation that retains only those
443+-- elements in the first stream that are present in the second stream.
445+-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
448+-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
451+-- 'intersectBy' is similar to but not the same as 'innerJoin':
453+-- >>> Stream.toList $ fmap fst $ Stream.innerJoin (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
456+-- Space: O(n) where @n@ is the number of elements in the second stream.
458+-- Time: O(m x n) where @m@ is the number of elements in the first stream and
459+-- @n@ is the number of elements in the second stream.
462+{-# INLINE intersectBy #-}
463+intersectBy :: (IsStream t, Monad m) =>
464+ (a -> a -> Bool) -> t m a -> t m a -> t m a
465+intersectBy eq s1 s2 =
468+ -- This may work well when s2 is small
469+ xs <- Stream.toListRev $ Stream.uniqBy eq $ StreamK.adapt s2
470+ return $ Stream.filter (\x -> List.any (eq x) xs) s1
472+-- | Like 'intersectBy' but works only on sorted streams.
479+{-# INLINE mergeIntersectBy #-}
480+mergeIntersectBy :: -- (IsStream t, Monad m) =>
481+ (a -> a -> Ordering) -> t m a -> t m a -> t m a
482+mergeIntersectBy _eq _s1 _s2 = undefined
484+-- Roughly leftJoin s1 s2 = s1 `difference` s2 + s1 `intersection` s2
486+-- | Delete first occurrences of those elements from the first stream that are
487+-- present in the second stream. If an element occurs multiple times in the
488+-- second stream as many occurrences of it are deleted from the first stream.
490+-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
493+-- The following laws hold:
496+-- (s1 `serial` s2) `differenceBy eq` s1 === s2
497+-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
500+-- Same as the list 'Data.List.//' operation.
502+-- Space: O(m) where @m@ is the number of elements in the first stream.
504+-- Time: O(m x n) where @m@ is the number of elements in the first stream and
505+-- @n@ is the number of elements in the second stream.
508+{-# INLINE differenceBy #-}
509+differenceBy :: (IsStream t, Monad m) =>
510+ (a -> a -> Bool) -> t m a -> t m a -> t m a
511+differenceBy eq s1 s2 =
514+ -- This may work well if s1 is small
515+ -- If s1 is big we can go through s1, deleting elements from s2 and
516+ -- not emitting an element if it was successfully deleted from s2.
517+ -- we will need a deleteBy that can return whether the element was
519+ xs <- Stream.toList $ StreamK.adapt s1
520+ fmap fromList $ foldl' (flip (List.deleteBy eq)) xs s2
522+-- | Like 'differenceBy' but works only on sorted streams.
527+{-# INLINE mergeDifferenceBy #-}
528+mergeDifferenceBy :: -- (IsStream t, Monad m) =>
529+ (a -> a -> Ordering) -> t m a -> t m a -> t m a
530+mergeDifferenceBy _eq _s1 _s2 = undefined
532+-- | This is essentially an append operation that appends all the extra
533+-- occurrences of elements from the second stream that are not already present
534+-- in the first stream.
536+-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
539+-- Equivalent to the following except that @s1@ is evaluated only once:
542+-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
545+-- Similar to 'outerJoin' but not the same.
552+{-# INLINE unionBy #-}
553+unionBy :: (IsStream t, MonadAsync m, Semigroup (t m a)) =>
554+ (a -> a -> Bool) -> t m a -> t m a -> t m a
558+ xs <- Stream.toList $ StreamK.adapt s2
559+ -- XXX we can use postscanlMAfter' instead of IORef
560+ ref <- liftIO $ newIORef $! List.nubBy eq xs
562+ liftIO $ modifyIORef' ref (List.deleteBy eq x)
566+ xs1 <- liftIO $ readIORef ref
567+ return $ fromList xs1
568+ return $ Stream.mapM f s1 <> s3
570+-- | Like 'unionBy' but works only on sorted streams.
575+{-# INLINE mergeUnionBy #-}
576+mergeUnionBy :: -- (IsStream t, Monad m) =>
577+ (a -> a -> Ordering) -> t m a -> t m a -> t m a
578+mergeUnionBy _eq _s1 _s2 = undefined