72a2b1f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement joinOuterHash API

72a2b1fRanjeet Kumar Ranjan5 years ago

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

compare with master

unifiedsplit

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

436 (fmap toKvMap (mkStreamLen val1))

437 (fmap toKvMap (mkStreamLen val2))

438

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+

439o_n_heap_buffering :: Int -> [Benchmark]

452o_n_heap_buffering :: Int -> [Benchmark]

440o_n_heap_buffering value =

453o_n_heap_buffering value =

441 [ bgroup "buffered"

454 [ bgroup "buffered"

442 [

455 [

443 benchIOSrc1 "joinInner" (joinInner sqrtVal sqrtVal)

456 benchIOSrc1 "joinInner" (joinInner sqrtVal sqrtVal)

444 , benchIOSrc1 "joinInnerMap" (joinInnerMap sqrtVal sqrtVal)

457 , benchIOSrc1 "joinInnerMap" (joinInnerMap sqrtVal sqrtVal)

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

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

445 ]

460 ]

446 ]

461 ]

447

462

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

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

42 , leftJoin

43 , mergeLeftJoin

44 , hashLeftJoin

45− , outerJoin

45+ , joinOuter

46 , mergeOuterJoin

47− , hashOuterJoin

47+ , joinOuterMap

48 )

49where

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

385 else Stream.nil

386 Nothing -> return (a, Nothing)

387

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

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

389--

390-- Space: O(n)

391--

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

423-- Time: O(m x n)

424--

425-- /Unimplemented/

426−{-# INLINE outerJoin #-}

426+{-# INLINE joinOuter #-}

427−outerJoin :: MonadIO m =>

427+joinOuter :: MonadIO m =>

428 (a -> b -> Bool)

429 -> SerialT m a

430 -> SerialT m b

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

432−outerJoin eq s1 s =

432+joinOuter eq s1 s =

433 Stream.concatM $ do

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

435 return $ go arr <> leftOver arr

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

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

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

472--

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

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

474--

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

476--

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

478--

479-- Time: O(m + n)

480--

481−-- /Unimplemented/

481+-- /Pre-release/

482−{-# INLINE hashOuterJoin #-}

482+{-# INLINE joinOuterMap #-}

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

483+joinOuterMap ::

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

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

485−hashOuterJoin _eq _s1 _s2 = undefined

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

486

511

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

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

488--

513--

489-- Space: O(1)

514-- Space: O(1)

490--

515--

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

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

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

608-- @

633-- @

609--

634--

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

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

611--

636--

612-- Space: O(n)

637-- Space: O(n)

613--

638--

streamly.cabal+1

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

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

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

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

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

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

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

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

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

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

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

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

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

1module Main (main) where

3−import Data.List (nub, sort)

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

4+import Data.Maybe (isNothing)

4import Test.QuickCheck

5import Test.QuickCheck

5 ( Gen

6 ( Gen

6 , Property

7 , Property

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

68 ]

69 ]

69 assert (sort v1 == sort v2)

70 assert (sort v1 == sort v2)

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+

71-------------------------------------------------------------------------------

115-------------------------------------------------------------------------------

72-- Main

116-- Main

73-------------------------------------------------------------------------------

117-------------------------------------------------------------------------------

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

126

83 prop "joinInner" Main.joinInner

127 prop "joinInner" Main.joinInner

84 prop "joinInnerMap" Main.joinInnerMap

128 prop "joinInnerMap" Main.joinInnerMap

129+ prop "joinOuter" Main.joinOuter

130+ prop "joinOuterMap" Main.joinOuterMap