@@ -37,8 +37,8 @@ module Streamly.Internal.Data.Stream.IsStream.Top
37 -- ** Join operations
37 -- ** Join operations
38 , crossJoin
38 , crossJoin
39 , joinInner
39 , joinInner
40 , joinInnerMerge
41 , joinInnerMerge
42 , leftJoin
42 , leftJoin
43 , mergeLeftJoin
43 , mergeLeftJoin
44 , hashLeftJoin
44 , hashLeftJoin
@@ -80,7 +80,6 @@ import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
81import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
81import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
84import Prelude hiding (filter, zipWith, concatMap, concat)
83import Prelude hiding (filter, zipWith, concatMap, concat)
@@ -261,12 +260,17 @@ crossJoin s1 s2 = do
261-- are equal by the given equality pedicate then return the tuple (a, b).
260-- are equal by the given equality pedicate then return the tuple (a, b).
263-- The second stream is evaluated multiple times. If the stream is a
262-- The second stream is evaluated multiple times. If the stream is a
264−-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
263+-- consume-once stream then the caller should cache it (e.g. in a
265−-- before calling this function. Caching may also improve performance if the
264+-- 'Data.Array.Array') before calling this function. Caching may also improve
266−-- stream is expensive to evaluate.
265+-- performance if the stream is expensive to evaluate.
268-- For space efficiency use the smaller stream as the second stream.
267-- For space efficiency use the smaller stream as the second stream.
269+-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
270+-- is an order of magnitude faster. joinInner may be used when the second
271+-- stream is generated from a seed, therefore, need not be stored in memory and
272+-- the amount of memory it takes is a concern.
270-- Space: O(n) assuming the second stream is cached in memory.
274-- Space: O(n) assuming the second stream is cached in memory.
272-- Time: O(m x n)
276-- Time: O(m x n)
@@ -275,20 +279,27 @@ crossJoin s1 s2 = do
275{-# INLINE joinInner #-}
279{-# INLINE joinInner #-}
276joinInner ::
280joinInner ::
277 forall (t :: (Type -> Type) -> Type -> Type) m a b.
281 forall (t :: (Type -> Type) -> Type -> Type) m a b.
278− (IsStream t, Monad (t m)) =>
282+ (IsStream t, Monad m) =>
279 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
283 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
280joinInner eq s1 s2 = do
284joinInner eq s1 s2 = do
281− -- XXX use concatMap instead?
285+ -- ConcatMap works faster than bind
282− a <- s1
286+ Stream.concatMap (\a ->
283− b <- s2
287+ Stream.concatMap (\b ->
284− if a `eq` b
288+ if a `eq` b
285− then return (a, b)
289+ then Stream.fromPure (a, b)
286− else Stream.nil
290+ else Stream.nil
288-- If the second stream is too big it can be partitioned based on hashes and
294-- If the second stream is too big it can be partitioned based on hashes and
289-- then we can process one parition at a time.
295-- then we can process one parition at a time.
297+-- XXX An IntMap may be faster when the keys are Int.
298+-- XXX Use hashmap instead of map?
291−-- | Like 'joinInner' but uses a hashmap for efficiency.
300+-- | Like 'joinInner' but uses a 'Map' for efficiency.
302+-- If the input streams have duplicate keys, the behavior is undefined.
293-- For space efficiency use the smaller stream as the second stream.
304-- For space efficiency use the smaller stream as the second stream.
@@ -297,20 +308,23 @@ joinInner eq s1 s2 = do
297-- Time: O(m + n)
308-- Time: O(m + n)
299-- /Pre-release/
310-- /Pre-release/
300−{-# INLINE joinInnerHash #-}
311+{-# INLINE joinInnerMap #-}
301−joinInnerHash :: (IsStream t, Monad m, Ord k) =>
312+joinInnerMap :: (IsStream t, Monad m, Ord k) =>
302 t m (k, a) -> t m (k, b) -> t m (k, a, b)
313 t m (k, a) -> t m (k, b) -> t m (k, a, b)
303−joinInnerHash s1 s2 =
314+joinInnerMap s1 s2 =
304− Stream.concatM $ do
315+ Stream.concatM $ do
305− km <- Stream.fold (Fold.classify Fold.toList) $ IsStream.adapt s2
316+ km <- kvFold $ IsStream.adapt s2
306− return $ Stream.mapMaybe (joinAB km) s1
317+ pure $ Stream.mapMaybe (joinAB km) s1
308− where
321+ -- XXX Generate error if a duplicate insertion is attempted?
322+ kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
310− joinAB km (k, a) =
324+ joinAB kvm (k, a) =
311− case k `Map.lookup` km of
325+ case k `Map.lookup` kvm of
312− Just b -> Just (k, a, head b)
326+ Just b -> Just (k, a, b)
313− Nothing -> Nothing
327+ Nothing -> Nothing
315-- | Like 'joinInner' but works only on sorted streams.
329-- | Like 'joinInner' but works only on sorted streams.