@@ -39,9 +39,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top
39 , joinInner
39 , joinInner
40 , joinInnerMap
40 , joinInnerMap
41 , joinInnerMerge
41 , joinInnerMerge
42− , leftJoin
42+ , joinLeft
43 , mergeLeftJoin
43 , mergeLeftJoin
44− , hashLeftJoin
44+ , joinLeftMap
45 , joinOuter
45 , joinOuter
46 , mergeOuterJoin
46 , mergeOuterJoin
47 , joinOuterMap
47 , joinOuterMap
@@ -291,6 +291,10 @@ joinInner eq s1 s2 = do
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
294-- If the second stream is too big it can be partitioned based on hashes and
298-- If the second stream is too big it can be partitioned based on hashes and
295-- then we can process one parition at a time.
299-- then we can process one parition at a time.
@@ -313,14 +317,11 @@ joinInnerMap :: (IsStream t, Monad m, Ord k) =>
313 t m (k, a) -> t m (k, b) -> t m (k, a, b)
317 t m (k, a) -> t m (k, b) -> t m (k, a, b)
314joinInnerMap s1 s2 =
318joinInnerMap s1 s2 =
315 Stream.concatM $ do
319 Stream.concatM $ do
316− km <- kvFold $ IsStream.adapt s2
320+ km <- toMap $ IsStream.adapt s2
317 pure $ Stream.mapMaybe (joinAB km) s1
321 pure $ Stream.mapMaybe (joinAB km) s1
321− -- XXX Generate error if a duplicate insertion is attempted?
322− kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
324 joinAB kvm (k, a) =
325 joinAB kvm (k, a) =
325 case k `Map.lookup` kvm of
326 case k `Map.lookup` kvm of
326 Just b -> Just (k, a, b)
327 Just b -> Just (k, a, b)
@@ -353,7 +354,7 @@ joinInnerMerge = undefined
353-- stream is expensive to evaluate.
354-- stream is expensive to evaluate.
356−-- rightJoin = flip leftJoin
357+-- rightJoin = flip joinLeft
359-- Space: O(n) assuming the second stream is cached in memory.
360-- Space: O(n) assuming the second stream is cached in memory.
@@ -361,10 +362,10 @@ joinInnerMerge = undefined
361-- Time: O(m x n)
362-- Time: O(m x n)
363-- /Unimplemented/
364-- /Unimplemented/
364−{-# INLINE leftJoin #-}
365+{-# INLINE joinLeft #-}
365−leftJoin :: Monad m =>
366+joinLeft :: Monad m =>
366 (a -> b -> Bool) -> SerialT m a -> SerialT m b -> SerialT m (a, Maybe b)
367 (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
368 a <- Stream.liftInner s1
369 a <- Stream.liftInner s1
369 -- XXX should we use StreamD monad here?
370 -- XXX should we use StreamD monad here?
370 -- XXX Is there a better way to perform some action at the end of a loop
371 -- 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
392-- Time: O(m + n)
393-- Time: O(m + n)
394−-- /Unimplemented/
395+-- /Pre-release/
395−{-# INLINE hashLeftJoin #-}
396+{-# INLINE joinLeftMap #-}
396−hashLeftJoin :: -- Hashable b =>
397+joinLeftMap :: (IsStream t, Ord k, Monad m) =>
397− (a -> b -> Bool) -> t m a -> t m b -> t m (a, Maybe b)
398+ t m (k, a) -> t m (k, b) -> t m (k, a, Maybe b)
398−hashLeftJoin = undefined
399+joinLeftMap s1 s2 =
401+ km <- toMap $ IsStream.adapt s2
402+ return $ Stream.map (joinAB km) s1
407+ case k `Map.lookup` km of
408+ Just b -> (k, a, Just b)
409+ Nothing -> (k, a, Nothing)
400−-- | Like 'leftJoin' but works only on sorted streams.
411+-- | Like 'joinLeft' but works only on sorted streams.
402-- Space: O(1)
413-- Space: O(1)
@@ -575,7 +586,7 @@ mergeIntersectBy :: -- (IsStream t, Monad m) =>
575 (a -> a -> Ordering) -> t m a -> t m a -> t m a
586 (a -> a -> Ordering) -> t m a -> t m a -> t m a
576mergeIntersectBy _eq _s1 _s2 = undefined
587mergeIntersectBy _eq _s1 _s2 = undefined
578−-- Roughly leftJoin s1 s2 = s1 `difference` s2 + s1 `intersection` s2
589+-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
580-- | Delete first occurrences of those elements from the first stream that are
591-- | Delete first occurrences of those elements from the first stream that are
581-- present in the second stream. If an element occurs multiple times in the
592-- present in the second stream. If an element occurs multiple times in the