anchor.diff:      let ofArray (source : 'T []) : CloudStream<'T> =
           { new CloudStream<'T> with
               member self.DegreeOfParallelism = None
  -            member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) (combiner : 'R -> 'R -> 'R) =
  +            member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) (combiner : 'R [] -> Local<'R>) =
                   cloud {
                       let! collector = collectorf 
                       let! workerCount = 
  
positive.code:                             partitions 
                              |> Array.map (fun (s, e) -> createTask [| for i in s..(e - 1L) do yield source.[int i] |] collectorf) 
                              |> Cloud.Parallel
                          return Array.reduce combiner results
                      else
                          return! projection collector.Result
                  } }
  
negative.code:     let ofCloudVector (source : CloudVector<'T>) : CloudStream<'T> =
          { new CloudStream<'T> with
              member self.DegreeOfParallelism = None
              member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) (combiner : 'R [] -> Local<'R>) =
                  cloud {
                      let useCache = source.IsCachingSupported
                      let partitions = getPartitionIndices source
  