160393c · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement intersectBySorted API

160393cRanjeet Kumar Ranjan5 years ago

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

compare with master

unifiedsplit

@@ -459,6 +459,10 @@ o_n_heap_buffering value =

459 $ joinWith Internal.joinOuter sqrtVal sqrtVal

460 , benchIOSrc1 "joinOuterMap"

461 $ joinMapWith Internal.joinOuterMap sqrtVal sqrtVal

462+ , benchIOSrc1 "intersectBy"

463+ $ joinWith Internal.intersectBy sqrtVal sqrtVal

464+ , benchIOSrc1 "intersectBySorted"

465+ $ joinMapWith Internal.intersectBySorted sqrtVal sqrtVal

462 ]

466 ]

463 ]

467 ]

464

468

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+8−5

@@ -28,7 +28,7 @@ module Streamly.Internal.Data.Stream.IsStream.Top

28 -- | These are not exactly set operations because streams are not

29 -- necessarily sets, they may have duplicated elements.

30 , intersectBy

31− , mergeIntersectBy

31+ , intersectBySorted

32 , differenceBy

33 , mergeDifferenceBy

34 , unionBy

@@ -65,6 +65,7 @@ import Streamly.Internal.Data.Stream.IsStream.Common (concatM)

65import Streamly.Internal.Data.Stream.IsStream.Type

66 (IsStream(..), adapt, foldl', fromList)

67import Streamly.Internal.Data.Stream.Serial (SerialT)

68+--import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

68import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

69import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

70import qualified Data.List as List

71import qualified Data.List as List

@@ -79,6 +80,7 @@ import qualified Streamly.Internal.Data.Stream.IsStream.Expand as Stream

79import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream

80import qualified Streamly.Internal.Data.Stream.IsStream.Reduce as Stream

80import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream

81import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream

81import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream

82import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream

83+import qualified Streamly.Internal.Data.Stream.StreamD as StreamD

83import Prelude hiding (filter, zipWith, concatMap, concat)

85import Prelude hiding (filter, zipWith, concatMap, concat)

@@ -580,11 +582,12 @@ intersectBy eq s1 s2 =

580--

582--

581-- Time: O(m+n)

583-- Time: O(m+n)

582--

584--

583−-- /Unimplemented/

585+-- /Pre-release/

584−{-# INLINE mergeIntersectBy #-}

586+{-# INLINE intersectBySorted #-}

585−mergeIntersectBy :: -- (IsStream t, Monad m) =>

587+intersectBySorted :: (IsStream t, MonadIO m, Eq a) =>

586 (a -> a -> Ordering) -> t m a -> t m a -> t m a

588 (a -> a -> Ordering) -> t m a -> t m a -> t m a

587−mergeIntersectBy _eq _s1 _s2 = undefined

589+intersectBySorted eq s1 =

590+ IsStream.fromStreamD . StreamD.intersectBySorted eq (IsStream.toStreamD s1) . IsStream.toStreamD

588

591

589-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2

592-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2

590

593

src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs+54

@@ -142,6 +142,7 @@ module Streamly.Internal.Data.Stream.StreamD.Nesting

142 -- | Opposite to compact in ArrayStream

143 , splitInnerBy

144 , splitInnerBySuffix

145+ , intersectBySorted

145 )

146 )

146where

147where

147

148

@@ -482,6 +483,59 @@ mergeBy

482 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

483 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

483mergeBy cmp = mergeByM (\a b -> return $ cmp a b)

484mergeBy cmp = mergeByM (\a b -> return $ cmp a b)

484

485

486+-------------------------------------------------------------------------------

487+-- Intersection of sorted streams ---------------------------------------------

488+-------------------------------------------------------------------------------

489+{-# INLINE_NORMAL intersectBySorted #-}

490+intersectBySorted

491+ :: (MonadIO m, Eq a)

492+ => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

493+intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =

494+ Stream step (Just ta, Just tb, Nothing, Nothing, Nothing)

495+

496+ where

497+ {-# INLINE_LATE step #-}

498+

499+ -- step 1

500+ step gst (Just sa, sb, Nothing, b, Nothing) = do

501+ r <- stepa gst sa

502+ return $ case r of

503+ Yield a sa' -> Skip (Just sa', sb, Just a, b, Nothing)

504+ Skip sa' -> Skip (Just sa', sb, Nothing, b, Nothing)

505+ Stop -> Stop

506+

507+ -- step 2

508+ step gst (sa, Just sb, a, Nothing, Nothing) = do

509+ r <- stepb gst sb

510+ return $ case r of

511+ Yield b sb' -> Skip (sa, Just sb', a, Just b, Nothing)

