20#ifndef _QUANTILES_SKETCH_IMPL_HPP_
21#define _QUANTILES_SKETCH_IMPL_HPP_
30#include "count_zeros.hpp"
31#include "conditional_forward.hpp"
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),
39is_base_buffer_sorted_(true),
43base_buffer_(allocator_),
50 base_buffer_.reserve(2 * std::min(quantiles_constants::MIN_K, k));
53template<
typename T,
typename C,
typename A>
55comparator_(other.comparator_),
56allocator_(other.allocator_),
57is_base_buffer_sorted_(other.is_base_buffer_sorted_),
60bit_pattern_(other.bit_pattern_),
61base_buffer_(other.base_buffer_),
62levels_(other.levels_),
63min_item_(other.min_item_),
64max_item_(other.max_item_),
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());
74template<
typename T,
typename C,
typename A>
76comparator_(other.comparator_),
77allocator_(other.allocator_),
78is_base_buffer_sorted_(other.is_base_buffer_sorted_),
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_)),
89template<
typename T,
typename C,
typename A>
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_);
106template<
typename T,
typename C,
typename A>
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_);
122template<
typename T,
typename C,
typename A>
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),
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)),
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());
143 if (item_count != compute_retained_items(k_, n_))
144 throw std::logic_error(
"Item count does not match value computed from k, n");
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),
156bit_pattern_(compute_bit_pattern(other.get_k(), other.get_n())),
157base_buffer_(allocator),
159min_item_(other.min_item_),
160max_item_(other.max_item_),
163 static_assert(std::is_constructible<T, From>::value,
164 "Type converting constructor requires new type to be constructible from existing type");
168 if (!other.is_empty()) {
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);
175 levels_.push_back(std::move(level));
179 for (
auto pair : other) {
180 const uint64_t wt = pair.second;
182 base_buffer_.push_back(T(pair.first));
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);
190 const uint8_t idx = count_trailing_zeros_in_u64(pair.second) - 1;
191 levels_[idx].push_back(T(pair.first));
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");
206template<
typename T,
typename C,
typename A>
207quantiles_sketch<T, C, A>::~quantiles_sketch() {
211template<
typename T,
typename C,
typename A>
212template<
typename FwdT>
214 if (!check_update_item(item)) {
return; }
216 min_item_.emplace(item);
217 max_item_.emplace(item);
219 if (comparator_(item, *min_item_)) *min_item_ = item;
220 if (comparator_(*max_item_, item)) *max_item_ = item;
224 if (base_buffer_.size() + 1 > base_buffer_.capacity()) grow_base_buffer();
226 base_buffer_.push_back(std::forward<FwdT>(item));
229 if (base_buffer_.size() > 1) is_base_buffer_sorted_ =
false;
230 if (base_buffer_.size() == 2 * k_) process_full_base_buffer();
234template<
typename T,
typename C,
typename A>
235template<
typename FwdSk>
237 if (other.is_empty()) {
239 }
else if (!other.is_estimation_mode()) {
241 for (
auto item : other.base_buffer_) {
242 update(conditional_forward<FwdSk>(item));
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()) {
254 downsampling_merge(sk_copy, std::move(*
this));
255 *
this = std::move(sk_copy);
257 downsampling_merge(*
this, std::forward<FwdSk>(other));
262 if (k_ <= other.get_k()) {
264 for (uint16_t i = 0; i < base_buffer_.size(); ++i) {
265 sk_copy.
update(std::move(base_buffer_[i]));
269 downsampling_merge(sk_copy, std::move(*
this));
271 *
this = std::move(sk_copy);
276template<
typename T,
typename C,
typename A>
277template<
typename SerDe>
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;
283 const uint8_t family = FAMILY;
287 std::sort(
const_cast<Level&
>(base_buffer_).begin(),
const_cast<Level&
>(base_buffer_).end(), comparator_);
291 const uint8_t flags_byte(
292 (is_empty() ? 1 << flags::IS_EMPTY : 0)
293 | (1 << flags::IS_SORTED)
294 | (1 << flags::IS_COMPACT)
296 write(os, flags_byte);
298 const uint16_t unused = 0;
309 serde.
serialize(os, base_buffer_.data(),
static_cast<unsigned>(base_buffer_.size()));
312 for (Level lvl : levels_) {
314 serde.
serialize(os, lvl.data(),
static_cast<unsigned>(lvl.size()));
319template<
typename T,
typename C,
typename A>
320template<
typename SerDe>
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;
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);
335 std::sort(
const_cast<Level&
>(base_buffer_).begin(),
const_cast<Level&
>(base_buffer_).end(), comparator_);
339 const uint8_t flags_byte(
340 (is_empty() ? 1 << flags::IS_EMPTY : 0)
341 | (1 << flags::IS_SORTED)
342 | (1 << flags::IS_COMPACT)
344 ptr += copy_to_mem(flags_byte, ptr);
345 ptr += copy_to_mem(k_, ptr);
346 ptr +=
sizeof(uint16_t);
350 ptr += copy_to_mem(n_, ptr);
357 if (base_buffer_.size() > 0)
358 ptr +=
serde.
serialize(ptr, end_ptr - ptr, base_buffer_.data(),
static_cast<unsigned>(base_buffer_.size()));
361 for (Level lvl : levels_) {
363 ptr +=
serde.
serialize(ptr, end_ptr - ptr, lvl.data(),
static_cast<unsigned>(lvl.size()));
370template<
typename T,
typename C,
typename A>
371template<
typename 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);
382 check_serial_version(serial_version);
383 check_family_id(family_id);
384 check_header_validity(preamble_longs, flags_byte, serial_version);
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;
389 return quantiles_sketch(k, comparator, allocator);
392 const auto items_seen = read<uint64_t>(is);
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;
397 optional<T> min_item;
398 optional<T> max_item;
402 typename std::aligned_storage<
sizeof(T),
alignof(T)>::type tmp_storage;
403 T* tmp =
reinterpret_cast<T*
>(&tmp_storage);
405 serde.deserialize(is, tmp, 1);
407 min_item.emplace(std::move(*tmp));
409 serde.deserialize(is, tmp, 1);
411 max_item.emplace(std::move(*tmp));
414 if (serial_version == 1) {
419 const uint8_t levels_needed = compute_levels_needed(k, items_seen);
420 const uint64_t bit_pattern = compute_bit_pattern(k, items_seen);
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) {
433 deserialize_array(is, items_to_read - bb_items, items_to_read - bb_items, serde, allocator);
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));
446 Level level(allocator);
448 levels.push_back(std::move(level));
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);
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 {
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);
465 items.get_deleter().set_destroy(
true);
466 if (!is.good())
throw std::runtime_error(
"error reading from std::istream");
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));
477template<
typename T,
typename C,
typename A>
478template<
typename 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;
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);
490 ptr += copy_from_mem(ptr, family_id);
492 ptr += copy_from_mem(ptr, flags_byte);
494 ptr += copy_from_mem(ptr, k);
496 ptr += copy_from_mem(ptr, unused);
499 check_serial_version(serial_version);
500 check_family_id(family_id);
501 check_header_validity(preamble_longs, flags_byte, serial_version);
503 const bool is_empty = (flags_byte & (1 << flags::IS_EMPTY)) > 0;
505 return quantiles_sketch(k, comparator, allocator);
508 ensure_minimum_memory(size, 16);
510 ptr += copy_from_mem(ptr, items_seen);
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;
515 optional<T> min_item;
516 optional<T> max_item;
520 typename std::aligned_storage<
sizeof(T),
alignof(T)>::type tmp_storage;
521 T* tmp =
reinterpret_cast<T*
>(&tmp_storage);
523 ptr += serde.deserialize(ptr, end_ptr - ptr, tmp, 1);
525 min_item.emplace(std::move(*tmp));
527 ptr += serde.deserialize(ptr, end_ptr - ptr, tmp, 1);
529 max_item.emplace(std::move(*tmp));
532 if (serial_version == 1) {
533 uint64_t unused_long;
534 ptr += copy_from_mem(ptr, unused_long);
538 const uint8_t levels_needed = compute_levels_needed(k, items_seen);
539 const uint64_t bit_pattern = compute_bit_pattern(k, items_seen);
547 const uint32_t bb_items = compute_base_buffer_items(k, items_seen);
548 uint32_t items_to_read = (levels_needed == 0 || is_compact) ? bb_items : 2 * k;
549 auto base_buffer_pair = deserialize_array(ptr, end_ptr - ptr, bb_items, 2 * k, serde, allocator);
550 ptr += base_buffer_pair.second;
551 if (items_to_read > bb_items) {
553 auto extras = deserialize_array(ptr, end_ptr - ptr, items_to_read - bb_items, items_to_read - bb_items, serde, allocator);
554 ptr += extras.second;
558 VectorLevels levels(allocator);
559 levels.reserve(levels_needed);
560 if (levels_needed > 0) {
561 uint64_t working_pattern = bit_pattern;
562 for (
size_t i = 0; i < levels_needed; ++i, working_pattern >>= 1) {
564 if ((working_pattern & 0x01) == 1) {
565 auto pair = deserialize_array(ptr, end_ptr - ptr, k, k, serde, allocator);
567 levels.push_back(std::move(pair.first));
569 Level level(allocator);
571 levels.push_back(std::move(level));
576 return quantiles_sketch(k, items_seen, bit_pattern,
577 std::move(base_buffer_pair.first), std::move(levels), std::move(min_item), std::move(max_item), is_sorted,
578 comparator, allocator);
581template<
typename T,
typename C,
typename A>
582template<
typename SerDe>
583auto 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)
584 -> std::pair<Level, size_t> {
585 const char* ptr =
static_cast<const char*
>(bytes);
586 const char* end_ptr =
static_cast<const char*
>(bytes) + size;
588 std::unique_ptr<T, items_deleter> items(alloc.allocate(num_items), items_deleter(allocator,
false, num_items));
589 ptr += serde.deserialize(ptr, end_ptr - ptr, items.get(), num_items);
591 items.get_deleter().set_destroy(
true);
594 Level level(allocator);
595 level.reserve(capacity);
596 level.insert(level.begin(),
597 std::make_move_iterator(items.get()),
598 std::make_move_iterator(items.get() + num_items));
600 return std::pair<Level, size_t>(std::move(level), ptr -
static_cast<const char*
>(bytes));
603template<
typename T,
typename C,
typename A>
607 std::ostringstream os;
608 os <<
"### Quantiles Sketch summary:" << std::endl;
609 os <<
" K : " << k_ << std::endl;
610 os <<
" N : " << n_ << std::endl;
611 os <<
" Epsilon : " << std::setprecision(3) << get_normalized_rank_error(
false) * 100 <<
"%" << std::endl;
612 os <<
" Epsilon PMF : " << get_normalized_rank_error(
true) * 100 <<
"%" << std::endl;
613 os <<
" Empty : " << (is_empty() ?
"true" :
"false") << std::endl;
614 os <<
" Estimation mode: " << (is_estimation_mode() ?
"true" :
"false") << std::endl;
615 os <<
" Levels (w/o BB): " << levels_.size() << std::endl;
616 os <<
" Used Levels : " << count_valid_levels(bit_pattern_) << std::endl;
617 os <<
" Retained items : " << get_num_retained() << std::endl;
619 os <<
" Min item : " << *min_item_ << std::endl;
620 os <<
" Max item : " << *max_item_ << std::endl;
622 os <<
"### End sketch summary" << std::endl;
625 os <<
"### Quantiles Sketch levels:" << std::endl;
626 os <<
" index: items in use" << std::endl;
627 os <<
" BB: " << base_buffer_.size() << std::endl;
628 for (uint8_t i = 0; i < levels_.size(); i++) {
629 os <<
" " <<
static_cast<unsigned int>(i) <<
": " << levels_[i].size() << std::endl;
631 os <<
"### End sketch levels" << std::endl;
635 os <<
"### Quantiles Sketch data:" << std::endl;
637 os <<
" BB:" << std::endl;
638 for (
const T& item : base_buffer_) {
639 os <<
" " << item << std::endl;
641 for (uint8_t i = 0; i < levels_.size(); ++i) {
642 os <<
" level " <<
static_cast<unsigned int>(level) <<
":" << std::endl;
643 for (
const T& item : levels_[i]) {
644 os <<
" " << item << std::endl;
647 os <<
"### End sketch data" << std::endl;
649 return string<A>(os.str().c_str(), allocator_);
652template<
typename T,
typename C,
typename A>
657template<
typename T,
typename C,
typename A>
662template<
typename T,
typename C,
typename A>
667template<
typename T,
typename C,
typename A>
669 return bit_pattern_ != 0;
672template<
typename T,
typename C,
typename A>
674 return compute_retained_items(k_, n_);
677template<
typename T,
typename C,
typename A>
679 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
683template<
typename T,
typename C,
typename A>
685 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
689template<
typename T,
typename C,
typename A>
694template<
typename T,
typename C,
typename A>
700template<
typename T,
typename C,
typename A>
701template<typename SerDe, typename TT, typename std::enable_if<std::is_arithmetic<TT>::value,
int>::type>
703 if (is_empty()) {
return EMPTY_SIZE_BYTES; }
704 return DATA_START + ((get_num_retained() + 2) *
sizeof(TT));
708template<
typename T,
typename C,
typename A>
709template<typename SerDe, typename TT, typename std::enable_if<!std::is_arithmetic<TT>::value,
int>::type>
711 if (is_empty()) {
return EMPTY_SIZE_BYTES; }
712 size_t size = DATA_START;
713 size += serde.size_of_item(*min_item_);
714 size += serde.size_of_item(*max_item_);
715 for (
auto it: *this) size += serde.size_of_item(it.first);
719template<
typename T,
typename C,
typename A>
721 return get_normalized_rank_error(k_, is_pmf);
724template<
typename T,
typename C,
typename A>
727 ? 1.854 / std::pow(k, 0.9657)
728 : 1.576 / std::pow(k, 0.9726);
731template<
typename T,
typename C,
typename A>
734 if (!is_base_buffer_sorted_) {
735 std::sort(
const_cast<Level&
>(base_buffer_).begin(),
const_cast<Level&
>(base_buffer_).end(), comparator_);
741 view.add(base_buffer_.begin(), base_buffer_.end(), weight);
742 for (
const auto& level: levels_) {
744 if (level.empty()) {
continue; }
745 view.add(level.begin(), level.end(), weight);
748 view.convert_to_cummulative();
752template<
typename T,
typename C,
typename A>
754 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
755 if ((rank < 0.0) || (rank > 1.0)) {
756 throw std::invalid_argument(
"Normalized rank cannot be less than 0 or greater than 1");
760 return sorted_view_->get_quantile(rank, inclusive);
763template<
typename T,
typename C,
typename A>
765 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
767 return sorted_view_->get_rank(item, inclusive);
770template<
typename T,
typename C,
typename A>
772 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
774 return sorted_view_->get_PMF(split_points, size, inclusive);
777template<
typename T,
typename C,
typename A>
779 if (is_empty())
throw std::runtime_error(
"operation is undefined for an empty sketch");
781 return sorted_view_->get_CDF(split_points, size, inclusive);
784template<
typename T,
typename C,
typename A>
786 const uint32_t bb_count = compute_base_buffer_items(k, n);
787 const uint64_t bit_pattern = compute_bit_pattern(k, n);
788 const uint32_t valid_levels = count_valid_levels(bit_pattern);
789 return bb_count + (k * valid_levels);
792template<
typename T,
typename C,
typename A>
793uint32_t quantiles_sketch<T, C, A>::compute_base_buffer_items(uint16_t k, uint64_t n) {
794 return n % (
static_cast<uint64_t
>(2) * k);
797template<
typename T,
typename C,
typename A>
798uint64_t quantiles_sketch<T, C, A>::compute_bit_pattern(uint16_t k, uint64_t n) {
799 return n / (
static_cast<uint64_t
>(2) * k);
802template<
typename T,
typename C,
typename A>
803uint32_t quantiles_sketch<T, C, A>::count_valid_levels(uint64_t bit_pattern) {
805 for (; bit_pattern > 0; ++count) bit_pattern &= bit_pattern - 1;
809template<
typename T,
typename C,
typename A>
810uint8_t quantiles_sketch<T, C, A>::compute_levels_needed(uint16_t k, uint64_t n) {
811 return static_cast<uint8_t
>(64U) - count_leading_zeros_in_u64(n / (2 * k));
814template<
typename T,
typename C,
typename A>
815void quantiles_sketch<T, C, A>::check_k(uint16_t k) {
817 throw std::invalid_argument(
"k must be a power of 2 that is >= "
823template<
typename T,
typename C,
typename A>
824void quantiles_sketch<T, C, A>::check_serial_version(uint8_t serial_version) {
825 if (serial_version == SERIAL_VERSION || serial_version == SERIAL_VERSION_1 || serial_version == SERIAL_VERSION_2)
828 throw std::invalid_argument(
"Possible corruption. Unrecognized serialization version: " + std::to_string(serial_version));
831template<
typename T,
typename C,
typename A>
832void quantiles_sketch<T, C, A>::check_family_id(uint8_t family_id) {
833 if (family_id == FAMILY)
836 throw std::invalid_argument(
"Possible corruption. Family id does not indicate quantiles sketch: " + std::to_string(family_id));
839template<
typename T,
typename C,
typename A>
840void quantiles_sketch<T, C, A>::check_header_validity(uint8_t preamble_longs, uint8_t flags_byte, uint8_t serial_version) {
841 const bool empty = (flags_byte & (1 << flags::IS_EMPTY)) > 0;
842 const bool compact = (flags_byte & (1 << flags::IS_COMPACT)) > 0;
844 const uint8_t sw = (compact ? 1 : 0) + (2 * (empty ? 1 : 0))
845 + (4 * (serial_version & 0xF)) + (32 * (preamble_longs & 0x3F));
864 std::ostringstream os;
865 os <<
"Possible sketch corruption. Inconsistent state: "
866 <<
"preamble_longs = " << preamble_longs
867 <<
", empty = " << (empty ?
"true" :
"false")
868 <<
", serialization_version = " << serial_version
869 <<
", compact = " << (compact ?
"true" :
"false");
870 throw std::invalid_argument(os.str());
874template <
typename T,
typename C,
typename A>
879template <
typename T,
typename C,
typename A>
884template<
typename T,
typename C,
typename A>
886 const size_t new_size = std::max(std::min(
static_cast<size_t>(2 * k_), 2 * base_buffer_.size()),
static_cast<size_t>(1));
887 base_buffer_.reserve(new_size);
890template<
typename T,
typename C,
typename A>
891void quantiles_sketch<T, C, A>::process_full_base_buffer() {
893 grow_levels_if_needed();
895 std::sort(base_buffer_.begin(), base_buffer_.end(), comparator_);
896 in_place_propagate_carry(0,
900 base_buffer_.clear();
901 is_base_buffer_sorted_ =
true;
902 if (n_ / (2 * k_) != bit_pattern_) {
903 throw std::logic_error(
"Internal error: n / 2k (" + std::to_string(n_ / 2 * k_)
904 +
" != bit_pattern " + std::to_string(bit_pattern_));
908template<
typename T,
typename C,
typename A>
909bool quantiles_sketch<T, C, A>::grow_levels_if_needed() {
910 const uint8_t levels_needed = compute_levels_needed(k_, n_);
911 if (levels_needed == 0)
915 if (levels_needed <= levels_.size())
918 Level empty_level(allocator_);
919 empty_level.reserve(k_);
920 levels_.push_back(std::move(empty_level));
924template<
typename T,
typename C,
typename A>
925template<
typename FwdV>
926void quantiles_sketch<T, C, A>::in_place_propagate_carry(uint8_t starting_level,
927 FwdV&& buf_size_k, Level& buf_size_2k,
928 bool apply_as_update,
929 quantiles_sketch& sketch) {
930 const uint64_t bit_pattern = sketch.bit_pattern_;
931 const int k = sketch.k_;
933 uint8_t ending_level = lowest_zero_bit_starting_at(bit_pattern, starting_level);
935 if (apply_as_update) {
938 zip_buffer(buf_size_2k, sketch.levels_[ending_level]);
941 for (uint16_t i = 0; i < k; ++i) {
942 sketch.levels_[ending_level].push_back(conditional_forward<FwdV>(buf_size_k[i]));
946 for (uint64_t lvl = starting_level; lvl < ending_level; lvl++) {
947 if ((bit_pattern & (
static_cast<uint64_t
>(1) << lvl)) == 0) {
948 throw std::logic_error(
"unexpected empty level in bit_pattern");
950 merge_two_size_k_buffers(
952 sketch.levels_[ending_level],
953 buf_size_2k, sketch.get_comparator());
954 sketch.levels_[lvl].clear();
955 sketch.levels_[ending_level].clear();
956 zip_buffer(buf_size_2k, sketch.levels_[ending_level]);
960 sketch.bit_pattern_ = bit_pattern + (
static_cast<uint64_t
>(1) << starting_level);
963template<
typename T,
typename C,
typename A>
964void quantiles_sketch<T, C, A>::zip_buffer(Level& buf_in, Level& buf_out) {
965#ifdef QUANTILES_VALIDATION
966 static uint32_t next_offset = 0;
967 uint32_t rand_offset = next_offset;
968 next_offset = 1 - next_offset;
970 uint32_t rand_offset = random_utils::random_bit();
972 if ((buf_in.size() != 2 * buf_out.capacity())
973 || (buf_out.size() > 0)) {
974 throw std::logic_error(
"zip_buffer requires buf_in.size() == "
975 "2*buf_out.capacity() and empty buf_out");
978 size_t k = buf_out.capacity();
979 for (uint32_t i = rand_offset, o = 0; o < k; i += 2, ++o) {
980 buf_out.push_back(std::move(buf_in[i]));
985template<
typename T,
typename C,
typename A>
986template<
typename FwdV>
987void quantiles_sketch<T, C, A>::zip_buffer_with_stride(FwdV&& buf_in, Level& buf_out, uint16_t stride) {
989 std::uniform_int_distribution<uint16_t> dist(0, stride - 1);
990 const uint16_t rand_offset = dist(random_utils::rand);
992 if ((buf_in.size() != stride * buf_out.capacity())
993 || (buf_out.size() > 0)) {
994 throw std::logic_error(
"zip_buffer_with_stride requires buf_in.size() == "
995 "stride*buf_out.capacity() and empty buf_out");
998 const size_t k = buf_out.capacity();
999 for (uint16_t i = rand_offset, o = 0; o < k; i += stride, ++o) {
1000 buf_out.push_back(conditional_forward<FwdV>(buf_in[i]));
1005template<
typename T,
typename C,
typename A>
1006void quantiles_sketch<T, C, A>::merge_two_size_k_buffers(Level& src_1, Level& src_2,
1007 Level& dst,
const C& comparator) {
1008 if (src_1.size() != src_2.size()
1009 || src_1.size() * 2 != dst.capacity()
1010 || dst.size() != 0) {
1011 throw std::logic_error(
"Input invariants violated in merge_two_size_k_buffers()");
1014 auto end1 = src_1.end(), end2 = src_2.end();
1015 auto it1 = src_1.begin(), it2 = src_2.begin();
1018 while (it1 != end1 && it2 != end2) {
1019 if (comparator(*it1, *it2)) {
1020 dst.push_back(std::move(*it1++));
1022 dst.push_back(std::move(*it2++));
1027 dst.insert(dst.end(), it1, end1);
1029 if (it2 == end2) {
throw std::logic_error(
"it2 unexpectedly already at end of range"); }
1030 dst.insert(dst.end(), it2, end2);
1034template<
typename T,
typename C,
typename A>
1035template<
typename FwdSk>
1036void quantiles_sketch<T, C, A>::standard_merge(quantiles_sketch& tgt, FwdSk&& src) {
1037 if (src.get_k() != tgt.get_k()) {
1038 throw std::invalid_argument(
"src.get_k() != tgt.get_k()");
1040 if (src.is_empty()) {
1044 uint64_t new_n = src.get_n() + tgt.get_n();
1047 for (uint16_t i = 0; i < src.base_buffer_.size(); ++i) {
1048 tgt.update(conditional_forward<FwdSk>(src.base_buffer_[i]));
1052 uint8_t levels_needed = compute_levels_needed(tgt.get_k(), new_n);
1053 if (levels_needed > tgt.levels_.size()) {
1054 tgt.levels_.reserve(levels_needed);
1055 while (tgt.levels_.size() < levels_needed) {
1056 Level empty_level(tgt.allocator_);
1057 empty_level.reserve(tgt.get_k());
1058 tgt.levels_.push_back(std::move(empty_level));
1062 Level scratch_buf(tgt.allocator_);
1063 scratch_buf.reserve(2 * tgt.get_k());
1065 uint64_t src_pattern = src.bit_pattern_;
1066 for (uint8_t src_lvl = 0; src_pattern != 0; ++src_lvl, src_pattern >>= 1) {
1067 if ((src_pattern & 1) > 0) {
1068 scratch_buf.clear();
1071 in_place_propagate_carry(src_lvl,
1072 src.levels_[src_lvl], scratch_buf,
1078 if ((tgt.get_n() / (2 * tgt.get_k())) != tgt.bit_pattern_) {
1079 throw std::logic_error(
"Failed internal consistency check after standard_merge()");
1085 if (!tgt.min_item_) {
1086 tgt.min_item_.emplace(conditional_forward<FwdSk>(*src.min_item_));
1088 if (tgt.comparator_(*src.min_item_, *tgt.min_item_))
1089 *tgt.min_item_ = conditional_forward<FwdSk>(*src.min_item_);
1091 if (!tgt.max_item_) {
1092 tgt.max_item_.emplace(conditional_forward<FwdSk>(*src.max_item_));
1094 if (tgt.comparator_(*tgt.max_item_, *src.max_item_))
1095 *tgt.max_item_ = conditional_forward<FwdSk>(*src.max_item_);
1099template<
typename T,
typename C,
typename A>
1100template<
typename FwdSk>
1101void quantiles_sketch<T, C, A>::downsampling_merge(quantiles_sketch& tgt, FwdSk&& src) {
1102 if (src.get_k() % tgt.get_k() != 0) {
1103 throw std::invalid_argument(
"src.get_k() is not a multiple of tgt.get_k()");
1105 if (src.is_empty()) {
1109 const uint16_t downsample_factor = src.get_k() / tgt.get_k();
1110 const uint8_t lg_sample_factor = count_trailing_zeros_in_u32(downsample_factor);
1112 const uint64_t new_n = src.get_n() + tgt.get_n();
1115 for (uint16_t i = 0; i < src.base_buffer_.size(); ++i) {
1116 tgt.update(conditional_forward<FwdSk>(src.base_buffer_[i]));
1120 const uint8_t levels_needed = compute_levels_needed(tgt.get_k(), new_n);
1121 if (levels_needed > tgt.levels_.size()) {
1122 tgt.levels_.reserve(levels_needed);
1123 while (tgt.levels_.size() < levels_needed) {
1124 Level empty_level(tgt.allocator_);
1125 empty_level.reserve(tgt.get_k());
1126 tgt.levels_.push_back(std::move(empty_level));
1130 Level down_buf(tgt.allocator_);
1131 down_buf.reserve(tgt.get_k());
1133 Level scratch_buf(tgt.allocator_);
1134 scratch_buf.reserve(2 * tgt.get_k());
1136 uint64_t src_pattern = src.bit_pattern_;
1137 for (uint8_t src_lvl = 0; src_pattern != 0; ++src_lvl, src_pattern >>= 1) {
1138 if ((src_pattern & 1) > 0) {
1140 scratch_buf.clear();
1143 zip_buffer_with_stride(src.levels_[src_lvl], down_buf, downsample_factor);
1146 in_place_propagate_carry(src_lvl + lg_sample_factor,
1147 down_buf, scratch_buf,
1153 if ((tgt.get_n() / (2 * tgt.get_k())) != tgt.bit_pattern_) {
1154 throw std::logic_error(
"Failed internal consistency check after downsampling_merge()");
1160 if (!tgt.min_item_) {
1161 tgt.min_item_.emplace(conditional_forward<FwdSk>(*src.min_item_));
1163 if (tgt.comparator_(*src.min_item_, *tgt.min_item_))
1164 *tgt.min_item_ = conditional_forward<FwdSk>(*src.min_item_);
1166 if (!tgt.max_item_) {
1167 tgt.max_item_.emplace(conditional_forward<FwdSk>(*src.max_item_));
1169 if (tgt.comparator_(*tgt.max_item_, *src.max_item_))
1170 *tgt.max_item_ = conditional_forward<FwdSk>(*src.max_item_);
1174template<
typename T,
typename C,
typename A>
1175uint8_t quantiles_sketch<T, C, A>::lowest_zero_bit_starting_at(uint64_t bits, uint8_t starting_bit) {
1176 uint8_t pos = starting_bit & 0X3F;
1177 uint64_t my_bits = bits >> pos;
1179 while ((my_bits &
static_cast<uint64_t
>(1)) != 0) {
1186template<
typename T,
typename C,
typename A>
1187class quantiles_sketch<T, C, A>::items_deleter {
1189 items_deleter(
const A& allocator,
bool destroy,
size_t num): allocator_(allocator), destroy_(destroy), num_(num) {}
1190 void operator() (T* ptr) {
1191 if (ptr !=
nullptr) {
1193 for (
size_t i = 0; i < num_; ++i) {
1197 allocator_.deallocate(ptr, num_);
1200 void set_destroy(
bool destroy) { destroy_ = destroy; }
1207template<
typename T,
typename C,
typename A>
1208void quantiles_sketch<T, C, A>::setup_sorted_view()
const {
1209 if (sorted_view_ ==
nullptr) {
1210 using AllocSortedView =
typename std::allocator_traits<A>::template rebind_alloc<quantiles_sorted_view<T, C, A>>;
1211 sorted_view_ =
new (AllocSortedView(allocator_).allocate(1)) quantiles_sorted_view<T, C, A>(get_sorted_view());
1215template<
typename T,
typename C,
typename A>
1216void quantiles_sketch<T, C, A>::reset_sorted_view() {
1217 if (sorted_view_ !=
nullptr) {
1218 sorted_view_->~quantiles_sorted_view();
1219 using AllocSortedView =
typename std::allocator_traits<A>::template rebind_alloc<quantiles_sorted_view<T, C, A>>;
1220 AllocSortedView(allocator_).deallocate(sorted_view_, 1);
1221 sorted_view_ =
nullptr;
1227template<
typename T,
typename C,
typename A>
1228quantiles_sketch<T, C, A>::const_iterator::const_iterator(
const Level& base_buffer,
1229 const std::vector<Level, AllocLevel>& levels,
1233base_buffer_(base_buffer),
1237bb_count_(compute_base_buffer_items(k, n)),
1238bit_pattern_(compute_bit_pattern(k, n)),
1245 if (bit_pattern_ == 0)
1246 index_ =
static_cast<uint32_t
>(n);
1248 level_ =
static_cast<int>(levels_.size());
1250 if (bb_count_ == 0 && bit_pattern_ > 0) {
1253 while ((bit_pattern_ & 0x01) == 0) {
1262template<
typename T,
typename C,
typename A>
1263typename quantiles_sketch<T, C, A>::const_iterator& quantiles_sketch<T, C, A>::const_iterator::operator++() {
1266 if ((level_ == -1 && index_ == base_buffer_.size() && levels_.size() > 0) || (level_ >= 0 && index_ == k_)) {
1270 if (level_ > 0) bit_pattern_ = bit_pattern_ >> 1;
1271 if (bit_pattern_ == 0)
return *
this;
1273 }
while ((bit_pattern_ &
static_cast<uint64_t
>(1)) == 0);
1278template<
typename T,
typename C,
typename A>
1279typename quantiles_sketch<T, C, A>::const_iterator& quantiles_sketch<T, C, A>::const_iterator::operator++(
int) {
1280 const_iterator tmp(*
this);
1285template<
typename T,
typename C,
typename A>
1286bool quantiles_sketch<T, C, A>::const_iterator::operator==(
const const_iterator& other)
const {
1287 return level_ == other.level_ && index_ == other.index_;
1290template<
typename T,
typename C,
typename A>
1291bool quantiles_sketch<T, C, A>::const_iterator::operator!=(
const const_iterator& other)
const {
1292 return !operator==(other);
1295template<
typename T,
typename C,
typename A>
1296auto quantiles_sketch<T, C, A>::const_iterator::operator*() const -> reference {
1297 return value_type(level_ == -1 ? base_buffer_[index_] : levels_[level_][index_], weight_);
1300template<
typename T,
typename C,
typename A>
1301auto quantiles_sketch<T, C, A>::const_iterator::operator->() const -> pointer {
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:690
quantiles_sorted_view< T, Comparator, Allocator > get_sorted_view() const
Gets the sorted view of this sketch.
Definition quantiles_sketch_impl.hpp:732
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:778
uint32_t get_num_retained() const
Returns the number of retained items (samples) in the sketch.
Definition quantiles_sketch_impl.hpp:673
double get_normalized_rank_error(bool is_pmf) const
Gets the normalized rank error for this sketch.
Definition quantiles_sketch_impl.hpp:720
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:771
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:604
uint16_t get_k() const
Returns configured parameter k.
Definition quantiles_sketch_impl.hpp:653
bool is_empty() const
Returns true if this sketch is empty.
Definition quantiles_sketch_impl.hpp:663
const T & get_max_item() const
Returns the max item of the stream.
Definition quantiles_sketch_impl.hpp:684
static quantiles_sketch deserialize(std::istream &is, const SerDe &sd=SerDe(), const Comparator &comparator=Comparator(), const Allocator &allocator=Allocator())
This method deserializes a sketch from a given stream.
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:764
allocator_type get_allocator() const
Returns the allocator for this sketch.
Definition quantiles_sketch_impl.hpp:695
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:753
const_iterator begin() const
Iterator pointing to the first item in the sketch.
Definition quantiles_sketch_impl.hpp:875
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:702
bool is_estimation_mode() const
Returns true if this sketch is in estimation mode.
Definition quantiles_sketch_impl.hpp:668
const T & get_min_item() const
Returns the min item of the stream.
Definition quantiles_sketch_impl.hpp:678
const_iterator end() const
Iterator pointing to the past-the-end item in the sketch.
Definition quantiles_sketch_impl.hpp:880
uint64_t get_n() const
Returns the length of the input stream.
Definition quantiles_sketch_impl.hpp:658
Sorted view for quantiles sketches (REQ, KLL and Quantiles)
Definition quantiles_sorted_view.hpp:38
const uint16_t MAX_K
maximum value of parameter K
Definition quantiles_sketch.hpp:41
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:34
void serialize(std::ostream &os, const T *items, unsigned num) const
Stream serialization.