72a2b1f · streamly · git.iko.soy

git.iko.soy/streamly

Implement joinOuterHash API

commit72a2b1fa43ecf731f23bd3575a6af4eb56bb951b authorRanjeet Kumar Ranjan <ranjeetr@composewell.com> dateWed 5 Jan 2022, 18:52 (5 years ago) parent7b0a343

4 files changed+102−15

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude/Serial/NestedStream.hs+15 src/Streamly/Internal/Data/Stream/IsStream/Top.hs+39−14 streamly.cabal+1 test/Streamly/Test/Prelude/Top.hs+47−1

benchmark/Streamly/Benchmark/Prelude/Serial/NestedStream.hs+15

@@ -436,12 +436,27 @@ joinInnerMap val1 val2 _ =

436436 (fmap toKvMap (mkStreamLen val1))

437437 (fmap toKvMap (mkStreamLen val2))

438438

439+{-# INLINE joinOuter #-}

440+joinOuter :: Int -> Int -> Int -> IO ()

441+joinOuter val1 val2 _ =

442+ S.drain $ Internal.joinOuter (==) (mkStreamLen val1) $ mkStreamLen val2

443+

444+{-# INLINE joinOuterMap #-}

445+joinOuterMap :: Int -> Int -> Int -> IO ()

446+joinOuterMap val1 val2 _ =

447+ S.drain $

448+ Internal.joinOuterMap

449+ (fmap toKvMap (mkStreamLen val1))

450+ (fmap toKvMap (mkStreamLen val2))

451+

439452o_n_heap_buffering :: Int -> [Benchmark]

440453o_n_heap_buffering value =

441454 [ bgroup "buffered"

442455 [

443456 benchIOSrc1 "joinInner" (joinInner sqrtVal sqrtVal)

444457 , benchIOSrc1 "joinInnerMap" (joinInnerMap sqrtVal sqrtVal)

458+ , benchIOSrc1 "joinOuter" (joinOuter sqrtVal sqrtVal)

459+ , benchIOSrc1 "joinOuterMap" (joinOuterMap sqrtVal sqrtVal)

445460 ]

446461 ]

447462

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+39−14

@@ -42,9 +42,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top

4242 , leftJoin

4343 , mergeLeftJoin

4444 , hashLeftJoin

45− , outerJoin

45+ , joinOuter

4646 , mergeOuterJoin

47− , hashOuterJoin

47+ , joinOuterMap

4848 )

4949where

5050

@@ -385,7 +385,7 @@ leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do

385385 else Stream.nil

386386 Nothing -> return (a, Nothing)

387387

388−-- | Like 'outerJoin' but uses a hashmap for efficiency.

388+-- | Like 'joinOuter' but uses a hashmap for efficiency.

389389--

390390-- Space: O(n)

391391--

@@ -423,13 +423,13 @@ mergeLeftJoin _eq _s1 _s2 = undefined

423423-- Time: O(m x n)

424424--

425425-- /Unimplemented/

426−{-# INLINE outerJoin #-}

427−outerJoin :: MonadIO m =>

426+{-# INLINE joinOuter #-}

427+joinOuter :: MonadIO m =>

428428 (a -> b -> Bool)

429429 -> SerialT m a

430430 -> SerialT m b

431431 -> SerialT m (Maybe a, Maybe b)

432−outerJoin eq s1 s =

432+joinOuter eq s1 s =

433433 Stream.concatM $ do

434434 arr <- Array.fromStream $ fmap (,False) s

435435 return $ go arr <> leftOver arr

@@ -470,7 +470,7 @@ outerJoin eq s1 s =

470470-- a flag. At the end go through @t m b@ and find those that are not in that

471471-- hash to return (Nothing, b).

472472--

473−-- | Like 'outerJoin' but uses a hashmap for efficiency.

473+-- | Like 'joinOuter' but uses a hashmap for efficiency.

474474--

475475-- For space efficiency use the smaller stream as the second stream.

476476--

@@ -478,13 +478,38 @@ outerJoin eq s1 s =

478478--

479479-- Time: O(m + n)

480480--

481−-- /Unimplemented/

482−{-# INLINE hashOuterJoin #-}

483−hashOuterJoin :: -- (Monad m, Hashable b) =>

484− (a -> b -> Ordering) -> t m a -> t m b -> t m (Maybe a, Maybe b)

485−hashOuterJoin _eq _s1 _s2 = undefined

481+-- /Pre-release/

482+{-# INLINE joinOuterMap #-}

483+joinOuterMap ::

484+ (IsStream t, Ord k, MonadIO m) =>

485+ t m (k, a) -> t m (k, b) -> t m (k, Maybe a, Maybe b)

486+joinOuterMap s1 s2 =

487+ Stream.concatM $ do

488+ km1 <- kvFold $ IsStream.adapt s1

489+ km2 <- kvFold $ IsStream.adapt s2

490+

491+ let res1 = Stream.map (joinAB km2) s1

492+ where

493+ joinAB km (k, a) =

494+ case k `Map.lookup` km of

495+ Just b -> (k, Just a, Just b)

496+ Nothing -> (k, Just a, Nothing)

497+

498+ let res2 = Stream.mapMaybe (joinAB km1) s2

499+ where

500+ joinAB km (k, b) =

501+ case k `Map.lookup` km of

502+ Just _ -> Nothing

503+ Nothing -> Just (k, Nothing, Just b)

504+

505+ return $ Stream.serial res1 res2

506+

507+ where

508+

509+ -- XXX Generate error if a duplicate insertion is attempted?

510+ kvFold = Stream.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty

486511

487−-- | Like 'outerJoin' but works only on sorted streams.

512+-- | Like 'joinOuter' but works only on sorted streams.

488513--

489514-- Space: O(1)

490515--

@@ -607,7 +632,7 @@ mergeDifferenceBy _eq _s1 _s2 = undefined

607632-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)

608633-- @

609634--

610−-- Similar to 'outerJoin' but not the same.

635+-- Similar to 'joinOuter' but not the same.

611636--

612637-- Space: O(n)

613638--

streamly.cabal+1

@@ -102,6 +102,7 @@ extra-source-files:

102102 test/Streamly/Test/Data/Array/Foreign.hs

103103 test/Streamly/Test/Data/Array/Stream/Foreign.hs

104104 test/Streamly/Test/Data/Parser/ParserD.hs

105+ test/Streamly/Test/Data/Stream/Top.hs

105106 test/Streamly/Test/FileSystem/Event.hs

106107 test/Streamly/Test/FileSystem/Event/Common.hs

107108 test/Streamly/Test/FileSystem/Event/Darwin.hs

test/Streamly/Test/Prelude/Top.hs+47−1

@@ -1,6 +1,7 @@

11module Main (main) where

3−import Data.List (nub, sort)

3+import Data.List (elem, nub, sort)

4+import Data.Maybe (isNothing)

45import Test.QuickCheck

56 ( Gen

67 , Property

@@ -68,6 +69,49 @@ joinInnerMap =

6869 ]

6970 assert (sort v1 == sort v2)

7071

72+-- XXX A bug need to be fixed in joinOuter function

73+joinOuter :: Property

74+joinOuter =

75+ forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

76+ forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

77+ monadicIO $ action ls0 ls1

78+ where

79+ action ls0 ls1 = do

80+ v1 <-

81+ run

82+ $ S.toList

83+ $ Top.joinOuter eq (S.fromList ls0) (S.fromList ls1)

84+ let v2 = [ (Just i, Just j) | i <- ls0, j <- ls1, i == j ]

85+ assert (v1 == v2)

86+

87+

88+joinOuterMap :: Property

89+joinOuterMap =

90+ forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

91+ forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

92+ monadicIO $

93+ action (map (\a -> (a,a)) ls0) (map (\a -> (a,a)) (nub ls1))

94+

95+ where

96+

97+ action ls0 ls1 = do

98+ v1 <-

99+ run

100+ $ S.toList

101+ $ Top.joinOuterMap (S.fromList ls0) (S.fromList ls1)

102+ let v2 = do

103+ i <- ls0

104+ if (elem i ls1)

105+ then return (fst i, Just (fst i), Just (fst i))

106+ else return (fst i, Just (fst i), Nothing)

107+ v3 = do

108+ j <- ls1

109+ if (elem j ls0)

110+ then return (fst j, Just (fst j), Just (fst j))

111+ else return (fst j, Nothing, Just (fst j))

112+ v4 = filter (\(_, a2, _) -> isNothing a2) v3

113+ assert (v1 == v2 ++ v4)

114+

71115-------------------------------------------------------------------------------

72116-- Main

73117-------------------------------------------------------------------------------

@@ -82,3 +126,5 @@ main = hspec $ do

82126

83127 prop "joinInner" Main.joinInner

84128 prop "joinInnerMap" Main.joinInnerMap

129+ prop "joinOuter" Main.joinOuter

130+ prop "joinOuterMap" Main.joinOuterMap