@@ -36,9 +36,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top
3636
3737 -- ** Join operations
3838 , crossJoin
39− , innerJoin
40− , mergeInnerJoin
41− , hashInnerJoin
39+ , joinInner
40+ , joinInnerMerge
41+ , joinInnerHash
4242 , leftJoin
4343 , mergeLeftJoin
4444 , hashLeftJoin
@@ -68,6 +68,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)
6868import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
6969
7070import qualified Data.List as List
71+import qualified Data.Map.Strict as Map
7172import qualified Streamly.Internal.Data.Array as Array
7273import qualified Streamly.Internal.Data.Fold as Fold
7374import qualified Streamly.Internal.Data.Parser as Parser
@@ -79,6 +80,7 @@ import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
7980import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
8081import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
8182
83+
8284import Prelude hiding (filter, zipWith, concatMap, concat)
8385
8486-- $setup
@@ -270,12 +272,12 @@ crossJoin s1 s2 = do
270272-- Time: O(m x n)
271273--
272274-- /Pre-release/
273−{-# INLINE innerJoin #-}
274−innerJoin ::
275+{-# INLINE joinInner #-}
276+joinInner ::
275277 forall (t :: (Type -> Type) -> Type -> Type) m a b.
276278 (IsStream t, Monad (t m)) =>
277279 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
278−innerJoin eq s1 s2 = do
280+joinInner eq s1 s2 = do
279281 -- XXX use concatMap instead?
280282 a <- s1
281283 b <- s2
@@ -286,7 +288,7 @@ innerJoin eq s1 s2 = do
286288-- If the second stream is too big it can be partitioned based on hashes and
287289-- then we can process one parition at a time.
288290--
289−-- | Like 'innerJoin' but uses a hashmap for efficiency.
291+-- | Like 'joinInner' but uses a hashmap for efficiency.
290292--
291293-- For space efficiency use the smaller stream as the second stream.
292294--
@@ -294,22 +296,32 @@ innerJoin eq s1 s2 = do
294296--
295297-- Time: O(m + n)
296298--
297−-- /Unimplemented/
298−{-# INLINE hashInnerJoin #-}
299−hashInnerJoin :: -- Hashable b =>
300− (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
301−hashInnerJoin = undefined
299+-- /Pre-release/
300+{-# INLINE joinInnerHash #-}
301+joinInnerHash :: (IsStream t, Monad m, Ord k) =>
302+ 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
307+
308+ where
309+
310+ joinAB km (k, a) =
311+ case k `Map.lookup` km of
312+ Just b -> Just (k, a, head b)
313+ Nothing -> Nothing
302314
303−-- | Like 'innerJoin' but works only on sorted streams.
315+-- | Like 'joinInner' but works only on sorted streams.
304316--
305317-- Space: O(1)
306318--
307319-- Time: O(m + n)
308320--
309321-- /Unimplemented/
310−{-# INLINE mergeInnerJoin #-}
311−mergeInnerJoin :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
312−mergeInnerJoin = undefined
322+{-# INLINE joinInnerMerge #-}
323+joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
324+joinInnerMerge = undefined
313325
314326-- XXX We can do this concurrently.
315327-- XXX If the second stream is sorted and passed as an Array or a seek capable
@@ -487,9 +499,9 @@ mergeOuterJoin _eq _s1 _s2 = undefined
487499-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
488500-- [2,1,1]
489501--
490−-- 'intersectBy' is similar to but not the same as 'innerJoin':
502+-- 'intersectBy' is similar to but not the same as 'joinInner':
491503--
492−-- >>> Stream.toList $ fmap fst $ Stream.innerJoin (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
504+-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
493505-- [1,1,2,2]
494506--
495507-- Space: O(n) where @n@ is the number of elements in the second stream.