3

目标是泛化Streaming.merge函数，

merge :: (Monad m, Ord a) => Stream (Of a) m r -> Stream (Of a) m s -> Stream (Of a) m (r, s)

到任意数量的源流。策略是使用Data.Heap.Heap按Stream (Of a) m r排序的a。即bigMerge会有签名

bigMerge :: (Monad m, Ord a) => [Stream (Of a) m r] -> Stream (Of a) m [r]

（该列表也可以替换为Heap。）

我所拥有的是一种不太正确的相当邪恶的混合物。它是这样的：

为了完整性，首先关闭导入：

import qualified Data.Heap as H
import Data.Heap (Heap)
import Data.List (sortBy)
import Data.Function (on)
import Streaming
import qualified Streaming.Prelude as S
import Streaming.Internal (Stream(..))  -- shouldn't!

为了使用Heap类的元素，Ord需要：

data Elt a m r = Elt Int (Maybe a) (Stream(Of a) m r)

引入extraInt以携带输入列表中流的索引，以便[r]可以使用正确顺序的元素构建返回的流。携带流的Maybe a当前值。

Eq和Ord实例是：

instance Eq a => Eq (Elt a m r) where
(Elt i ma _) == (Elt i' ma' _) = 
    if i == i' then error "Internal error: Index clash in ==" 
    else ma == ma'

instance Ord a => Ord (Elt a m r) where
(Elt i ma s) <= (Elt i' ma' s') | i==i' = error "Internal error: Index clash in <="
                                | otherwise = cmp (i, ma, s) (i', ma', s')
    where 
    cmp _                      (_, Nothing, Return _) = True
    cmp (_, Nothing, Return _) _                      = False
    cmp (i, Just a, _)         (i', Just a', _)       = if a == a' then i <= i' else a <= a'
    cmp (i, _, _)              (i', _, _)             = i <= i'

基本上，任何东西都是<=a Return，所有其他情况都使用 thea和/或ito 排序Elts。（errors用于调试目的。）

一些辅助函数Elt从.StreamHeapStream

eltFromStream :: (Monad m, Ord a) => Int -> Stream (Of a) m r -> m (Elt a m r)
eltFromStream i (Return r) = return $ Elt i Nothing (Return r)
eltFromStream i (Effect m) = do
    stream' <- m
    return $ Elt i Nothing stream'
eltFromStream i (Step (a :> rest)) = return $ Elt i (Just a) rest

heapFromStreams :: (Monad m, Ord a) => [Stream (Of a) m r] -> m (Heap (Elt a m r))
heapFromStreams strs = H.fromList <$> (sequence $ fmap (uncurry eltFromStream) (zip [0..] strs))

核心是loop功能

loop :: (Monad m, Ord a) => Heap (Elt a m r) -> m (Heap (Elt a m r))
loop h = do
let (Elt i ma s, h') = unsafeUncons h
elt <- case s of
    Return r         -> return $ Elt i Nothing (Return r)
    Effect m         -> Elt i Nothing <$> m
    Step (a :> rest) -> return $ Elt i (Just a) rest
return $ H.insert elt h'

带着厚脸皮unsafeUncons的存在

unsafeUncons :: Heap a -> (a, Heap a)
unsafeUncons h = case H.uncons h of
Nothing -> error "Internal error"
Just x -> x

该函数loop用于heapMerge将HeapStream

heapMerge :: (Monad m, Ord a) => Heap (Elt a m r) -> Stream (Of a) m [r]
heapMerge h = case (ma,s) of
    (Nothing, Return _) -> Return $ getRs h
    (_, Effect m) -> error "TODO"
    (Just a, _)  -> do
        h' <- lift $ loop h
        Step (a :> heapMerge h')
    where
        Elt i ma s = H.minimum h

getRs只是将Return值组合成一个列表

getRs :: (Monad m, Ord a) => Heap (Elt a m r) -> [r]
getRs h = snd <$> sortBy (compare `on` fst) (map f (H.toUnsortedList h))
where
    f :: Monad m => Elt a m r -> (Int, r)
    f (Elt i _ (Return r)) = (i,r)
    f _ = error "Internal error: Call getR only after stream has finished!"

然后，最后，

bigMerge :: (Monad m, Ord a) => [Stream (Of a) m r] -> Stream (Of a) m [r]
bigMerge streams = 
if null streams then Return [] 
else do
    h <- lift $ heapFromStreams streams
    heapMerge h

这是令人费解的，Effect没有得到正确处理，它依赖于Return, Step,Effect而不是inspectand next。它确实会在简单的输入上产生正确的结果，例如

s1 = S.each [2,4,5::Int]
s2 = S.each [1,2,4,5::Int]
s3 = S.each [3::Int]
S.print $ merge [s1,s2,s3]

我确信有一种方法可以正确且更惯用地执行此操作。一方面，Maybe ainElt可能是多余的，我可以直接做(Stream (Of a) m r)一个实例Ord，如果Effects 只是模式匹配，没有执行，那么这应该没问题。但Stream (Of (Heap (Stream (Of a) m r, Int))) (Heap (Int,r))看起来很奇怪。“带索引的流”IStream a m r = IStream Int ((Heap (Stream (Of a) m r) deriving Functor是 中的函子r，所以，如果有适当的==and <=，我会看Stream (IStream a m) m (Heap (Int, r))?

图书馆的这个功能方面对streaming我来说仍然有点困惑，所以任何帮助都将不胜感激。

4

1 回答 1

3

的签名bigMerge看起来非常像sequenceAfrom的签名Data.Traversable：

sequenceA :: Applicative f => [f r] -> f [r]

问题当然是我们不能使用 s 的标准Applicative实例Stream，因为它是连接而不是合并。但我们可以尝试通过新类型创建自己的实例：

{-# LANGUAGE DeriveFunctor #-}
import Streaming
import qualified Streaming.Prelude as S

newtype MergeStream a m r = 
    MergeStream { getMergeStream :: Stream (Of a) m r } deriving Functor

-- BEWARE! Only valid for ORDERED streams!
instance (Monad m, Ord a) => Applicative (MergeStream a m) where
    pure x = MergeStream (pure x)
    MergeStream f <*> MergeStream x = MergeStream (uncurry ($) <$> S.merge f x)

现在，使用s1,s2和s3您的示例以及标准Traversable函数：

ghci> S.toList_ $ getMergeStream . traverse MergeStream $ [s1,s2,s3]
[1,2,2,3,4,4,5,5]

这似乎有效。也就是说，出于效率原因，您尝试bigMerge使用Stream内部和堆实现可能仍然值得。

于 2019-12-04T07:21:14.247 回答