Sha256: e2af4d85479b318e9df8d857e6bc2c187d55cc377ebe42d35e78dd12682775b1
Contents?: true
Size: 1.41 KB
Versions: 10
Compression:
Stored size: 1.41 KB
Contents
module Riak
class MapReduce
# @api private
# Collects and normalizes results from MapReduce requests
class Results
# Creates a new result collector
# @param [MapReduce] mr the MapReduce query
def initialize(mr)
@keep_count = mr.query.select {|p| p.keep }.size
@hash = create_results_hash(mr.query)
end
# Adds a new result to the collector
# @param [Fixnum] phase the phase index
# @param [Array] data the phase result
def add(phase, result)
@hash[phase] += result
end
# Coalesces the query results
# @return [Array] the query results, coalesced according to the
# phase configuration
def report
if @keep_count > 1
@hash.to_a.sort.map {|(num, results)| results }
else
@hash[@hash.keys.first]
end
end
private
def create_results_hash(query)
# When the query is empty, only bucket/key pairs are returned,
# but implicitly in phase 0.
return { 0 => [] } if query.empty?
# Pre-populate the hash with empty results for kept phases.
# Additionally, the last phase is always implictly kept, even
# when keep is false.
query.inject({}) do |hash, phase|
if phase.keep || query[-1] == phase
hash[query.index(phase)] = []
end
hash
end
end
end
end
end
Version data entries
10 entries across 10 versions & 1 rubygems