512+ Skip sb' -> Skip (sa, Just sb', a, Nothing, Nothing)

513+ Stop -> Stop

514+

515+ -- step 3

516+ -- both the values are available compare it

517+ step _ (sa, sb, Just a, Just b, Nothing) = do

518+ let res = cmp a b

519+ return $ case res of

520+ GT -> Skip (sa, sb, Just a, Nothing, Nothing)

521+ LT -> Skip (sa, sb, Nothing, Just b, Nothing)

522+ EQ -> Yield a (sa, sb, Nothing, Just a, Just b) -- step 4

523+

524+ -- step 4

525+ -- Matching element

526+ step gst (Just sa, Just sb, Nothing, Just _, Just b) = do

527+ r1 <- stepa gst sa

528+ return $ case r1 of

529+ Yield a' sa' -> do

530+ if a' == b -- match with prev a

531+ then Yield a' (Just sa', Just sb, Nothing, Just b, Just b) --step 1

532+ else Skip (Just sa', Just sb, Just a', Nothing, Nothing)

533+

534+ Skip sa' -> Skip (Just sa', Just sb, Nothing, Nothing, Nothing)

535+ Stop -> Stop

536+

537+ step _ (_, _, _, _, _) = return Stop

538+

485------------------------------------------------------------------------------

539------------------------------------------------------------------------------

486-- Combine N Streams - unfoldMany

540-- Combine N Streams - unfoldMany

487------------------------------------------------------------------------------

541------------------------------------------------------------------------------

streamly.cabal+1−1

@@ -101,7 +101,7 @@ extra-source-files:

101 test/Streamly/Test/Data/Array/Prim/Pinned.hs

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

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

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

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

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

test/Streamly/Test/Prelude/Top.hs+44−5

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

1−module Main (main) where

1+module Main (main)

2+ where

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

4+import Data.List (elem, intersect, nub, sort)

4import Data.Maybe (isNothing)

5import Data.Maybe (isNothing)

5import Test.QuickCheck

6import Test.QuickCheck

6 ( Gen

7 ( Gen

@@ -168,10 +169,45 @@ joinLeftMap =

168 let v2 = joinLeftList ls0 ls1

169 let v2 = joinLeftList ls0 ls1

169 assert (v1 == v2)

170 assert (v1 == v2)

170

171

171−-------------------------------------------------------------------------------

172+intersectBy :: Property

172−-- Main

173+intersectBy =

173−-------------------------------------------------------------------------------

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

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

176+ monadicIO $ action (sort ls0) (sort ls1)

177+

178+ where

174

179

180+ action ls0 ls1 = do

181+ v1 <-

182+ run

183+ $ S.toList

184+ $ Top.intersectBy

185+ (==)

186+ (S.fromList ls0)

187+ (S.fromList ls1)

188+ let v2 = intersect ls0 ls1

189+ assert (v1 == sort v2)

190+

191+intersectBySorted :: Property

192+intersectBySorted =

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

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

195+ monadicIO $ action (sort ls0) (sort ls1)

196+

197+ where

198+

199+ action ls0 ls1 = do

200+ v1 <-

201+ run

202+ $ S.toList

203+ $ Top.intersectBySorted

204+ compare

205+ (S.fromList ls0)

206+ (S.fromList ls1)

207+ let v2 = intersect ls0 ls1

208+ assert (v1 == sort v2)

209+

210+-------------------------------------------------------------------------------

175moduleName :: String

211moduleName :: String

176moduleName = "Prelude.Top"

212moduleName = "Prelude.Top"

177

213

@@ -187,3 +223,6 @@ main = hspec $ do

187 prop "joinOuterMap" Main.joinOuterMap

223 prop "joinOuterMap" Main.joinOuterMap

188 prop "joinLeft" Main.joinLeft

224 prop "joinLeft" Main.joinLeft

189 prop "joinLeftMap" Main.joinLeftMap

225 prop "joinLeftMap" Main.joinLeftMap

226+ -- intersect

227+ prop "intersectBy" Main.intersectBy

228+ prop "intersectBySorted" Main.intersectBySorted