@@ -0,0 +1,630 @@
1+-- |
2+-- Module : Streamly.Internal.Data.Stream.Top
3+-- Copyright : (c) 2020 Composewell Technologies
4+-- License : BSD-3-Clause
5+-- Maintainer : streamly@composewell.com
6+-- Stability : experimental
7+-- Portability : GHC
8+--
9+-- Top level module that can depend on all other lower level Stream modules.
10+
11+module Streamly.Internal.Data.Stream.Top
12+ (
13+ -- * Transformation
14+ -- ** Sampling
15+ -- | Value agnostic filtering.
16+ sampleFromThen
17+
18+ -- ** Reordering
19+ , sortBy
20+
21+ -- * Nesting
22+ -- ** Set like operations
23+ -- | These are not exactly set operations because streams are not
24+ -- necessarily sets, they may have duplicated elements.
25+ , intersectBy
26+ , intersectBySorted
27+ , differenceBy
28+ , mergeDifferenceBy
29+ , unionBy
30+ , mergeUnionBy
31+
32+ -- ** Join operations
33+ , crossJoin
34+ , joinInner
35+ , joinInnerMap
36+ , joinInnerMerge
37+ , joinLeft
38+ , mergeLeftJoin
39+ , joinLeftMap
40+ , joinOuter
41+ , mergeOuterJoin
42+ , joinOuterMap
43+ )
44+where
45+
46+#include "inline.hs"
47+
48+import Control.Monad.Catch (MonadCatch)
49+import Control.Monad.IO.Class (MonadIO(..))
50+import Control.Monad.Trans.Class (lift)
51+import Control.Monad.Trans.State.Strict (get, put)
52+import Data.Function ((&))
53+import Data.IORef (newIORef, readIORef, modifyIORef')
54+import Data.Maybe (isJust)
55+import Streamly.Internal.Data.Stream.Common ()
56+import Streamly.Internal.Data.Stream.Type (Stream, fromStreamD, toStreamD)
57+
58+import qualified Data.List as List
59+import qualified Data.Map.Strict as Map
60+import qualified Streamly.Internal.Data.Array as Array
61+import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
62+import qualified Streamly.Internal.Data.Fold as Fold
63+import qualified Streamly.Internal.Data.Parser as Parser
64+import qualified Streamly.Internal.Data.Stream.Lift as Stream
65+import qualified Streamly.Internal.Data.Stream.Eliminate as Stream
66+import qualified Streamly.Internal.Data.Stream.Generate as Stream
67+import qualified Streamly.Internal.Data.Stream.Expand as Stream
68+import qualified Streamly.Internal.Data.Stream.Reduce as Stream
69+import qualified Streamly.Internal.Data.Stream.Transform as Stream
70+import qualified Streamly.Internal.Data.Stream.StreamD as StreamD
71+import qualified Streamly.Internal.Data.Unfold as Unfold
72+
73+import Prelude hiding (filter, zipWith, concatMap, concat)
74+
75+-- $setup
76+-- >>> :m
77+-- >>> import Prelude hiding (filter, zipWith, concatMap, concat)
78+-- >>> import qualified Streamly.Prelude as Stream
79+-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
80+
81+------------------------------------------------------------------------------
82+-- Sampling
83+------------------------------------------------------------------------------
84+
85+-- XXX We can implement this using addition instead of "mod" to make it more
86+-- efficient.
87+--
88+-- | @sampleFromthen offset stride@ samples the element at @offset@ index and
89+-- then every element at strides of @stride@.
90+--
91+-- >>> Stream.toList $ Stream.sampleFromThen 2 3 $ Stream.enumerateFromTo 0 10
92+-- [2,5,8]
93+--
94+-- /Pre-release/
95+--
96+{-# INLINE sampleFromThen #-}
97+sampleFromThen :: Monad m => Int -> Int -> Stream m a -> Stream m a
98+sampleFromThen offset stride =
99+ Stream.with Stream.indexed Stream.filter
100+ (\(i, _) -> i >= offset && (i - offset) `mod` stride == 0)
101+
102+------------------------------------------------------------------------------
103+-- Reordering
104+------------------------------------------------------------------------------
105+--
106+-- We could possibly choose different algorithms depending on whether the
107+-- input stream is almost sorted (ascending/descending) or random. We could
108+-- serialize the stream to an array and use quicksort.
109+--
110+-- | Sort the input stream using a supplied comparison function.
111+--
112+-- /O(n) space/
113+--
114+-- Note: this is not the fastest possible implementation as of now.
115+--
116+-- /Pre-release/
117+--
118+{-# INLINE sortBy #-}
119+sortBy :: MonadCatch m => (a -> a -> Ordering) -> Stream m a -> Stream m a
120+-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure
121+sortBy cmp =
122+ let p =
123+ Parser.groupByRollingEither
124+ (\x -> (< GT) . cmp x)
125+ Fold.toStreamRev
126+ Fold.toStream
127+ in Stream.concatPairsWith (Stream.mergeBy cmp) id
128+ . Stream.parseMany (fmap (either id id) p)
129+
130+------------------------------------------------------------------------------
131+-- SQL Joins
132+------------------------------------------------------------------------------
133+--
134+-- Some references:
135+-- * https://en.wikipedia.org/wiki/Relational_algebra
136+-- * https://en.wikipedia.org/wiki/Join_(SQL)
137+
138+-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only
139+-- constraint, the best would be to use an Array with linear search. If the
140+-- second stream is sorted we can also use a binary search, using Ord
141+-- constraint or an ordering function.
142+--
143+-- For Storables we can cache the second stream into an unboxed array for
144+-- possibly faster access/compact representation?
145+--
146+-- If we do not want to keep the stream in memory but always read it from the
147+-- source (disk/network) every time we iterate through it then we can do that
148+-- too by reading the stream every time, the stream must have immutable state
149+-- in that case and the user is responsible for the behavior if the stream
150+-- source changes during iterations. We can also use an Unfold instead of
151+-- stream. We probably need a way to distinguish streams that can be read
152+-- mutliple times without any interference (e.g. unfolding a stream using an
153+-- immutable handle would work i.e. using pread/pwrite instead of maintianing
154+-- an offset in the handle).
155+
156+-- XXX We can do this concurrently.
157+--
158+-- | This is the same as 'Streamly.Internal.Data.Unfold.outerProduct' but less
159+-- efficient.
160+--
161+-- The second stream is evaluated multiple times. If the second stream is
162+-- consume-once stream then it can be cached in an 'Data.Array.Array' before
163+-- calling this function. Caching may also improve performance if the stream is
164+-- expensive to evaluate.
165+--
166+-- Time: O(m x n)
167+--
168+-- /Pre-release/
169+{-# INLINE crossJoin #-}
170+crossJoin :: Monad m => Stream m a -> Stream m b -> Stream m (a, b)
171+crossJoin s1 s2 = do
172+ -- XXX use concatMap instead?
173+ a <- s1
174+ b <- s2
175+ return (a, b)
176+
177+-- XXX We can do this concurrently.
178+-- XXX If the second stream is sorted and passed as an Array we could use
179+-- binary search if we have an Ord instance or Ordering returning function. The
180+-- time complexity would then become (m x log n).
181+--
182+-- | For all elements in @Stream m a@, for all elements in @Stream m b@ if @a@ and @b@
183+-- are equal by the given equality pedicate then return the tuple (a, b).
184+--
185+-- The second stream is evaluated multiple times. If the stream is a
186+-- consume-once stream then the caller should cache it (e.g. in a
187+-- 'Data.Array.Array') before calling this function. Caching may also improve
188+-- performance if the stream is expensive to evaluate.
189+--
190+-- For space efficiency use the smaller stream as the second stream.
191+--
192+-- You should almost always use joinInnerMap instead of joinInner. joinInnerMap
193+-- is an order of magnitude faster. joinInner may be used when the second
194+-- stream is generated from a seed, therefore, need not be stored in memory and
195+-- the amount of memory it takes is a concern.
196+--
197+-- Space: O(n) assuming the second stream is cached in memory.
198+--
199+-- Time: O(m x n)
200+--
201+-- /Pre-release/
202+{-# INLINE joinInner #-}
203+joinInner :: Monad m =>
204+ (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, b)
205+joinInner eq s1 s2 = do
206+ -- ConcatMap works faster than bind
207+ Stream.concatMap (\a ->
208+ Stream.concatMap (\b ->
209+ if a `eq` b
210+ then Stream.fromPure (a, b)
211+ else Stream.nil
212+ ) s2
213+ ) s1
214+
215+-- XXX Generate error if a duplicate insertion is attempted?
216+toMap :: (Monad m, Ord k) => Stream m (k, v) -> m (Map.Map k v)
217+toMap =
218+ let f = Fold.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
219+ in Stream.fold f
220+
221+-- If the second stream is too big it can be partitioned based on hashes and
222+-- then we can process one parition at a time.
223+--
224+-- XXX An IntMap may be faster when the keys are Int.
225+-- XXX Use hashmap instead of map?
226+--
227+-- | Like 'joinInner' but uses a 'Map' for efficiency.
228+--
229+-- If the input streams have duplicate keys, the behavior is undefined.
230+--
231+-- For space efficiency use the smaller stream as the second stream.
232+--
233+-- Space: O(n)
234+--
235+-- Time: O(m + n)
236+--
237+-- /Pre-release/
238+{-# INLINE joinInnerMap #-}
239+joinInnerMap :: (Monad m, Ord k) =>
240+ Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, b)
241+joinInnerMap s1 s2 =
242+ Stream.concatM $ do
243+ km <- toMap s2
244+ pure $ Stream.mapMaybe (joinAB km) s1
245+
246+ where
247+
248+ joinAB kvm (k, a) =
249+ case k `Map.lookup` kvm of
250+ Just b -> Just (k, a, b)
251+ Nothing -> Nothing
252+
253+-- | Like 'joinInner' but works only on sorted streams.
254+--
255+-- Space: O(1)
256+--
257+-- Time: O(m + n)
258+--
259+-- /Unimplemented/
260+{-# INLINE joinInnerMerge #-}
261+joinInnerMerge ::
262+ (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, b)
263+joinInnerMerge = undefined
264+
265+-- XXX We can do this concurrently.
266+-- XXX If the second stream is sorted and passed as an Array or a seek capable
267+-- stream then we could use binary search if we have an Ord instance or
268+-- Ordering returning function. The time complexity would then become (m x log
269+-- n).
270+--
271+-- | For all elements in @Stream m a@, for all elements in @Stream m b@ if @a@ and @b@
272+-- are equal then return the tuple @(a, Just b)@. If @a@ is not present in @t
273+-- m b@ then return @(a, Nothing)@.
274+--
275+-- The second stream is evaluated multiple times. If the stream is a
276+-- consume-once stream then the caller should cache it in an 'Data.Array.Array'
277+-- before calling this function. Caching may also improve performance if the
278+-- stream is expensive to evaluate.
279+--
280+-- @
281+-- rightJoin = flip joinLeft
282+-- @
283+--
284+-- Space: O(n) assuming the second stream is cached in memory.
285+--
286+-- Time: O(m x n)
287+--
288+-- /Unimplemented/
289+{-# INLINE joinLeft #-}
290+joinLeft :: Monad m =>
291+ (a -> b -> Bool) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
292+joinLeft eq s1 s2 = Stream.evalStateT (return False) $ do
293+ a <- Stream.liftInner s1
294+ -- XXX should we use StreamD monad here?
295+ -- XXX Is there a better way to perform some action at the end of a loop
296+ -- iteration?
297+ lift $ put False
298+ let final = do
299+ r <- lift get
300+ if r
301+ then Stream.nil
302+ else Stream.fromPure Nothing
303+ b <- fmap Just (Stream.liftInner s2) <> final
304+ case b of
305+ Just b1 ->
306+ if a `eq` b1
307+ then do
308+ lift $ put True
309+ return (a, Just b1)
310+ else Stream.nil
311+ Nothing -> return (a, Nothing)
312+
313+-- | Like 'joinLeft' but uses a hashmap for efficiency.
314+--
315+-- Space: O(n)
316+--
317+-- Time: O(m + n)
318+--
319+-- /Pre-release/
320+{-# INLINE joinLeftMap #-}
321+joinLeftMap :: (Ord k, Monad m) =>
322+ Stream m (k, a) -> Stream m (k, b) -> Stream m (k, a, Maybe b)
323+joinLeftMap s1 s2 =
324+ Stream.concatM $ do
325+ km <- toMap s2
326+ return $ fmap (joinAB km) s1
327+
328+ where
329+
330+ joinAB km (k, a) =
331+ case k `Map.lookup` km of
332+ Just b -> (k, a, Just b)
333+ Nothing -> (k, a, Nothing)
334+
335+-- | Like 'joinLeft' but works only on sorted streams.
336+--
337+-- Space: O(1)
338+--
339+-- Time: O(m + n)
340+--
341+-- /Unimplemented/
342+{-# INLINE mergeLeftJoin #-}
343+mergeLeftJoin :: -- Monad m =>
344+ (a -> b -> Ordering) -> Stream m a -> Stream m b -> Stream m (a, Maybe b)
345+mergeLeftJoin _eq _s1 _s2 = undefined
346+
347+-- XXX We can do this concurrently.
348+--
349+-- | For all elements in @t m a@, for all elements in @t m b@ if @a@ and @b@
350+-- are equal by the given equality pedicate then return the tuple (Just a, Just
351+-- b). If @a@ is not found in @t m b@ then return (a, Nothing), return
352+-- (Nothing, b) for vice-versa.
353+--
354+-- For space efficiency use the smaller stream as the second stream.
355+--
356+-- Space: O(n)
357+--
358+-- Time: O(m x n)
359+--
360+-- /Pre-release/
361+{-# INLINE joinOuter #-}
362+joinOuter :: MonadIO m =>
363+ (a -> b -> Bool)
364+ -> Stream m a
365+ -> Stream m b
366+ -> Stream m (Maybe a, Maybe b)
367+joinOuter eq s1 s =
368+ Stream.concatM $ do
369+ inputArr <- Array.fromStream s
370+ let len = length inputArr
371+ foundArr <-
372+ Stream.fold
373+ (MA.writeN len)
374+ (Stream.unfold Unfold.fromList (Prelude.replicate len False))
375+ return $ go inputArr foundArr <> leftOver inputArr foundArr
376+
377+ where
378+
379+ leftOver inputArr foundArr =
380+ let stream1 = Array.toStream inputArr
381+ stream2 = Stream.unfold MA.read foundArr
382+ in Stream.filter
383+ isJust
384+ ( Stream.zipWith (\x y ->
385+ if y
386+ then Nothing
387+ else Just (Nothing, Just x)
388+ ) stream1 stream2
389+ ) & Stream.catMaybes
390+
391+ go inputArr foundArr = Stream.evalStateT (return False) $ do
392+ a <- Stream.liftInner s1
393+ -- XXX should we use StreamD monad here?
394+ -- XXX Is there a better way to perform some action at the end of a loop
395+ -- iteration?
396+ lift $ put False
397+ let final = do
398+ r <- lift get
399+ if r
400+ then Stream.nil
401+ else Stream.fromPure Nothing
402+ (i, b) <-
403+ let stream = Array.toStream inputArr
404+ in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
405+
406+ case b of
407+ Just b1 ->
408+ if a `eq` b1
409+ then do
410+ lift $ put True
411+ MA.putIndex i True foundArr
412+ return (Just a, Just b1)
413+ else Stream.nil
414+ Nothing -> return (Just a, Nothing)
415+
416+-- Put the b's that have been paired, in another hash or mutate the hash to set
417+-- a flag. At the end go through @Stream m b@ and find those that are not in that
418+-- hash to return (Nothing, b).
419+--
420+-- | Like 'joinOuter' but uses a 'Map' for efficiency.
421+--
422+-- Space: O(m + n)
423+--
424+-- Time: O(m + n)
425+--
426+-- /Pre-release/
427+{-# INLINE joinOuterMap #-}
428+joinOuterMap ::
429+ (Ord k, MonadIO m) =>
430+ Stream m (k, a) -> Stream m (k, b) -> Stream m (k, Maybe a, Maybe b)
431+joinOuterMap s1 s2 =
432+ Stream.concatM $ do
433+ km1 <- kvFold s1
434+ km2 <- kvFold s2
435+
436+ -- XXX Not sure if toList/fromList would fuse optimally. We may have to
437+ -- create a fused Map.toStream function.
438+ let res1 = fmap (joinAB km2)
439+ $ Stream.unfold Unfold.fromList $ Map.toList km1
440+ where
441+ joinAB km (k, a) =
442+ case k `Map.lookup` km of
443+ Just b -> (k, Just a, Just b)
444+ Nothing -> (k, Just a, Nothing)
445+
446+ -- XXX We can take advantage of the lookups in the first pass above to
447+ -- reduce the number of lookups in this pass. If we keep mutable cells
448+ -- in the second Map, we can flag it in the first pass and not do any
449+ -- lookup in the second pass if it is flagged.
450+ let res2 = Stream.mapMaybe (joinAB km1)
451+ $ Stream.unfold Unfold.fromList $ Map.toList km2
452+ where
453+ joinAB km (k, b) =
454+ case k `Map.lookup` km of
455+ Just _ -> Nothing
456+ Nothing -> Just (k, Nothing, Just b)
457+
458+ return $ Stream.append res1 res2
459+
460+ where
461+
462+ -- XXX Generate error if a duplicate insertion is attempted?
463+ kvFold =
464+ let f = Fold.foldl' (\kv (k, b) -> Map.insert k b kv) Map.empty
465+ in Stream.fold f
466+
467+-- | Like 'joinOuter' but works only on sorted streams.
468+--
469+-- Space: O(1)
470+--
471+-- Time: O(m + n)
472+--
473+-- /Unimplemented/
474+{-# INLINE mergeOuterJoin #-}
475+mergeOuterJoin :: -- Monad m =>
476+ (a -> b -> Ordering)
477+ -> Stream m a
478+ -> Stream m b
479+ -> Stream m (Maybe a, Maybe b)
480+mergeOuterJoin _eq _s1 _s2 = undefined
481+
482+------------------------------------------------------------------------------
483+-- Set operations (special joins)
484+------------------------------------------------------------------------------
485+--
486+-- TODO: OrdSet/IntSet/hashmap based versions of these. With Eq only constraint
487+-- the best would be to use an Array with linear search. If the second stream
488+-- is sorted we can also use a binary search, using Ord constraint.
489+
490+-- | 'intersectBy' is essentially a filtering operation that retains only those
491+-- elements in the first stream that are present in the second stream.
492+--
493+-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
494+-- [1,2,2]
495+--
496+-- >>> Stream.toList $ Stream.intersectBy (==) (Stream.fromList [2,1,1,3]) (Stream.fromList [1,2,2,4])
497+-- [2,1,1]
498+--
499+-- 'intersectBy' is similar to but not the same as 'joinInner':
500+--
501+-- >>> Stream.toList $ fmap fst $ Stream.joinInner (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [2,1,1,3])
502+-- [1,1,2,2]
503+--
504+-- Space: O(n) where @n@ is the number of elements in the second stream.
505+--
506+-- Time: O(m x n) where @m@ is the number of elements in the first stream and
507+-- @n@ is the number of elements in the second stream.
508+--
509+-- /Pre-release/
510+{-# INLINE intersectBy #-}
511+intersectBy :: Monad m =>
512+ (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
513+intersectBy eq s1 s2 =
514+ Stream.concatM
515+ $ do
516+ -- This may work well when s2 is small
517+ xs <- Stream.fold Fold.toListRev $ Stream.uniqBy eq s2
518+ return $ Stream.filter (\x -> List.any (eq x) xs) s1
519+
520+-- | Like 'intersectBy' but works only on streams sorted in ascending order.
521+--
522+-- Space: O(1)
523+--
524+-- Time: O(m+n)
525+--
526+-- /Pre-release/
527+{-# INLINE intersectBySorted #-}
528+intersectBySorted :: (Monad m) =>
529+ (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
530+intersectBySorted eq s1 =
531+ fromStreamD
532+ . StreamD.intersectBySorted eq (toStreamD s1)
533+ . toStreamD
534+
535+-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2
536+
537+-- | Delete first occurrences of those elements from the first stream that are
538+-- present in the second stream. If an element occurs multiple times in the
539+-- second stream as many occurrences of it are deleted from the first stream.
540+--
541+-- >>> Stream.toList $ Stream.differenceBy (==) (Stream.fromList [1,2,2]) (Stream.fromList [1,2,3])
542+-- [2]
543+--
544+-- The following laws hold:
545+--
546+-- @
547+-- (s1 `serial` s2) `differenceBy eq` s1 === s2
548+-- (s1 `wSerial` s2) `differenceBy eq` s1 === s2
549+-- @
550+--
551+-- Same as the list 'Data.List.//' operation.
552+--
553+-- Space: O(m) where @m@ is the number of elements in the first stream.
554+--
555+-- Time: O(m x n) where @m@ is the number of elements in the first stream and
556+-- @n@ is the number of elements in the second stream.
557+--
558+-- /Pre-release/
559+{-# INLINE differenceBy #-}
560+differenceBy :: (Monad m) =>
561+ (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
562+differenceBy eq s1 s2 =
563+ Stream.concatM
564+ $ do
565+ -- This may work well if s1 is small
566+ -- If s1 is big we can go through s1, deleting elements from s2 and
567+ -- not emitting an element if it was successfully deleted from s2.
568+ -- we will need a deleteBy that can return whether the element was
569+ -- deleted or not.
570+ xs <- Stream.fold Fold.toList s1
571+ let f = Fold.foldl' (flip (List.deleteBy eq)) xs
572+ fmap (Stream.unfold Unfold.fromList) $ Stream.fold f s2
573+
574+-- | Like 'differenceBy' but works only on sorted streams.
575+--
576+-- Space: O(1)
577+--
578+-- /Unimplemented/
579+{-# INLINE mergeDifferenceBy #-}
580+mergeDifferenceBy :: -- (Monad m) =>
581+ (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
582+mergeDifferenceBy _eq _s1 _s2 = undefined
583+
584+-- | This is essentially an append operation that appends all the extra
585+-- occurrences of elements from the second stream that are not already present
586+-- in the first stream.
587+--
588+-- >>> Stream.toList $ Stream.unionBy (==) (Stream.fromList [1,2,2,4]) (Stream.fromList [1,1,2,3])
589+-- [1,2,2,4,3]
590+--
591+-- Equivalent to the following except that @s1@ is evaluated only once:
592+--
593+-- @
594+-- unionBy eq s1 s2 = s1 \`serial` (s2 `differenceBy eq` s1)
595+-- @
596+--
597+-- Similar to 'joinOuter' but not the same.
598+--
599+-- Space: O(n)
600+--
601+-- Time: O(m x n)
602+--
603+-- /Pre-release/
604+{-# INLINE unionBy #-}
605+unionBy :: MonadIO m =>
606+ (a -> a -> Bool) -> Stream m a -> Stream m a -> Stream m a
607+unionBy eq s1 s2 =
608+ Stream.concatM
609+ $ do
610+ xs <- Stream.fold Fold.toList s2
611+ -- XXX we can use postscanlMAfter' instead of IORef
612+ ref <- liftIO $ newIORef $! List.nubBy eq xs
613+ let f x = do
614+ liftIO $ modifyIORef' ref (List.deleteBy eq x)
615+ return x
616+ s3 = Stream.concatM
617+ $ do
618+ xs1 <- liftIO $ readIORef ref
619+ return $ Stream.unfold Unfold.fromList xs1
620+ return $ Stream.mapM f s1 <> s3
621+
622+-- | Like 'unionBy' but works only on sorted streams.
623+--
624+-- Space: O(1)
625+--
626+-- /Unimplemented/
627+{-# INLINE mergeUnionBy #-}
628+mergeUnionBy :: -- (Monad m) =>
629+ (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a
630+mergeUnionBy _eq _s1 _s2 = undefined