1b8043f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix joinOuter

1b8043fRanjeet Kumar Ranjan5 years ago

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

compare with master

unifiedsplit

@@ -54,9 +54,10 @@ import Control.Monad.Catch (MonadCatch)

54import Control.Monad.IO.Class (MonadIO(..))

55import Control.Monad.Trans.Class (lift)

56import Control.Monad.Trans.State.Strict (get, put)

57−-- import Data.Hashable (Hashable)

57+import Data.Function ((&))

58import Data.IORef (newIORef, readIORef, modifyIORef')

59import Data.Kind (Type)

60+import Data.Maybe (isJust)

60#if !(MIN_VERSION_base(4,11,0))

61#if !(MIN_VERSION_base(4,11,0))

61import Data.Semigroup (Semigroup(..))

62import Data.Semigroup (Semigroup(..))

62#endif

63#endif

@@ -70,6 +71,7 @@ import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

70import qualified Data.List as List

71import qualified Data.List as List

71import qualified Data.Map.Strict as Map

72import qualified Data.Map.Strict as Map

72import qualified Streamly.Internal.Data.Array as Array

73import qualified Streamly.Internal.Data.Array as Array

74+import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA

73import qualified Streamly.Internal.Data.Fold as Fold

75import qualified Streamly.Internal.Data.Fold as Fold

74import qualified Streamly.Internal.Data.Parser as Parser

76import qualified Streamly.Internal.Data.Parser as Parser

75import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

77import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

@@ -443,17 +445,29 @@ joinOuter :: MonadIO m =>

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

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

444joinOuter eq s1 s =

446joinOuter eq s1 s =

445 Stream.concatM $ do

447 Stream.concatM $ do

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

448+ inputArr <- Array.fromStream s

449+ let len = length inputArr

450+ foundArr <-

451+ Stream.fold

452+ (MA.writeN len)

453+ (Stream.fromList (Prelude.replicate len False))

447− return $ go arr <> leftOver arr

454+ return $ go inputArr foundArr <> leftOver inputArr foundArr

448

455

449 where

456 where

450

457

451− leftOver =

458+ leftOver inputArr foundArr =

452− fmap (\(x, _) -> (Nothing, Just x))

453− . Stream.filter (not . snd)

454− . Array.toStream

459+ let stream1 = IsStream.fromSerial $ Array.toStream inputArr

455−

460+ stream2 = Stream.unfold MA.read foundArr

461+ in Stream.filter

462+ isJust

463+ ( Stream.zipWith (\x y ->

464+ if y

465+ then Nothing

466+ else Just (Nothing, Just x)

467+ ) stream1 stream2

468+ ) & Stream.catMaybes

469+

456− go arr = Stream.evalStateT (return False) $ do

470+ go inputArr foundArr = Stream.evalStateT (return False) $ do

457 a <- Stream.liftInner s1

471 a <- Stream.liftInner s1

458 -- XXX should we use StreamD monad here?

472 -- XXX should we use StreamD monad here?

459 -- XXX Is there a better way to perform some action at the end of a loop

473 -- XXX Is there a better way to perform some action at the end of a loop

@@ -464,16 +478,16 @@ joinOuter eq s1 s =

464 if r

478 if r

465 then Stream.nil

479 then Stream.nil

466 else Stream.fromPure Nothing

480 else Stream.fromPure Nothing

467− (_i, b) <-

481+ (i, b) <-

468− let stream = IsStream.fromSerial $ Array.toStream arr

482+ let stream = IsStream.fromSerial $ Array.toStream inputArr

469 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final

483 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final

484+

470 case b of

485 case b of

471− Just (b1, _used) ->

486+ Just b1 ->

472 if a `eq` b1

487 if a `eq` b1

473 then do

488 then do

474 lift $ put True

489 lift $ put True

475− -- XXX Need to use a mutable array

490+ MA.putIndex i True foundArr

476− -- when (not used) $ Array.writeIndex i True

477 return (Just a, Just b1)

491 return (Just a, Just b1)

478 else Stream.nil

492 else Stream.nil

479 Nothing -> return (Just a, Nothing)

493 Nothing -> return (Just a, Nothing)

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

@@ -85,8 +85,6 @@ joinOuterList ls0 ls1 =

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

86 in v2 ++ v4

88−-- XXX A bug need to be fixed in joinOuter function

89−{-

90joinOuter :: Property

88joinOuter :: Property

91joinOuter =

89joinOuter =

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

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

@@ -103,7 +101,7 @@ joinOuter =

103 (map (\a -> (a, a)) ls1)

101 (map (\a -> (a, a)) ls1)

104 v3 = map (\(_, v10, v20) -> (v10, v20)) v2

102 v3 = map (\(_, v10, v20) -> (v10, v20)) v2

105 assert (sort v1 == sort v3)

103 assert (sort v1 == sort v3)

106−-}

104+

107

105

108joinOuterMap :: Property

106joinOuterMap :: Property

109joinOuterMap =

107joinOuterMap =

@@ -209,8 +207,7 @@ main = hspec $ do

209 -- Joins

207 -- Joins

210 prop "joinInner" Main.joinInner

208 prop "joinInner" Main.joinInner

211 prop "joinInnerMap" Main.joinInnerMap

209 prop "joinInnerMap" Main.joinInnerMap

212− -- XXX currently API is broken https://github.com/composewell/streamly/issues/1032

213− --prop "joinOuter" Main.joinOuter

210+ prop "joinOuter" Main.joinOuter

214 prop "joinOuterMap" Main.joinOuterMap

211 prop "joinOuterMap" Main.joinOuterMap

215 prop "joinLeft" Main.joinLeft

212 prop "joinLeft" Main.joinLeft

216 prop "joinLeftMap" Main.joinLeftMap

213 prop "joinLeftMap" Main.joinLeftMap