merge with other tdigests
| 187 | |
| 188 | // merge with other tdigests |
| 189 | void Merge(const std::vector<const TDigestImpl*>& tdigest_impls) { |
| 190 | // current and end iterator |
| 191 | using CentroidIter = std::vector<Centroid>::const_iterator; |
| 192 | using CentroidIterPair = std::pair<CentroidIter, CentroidIter>; |
| 193 | // use a min-heap to find next minimal centroid from all tdigests |
| 194 | auto centroid_gt = [](const CentroidIterPair& lhs, const CentroidIterPair& rhs) { |
| 195 | return lhs.first->mean > rhs.first->mean; |
| 196 | }; |
| 197 | using CentroidQueue = |
| 198 | std::priority_queue<CentroidIterPair, std::vector<CentroidIterPair>, |
| 199 | decltype(centroid_gt)>; |
| 200 | |
| 201 | // trivial dynamic memory allocated at runtime |
| 202 | std::vector<CentroidIterPair> queue_buffer; |
| 203 | queue_buffer.reserve(tdigest_impls.size() + 1); |
| 204 | CentroidQueue queue(std::move(centroid_gt), std::move(queue_buffer)); |
| 205 | |
| 206 | const auto& this_tdigest = tdigests_[current_]; |
| 207 | if (this_tdigest.size() > 0) { |
| 208 | queue.emplace(this_tdigest.cbegin(), this_tdigest.cend()); |
| 209 | } |
| 210 | for (const TDigestImpl* td : tdigest_impls) { |
| 211 | const auto& other_tdigest = td->tdigests_[td->current_]; |
| 212 | if (other_tdigest.size() > 0) { |
| 213 | queue.emplace(other_tdigest.cbegin(), other_tdigest.cend()); |
| 214 | total_weight_ += td->total_weight_; |
| 215 | min_ = std::min(min_, td->min_); |
| 216 | max_ = std::max(max_, td->max_); |
| 217 | } |
| 218 | } |
| 219 | |
| 220 | merger_.Reset(total_weight_, &tdigests_[1 - current_]); |
| 221 | CentroidIter current_iter, end_iter; |
| 222 | // do k-way merge till one buffer left |
| 223 | while (queue.size() > 1) { |
| 224 | std::tie(current_iter, end_iter) = queue.top(); |
| 225 | merger_.Add(*current_iter); |
| 226 | queue.pop(); |
| 227 | if (++current_iter != end_iter) { |
| 228 | queue.emplace(current_iter, end_iter); |
| 229 | } |
| 230 | } |
| 231 | // merge last buffer |
| 232 | if (!queue.empty()) { |
| 233 | std::tie(current_iter, end_iter) = queue.top(); |
| 234 | while (current_iter != end_iter) { |
| 235 | merger_.Add(*current_iter++); |
| 236 | } |
| 237 | } |
| 238 | merger_.Reset(0, nullptr); |
| 239 | |
| 240 | current_ = 1 - current_; |
| 241 | } |
| 242 | |
| 243 | // merge input data with current tdigest |
| 244 | void MergeInput(std::vector<double>& input) { |