on master help sign in
core/src/DocTestDataStream.hs+2
since splitsnapshots
unifiedsplit
@@ -6,6 +6,7 @@
6>>> import Control.Monad.IO.Class (MonadIO (liftIO))
6>>> import Control.Monad.IO.Class (MonadIO (liftIO))
7>>> import Control.Monad.Trans.Class (lift)
7>>> import Control.Monad.Trans.Class (lift)
8>>> import Control.Monad.Trans.Identity (runIdentityT)
8>>> import Control.Monad.Trans.Identity (runIdentityT)
9+>>> import Data.Char (isSpace)
9>>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
10>>> import Data.Either (fromLeft, fromRight, isLeft, isRight, either)
10>>> import Data.Maybe (fromJust, isJust)
11>>> import Data.Maybe (fromJust, isJust)
11>>> import Data.Function (fix, (&))
12>>> import Data.Function (fix, (&))
@@ -32,6 +33,7 @@ For APIs that have not been released yet.
32>>> import qualified Streamly.Internal.Data.Fold as Fold
33>>> import qualified Streamly.Internal.Data.Fold as Fold
33>>> import qualified Streamly.Internal.Data.Parser as Parser
34>>> import qualified Streamly.Internal.Data.Parser as Parser
34>>> import qualified Streamly.Internal.Data.Stream as Stream
35>>> import qualified Streamly.Internal.Data.Stream as Stream
36+>>> import qualified Streamly.Internal.Data.StreamK as StreamK
35>>> import qualified Streamly.Internal.Data.Unfold as Unfold
37>>> import qualified Streamly.Internal.Data.Unfold as Unfold
36>>> import qualified Streamly.Internal.FileSystem.Dir as Dir
38>>> import qualified Streamly.Internal.FileSystem.Dir as Dir
37-}
39-}
core/src/Streamly/Data/Stream.hs+129−20
@@ -70,6 +70,15 @@ module Streamly.Data.Stream
70
70
71 -- ** From Values
71 -- ** From Values
72 -- | Generate a monadic stream from a seed value or values.
72 -- | Generate a monadic stream from a seed value or values.
73+ --
74+ -- Non-empty pure streams can be cycled using repeat and unfold:
75+ --
76+ -- >>> cycle = Stream.unfoldMany Unfold.fromList . Stream.repeat
77+ --
78+ -- Non-empty StreamK can be cycled using append:
79+ --
80+ -- >>> cycle xs = let ys = xs `StreamK.append` ys in ys
81+ --
73 , fromPure
82 , fromPure
74 , fromEffect
83 , fromEffect
75 , repeat
84 , repeat
@@ -212,21 +221,57 @@ module Streamly.Data.Stream
212 -- operations like mapping a function over the stream.
221 -- operations like mapping a function over the stream.
213 , foldrM
222 , foldrM
214 , foldr
223 , foldr
224+ -- foldr1
215
225
216 -- ** Specific Folds
226 -- ** Specific Folds
217− -- | Usually you can use the folds in "Streamly.Data.Fold". However, some
227+ -- | Streams are folded using folds in "Streamly.Data.Fold". Here are some
218− -- folds that may be commonly used or may have an edge in performance in
228+ -- idioms and equivalents of Data.List APIs using folds:
219− -- some cases are provided here.
220− --
221− -- Useful idioms:
222 --
229 --
223 -- >>> foldlM' f a = Stream.fold (Fold.foldlM' f a)
230 -- >>> foldlM' f a = Stream.fold (Fold.foldlM' f a)
224− -- >>> foldl1 f = Stream.fold (Fold.foldl1' f)
231+ -- >>> foldl1' f = Stream.fold (Fold.foldl1' f)
225 -- >>> foldl' f a = Stream.fold (Fold.foldl' f a)
232 -- >>> foldl' f a = Stream.fold (Fold.foldl' f a)
226 -- >>> drain = Stream.fold Fold.drain
233 -- >>> drain = Stream.fold Fold.drain
227 -- >>> mapM_ f = Stream.fold (Fold.drainMapM f)
234 -- >>> mapM_ f = Stream.fold (Fold.drainMapM f)
228 -- >>> length = Stream.fold Fold.length
235 -- >>> length = Stream.fold Fold.length
236+ -- >>> genericLength = Stream.fold Fold.lengthGeneric
229 -- >>> head = Stream.fold Fold.one
237 -- >>> head = Stream.fold Fold.one
238+ -- >>> last = Stream.fold Fold.latest
239+ -- >>> null = Stream.fold Fold.null
240+ -- >>> and = Stream.fold Fold.and
241+ -- >>> or = Stream.fold Fold.or
242+ -- >>> any p = Stream.fold (Fold.any p)
243+ -- >>> all p = Stream.fold (Fold.all p)
244+ -- >>> sum = Stream.fold Fold.sum
245+ -- >>> product = Stream.fold Fold.product
246+ -- >>> maximum = Stream.fold Fold.maximum
247+ -- >>> maximumBy cmp = Stream.fold (Fold.maximumBy cmp)
248+ -- >>> minimum = Stream.fold Fold.minimum
249+ -- >>> minimumBy cmp = Stream.fold (Fold.minimumBy cmp)
250+ -- >>> elem x = Stream.fold (Fold.elem x)
251+ -- >>> notElem x = Stream.fold (Fold.notElem x)
252+ -- >>> lookup x = Stream.fold (Fold.lookup x)
253+ -- >>> find p = Stream.fold (Fold.find p)
254+ -- >>> (!?) i = Stream.fold (Fold.index i)
255+ -- >>> genericIndex i = Stream.fold (Fold.indexGeneric i)
256+ -- >>> elemIndex x = Stream.fold (Fold.elemIndex x)
257+ -- >>> findIndex p = Stream.fold (Fold.findIndex p)
258+ --
259+ -- Some equivalents of Data.List APIs from the Stream module:
260+ --
261+ -- >>> head = fmap (fmap fst) . Stream.uncons
262+ -- >>> tail = fmap (fmap snd) . Stream.uncons
263+ -- >>> tail = Stream.tail -- unreleased API
264+ -- >>> init = Stream.init -- unreleased API
265+ --
266+ -- A Stream based toList fold implementation is provided below because it
267+ -- has a better performance compared to the fold.
268+
269+ -- Functions in Data.List, missing here:
270+ -- unsnoc = Stream.parseBreak (Parser.init Fold.toList)
271+ -- genericTake
272+ -- genericDrop
273+ -- genericSplitAt
274+ -- genericReplicate
230 , toList
275 , toList
231
276
232 -- * Mapping
277 -- * Mapping
@@ -280,6 +325,15 @@ module Streamly.Data.Stream
280 , scan
325 , scan
281 , postscan
326 , postscan
282 -- XXX postscan1 can be implemented using Monoids or Refolds.
327 -- XXX postscan1 can be implemented using Monoids or Refolds.
328+ -- The following scans from Data.List are not provided.
329+ -- XXX scanl
330+ -- XXX scanl1
331+ -- XXX scanr
332+ -- XXX scanr1
333+ -- XXX mapAccumL
334+ -- XXX mapAccumR
335+ -- XXX inits
336+ -- XXX tails
283
337
284 -- ** Specific scans
338 -- ** Specific scans
285 -- Indexing can be considered as a special type of zipping where we zip a
339 -- Indexing can be considered as a special type of zipping where we zip a
@@ -288,6 +342,8 @@ module Streamly.Data.Stream
288
342
289 -- * Insertion
343 -- * Insertion
290 -- | Add elements to the stream.
344 -- | Add elements to the stream.
345+ --
346+ -- >>> insert = Stream.insertBy compare
291
347
292 -- Inserting elements is a special case of interleaving/merging streams.
348 -- Inserting elements is a special case of interleaving/merging streams.
293 , insertBy
349 , insertBy
@@ -323,12 +379,15 @@ module Streamly.Data.Stream
323 -- "Streamly.Internal.Data.Fold" can be used along with 'scanMaybe' to
379 -- "Streamly.Internal.Data.Fold" can be used along with 'scanMaybe' to
324 -- perform stateful filtering operations in general.
380 -- perform stateful filtering operations in general.
325 --
381 --
326− -- Useful idioms:
382+ -- Idioms and equivalents of Data.List APIs:
327 --
383 --
328 -- >>> deleteBy cmp x = Stream.scanMaybe (Fold.deleteBy cmp x)
384 -- >>> deleteBy cmp x = Stream.scanMaybe (Fold.deleteBy cmp x)
385+ -- >>> deleteBy = Stream.deleteBy -- unreleased API
386+ -- >>> delete = deleteBy (==)
329 -- >>> findIndices p = Stream.scanMaybe (Fold.findIndices p)
387 -- >>> findIndices p = Stream.scanMaybe (Fold.findIndices p)
330 -- >>> elemIndices a = findIndices (== a)
388 -- >>> elemIndices a = findIndices (== a)
331 -- >>> uniq = Stream.scanMaybe (Fold.uniqBy (==))
389 -- >>> uniq = Stream.scanMaybe (Fold.uniqBy (==))
390+ -- >>> partition p = Stream.fold (Fold.partition Fold.toList Fold.toList) . fmap (if p then Left else Right)
332 , scanMaybe
391 , scanMaybe
333 , take
392 , take
334 , takeWhile
393 , takeWhile
@@ -336,6 +395,8 @@ module Streamly.Data.Stream
336 , drop
395 , drop
337 , dropWhile
396 , dropWhile
338 , dropWhileM
397 , dropWhileM
398+ -- XXX write to an array in reverse and then read in reverse
399+ -- > dropWhileEnd = reverse . dropWhile p . reverse
339
400
340 -- XXX These are available as scans in folds. We need to check the
401 -- XXX These are available as scans in folds. We need to check the
341 -- performance though. If these are common and we need convenient stream
402 -- performance though. If these are common and we need convenient stream
@@ -372,6 +433,9 @@ module Streamly.Data.Stream
372 -- >>> s = s1 `Stream.append` s2
433 -- >>> s = s1 `Stream.append` s2
373
434
374 -- ** Appending
435 -- ** Appending
436+ -- | Equivalent of Data.List append:
437+ --
438+ -- >>> (++) = Stream.append
375 , append
439 , append
376
440
377 -- ** Interleaving
441 -- ** Interleaving
@@ -382,8 +446,14 @@ module Streamly.Data.Stream
382 , mergeByM
446 , mergeByM
383
447
384 -- ** Zipping
448 -- ** Zipping
449+ -- | Idioms and equivalents of Data.List APIs:
450+ --
451+ -- >>> zip = Stream.zipWith (,)
452+ -- >>> unzip = Stream.fold (Fold.unzip Fold.toList Fold.toList)
385 , zipWith
453 , zipWith
386 , zipWithM
454 , zipWithM
455+ -- XXX zipWith3,4,5,6,7
456+ -- XXX unzip3,4,5,6,7
387 -- , ZipStream (..)
457 -- , ZipStream (..)
388
458
389 -- ** Cross Product
459 -- ** Cross Product
@@ -416,48 +486,89 @@ module Streamly.Data.Stream
416 -- One dimension loops are just a special case of nested loops. For
486 -- One dimension loops are just a special case of nested loops. For
417 -- example, 'concatMap' can degenerate to a simple map operation:
487 -- example, 'concatMap' can degenerate to a simple map operation:
418 --
488 --
419− -- > map f m = S.concatMap (\x -> S.fromPure (f x)) m
489+ -- >>> map f m = Stream.concatMap (\x -> Stream.fromPure (f x)) m
420 --
490 --
421 -- Similarly, 'concatMap' can perform filtering by mapping an element to a
491 -- Similarly, 'concatMap' can perform filtering by mapping an element to a
422 -- 'nil' stream:
492 -- 'nil' stream:
423 --
493 --
424− -- > filter p m = S.concatMap (\x -> if p x then S.fromPure x else S.nil) m
494+ -- >>> filter p m = Stream.concatMap (\x -> if p x then Stream.fromPure x else Stream.nil) m
495+ --
496+ -- Idioms and equivalents of Data.List APIs:
425 --
497 --
498+ -- >>> concat = Stream.concatMap id
499+ -- >>> unlines = Stream.interposeSuffix '\n'
500+ -- >>> unlines = Stream.intercalateSuffix Unfold.fromList "\n"
501+ -- >>> unwords = Stream.interpose ' '
502+ -- >>> unwords = Stream.intercalate Unfold.fromList " "
426
503
427 , concatEffect
504 , concatEffect
428 , concatMap
505 , concatMap
429 , concatMapM
506 , concatMapM
430
507
431 -- * Repeated Fold
508 -- * Repeated Fold
432− -- | Useful idioms:
509+ -- | Idioms and equivalents of Data.List APIs:
433 --
510 --
434− -- >>> splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
511+ -- >>> groupsOf n = Stream.foldMany (Fold.take n Fold.toList)
435− -- >>> splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
512+ -- >>> groupBy eq = Stream.groupsWhile eq Fold.toList
436− -- >>> groupsBy eq f = Stream.parseMany (Parser.groupBy eq f)
513+ -- >>> groupBy eq = Stream.parseMany (Parser.groupBy eq Fold.toList)
437− -- >>> groupsByRolling eq f = Stream.parseMany (Parser.groupByRolling eq f)
514+ -- >>> groupsByRolling eq = Stream.parseMany (Parser.groupByRolling eq Fold.toList)
438− -- >>> groupsOf n f = Stream.foldMany (Fold.take n f)
515+ -- >>> groups = groupBy (==)
439 , foldMany -- XXX Rename to foldRepeat
516 , foldMany -- XXX Rename to foldRepeat
440 , groupsOf
517 , groupsOf
441 , parseMany
518 , parseMany
442
519
443 -- * Splitting
520 -- * Splitting
521+ -- | Idioms and equivalents of Data.List APIs:
522+ --
523+ -- >>> splitWithSuffix p f = Stream.foldMany (Fold.takeEndBy p f)
524+ -- >>> splitOnSuffix p f = Stream.foldMany (Fold.takeEndBy_ p f)
525+ -- >>> lines = splitOnSuffix (== '\n')
526+ -- >>> words = Stream.wordsBy isSpace
527+ -- >>> splitAt n = Stream.fold (Fold.splitAt n Fold.toList Fold.toList)
528+ -- >>> span p = Parser.splitWith (,) (Parser.takeWhile p Fold.toList) (Parser.fromFold Fold.toList)
529+ -- >>> break p = span (not . p)
444 , splitOn
530 , splitOn
445 , wordsBy
531 , wordsBy
446
532
447 -- * Buffered Operations
533 -- * Buffered Operations
448 -- | Operations that require buffering of the stream.
534 -- | Operations that require buffering of the stream.
449 -- Reverse is essentially a left fold followed by an unfold.
535 -- Reverse is essentially a left fold followed by an unfold.
536+ --
537+ -- Idioms and equivalents of Data.List APIs:
538+ --
539+ -- >>> nub = Stream.fold Fold.toList . Stream.scanMaybe Fold.nub
540+ -- >>> nub = Stream.ordNub -- unreleased API
541+ -- >>> sortBy = StreamK.sortBy
542+ -- >>> sortOn f = StreamK.sortOn -- unreleased API
543+ -- >>> deleteFirstsBy = Stream.deleteFirstsBy -- unreleased
544+ -- >>> (\\) = Stream.deleteFirstsBy (==) -- unreleased
545+ -- >>> intersectBy = Stream.intersectBy -- unreleased
546+ -- >>> intersect = Stream.intersectBy (==) -- unreleased
547+ -- >>> unionBy = Stream.unionBy -- unreleased
548+ -- >>> union = Stream.unionBy (==) -- unreleased
549+ --
450 , reverse
550 , reverse
551+ -- XXX transpose: write the streams to arrays and then stream transposed.
552+ -- XXX subsequences
553+ -- XXX permutations
554+ -- , nub
555+ -- , ordNub
556+ -- , nubBy
451
557
452 -- * Multi-Stream folds
558 -- * Multi-Stream folds
453 -- | Operations that consume multiple streams at the same time.
559 -- | Operations that consume multiple streams at the same time.
454 , eqBy
560 , eqBy
455 , cmpBy
561 , cmpBy
456 , isPrefixOf
562 , isPrefixOf
563+ , isInfixOf
564+ -- , isSuffixOf
565+ -- , isSuffixOfUnbox
457 , isSubsequenceOf
566 , isSubsequenceOf
458
567
459 -- trimming sequences
568 -- trimming sequences
460− , stripPrefix
569+ -- , stripPrefix
570+ -- , stripSuffix
571+ -- , stripSuffixUnbox
461
572
462 -- Exceptions and resource management depend on the "exceptions" package
573 -- Exceptions and resource management depend on the "exceptions" package
463 -- XXX We can have IO Stream operations not depending on "exceptions"
574 -- XXX We can have IO Stream operations not depending on "exceptions"
@@ -521,10 +632,8 @@ where
521import Streamly.Internal.Data.Stream
632import Streamly.Internal.Data.Stream
522import Prelude
633import Prelude
523 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
634 hiding (filter, drop, dropWhile, take, takeWhile, zipWith, foldr,
524− foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem,
525− notElem, maximum, minimum, head, last, tail, length, null,
526− reverse, iterate, init, and, or, lookup, foldr1, (!!),
635+ mapM, sequence, reverse, iterate, foldr1, repeat, replicate,
527− scanl, scanl1, repeat, replicate, concatMap, span)
636+ concatMap)
528
637
529import qualified Streamly.Internal.Data.Array.Type as Array
638import qualified Streamly.Internal.Data.Array.Type as Array
530
639
core/src/Streamly/Internal/Data/MutByteArray/Type.hs+6
@@ -196,6 +196,12 @@ newBytesAs Pinned = pinnedNew
196-- Copying
196-- Copying
197-------------------------------------------------------------------------------
197-------------------------------------------------------------------------------
198
198
199+-- Note: Array copy is more efficient than streaming copy.
200+-- CopyMutableByteArray# translates to genMemcpy in GHC/CmmToAsm/X86/CodeGen.hs
201+-- glibc memcpy copies bytes/words/pages - unrolls the loops:
202+-- https://github.com/bminor/glibc/blob/4290aed05135ae4c0272006442d147f2155e70d7/string/memcpy.c
203+-- https://github.com/bminor/glibc/blob/4290aed05135ae4c0272006442d147f2155e70d7/string/wordcopy.c
204+
199-- | Put a sub range of a source array into a subrange of a destination array.
205-- | Put a sub range of a source array into a subrange of a destination array.
200-- This is not safe as it does not check the bounds of neither the src array
206-- This is not safe as it does not check the bounds of neither the src array
201-- nor the destination array.
207-- nor the destination array.
core/src/Streamly/Internal/Data/Serialize/TH.hs+5−1
@@ -7,7 +7,11 @@
7-- Maintainer : streamly@composewell.com
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
8-- Stability : experimental
9-- Portability : GHC
9-- Portability : GHC
10−--
10+
11+-- XXX Instead of applying the product constructor in one go can we apply it
12+-- one at a time in case of too many arguments, compiler may not have to save
13+-- them in local vars.
14+
11module Streamly.Internal.Data.Serialize.TH
15module Streamly.Internal.Data.Serialize.TH
12 (
16 (
13 -- Deriving
17 -- Deriving
core/src/Streamly/Internal/Data/Stream/Container.hs+56−48
@@ -11,16 +11,17 @@
11
11
12module Streamly.Internal.Data.Stream.Container
12module Streamly.Internal.Data.Stream.Container
13 (
13 (
14− nub
14+ -- * Deduplication
15+ ordNub
15
16
16− -- * Joins for unconstrained types
17+ -- * Joins
17− , joinLeftGeneric
18+ , leftJoin
18− , joinOuterGeneric
19+ , outerJoin
19
20
20− -- * Joins with Ord constraint
21+ -- * Ord Joins
21− , joinInner
22+ , innerOrdJoin
22− , joinLeft
23+ , leftOrdJoin
23− , joinOuter
24+ , outerOrdJoin
24 )
25 )
25where
26where
26
27
@@ -47,12 +48,15 @@ import qualified Streamly.Internal.Data.Stream.Transformer as Stream
47
48
48#include "DocTestDataStream.hs"
49#include "DocTestDataStream.hs"
49
50
51+-- | @nub@ specialized to 'Ord' types for better performance. Returns a
52+-- subsequence of the stream removing any duplicate elements.
53+--
50−-- | The memory used is proportional to the number of unique elements in the
54+-- The memory used is proportional to the number of unique elements in the
51−-- stream. If we want to limit the memory we can just use "take" to limit the
55+-- stream. One way to limit the memory is to use @take@ on the resulting
52−-- uniq elements in the stream.
56+-- stream to limit the unique elements in the stream.
53−{-# INLINE_NORMAL nub #-}
57+{-# INLINE_NORMAL ordNub #-}
54−nub :: (Monad m, Ord a) => Stream m a -> Stream m a
58+ordNub :: (Monad m, Ord a) => Stream m a -> Stream m a
55−nub (Stream step1 state1) = Stream step (Set.empty, state1)
59+ordNub (Stream step1 state1) = Stream step (Set.empty, state1)
56
60
57 where
61 where
58
62
@@ -78,8 +82,8 @@ toMap =
78--
82--
79-- XXX An IntMap may be faster when the keys are Int.
83-- XXX An IntMap may be faster when the keys are Int.
80-- XXX Use hashmap instead of map?
84-- XXX Use hashmap instead of map?
81−--
85+
82−-- | Like 'joinInner' but uses a 'Map' for efficiency.
86+-- | 'innerJoin' specialized to 'Ord' types for better performance.
83--
87--
84-- If the input streams have duplicate keys, the behavior is undefined.
88-- If the input streams have duplicate keys, the behavior is undefined.
85--
89--
@@ -90,10 +94,10 @@ toMap =
90-- Time: O(m + n)
94-- Time: O(m + n)
91--
95--
92-- /Pre-release/
96-- /Pre-release/
93−{-# INLINE joinInner #-}
97+{-# INLINE innerOrdJoin #-}
94−joinInner :: (Monad m, Ord k) =>
98+innerOrdJoin :: (Monad m, Ord k) =>
95 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, b)
99 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, b)
96−joinInner s1 s2 =
100+innerOrdJoin s1 s2 =
97 Stream.concatEffect $ do
101 Stream.concatEffect $ do
98 km <- toMap s2
102 km <- toMap s2
99 pure $ Stream.mapMaybe (joinAB km) s1
103 pure $ Stream.mapMaybe (joinAB km) s1
@@ -106,32 +110,33 @@ joinInner s1 s2 =
106 Nothing -> Nothing
110 Nothing -> Nothing
107
111
108-- XXX We can do this concurrently.
112-- XXX We can do this concurrently.
113+-- XXX Check performance of StreamD vs StreamK
109-- XXX If the second stream is sorted and passed as an Array or a seek capable
114-- XXX If the second stream is sorted and passed as an Array or a seek capable
110-- stream then we could use binary search if we have an Ord instance or
115-- stream then we could use binary search if we have an Ord instance or
111-- Ordering returning function. The time complexity would then become (m x log
116-- Ordering returning function. The time complexity would then become (m x log
112-- n).
117-- n).
113
118
114−-- XXX Check performance of StreamD vs StreamK
119+-- | Like 'innerJoin' but emits @(a, Just b)@ whenever a and b are equal, for
115−
116−-- | Like 'joinInner' but emit @(a, Just b)@, and additionally, for those @a@'s
117−-- that are not equal to any @b@ emit @(a, Nothing)@.
120+-- those @a@'s that are not equal to any @b@ emits @(a, Nothing)@.
118--
121--
119−-- The second stream is evaluated multiple times. If the stream is a
122+-- This is a generalization of 'innerJoin' to include all elements from the
120−-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
123+-- left stream and not just those which have an equal in the right stream. This
121−-- before calling this function. Caching may also improve performance if the
124+-- is not a commutative operation, the order of the stream arguments matters.
122−-- stream is expensive to evaluate.
123--
125--
124−-- >>> joinRightGeneric eq = flip (Stream.joinLeftGeneric eq)
126+-- All the caveats mentioned in 'innerJoin' apply here as well. Right join is
127+-- not provided because it is just a flipped left join:
128+--
129+-- >>> rightJoin eq = flip (Stream.leftJoin eq)
125--
130--
126-- Space: O(n) assuming the second stream is cached in memory.
131-- Space: O(n) assuming the second stream is cached in memory.
127--
132--
128-- Time: O(m x n)
133-- Time: O(m x n)
129--
134--
130-- /Unimplemented/
135-- /Unimplemented/
131−{-# INLINE joinLeftGeneric #-}
136+{-# INLINE leftJoin #-}
132−joinLeftGeneric :: Monad m =>
137+leftJoin :: Monad m =>
133 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
138 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
134−joinLeftGeneric eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
139+leftJoin eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
135 a <- mkCross (Stream.liftInner s1)
140 a <- mkCross (Stream.liftInner s1)
136 -- XXX should we use StreamD monad here?
141 -- XXX should we use StreamD monad here?
137 -- XXX Is there a better way to perform some action at the end of a loop
142 -- XXX Is there a better way to perform some action at the end of a loop
@@ -152,19 +157,17 @@ joinLeftGeneric eq s1 s2 = Stream.evalStateT (return False) $ unCross $ do
152 else mkCross Stream.nil
157 else mkCross Stream.nil
153 Nothing -> return (a, Nothing)
158 Nothing -> return (a, Nothing)
154
159
155−-- XXX rename to joinLeftOrd?
160+-- | 'leftJoin' specialized to 'Ord' types for better performance.
156−
157−-- | A more efficient 'joinLeft' using a hashmap for efficiency.
158--
161--
159-- Space: O(n)
162-- Space: O(n)
160--
163--
161-- Time: O(m + n)
164-- Time: O(m + n)
162--
165--
163-- /Pre-release/
166-- /Pre-release/
164−{-# INLINE joinLeft #-}
167+{-# INLINE leftOrdJoin #-}
165−joinLeft :: (Ord k, Monad m) =>
168+leftOrdJoin :: (Ord k, Monad m) =>
166 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, Maybe b)
169 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, Maybe b)
167−joinLeft s1 s2 =
170+leftOrdJoin s1 s2 =
168 Stream.concatEffect $ do
171 Stream.concatEffect $ do
169 km <- toMap s2
172 km <- toMap s2
170 return $ fmap (joinAB km) s1
173 return $ fmap (joinAB km) s1
@@ -177,14 +180,19 @@ joinLeft s1 s2 =
177 Nothing -> (k, a, Nothing)
180 Nothing -> (k, a, Nothing)
178
181
179-- XXX We can do this concurrently.
182-- XXX We can do this concurrently.
183+-- XXX Check performance of StreamD vs StreamK cross operation.
180
184
181−-- XXX Check performance of StreamD vs StreamK
182−
183−-- | Like 'joinLeft' but emits a @(Just a, Just b)@. Like 'joinLeft', for those
185+-- | Like 'leftJoin' but emits a @(Just a, Just b)@. Like 'leftJoin', for those
184-- @a@'s that are not equal to any @b@ emit @(Just a, Nothing)@, but
186-- @a@'s that are not equal to any @b@ emit @(Just a, Nothing)@, but
185-- additionally, for those @b@'s that are not equal to any @a@ emit @(Nothing,
187-- additionally, for those @b@'s that are not equal to any @a@ emit @(Nothing,
186-- Just b)@.
188-- Just b)@.
187--
189--
190+-- This is a generalization of left join to include all the elements from the
191+-- right stream as well, in other words it is a combination of left and right
192+-- joins. This is a commutative operation. The order of stream arguments can be
193+-- changed without affecting results, except for the ordering of elements in
194+-- the resulting tuple.
195+--
188-- For space efficiency use the smaller stream as the second stream.
196-- For space efficiency use the smaller stream as the second stream.
189--
197--
190-- Space: O(n)
198-- Space: O(n)
@@ -192,15 +200,15 @@ joinLeft s1 s2 =
192-- Time: O(m x n)
200-- Time: O(m x n)
193--
201--
194-- /Pre-release/
202-- /Pre-release/
195−{-# INLINE joinOuterGeneric #-}
203+{-# INLINE outerJoin #-}
196−joinOuterGeneric :: MonadIO m =>
204+outerJoin :: MonadIO m =>
197 (a -> b -> Bool)
205 (a -> b -> Bool)
198 -> Stream m a
206 -> Stream m a
199 -> Stream m b
207 -> Stream m b
200 -> Stream m (Maybe a, Maybe b)
208 -> Stream m (Maybe a, Maybe b)
201−joinOuterGeneric eq s1 s =
209+outerJoin eq s1 s2 =
202 Stream.concatEffect $ do
210 Stream.concatEffect $ do
203− inputArr <- Array.fromStream s
211+ inputArr <- Array.fromStream s2
204 let len = Array.length inputArr
212 let len = Array.length inputArr
205 foundArr <-
213 foundArr <-
206 Stream.fold
214 Stream.fold
@@ -254,18 +262,18 @@ joinOuterGeneric eq s1 s =
254-- a flag. At the end go through @Stream m b@ and find those that are not in that
262-- a flag. At the end go through @Stream m b@ and find those that are not in that
255-- hash to return (Nothing, b).
263-- hash to return (Nothing, b).
256
264
257−-- | Like 'joinOuter' but uses a 'Map' for efficiency.
265+-- | 'outerJoin' specialized to 'Ord' types for better performance.
258--
266--
259-- Space: O(m + n)
267-- Space: O(m + n)
260--
268--
261-- Time: O(m + n)
269-- Time: O(m + n)
262--
270--
263-- /Pre-release/
271-- /Pre-release/
264−{-# INLINE joinOuter #-}
272+{-# INLINE outerOrdJoin #-}
265−joinOuter ::
273+outerOrdJoin ::
266 (Ord k, MonadIO m) =>
274 (Ord k, MonadIO m) =>
267 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, Maybe a, Maybe b)
275 Stream m (k, a) -> Stream m (k, b) -> Stream m (k, Maybe a, Maybe b)
268−joinOuter s1 s2 =
276+outerOrdJoin s1 s2 =
269 Stream.concatEffect $ do
277 Stream.concatEffect $ do
270 km1 <- kvFold s1
278 km1 <- kvFold s1
271 km2 <- kvFold s2
279 km2 <- kvFold s2
core/src/Streamly/Internal/Data/Stream/Eliminate.hs+33−25
@@ -13,36 +13,24 @@
13--
13--
14module Streamly.Internal.Data.Stream.Eliminate
14module Streamly.Internal.Data.Stream.Eliminate
15 (
15 (
16− -- * Running a 'Fold'
16+ -- * Running a Parser
17− fold
18−
19− -- -- * Running a 'Parser'
20− , parse
17+ parse
21 , parseD
18 , parseD
22 , parseBreak
19 , parseBreak
23 , parseBreakD
20 , parseBreakD
24
21
25− -- * Stream Deconstruction
22+ -- * Deconstruction
26 , uncons
23 , uncons
27
24
28 -- * Right Folds
25 -- * Right Folds
29− , foldrM
30− , foldr
31− , foldrMx
32 , foldr1
26 , foldr1
33
27
34− -- * Left Folds
35− , foldlM'
36− , foldl'
37− , foldlMx'
38− , foldlx'
39−
40 -- * Specific Fold Functions
28 -- * Specific Fold Functions
41− , drain
42 , mapM_ -- Map and Fold
29 , mapM_ -- Map and Fold
43 , null
30 , null
44 , head
31 , head
45 , headElse
32 , headElse
33+ , init
46 , tail
34 , tail
47 , last
35 , last
48 , elem
36 , elem
@@ -60,16 +48,9 @@ module Streamly.Internal.Data.Stream.Eliminate
60 , the
48 , the
61
49
62 -- * To containers
50 -- * To containers
63− , toList
64 , toListRev
51 , toListRev
65
52
66 -- * Multi-Stream Folds
53 -- * Multi-Stream Folds
67− -- ** Comparisons
68− -- | These should probably be expressed using zipping operations.
69− , eqBy
70− , cmpBy
71−
72− -- ** Substreams
73 -- | These should probably be expressed using parsers.
54 -- | These should probably be expressed using parsers.
74 , isPrefixOf
55 , isPrefixOf
75 , isInfixOf
56 , isInfixOf
@@ -90,7 +71,7 @@ import Foreign.Storable (Storable)
90import GHC.Exts (SpecConstrAnnotation(..))
71import GHC.Exts (SpecConstrAnnotation(..))
91import GHC.Types (SPEC(..))
72import GHC.Types (SPEC(..))
92import Streamly.Internal.Data.Parser (ParseError(..))
73import Streamly.Internal.Data.Parser (ParseError(..))
93−import Streamly.Internal.Data.SVar.Type (defState)
74+import Streamly.Internal.Data.SVar.Type (adaptState, defState)
94import Streamly.Internal.Data.Unbox (Unbox)
75import Streamly.Internal.Data.Unbox (Unbox)
95
76
96import Streamly.Internal.Data.Maybe.Strict (Maybe'(..))
77import Streamly.Internal.Data.Maybe.Strict (Maybe'(..))
@@ -105,7 +86,7 @@ import qualified Streamly.Internal.Data.Stream.Transform as StreamD
105
86
106import Prelude hiding
87import Prelude hiding
107 ( Foldable(..), all, any, head, last, lookup, mapM, mapM_
88 ( Foldable(..), all, any, head, last, lookup, mapM, mapM_
108− , notElem, splitAt, tail, (!!))
89+ , notElem, splitAt, init, tail, (!!))
109import Data.Foldable (length)
90import Data.Foldable (length)
110import Streamly.Internal.Data.Stream.Type
91import Streamly.Internal.Data.Stream.Type
111
92
@@ -391,9 +372,35 @@ head = foldrM (\x _ -> return (Just x)) (return Nothing)
391headElse :: Monad m => a -> Stream m a -> m a
372headElse :: Monad m => a -> Stream m a -> m a
392headElse a = foldrM (\x _ -> return x) (return a)
373headElse a = foldrM (\x _ -> return x) (return a)
393
374
375+{-# INLINE_NORMAL init #-}
376+init :: Monad m => Stream m a -> m (Maybe (Stream m a))
377+init stream = do
378+ r <- uncons stream
379+ case r of
380+ Nothing -> return Nothing
381+ Just (h, Stream step1 state1) ->
382+ return $ Just $ Stream step (h, state1)
383+
384+ where
385+
386+ step gst (a, s1) = do
387+ res <- step1 (adaptState gst) s1
388+ return $
389+ case res of
390+ Yield x s -> Yield a (x, s)
391+ Skip s -> Skip (a, s)
392+ Stop -> Stop
393+
394+-- | Same as:
395+--
396+-- >>> tail = fmap (fmap snd) . Stream.uncons
397+--
394-- Does not fuse, has the same performance as the StreamK version.
398-- Does not fuse, has the same performance as the StreamK version.
399+--
395{-# INLINE_NORMAL tail #-}
400{-# INLINE_NORMAL tail #-}
396tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
401tail :: Monad m => Stream m a -> m (Maybe (Stream m a))
402+tail = fmap (fmap snd) . uncons
403+{-
397tail (UnStream step state) = go SPEC state
404tail (UnStream step state) = go SPEC state
398 where
405 where
399 go !_ st = do
406 go !_ st = do
@@ -402,6 +409,7 @@ tail (UnStream step state) = go SPEC state
402 Yield _ s -> return (Just $ Stream step s)
409 Yield _ s -> return (Just $ Stream step s)
403 Skip s -> go SPEC s
410 Skip s -> go SPEC s
404 Stop -> return Nothing
411 Stop -> return Nothing
412+-}
405
413
406-- XXX will it fuse? need custom impl?
414-- XXX will it fuse? need custom impl?
407{-# INLINE_NORMAL last #-}
415{-# INLINE_NORMAL last #-}
core/src/Streamly/Internal/Data/Stream/Exception.hs+9−6
@@ -9,17 +9,20 @@
9
9
10module Streamly.Internal.Data.Stream.Exception
10module Streamly.Internal.Data.Stream.Exception
11 (
11 (
12− gbracket_
12+ -- * Resources
13− , gbracket
14− , before
13+ before
15− , afterUnsafe
16 , afterIO
14 , afterIO
15+ , afterUnsafe
16+ , finallyIO
17+ , finallyUnsafe
18+ , gbracket_
19+ , gbracket
17 , bracketUnsafe
20 , bracketUnsafe
18 , bracketIO3
21 , bracketIO3
19 , bracketIO
22 , bracketIO
23+
24+ -- * Exceptions
20 , onException
25 , onException
21− , finallyUnsafe
22− , finallyIO
23 , ghandle
26 , ghandle
24 , handle
27 , handle
25 )
28 )
core/src/Streamly/Internal/Data/Stream/Generate.hs−12
@@ -16,20 +16,13 @@ module Streamly.Internal.Data.Stream.Generate
16 (
16 (
17 -- * Primitives
17 -- * Primitives
18 nil
18 nil
19− , nilM
20 , cons
19 , cons
21− , consM
22−
23− -- * From 'Unfold'
24− , unfold
25
20
26 -- * Unfolding
21 -- * Unfolding
27 , unfoldr
22 , unfoldr
28 , unfoldrM
23 , unfoldrM
29
24
30 -- * From Values
25 -- * From Values
31− , fromPure
32− , fromEffect
33 , repeat
26 , repeat
34 , repeatM
27 , repeatM
35 , replicate
28 , replicate
@@ -95,7 +88,6 @@ module Streamly.Internal.Data.Stream.Generate
95 -- * From Containers
88 -- * From Containers
96 -- | Transform an input structure into a stream.
89 -- | Transform an input structure into a stream.
97
90
98− , fromList
99 , fromListM
91 , fromListM
100 , fromFoldable
92 , fromFoldable
101 , fromFoldableM
93 , fromFoldableM
@@ -104,10 +96,6 @@ module Streamly.Internal.Data.Stream.Generate
104 , fromPtr
96 , fromPtr
105 , fromPtrN
97 , fromPtrN
106 , fromByteStr#
98 , fromByteStr#
107−
108− -- * Conversions
109− , fromStreamK
110− , toStreamK
111 )
99 )
112where
100where
113
101
core/src/Streamly/Internal/Data/Stream/Nesting.hs+10−64
@@ -55,11 +55,6 @@ module Streamly.Internal.Data.Stream.Nesting
55 -- yields an element. A special case of unfoldRoundRobin.
55 -- yields an element. A special case of unfoldRoundRobin.
56 , roundRobin -- interleaveFair?/ParallelFair
56 , roundRobin -- interleaveFair?/ParallelFair
57
57
58− -- *** Zipping
59− -- | Zip corresponding elements of two streams.
60− , zipWith
61− , zipWithM
62−
63 -- *** Merging
58 -- *** Merging
64 -- | Interleave elements from two streams based on a condition.
59 -- | Interleave elements from two streams based on a condition.
65 , mergeBy
60 , mergeBy
@@ -76,17 +71,10 @@ module Streamly.Internal.Data.Stream.Nesting
76 -- unfoldMany: Unfold m a b -> t m a -> t m b
71 -- unfoldMany: Unfold m a b -> t m a -> t m b
77 -- @
72 -- @
78
73
79− -- *** ConcatMap
80− -- | Generate streams by mapping a stream generator on each element of an
81− -- input stream, append the resulting streams and flatten.
82− , concatMap
83− , concatMapM
84−
85 -- *** ConcatUnfold
74 -- *** ConcatUnfold
86 -- | Generate streams by using an unfold on each element of an input
75 -- | Generate streams by using an unfold on each element of an input
87 -- stream, append the resulting streams and flatten. A special case of
76 -- stream, append the resulting streams and flatten. A special case of
88 -- gintercalate.
77 -- gintercalate.
89− , unfoldMany
90 , ConcatUnfoldInterleaveState (..)
78 , ConcatUnfoldInterleaveState (..)
91 , unfoldInterleave
79 , unfoldInterleave
92 , unfoldRoundRobin
80 , unfoldRoundRobin
@@ -118,11 +106,8 @@ module Streamly.Internal.Data.Stream.Nesting
118
106
119 -- ** Folding
107 -- ** Folding
120 -- | Apply folds on a stream.
108 -- | Apply folds on a stream.
121− , foldMany
122− , refoldMany
123 , foldSequence
109 , foldSequence
124 , foldIterateM
110 , foldIterateM
125− , refoldIterateM
126
111
127 -- ** Parsing
112 -- ** Parsing
128 -- | Parsing is opposite to flattening. 'parseMany' is dual to concatMap or
113 -- | Parsing is opposite to flattening. 'parseMany' is dual to concatMap or
@@ -139,7 +124,6 @@ module Streamly.Internal.Data.Stream.Nesting
139
124
140 -- ** Grouping
125 -- ** Grouping
141 -- | Group segments of a stream and fold. Special case of parsing.
126 -- | Group segments of a stream and fold. Special case of parsing.
142− , groupsOf
143 , groupsBy
127 , groupsBy
144 , groupsWhile
128 , groupsWhile
145 , groupsRollingBy
129 , groupsRollingBy
@@ -147,8 +131,8 @@ module Streamly.Internal.Data.Stream.Nesting
147 -- ** Splitting
131 -- ** Splitting
148 -- | A special case of parsing.
132 -- | A special case of parsing.
149 , wordsBy
133 , wordsBy
150− , splitOnSeq
134+ , splitOnSeq -- XXX splitOnSeg
151− , splitOnSuffixSeq
135+ , splitOnSuffixSeq -- XXX splitOnSegSuffix, splitOnTrailer
152
136
153 -- XXX Implement these as folds or parsers instead.
137 -- XXX Implement these as folds or parsers instead.
154 , splitOnSuffixSeqAny
138 , splitOnSuffixSeqAny
@@ -157,9 +141,8 @@ module Streamly.Internal.Data.Stream.Nesting
157
141
158 -- * Transform (Nested Containers)
142 -- * Transform (Nested Containers)
159 -- | Opposite to compact in ArrayStream
143 -- | Opposite to compact in ArrayStream
160− , splitInnerBy
144+ , splitInnerBy -- XXX innerSplitOn
161− , splitInnerBySuffix
145+ , splitInnerBySuffix -- XXX innerSplitOnSuffix
162− , intersectBySorted
163
146
164 -- * Reduce By Streams
147 -- * Reduce By Streams
165 , dropPrefix
148 , dropPrefix
@@ -609,49 +592,6 @@ mergeFstBy :: -- Monad m =>
609mergeFstBy _f _m1 _m2 = undefined
592mergeFstBy _f _m1 _m2 = undefined
610 -- fromStreamK $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
593 -- fromStreamK $ D.mergeFstBy f (toStreamD m1) (toStreamD m2)
611
594
612−-------------------------------------------------------------------------------
613−-- Intersection of sorted streams
614−-------------------------------------------------------------------------------
615−
616−-- Assuming the streams are sorted in ascending order
617−{-# INLINE_NORMAL intersectBySorted #-}
618−intersectBySorted :: Monad m
619− => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
620−intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =
621− Stream step
622− ( ta -- left stream state
623− , tb -- right stream state
624− , Nothing -- left value
625− , Nothing -- right value
626− )
627−
628− where
629−
630− {-# INLINE_LATE step #-}
631− -- step 1, fetch the first value
632− step gst (sa, sb, Nothing, b) = do
633− r <- stepa gst sa
634− return $ case r of
635− Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3
636− Skip sa' -> Skip (sa', sb, Nothing, b)
637− Stop -> Stop
638−
639− -- step 2, fetch the second value
640− step gst (sa, sb, a@(Just _), Nothing) = do
641− r <- stepb gst sb
642− return $ case r of
643− Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3
644− Skip sb' -> Skip (sa, sb', a, Nothing)
645− Stop -> Stop
646−
647− -- step 3, compare the two values
648− step _ (sa, sb, Just a, Just b) = do
649− let res = cmp a b
650− return $ case res of
651− GT -> Skip (sa, sb, Just a, Nothing) -- step 2
652− LT -> Skip (sa, sb, Nothing, Just b) -- step 1
653− EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1
654−
655------------------------------------------------------------------------------
595------------------------------------------------------------------------------
656-- Combine N Streams - unfoldMany
596-- Combine N Streams - unfoldMany
657------------------------------------------------------------------------------
597------------------------------------------------------------------------------
@@ -3046,6 +2986,8 @@ splitInnerBySuffix isEmpty splitter joiner (Stream step1 state1) =
3046--
2986--
3047-- Space: @O(1)@
2987-- Space: @O(1)@
3048--
2988--
2989+-- See also stripPrefix.
2990+--
3049-- /Unimplemented/
2991-- /Unimplemented/
3050{-# INLINE dropPrefix #-}
2992{-# INLINE dropPrefix #-}
3051dropPrefix ::
2993dropPrefix ::
@@ -3058,6 +3000,8 @@ dropPrefix = error "Not implemented yet!"
3058--
3000--
3059-- Space: @O(n)@ where n is the length of the infix.
3001-- Space: @O(n)@ where n is the length of the infix.
3060--
3002--
3003+-- See also stripInfix.
3004+--
3061-- /Unimplemented/
3005-- /Unimplemented/
3062{-# INLINE dropInfix #-}
3006{-# INLINE dropInfix #-}
3063dropInfix ::
3007dropInfix ::
@@ -3070,6 +3014,8 @@ dropInfix = error "Not implemented yet!"
3070--
3014--
3071-- Space: @O(n)@ where n is the length of the suffix.
3015-- Space: @O(n)@ where n is the length of the suffix.
3072--
3016--
3017+-- See also stripSuffix.
3018+--
3073-- /Unimplemented/
3019-- /Unimplemented/
3074{-# INLINE dropSuffix #-}
3020{-# INLINE dropSuffix #-}
3075dropSuffix ::
3021dropSuffix ::
core/src/Streamly/Internal/Data/Stream/Top.hs+243−115
@@ -8,38 +8,73 @@
8-- Portability : GHC
8-- Portability : GHC
9--
9--
10-- Top level module that can depend on all other lower level Stream modules.
10-- Top level module that can depend on all other lower level Stream modules.
11+--
12+-- Design notes:
13+--
14+-- The order of arguments in the join operations should ideally be opposite. It
15+-- should be such that the infinite stream is the last one. The transformation
16+-- should be on the last argument, so if you curry the functions with all other
17+-- arguments we get a @Stream -> Stream@ function. The first stream argument
18+-- may be considered as a config or modifier for the operation.
19+--
20+-- Benefit of changing the order is that we get a more intuitive Stream ->
21+-- Stream transformation after currying all other arguments. The inner loop
22+-- streams become arguments for the transformation, more like local modifiers
23+-- for the global outer stream as the last argument. Thus we can continue using
24+-- transformations on the outer stream in a composed pipeline. Otherwise we can
25+-- use flip to flip the order.
26+--
27+-- The fact that the inner stream can be used in the loop multiple times also
28+-- tells that this is not the real effectful stream, it is more like a pure
29+-- stream or an array. In fact we may consider using an Identity streams as
30+-- inner streams in which case these functions will not look nice.
31+--
32+-- Downsides:
33+--
34+-- * Maybe less intuitive to think about, because we usually think the first
35+-- stream as the outer loop and second as the inner.
36+-- * Zip and merge operations will continue using the opposite order.
37+-- * Need to change the order of cross, crossWith operations as well
38+-- * It will be inconsistent with Data.List. The functions cannot be used as
39+-- intuitive operators.
40+--
41+-- The choice is similar to concatMap vs bind. concatMap is pipeline
42+-- composition friendly but bind is user intuition friendly. Another option is
43+-- to have other functions with a different argument order e.g. flippedCross
44+-- instead of cross.
45+--
46+-- If we change the order we have to make sure that we have a consistent
47+-- convention for set-like and the cross join operations.
11
48
12module Streamly.Internal.Data.Stream.Top
49module Streamly.Internal.Data.Stream.Top
13 (
50 (
14− -- * Transformation
15− -- ** Sampling
51+ -- * Sampling
16 -- | Value agnostic filtering.
52 -- | Value agnostic filtering.
17 strideFromThen
53 strideFromThen
18
54
19− -- * Nesting
55+ -- * Straight Joins
20− -- ** Set like operations
21− -- | These are not exactly set operations because streams are not
56+ -- | These are set-like operations but not exactly set operations because
22− -- necessarily sets, they may have duplicated elements. These operations
57+ -- streams are not necessarily sets, they may have duplicated elements.
23− -- are generic i.e. they work on streams of unconstrained types, therefore,
58+ -- These operations are generic i.e. they work on streams of unconstrained
24− -- they have quadratic performance characterstics. For better performance
59+ -- types, therefore, they have quadratic performance characterstics. For
60+ -- better performance using Set or Map structures see the
25− -- using Set structures see the Streamly.Internal.Data.Stream.Container
61+ -- Streamly.Internal.Data.Stream.Container module.
26− -- module.
62+ , intersectBy
27− , filterInStreamGenericBy
63+ , deleteFirstsBy
28− , deleteInStreamGenericBy
64+ , unionBy
29− , unionWithStreamGenericBy
65+
30−
31− -- ** Set like operations on sorted streams
66+ -- Set like operations on sorted streams
32− , filterInStreamAscBy
67+ , sortedIntersectBy
33− , deleteInStreamAscBy
68+ , sortedDeleteFirstsBy
34− , unionWithStreamAscBy
69+ , sortedUnionBy
35−
70+
36− -- ** Join operations
71+ -- * Cross Joins
37− , joinInnerGeneric
72+ , innerJoin
38−
73+
39− -- * Joins on sorted stream
74+ -- Joins on sorted stream
40− , joinInnerAscBy
75+ , innerSortedJoin
41− , joinLeftAscBy
76+ , leftSortedJoin
42− , joinOuterAscBy
77+ , outerSortedJoin
43 )
78 )
44where
79where
45
80
@@ -48,7 +83,7 @@ where
48import Control.Monad.IO.Class (MonadIO(..))
83import Control.Monad.IO.Class (MonadIO(..))
49import Data.IORef (newIORef, readIORef, modifyIORef')
84import Data.IORef (newIORef, readIORef, modifyIORef')
50import Streamly.Internal.Data.Fold.Type (Fold)
85import Streamly.Internal.Data.Fold.Type (Fold)
51−import Streamly.Internal.Data.Stream.Type (Stream, cross)
86+import Streamly.Internal.Data.Stream.Type (Stream(..), Step(..), cross)
52
87
53import qualified Data.List as List
88import qualified Data.List as List
54import qualified Streamly.Internal.Data.Fold as Fold
89import qualified Streamly.Internal.Data.Fold as Fold
@@ -110,29 +145,40 @@ strideFromThen offset stride =
110-- binary search if we have an Ord instance or Ordering returning function. The
145-- binary search if we have an Ord instance or Ordering returning function. The
111-- time complexity would then become (m x log n).
146-- time complexity would then become (m x log n).
112
147
113−-- | Like 'cross' but emits only those tuples where @a == b@ using the
148+-- | Like 'cross' but emits only those tuples where @a == b@ using the supplied
114−-- supplied equality predicate.
149+-- equality predicate. This is essentially a @cross intersection@ of two
150+-- streams.
115--
151--
116-- Definition:
152-- Definition:
117--
153--
118−-- >>> joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ Stream.cross s1 s2
154+-- >>> innerJoin eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ Stream.cross s1 s2
155+--
156+-- The second (inner) stream must be finite. Moreover, it must be either pure
157+-- or capable of multiple evaluations. If not then the caller should cache it
158+-- in an 'Data.Array.Array', if the type does not have an 'Unbox' instance then
159+-- use the Generic 'Data.Array.Generic.Array'. Convert the array to stream
160+-- before calling this function. Caching may also improve performance if the
161+-- stream is expensive to evaluate.
119--
162--
120−-- You should almost always prefer @joinInnerOrd@ over 'joinInnerGeneric' if
163+-- If you care about performance this function should be your last choice among
164+-- all inner joins. 'Streamly.Internal.Data.Unfold.innerJoin' is a much faster
165+-- fused alternative. 'innerSortedJoin' is a faster alternative when streams
121−-- possible. @joinInnerOrd@ is an order of magnitude faster but may take more
166+-- are sorted. 'innerOrdJoin' is an order of magnitude faster alternative when
122−-- space for caching the second stream.
167+-- the type has an 'Ord' instance.
123--
168--
124−-- See 'Streamly.Internal.Data.Unfold.joinInnerGeneric' for a much faster fused
169+-- Note: Conceptually, this is a commutative operation. Result includes all the
125−-- alternative.
170+-- elements from the left and the right stream. The order of streams can be
171+-- changed without affecting results, except for the ordering within the tuple.
126--
172--
127-- Time: O(m x n)
173-- Time: O(m x n)
128--
174--
129-- /Pre-release/
175-- /Pre-release/
130−{-# INLINE joinInnerGeneric #-}
176+{-# INLINE innerJoin #-}
131−joinInnerGeneric :: Monad m =>
177+innerJoin :: Monad m =>
132 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
178 (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
133−joinInnerGeneric eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
179+innerJoin eq s1 s2 = Stream.filter (\(a, b) -> a `eq` b) $ cross s1 s2
134{-
180{-
135−joinInnerGeneric eq s1 s2 = do
181+innerJoin eq s1 s2 = do
136 -- ConcatMap works faster than bind
182 -- ConcatMap works faster than bind
137 Stream.concatMap (\a ->
183 Stream.concatMap (\a ->
138 Stream.concatMap (\b ->
184 Stream.concatMap (\b ->
@@ -143,44 +189,44 @@ joinInnerGeneric eq s1 s2 = do
143 ) s1
189 ) s1
144-}
190-}
145
191
146−-- | A more efficient 'joinInner' for sorted streams.
192+-- | A more efficient 'innerJoin' for sorted streams.
147--
193--
148-- Space: O(1)
194-- Space: O(1)
149--
195--
150-- Time: O(m + n)
196-- Time: O(m + n)
151--
197--
152-- /Unimplemented/
198-- /Unimplemented/
153−{-# INLINE joinInnerAscBy #-}
199+{-# INLINE innerSortedJoin #-}
154−joinInnerAscBy ::
200+innerSortedJoin ::
155 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
201 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
156−joinInnerAscBy = undefined
202+innerSortedJoin = undefined
157
203
158−-- | A more efficient 'joinLeft' for sorted streams.
204+-- | A more efficient 'leftJoin' for sorted streams.
159--
205--
160-- Space: O(1)
206-- Space: O(1)
161--
207--
162-- Time: O(m + n)
208-- Time: O(m + n)
163--
209--
164-- /Unimplemented/
210-- /Unimplemented/
165−{-# INLINE joinLeftAscBy #-}
211+{-# INLINE leftSortedJoin #-}
166−joinLeftAscBy :: -- Monad m =>
212+leftSortedJoin :: -- Monad m =>
167 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
213 (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
168−joinLeftAscBy _eq _s1 _s2 = undefined
214+leftSortedJoin _eq _s1 _s2 = undefined
169
215
170−-- | A more efficient 'joinOuter' for sorted streams.
216+-- | A more efficient 'outerJoin' for sorted streams.
171--
217--
172-- Space: O(1)
218-- Space: O(1)
173--
219--
174-- Time: O(m + n)
220-- Time: O(m + n)
175--
221--
176-- /Unimplemented/
222-- /Unimplemented/
177−{-# INLINE joinOuterAscBy #-}
223+{-# INLINE outerSortedJoin #-}
178−joinOuterAscBy :: -- Monad m =>
224+outerSortedJoin :: -- Monad m =>
179 (a -> b -> Ordering)
225 (a -> b -> Ordering)
180 -> Stream m a
226 -> Stream m a
181 -> Stream m b
227 -> Stream m b
182 -> Stream m (Maybe a, Maybe b)
228 -> Stream m (Maybe a, Maybe b)
183−joinOuterAscBy _eq _s1 _s2 = undefined
229+outerSortedJoin _eq _s1 _s2 = undefined
184
230
185------------------------------------------------------------------------------
231------------------------------------------------------------------------------
186-- Set operations (special joins)
232-- Set operations (special joins)
@@ -190,8 +236,8 @@ joinOuterAscBy _eq _s1 _s2 = undefined
190-- the best would be to use an Array with linear search. If the second stream
236-- the best would be to use an Array with linear search. If the second stream
191-- is sorted we can also use a binary search, using Ord constraint.
237-- is sorted we can also use a binary search, using Ord constraint.
192
238
193−-- | Keep only those elements in the second stream that are present in the
239+-- | Keep only those elements in the first stream that are present in the
194−-- first stream too. The first stream is folded to a container using the
240+-- second stream too. The second stream is folded to a container using the
195-- supplied fold and then the elements in the container are looked up using the
241-- supplied fold and then the elements in the container are looked up using the
196-- supplied lookup function.
242-- supplied lookup function.
197--
243--
@@ -206,23 +252,30 @@ filterStreamWith :: Monad m =>
206filterStreamWith fld member s1 s2 =
252filterStreamWith fld member s1 s2 =
207 Stream.concatEffect
253 Stream.concatEffect
208 $ do
254 $ do
209− xs <- Stream.fold fld s1
255+ xs <- Stream.fold fld s2
210− return $ Stream.filter (`member` xs) s2
256+ return $ Stream.filter (`member` xs) s1
211
257
212−-- | 'filterInStreamGenericBy' retains only those elements in the second stream that
258+-- XXX instead of folding the second stream to a list we could use it directly.
213−-- are present in the first stream.
259+-- If the user wants they can generate the stream from an array and also call
214−--
260+-- uniq or nub on it. We can provide a convenience Stream -> Stream to cache
215−-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
261+-- a finite stream in an array and serve it from the cache. The user can decide
216−-- [2,1,1]
262+-- what is best based on the context. They can also choose to use a boxed or
217−--
263+-- unboxed array for caching. To force caching we can make the second stream
218−-- >>> Stream.fold Fold.toList $ Stream.filterInStreamGenericBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
264+-- monad type Identity. But that may be less flexible. One option is to use
219−-- [1,2,2]
265+-- cachedIntersectBy etc for automatic caching.
266+
267+-- | 'intersectBy' returns a subsequence of the first stream which intersects
268+-- with the second stream. Note that this is not a commutative operation unlike
269+-- a set intersection, because of duplicate elements in the stream the order of
270+-- the streams matters. This is similar to 'Data.List.intersectBy'. Note that
271+-- intersectBy is a special case of 'innerJoin'.
220--
272--
221−-- Similar to the list intersectBy operation but with the stream argument order
273+-- >>> f s1 s2 = Stream.fold Fold.toList $ Stream.intersectBy (==) (Stream.fromList s1) (Stream.fromList s2)
222−-- flipped.
274+-- >>> f [1,3,4,4,5]) [2,3,4,5,5]
275+-- [3,4,4,5]
223--
276--
224−-- The first stream must be finite and must not block. Second stream is
277+-- First stream can be infinite, the second stream must be finite and must be
225−-- processed only after the first stream is fully realized.
278+-- capable of multiple evaluations.
226--
279--
227-- Space: O(n) where @n@ is the number of elements in the second stream.
280-- Space: O(n) where @n@ is the number of elements in the second stream.
228--
281--
@@ -230,46 +283,87 @@ filterStreamWith fld member s1 s2 =
230-- @n@ is the number of elements in the second stream.
283-- @n@ is the number of elements in the second stream.
231--
284--
232-- /Pre-release/
285-- /Pre-release/
233−{-# INLINE filterInStreamGenericBy #-}
286+{-# INLINE intersectBy #-}
234−filterInStreamGenericBy :: Monad m =>
287+intersectBy :: Monad m =>
235 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
288 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
236−filterInStreamGenericBy eq =
289+intersectBy eq =
237 -- XXX Use an (unboxed) array instead.
290 -- XXX Use an (unboxed) array instead.
238 filterStreamWith
291 filterStreamWith
239 (Fold.scanMaybe (Fold.uniqBy eq) Fold.toListRev)
292 (Fold.scanMaybe (Fold.uniqBy eq) Fold.toListRev)
240 (List.any . eq)
293 (List.any . eq)
241
294
295+-------------------------------------------------------------------------------
296+-- Intersection of sorted streams
297+-------------------------------------------------------------------------------
298+
299+-- XXX The sort order is not important as long both the streams have the same
300+-- sort order. We need to move only in one direction in each stream.
301+-- XXX Fix the argument order to use the same behavior as intersectBy.
302+
242−-- | Like 'filterInStreamGenericBy' but assumes that the input streams are sorted in
303+-- | Like 'intersectBy' but assumes that the input streams are sorted in
243-- ascending order. To use it on streams sorted in descending order pass an
304-- ascending order. To use it on streams sorted in descending order pass an
244-- inverted comparison function returning GT for less than and LT for greater
305-- inverted comparison function returning GT for less than and LT for greater
245-- than.
306-- than.
246--
307--
308+-- Both streams can be infinite.
309+--
247-- Space: O(1)
310-- Space: O(1)
248--
311--
249-- Time: O(m+n)
312-- Time: O(m+n)
250--
313--
251-- /Pre-release/
314-- /Pre-release/
252−{-# INLINE filterInStreamAscBy #-}
315+{-# INLINE_NORMAL sortedIntersectBy #-}
253−filterInStreamAscBy :: Monad m =>
316+sortedIntersectBy :: Monad m =>
254 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
317 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
255−filterInStreamAscBy eq s1 s2 = Stream.intersectBySorted eq s2 s1
318+sortedIntersectBy cmp (Stream stepa ta) (Stream stepb tb) =
319+ Stream step
320+ ( ta -- left stream state
321+ , tb -- right stream state
322+ , Nothing -- left value
323+ , Nothing -- right value
324+ )
256
325
257−-- | Delete all elements of the first stream from the seconds stream. If an
326+ where
258−-- element occurs multiple times in the first stream as many occurrences of it
327+
259−-- are deleted from the second stream.
328+ {-# INLINE_LATE step #-}
260−--
329+ -- step 1, fetch the first value
261−-- >>> Stream.fold Fold.toList $ Stream.deleteInStreamGenericBy (==) (Stream.fromList [1,2,3]) (Stream.fromList [1,2,2])
330+ step gst (sa, sb, Nothing, b) = do
262−-- [2]
331+ r <- stepa gst sa
332+ return $ case r of
333+ Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3
334+ Skip sa' -> Skip (sa', sb, Nothing, b)
335+ Stop -> Stop
336+
337+ -- step 2, fetch the second value
338+ step gst (sa, sb, a@(Just _), Nothing) = do
339+ r <- stepb gst sb
340+ return $ case r of
341+ Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3
342+ Skip sb' -> Skip (sa, sb', a, Nothing)
343+ Stop -> Stop
344+
345+ -- step 3, compare the two values
346+ step _ (sa, sb, Just a, Just b) = do
347+ let res = cmp a b
348+ return $ case res of
349+ GT -> Skip (sa, sb, Just a, Nothing) -- step 2
350+ LT -> Skip (sa, sb, Nothing, Just b) -- step 1
351+ EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1
352+
353+-- | Returns a subsequence of the first stream, deleting first occurrences of
354+-- those elements that are present in the second stream. Note that this is not
355+-- a commutative operation. This is similar to the 'Data.List.deleteFirstsBy'.
263--
356--
264−-- The following laws hold:
357+-- >>> f xs ys = Stream.fold Fold.toList $ Stream.deleteFirstsBy (==) (Stream.fromList xs) (Stream.fromList ys)
358+-- >>> f [1,2,2,3,3,5] [1,2,2,3,4]
359+-- [2,3,5]
265--
360--
266−-- > deleteInStreamGenericBy (==) s1 (s1 `append` s2) === s2
361+-- The following holds:
267−-- > deleteInStreamGenericBy (==) s1 (s1 `interleave` s2) === s2
268--
362--
269−-- Same as the list 'Data.List.//' operation but with argument order flipped.
363+-- > deleteFirstsBy (==) (Stream.nub s2 `append` s1) s2 === s1
364+-- > deleteFirstsBy (==) (Stream.nub s2 `interleave` s1) s2 === s1
270--
365--
271−-- The first stream must be finite and must not block. Second stream is
366+-- First stream can be infinite, second stream must be finite.
272−-- processed only after the first stream is fully realized.
273--
367--
274-- Space: O(m) where @m@ is the number of elements in the first stream.
368-- Space: O(m) where @m@ is the number of elements in the first stream.
275--
369--
@@ -277,58 +371,83 @@ filterInStreamAscBy eq s1 s2 = Stream.intersectBySorted eq s2 s1
277-- @n@ is the number of elements in the second stream.
371-- @n@ is the number of elements in the second stream.
278--
372--
279-- /Pre-release/
373-- /Pre-release/
280−{-# INLINE deleteInStreamGenericBy #-}
374+{-# INLINE deleteFirstsBy #-}
281−deleteInStreamGenericBy :: Monad m =>
375+deleteFirstsBy :: Monad m =>
282 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
376 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
283−deleteInStreamGenericBy eq s1 s2 =
377+deleteFirstsBy eq s2 s1 =
284− Stream.concatEffect
378+ -- XXX s2 can be a sorted mutable array and we can use binary
285− $ do
379+ -- search to find. Mark the element deleted, count the deletions
286− -- This may work well if s1 is small
380+ -- and reconsolidate the array when a min number of elements is
287− -- If s1 is big we can go through s1, deleting elements from s2 and
381+ -- deleted.
288− -- not emitting an element if it was successfully deleted from s2.
289− -- we will need a deleteBy that can return whether the element was
290− -- deleted or not.
291− xs <- Stream.fold Fold.toList s2
292− let f = Fold.foldl' (flip (List.deleteBy eq)) xs
293− fmap Stream.fromList $ Stream.fold f s1
294
382
383+ -- XXX Use StreamK or list as second argument instead of Stream to avoid
384+ -- concatEffect?
385+ Stream.concatEffect $ do
386+ xs <- Stream.toList s1
387+ -- It reverses the list but that is fine.
388+ let del x =
389+ List.foldl' (\(ys,res) y ->
390+ if x `eq` y
391+ then (ys, True)
392+ else (x:ys, res)) ([], False)
393+ g (ys,_) x =
394+ let (ys1, deleted) = del x ys
395+ in if deleted
396+ then (ys1, Nothing)
397+ else (ys1, Just x)
398+ in return
399+ $ Stream.catMaybes
400+ $ fmap snd
401+ $ Stream.postscanl' g (xs, Nothing) s2
402+
295−-- | A more efficient 'deleteInStreamGenericBy' for streams sorted in ascending order.
403+-- | A more efficient 'deleteFirstsBy' for streams sorted in ascending order.
404+--
405+-- Both streams can be infinite.
296--
406--
297-- Space: O(1)
407-- Space: O(1)
298--
408--
299-- /Unimplemented/
409-- /Unimplemented/
300−{-# INLINE deleteInStreamAscBy #-}
410+{-# INLINE sortedDeleteFirstsBy #-}
301−deleteInStreamAscBy :: -- (Monad m) =>
411+sortedDeleteFirstsBy :: -- (Monad m) =>
302 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
412 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
303−deleteInStreamAscBy _eq _s1 _s2 = undefined
413+sortedDeleteFirstsBy _eq _s1 _s2 = undefined
304
414
305-- XXX Remove the MonadIO constraint. We can just cache one stream and then
415-- XXX Remove the MonadIO constraint. We can just cache one stream and then
306-- implement using differenceEqBy.
416-- implement using differenceEqBy.
307
417
308−-- | This essentially appends to the second stream all the occurrences of
418+-- | Returns the first stream appended with those unique elements from the
309−-- elements in the first stream that are not already present in the second
419+-- second stream that are not already present in the first stream. Note that
310−-- stream.
420+-- this is not a commutative operation unlike a set union, argument order
421+-- matters. The behavior is similar to the 'Data.List.unionBy'.
311--
422--
312-- Equivalent to the following except that @s2@ is evaluated only once:
423-- Equivalent to the following except that @s2@ is evaluated only once:
313--
424--
314−-- >>> unionWithStreamGenericBy eq s1 s2 = s2 `Stream.append` (Stream.deleteInStreamGenericBy eq s2 s1)
425+-- >>> unionBy eq s1 s2 = s1 `Stream.append` Stream.deleteFirstsBy eq s1 (Stream.nub s2)
315--
426--
316-- Example:
427-- Example:
317--
428--
318−-- >>> Stream.fold Fold.toList $ Stream.unionWithStreamGenericBy (==) (Stream.fromList [1,1,2,3]) (Stream.fromList [1,2,2,4])
429+-- >>> f s1 s2 = Stream.fold Fold.toList $ Stream.unionBy (==) (Stream.fromList s1) (Stream.fromList s2)
430+-- >>> f [1,2,2,4] [1,1,2,3,3]
319-- [1,2,2,4,3]
431-- [1,2,2,4,3]
320--
432--
433+-- First stream can be infinite, but second stream must be finite. Note that if
434+-- the first stream is infinite the union means just the first stream. Thus
435+-- union is useful only when both streams are finite. See 'sortedUnionBy' where
436+-- union can work on infinite streams if they are sorted.
437+--
321-- Space: O(n)
438-- Space: O(n)
322--
439--
323-- Time: O(m x n)
440-- Time: O(m x n)
324--
441--
325-- /Pre-release/
442-- /Pre-release/
326−{-# INLINE unionWithStreamGenericBy #-}
443+{-# INLINE unionBy #-}
327−unionWithStreamGenericBy :: MonadIO m =>
444+unionBy :: MonadIO m =>
328 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
445 (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
329−unionWithStreamGenericBy eq s1 s2 =
446+unionBy eq s2 s1 =
330 Stream.concatEffect
447 Stream.concatEffect
331 $ do
448 $ do
449+ -- XXX use a rewrite rule such that if a list converted to stream
450+ -- is passed to unionBy then this becomes an identity operation.
332 xs <- Stream.fold Fold.toList s1
451 xs <- Stream.fold Fold.toList s1
333 -- XXX we can use postscanlMAfter' instead of IORef
452 -- XXX we can use postscanlMAfter' instead of IORef
334 ref <- liftIO $ newIORef $! List.nubBy eq xs
453 ref <- liftIO $ newIORef $! List.nubBy eq xs
@@ -341,12 +460,21 @@ unionWithStreamGenericBy eq s1 s2 =
341 return $ Stream.fromList xs1
460 return $ Stream.fromList xs1
342 return $ Stream.mapM f s2 `Stream.append` s3
461 return $ Stream.mapM f s2 `Stream.append` s3
343
462
344−-- | A more efficient 'unionWithStreamGenericBy' for sorted streams.
463+-- | A more efficient 'unionBy' for sorted streams.
464+--
465+-- Note that the behavior is different from 'unionBy'. In 'unionBy' we append
466+-- the unique elements from second stream only after exhausting the first one
467+-- whereas in sorted streams we can determine unique elements early even when
468+-- we are going through the first stream. Thus the result is an interleaving of
469+-- the two streams, merging those elements from the second stream that are not
470+-- present in the first.
345--
471--
346-- Space: O(1)
472-- Space: O(1)
347--
473--
474+-- Both streams can be infinite.
475+--
348-- /Unimplemented/
476-- /Unimplemented/
349−{-# INLINE unionWithStreamAscBy #-}
477+{-# INLINE sortedUnionBy #-}
350−unionWithStreamAscBy :: -- (Monad m) =>
478+sortedUnionBy :: -- (Monad m) =>
351 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
479 (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
352−unionWithStreamAscBy _eq _s1 _s2 = undefined
480+sortedUnionBy _eq _s1 _s2 = undefined
core/src/Streamly/Internal/Data/Stream/Transform.hs+52−26
@@ -21,8 +21,6 @@ module Streamly.Internal.Data.Stream.Transform
21
21
22 -- * Mapping
22 -- * Mapping
23 -- | Stateless one-to-one maps.
23 -- | Stateless one-to-one maps.
24− , map
25− , mapM
26 , sequence
24 , sequence
27
25
28 -- * Mapping Effects
26 -- * Mapping Effects
@@ -75,7 +73,7 @@ module Streamly.Internal.Data.Stream.Transform
75 , scanMaybe
73 , scanMaybe
76 , filter
74 , filter
77 , filterM
75 , filterM
78− , deleteBy
76+ , deleteBy -- deleteOnceBy?
79 , uniqBy
77 , uniqBy
80 , uniq
78 , uniq
81 , prune
79 , prune
@@ -83,9 +81,8 @@ module Streamly.Internal.Data.Stream.Transform
83
81
84 -- * Trimming
82 -- * Trimming
85 -- | Produce a subset of the stream trimmed at ends.
83 -- | Produce a subset of the stream trimmed at ends.
86− , take
84+ , initNonEmpty
87− , takeWhile
85+ , tailNonEmpty
88− , takeWhileM
89 , takeWhileLast
86 , takeWhileLast
90 , takeWhileAround
87 , takeWhileAround
91 , drop
88 , drop
@@ -132,7 +129,6 @@ module Streamly.Internal.Data.Stream.Transform
132 -- * Searching
129 -- * Searching
133 , findIndices
130 , findIndices
134 , elemIndices
131 , elemIndices
135− , slicesBy
136
132
137 -- * Rolling map
133 -- * Rolling map
138 -- | Map using the previous element.
134 -- | Map using the previous element.
@@ -1074,6 +1070,55 @@ repeated = undefined
1074-- Trimming
1070-- Trimming
1075------------------------------------------------------------------------------
1071------------------------------------------------------------------------------
1076
1072
1073+-- | init for non-empty streams, fails for empty stream case.
1074+--
1075+{-# INLINE initNonEmpty #-}
1076+initNonEmpty :: Monad m => Stream m a -> Stream m a
1077+initNonEmpty (Stream step1 state1) = Stream step (Nothing, state1)
1078+
1079+ where
1080+
1081+ step gst (Nothing, s1) = do
1082+ r <- step1 (adaptState gst) s1
1083+ return $
1084+ case r of
1085+ Yield x s -> Skip (Just x, s)
1086+ Skip s -> Skip (Nothing, s)
1087+ Stop -> error "initNonEmpty: empty Stream"
1088+
1089+ step gst (Just a, s1) = do
1090+ r <- step1 (adaptState gst) s1
1091+ return $
1092+ case r of
1093+ Yield x s -> Yield a (Just x, s)
1094+ Skip s -> Skip (Just a, s)
1095+ Stop -> Stop
1096+
1097+-- | tail for non-empty streams, fails for empty stream case.
1098+--
1099+-- See also 'tail' for a non-partial version of this function..
1100+{-# INLINE tailNonEmpty #-}
1101+tailNonEmpty :: Monad m => Stream m a -> Stream m a
1102+tailNonEmpty (Stream step1 state1) = Stream step (Nothing, state1)
1103+
1104+ where
1105+
1106+ step gst (Nothing, s1) = do
1107+ r <- step1 (adaptState gst) s1
1108+ return $
1109+ case r of
1110+ Yield x s -> Skip (Just x, s)
1111+ Skip s -> Skip (Nothing, s)
1112+ Stop -> error "tailNonEmpty: empty Stream"
1113+
1114+ step gst (Just a, s1) = do
1115+ r <- step1 (adaptState gst) s1
1116+ return $
1117+ case r of
1118+ Yield x s -> Yield x (Just x, s)
1119+ Skip s -> Skip (Just a, s)
1120+ Stop -> Stop
1121+
1077-- | Take all consecutive elements at the end of the stream for which the
1122-- | Take all consecutive elements at the end of the stream for which the
1078-- predicate is true.
1123-- predicate is true.
1079--
1124--
@@ -1709,25 +1754,6 @@ findIndices p (Stream step state) = Stream step' (state, 0)
1709elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
1754elemIndices :: (Monad m, Eq a) => a -> Stream m a -> Stream m Int
1710elemIndices a = findIndices (== a)
1755elemIndices a = findIndices (== a)
1711
1756
1712−{-# INLINE_NORMAL slicesBy #-}
1713−slicesBy :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
1714−slicesBy p (Stream step1 state1) = Stream step (Just (state1, 0, 0))
1715−
1716− where
1717−
1718− {-# INLINE_LATE step #-}
1719− step gst (Just (st, i, len)) = i `seq` len `seq` do
1720− r <- step1 (adaptState gst) st
1721− return
1722− $ case r of
1723− Yield x s ->
1724− if p x
1725− then Yield (i, len + 1) (Just (s, i + len + 1, 0))
1726− else Skip (Just (s, i, len + 1))
1727− Skip s -> Skip (Just (s, i, len))
1728− Stop -> if len == 0 then Stop else Yield (i, len) Nothing
1729− step _ Nothing = return Stop
1730−
1731------------------------------------------------------------------------------
1757------------------------------------------------------------------------------
1732-- Rolling map
1758-- Rolling map
1733------------------------------------------------------------------------------
1759------------------------------------------------------------------------------
core/src/Streamly/Internal/Data/Stream/Transformer.hs+2−1
@@ -11,10 +11,11 @@
11
11
12module Streamly.Internal.Data.Stream.Transformer
12module Streamly.Internal.Data.Stream.Transformer
13 (
13 (
14+ -- * Fold to Transformer Monad
14 foldlT
15 foldlT
15 , foldrT
16 , foldrT
16
17
17− -- * Transform Inner Monad
18+ -- * Inner Monad Operations
18 , liftInner
19 , liftInner
19 , runReaderT
20 , runReaderT
20 , usingReaderT
21 , usingReaderT
core/src/Streamly/Internal/Data/Stream/Type.hs+31−3
@@ -22,18 +22,18 @@
22
22
23module Streamly.Internal.Data.Stream.Type
23module Streamly.Internal.Data.Stream.Type
24 (
24 (
25− -- * The stream type
25+ -- * Type
26 Step (..)
26 Step (..)
27 -- XXX UnStream is exported to avoid a performance issue in some
27 -- XXX UnStream is exported to avoid a performance issue in some
28 -- combinators if we use the pattern synonym "Stream".
28 -- combinators if we use the pattern synonym "Stream".
29 , Stream (Stream, UnStream)
29 , Stream (Stream, UnStream)
30
30
31− -- * CrossStream type wrapper
31+ -- * CrossStream
32 , CrossStream
32 , CrossStream
33 , unCross
33 , unCross
34 , mkCross
34 , mkCross
35
35
36− -- * Conversion to StreamK
36+ -- * To StreamK
37 , fromStreamK
37 , fromStreamK
38 , toStreamK
38 , toStreamK
39
39
@@ -91,6 +91,7 @@ module Streamly.Internal.Data.Stream.Type
91
91
92 -- * Combining Two Streams
92 -- * Combining Two Streams
93 -- ** Zipping
93 -- ** Zipping
94+ -- | Zip corresponding elements of two streams.
94 , zipWithM
95 , zipWithM
95 , zipWith
96 , zipWith
96
97
@@ -106,6 +107,8 @@ module Streamly.Internal.Data.Stream.Type
106 , unfoldMany
107 , unfoldMany
107
108
108 -- * Concat
109 -- * Concat
110+ -- | Generate streams by mapping a stream generator on each element of an
111+ -- input stream, append the resulting streams and flatten.
109 , concatEffect
112 , concatEffect
110 , concatMap
113 , concatMap
111 , concatMapM
114 , concatMapM
@@ -139,6 +142,7 @@ module Streamly.Internal.Data.Stream.Type
139 , indexOnSuffix
142 , indexOnSuffix
140
143
141 -- * Multi-stream folds
144 -- * Multi-stream folds
145+ -- | These should probably be expressed using zipping operations.
142 , eqBy
146 , eqBy
143 , cmpBy
147 , cmpBy
144
148
@@ -568,8 +572,12 @@ foldAdd f =
568-- >>> Stream.foldrM step (return False) s
572-- >>> Stream.foldrM step (return False) s
569-- True
573-- True
570--
574--
575+-- >>> import Control.Monad (join)
576+-- >>> foldrM f z = join $ Stream.foldr f z
577+--
571{-# INLINE_NORMAL foldrM #-}
578{-# INLINE_NORMAL foldrM #-}
572foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b
579foldrM :: Monad m => (a -> m b -> m b) -> m b -> Stream m a -> m b
580+-- foldrM f z = join . Streamly.Internal.Data.Stream.StreamD.Type.foldr f z
573foldrM f z (Stream step state) = go SPEC state
581foldrM f z (Stream step state) = go SPEC state
574 where
582 where
575 {-# INLINE_LATE go #-}
583 {-# INLINE_LATE go #-}
@@ -2068,6 +2076,26 @@ indexOnSuffix predicate =
2068 (indexerBy (FL.takeEndBy_ predicate FL.length) 1)
2076 (indexerBy (FL.takeEndBy_ predicate FL.length) 1)
2069 (return (-1, 0))
2077 (return (-1, 0))
2070
2078
2079+-- Alternate implementation
2080+{-# INLINE_NORMAL _indexOnSuffix #-}
2081+_indexOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
2082+_indexOnSuffix p (Stream step1 state1) = Stream step (Just (state1, 0, 0))
2083+
2084+ where
2085+
2086+ {-# INLINE_LATE step #-}
2087+ step gst (Just (st, i, len)) = i `seq` len `seq` do
2088+ r <- step1 (adaptState gst) st
2089+ return
2090+ $ case r of
2091+ Yield x s ->
2092+ if p x
2093+ then Yield (i, len + 1) (Just (s, i + len + 1, 0))
2094+ else Skip (Just (s, i, len + 1))
2095+ Skip s -> Skip (Just (s, i, len))
2096+ Stop -> if len == 0 then Stop else Yield (i, len) Nothing
2097+ step _ Nothing = return Stop
2098+
2071{-# DEPRECATED sliceOnSuffix "Please use indexOnSuffix instead." #-}
2099{-# DEPRECATED sliceOnSuffix "Please use indexOnSuffix instead." #-}
2072sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
2100sliceOnSuffix :: Monad m => (a -> Bool) -> Stream m a -> Stream m (Int, Int)
2073sliceOnSuffix = indexOnSuffix
2101sliceOnSuffix = indexOnSuffix
core/src/Streamly/Internal/Data/StreamK.hs+19−8
@@ -16,16 +16,15 @@ module Streamly.Internal.Data.StreamK
16 --
16 --
17 -- $setup
17 -- $setup
18
18
19− -- * The stream type
20 module Streamly.Internal.Data.StreamK.Type
19 module Streamly.Internal.Data.StreamK.Type
20+ -- * Transformer
21 , module Streamly.Internal.Data.StreamK.Transformer
21 , module Streamly.Internal.Data.StreamK.Transformer
22
22
23− , StreamK(..)
23+ -- * From containers
24 , fromList
24 , fromList
25 , fromStream
25 , fromStream
26− , toStream
27
26
28− -- ** Specialized Generation
27+ -- * Specialized Generation
29 , repeatM
28 , repeatM
30 , replicate
29 , replicate
31 , replicateM
30 , replicateM
@@ -67,13 +66,13 @@ module Streamly.Internal.Data.StreamK
67 , find
66 , find
68 , (!!)
67 , (!!)
69
68
69+ -- ** To Containers
70+ , toList
71+ , toStream
72+
70 -- ** Map and Fold
73 -- ** Map and Fold
71 , mapM_
74 , mapM_
72
75
73− -- ** Conversions
74− , toList
75− , hoist
76−
77 -- * Transformation
76 -- * Transformation
78 -- ** By folding (scans)
77 -- ** By folding (scans)
79 , scanl'
78 , scanl'
@@ -100,6 +99,7 @@ module Streamly.Internal.Data.StreamK
100
99
101 -- ** Reordering
100 -- ** Reordering
102 , sortBy
101 , sortBy
102+ , sortOn
103
103
104 -- ** Map and Filter
104 -- ** Map and Filter
105 , mapMaybe
105 , mapMaybe
@@ -115,6 +115,9 @@ module Streamly.Internal.Data.StreamK
115 -- ** Transformation comprehensions
115 -- ** Transformation comprehensions
116 , the
116 , the
117
117
118+ -- ** Transforming Inner Monad
119+ , hoist
120+
118 -- * Exceptions
121 -- * Exceptions
119 , handle
122 , handle
120
123
@@ -131,6 +134,7 @@ import Control.Exception (mask_, Exception)
131import Control.Monad (void, join)
134import Control.Monad (void, join)
132import Control.Monad.Catch (MonadCatch)
135import Control.Monad.Catch (MonadCatch)
133import Control.Monad.IO.Class (MonadIO(..))
136import Control.Monad.IO.Class (MonadIO(..))
137+import Data.Ord (comparing)
134import Data.Proxy (Proxy(..))
138import Data.Proxy (Proxy(..))
135import GHC.Types (SPEC(..))
139import GHC.Types (SPEC(..))
136import Streamly.Internal.Data.Array.Type (Array(..))
140import Streamly.Internal.Data.Array.Type (Array(..))
@@ -1717,3 +1721,10 @@ sortBy cmp =
1717 . Stream.catRights -- its a non-failing backtracking parser
1721 . Stream.catRights -- its a non-failing backtracking parser
1718 . Stream.parseMany (fmap (either id id) p)
1722 . Stream.parseMany (fmap (either id id) p)
1719 . Stream.fromStreamK
1723 . Stream.fromStreamK
1724+
1725+{-# INLINE sortOn #-}
1726+sortOn :: (Monad m, Ord b) => (a -> b) -> StreamK m a -> StreamK m a
1727+sortOn f =
1728+ fmap snd
1729+ . sortBy (comparing fst)
1730+ . fmap (\x -> let y = f x in y `seq` (y, x))
core/src/Streamly/Internal/Data/StreamK/Type.hs+56−11
@@ -97,8 +97,11 @@ module Streamly.Internal.Data.StreamK.Type
97 -- ** Specific Folds
97 -- ** Specific Folds
98 , drain
98 , drain
99 , null
99 , null
100+ , headNonEmpty
100 , tail
101 , tail
102+ , tailNonEmpty
101 , init
103 , init
104+ , initNonEmpty
102
105
103 -- * Mapping
106 -- * Mapping
104 , map
107 , map
@@ -1811,14 +1814,42 @@ iterateMWith cns step = go
1811 !next <- s
1814 !next <- s
1812 foldStreamShared st stp sng yld (return next `cns` go (step next))
1815 foldStreamShared st stp sng yld (return next `cns` go (step next))
1813
1816
1814−{-# INLINE headPartial #-}
1817+-- | head for non-empty streams, fails for empty stream case.
1818+--
1819+{-# INLINE headNonEmpty #-}
1815−headPartial :: Monad m => StreamK m a -> m a
1820+headNonEmpty :: Monad m => StreamK m a -> m a
1816−headPartial = foldrM (\x _ -> return x) (error "head of nil")
1821+headNonEmpty = foldrM (\x _ -> return x) (error "headNonEmpty: empty stream")
1822+
1823+-- | init for non-empty streams, fails for empty stream case.
1824+--
1825+-- See also 'init' for a non-partial version of this function..
1826+{-# INLINE initNonEmpty #-}
1827+initNonEmpty :: Stream m a -> Stream m a
1828+initNonEmpty = go0
1829+
1830+ where
1831+
1832+ go0 m = mkStream $ \st yld sng stp ->
1833+ let stop = error "initNonEmpty: Empty Stream."
1834+ single _ = stp
1835+ yieldk a r = foldStream st yld sng stp (go1 a r)
1836+ in foldStream st yieldk single stop m
1837+
1838+ go1 a r = mkStream $ \st yld sng stp ->
1839+ let stop = stp
1840+ single _ = sng a
1841+ yieldk a1 r1 = yld a (go1 a1 r1)
1842+ in foldStream st yieldk single stop r
1817
1843
1818−{-# INLINE tailPartial #-}
1844+-- | tail for non-empty streams, fails for empty stream case.
1845+--
1846+-- See also 'tail' for a non-partial version of this function..
1847+--
1848+-- Note: this is same as "drop 1" with error on empty stream.
1849+{-# INLINE tailNonEmpty #-}
1819−tailPartial :: StreamK m a -> StreamK m a
1850+tailNonEmpty :: StreamK m a -> StreamK m a
1820−tailPartial m = mkStream $ \st yld sng stp ->
1851+tailNonEmpty m = mkStream $ \st yld sng stp ->
1821− let stop = error "tail of nil"
1852+ let stop = error "tailNonEmpty: empty stream"
1822 single _ = stp
1853 single _ = stp
1823 yieldk _ r = foldStream st yld sng stp r
1854 yieldk _ r = foldStream st yld sng stp r
1824 in foldStream st yieldk single stop m
1855 in foldStream st yieldk single stop m
@@ -1883,10 +1914,10 @@ mfix f = mkStream $ \st yld sng stp ->
1883 where
1914 where
1884
1915
1885 -- fix the head element of the stream
1916 -- fix the head element of the stream
1886− xs = fix (f . headPartial)
1917+ xs = fix (f . headNonEmpty)
1887
1918
1888 -- now fix the tail recursively
1919 -- now fix the tail recursively
1889− ys = mfix (tailPartial . f)
1920+ ys = mfix (tailNonEmpty . f)
1890
1921
1891-------------------------------------------------------------------------------
1922-------------------------------------------------------------------------------
1892-- Conversions
1923-- Conversions
@@ -1917,6 +1948,15 @@ uncons m =
1917 yieldk a r = pure (Just (a, r))
1948 yieldk a r = pure (Just (a, r))
1918 in foldStream defState yieldk single stop m
1949 in foldStream defState yieldk single stop m
1919
1950
1951+-- Note that this is not a StreamK -> StreamK because then we cannot handle the
1952+-- empty stream case without making this a partial function.
1953+--
1954+-- See tailNonEmpty as well above.
1955+
1956+-- | Same as:
1957+--
1958+-- >>> tail = fmap (fmap snd) StreamK.uncons
1959+--
1920{-# INLINE tail #-}
1960{-# INLINE tail #-}
1921tail :: Applicative m => StreamK m a -> m (Maybe (StreamK m a))
1961tail :: Applicative m => StreamK m a -> m (Maybe (StreamK m a))
1922tail =
1962tail =
@@ -1925,9 +1965,14 @@ tail =
1925 yieldk _ r = pure $ Just r
1965 yieldk _ r = pure $ Just r
1926 in foldStream defState yieldk single stop
1966 in foldStream defState yieldk single stop
1927
1967
1928−-- | Extract all but the last element of the stream, if any.
1968+-- Note that this is not a StreamK -> StreamK because then we cannot handle the
1969+-- empty stream case without making this a partial function.
1929--
1970--
1930−-- Note: This will end up buffering the entire stream.
1971+-- XXX How do we implement unsnoc? Make StreamK a monad and return the
1972+-- remaining stream as a result value in the monad?
1973+
1974+-- | Extract all but the last element of the stream, if any. This will end up
1975+-- evaluating the last element as well to find out that it is last.
1931--
1976--
1932-- /Pre-release/
1977-- /Pre-release/
1933{-# INLINE init #-}
1978{-# INLINE init #-}
core/src/Streamly/Internal/Data/Unfold.hs+6−4
@@ -82,7 +82,7 @@ module Streamly.Internal.Data.Unfold
82 , dropWhileM
82 , dropWhileM
83
83
84 -- ** Cross product
84 -- ** Cross product
85− , joinInnerGeneric
85+ , innerJoin
86
86
87 -- ** Resource Management
87 -- ** Resource Management
88 -- | 'bracket' is the most general resource management operation, all other
88 -- | 'bracket' is the most general resource management operation, all other
@@ -677,10 +677,12 @@ dropWhileM f (Unfold step inject) = Unfold step' inject'
677dropWhile :: Monad m => (b -> Bool) -> Unfold m a b -> Unfold m a b
677dropWhile :: Monad m => (b -> Bool) -> Unfold m a b -> Unfold m a b
678dropWhile f = dropWhileM (return . f)
678dropWhile f = dropWhileM (return . f)
679
679
680+-- | Cross intersection of two unfolds. See
681+-- 'Streamly.Internal.Data.Stream.innerJoin' for more details.
680−{-# INLINE_NORMAL joinInnerGeneric #-}
682+{-# INLINE_NORMAL innerJoin #-}
681−joinInnerGeneric :: Monad m =>
683+innerJoin :: Monad m =>
682 (b -> c -> Bool) -> Unfold m a b -> Unfold m a c -> Unfold m a (b, c)
684 (b -> c -> Bool) -> Unfold m a b -> Unfold m a c -> Unfold m a (b, c)
683−joinInnerGeneric eq s1 s2 = filter (\(a, b) -> a `eq` b) $ cross s1 s2
685+innerJoin eq s1 s2 = filter (\(a, b) -> a `eq` b) $ cross s1 s2
684
686
685------------------------------------------------------------------------------
687------------------------------------------------------------------------------
686-- Exceptions
688-- Exceptions
core/src/Streamly/Internal/Unicode/Stream.hs+1−1
@@ -1057,7 +1057,7 @@ stripHead = Stream.dropWhile isSpace
1057-- >>> Stream.fold Fold.toList $ Unicode.lines Fold.toList (Stream.fromList "lines\nthis\nstring\n\n\n")
1057-- >>> Stream.fold Fold.toList $ Unicode.lines Fold.toList (Stream.fromList "lines\nthis\nstring\n\n\n")
1058-- ["lines","this","string","",""]
1058-- ["lines","this","string","",""]
1059--
1059--
1060−-- > lines = Stream.splitOnSuffix (== '\n')
1060+-- >>> lines f = Stream.foldMany (Fold.takeEndBy_ (== '\n') f)
1061--
1061--
1062-- /Pre-release/
1062-- /Pre-release/
1063{-# INLINE lines #-}
1063{-# INLINE lines #-}
docs/Developer/optimization-guidelines.md+12
@@ -191,6 +191,18 @@ the stages of a loop, it is often not useful to inline the whole loop
191itself, in fact we may have to occasionally use a `NOINLINE` so that the
191itself, in fact we may have to occasionally use a `NOINLINE` so that the
192compiler does not inline it.
192compiler does not inline it.
193
193
194+We need to be careful about NOINLINE on polymorphic functions. When
195+NOINLINE is used, the function cannot be specialized causing huge
196+performance degradation. To avoid that we can either make the
197+function monomorphic or sometimes using `NOINLINE [0]` can do the
198+trick. `NOINLINE [0]` blocks inlining in earlier phases but lets it
199+specialize in the last phase, inlining in the last phase is left to
200+the compiler. `INLINE [0]` may sound the same as `NOINLINE [0]`,
201+however, it behaves differently because we ask the compiler to INLINE
202+it compulsorily and it may not give us the desired results. The
203+Data.Fold.writeLastN benchmark is one such case where `NOINLINE [0]`
204+provides much better performance than `INLINE [0]`.
205+
194### NoSpecConstr
206### NoSpecConstr
195
207
196It is not always useful to specialize function calls for all constructors, some
208It is not always useful to specialize function calls for all constructors, some
src/Streamly/Internal/Data/Stream/IsStream/Top.hs+1−1
@@ -649,7 +649,7 @@ intersectBySorted :: (IsStream t, Monad m) =>
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
649 (a -> a -> Ordering) -> t m a -> t m a -> t m a
650intersectBySorted eq s1 =
650intersectBySorted eq s1 =
651 IsStream.fromStreamD
651 IsStream.fromStreamD
652− . StreamD.intersectBySorted eq (IsStream.toStreamD s1)
652+ . StreamD.sortedIntersectBy eq (IsStream.toStreamD s1)
653 . IsStream.toStreamD
653 . IsStream.toStreamD
654
654
655-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
655-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2