#ifndef _VAR_OPT_UNION_IMPL_HPP_
#define _VAR_OPT_UNION_IMPL_HPP_
#include "var_opt_union.hpp"
#include <cmath>
#include <sstream>
#include <stdexcept>
namespace datasketches {
template<typename T, typename A>
var_opt_union<T, A>::var_opt_union(uint32_t max_k, const A& allocator) :
n_(0),
outer_tau_numer_(0.0),
outer_tau_denom_(0),
max_k_(max_k),
allocator_(allocator),
gadget_(max_k, var_opt_sketch<T, A>::DEFAULT_RESIZE_FACTOR, true, allocator)
{}
template<typename T, typename A>
var_opt_union<T, A>::var_opt_union(const var_opt_union& other) :
n_(other.n_),
outer_tau_numer_(other.outer_tau_numer_),
outer_tau_denom_(other.outer_tau_denom_),
max_k_(other.max_k_),
allocator_(other.allocator_),
gadget_(other.gadget_)
{}
template<typename T, typename A>
var_opt_union<T, A>::var_opt_union(var_opt_union&& other) noexcept :
n_(other.n_),
outer_tau_numer_(other.outer_tau_numer_),
outer_tau_denom_(other.outer_tau_denom_),
max_k_(other.max_k_),
allocator_(other.allocator_),
gadget_(std::move(other.gadget_))
{}
template<typename T, typename A>
var_opt_union<T, A>::var_opt_union(uint64_t n, double outer_tau_numer, uint64_t outer_tau_denom,
uint32_t max_k, var_opt_sketch<T, A>&& gadget, const A& allocator) :
n_(n),
outer_tau_numer_(outer_tau_numer),
outer_tau_denom_(outer_tau_denom),
max_k_(max_k),
allocator_(allocator),
gadget_(gadget)
{}
template<typename T, typename A>
var_opt_union<T, A>::~var_opt_union() {}
template<typename T, typename A>
var_opt_union<T, A>& var_opt_union<T, A>::operator=(const var_opt_union& other) {
var_opt_union union_copy(other);
std::swap(n_, union_copy.n_);
std::swap(outer_tau_numer_, union_copy.outer_tau_numer_);
std::swap(outer_tau_denom_, union_copy.outer_tau_denom_);
std::swap(max_k_, union_copy.max_k_);
std::swap(allocator_, other.allocator_);
std::swap(gadget_, union_copy.gadget_);
return *this;
}
template<typename T, typename A>
var_opt_union<T, A>& var_opt_union<T, A>::operator=(var_opt_union&& other) {
std::swap(n_, other.n_);
std::swap(outer_tau_numer_, other.outer_tau_numer_);
std::swap(outer_tau_denom_, other.outer_tau_denom_);
std::swap(max_k_, other.max_k_);
std::swap(allocator_, other.allocator_);
std::swap(gadget_, other.gadget_);
return *this;
}
template<typename T, typename A>
template<typename SerDe>
var_opt_union<T, A> var_opt_union<T, A>::deserialize(std::istream& is, const SerDe& sd, const A& allocator) {
const auto preamble_longs = read<uint8_t>(is);
const auto serial_version = read<uint8_t>(is);
const auto family_id = read<uint8_t>(is);
const auto flags = read<uint8_t>(is);
const auto max_k = read<uint32_t>(is);
check_preamble_longs(preamble_longs, flags);
check_family_and_serialization_version(family_id, serial_version);
if (max_k == 0 || max_k > var_opt_constants::MAX_K) {
throw std::invalid_argument("k must be at least 1 and less than 2^31 - 1");
}
bool is_empty = flags & EMPTY_FLAG_MASK;
if (is_empty) {
if (!is.good())
throw std::runtime_error("error reading from std::istream");
else
return var_opt_union(max_k);
}
const auto items_seen = read<uint64_t>(is);
const auto outer_tau_numer = read<double>(is);
const auto outer_tau_denom = read<uint64_t>(is);
var_opt_sketch<T, A> gadget = var_opt_sketch<T, A>::deserialize(is, sd, allocator);
if (!is.good())
throw std::runtime_error("error reading from std::istream");
return var_opt_union(items_seen, outer_tau_numer, outer_tau_denom, max_k, std::move(gadget), allocator);
}
template<typename T, typename A>
template<typename SerDe>
var_opt_union<T, A> var_opt_union<T, A>::deserialize(const void* bytes, size_t size, const SerDe& sd, const A& allocator) {
ensure_minimum_memory(size, 8);
const char* ptr = static_cast<const char*>(bytes);
uint8_t preamble_longs;
ptr += copy_from_mem(ptr, preamble_longs);
uint8_t serial_version;
ptr += copy_from_mem(ptr, serial_version);
uint8_t family_id;
ptr += copy_from_mem(ptr, family_id);
uint8_t flags;
ptr += copy_from_mem(ptr, flags);
uint32_t max_k;
ptr += copy_from_mem(ptr, max_k);
check_preamble_longs(preamble_longs, flags);
check_family_and_serialization_version(family_id, serial_version);
if (max_k == 0 || max_k > var_opt_constants::MAX_K) {
throw std::invalid_argument("k must be at least 1 and less than 2^31 - 1");
}
bool is_empty = flags & EMPTY_FLAG_MASK;
if (is_empty) {
return var_opt_union(max_k);
}
uint64_t items_seen;
ptr += copy_from_mem(ptr, items_seen);
double outer_tau_numer;
ptr += copy_from_mem(ptr, outer_tau_numer);
uint64_t outer_tau_denom;
ptr += copy_from_mem(ptr, outer_tau_denom);
const size_t gadget_size = size - (PREAMBLE_LONGS_NON_EMPTY << 3);
var_opt_sketch<T, A> gadget = var_opt_sketch<T, A>::deserialize(ptr, gadget_size, sd, allocator);
return var_opt_union(items_seen, outer_tau_numer, outer_tau_denom, max_k, std::move(gadget), allocator);
}
template<typename T, typename A>
template<typename SerDe>
size_t var_opt_union<T, A>::get_serialized_size_bytes(const SerDe& sd) const {
if (n_ == 0) {
return PREAMBLE_LONGS_EMPTY << 3;
} else {
return (PREAMBLE_LONGS_NON_EMPTY << 3) + gadget_.get_serialized_size_bytes(sd);
}
}
template<typename T, typename A>
template<typename SerDe>
void var_opt_union<T, A>::serialize(std::ostream& os, const SerDe& sd) const {
bool empty = (n_ == 0);
const uint8_t serialization_version(SER_VER);
const uint8_t family_id(FAMILY_ID);
uint8_t preamble_longs;
uint8_t flags;
if (empty) {
preamble_longs = PREAMBLE_LONGS_EMPTY;
flags = EMPTY_FLAG_MASK;
} else {
preamble_longs = PREAMBLE_LONGS_NON_EMPTY;
flags = 0;
}
write(os, preamble_longs);
write(os, serialization_version);
write(os, family_id);
write(os, flags);
write(os, max_k_);
if (!empty) {
write(os, n_);
write(os, outer_tau_numer_);
write(os, outer_tau_denom_);
gadget_.serialize(os, sd);
}
}
template<typename T, typename A>
template<typename SerDe>
std::vector<uint8_t, AllocU8<A>> var_opt_union<T, A>::serialize(unsigned header_size_bytes, const SerDe& sd) const {
const size_t size = header_size_bytes + get_serialized_size_bytes(sd);
std::vector<uint8_t, AllocU8<A>> bytes(size, 0, gadget_.allocator_);
uint8_t* ptr = bytes.data() + header_size_bytes;
const bool empty = n_ == 0;
const uint8_t serialization_version(SER_VER);
const uint8_t family_id(FAMILY_ID);
uint8_t preamble_longs;
uint8_t flags;
if (empty) {
preamble_longs = PREAMBLE_LONGS_EMPTY;
flags = EMPTY_FLAG_MASK;
} else {
preamble_longs = PREAMBLE_LONGS_NON_EMPTY;
flags = 0;
}
ptr += copy_to_mem(preamble_longs, ptr);
ptr += copy_to_mem(serialization_version, ptr);
ptr += copy_to_mem(family_id, ptr);
ptr += copy_to_mem(flags, ptr);
ptr += copy_to_mem(max_k_, ptr);
if (!empty) {
ptr += copy_to_mem(n_, ptr);
ptr += copy_to_mem(outer_tau_numer_, ptr);
ptr += copy_to_mem(outer_tau_denom_, ptr);
auto gadget_bytes = gadget_.serialize(0, sd);
ptr += copy_to_mem(gadget_bytes.data(), ptr, gadget_bytes.size() * sizeof(uint8_t));
}
return bytes;
}
template<typename T, typename A>
void var_opt_union<T, A>::reset() {
n_ = 0;
outer_tau_numer_ = 0.0;
outer_tau_denom_ = 0;
gadget_.reset();
}
template<typename T, typename A>
string<A> var_opt_union<T, A>::to_string() const {
std::ostringstream os;
os << "### VarOpt Union SUMMARY:" << std::endl;
os << " n : " << n_ << std::endl;
os << " Max k : " << max_k_ << std::endl;
os << " Gadget Summary:" << std::endl;
os << gadget_.to_string();
os << "### END VarOpt Union SUMMARY" << std::endl;
return string<A>(os.str().c_str(), gadget_.allocator_);
}
template<typename T, typename A>
void var_opt_union<T, A>::update(const var_opt_sketch<T, A>& sk) {
merge_items(sk);
resolve_tau(sk);
}
template<typename T, typename A>
void var_opt_union<T, A>::update(var_opt_sketch<T, A>&& sk) {
merge_items(std::move(sk));
resolve_tau(sk); }
template<typename T, typename A>
double var_opt_union<T, A>::get_outer_tau() const {
if (outer_tau_denom_ == 0) {
return 0.0;
} else {
return outer_tau_numer_ / outer_tau_denom_;
}
}
template<typename T, typename A>
void var_opt_union<T, A>::merge_items(const var_opt_sketch<T, A>& sketch) {
if (sketch.n_ == 0) {
return;
}
n_ += sketch.n_;
typename var_opt_sketch<T, A>::const_iterator h_itr(sketch, false, false);
typename var_opt_sketch<T, A>::const_iterator h_end(sketch, true, false);
while (h_itr != h_end) {
std::pair<const T&, const double> sample = *h_itr;
gadget_.update(sample.first, sample.second, false);
++h_itr;
}
typename var_opt_sketch<T, A>::iterator r_itr(sketch, false, true);
typename var_opt_sketch<T, A>::iterator r_end(sketch, true, true);
while (r_itr != r_end) {
std::pair<const T&, const double> sample = *r_itr;
gadget_.update(sample.first, sample.second, true);
++r_itr;
}
}
template<typename T, typename A>
void var_opt_union<T, A>::merge_items(var_opt_sketch<T, A>&& sketch) {
if (sketch.n_ == 0) {
return;
}
n_ += sketch.n_;
typename var_opt_sketch<T, A>::iterator h_itr(sketch, false, false);
typename var_opt_sketch<T, A>::iterator h_end(sketch, true, false);
while (h_itr != h_end) {
std::pair<T&, double> sample = *h_itr;
gadget_.update(std::move(sample.first), sample.second, false);
++h_itr;
}
typename var_opt_sketch<T, A>::iterator r_itr(sketch, false, true);
typename var_opt_sketch<T, A>::iterator r_end(sketch, true, true);
while (r_itr != r_end) {
std::pair<T&, double> sample = *r_itr;
gadget_.update(std::move(sample.first), sample.second, true);
++r_itr;
}
}
template<typename T, typename A>
void var_opt_union<T, A>::resolve_tau(const var_opt_sketch<T, A>& sketch) {
if (sketch.r_ > 0) {
const double sketch_tau = sketch.get_tau();
const double outer_tau = get_outer_tau();
if (outer_tau_denom_ == 0) {
outer_tau_numer_ = sketch.total_wt_r_;
outer_tau_denom_ = sketch.r_;
} else if (sketch_tau > outer_tau) {
outer_tau_numer_ = sketch.total_wt_r_;
outer_tau_denom_ = sketch.r_;
} else if (sketch_tau == outer_tau) {
outer_tau_numer_ += sketch.total_wt_r_;
outer_tau_denom_ += sketch.r_;
}
}
}
template<typename T, typename A>
var_opt_sketch<T, A> var_opt_union<T, A>::get_result() const {
if (gadget_.num_marks_in_h_ == 0) {
return simple_gadget_coercer();
} else {
var_opt_sketch<T, A> gcopy(gadget_, false, n_);
const bool is_pseudo_exact = detect_and_handle_subcase_of_pseudo_exact(gcopy);
if (!is_pseudo_exact) {
migrate_marked_items_by_decreasing_k(gcopy);
}
return gcopy;
}
}
template<typename T, typename A>
var_opt_sketch<T, A> var_opt_union<T, A>::simple_gadget_coercer() const {
if (gadget_.num_marks_in_h_ != 0) throw std::logic_error("simple gadget coercer only applies if no marks");
return var_opt_sketch<T, A>(gadget_, true, n_);
}
template<typename T, typename A>
bool var_opt_union<T, A>::there_exist_unmarked_h_items_lighter_than_target(double threshold) const {
for (uint32_t i = 0; i < gadget_.h_; ++i) {
if ((gadget_.weights_[i] < threshold) && !gadget_.marks_[i]) {
return true;
}
}
return false;
}
template<typename T, typename A>
bool var_opt_union<T, A>::detect_and_handle_subcase_of_pseudo_exact(var_opt_sketch<T, A>& sk) const {
const bool condition1 = gadget_.r_ == 0;
const bool condition2 = gadget_.num_marks_in_h_ > 0;
const bool condition3 = gadget_.num_marks_in_h_ == outer_tau_denom_;
if (!(condition1 && condition2 && condition3)) {
return false;
} else {
const bool anti_condition4 = there_exist_unmarked_h_items_lighter_than_target(gadget_.get_tau());
if (anti_condition4) {
return false;
} else {
mark_moving_gadget_coercer(sk);
return true;
}
}
}
template<typename T, typename A>
void var_opt_union<T, A>::mark_moving_gadget_coercer(var_opt_sketch<T, A>& sk) const {
const uint32_t result_k = gadget_.h_ + gadget_.r_;
uint32_t result_h = 0;
uint32_t result_r = 0;
size_t next_r_pos = result_k;
double* wts = AllocDouble(allocator_).allocate(result_k + 1);
T* data = A(allocator_).allocate(result_k + 1);
const size_t final_idx = gadget_.get_num_samples();
for (size_t idx = gadget_.h_ + 1; idx <= final_idx; ++idx) {
new (&data[next_r_pos]) T(gadget_.data_[idx]);
wts[next_r_pos] = gadget_.weights_[idx];
++result_r;
--next_r_pos;
}
double transferred_weight = 0;
for (size_t idx = 0; idx < gadget_.h_; ++idx) {
if (gadget_.marks_[idx]) {
new (&data[next_r_pos]) T(gadget_.data_[idx]);
wts[next_r_pos] = -1.0;
transferred_weight += gadget_.weights_[idx];
++result_r;
--next_r_pos;
} else {
new (&data[result_h]) T(gadget_.data_[idx]);
wts[result_h] = gadget_.weights_[idx];
++result_h;
}
}
if (result_h + result_r != result_k) throw std::logic_error("H + R counts must equal k");
if (std::abs(transferred_weight - outer_tau_numer_) > 1e-10) {
throw std::logic_error("unexpected mismatch in transferred weight");
}
const double result_r_weight = gadget_.total_wt_r_ + transferred_weight;
const uint64_t result_n = n_;
wts[result_h] = -1.0;
AllocBool(allocator_).deallocate(sk.marks_, sk.curr_items_alloc_);
AllocDouble(allocator_).deallocate(sk.weights_, sk.curr_items_alloc_);
for (size_t i = 0; i < result_k; ++i) { sk.data_[i].~T(); } A(allocator_).deallocate(sk.data_, sk.curr_items_alloc_);
sk.data_ = data;
sk.weights_ = wts;
sk.marks_ = nullptr;
sk.num_marks_in_h_ = 0;
sk.curr_items_alloc_ = result_k + 1;
sk.k_ = result_k;
sk.n_ = result_n;
sk.h_ = result_h;
sk.r_ = result_r;
sk.total_wt_r_ = result_r_weight;
}
template<typename T, typename A>
void var_opt_union<T, A>::migrate_marked_items_by_decreasing_k(var_opt_sketch<T, A>& gcopy) const {
const uint32_t r_count = gcopy.r_;
const uint32_t h_count = gcopy.h_;
const uint32_t k = gcopy.k_;
if (gcopy.num_marks_in_h_ == 0) throw std::logic_error("unexpectedly found no marked items to migrate");
if ((r_count != 0) && ((h_count + r_count) != k)) throw std::logic_error("invalid gadget state");
if ((r_count == 0) && (h_count < k)) {
gcopy.k_ = h_count; }
gcopy.decrease_k_by_1();
if (gcopy.get_tau() == 0.0) throw std::logic_error("gadget must be in sampling mode");
while (gcopy.num_marks_in_h_ > 0) {
gcopy.decrease_k_by_1();
}
gcopy.strip_marks();
}
template<typename T, typename A>
void var_opt_union<T, A>::check_preamble_longs(uint8_t preamble_longs, uint8_t flags) {
bool is_empty(flags & EMPTY_FLAG_MASK);
if (is_empty) {
if (preamble_longs != PREAMBLE_LONGS_EMPTY) {
throw std::invalid_argument("Possible corruption: Preamble longs must be "
+ std::to_string(PREAMBLE_LONGS_EMPTY) + " for an empty sketch. Found: "
+ std::to_string(preamble_longs));
}
} else {
if (preamble_longs != PREAMBLE_LONGS_NON_EMPTY) {
throw std::invalid_argument("Possible corruption: Preamble longs must be "
+ std::to_string(PREAMBLE_LONGS_NON_EMPTY)
+ " for a non-empty sketch. Found: " + std::to_string(preamble_longs));
}
}
}
template<typename T, typename A>
void var_opt_union<T, A>::check_family_and_serialization_version(uint8_t family_id, uint8_t ser_ver) {
if (family_id == FAMILY_ID) {
if (ser_ver != SER_VER) {
throw std::invalid_argument("Possible corruption: VarOpt Union serialization version must be "
+ std::to_string(SER_VER) + ". Found: " + std::to_string(ser_ver));
}
return;
}
throw std::invalid_argument("Possible corruption: VarOpt Union family id must be "
+ std::to_string(FAMILY_ID) + ". Found: " + std::to_string(family_id));
}
}
#endif