@@ -37,8 +37,8 @@ module Streamly.Internal.Data.Stream.IsStream.Top
3737 -- ** Join operations
3838 , crossJoin
3939 , joinInner
40+ , joinInnerMap
4041 , joinInnerMerge
41− , joinInnerHash
4242 , leftJoin
4343 , mergeLeftJoin
4444 , hashLeftJoin
@@ -80,7 +80,6 @@ import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
8080import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
8181import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
8282
83−
8483import Prelude hiding (filter, zipWith, concatMap, concat)
8584
8685-- $setup
@@ -261,12 +260,17 @@ crossJoin s1 s2 = do
261260-- are equal by the given equality pedicate then return the tuple (a, b).
262261--
263262-- 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'
265−-- before calling this function. Caching may also improve performance if the
266−-- stream is expensive to evaluate.
263+-- consume-once stream then the caller should cache it (e.g. in a
264+-- 'Data.Array.Array') before calling this function. Caching may also improve
265+-- performance if the stream is expensive to evaluate.
267266--
268267-- For space efficiency use the smaller stream as the second stream.
269268--
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.
273+--
270274-- Space: O(n) assuming the second stream is cached in memory.
271275--
272276-- Time: O(m x n)
@@ -275,20 +279,27 @@ crossJoin s1 s2 = do
275279{-# INLINE joinInner #-}
276280joinInner ::
277281 forall (t :: (Type -> Type) -> Type -> Type) m a b.
278− (IsStream t, Monad (t m)) =>
282+ (IsStream t, Monad m) =>
279283 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
280284joinInner eq s1 s2 = do
281− -- XXX use concatMap instead?
282− a <- s1
283− b <- s2
284− if a `eq` b
285− then return (a, b)
286− else Stream.nil
285+ -- ConcatMap works faster than bind
286+ Stream.concatMap (\a ->
287+ Stream.concatMap (\b ->
288+ if a `eq` b
289+ then Stream.fromPure (a, b)
290+ else Stream.nil
291+ ) s2
292+ ) s1
287293
288294-- If the second stream is too big it can be partitioned based on hashes and
289295-- then we can process one parition at a time.
290296--
291−-- | Like 'joinInner' but uses a hashmap for efficiency.
297+-- XXX An IntMap may be faster when the keys are Int.
298+-- XXX Use hashmap instead of map?
299+--
300+-- | Like 'joinInner' but uses a 'Map' for efficiency.
301+--
302+-- If the input streams have duplicate keys, the behavior is undefined.
292303--
293304-- For space efficiency use the smaller stream as the second stream.
294305--
@@ -297,20 +308,23 @@ joinInner eq s1 s2 = do
297308-- Time: O(m + n)
298309--
299310-- /Pre-release/
300−{-# INLINE joinInnerHash #-}
301−joinInnerHash :: (IsStream t, Monad m, Ord k) =>
311+{-# INLINE joinInnerMap #-}
312+joinInnerMap :: (IsStream t, Monad m, Ord k) =>
302313 t m (k, a) -> t m (k, b) -> t m (k, a, b)
303−joinInnerHash s1 s2 =
304− Stream.concatM $ do
305− km <- Stream.fold (Fold.classify Fold.toList) $ IsStream.adapt s2
306− return $ Stream.mapMaybe (joinAB km) s1
314+joinInnerMap s1 s2 =
315+ Stream.concatM $ do
316+ km <- kvFold $ IsStream.adapt s2
317+ pure $ Stream.mapMaybe (joinAB km) s1
318+
319+ where
307320
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
309323
310− joinAB km (k, a) =
311− case k `Map.lookup` km of
312− Just b -> Just (k, a, head b)
313− Nothing -> Nothing
324+ joinAB kvm (k, a) =
325+ case k `Map.lookup` kvm of
326+ Just b -> Just (k, a, b)
327+ Nothing -> Nothing
314328
315329-- | Like 'joinInner' but works only on sorted streams.
316330--