@@ -39,9 +39,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top
3939 , joinInner
4040 , joinInnerMap
4141 , joinInnerMerge
42− , leftJoin
42+ , joinLeft
4343 , mergeLeftJoin
44− , hashLeftJoin
44+ , joinLeftMap
4545 , joinOuter
4646 , mergeOuterJoin
4747 , joinOuterMap
@@ -291,6 +291,10 @@ joinInner eq s1 s2 = do
291291 ) s2
292292 ) s1
293293
294+-- XXX Generate error if a duplicate insertion is attempted?
295+toMap :: (Monad m, Ord k) => IsStream.SerialT m (k, v) -> m (Map.Map k v)
296+toMap = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
297+
294298-- If the second stream is too big it can be partitioned based on hashes and
295299-- then we can process one parition at a time.
296300--
@@ -313,14 +317,11 @@ joinInnerMap :: (IsStream t, Monad m, Ord k) =>
313317 t m (k, a) -> t m (k, b) -> t m (k, a, b)
314318joinInnerMap s1 s2 =
315319 Stream.concatM $ do
316− km <- kvFold $ IsStream.adapt s2
320+ km <- toMap $ IsStream.adapt s2
317321 pure $ Stream.mapMaybe (joinAB km) s1
318322
319323 where
320324
321− -- XXX Generate error if a duplicate insertion is attempted?
322− kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
323−
324325 joinAB kvm (k, a) =
325326 case k `Map.lookup` kvm of
326327 Just b -> Just (k, a, b)
@@ -353,7 +354,7 @@ joinInnerMerge = undefined
353354-- stream is expensive to evaluate.
354355--
355356-- @
356−-- rightJoin = flip leftJoin
357+-- rightJoin = flip joinLeft
357358-- @
358359--
359360-- Space: O(n) assuming the second stream is cached in memory.
@@ -361,10 +362,10 @@ joinInnerMerge = undefined
361362-- Time: O(m x n)
362363--
363364-- /Unimplemented/
364−{-# INLINE leftJoin #-}
365−leftJoin :: Monad m =>
365+{-# INLINE joinLeft #-}
366+joinLeft :: Monad m =>
366367 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
367−leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do
368+joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
368369 a <- Stream.liftInner s1
369370 -- XXX should we use StreamD monad here?
370371 -- XXX Is there a better way to perform some action at the end of a loop
@@ -391,13 +392,23 @@ leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do
391392--
392393-- Time: O(m + n)
393394--
394−-- /Unimplemented/
395−{-# INLINE hashLeftJoin #-}
396−hashLeftJoin :: -- Hashable b =>
397− (a -> b -> Bool) -> t m a -> t m b -> t m (a, Maybe b)
398−hashLeftJoin = undefined
395+-- /Pre-release/
396+{-# INLINE joinLeftMap #-}
397+joinLeftMap :: (IsStream t, Ord k, Monad m) =>
398+ t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
399+joinLeftMap s1 s2 =
400+ Stream.concatM $ do
401+ km <- toMap $ IsStream.adapt s2
402+ return $ Stream.map (joinAB km) s1
403+
404+ where
405+
406+ joinAB km (k, a) =
407+ case k `Map.lookup` km of
408+ Just b -> (k, a, Just b)
409+ Nothing -> (k, a, Nothing)
399410
400−-- | Like 'leftJoin' but works only on sorted streams.
411+-- | Like 'joinLeft' but works only on sorted streams.
401412--
402413-- Space: O(1)
403414--
@@ -575,7 +586,7 @@ mergeIntersectBy :: -- (IsStream t, Monad m) =>
575586 (a -> a -> Ordering) -> t m a -> t m a -> t m a
576587mergeIntersectBy _eq _s1 _s2 = undefined
577588
578−-- Roughly leftJoin s1 s2 = s1 `difference` s2 + s1 `intersection` s2
589+-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
579590
580591-- | Delete first occurrences of those elements from the first stream that are
581592-- present in the second stream. If an element occurs multiple times in the