@@ -36,9 +36,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top
37 -- ** Join operations
37 -- ** Join operations
38 , crossJoin
38 , crossJoin
39− , innerJoin
39+ , joinInner
40− , mergeInnerJoin
40+ , joinInnerMerge
41− , hashInnerJoin
41+ , joinInnerHash
42 , leftJoin
42 , leftJoin
43 , mergeLeftJoin
43 , mergeLeftJoin
44 , hashLeftJoin
44 , hashLeftJoin
@@ -68,6 +68,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT)
68import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
68import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
70import qualified Data.List as List
70import qualified Data.List as List
71+import qualified Data.Map.Strict as Map
71import qualified Streamly.Internal.Data.Array as Array
72import qualified Streamly.Internal.Data.Array as Array
72import qualified Streamly.Internal.Data.Fold as Fold
73import qualified Streamly.Internal.Data.Fold as Fold
73import qualified Streamly.Internal.Data.Parser as Parser
74import qualified Streamly.Internal.Data.Parser as Parser
@@ -79,6 +80,7 @@ import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream
79import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream
80import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
81import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream
82import Prelude hiding (filter, zipWith, concatMap, concat)
84import Prelude hiding (filter, zipWith, concatMap, concat)
@@ -270,12 +272,12 @@ crossJoin s1 s2 = do
270-- Time: O(m x n)
272-- Time: O(m x n)
272-- /Pre-release/
274-- /Pre-release/
273−{-# INLINE innerJoin #-}
275+{-# INLINE joinInner #-}
274−innerJoin ::
276+joinInner ::
275 forall (t :: (Type -> Type) -> Type -> Type) m a b.
277 forall (t :: (Type -> Type) -> Type -> Type) m a b.
276 (IsStream t, Monad (t m)) =>
278 (IsStream t, Monad (t m)) =>
277 (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
279 (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
279 -- XXX use concatMap instead?
281 -- XXX use concatMap instead?
@@ -286,7 +288,7 @@ innerJoin eq s1 s2 = do
286-- If the second stream is too big it can be partitioned based on hashes and
288-- If the second stream is too big it can be partitioned based on hashes and
287-- then we can process one parition at a time.
289-- then we can process one parition at a time.
289−-- | Like 'innerJoin' but uses a hashmap for efficiency.
291+-- | Like 'joinInner' but uses a hashmap for efficiency.
291-- For space efficiency use the smaller stream as the second stream.
293-- For space efficiency use the smaller stream as the second stream.
@@ -294,22 +296,32 @@ innerJoin eq s1 s2 = do
295-- Time: O(m + n)
297-- Time: O(m + n)
297−-- /Unimplemented/
299+-- /Pre-release/
298−{-# INLINE hashInnerJoin #-}
300+{-# INLINE joinInnerHash #-}
299−hashInnerJoin :: -- Hashable b =>
301+joinInnerHash :: (IsStream t, Monad m, Ord k) =>
300− (a -> b -> Bool) -> t m a -> t m b -> t m (a, b)
302+ t m (k, a) -> t m (k, b) -> t m (k, a, b)
301−hashInnerJoin = undefined
303+joinInnerHash s1 s2 =
305+ km <- Stream.fold (Fold.classify Fold.toList) $ IsStream.adapt s2
306+ return $ Stream.mapMaybe (joinAB km) s1
311+ case k `Map.lookup` km of
312+ Just b -> Just (k, a, head b)
303−-- | Like 'innerJoin' but works only on sorted streams.
315+-- | Like 'joinInner' but works only on sorted streams.
305-- Space: O(1)
317-- Space: O(1)
307-- Time: O(m + n)
319-- Time: O(m + n)
309-- /Unimplemented/
321-- /Unimplemented/
310−{-# INLINE mergeInnerJoin #-}
322+{-# INLINE joinInnerMerge #-}
311−mergeInnerJoin :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
323+joinInnerMerge :: (a -> b -> Ordering) -> t m a -> t m b -> t m (a, b)
312−mergeInnerJoin = undefined
324+joinInnerMerge = undefined
314-- XXX We can do this concurrently.
326-- XXX We can do this concurrently.
315-- XXX If the second stream is sorted and passed as an Array or a seek capable
327-- 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
487-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
499-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
488-- [2,1,1]
500-- [2,1,1]
490−-- 'intersectBy' is similar to but not the same as 'innerJoin':
502+-- 'intersectBy' is similar to but not the same as 'joinInner':
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])
493-- [1,1,2,2]
505-- [1,1,2,2]
495-- Space: O(n) where @n@ is the number of elements in the second stream.
507-- Space: O(n) where @n@ is the number of elements in the second stream.