@@ -8,73 +8,38 @@
88-- Portability : GHC
99--
1010-- Top level module that can depend on all other lower level Stream modules.
11−--
12−-- Design notes:
13−--
14−-- The order of arguments in the join operations should ideally be opposite. It
15−-- should be such that the infinite stream is the last one. The transformation
16−-- should be on the last argument, so if you curry the functions with all other
17−-- arguments we get a @Stream -> Stream@ function. The first stream argument
18−-- may be considered as a config or modifier for the operation.
19−--
20−-- Benefit of changing the order is that we get a more intuitive Stream ->
21−-- Stream transformation after currying all other arguments. The inner loop
22−-- streams become arguments for the transformation, more like local modifiers
23−-- for the global outer stream as the last argument. Thus we can continue using
24−-- transformations on the outer stream in a composed pipeline. Otherwise we can
25−-- use flip to flip the order.
26−--
27−-- The fact that the inner stream can be used in the loop multiple times also
28−-- tells that this is not the real effectful stream, it is more like a pure
29−-- stream or an array. In fact we may consider using an Identity streams as
30−-- inner streams in which case these functions will not look nice.
31−--
32−-- Downsides:
33−--
34−-- * Maybe less intuitive to think about, because we usually think the first
35−-- stream as the outer loop and second as the inner.
36−-- * Zip and merge operations will continue using the opposite order.
37−-- * Need to change the order of cross, crossWith operations as well
38−-- * It will be inconsistent with Data.List. The functions cannot be used as
39−-- intuitive operators.
40−--
41−-- The choice is similar to concatMap vs bind. concatMap is pipeline
42−-- composition friendly but bind is user intuition friendly. Another option is
43−-- to have other functions with a different argument order e.g. flippedCross
44−-- instead of cross.
45−--
46−-- If we change the order we have to make sure that we have a consistent
47−-- convention for set-like and the cross join operations.
4811
4912module Streamly.Internal.Data.Stream.Top
5013 (
51− -- * Sampling
14+ -- * Transformation
15+ -- ** Sampling
5216 -- | Value agnostic filtering.
5317 strideFromThen
5418
55− -- * Straight Joins
56− -- | These are set-like operations but not exactly set operations because
57− -- streams are not necessarily sets, they may have duplicated elements.
58− -- These operations are generic i.e. they work on streams of unconstrained
59− -- types, therefore, they have quadratic performance characterstics. For
60− -- better performance using Set or Map structures see the
61− -- Streamly.Internal.Data.Stream.Container module.
62− , intersectBy
63− , deleteFirstsBy
64− , unionBy
65−
66− -- Set like operations on sorted streams
67− , sortedIntersectBy
68− , sortedDeleteFirstsBy
69− , sortedUnionBy
70−
71− -- * Cross Joins
72− , innerJoin
73−
74− -- Joins on sorted stream
75− , innerSortedJoin
76− , leftSortedJoin
77− , outerSortedJoin
19+ -- * Nesting
20+ -- ** Set like operations
21+ -- | These are not exactly set operations because streams are not
22+ -- necessarily sets, they may have duplicated elements. These operations
23+ -- are generic i.e. they work on streams of unconstrained types, therefore,
24+ -- they have quadratic performance characterstics. For better performance
25+ -- using Set structures see the Streamly.Internal.Data.Stream.Container
26+ -- module.
27+ , filterInStreamGenericBy
28+ , deleteInStreamGenericBy
29+ , unionWithStreamGenericBy
30+
31+ -- ** Set like operations on sorted streams
32+ , filterInStreamAscBy
33+ , deleteInStreamAscBy
34+ , unionWithStreamAscBy
35+
36+ -- ** Join operations
37+ , joinInnerGeneric
38+
39+ -- * Joins on sorted stream
40+ , joinInnerAscBy
41+ , joinLeftAscBy
42+ , joinOuterAscBy
7843 )
7944where
8045
@@ -83,7 +48,7 @@ where
8348import Control.Monad.IO.Class (MonadIO(..))
8449import Data.IORef (newIORef, readIORef, modifyIORef')
8550import Streamly.Internal.Data.Fold.Type (Fold)
86−import Streamly.Internal.Data.Stream.Type (Stream(..), Step(..), cross)
51+import Streamly.Internal.Data.Stream.Type (Stream, cross)
8752
8853import qualified Data.List as List
8954import qualified Streamly.Internal.Data.Fold as Fold
@@ -145,40 +110,29 @@ strideFromThen offset stride =
145110-- binary search if we have an Ord instance or Ordering returning function. The
146111-- time complexity would then become (m x log n).
147112
148−-- | Like 'cross' but emits only those tuples where @a == b@ using the supplied
149−-- equality predicate. This is essentially a @cross intersection@ of two
150−-- streams.
113+-- | Like 'cross' but emits only those tuples where @a == b@ using the
114+-- supplied equality predicate.
151115--
152116-- Definition:
153117--
154−-- >>> innerJoin eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ Stream.cross s1 s2
155−--
156−-- The second (inner) stream must be finite. Moreover, it must be either pure
157−-- or capable of multiple evaluations. If not then the caller should cache it
158−-- in an 'Data.Array.Array', if the type does not have an 'Unbox' instance then
159−-- use the Generic 'Data.Array.Generic.Array'. Convert the array to stream
160−-- before calling this function. Caching may also improve performance if the
161−-- stream is expensive to evaluate.
118+-- >>> joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ Stream.cross s1 s2
162119--
163−-- If you care about performance this function should be your last choice among
164−-- all inner joins. 'Streamly.Internal.Data.Unfold.innerJoin' is a much faster
165−-- fused alternative. 'innerSortedJoin' is a faster alternative when streams
166−-- are sorted. 'innerOrdJoin' is an order of magnitude faster alternative when
167−-- the type has an 'Ord' instance.
120+-- You should almost always prefer @joinInnerOrd@ over 'joinInnerGeneric' if
121+-- possible. @joinInnerOrd@ is an order of magnitude faster but may take more
122+-- space for caching the second stream.
168123--
169−-- Note: Conceptually, this is a commutative operation. Result includes all the
170−-- elements from the left and the right stream. The order of streams can be
171−-- changed without affecting results, except for the ordering within the tuple.
124+-- See 'Streamly.Internal.Data.Unfold.joinInnerGeneric' for a much faster fused
125+-- alternative.
172126--
173127-- Time: O(m x n)
174128--
175129-- /Pre-release/
176−{-# INLINE innerJoin #-}
177−innerJoin :: Monad m =>
130+{-# INLINE joinInnerGeneric #-}
131+joinInnerGeneric :: Monad m =>
178132 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
179−innerJoin eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
133+joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
180134{-
181−innerJoin eq s1 s2 = do
135+joinInnerGeneric eq s1 s2 = do
182136 -- ConcatMap works faster than bind
183137 Stream.concatMap (\a ->
184138 Stream.concatMap (\b ->
@@ -189,44 +143,44 @@ innerJoin eq s1 s2 = do
189143 ) s1
190144-}
191145
192−-- | A more efficient 'innerJoin' for sorted streams.
146+-- | A more efficient 'joinInner' for sorted streams.
193147--
194148-- Space: O(1)
195149--
196150-- Time: O(m + n)
197151--
198152-- /Unimplemented/
199−{-# INLINE innerSortedJoin #-}
200−innerSortedJoin ::
153+{-# INLINE joinInnerAscBy #-}
154+joinInnerAscBy ::
201155 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
202−innerSortedJoin = undefined
156+joinInnerAscBy = undefined
203157
204−-- | A more efficient 'leftJoin' for sorted streams.
158+-- | A more efficient 'joinLeft' for sorted streams.
205159--
206160-- Space: O(1)
207161--
208162-- Time: O(m + n)
209163--
210164-- /Unimplemented/
211−{-# INLINE leftSortedJoin #-}
212−leftSortedJoin :: -- Monad m =>
165+{-# INLINE joinLeftAscBy #-}
166+joinLeftAscBy :: -- Monad m =>
213167 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
214−leftSortedJoin _eq _s1 _s2 = undefined
168+joinLeftAscBy _eq _s1 _s2 = undefined
215169
216−-- | A more efficient 'outerJoin' for sorted streams.
170+-- | A more efficient 'joinOuter' for sorted streams.
217171--
218172-- Space: O(1)
219173--
220174-- Time: O(m + n)
221175--
222176-- /Unimplemented/
223−{-# INLINE outerSortedJoin #-}
224−outerSortedJoin :: -- Monad m =>
177+{-# INLINE joinOuterAscBy #-}
178+joinOuterAscBy :: -- Monad m =>
225179 (a -> b -> Ordering)
226180 -> Stream m a
227181 -> Stream m b
228182 -> Stream m (Maybe a, Maybe b)
229−outerSortedJoin _eq _s1 _s2 = undefined
183+joinOuterAscBy _eq _s1 _s2 = undefined
230184
231185------------------------------------------------------------------------------
232186-- Set operations (special joins)
@@ -236,8 +190,8 @@ outerSortedJoin _eq _s1 _s2 = undefined
236190-- the best would be to use an Array with linear search. If the second stream
237191-- is sorted we can also use a binary search, using Ord constraint.
238192
239−-- | Keep only those elements in the first stream that are present in the
240−-- second stream too. The second stream is folded to a container using the
193+-- | Keep only those elements in the second stream that are present in the
194+-- first stream too. The first stream is folded to a container using the
241195-- supplied fold and then the elements in the container are looked up using the
242196-- supplied lookup function.
243197--
@@ -252,30 +206,23 @@ filterStreamWith :: Monad m =>
252206filterStreamWith fld member s1 s2 =
253207 Stream.concatEffect
254208 $ do
255− xs <- Stream.fold fld s2
256− return $ Stream.filter (`member` xs) s1
257−
258−-- XXX instead of folding the second stream to a list we could use it directly.
259−-- If the user wants they can generate the stream from an array and also call
260−-- uniq or nub on it. We can provide a convenience Stream -> Stream to cache
261−-- a finite stream in an array and serve it from the cache. The user can decide
262−-- what is best based on the context. They can also choose to use a boxed or
263−-- unboxed array for caching. To force caching we can make the second stream
264−-- monad type Identity. But that may be less flexible. One option is to use
265−-- cachedIntersectBy etc for automatic caching.
209+ xs <- Stream.fold fld s1
210+ return $ Stream.filter (`member` xs) s2
266211
267−-- | 'intersectBy' returns a subsequence of the first stream which intersects
268−-- with the second stream. Note that this is not a commutative operation unlike
269−-- a set intersection, because of duplicate elements in the stream the order of
270−-- the streams matters. This is similar to 'Data.List.intersectBy'. Note that
271−-- intersectBy is a special case of 'innerJoin'.
212+-- | 'filterInStreamGenericBy' retains only those elements in the second stream that
213+-- are present in the first stream.
272214--
273−-- >>> f s1 s2 = Stream.fold Fold.toList $ Stream.intersectBy (==) (Stream.fromList s1) (Stream.fromList s2)
274−-- >>> f [1,3,4,4,5]) [2,3,4,5,5]
275−-- [3,4,4,5]
215+-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
216+-- [2,1,1]
276217--
277−-- First stream can be infinite, the second stream must be finite and must be
278−-- capable of multiple evaluations.
218+-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
219+-- [1,2,2]
220+--
221+-- Similar to the list intersectBy operation but with the stream argument order
222+-- flipped.
223+--
224+-- The first stream must be finite and must not block. Second stream is
225+-- processed only after the first stream is fully realized.
279226--
280227-- Space: O(n) where @n@ is the number of elements in the second stream.
281228--
@@ -283,87 +230,46 @@ filterStreamWith fld member s1 s2 =
283230-- @n@ is the number of elements in the second stream.
284231--
285232-- /Pre-release/
286−{-# INLINE intersectBy #-}
287−intersectBy :: Monad m =>
233+{-# INLINE filterInStreamGenericBy #-}
234+filterInStreamGenericBy :: Monad m =>
288235 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
289−intersectBy eq =
236+filterInStreamGenericBy eq =
290237 -- XXX Use an (unboxed) array instead.
291238 filterStreamWith
292239 (Fold.scanMaybe (Fold.uniqBy eq) Fold.toListRev)
293240 (List.any . eq)
294241
295−-------------------------------------------------------------------------------
296−-- Intersection of sorted streams
297−-------------------------------------------------------------------------------
298−
299−-- XXX The sort order is not important as long both the streams have the same
300−-- sort order. We need to move only in one direction in each stream.
301−-- XXX Fix the argument order to use the same behavior as intersectBy.
302−
303−-- | Like 'intersectBy' but assumes that the input streams are sorted in
242+-- | Like 'filterInStreamGenericBy' but assumes that the input streams are sorted in
304243-- ascending order. To use it on streams sorted in descending order pass an
305244-- inverted comparison function returning GT for less than and LT for greater
306245-- than.
307246--
308−-- Both streams can be infinite.
309−--
310247-- Space: O(1)
311248--
312249-- Time: O(m+n)
313250--
314251-- /Pre-release/
315−{-# INLINE_NORMAL sortedIntersectBy #-}
316−sortedIntersectBy :: Monad m =>
252+{-# INLINE filterInStreamAscBy #-}
253+filterInStreamAscBy :: Monad m =>
317254 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
318−sortedIntersectBy cmp (Stream stepa ta) (Stream stepb tb) =
319− Stream step
320− ( ta -- left stream state
321− , tb -- right stream state
322− , Nothing -- left value
323− , Nothing -- right value
324− )
325−
326− where
255+filterInStreamAscBy eq s1 s2 = Stream.intersectBySorted eq s2 s1
327256
328− {-# INLINE_LATE step #-}
329− -- step 1, fetch the first value
330− step gst (sa, sb, Nothing, b) = do
331− r <- stepa gst sa
332− return $ case r of
333− Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3
334− Skip sa' -> Skip (sa', sb, Nothing, b)
335− Stop -> Stop
336−
337− -- step 2, fetch the second value
338− step gst (sa, sb, a@(Just _), Nothing) = do
339− r <- stepb gst sb
340− return $ case r of
341− Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3
342− Skip sb' -> Skip (sa, sb', a, Nothing)
343− Stop -> Stop
344−
345− -- step 3, compare the two values
346− step _ (sa, sb, Just a, Just b) = do
347− let res = cmp a b
348− return $ case res of
349− GT -> Skip (sa, sb, Just a, Nothing) -- step 2
350− LT -> Skip (sa, sb, Nothing, Just b) -- step 1
351− EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1
352−
353−-- | Returns a subsequence of the first stream, deleting first occurrences of
354−-- those elements that are present in the second stream. Note that this is not
355−-- a commutative operation. This is similar to the 'Data.List.deleteFirstsBy'.
257+-- | Delete all elements of the first stream from the seconds stream. If an
258+-- element occurs multiple times in the first stream as many occurrences of it
259+-- are deleted from the second stream.
356260--
357−-- >>> f xs ys = Stream.fold Fold.toList $ Stream.deleteFirstsBy (==) (Stream.fromList xs) (Stream.fromList ys)
358−-- >>> f [1,2,2,3,3,5] [1,2,2,3,4]
359−-- [2,3,5]
261+-- >>> Stream.fold Fold.toList $ Stream.deleteInStreamGenericBy (==) (Stream.fromList [1,2,3]) (Stream.fromList [1,2,2])
262+-- [2]
360263--
361−-- The following holds:
264+-- The following laws hold:
362265--
363−-- > deleteFirstsBy (==) (Stream.nub s2 `append` s1) s2 === s1
364−-- > deleteFirstsBy (==) (Stream.nub s2 `interleave` s1) s2 === s1
266+-- > deleteInStreamGenericBy (==) s1 (s1 `append` s2) === s2
267+-- > deleteInStreamGenericBy (==) s1 (s1 `interleave` s2) === s2
365268--
366−-- First stream can be infinite, second stream must be finite.
269+-- Same as the list 'Data.List.//' operation but with argument order flipped.
270+--
271+-- The first stream must be finite and must not block. Second stream is
272+-- processed only after the first stream is fully realized.
367273--
368274-- Space: O(m) where @m@ is the number of elements in the first stream.
369275--
@@ -371,83 +277,58 @@ sortedIntersectBy cmp (Stream stepa ta) (Stream stepb tb) =
371277-- @n@ is the number of elements in the second stream.
372278--
373279-- /Pre-release/
374−{-# INLINE deleteFirstsBy #-}
375−deleteFirstsBy :: Monad m =>
280+{-# INLINE deleteInStreamGenericBy #-}
281+deleteInStreamGenericBy :: Monad m =>
376282 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
377−deleteFirstsBy eq s2 s1 =
378− -- XXX s2 can be a sorted mutable array and we can use binary
379− -- search to find. Mark the element deleted, count the deletions
380− -- and reconsolidate the array when a min number of elements is
381− -- deleted.
382−
383− -- XXX Use StreamK or list as second argument instead of Stream to avoid
384− -- concatEffect?
385− Stream.concatEffect $ do
386− xs <- Stream.toList s1
387− -- It reverses the list but that is fine.
388− let del x =
389− List.foldl' (\(ys,res) y ->
390− if x `eq` y
391− then (ys, True)
392− else (x:ys, res)) ([], False)
393− g (ys,_) x =
394− let (ys1, deleted) = del x ys
395− in if deleted
396− then (ys1, Nothing)
397− else (ys1, Just x)
398− in return
399− $ Stream.catMaybes
400− $ fmap snd
401− $ Stream.postscanl' g (xs, Nothing) s2
283+deleteInStreamGenericBy eq s1 s2 =
284+ Stream.concatEffect
285+ $ do
286+ -- This may work well if s1 is small
287+ -- If s1 is big we can go through s1, deleting elements from s2 and
288+ -- not emitting an element if it was successfully deleted from s2.
289+ -- we will need a deleteBy that can return whether the element was
290+ -- deleted or not.
291+ xs <- Stream.fold Fold.toList s2
292+ let f = Fold.foldl' (flip (List.deleteBy eq)) xs
293+ fmap Stream.fromList $ Stream.fold f s1
402294
403−-- | A more efficient 'deleteFirstsBy' for streams sorted in ascending order.
404−--
405−-- Both streams can be infinite.
295+-- | A more efficient 'deleteInStreamGenericBy' for streams sorted in ascending order.
406296--
407297-- Space: O(1)
408298--
409299-- /Unimplemented/
410−{-# INLINE sortedDeleteFirstsBy #-}
411−sortedDeleteFirstsBy :: -- (Monad m) =>
300+{-# INLINE deleteInStreamAscBy #-}
301+deleteInStreamAscBy :: -- (Monad m) =>
412302 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
413−sortedDeleteFirstsBy _eq _s1 _s2 = undefined
303+deleteInStreamAscBy _eq _s1 _s2 = undefined
414304
415305-- XXX Remove the MonadIO constraint. We can just cache one stream and then
416306-- implement using differenceEqBy.
417307
418−-- | Returns the first stream appended with those unique elements from the
419−-- second stream that are not already present in the first stream. Note that
420−-- this is not a commutative operation unlike a set union, argument order
421−-- matters. The behavior is similar to the 'Data.List.unionBy'.
308+-- | This essentially appends to the second stream all the occurrences of
309+-- elements in the first stream that are not already present in the second
310+-- stream.
422311--
423312-- Equivalent to the following except that @s2@ is evaluated only once:
424313--
425−-- >>> unionBy eq s1 s2 = s1 `Stream.append` Stream.deleteFirstsBy eq s1 (Stream.nub s2)
314+-- >>> unionWithStreamGenericBy eq s1 s2 = s2 `Stream.append` (Stream.deleteInStreamGenericBy eq s2 s1)
426315--
427316-- Example:
428317--
429−-- >>> f s1 s2 = Stream.fold Fold.toList $ Stream.unionBy (==) (Stream.fromList s1) (Stream.fromList s2)
430−-- >>> f [1,2,2,4] [1,1,2,3,3]
318+-- >>> Stream.fold Fold.toList $ Stream.unionWithStreamGenericBy (==) (Stream.fromList [1,1,2,3]) (Stream.fromList [1,2,2,4])
431319-- [1,2,2,4,3]
432320--
433−-- First stream can be infinite, but second stream must be finite. Note that if
434−-- the first stream is infinite the union means just the first stream. Thus
435−-- union is useful only when both streams are finite. See 'sortedUnionBy' where
436−-- union can work on infinite streams if they are sorted.
437−--
438321-- Space: O(n)
439322--
440323-- Time: O(m x n)
441324--
442325-- /Pre-release/
443−{-# INLINE unionBy #-}
444−unionBy :: MonadIO m =>
326+{-# INLINE unionWithStreamGenericBy #-}
327+unionWithStreamGenericBy :: MonadIO m =>
445328 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
446−unionBy eq s2 s1 =
329+unionWithStreamGenericBy eq s1 s2 =
447330 Stream.concatEffect
448331 $ do
449− -- XXX use a rewrite rule such that if a list converted to stream
450− -- is passed to unionBy then this becomes an identity operation.
451332 xs <- Stream.fold Fold.toList s1
452333 -- XXX we can use postscanlMAfter' instead of IORef
453334 ref <- liftIO $ newIORef $! List.nubBy eq xs
@@ -460,21 +341,12 @@ unionBy eq s2 s1 =
460341 return $ Stream.fromList xs1
461342 return $ Stream.mapM f s2 `Stream.append` s3
462343
463−-- | A more efficient 'unionBy' for sorted streams.
464−--
465−-- Note that the behavior is different from 'unionBy'. In 'unionBy' we append
466−-- the unique elements from second stream only after exhausting the first one
467−-- whereas in sorted streams we can determine unique elements early even when
468−-- we are going through the first stream. Thus the result is an interleaving of
469−-- the two streams, merging those elements from the second stream that are not
470−-- present in the first.
344+-- | A more efficient 'unionWithStreamGenericBy' for sorted streams.
471345--
472346-- Space: O(1)
473347--
474−-- Both streams can be infinite.
475−--
476348-- /Unimplemented/
477−{-# INLINE sortedUnionBy #-}
478−sortedUnionBy :: -- (Monad m) =>
349+{-# INLINE unionWithStreamAscBy #-}
350+unionWithStreamAscBy :: -- (Monad m) =>
479351 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
480−sortedUnionBy _eq _s1 _s2 = undefined
352+unionWithStreamAscBy _eq _s1 _s2 = undefined