Encode decode store

Author: geonove

ddsketch/include/collapsing_dense_store.hpp

@@ -32,6 +32,11 @@ class CollapsingDenseStore : public DenseStore<Derived, Allocator> {
   explicit CollapsingDenseStore(size_type max_num_bins);
   CollapsingDenseStore(const CollapsingDenseStore<Derived, Allocator>& other) = default;
 
+  CollapsingDenseStore<Derived,Allocator>& operator=(const CollapsingDenseStore<Derived, Allocator>& other);
+
+  void serialize(std::ostream& os) const;
+  static Derived deserialize(std::istream& is);
+
   ~CollapsingDenseStore() = default;
 
   void clear();

ddsketch/include/collapsing_dense_store_impl.hpp

@@ -26,7 +26,20 @@ namespace datasketches {
 template<class Derived, typename Allocator>
 CollapsingDenseStore<Derived, Allocator>::CollapsingDenseStore(size_type max_num_bins): DenseStore<Derived, Allocator>(), max_num_bins(max_num_bins), is_collapsed(false) {}
 
-// TODO implement copy constructor
+template<class Derived, typename Allocator>
+CollapsingDenseStore<Derived, Allocator>& CollapsingDenseStore<Derived, Allocator>::operator=(const CollapsingDenseStore<Derived, Allocator>& other) {
+  if (max_num_bins != other.max_num_bins) {
+    throw std::runtime_error("stores must have same maximum number of bins");
+  }
+
+  this->bins = other.bins;
+  this->offset = other.offset;
+  this->min_index = other.min_index;
+  this->max_index = other.max_index;
+
+
+  return *this;
+}
 
 template<class Derived, typename Allocator>
 typename CollapsingDenseStore<Derived, Allocator>::size_type CollapsingDenseStore<Derived, Allocator>::get_new_length(size_type new_min_index, size_type new_max_index) const {
@@ -39,5 +52,29 @@ void CollapsingDenseStore<Derived, Allocator>::clear() {
   is_collapsed = false;
 }
 
+template<class Derived, typename Allocator>
+void CollapsingDenseStore<Derived, Allocator>::serialize(std::ostream& os) const {
+  write(os, max_num_bins);
+  if (this->is_empty()) {
+    return;
+  }
+  write(os, is_collapsed);
+
+  this->serialize_common(os);
+}
+
+template<class Derived, typename Allocator>
+Derived CollapsingDenseStore<Derived, Allocator>::deserialize(std::istream& is) {
+  Derived store(read<int>(is));
+
+  if (is.peek() == std::istream::traits_type::eof()) {
+    return store;
+  }
+  store.is_collapsed = read<bool>(is);
+  Derived::deserialize_common(store, is);
+
+  return store;
+}
+
 }
 #endif //COLLAPSING_DENSE_STORE_IMPL_HPP

ddsketch/include/dense_store.hpp

@@ -50,8 +50,10 @@ class DenseStore {
 
   void merge(const SparseStore<Allocator>& other);
 
-  void serialize(std::ostream& is) const;
-  DenseStore<Derived, Allocator> deserialize(std::ostream& is);
+  void serialize(std::ostream& os) const;
+  static Derived deserialize(std::istream& is);
+
+  bool operator==(const DenseStore<Derived, Allocator>& other) const;
 
   iterator begin() const;
   iterator end() const;
@@ -132,6 +134,9 @@ class DenseStore {
   void reset_bins();
   void reset_bins(size_type from_index, size_type to_index);
 
+  void serialize_common(std::ostream& os) const;
+  static void deserialize_common(Derived& store, std::istream& is);
+
   Derived& derived();
   const Derived& derived() const;
 };

ddsketch/include/dense_store_impl.hpp

@@ -21,6 +21,9 @@
 #define DENSE_STORE_IMPL_HPP
 
 #include <algorithm>
+#include <filesystem>
+
+#include "common_defs.hpp"
 
 namespace datasketches {
 
@@ -247,6 +250,14 @@ void DenseStore<Derived, Allocator>::reset_bins(size_type from_index, size_type
   std::fill(bins.begin() + from_index - offset, bins.begin() + to_index - offset + 1, 0);
 }
 
+template<class Derived, typename Allocator>
+bool DenseStore<Derived, Allocator>::operator==(const DenseStore<Derived, Allocator>& other) const {
+  return offset == other.offset &&
+    min_index == other.min_index &&
+    max_index == other.max_index &&
+    bins == other.bins;
+}
+
 template<class Derived, typename Allocator>
 Derived& DenseStore<Derived, Allocator>::derived() {
   return static_cast<Derived&>(*this);
@@ -338,11 +349,62 @@ typename DenseStore<Derived, Allocator>::reverse_iterator::reference DenseStore<
 
 template<class Derived, typename Allocator>
 void DenseStore<Derived, Allocator>::serialize(std::ostream& os) const {
+  derived().serialize(os);
+}
+
+template<class Derived, typename Allocator>
+Derived DenseStore<Derived, Allocator>::deserialize(std::istream& is) {
+  return Derived::deserialize(is);
+}
+
+template<class Derived, typename Allocator>
+void DenseStore<Derived, Allocator>::serialize_common(std::ostream& os) const {
   if (is_empty()) {
     return;
   }
 
+  // Serialize the range information
+  write(os, min_index);
+  write(os, max_index);
+  write(os, offset);
 
+  // Serialize the bins array (only the used portion)
+  const size_type num_bins = bins.size();
+  write(os, num_bins);
+
+  size_type non_empty_bins = 0;
+  for (const double& count : bins) {
+    non_empty_bins += (count > 0.0);
+  }
+  write(os, non_empty_bins);
+
+  for (const Bin& bin : *this) {
+    write(os, bin.getIndex());
+    write(os, bin.getCount());
+  }
+}
+
+template<class Derived, typename Allocator>
+void DenseStore<Derived, Allocator>::deserialize_common(Derived& store, std::istream& is) {
+  if (is.peek() == std::istream::traits_type::eof()) {
+    return;
+  }
+  // Deserialize the range information
+  store.min_index = read<size_type>(is);
+  store.max_index = read<size_type>(is);
+  store.offset = read<size_type>(is);
+
+  // Deserialize the bins array
+  const auto num_bins = read<size_type>(is);
+  store.bins.resize(num_bins, 0.0);
+
+  const auto non_empty_bins = read<size_type>(is);
+  // Read the actual bin counts
+  for (size_type i = 0; i < non_empty_bins; ++i) {
+    const auto index =  read<int>(is);
+    const auto count = read<double>(is);
+    store.add(index, count);
+  }
 }
 }
 

ddsketch/include/sparse_store.hpp

@@ -43,6 +43,8 @@ class SparseStore {
 
   SparseStore() = default;
 
+  bool operator==(const SparseStore &other) const;
+
   void add(int index);
   void add(int index, double count);
   void add(const Bin& bin);
@@ -57,6 +59,9 @@ class SparseStore {
   bool is_empty() const;
   double get_total_count() const;
 
+  void serialize(std::ostream& os) const;
+  static SparseStore deserialize(std::istream& is);
+
   iterator begin() const;
   iterator end() const;
   reverse_iterator rbegin() const;

ddsketch/include/sparse_store_impl.hpp

@@ -23,6 +23,12 @@
 #include "sparse_store.hpp"
 
 namespace datasketches {
+
+template<typename Allocator>
+bool SparseStore<Allocator>::operator==(const SparseStore &other) const {
+  return bins == other.bins;
+}
+
 template<typename Allocator>
 void SparseStore<Allocator>::add(int index) {
   add(index, 1);
@@ -190,6 +196,29 @@ double SparseStore<Allocator>::get_total_count() const {
   }
   return total_count;
 }
+
+template<typename Allocator>
+void SparseStore<Allocator>::serialize(std::ostream &os) const {
+  write(os, bins.size());
+  for (const auto& [index, count] : bins) {
+    write(os, index);
+    write(os, count);
+  }
+}
+
+template<typename Allocator>
+SparseStore<Allocator> SparseStore<Allocator>::deserialize(std::istream& is) {
+  SparseStore<Allocator> store;
+  const auto num_bins = read<typename bins_type::size_type>(is);
+  for (typename bins_type::size_type i = 0; i < num_bins; ++i) {
+    const auto index = read<typename bins_type::key_type>(is);
+    const auto count = read<typename bins_type::mapped_type>(is);
+    store.bins[index] = count;
+  }
+
+  return store;
+}
+
 }
 
 #endif //SPARSE_STORE_IMPL_HPP

ddsketch/include/unbounded_size_dense_store.hpp

@@ -35,9 +35,15 @@ class UnboundedSizeDenseStore: public DenseStore<UnboundedSizeDenseStore<Allocat
   UnboundedSizeDenseStore* copy() const;
   ~UnboundedSizeDenseStore() = default;
 
+  UnboundedSizeDenseStore& operator=(const UnboundedSizeDenseStore& other);
+
   void merge(const UnboundedSizeDenseStore<Allocator>& other);
   using DenseStore<UnboundedSizeDenseStore, Allocator>::merge;
 
+
+  void serialize(std::ostream& os) const;
+  static UnboundedSizeDenseStore deserialize(std::istream& is);
+
 protected:
   size_type normalize(size_type index);
   void adjust(size_type new_min_index, size_type new_max_index);

ddsketch/include/unbounded_size_dense_store_impl.hpp

@@ -48,6 +48,19 @@ UnboundedSizeDenseStore<Allocator> *UnboundedSizeDenseStore<Allocator>::copy() c
   return new (alloc.allocate(1)) UnboundedSizeDenseStore<Allocator>(*this);
 }
 
+template<typename Allocator>
+UnboundedSizeDenseStore<Allocator> &UnboundedSizeDenseStore<Allocator>::operator=(const UnboundedSizeDenseStore &other) {
+
+  this->bins = other.bins;
+  this->offset = other.offset;
+  this->min_index = other.min_index;
+  this->max_index = other.max_index;
+
+
+  return *this;
+}
+
+
 
 template<typename Allocator>
 void UnboundedSizeDenseStore<Allocator>::adjust(size_type new_min_index, size_type new_max_index) {
@@ -69,6 +82,21 @@ void UnboundedSizeDenseStore<Allocator>::merge(const UnboundedSizeDenseStore<All
   }
 }
 
+template<typename Allocator>
+void UnboundedSizeDenseStore<Allocator>::serialize(std::ostream &os) const {
+  this->serialize_common(os);
+}
+
+template<typename Allocator>
+UnboundedSizeDenseStore<Allocator> UnboundedSizeDenseStore<Allocator>::deserialize(std::istream &is) {
+  UnboundedSizeDenseStore<Allocator> store;
+  UnboundedSizeDenseStore::deserialize_common(store, is);
+
+  return store;
+}
+
+
+
 
 
 }

ddsketch/test/StoreTest.cpp

@@ -18,6 +18,7 @@
  */
 
 #include <iostream>
+#include <sstream>
 #include <catch2/catch.hpp>
 
 #include "collapsing_highest_dense_store.hpp"
@@ -543,5 +544,44 @@ TEST_CASE("merge test", "[mergetest]") {
 
   //DDSketch<CollapsingHighestDenseStore<A>, LinearlyInterpolatedMapping> sketch4<1024>(0.99);
 }
+
+TEMPLATE_TEST_CASE("dense store serialization test", "[serialization]",
+  CollapsingLowestStoreTestCase<8>,
+  CollapsingLowestStoreTestCase<128>,
+  CollapsingLowestStoreTestCase<1024>,
+  CollapsingHighestStoreTestCase<8>,
+  CollapsingHighestStoreTestCase<128>,
+  CollapsingHighestStoreTestCase<1024>,
+  UnboundedStoreSizeTestCase,
+  SparseStoreTestCase
+) {
+  // Test empty store serialization
+  auto store = *TestType::new_store();
+  using StoreType = decltype(store);
+  std::stringstream stream;
+
+  store.serialize(stream);
+  StoreType deserialized_empty_store = StoreType::deserialize(stream);
+  REQUIRE(store.is_empty());
+  REQUIRE(deserialized_empty_store.is_empty());
+  REQUIRE(stream.peek() == std::istream::traits_type::eof());
+  REQUIRE(store == deserialized_empty_store);
+  stream.clear();
+
+  std::vector<int> indexes{-1000, -1, 0, 1, 1000};
+  std::vector<double> counts{0, 1, 2, 4, 5, 10, 20, 100, 1000, 10000};
+  for (int idx: indexes) {
+    for (double count: counts) {
+      store.add(idx, count);
+    }
+  }
+
+  store.serialize(stream);
+  auto deserialized_store = StoreType::deserialize(stream);
+  REQUIRE_FALSE(store.is_empty());
+  REQUIRE_FALSE(deserialized_store.is_empty());
+  REQUIRE(stream.peek() == std::istream::traits_type::eof());
+  REQUIRE(store == deserialized_store);
+}
 }