datasketches-cpp
Loading...
Searching...
No Matches
quantiles_sketch_impl.hpp
1/*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 */
19
20#ifndef _QUANTILES_SKETCH_IMPL_HPP_
21#define _QUANTILES_SKETCH_IMPL_HPP_
22
23#include <cmath>
24#include <algorithm>
25#include <stdexcept>
26#include <iomanip>
27#include <sstream>
28#include <type_traits>
29
30#include "count_zeros.hpp"
31#include "conditional_forward.hpp"
32
33namespace datasketches {
34
35template<typename T, typename C, typename A>
36quantiles_sketch<T, C, A>::quantiles_sketch(uint16_t k, const C& comparator, const A& allocator):
37comparator_(comparator),
38allocator_(allocator),
39is_base_buffer_sorted_(true),
40k_(k),
41n_(0),
42bit_pattern_(0),
43base_buffer_(allocator_),
44levels_(allocator_),
45min_item_(),
46max_item_(),
47sorted_view_(nullptr)
48{
49 check_k(k_);
50 base_buffer_.reserve(2 * std::min(quantiles_constants::MIN_K, k));
51}
52
53template<typename T, typename C, typename A>
54quantiles_sketch<T, C, A>::quantiles_sketch(const quantiles_sketch& other):
55comparator_(other.comparator_),
56allocator_(other.allocator_),
57is_base_buffer_sorted_(other.is_base_buffer_sorted_),
58k_(other.k_),
59n_(other.n_),
60bit_pattern_(other.bit_pattern_),
61base_buffer_(other.base_buffer_),
62levels_(other.levels_),
63min_item_(other.min_item_),
64max_item_(other.max_item_),
65sorted_view_(nullptr)
66{
67 for (size_t i = 0; i < levels_.size(); ++i) {
68 if (levels_[i].capacity() != other.levels_[i].capacity()) {
69 levels_[i].reserve(other.levels_[i].capacity());
70 }
71 }
72}
73
74template<typename T, typename C, typename A>
75quantiles_sketch<T, C, A>::quantiles_sketch(quantiles_sketch&& other) noexcept:
76comparator_(other.comparator_),
77allocator_(other.allocator_),
78is_base_buffer_sorted_(other.is_base_buffer_sorted_),
79k_(other.k_),
80n_(other.n_),
81bit_pattern_(other.bit_pattern_),
82base_buffer_(std::move(other.base_buffer_)),
83levels_(std::move(other.levels_)),
84min_item_(std::move(other.min_item_)),
85max_item_(std::move(other.max_item_)),
86sorted_view_(nullptr)
87{}
88
89template<typename T, typename C, typename A>
90quantiles_sketch<T, C, A>& quantiles_sketch<T, C, A>::operator=(const quantiles_sketch& other) {
91 quantiles_sketch<T, C, A> copy(other);
92 std::swap(comparator_, copy.comparator_);
93 std::swap(allocator_, copy.allocator_);
94 std::swap(is_base_buffer_sorted_, copy.is_base_buffer_sorted_);
95 std::swap(k_, copy.k_);
96 std::swap(n_, copy.n_);
97 std::swap(bit_pattern_, copy.bit_pattern_);
98 std::swap(base_buffer_, copy.base_buffer_);
99 std::swap(levels_, copy.levels_);
100 std::swap(min_item_, copy.min_item_);
101 std::swap(max_item_, copy.max_item_);
102 reset_sorted_view();
103 return *this;
104}
105
106template<typename T, typename C, typename A>
107quantiles_sketch<T, C, A>& quantiles_sketch<T, C, A>::operator=(quantiles_sketch&& other) noexcept {
108 std::swap(comparator_, other.comparator_);
109 std::swap(allocator_, other.allocator_);
110 std::swap(is_base_buffer_sorted_, other.is_base_buffer_sorted_);
111 std::swap(k_, other.k_);
112 std::swap(n_, other.n_);
113 std::swap(bit_pattern_, other.bit_pattern_);
114 std::swap(base_buffer_, other.base_buffer_);
115 std::swap(levels_, other.levels_);
116 std::swap(min_item_, other.min_item_);
117 std::swap(max_item_, other.max_item_);
118 reset_sorted_view();
119 return *this;
120}
121
122template<typename T, typename C, typename A>
123quantiles_sketch<T, C, A>::quantiles_sketch(uint16_t k, uint64_t n, uint64_t bit_pattern,
124 Level&& base_buffer, VectorLevels&& levels,
125 optional<T>&& min_item, optional<T>&& max_item,
126 bool is_sorted, const C& comparator, const A& allocator):
127comparator_(comparator),
128allocator_(allocator),
129is_base_buffer_sorted_(is_sorted),
130k_(k),
131n_(n),
132bit_pattern_(bit_pattern),
133base_buffer_(std::move(base_buffer)),
134levels_(std::move(levels)),
135min_item_(std::move(min_item)),
136max_item_(std::move(max_item)),
137sorted_view_(nullptr)
138{
139 uint32_t item_count = static_cast<uint32_t>(base_buffer_.size());
140 for (Level& lvl : levels_) {
141 item_count += static_cast<uint32_t>(lvl.size());
142 }
143 if (item_count != compute_retained_items(k_, n_))
144 throw std::logic_error("Item count does not match value computed from k, n");
145}
146
147template<typename T, typename C, typename A>
148template<typename From, typename FC, typename FA>
149quantiles_sketch<T, C, A>::quantiles_sketch(const quantiles_sketch<From, FC, FA>& other,
150 const C& comparator, const A& allocator):
151comparator_(comparator),
152allocator_(allocator),
153is_base_buffer_sorted_(false),
154k_(other.get_k()),
155n_(other.get_n()),
156bit_pattern_(compute_bit_pattern(other.get_k(), other.get_n())),
157base_buffer_(allocator),
158levels_(allocator),
159min_item_(other.min_item_),
160max_item_(other.max_item_),
161sorted_view_(nullptr)
162{
163 static_assert(std::is_constructible<T, From>::value,
164 "Type converting constructor requires new type to be constructible from existing type");
165
166 base_buffer_.reserve(2 * std::min(quantiles_constants::MIN_K, k_));
167
168 if (!other.is_empty()) {
169 // reserve space in levels
170 const uint8_t num_levels = compute_levels_needed(k_, n_);
171 levels_.reserve(num_levels);
172 for (int i = 0; i < num_levels; ++i) {
173 Level level(allocator);
174 level.reserve(k_);
175 levels_.push_back(std::move(level));
176 }
177
178 // iterate through points, assigning to the correct level as needed
179 for (auto pair : other) {
180 const uint64_t wt = pair.second;
181 if (wt == 1) {
182 base_buffer_.push_back(T(pair.first));
183 // resize where needed as if adding points via update()
184 if (base_buffer_.size() + 1 > base_buffer_.capacity()) {
185 const size_t new_size = std::max(std::min(static_cast<size_t>(2 * k_), 2 * base_buffer_.size()), static_cast<size_t>(1));
186 base_buffer_.reserve(new_size);
187 }
188 }
189 else {
190 const uint8_t idx = count_trailing_zeros_in_u64(pair.second) - 1;
191 levels_[idx].push_back(T(pair.first));
192 }
193 }
194
195 // validate that ordering within each level is preserved
196 // base_buffer_ can be considered unsorted for this purpose
197 for (int i = 0; i < num_levels; ++i) {
198 if (!std::is_sorted(levels_[i].begin(), levels_[i].end(), comparator_)) {
199 throw std::logic_error("Copy construction across types produces invalid sorting");
200 }
201 }
202 }
203}
204
205
206template<typename T, typename C, typename A>
207quantiles_sketch<T, C, A>::~quantiles_sketch() {
208 reset_sorted_view();
209}
210
211template<typename T, typename C, typename A>
212template<typename FwdT>
214 if (!check_update_item(item)) { return; }
215 if (is_empty()) {
216 min_item_.emplace(item);
217 max_item_.emplace(item);
218 } else {
219 if (comparator_(item, *min_item_)) *min_item_ = item;
220 if (comparator_(*max_item_, item)) *max_item_ = item;
221 }
222
223 // if exceed capacity, grow until size 2k -- assumes eager processing
224 if (base_buffer_.size() + 1 > base_buffer_.capacity()) grow_base_buffer();
225
226 base_buffer_.push_back(std::forward<FwdT>(item));
227 ++n_;
228
229 if (base_buffer_.size() > 1) is_base_buffer_sorted_ = false;
230 if (base_buffer_.size() == 2 * k_) process_full_base_buffer();
231 reset_sorted_view();
232}
233
234template<typename T, typename C, typename A>
235template<typename FwdSk>
237 if (other.is_empty()) {
238 return; // nothing to do
239 } else if (!other.is_estimation_mode()) {
240 // other is exact, stream in regardless of k
241 for (auto item : other.base_buffer_) {
242 update(conditional_forward<FwdSk>(item));
243 }
244 reset_sorted_view();
245 return;
246 }
247
248 // other has data and is in estimation mode
249 if (is_estimation_mode()) {
250 if (k_ == other.get_k()) {
251 standard_merge(*this, std::forward<FwdSk>(other));
252 } else if (k_ > other.get_k()) {
253 quantiles_sketch sk_copy(std::forward<FwdSk>(other));
254 downsampling_merge(sk_copy, std::move(*this));
255 *this = std::move(sk_copy);
256 } else { // k_ < other.get_k()
257 downsampling_merge(*this, std::forward<FwdSk>(other));
258 }
259 } else {
260 // exact or empty
261 quantiles_sketch sk_copy(std::forward<FwdSk>(other));
262 if (k_ <= other.get_k()) {
263 if (!is_empty()) {
264 for (uint16_t i = 0; i < base_buffer_.size(); ++i) {
265 sk_copy.update(std::move(base_buffer_[i]));
266 }
267 }
268 } else { // k_ > other.get_k()
269 downsampling_merge(sk_copy, std::move(*this));
270 }
271 *this = std::move(sk_copy);
272 }
273 reset_sorted_view();
274}
275
276template<typename T, typename C, typename A>
277template<typename SerDe>
278void quantiles_sketch<T, C, A>::serialize(std::ostream& os, const SerDe& serde) const {
279 const uint8_t preamble_longs = is_empty() ? PREAMBLE_LONGS_SHORT : PREAMBLE_LONGS_FULL;
280 write(os, preamble_longs);
281 const uint8_t ser_ver = SERIAL_VERSION;
282 write(os, ser_ver);
283 const uint8_t family = FAMILY;
284 write(os, family);
285
286 // side-effect: sort base buffer since always compact
287 std::sort(const_cast<Level&>(base_buffer_).begin(), const_cast<Level&>(base_buffer_).end(), comparator_);
288 const_cast<quantiles_sketch*>(this)->is_base_buffer_sorted_ = true;
289
290 // empty, ordered, compact are valid flags
291 const uint8_t flags_byte(
292 (is_empty() ? 1 << flags::IS_EMPTY : 0)
293 | (1 << flags::IS_SORTED) // always sorted as side effect noted above
294 | (1 << flags::IS_COMPACT) // always compact -- could be optional for numeric types?
295 );
296 write(os, flags_byte);
297 write(os, k_);
298 const uint16_t unused = 0;
299 write(os, unused);
300
301 if (!is_empty()) {
302 write(os, n_);
303
304 // min and max
305 serde.serialize(os, &*min_item_, 1);
306 serde.serialize(os, &*max_item_, 1);
307
308 // base buffer items
309 serde.serialize(os, base_buffer_.data(), static_cast<unsigned>(base_buffer_.size()));
310
311 // levels, only when data is present
312 for (Level lvl : levels_) {
313 if (lvl.size() > 0)
314 serde.serialize(os, lvl.data(), static_cast<unsigned>(lvl.size()));
315 }
316 }
317}
318
319template<typename T, typename C, typename A>
320template<typename SerDe>
321auto quantiles_sketch<T, C, A>::serialize(unsigned header_size_bytes, const SerDe& serde) const -> vector_bytes {
322 const size_t size = get_serialized_size_bytes(serde) + header_size_bytes;
323 vector_bytes bytes(size, 0, allocator_);
324 uint8_t* ptr = bytes.data() + header_size_bytes;
325 const uint8_t* end_ptr = ptr + size;
326
327 const uint8_t preamble_longs = is_empty() ? PREAMBLE_LONGS_SHORT : PREAMBLE_LONGS_FULL;
328 ptr += copy_to_mem(preamble_longs, ptr);
329 const uint8_t ser_ver = SERIAL_VERSION;
330 ptr += copy_to_mem(ser_ver, ptr);
331 const uint8_t family = FAMILY;
332 ptr += copy_to_mem(family, ptr);
333
334 // side-effect: sort base buffer since always compact
335 std::sort(const_cast<Level&>(base_buffer_).begin(), const_cast<Level&>(base_buffer_).end(), comparator_);
336 const_cast<quantiles_sketch*>(this)->is_base_buffer_sorted_ = true;
337
338 // empty, ordered, compact are valid flags
339 const uint8_t flags_byte(
340 (is_empty() ? 1 << flags::IS_EMPTY : 0)
341 | (1 << flags::IS_SORTED) // always sorted as side effect noted above
342 | (1 << flags::IS_COMPACT) // always compact
343 );
344 ptr += copy_to_mem(flags_byte, ptr);
345 ptr += copy_to_mem(k_, ptr);
346 ptr += sizeof(uint16_t); // 2 unused bytes
347
348 if (!is_empty()) {
349
350 ptr += copy_to_mem(n_, ptr);
351
352 // min and max
353 ptr += serde.serialize(ptr, end_ptr - ptr, &*min_item_, 1);
354 ptr += serde.serialize(ptr, end_ptr - ptr, &*max_item_, 1);
355
356 // base buffer items
357 if (base_buffer_.size() > 0)
358 ptr += serde.serialize(ptr, end_ptr - ptr, base_buffer_.data(), static_cast<unsigned>(base_buffer_.size()));
359
360 // levels, only when data is present
361 for (Level lvl : levels_) {
362 if (lvl.size() > 0)
363 ptr += serde.serialize(ptr, end_ptr - ptr, lvl.data(), static_cast<unsigned>(lvl.size()));
364 }
365 }
366
367 return bytes;
368}
369
370template<typename T, typename C, typename A>
371template<typename SerDe>
372auto quantiles_sketch<T, C, A>::deserialize(std::istream &is, const SerDe& serde,
373 const C& comparator, const A &allocator) -> quantiles_sketch {
374 const auto preamble_longs = read<uint8_t>(is);
375 const auto serial_version = read<uint8_t>(is);
376 const auto family_id = read<uint8_t>(is);
377 const auto flags_byte = read<uint8_t>(is);
378 const auto k = read<uint16_t>(is);
379 read<uint16_t>(is); // unused
380
381 check_k(k);
382 check_serial_version(serial_version); // a little redundant with the header check
383 check_family_id(family_id);
384 check_header_validity(preamble_longs, flags_byte, serial_version);
385
386 if (!is.good()) throw std::runtime_error("error reading from std::istream");
387 const bool is_empty = (flags_byte & (1 << flags::IS_EMPTY)) > 0;
388 if (is_empty) {
389 return quantiles_sketch(k, comparator, allocator);
390 }
391
392 const auto items_seen = read<uint64_t>(is);
393
394 const bool is_compact = (serial_version == 2) | ((flags_byte & (1 << flags::IS_COMPACT)) > 0);
395 const bool is_sorted = (flags_byte & (1 << flags::IS_SORTED)) > 0;
396
397 optional<T> min_item;
398 optional<T> max_item;
399
400 // Space to deserialize min and max.
401 // serde::deserialize expects allocated but not initialized storage.
402 typename std::aligned_storage<sizeof(T), alignof(T)>::type tmp_storage;
403 T* tmp = reinterpret_cast<T*>(&tmp_storage);
404
405 serde.deserialize(is, tmp, 1);
406 // serde call did not throw, repackage and cleanup
407 min_item.emplace(std::move(*tmp));
408 tmp->~T();
409 serde.deserialize(is, tmp, 1);
410 // serde call did not throw, repackage and cleanup
411 max_item.emplace(std::move(*tmp));
412 tmp->~T();
413
414 if (serial_version == 1) {
415 read<uint64_t>(is); // no longer used
416 }
417
418 // allocate buffers as needed
419 const uint8_t levels_needed = compute_levels_needed(k, items_seen);
420 const uint64_t bit_pattern = compute_bit_pattern(k, items_seen);
421
422 // Java provides a compact storage layout for a sketch of primitive doubles. The C++ version
423 // does not currently operate sketches in compact mode, but will only serialize as compact
424 // to avoid complications around serialization of empty values for generic type T. We also need
425 // to be able to ingest either serialized format from Java.
426
427 // load base buffer
428 const uint32_t bb_items = compute_base_buffer_items(k, items_seen);
429 uint32_t items_to_read = (levels_needed == 0 || is_compact) ? bb_items : 2 * k;
430 Level base_buffer = deserialize_array(is, bb_items, 2 * k, serde, allocator);
431 if (items_to_read > bb_items) { // either equal or greater, never read fewer items
432 // read remaining items, but don't store them
433 deserialize_array(is, items_to_read - bb_items, items_to_read - bb_items, serde, allocator);
434 }
435
436 // populate vector of Levels directly
437 VectorLevels levels(allocator);
438 levels.reserve(levels_needed);
439 if (levels_needed > 0) {
440 uint64_t working_pattern = bit_pattern;
441 for (size_t i = 0; i < levels_needed; ++i, working_pattern >>= 1) {
442 if ((working_pattern & 0x01) == 1) {
443 Level level = deserialize_array(is, k, k, serde, allocator);
444 levels.push_back(std::move(level));
445 } else {
446 Level level(allocator);
447 level.reserve(k);
448 levels.push_back(std::move(level));
449 }
450 }
451 }
452
453 return quantiles_sketch(k, items_seen, bit_pattern,
454 std::move(base_buffer), std::move(levels), std::move(min_item), std::move(max_item), is_sorted,
455 comparator, allocator);
456}
457
458template<typename T, typename C, typename A>
459template<typename SerDe>
460auto quantiles_sketch<T, C, A>::deserialize_array(std::istream& is, uint32_t num_items, uint32_t capacity, const SerDe& serde, const A& allocator) -> Level {
461 A alloc(allocator);
462 std::unique_ptr<T, items_deleter> items(alloc.allocate(num_items), items_deleter(allocator, false, num_items));
463 serde.deserialize(is, items.get(), num_items);
464 // serde did not throw, enable destructors
465 items.get_deleter().set_destroy(true);
466 if (!is.good()) throw std::runtime_error("error reading from std::istream");
467
468 // successfully read, now put into a Level
469 Level level(allocator);
470 level.reserve(capacity);
471 level.insert(level.begin(),
472 std::make_move_iterator(items.get()),
473 std::make_move_iterator(items.get() + num_items));
474 return level;
475}
476
477template<typename T, typename C, typename A>
478template<typename SerDe>
479auto quantiles_sketch<T, C, A>::deserialize(const void* bytes, size_t size, const SerDe& serde,
480 const C& comparator, const A &allocator) -> quantiles_sketch {
481 ensure_minimum_memory(size, 8);
482 const char* ptr = static_cast<const char*>(bytes);
483 const char* end_ptr = static_cast<const char*>(bytes) + size;
484
485 uint8_t preamble_longs;
486 ptr += copy_from_mem(ptr, preamble_longs);
487 uint8_t serial_version;
488 ptr += copy_from_mem(ptr, serial_version);
489 uint8_t family_id;
490 ptr += copy_from_mem(ptr, family_id);
491 uint8_t flags_byte;
492 ptr += copy_from_mem(ptr, flags_byte);
493 uint16_t k;
494 ptr += copy_from_mem(ptr, k);
495 uint16_t unused;
496 ptr += copy_from_mem(ptr, unused);
497
498 check_k(k);
499 check_serial_version(serial_version); // a little redundant with the header check
500 check_family_id(family_id);
501 check_header_validity(preamble_longs, flags_byte, serial_version);
502
503 const bool is_empty = (flags_byte & (1 << flags::IS_EMPTY)) > 0;
504 if (is_empty) {
505 return quantiles_sketch(k, comparator, allocator);
506 }
507
508 ensure_minimum_memory(size, 16);
509 uint64_t items_seen;
510 ptr += copy_from_mem(ptr, items_seen);
511
512 const bool is_compact = (serial_version == 2) | ((flags_byte & (1 << flags::IS_COMPACT)) > 0);
513 const bool is_sorted = (flags_byte & (1 << flags::IS_SORTED)) > 0;
514
515 optional<T> min_item;
516 optional<T> max_item;
517
518 // Space to deserialize min and max.
519 // serde::deserialize expects allocated but not initialized storage.
520 typename std::aligned_storage<sizeof(T), alignof(T)>::type tmp_storage;
521 T* tmp = reinterpret_cast<T*>(&tmp_storage);
522
523 ptr += serde.deserialize(ptr, end_ptr - ptr, tmp, 1);
524 // serde call did not throw, repackage and cleanup
525 min_item.emplace(std::move(*tmp));
526 tmp->~T();
527 ptr += serde.deserialize(ptr, end_ptr - ptr, tmp, 1);
528 // serde call did not throw, repackage and cleanup
529 max_item.emplace(std::move(*tmp));
530 tmp->~T();
531
532 if (serial_version == 1) {
533 uint64_t unused_long;
534 ensure_minimum_memory(end_ptr - ptr, sizeof(unused_long));
535 ptr += copy_from_mem(ptr, unused_long); // no longer used
536 }
537
538 // allocate buffers as needed
539 const uint8_t levels_needed = compute_levels_needed(k, items_seen);
540 const uint64_t bit_pattern = compute_bit_pattern(k, items_seen);
541
542 // Java provides a compact storage layout for a sketch of primitive doubles. The C++ version
543 // does not currently operate sketches in compact mode, but will only serialize as compact
544 // to avoid complications around serialization of empty values for generic type T. We also need
545 // to be able to ingest either serialized format from Java.
546
547 // load base buffer
548 const uint32_t bb_items = compute_base_buffer_items(k, items_seen);
549 uint32_t items_to_read = (levels_needed == 0 || is_compact) ? bb_items : 2 * k;
550 auto base_buffer_pair = deserialize_array(ptr, end_ptr - ptr, bb_items, 2 * k, serde, allocator);
551 ptr += base_buffer_pair.second;
552 if (items_to_read > bb_items) { // either equal or greater, never read fewer items
553 // read remaining items, only use to advance the pointer
554 auto extras = deserialize_array(ptr, end_ptr - ptr, items_to_read - bb_items, items_to_read - bb_items, serde, allocator);
555 ptr += extras.second;
556 }
557
558 // populate vector of Levels directly
559 VectorLevels levels(allocator);
560 levels.reserve(levels_needed);
561 if (levels_needed > 0) {
562 uint64_t working_pattern = bit_pattern;
563 for (size_t i = 0; i < levels_needed; ++i, working_pattern >>= 1) {
564
565 if ((working_pattern & 0x01) == 1) {
566 auto pair = deserialize_array(ptr, end_ptr - ptr, k, k, serde, allocator);
567 ptr += pair.second;
568 levels.push_back(std::move(pair.first));
569 } else {
570 Level level(allocator);
571 level.reserve(k);
572 levels.push_back(std::move(level));
573 }
574 }
575 }
576
577 return quantiles_sketch(k, items_seen, bit_pattern,
578 std::move(base_buffer_pair.first), std::move(levels), std::move(min_item), std::move(max_item), is_sorted,
579 comparator, allocator);
580}
581
582template<typename T, typename C, typename A>
583template<typename SerDe>
584auto quantiles_sketch<T, C, A>::deserialize_array(const void* bytes, size_t size, uint32_t num_items, uint32_t capacity, const SerDe& serde, const A& allocator)
585 -> std::pair<Level, size_t> {
586 const char* ptr = static_cast<const char*>(bytes);
587 const char* end_ptr = static_cast<const char*>(bytes) + size;
588 A alloc(allocator);
589 std::unique_ptr<T, items_deleter> items(alloc.allocate(num_items), items_deleter(allocator, false, num_items));
590 ptr += serde.deserialize(ptr, end_ptr - ptr, items.get(), num_items);
591 // serde did not throw, enable destructors
592 items.get_deleter().set_destroy(true);
593
594 // successfully read, now put into a Level
595 Level level(allocator);
596 level.reserve(capacity);
597 level.insert(level.begin(),
598 std::make_move_iterator(items.get()),
599 std::make_move_iterator(items.get() + num_items));
600
601 return std::pair<Level, size_t>(std::move(level), ptr - static_cast<const char*>(bytes));
602}
603
604template<typename T, typename C, typename A>
605string<A> quantiles_sketch<T, C, A>::to_string(bool print_levels, bool print_items) const {
606 // Using a temporary stream for implementation here does not comply with AllocatorAwareContainer requirements.
607 // The stream does not support passing an allocator instance, and alternatives are complicated.
608 std::ostringstream os;
609 os << "### Quantiles Sketch summary:" << std::endl;
610 os << " K : " << k_ << std::endl;
611 os << " N : " << n_ << std::endl;
612 os << " Epsilon : " << std::setprecision(3) << get_normalized_rank_error(false) * 100 << "%" << std::endl;
613 os << " Epsilon PMF : " << get_normalized_rank_error(true) * 100 << "%" << std::endl;
614 os << " Empty : " << (is_empty() ? "true" : "false") << std::endl;
615 os << " Estimation mode: " << (is_estimation_mode() ? "true" : "false") << std::endl;
616 os << " Levels (w/o BB): " << levels_.size() << std::endl;
617 os << " Used Levels : " << count_valid_levels(bit_pattern_) << std::endl;
618 os << " Retained items : " << get_num_retained() << std::endl;
619 if (!is_empty()) {
620 os << " Min item : " << *min_item_ << std::endl;
621 os << " Max item : " << *max_item_ << std::endl;
622 }
623 os << "### End sketch summary" << std::endl;
624
625 if (print_levels) {
626 os << "### Quantiles Sketch levels:" << std::endl;
627 os << " index: items in use" << std::endl;
628 os << " BB: " << base_buffer_.size() << std::endl;
629 for (uint8_t i = 0; i < levels_.size(); i++) {
630 os << " " << static_cast<unsigned int>(i) << ": " << levels_[i].size() << std::endl;
631 }
632 os << "### End sketch levels" << std::endl;
633 }
634
635 if (print_items) {
636 os << "### Quantiles Sketch data:" << std::endl;
637 uint8_t level = 0;
638 os << " BB:" << std::endl;
639 for (const T& item : base_buffer_) {
640 os << " " << item << std::endl;
641 }
642 for (uint8_t i = 0; i < levels_.size(); ++i) {
643 os << " level " << static_cast<unsigned int>(level) << ":" << std::endl;
644 for (const T& item : levels_[i]) {
645 os << " " << item << std::endl;
646 }
647 }
648 os << "### End sketch data" << std::endl;
649 }
650 return string<A>(os.str().c_str(), allocator_);
651}
652
653template<typename T, typename C, typename A>
655 return k_;
656}
657
658template<typename T, typename C, typename A>
660 return n_;
661}
662
663template<typename T, typename C, typename A>
665 return n_ == 0;
666}
667
668template<typename T, typename C, typename A>
670 return bit_pattern_ != 0;
671}
672
673template<typename T, typename C, typename A>
675 return compute_retained_items(k_, n_);
676}
677
678template<typename T, typename C, typename A>
680 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
681 return *min_item_;
682}
683
684template<typename T, typename C, typename A>
686 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
687 return *max_item_;
688}
689
690template<typename T, typename C, typename A>
692 return comparator_;
693}
694
695template<typename T, typename C, typename A>
697 return allocator_;
698}
699
700// implementation for fixed-size arithmetic types (integral and floating point)
701template<typename T, typename C, typename A>
702template<typename SerDe, typename TT, typename std::enable_if<std::is_arithmetic<TT>::value, int>::type>
704 if (is_empty()) { return EMPTY_SIZE_BYTES; }
705 return DATA_START + ((get_num_retained() + 2) * sizeof(TT));
706}
707
708// implementation for all other types
709template<typename T, typename C, typename A>
710template<typename SerDe, typename TT, typename std::enable_if<!std::is_arithmetic<TT>::value, int>::type>
712 if (is_empty()) { return EMPTY_SIZE_BYTES; }
713 size_t size = DATA_START;
714 size += serde.size_of_item(*min_item_);
715 size += serde.size_of_item(*max_item_);
716 for (auto it: *this) size += serde.size_of_item(it.first);
717 return size;
718}
719
720template<typename T, typename C, typename A>
722 return get_normalized_rank_error(k_, is_pmf);
723}
724
725template<typename T, typename C, typename A>
727 return is_pmf
728 ? 1.854 / std::pow(k, 0.9657)
729 : 1.576 / std::pow(k, 0.9726);
730}
731
732template<typename T, typename C, typename A>
734 // allow side-effect of sorting the base buffer
735 if (!is_base_buffer_sorted_) {
736 std::sort(const_cast<Level&>(base_buffer_).begin(), const_cast<Level&>(base_buffer_).end(), comparator_);
737 const_cast<quantiles_sketch*>(this)->is_base_buffer_sorted_ = true;
738 }
739 quantiles_sorted_view<T, C, A> view(get_num_retained(), comparator_, allocator_);
740
741 uint64_t weight = 1;
742 view.add(base_buffer_.begin(), base_buffer_.end(), weight);
743 for (const auto& level: levels_) {
744 weight <<= 1;
745 if (level.empty()) { continue; }
746 view.add(level.begin(), level.end(), weight);
747 }
748
749 view.convert_to_cummulative();
750 return view;
751}
752
753template<typename T, typename C, typename A>
754auto quantiles_sketch<T, C, A>::get_quantile(double rank, bool inclusive) const -> quantile_return_type {
755 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
756 if ((rank < 0.0) || (rank > 1.0)) {
757 throw std::invalid_argument("Normalized rank cannot be less than 0 or greater than 1");
758 }
759 // possible side-effect: sorting base buffer
760 setup_sorted_view();
761 return sorted_view_->get_quantile(rank, inclusive);
762}
763
764template<typename T, typename C, typename A>
765double quantiles_sketch<T, C, A>::get_rank(const T& item, bool inclusive) const {
766 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
767 setup_sorted_view();
768 return sorted_view_->get_rank(item, inclusive);
769}
770
771template<typename T, typename C, typename A>
772auto quantiles_sketch<T, C, A>::get_PMF(const T* split_points, uint32_t size, bool inclusive) const -> vector_double {
773 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
774 setup_sorted_view();
775 return sorted_view_->get_PMF(split_points, size, inclusive);
776}
777
778template<typename T, typename C, typename A>
779auto quantiles_sketch<T, C, A>::get_CDF(const T* split_points, uint32_t size, bool inclusive) const -> vector_double {
780 if (is_empty()) throw std::runtime_error("operation is undefined for an empty sketch");
781 setup_sorted_view();
782 return sorted_view_->get_CDF(split_points, size, inclusive);
783}
784
785template<typename T, typename C, typename A>
786uint32_t quantiles_sketch<T, C, A>::compute_retained_items(uint16_t k, uint64_t n) {
787 const uint32_t bb_count = compute_base_buffer_items(k, n);
788 const uint64_t bit_pattern = compute_bit_pattern(k, n);
789 const uint32_t valid_levels = count_valid_levels(bit_pattern);
790 return bb_count + (k * valid_levels);
791}
792
793template<typename T, typename C, typename A>
794uint32_t quantiles_sketch<T, C, A>::compute_base_buffer_items(uint16_t k, uint64_t n) {
795 return n % (static_cast<uint64_t>(2) * k);
796}
797
798template<typename T, typename C, typename A>
799uint64_t quantiles_sketch<T, C, A>::compute_bit_pattern(uint16_t k, uint64_t n) {
800 return n / (static_cast<uint64_t>(2) * k);
801}
802
803template<typename T, typename C, typename A>
804uint32_t quantiles_sketch<T, C, A>::count_valid_levels(uint64_t bit_pattern) {
805 uint32_t count = 0;
806 for (; bit_pattern > 0; ++count) bit_pattern &= bit_pattern - 1;
807 return count;
808}
809
810template<typename T, typename C, typename A>
811uint8_t quantiles_sketch<T, C, A>::compute_levels_needed(uint16_t k, uint64_t n) {
812 return static_cast<uint8_t>(64U) - count_leading_zeros_in_u64(n / (2 * k));
813}
814
815template<typename T, typename C, typename A>
816void quantiles_sketch<T, C, A>::check_k(uint16_t k) {
817 if (k < quantiles_constants::MIN_K || k > quantiles_constants::MAX_K || (k & (k - 1)) != 0) {
818 throw std::invalid_argument("k must be a power of 2 that is >= "
819 + std::to_string(quantiles_constants::MIN_K) + " and <= "
820 + std::to_string(quantiles_constants::MAX_K) + ". Found: " + std::to_string(k));
821 }
822}
823
824template<typename T, typename C, typename A>
825void quantiles_sketch<T, C, A>::check_serial_version(uint8_t serial_version) {
826 if (serial_version == SERIAL_VERSION || serial_version == SERIAL_VERSION_1 || serial_version == SERIAL_VERSION_2)
827 return;
828 else
829 throw std::invalid_argument("Possible corruption. Unrecognized serialization version: " + std::to_string(serial_version));
830}
831
832template<typename T, typename C, typename A>
833void quantiles_sketch<T, C, A>::check_family_id(uint8_t family_id) {
834 if (family_id == FAMILY)
835 return;
836 else
837 throw std::invalid_argument("Possible corruption. Family id does not indicate quantiles sketch: " + std::to_string(family_id));
838}
839
840template<typename T, typename C, typename A>
841void quantiles_sketch<T, C, A>::check_header_validity(uint8_t preamble_longs, uint8_t flags_byte, uint8_t serial_version) {
842 const bool empty = (flags_byte & (1 << flags::IS_EMPTY)) > 0;
843 const bool compact = (flags_byte & (1 << flags::IS_COMPACT)) > 0;
844
845 const uint8_t sw = (compact ? 1 : 0) + (2 * (empty ? 1 : 0))
846 + (4 * (serial_version & 0xF)) + (32 * (preamble_longs & 0x3F));
847 bool valid = true;
848
849 switch (sw) { // exhaustive list and description of all valid cases
850 case 38 : break;
851 case 164 : break;
852 case 42 : break;
853 case 72 : break;
854 case 47 : break; // compact, empty, serVer = 3, preLongs = 1;
855 case 46 : break;
856 case 79 : break; // compact, empty, serVer = 3, preLongs = 2;
857 case 78 : break;
858 case 77 : break; // compact, !empty, serVer = 3, preLongs = 2;
859 case 76 : break;
860 default : //all other case values are invalid
861 valid = false;
862 }
863
864 if (!valid) {
865 std::ostringstream os;
866 os << "Possible sketch corruption. Inconsistent state: "
867 << "preamble_longs = " << preamble_longs
868 << ", empty = " << (empty ? "true" : "false")
869 << ", serialization_version = " << serial_version
870 << ", compact = " << (compact ? "true" : "false");
871 throw std::invalid_argument(os.str());
872 }
873}
874
875template <typename T, typename C, typename A>
876typename quantiles_sketch<T, C, A>::const_iterator quantiles_sketch<T, C, A>::begin() const {
877 return quantiles_sketch<T, C, A>::const_iterator(base_buffer_, levels_, k_, n_, false);
878}
879
880template <typename T, typename C, typename A>
881typename quantiles_sketch<T, C, A>::const_iterator quantiles_sketch<T, C, A>::end() const {
882 return quantiles_sketch<T, C, A>::const_iterator(base_buffer_, levels_, k_, n_, true);
883}
884
885template<typename T, typename C, typename A>
886void quantiles_sketch<T, C, A>::grow_base_buffer() {
887 const size_t new_size = std::max(std::min(static_cast<size_t>(2 * k_), 2 * base_buffer_.size()), static_cast<size_t>(1));
888 base_buffer_.reserve(new_size);
889}
890
891template<typename T, typename C, typename A>
892void quantiles_sketch<T, C, A>::process_full_base_buffer() {
893 // make sure there will be enough levels for the propagation
894 grow_levels_if_needed(); // note: n_ was already incremented by update() before this
895
896 std::sort(base_buffer_.begin(), base_buffer_.end(), comparator_);
897 in_place_propagate_carry(0,
898 levels_[0], // unused here, but 0 is guaranteed to exist
899 base_buffer_,
900 true, *this);
901 base_buffer_.clear();
902 is_base_buffer_sorted_ = true;
903 if (n_ / (2 * k_) != bit_pattern_) {
904 throw std::logic_error("Internal error: n / 2k (" + std::to_string(n_ / 2 * k_)
905 + " != bit_pattern " + std::to_string(bit_pattern_));
906 }
907}
908
909template<typename T, typename C, typename A>
910bool quantiles_sketch<T, C, A>::grow_levels_if_needed() {
911 const uint8_t levels_needed = compute_levels_needed(k_, n_);
912 if (levels_needed == 0)
913 return false; // don't need levels and might have small base buffer. Possible during merges.
914
915 // from here on, assume full size base buffer (2k) and at least one additional level
916 if (levels_needed <= levels_.size())
917 return false;
918
919 Level empty_level(allocator_);
920 empty_level.reserve(k_);
921 levels_.push_back(std::move(empty_level));
922 return true;
923}
924
925template<typename T, typename C, typename A>
926template<typename FwdV>
927void quantiles_sketch<T, C, A>::in_place_propagate_carry(uint8_t starting_level,
928 FwdV&& buf_size_k, Level& buf_size_2k,
929 bool apply_as_update,
930 quantiles_sketch& sketch) {
931 const uint64_t bit_pattern = sketch.bit_pattern_;
932 const int k = sketch.k_;
933
934 uint8_t ending_level = lowest_zero_bit_starting_at(bit_pattern, starting_level);
935
936 if (apply_as_update) {
937 // update version of computation
938 // its is okay for buf_size_k to be null in this case
939 zip_buffer(buf_size_2k, sketch.levels_[ending_level]);
940 } else {
941 // merge_into version of computation
942 for (uint16_t i = 0; i < k; ++i) {
943 sketch.levels_[ending_level].push_back(conditional_forward<FwdV>(buf_size_k[i]));
944 }
945 }
946
947 for (uint64_t lvl = starting_level; lvl < ending_level; lvl++) {
948 if ((bit_pattern & (static_cast<uint64_t>(1) << lvl)) == 0) {
949 throw std::logic_error("unexpected empty level in bit_pattern");
950 }
951 merge_two_size_k_buffers(
952 sketch.levels_[lvl],
953 sketch.levels_[ending_level],
954 buf_size_2k, sketch.get_comparator());
955 sketch.levels_[lvl].clear();
956 sketch.levels_[ending_level].clear();
957 zip_buffer(buf_size_2k, sketch.levels_[ending_level]);
958 } // end of loop over lower levels
959
960 // update bit pattern with binary-arithmetic ripple carry
961 sketch.bit_pattern_ = bit_pattern + (static_cast<uint64_t>(1) << starting_level);
962}
963
964template<typename T, typename C, typename A>
965void quantiles_sketch<T, C, A>::zip_buffer(Level& buf_in, Level& buf_out) {
966#ifdef QUANTILES_VALIDATION
967 static uint32_t next_offset = 0;
968 uint32_t rand_offset = next_offset;
969 next_offset = 1 - next_offset;
970#else
971 uint32_t rand_offset = random_utils::random_bit();
972#endif
973 if ((buf_in.size() != 2 * buf_out.capacity())
974 || (buf_out.size() > 0)) {
975 throw std::logic_error("zip_buffer requires buf_in.size() == "
976 "2*buf_out.capacity() and empty buf_out");
977 }
978
979 size_t k = buf_out.capacity();
980 for (uint32_t i = rand_offset, o = 0; o < k; i += 2, ++o) {
981 buf_out.push_back(std::move(buf_in[i]));
982 }
983 buf_in.clear();
984}
985
986template<typename T, typename C, typename A>
987template<typename FwdV>
988void quantiles_sketch<T, C, A>::zip_buffer_with_stride(FwdV&& buf_in, Level& buf_out, uint16_t stride) {
989 // Random offset in range [0, stride)
990 std::uniform_int_distribution<uint16_t> dist(0, stride - 1);
991 const uint16_t rand_offset = dist(random_utils::rand);
992
993 if ((buf_in.size() != stride * buf_out.capacity())
994 || (buf_out.size() > 0)) {
995 throw std::logic_error("zip_buffer_with_stride requires buf_in.size() == "
996 "stride*buf_out.capacity() and empty buf_out");
997 }
998
999 const size_t k = buf_out.capacity();
1000 for (uint16_t i = rand_offset, o = 0; o < k; i += stride, ++o) {
1001 buf_out.push_back(conditional_forward<FwdV>(buf_in[i]));
1002 }
1003 // do not clear input buffer
1004}
1005
1006template<typename T, typename C, typename A>
1007void quantiles_sketch<T, C, A>::merge_two_size_k_buffers(Level& src_1, Level& src_2,
1008 Level& dst, const C& comparator) {
1009 if (src_1.size() != src_2.size()
1010 || src_1.size() * 2 != dst.capacity()
1011 || dst.size() != 0) {
1012 throw std::logic_error("Input invariants violated in merge_two_size_k_buffers()");
1013 }
1014
1015 auto end1 = src_1.end(), end2 = src_2.end();
1016 auto it1 = src_1.begin(), it2 = src_2.begin();
1017
1018 // TODO: probably actually doing copies given Level&?
1019 while (it1 != end1 && it2 != end2) {
1020 if (comparator(*it1, *it2)) {
1021 dst.push_back(std::move(*it1++));
1022 } else {
1023 dst.push_back(std::move(*it2++));
1024 }
1025 }
1026
1027 if (it1 != end1) {
1028 dst.insert(dst.end(), it1, end1);
1029 } else {
1030 if (it2 == end2) { throw std::logic_error("it2 unexpectedly already at end of range"); }
1031 dst.insert(dst.end(), it2, end2);
1032 }
1033}
1034
1035template<typename T, typename C, typename A>
1036template<typename FwdSk>
1037void quantiles_sketch<T, C, A>::standard_merge(quantiles_sketch& tgt, FwdSk&& src) {
1038 if (src.get_k() != tgt.get_k()) {
1039 throw std::invalid_argument("src.get_k() != tgt.get_k()");
1040 }
1041 if (src.is_empty()) {
1042 return;
1043 }
1044
1045 uint64_t new_n = src.get_n() + tgt.get_n();
1046
1047 // move items from src's base buffer
1048 for (uint16_t i = 0; i < src.base_buffer_.size(); ++i) {
1049 tgt.update(conditional_forward<FwdSk>(src.base_buffer_[i]));
1050 }
1051
1052 // check (after moving raw items) if we need to extend levels array
1053 uint8_t levels_needed = compute_levels_needed(tgt.get_k(), new_n);
1054 if (levels_needed > tgt.levels_.size()) {
1055 tgt.levels_.reserve(levels_needed);
1056 while (tgt.levels_.size() < levels_needed) {
1057 Level empty_level(tgt.allocator_);
1058 empty_level.reserve(tgt.get_k());
1059 tgt.levels_.push_back(std::move(empty_level));
1060 }
1061 }
1062
1063 Level scratch_buf(tgt.allocator_);
1064 scratch_buf.reserve(2 * tgt.get_k());
1065
1066 uint64_t src_pattern = src.bit_pattern_;
1067 for (uint8_t src_lvl = 0; src_pattern != 0; ++src_lvl, src_pattern >>= 1) {
1068 if ((src_pattern & 1) > 0) {
1069 scratch_buf.clear();
1070
1071 // propagate-carry
1072 in_place_propagate_carry(src_lvl,
1073 src.levels_[src_lvl], scratch_buf,
1074 false, tgt);
1075 // update n_ at the end
1076 }
1077 }
1078 tgt.n_ = new_n;
1079 if ((tgt.get_n() / (2 * tgt.get_k())) != tgt.bit_pattern_) {
1080 throw std::logic_error("Failed internal consistency check after standard_merge()");
1081 }
1082
1083 // update min and max items
1084 // can't just check is_empty() since min and max might not have been set if
1085 // there were no base buffer items added via update()
1086 if (!tgt.min_item_) {
1087 tgt.min_item_.emplace(conditional_forward<FwdSk>(*src.min_item_));
1088 } else {
1089 if (tgt.comparator_(*src.min_item_, *tgt.min_item_))
1090 *tgt.min_item_ = conditional_forward<FwdSk>(*src.min_item_);
1091 }
1092 if (!tgt.max_item_) {
1093 tgt.max_item_.emplace(conditional_forward<FwdSk>(*src.max_item_));
1094 } else {
1095 if (tgt.comparator_(*tgt.max_item_, *src.max_item_))
1096 *tgt.max_item_ = conditional_forward<FwdSk>(*src.max_item_);
1097 }
1098}
1099
1100template<typename T, typename C, typename A>
1101template<typename FwdSk>
1102void quantiles_sketch<T, C, A>::downsampling_merge(quantiles_sketch& tgt, FwdSk&& src) {
1103 if (src.get_k() % tgt.get_k() != 0) {
1104 throw std::invalid_argument("src.get_k() is not a multiple of tgt.get_k()");
1105 }
1106 if (src.is_empty()) {
1107 return;
1108 }
1109
1110 const uint16_t downsample_factor = src.get_k() / tgt.get_k();
1111 const uint8_t lg_sample_factor = count_trailing_zeros_in_u32(downsample_factor);
1112
1113 const uint64_t new_n = src.get_n() + tgt.get_n();
1114
1115 // move items from src's base buffer
1116 for (uint16_t i = 0; i < src.base_buffer_.size(); ++i) {
1117 tgt.update(conditional_forward<FwdSk>(src.base_buffer_[i]));
1118 }
1119
1120 // check (after moving raw items) if we need to extend levels array
1121 const uint8_t levels_needed = compute_levels_needed(tgt.get_k(), new_n);
1122 if (levels_needed > tgt.levels_.size()) {
1123 tgt.levels_.reserve(levels_needed);
1124 while (tgt.levels_.size() < levels_needed) {
1125 Level empty_level(tgt.allocator_);
1126 empty_level.reserve(tgt.get_k());
1127 tgt.levels_.push_back(std::move(empty_level));
1128 }
1129 }
1130
1131 Level down_buf(tgt.allocator_);
1132 down_buf.reserve(tgt.get_k());
1133
1134 Level scratch_buf(tgt.allocator_);
1135 scratch_buf.reserve(2 * tgt.get_k());
1136
1137 uint64_t src_pattern = src.bit_pattern_;
1138 for (uint8_t src_lvl = 0; src_pattern != 0; ++src_lvl, src_pattern >>= 1) {
1139 if ((src_pattern & 1) > 0) {
1140 down_buf.clear();
1141 scratch_buf.clear();
1142
1143 // zip with stride, leaving input buffer intact
1144 zip_buffer_with_stride(src.levels_[src_lvl], down_buf, downsample_factor);
1145
1146 // propagate-carry
1147 in_place_propagate_carry(src_lvl + lg_sample_factor,
1148 down_buf, scratch_buf,
1149 false, tgt);
1150 // update n_ at the end
1151 }
1152 }
1153 tgt.n_ = new_n;
1154 if ((tgt.get_n() / (2 * tgt.get_k())) != tgt.bit_pattern_) {
1155 throw std::logic_error("Failed internal consistency check after downsampling_merge()");
1156 }
1157
1158 // update min and max items
1159 // can't just check is_empty() since min and max might not have been set if
1160 // there were no base buffer items added via update()
1161 if (!tgt.min_item_) {
1162 tgt.min_item_.emplace(conditional_forward<FwdSk>(*src.min_item_));
1163 } else {
1164 if (tgt.comparator_(*src.min_item_, *tgt.min_item_))
1165 *tgt.min_item_ = conditional_forward<FwdSk>(*src.min_item_);
1166 }
1167 if (!tgt.max_item_) {
1168 tgt.max_item_.emplace(conditional_forward<FwdSk>(*src.max_item_));
1169 } else {
1170 if (tgt.comparator_(*tgt.max_item_, *src.max_item_))
1171 *tgt.max_item_ = conditional_forward<FwdSk>(*src.max_item_);
1172 }
1173}
1174
1175template<typename T, typename C, typename A>
1176uint8_t quantiles_sketch<T, C, A>::lowest_zero_bit_starting_at(uint64_t bits, uint8_t starting_bit) {
1177 uint8_t pos = starting_bit & 0X3F;
1178 uint64_t my_bits = bits >> pos;
1179
1180 while ((my_bits & static_cast<uint64_t>(1)) != 0) {
1181 my_bits >>= 1;
1182 pos++;
1183 }
1184 return pos;
1185}
1186
1187template<typename T, typename C, typename A>
1188class quantiles_sketch<T, C, A>::items_deleter {
1189 public:
1190 items_deleter(const A& allocator, bool destroy, size_t num): allocator_(allocator), destroy_(destroy), num_(num) {}
1191 void operator() (T* ptr) {
1192 if (ptr != nullptr) {
1193 if (destroy_) {
1194 for (size_t i = 0; i < num_; ++i) {
1195 ptr[i].~T();
1196 }
1197 }
1198 allocator_.deallocate(ptr, num_);
1199 }
1200 }
1201 void set_destroy(bool destroy) { destroy_ = destroy; }
1202 private:
1203 A allocator_;
1204 bool destroy_;
1205 size_t num_;
1206};
1207
1208template<typename T, typename C, typename A>
1209void quantiles_sketch<T, C, A>::setup_sorted_view() const {
1210 if (sorted_view_ == nullptr) {
1211 using AllocSortedView = typename std::allocator_traits<A>::template rebind_alloc<quantiles_sorted_view<T, C, A>>;
1212 sorted_view_ = new (AllocSortedView(allocator_).allocate(1)) quantiles_sorted_view<T, C, A>(get_sorted_view());
1213 }
1214}
1215
1216template<typename T, typename C, typename A>
1217void quantiles_sketch<T, C, A>::reset_sorted_view() {
1218 if (sorted_view_ != nullptr) {
1219 sorted_view_->~quantiles_sorted_view();
1220 using AllocSortedView = typename std::allocator_traits<A>::template rebind_alloc<quantiles_sorted_view<T, C, A>>;
1221 AllocSortedView(allocator_).deallocate(sorted_view_, 1);
1222 sorted_view_ = nullptr;
1223 }
1224}
1225
1226// quantiles_sketch::const_iterator implementation
1227
1228template<typename T, typename C, typename A>
1229quantiles_sketch<T, C, A>::const_iterator::const_iterator(const Level& base_buffer,
1230 const std::vector<Level, AllocLevel>& levels,
1231 uint16_t k,
1232 uint64_t n,
1233 bool is_end):
1234base_buffer_(base_buffer),
1235levels_(levels),
1236level_(-1),
1237index_(0),
1238bb_count_(compute_base_buffer_items(k, n)),
1239bit_pattern_(compute_bit_pattern(k, n)),
1240weight_(1),
1241k_(k)
1242{
1243 if (is_end) {
1244 // if exact mode: index_ = n is end
1245 // if sampling, level_ = max_level + 1 and index_ = 0 is end
1246 if (bit_pattern_ == 0) // only a valid check for exact mode in constructor
1247 index_ = static_cast<uint32_t>(n);
1248 else
1249 level_ = static_cast<int>(levels_.size());
1250 } else { // find first non-empty item
1251 if (bb_count_ == 0 && bit_pattern_ > 0) {
1252 level_ = 0;
1253 weight_ = 2;
1254 while ((bit_pattern_ & 0x01) == 0) {
1255 weight_ *= 2;
1256 ++level_;
1257 bit_pattern_ >>= 1;
1258 }
1259 }
1260 }
1261}
1262
1263template<typename T, typename C, typename A>
1264typename quantiles_sketch<T, C, A>::const_iterator& quantiles_sketch<T, C, A>::const_iterator::operator++() {
1265 ++index_;
1266
1267 if ((level_ == -1 && index_ == base_buffer_.size() && levels_.size() > 0) || (level_ >= 0 && index_ == k_)) { // go to the next non-empty level
1268 index_ = 0;
1269 do {
1270 ++level_;
1271 if (level_ > 0) bit_pattern_ = bit_pattern_ >> 1;
1272 if (bit_pattern_ == 0) return *this;
1273 weight_ *= 2;
1274 } while ((bit_pattern_ & static_cast<uint64_t>(1)) == 0);
1275 }
1276 return *this;
1277}
1278
1279template<typename T, typename C, typename A>
1280typename quantiles_sketch<T, C, A>::const_iterator& quantiles_sketch<T, C, A>::const_iterator::operator++(int) {
1281 const_iterator tmp(*this);
1282 operator++();
1283 return tmp;
1284}
1285
1286template<typename T, typename C, typename A>
1287bool quantiles_sketch<T, C, A>::const_iterator::operator==(const const_iterator& other) const {
1288 return level_ == other.level_ && index_ == other.index_;
1289}
1290
1291template<typename T, typename C, typename A>
1292bool quantiles_sketch<T, C, A>::const_iterator::operator!=(const const_iterator& other) const {
1293 return !operator==(other);
1294}
1295
1296template<typename T, typename C, typename A>
1297auto quantiles_sketch<T, C, A>::const_iterator::operator*() const -> reference {
1298 return value_type(level_ == -1 ? base_buffer_[index_] : levels_[level_][index_], weight_);
1299}
1300
1301template<typename T, typename C, typename A>
1302auto quantiles_sketch<T, C, A>::const_iterator::operator->() const -> pointer {
1303 return **this;
1304}
1305
1306} /* namespace datasketches */
1307
1308#endif // _QUANTILES_SKETCH_IMPL_HPP_
This is a stochastic streaming sketch that enables near-real time analysis of the approximate distrib...
Definition quantiles_sketch.hpp:158
Comparator get_comparator() const
Returns an instance of the comparator for this sketch.
Definition quantiles_sketch_impl.hpp:691
quantiles_sorted_view< T, Comparator, Allocator > get_sorted_view() const
Gets the sorted view of this sketch.
Definition quantiles_sketch_impl.hpp:733
quantiles_sketch & operator=(const quantiles_sketch &other)
Copy assignment.
Definition quantiles_sketch_impl.hpp:90
vector_double get_CDF(const T *split_points, uint32_t size, bool inclusive=true) const
Returns an approximation to the Cumulative Distribution Function (CDF), which is the cumulative analo...
Definition quantiles_sketch_impl.hpp:779
uint32_t get_num_retained() const
Returns the number of retained items (samples) in the sketch.
Definition quantiles_sketch_impl.hpp:674
double get_normalized_rank_error(bool is_pmf) const
Gets the normalized rank error for this sketch.
Definition quantiles_sketch_impl.hpp:721
vector_double get_PMF(const T *split_points, uint32_t size, bool inclusive=true) const
Returns an approximation to the Probability Mass Function (PMF) of the input stream given a set of sp...
Definition quantiles_sketch_impl.hpp:772
void merge(FwdSk &&other)
Merges another sketch into this one.
Definition quantiles_sketch_impl.hpp:236
void update(FwdT &&item)
Updates this sketch with the given data item.
Definition quantiles_sketch_impl.hpp:213
void serialize(std::ostream &os, const SerDe &sd=SerDe()) const
This method serializes the sketch into a given stream in a binary form.
Definition quantiles_sketch_impl.hpp:278
string< Allocator > to_string(bool print_levels=false, bool print_items=false) const
Prints a summary of the sketch.
Definition quantiles_sketch_impl.hpp:605
uint16_t get_k() const
Returns configured parameter k.
Definition quantiles_sketch_impl.hpp:654
bool is_empty() const
Returns true if this sketch is empty.
Definition quantiles_sketch_impl.hpp:664
const T & get_max_item() const
Returns the max item of the stream.
Definition quantiles_sketch_impl.hpp:685
double get_rank(const T &item, bool inclusive=true) const
Returns an approximation to the normalized rank of the given item from 0 to 1, inclusive.
Definition quantiles_sketch_impl.hpp:765
allocator_type get_allocator() const
Returns the allocator for this sketch.
Definition quantiles_sketch_impl.hpp:696
quantile_return_type get_quantile(double rank, bool inclusive=true) const
Returns an approximation to the data item associated with the given rank of a hypothetical sorted ver...
Definition quantiles_sketch_impl.hpp:754
const_iterator begin() const
Iterator pointing to the first item in the sketch.
Definition quantiles_sketch_impl.hpp:876
size_t get_serialized_size_bytes(const SerDe &sd=SerDe()) const
Computes size needed to serialize the current state of the sketch.
Definition quantiles_sketch_impl.hpp:703
bool is_estimation_mode() const
Returns true if this sketch is in estimation mode.
Definition quantiles_sketch_impl.hpp:669
const T & get_min_item() const
Returns the min item of the stream.
Definition quantiles_sketch_impl.hpp:679
const_iterator end() const
Iterator pointing to the past-the-end item in the sketch.
Definition quantiles_sketch_impl.hpp:881
uint64_t get_n() const
Returns the length of the input stream.
Definition quantiles_sketch_impl.hpp:659
Sorted view for quantiles sketches (REQ, KLL and Quantiles).
Definition quantiles_sorted_view.hpp:38
const uint16_t MIN_K
minimum value of parameter K
Definition quantiles_sketch.hpp:39
DataSketches namespace.
Definition binomial_bounds.hpp:38
Interface for serializing and deserializing items.
Definition serde.hpp:35
void serialize(std::ostream &os, const T *items, unsigned num) const
Stream serialization.