on master help sign in
Rename, document, update joins/set operations
a6a15e4Harendra Kumar
core/src/Streamly/Internal/Data/Stream/Container.hs+56−48
@@ -11,16 +11,17 @@
11
11
12module Streamly.Internal.Data.Stream.Container
12module Streamly.Internal.Data.Stream.Container
13 (
13 (
14− nub
14+ -- * Deduplication
15+ ordNub
15
16
16− -- * Joins for unconstrained types
17+ -- * Joins
17− , joinLeftGeneric
18+ , leftJoin
18− , joinOuterGeneric
19+ , outerJoin
19
20
20− -- * Joins with Ord constraint
21+ -- * Ord Joins
21− , joinInner
22+ , innerOrdJoin
22− , joinLeft
23+ , leftOrdJoin
23− , joinOuter
24+ , outerOrdJoin
24 )
25 )
25where
26where
26
27
@@ -47,12 +48,15 @@ import qualified Streamly.Internal.Data.Stream.Transformer as Stream
47
48
48#include "DocTestDataStream.hs"
49#include "DocTestDataStream.hs"
49
50
51+-- | @nub@ specialized to 'Ord' types for better performance. Returns a
52+-- subsequence of the stream removing any duplicate elements.
53+--
50−-- | The memory used is proportional to the number of unique elements in the
54+-- The memory used is proportional to the number of unique elements in the
51−-- stream. If we want to limit the memory we can just use "take" to limit the
55+-- stream. One way to limit the memory is to use @take@ on the resulting
52−-- uniq elements in the stream.
56+-- stream to limit the unique elements in the stream.
53−{-# INLINE_NORMAL nub #-}
57+{-# INLINE_NORMAL ordNub #-}
54−nub :: (Monad m, Ord a) => Stream m a -> Stream m a
58+ordNub :: (Monad m, Ord a) => Stream m a -> Stream m a
55−nub (Stream step1 state1) = Stream step (Set.empty, state1)
59+ordNub (Stream step1 state1) = Stream step (Set.empty, state1)
56
60
57 where
61 where
58
62
@@ -78,8 +82,8 @@ toMap =
78--
82--
79-- XXX An IntMap may be faster when the keys are Int.
83-- XXX An IntMap may be faster when the keys are Int.
80-- XXX Use hashmap instead of map?
84-- XXX Use hashmap instead of map?
81−--
85+
82−-- | Like 'joinInner' but uses a 'Map' for efficiency.
86+-- | 'innerJoin' specialized to 'Ord' types for better performance.
83--
87--
84-- If the input streams have duplicate keys, the behavior is undefined.
88-- If the input streams have duplicate keys, the behavior is undefined.
85--
89--
@@ -90,10 +94,10 @@ toMap =
90-- Time: O(m + n)
94-- Time: O(m + n)
91--
95--
92-- /Pre-release/
96-- /Pre-release/
93−{-# INLINE joinInner #-}
97+{-# INLINE innerOrdJoin #-}
94−joinInner :: (Monad m, Ord k) =>
98+innerOrdJoin :: (Monad m, Ord k) =>
95 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, b)
99 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, b)
96−joinInner s1 s2 =
100+innerOrdJoin s1 s2 =
97 Stream.concatEffect $ do
101 Stream.concatEffect $ do
98 km <- toMap s2
102 km <- toMap s2
99 pure $ Stream.mapMaybe (joinAB km) s1
103 pure $ Stream.mapMaybe (joinAB km) s1
@@ -106,32 +110,33 @@ joinInner s1 s2 =
106 Nothing -> Nothing
110 Nothing -> Nothing
107
111
108-- XXX We can do this concurrently.
112-- XXX We can do this concurrently.
113+-- XXX Check performance of StreamD vs StreamK
109-- XXX If the second stream is sorted and passed as an Array or a seek capable
114-- XXX If the second stream is sorted and passed as an Array or a seek capable
110-- stream then we could use binary search if we have an Ord instance or
115-- stream then we could use binary search if we have an Ord instance or
111-- Ordering returning function. The time complexity would then become (m x log
116-- Ordering returning function. The time complexity would then become (m x log
112-- n).
117-- n).
113
118
114−-- XXX Check performance of StreamD vs StreamK
119+-- | Like 'innerJoin' but emits @(a, Just b)@ whenever a and b are equal, for
115−
116−-- | Like 'joinInner' but emit @(a, Just b)@, and additionally, for those @a@'s
117−-- that are not equal to any @b@ emit @(a, Nothing)@.
120+-- those @a@'s that are not equal to any @b@ emits @(a, Nothing)@.
118--
121--
119−-- The second stream is evaluated multiple times. If the stream is a
122+-- This is a generalization of 'innerJoin' to include all elements from the
120−-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
123+-- left stream and not just those which have an equal in the right stream. This
121−-- before calling this function. Caching may also improve performance if the
124+-- is not a commutative operation, the order of the stream arguments matters.
122−-- stream is expensive to evaluate.
123--
125--
124−-- >>> joinRightGeneric eq = flip (Stream.joinLeftGeneric eq)
126+-- All the caveats mentioned in 'innerJoin' apply here as well. Right join is
127+-- not provided because it is just a flipped left join:
128+--
129+-- >>> rightJoin eq = flip (Stream.leftJoin eq)
125--
130--
126-- Space: O(n) assuming the second stream is cached in memory.
131-- Space: O(n) assuming the second stream is cached in memory.
127--
132--
128-- Time: O(m x n)
133-- Time: O(m x n)
129--
134--
130-- /Unimplemented/
135-- /Unimplemented/
131−{-# INLINE joinLeftGeneric #-}
136+{-# INLINE leftJoin #-}
132−joinLeftGeneric :: Monad m =>
137+leftJoin :: Monad m =>
133 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
138 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
134−joinLeftGeneric eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
139+leftJoin eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
135 a <- mkCross (Stream.liftInner s1)
140 a <- mkCross (Stream.liftInner s1)
136 -- XXX should we use StreamD monad here?
141 -- XXX should we use StreamD monad here?
137 -- XXX Is there a better way to perform some action at the end of a loop
142 -- XXX Is there a better way to perform some action at the end of a loop
@@ -152,19 +157,17 @@ joinLeftGeneric eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
152 else mkCross Stream.nil
157 else mkCross Stream.nil
153 Nothing -> return (a, Nothing)
158 Nothing -> return (a, Nothing)
154
159
155−-- XXX rename to joinLeftOrd?
160+-- | 'leftJoin' specialized to 'Ord' types for better performance.
156−
157−-- | A more efficient 'joinLeft' using a hashmap for efficiency.
158--
161--
159-- Space: O(n)
162-- Space: O(n)
160--
163--
161-- Time: O(m + n)
164-- Time: O(m + n)
162--
165--
163-- /Pre-release/
166-- /Pre-release/
164−{-# INLINE joinLeft #-}
167+{-# INLINE leftOrdJoin #-}
165−joinLeft :: (Ord k, Monad m) =>
168+leftOrdJoin :: (Ord k, Monad m) =>
166 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, Maybe b)
169 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, Maybe b)
167−joinLeft s1 s2 =
170+leftOrdJoin s1 s2 =
168 Stream.concatEffect $ do
171 Stream.concatEffect $ do
169 km <- toMap s2
172 km <- toMap s2
170 return $ fmap (joinAB km) s1
173 return $ fmap (joinAB km) s1
@@ -177,14 +180,19 @@ joinLeft s1 s2 =
177 Nothing -> (k, a, Nothing)
180 Nothing -> (k, a, Nothing)
178
181
179-- XXX We can do this concurrently.
182-- XXX We can do this concurrently.
183+-- XXX Check performance of StreamD vs StreamK cross operation.
180
184
181−-- XXX Check performance of StreamD vs StreamK
182−
183−-- | Like 'joinLeft' but emits a @(Just a, Just b)@. Like 'joinLeft', for those
185+-- | Like 'leftJoin' but emits a @(Just a, Just b)@. Like 'leftJoin', for those
184-- @a@'s that are not equal to any @b@ emit @(Just a, Nothing)@, but
186-- @a@'s that are not equal to any @b@ emit @(Just a, Nothing)@, but
185-- additionally, for those @b@'s that are not equal to any @a@ emit @(Nothing,
187-- additionally, for those @b@'s that are not equal to any @a@ emit @(Nothing,
186-- Just b)@.
188-- Just b)@.
187--
189--
190+-- This is a generalization of left join to include all the elements from the
191+-- right stream as well, in other words it is a combination of left and right
192+-- joins. This is a commutative operation. The order of stream arguments can be
193+-- changed without affecting results, except for the ordering of elements in
194+-- the resulting tuple.
195+--
188-- For space efficiency use the smaller stream as the second stream.
196-- For space efficiency use the smaller stream as the second stream.
189--
197--
190-- Space: O(n)
198-- Space: O(n)
@@ -192,15 +200,15 @@ joinLeft s1 s2 =
192-- Time: O(m x n)
200-- Time: O(m x n)
193--
201--
194-- /Pre-release/
202-- /Pre-release/
195−{-# INLINE joinOuterGeneric #-}
203+{-# INLINE outerJoin #-}
196−joinOuterGeneric :: MonadIO m =>
204+outerJoin :: MonadIO m =>
197 (a -> b -> Bool)
205 (a -> b -> Bool)
198 -> Stream m a
206 -> Stream m a
199 -> Stream m b
207 -> Stream m b
200 -> Stream m (Maybe a, Maybe b)
208 -> Stream m (Maybe a, Maybe b)
201−joinOuterGeneric eq s1 s =
209+outerJoin eq s1 s2 =
202 Stream.concatEffect $ do
210 Stream.concatEffect $ do
203− inputArr <- Array.fromStream s
211+ inputArr <- Array.fromStream s2
204 let len = Array.length inputArr
212 let len = Array.length inputArr
205 foundArr <-
213 foundArr <-
206 Stream.fold
214 Stream.fold
@@ -254,18 +262,18 @@ joinOuterGeneric eq s1 s =
254-- a flag. At the end go through @Stream m b@ and find those that are not in that
262-- a flag. At the end go through @Stream m b@ and find those that are not in that
255-- hash to return (Nothing, b).
263-- hash to return (Nothing, b).
256
264
257−-- | Like 'joinOuter' but uses a 'Map' for efficiency.
265+-- | 'outerJoin' specialized to 'Ord' types for better performance.
258--
266--
259-- Space: O(m + n)
267-- Space: O(m + n)
260--
268--
261-- Time: O(m + n)
269-- Time: O(m + n)
262--
270--
263-- /Pre-release/
271-- /Pre-release/
264−{-# INLINE joinOuter #-}
272+{-# INLINE outerOrdJoin #-}
265−joinOuter ::
273+outerOrdJoin ::
266 (Ord k, MonadIO m) =>
274 (Ord k, MonadIO m) =>
267 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, Maybe a, Maybe b)
275 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, Maybe a, Maybe b)
268−joinOuter s1 s2 =
276+outerOrdJoin s1 s2 =
269 Stream.concatEffect $ do
277 Stream.concatEffect $ do
270 km1 <- kvFold s1
278 km1 <- kvFold s1
271 km2 <- kvFold s2
279 km2 <- kvFold s2
core/src/Streamly/Internal/Data/Stream/Nesting.hs+2−46
@@ -141,9 +141,8 @@ module Streamly.Internal.Data.Stream.Nesting
141
141
142 -- * Transform (Nested Containers)
142 -- * Transform (Nested Containers)
143 -- | Opposite to compact in ArrayStream
143 -- | Opposite to compact in ArrayStream
144− , splitInnerBy
144+ , splitInnerBy -- XXX innerSplitOn
145− , splitInnerBySuffix
145+ , splitInnerBySuffix -- XXX innerSplitOnSuffix
146− , intersectBySorted
147
146
148 -- * Reduce By Streams
147 -- * Reduce By Streams
149 , dropPrefix
148 , dropPrefix
@@ -593,49 +592,6 @@ mergeFstBy :: -- Monad m =>
593mergeFstBy _f _m1 _m2 = undefined
592mergeFstBy _f _m1 _m2 = undefined
594 -- fromStreamK $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
593 -- fromStreamK $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
595
594
596−-------------------------------------------------------------------------------
597−-- Intersection of sorted streams
598−-------------------------------------------------------------------------------
599−
600−-- Assuming the streams are sorted in ascending order
601−{-# INLINE_NORMAL intersectBySorted #-}
602−intersectBySorted :: Monad m
603− => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
604−intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =
605− Stream step
606− ( ta -- left stream state
607− , tb -- right stream state
608− , Nothing -- left value
609− , Nothing -- right value
610− )
611−
612− where
613−
614− {-# INLINE_LATE step #-}
615− -- step 1, fetch the first value
616− step gst (sa, sb, Nothing, b) = do
617− r <- stepa gst sa
618− return $ case r of
619− Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3
620− Skip sa' -> Skip (sa', sb, Nothing, b)
621− Stop -> Stop
622−
623− -- step 2, fetch the second value
624− step gst (sa, sb, a@(Just _), Nothing) = do
625− r <- stepb gst sb
626− return $ case r of
627− Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3
628− Skip sb' -> Skip (sa, sb', a, Nothing)
629− Stop -> Stop
630−
631− -- step 3, compare the two values
632− step _ (sa, sb, Just a, Just b) = do
633− let res = cmp a b
634− return $ case res of
635− GT -> Skip (sa, sb, Just a, Nothing) -- step 2
636− LT -> Skip (sa, sb, Nothing, Just b) -- step 1
637− EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1
638−
639------------------------------------------------------------------------------
595------------------------------------------------------------------------------
640-- Combine N Streams - unfoldMany
596-- Combine N Streams - unfoldMany
641------------------------------------------------------------------------------
597------------------------------------------------------------------------------
core/src/Streamly/Internal/Data/Stream/Top.hs+243−115
@@ -8,38 +8,73 @@
8-- Portability : GHC
8-- Portability : GHC
9--
9--
10-- Top level module that can depend on all other lower level Stream modules.
10-- 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.
11
48
12module Streamly.Internal.Data.Stream.Top
49module Streamly.Internal.Data.Stream.Top
13 (
50 (
14− -- * Transformation
15− -- ** Sampling
51+ -- * Sampling
16 -- | Value agnostic filtering.
52 -- | Value agnostic filtering.
17 strideFromThen
53 strideFromThen
18
54
19− -- * Nesting
55+ -- * Straight Joins
20− -- ** Set like operations
21− -- | These are not exactly set operations because streams are not
56+ -- | These are set-like operations but not exactly set operations because
22− -- necessarily sets, they may have duplicated elements. These operations
57+ -- streams are not necessarily sets, they may have duplicated elements.
23− -- are generic i.e. they work on streams of unconstrained types, therefore,
58+ -- These operations are generic i.e. they work on streams of unconstrained
24− -- they have quadratic performance characterstics. For better performance
59+ -- types, therefore, they have quadratic performance characterstics. For
60+ -- better performance using Set or Map structures see the
25− -- using Set structures see the Streamly.Internal.Data.Stream.Container
61+ -- Streamly.Internal.Data.Stream.Container module.
26− -- module.
62+ , intersectBy
27− , filterInStreamGenericBy
63+ , deleteFirstsBy
28− , deleteInStreamGenericBy
64+ , unionBy
29− , unionWithStreamGenericBy
65+
30−
31− -- ** Set like operations on sorted streams
66+ -- Set like operations on sorted streams
32− , filterInStreamAscBy
67+ , sortedIntersectBy
33− , deleteInStreamAscBy
68+ , sortedDeleteFirstsBy
34− , unionWithStreamAscBy
69+ , sortedUnionBy
35−
70+
36− -- ** Join operations
71+ -- * Cross Joins
37− , joinInnerGeneric
72+ , innerJoin
38−
73+
39− -- * Joins on sorted stream
74+ -- Joins on sorted stream
40− , joinInnerAscBy
75+ , innerSortedJoin
41− , joinLeftAscBy
76+ , leftSortedJoin
42− , joinOuterAscBy
77+ , outerSortedJoin
43 )
78 )
44where
79where
45
80
@@ -48,7 +83,7 @@ where
48import Control.Monad.IO.Class (MonadIO(..))
83import Control.Monad.IO.Class (MonadIO(..))
49import Data.IORef (newIORef, readIORef, modifyIORef')
84import Data.IORef (newIORef, readIORef, modifyIORef')
50import Streamly.Internal.Data.Fold.Type (Fold)
85import Streamly.Internal.Data.Fold.Type (Fold)
51−import Streamly.Internal.Data.Stream.Type (Stream, cross)
86+import Streamly.Internal.Data.Stream.Type (Stream(..), Step(..), cross)
52
87
53import qualified Data.List as List
88import qualified Data.List as List
54import qualified Streamly.Internal.Data.Fold as Fold
89import qualified Streamly.Internal.Data.Fold as Fold
@@ -110,29 +145,40 @@ strideFromThen offset stride =
110-- binary search if we have an Ord instance or Ordering returning function. The
145-- binary search if we have an Ord instance or Ordering returning function. The
111-- time complexity would then become (m x log n).
146-- time complexity would then become (m x log n).
112
147
113−-- | Like 'cross' but emits only those tuples where @a == b@ using the
148+-- | Like 'cross' but emits only those tuples where @a == b@ using the supplied
114−-- supplied equality predicate.
149+-- equality predicate. This is essentially a @cross intersection@ of two
150+-- streams.
115--
151--
116-- Definition:
152-- Definition:
117--
153--
118−-- >>> joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ Stream.cross s1 s2
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.
119--
162--
120−-- You should almost always prefer @joinInnerOrd@ over 'joinInnerGeneric' if
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
121−-- possible. @joinInnerOrd@ is an order of magnitude faster but may take more
166+-- are sorted. 'innerOrdJoin' is an order of magnitude faster alternative when
122−-- space for caching the second stream.
167+-- the type has an 'Ord' instance.
123--
168--
124−-- See 'Streamly.Internal.Data.Unfold.joinInnerGeneric' for a much faster fused
169+-- Note: Conceptually, this is a commutative operation. Result includes all the
125−-- alternative.
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.
126--
172--
127-- Time: O(m x n)
173-- Time: O(m x n)
128--
174--
129-- /Pre-release/
175-- /Pre-release/
130−{-# INLINE joinInnerGeneric #-}
176+{-# INLINE innerJoin #-}
131−joinInnerGeneric :: Monad m =>
177+innerJoin :: Monad m =>
132 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
178 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
133−joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
179+innerJoin eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
134{-
180{-
135−joinInnerGeneric eq s1 s2 = do
181+innerJoin eq s1 s2 = do
136 -- ConcatMap works faster than bind
182 -- ConcatMap works faster than bind
137 Stream.concatMap (\a ->
183 Stream.concatMap (\a ->
138 Stream.concatMap (\b ->
184 Stream.concatMap (\b ->
@@ -143,44 +189,44 @@ joinInnerGeneric eq s1 s2 = do
143 ) s1
189 ) s1
144-}
190-}
145
191
146−-- | A more efficient 'joinInner' for sorted streams.
192+-- | A more efficient 'innerJoin' for sorted streams.
147--
193--
148-- Space: O(1)
194-- Space: O(1)
149--
195--
150-- Time: O(m + n)
196-- Time: O(m + n)
151--
197--
152-- /Unimplemented/
198-- /Unimplemented/
153−{-# INLINE joinInnerAscBy #-}
199+{-# INLINE innerSortedJoin #-}
154−joinInnerAscBy ::
200+innerSortedJoin ::
155 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
201 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
156−joinInnerAscBy = undefined
202+innerSortedJoin = undefined
157
203
158−-- | A more efficient 'joinLeft' for sorted streams.
204+-- | A more efficient 'leftJoin' for sorted streams.
159--
205--
160-- Space: O(1)
206-- Space: O(1)
161--
207--
162-- Time: O(m + n)
208-- Time: O(m + n)
163--
209--
164-- /Unimplemented/
210-- /Unimplemented/
165−{-# INLINE joinLeftAscBy #-}
211+{-# INLINE leftSortedJoin #-}
166−joinLeftAscBy :: -- Monad m =>
212+leftSortedJoin :: -- Monad m =>
167 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
213 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
168−joinLeftAscBy _eq _s1 _s2 = undefined
214+leftSortedJoin _eq _s1 _s2 = undefined
169
215
170−-- | A more efficient 'joinOuter' for sorted streams.
216+-- | A more efficient 'outerJoin' for sorted streams.
171--
217--
172-- Space: O(1)
218-- Space: O(1)
173--
219--
174-- Time: O(m + n)
220-- Time: O(m + n)
175--
221--
176-- /Unimplemented/
222-- /Unimplemented/
177−{-# INLINE joinOuterAscBy #-}
223+{-# INLINE outerSortedJoin #-}
178−joinOuterAscBy :: -- Monad m =>
224+outerSortedJoin :: -- Monad m =>
179 (a -> b -> Ordering)
225 (a -> b -> Ordering)
180 -> Stream m a
226 -> Stream m a
181 -> Stream m b
227 -> Stream m b
182 -> Stream m (Maybe a, Maybe b)
228 -> Stream m (Maybe a, Maybe b)
183−joinOuterAscBy _eq _s1 _s2 = undefined
229+outerSortedJoin _eq _s1 _s2 = undefined
184
230
185------------------------------------------------------------------------------
231------------------------------------------------------------------------------
186-- Set operations (special joins)
232-- Set operations (special joins)
@@ -190,8 +236,8 @@ joinOuterAscBy _eq _s1 _s2 = undefined
190-- the best would be to use an Array with linear search. If the second stream
236-- the best would be to use an Array with linear search. If the second stream
191-- is sorted we can also use a binary search, using Ord constraint.
237-- is sorted we can also use a binary search, using Ord constraint.
192
238
193−-- | Keep only those elements in the second stream that are present in the
239+-- | Keep only those elements in the first stream that are present in the
194−-- first stream too. The first stream is folded to a container using the
240+-- second stream too. The second stream is folded to a container using the
195-- supplied fold and then the elements in the container are looked up using the
241-- supplied fold and then the elements in the container are looked up using the
196-- supplied lookup function.
242-- supplied lookup function.
197--
243--
@@ -206,23 +252,30 @@ filterStreamWith :: Monad m =>
206filterStreamWith fld member s1 s2 =
252filterStreamWith fld member s1 s2 =
207 Stream.concatEffect
253 Stream.concatEffect
208 $ do
254 $ do
209− xs <- Stream.fold fld s1
255+ xs <- Stream.fold fld s2
210− return $ Stream.filter (`member` xs) s2
256+ return $ Stream.filter (`member` xs) s1
211
257
212−-- | 'filterInStreamGenericBy' retains only those elements in the second stream that
258+-- XXX instead of folding the second stream to a list we could use it directly.
213−-- are present in the first stream.
259+-- If the user wants they can generate the stream from an array and also call
214−--
260+-- uniq or nub on it. We can provide a convenience Stream -> Stream to cache
215−-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
261+-- a finite stream in an array and serve it from the cache. The user can decide
216−-- [2,1,1]
262+-- what is best based on the context. They can also choose to use a boxed or
217−--
263+-- unboxed array for caching. To force caching we can make the second stream
218−-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
264+-- monad type Identity. But that may be less flexible. One option is to use
219−-- [1,2,2]
265+-- cachedIntersectBy etc for automatic caching.
266+
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'.
220--
272--
221−-- Similar to the list intersectBy operation but with the stream argument order
273+-- >>> f s1 s2 = Stream.fold Fold.toList $ Stream.intersectBy (==) (Stream.fromList s1) (Stream.fromList s2)
222−-- flipped.
274+-- >>> f [1,3,4,4,5]) [2,3,4,5,5]
275+-- [3,4,4,5]
223--
276--
224−-- The first stream must be finite and must not block. Second stream is
277+-- First stream can be infinite, the second stream must be finite and must be
225−-- processed only after the first stream is fully realized.
278+-- capable of multiple evaluations.
226--
279--
227-- Space: O(n) where @n@ is the number of elements in the second stream.
280-- Space: O(n) where @n@ is the number of elements in the second stream.
228--
281--
@@ -230,46 +283,87 @@ filterStreamWith fld member s1 s2 =
230-- @n@ is the number of elements in the second stream.
283-- @n@ is the number of elements in the second stream.
231--
284--
232-- /Pre-release/
285-- /Pre-release/
233−{-# INLINE filterInStreamGenericBy #-}
286+{-# INLINE intersectBy #-}
234−filterInStreamGenericBy :: Monad m =>
287+intersectBy :: Monad m =>
235 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
288 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
236−filterInStreamGenericBy eq =
289+intersectBy eq =
237 -- XXX Use an (unboxed) array instead.
290 -- XXX Use an (unboxed) array instead.
238 filterStreamWith
291 filterStreamWith
239 (Fold.scanMaybe (Fold.uniqBy eq) Fold.toListRev)
292 (Fold.scanMaybe (Fold.uniqBy eq) Fold.toListRev)
240 (List.any . eq)
293 (List.any . eq)
241
294
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+
242−-- | Like 'filterInStreamGenericBy' but assumes that the input streams are sorted in
303+-- | Like 'intersectBy' but assumes that the input streams are sorted in
243-- ascending order. To use it on streams sorted in descending order pass an
304-- ascending order. To use it on streams sorted in descending order pass an
244-- inverted comparison function returning GT for less than and LT for greater
305-- inverted comparison function returning GT for less than and LT for greater
245-- than.
306-- than.
246--
307--
308+-- Both streams can be infinite.
309+--
247-- Space: O(1)
310-- Space: O(1)
248--
311--
249-- Time: O(m+n)
312-- Time: O(m+n)
250--
313--
251-- /Pre-release/
314-- /Pre-release/
252−{-# INLINE filterInStreamAscBy #-}
315+{-# INLINE_NORMAL sortedIntersectBy #-}
253−filterInStreamAscBy :: Monad m =>
316+sortedIntersectBy :: Monad m =>
254 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
317 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
255−filterInStreamAscBy eq s1 s2 = Stream.intersectBySorted eq s2 s1
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+ )
256
325
257−-- | Delete all elements of the first stream from the seconds stream. If an
326+ where
258−-- element occurs multiple times in the first stream as many occurrences of it
327+
259−-- are deleted from the second stream.
328+ {-# INLINE_LATE step #-}
260−--
329+ -- step 1, fetch the first value
261−-- >>> Stream.fold Fold.toList $ Stream.deleteInStreamGenericBy (==) (Stream.fromList [1,2,3]) (Stream.fromList [1,2,2])
330+ step gst (sa, sb, Nothing, b) = do
262−-- [2]
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'.
263--
356--
264−-- The following laws hold:
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]
265--
360--
266−-- > deleteInStreamGenericBy (==) s1 (s1 `append` s2) === s2
361+-- The following holds:
267−-- > deleteInStreamGenericBy (==) s1 (s1 `interleave` s2) === s2
268--
362--
269−-- Same as the list 'Data.List.//' operation but with argument order flipped.
363+-- > deleteFirstsBy (==) (Stream.nub s2 `append` s1) s2 === s1
364+-- > deleteFirstsBy (==) (Stream.nub s2 `interleave` s1) s2 === s1
270--
365--
271−-- The first stream must be finite and must not block. Second stream is
366+-- First stream can be infinite, second stream must be finite.
272−-- processed only after the first stream is fully realized.
273--
367--
274-- Space: O(m) where @m@ is the number of elements in the first stream.
368-- Space: O(m) where @m@ is the number of elements in the first stream.
275--
369--
@@ -277,58 +371,83 @@ filterInStreamAscBy eq s1 s2 = Stream.intersectBySorted eq s2 s1
277-- @n@ is the number of elements in the second stream.
371-- @n@ is the number of elements in the second stream.
278--
372--
279-- /Pre-release/
373-- /Pre-release/
280−{-# INLINE deleteInStreamGenericBy #-}
374+{-# INLINE deleteFirstsBy #-}
281−deleteInStreamGenericBy :: Monad m =>
375+deleteFirstsBy :: Monad m =>
282 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
376 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
283−deleteInStreamGenericBy eq s1 s2 =
377+deleteFirstsBy eq s2 s1 =
284− Stream.concatEffect
378+ -- XXX s2 can be a sorted mutable array and we can use binary
285− $ do
379+ -- search to find. Mark the element deleted, count the deletions
286− -- This may work well if s1 is small
380+ -- and reconsolidate the array when a min number of elements is
287− -- If s1 is big we can go through s1, deleting elements from s2 and
381+ -- deleted.
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
294
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
402+
295−-- | A more efficient 'deleteInStreamGenericBy' for streams sorted in ascending order.
403+-- | A more efficient 'deleteFirstsBy' for streams sorted in ascending order.
404+--
405+-- Both streams can be infinite.
296--
406--
297-- Space: O(1)
407-- Space: O(1)
298--
408--
299-- /Unimplemented/
409-- /Unimplemented/
300−{-# INLINE deleteInStreamAscBy #-}
410+{-# INLINE sortedDeleteFirstsBy #-}
301−deleteInStreamAscBy :: -- (Monad m) =>
411+sortedDeleteFirstsBy :: -- (Monad m) =>
302 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
412 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
303−deleteInStreamAscBy _eq _s1 _s2 = undefined
413+sortedDeleteFirstsBy _eq _s1 _s2 = undefined
304
414
305-- XXX Remove the MonadIO constraint. We can just cache one stream and then
415-- XXX Remove the MonadIO constraint. We can just cache one stream and then
306-- implement using differenceEqBy.
416-- implement using differenceEqBy.
307
417
308−-- | This essentially appends to the second stream all the occurrences of
418+-- | Returns the first stream appended with those unique elements from the
309−-- elements in the first stream that are not already present in the second
419+-- second stream that are not already present in the first stream. Note that
310−-- stream.
420+-- this is not a commutative operation unlike a set union, argument order
421+-- matters. The behavior is similar to the 'Data.List.unionBy'.
311--
422--
312-- Equivalent to the following except that @s2@ is evaluated only once:
423-- Equivalent to the following except that @s2@ is evaluated only once:
313--
424--
314−-- >>> unionWithStreamGenericBy eq s1 s2 = s2 `Stream.append` (Stream.deleteInStreamGenericBy eq s2 s1)
425+-- >>> unionBy eq s1 s2 = s1 `Stream.append` Stream.deleteFirstsBy eq s1 (Stream.nub s2)
315--
426--
316-- Example:
427-- Example:
317--
428--
318−-- >>> Stream.fold Fold.toList $ Stream.unionWithStreamGenericBy (==) (Stream.fromList [1,1,2,3]) (Stream.fromList [1,2,2,4])
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]
319-- [1,2,2,4,3]
431-- [1,2,2,4,3]
320--
432--
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+--
321-- Space: O(n)
438-- Space: O(n)
322--
439--
323-- Time: O(m x n)
440-- Time: O(m x n)
324--
441--
325-- /Pre-release/
442-- /Pre-release/
326−{-# INLINE unionWithStreamGenericBy #-}
443+{-# INLINE unionBy #-}
327−unionWithStreamGenericBy :: MonadIO m =>
444+unionBy :: MonadIO m =>
328 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
445 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
329−unionWithStreamGenericBy eq s1 s2 =
446+unionBy eq s2 s1 =
330 Stream.concatEffect
447 Stream.concatEffect
331 $ do
448 $ 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.
332 xs <- Stream.fold Fold.toList s1
451 xs <- Stream.fold Fold.toList s1
333 -- XXX we can use postscanlMAfter' instead of IORef
452 -- XXX we can use postscanlMAfter' instead of IORef
334 ref <- liftIO $ newIORef $! List.nubBy eq xs
453 ref <- liftIO $ newIORef $! List.nubBy eq xs
@@ -341,12 +460,21 @@ unionWithStreamGenericBy eq s1 s2 =
341 return $ Stream.fromList xs1
460 return $ Stream.fromList xs1
342 return $ Stream.mapM f s2 `Stream.append` s3
461 return $ Stream.mapM f s2 `Stream.append` s3
343
462
344−-- | A more efficient 'unionWithStreamGenericBy' for sorted streams.
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.
345--
471--
346-- Space: O(1)
472-- Space: O(1)
347--
473--
474+-- Both streams can be infinite.
475+--
348-- /Unimplemented/
476-- /Unimplemented/
349−{-# INLINE unionWithStreamAscBy #-}
477+{-# INLINE sortedUnionBy #-}
350−unionWithStreamAscBy :: -- (Monad m) =>
478+sortedUnionBy :: -- (Monad m) =>
351 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
479 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
352−unionWithStreamAscBy _eq _s1 _s2 = undefined
480+sortedUnionBy _eq _s1 _s2 = undefined
core/src/Streamly/Internal/Data/Unfold.hs+6−4
@@ -82,7 +82,7 @@ module Streamly.Internal.Data.Unfold
82 , dropWhileM
82 , dropWhileM
83
83
84 -- ** Cross product
84 -- ** Cross product
85− , joinInnerGeneric
85+ , innerJoin
86
86
87 -- ** Resource Management
87 -- ** Resource Management
88 -- | 'bracket' is the most general resource management operation, all other
88 -- | 'bracket' is the most general resource management operation, all other
@@ -677,10 +677,12 @@ dropWhileM f (Unfold step inject) = Unfold step' inject'
677dropWhile :: Monad m => (b -> Bool) -> Unfold m a b -> Unfold m a b
677dropWhile :: Monad m => (b -> Bool) -> Unfold m a b -> Unfold m a b
678dropWhile f = dropWhileM (return . f)
678dropWhile f = dropWhileM (return . f)
679
679
680+-- | Cross intersection of two unfolds. See
681+-- 'Streamly.Internal.Data.Stream.innerJoin' for more details.
680−{-# INLINE_NORMAL joinInnerGeneric #-}
682+{-# INLINE_NORMAL innerJoin #-}
681−joinInnerGeneric :: Monad m =>
683+innerJoin :: Monad m =>
682 (b -> c -> Bool) -> Unfold m a b -> Unfold m a c -> Unfold m a (b, c)
684 (b -> c -> Bool) -> Unfold m a b -> Unfold m a c -> Unfold m a (b, c)
683−joinInnerGeneric eq s1 s2 = filter (\(a, b) -> a `eq` b) $ cross s1 s2
685+innerJoin eq s1 s2 = filter (\(a, b) -> a `eq` b) $ cross s1 s2
684
686
685------------------------------------------------------------------------------
687------------------------------------------------------------------------------
686-- Exceptions
688-- Exceptions
src/Streamly/Internal/Data/Stream/IsStream/Top.hs+1−1
@@ -649,7 +649,7 @@ intersectBySorted :: (IsStream t, Monad m) =>
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
650intersectBySorted eq s1 =
650intersectBySorted eq s1 =
651 IsStream.fromStreamD
651 IsStream.fromStreamD
652− . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
652+ . StreamD.sortedIntersectBy eq (IsStream.toStreamD s1)
653 . IsStream.toStreamD
653 . IsStream.toStreamD
654
654
655-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
655-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2