Implement ddsketch serialize and deserialize

Author: geonove

ddsketch/include/collapsing_dense_store_impl.hpp

@@ -68,7 +68,7 @@ void CollapsingDenseStore<Derived, Allocator>::serialize(std::ostream& os) const
 
 template<class Derived, typename Allocator>
 Derived CollapsingDenseStore<Derived, Allocator>::deserialize(std::istream& is) {
-  Derived store(read<int>(is));
+  Derived store(read<size_type>(is));
 
   if (is.peek() == std::istream::traits_type::eof()) {
     return store;
@@ -81,22 +81,14 @@ Derived CollapsingDenseStore<Derived, Allocator>::deserialize(std::istream& is)
 
 template<class Derived, typename Allocator>
 int CollapsingDenseStore<Derived, Allocator>::get_serialized_size_bytes() const {
-  int size_bytes = 0;
-  size_bytes += sizeof(max_num_bins);
-  size_bytes += sizeof(is_collapsed);
-  size_bytes += sizeof(this->min_index);
-  size_bytes += sizeof(this->max_index);
-  size_bytes += sizeof(size_type);
-
-  size_type non_empty_bins = 0;
-  for (const double& count : this->bins) {
-    non_empty_bins += (count > 0.0);
+  // Header written by serialize(): max_num_bins always present
+  int size_bytes = static_cast<int>(sizeof(max_num_bins));
+  if (this->is_empty()) {
+    return size_bytes;
   }
-
-  size_bytes += sizeof(non_empty_bins);
-  size_bytes += non_empty_bins * sizeof(size_type);
-  size_bytes += non_empty_bins * sizeof(double);
-
+  // is_collapsed flag, then the common section (range + bins)
+  size_bytes += static_cast<int>(sizeof(is_collapsed));
+  size_bytes += this->get_serialized_size_bytes_common();
   return size_bytes;
 }
 

ddsketch/include/ddsketch.hpp

@@ -30,8 +30,9 @@
 
 namespace datasketches {
 
+
 template<store_concept Store, class Mapping>
-class DDSketch {
+class  DDSketch {
 public:
   // TODO
   // DDSketch() = default;
@@ -66,6 +67,8 @@ class DDSketch {
   void serialize(std::ostream& os) const;
   static DDSketch deserialize(std::istream& is);
   int get_serialized_size_bytes() const;
+
+  bool operator==(const DDSketch<Store, Mapping>& other) const;
 protected:
   Store positive_store;
   Store negative_store;
@@ -84,11 +87,19 @@ class DDSketch {
   template<store_concept OtherStore>
   void check_mergeability(const DDSketch<OtherStore, Mapping>& other) const;
   double get_quantile(const double& rank, const double& count) const;
+
+  enum Flag : uint8_t {
+    POSITIVE_STORE,
+    NEGATIVE_STORE,
+    INDEX_MAPPING,
+    ZERO_COUNT,
+  };
 };
 
 // CTA (class template argument deduction) deduction guides (so you can write `ddsketch sketch(s1);`)
 template<store_concept Store, class Mapping>
-DDSketch(Store, Mapping) -> DDSketch<Store, Mapping>;
+DDSketch(Store, Mapping) -> DDSketch<Store, Mapping>;\
+
 
 } /* namespace datasketches */
 

ddsketch/include/ddsketch_impl.hpp

@@ -198,17 +198,50 @@ double DDSketch<Store, Mapping>::get_quantile(const double& rank, const double&
 template<store_concept Store, class Mapping>
 void DDSketch<Store, Mapping>::serialize(std::ostream& os) const {
   index_mapping.serialize(os);
-  if (zero_count > 0.0) {
-    write(os, zero_count);
-  }
 
+  write(os, zero_count);
+
+
+  auto val = positive_store.get_serialized_size_bytes();
+  write(os, positive_store.get_serialized_size_bytes());
   positive_store.serialize(os);
+
+  val = negative_store.get_serialized_size_bytes();
+  write(os, negative_store.get_serialized_size_bytes());
   negative_store.serialize(os);
 }
 
 template<store_concept Store, class Mapping>
 DDSketch<Store, Mapping> DDSketch<Store, Mapping>::deserialize(std::istream &is) {
+  Mapping deserialized_index_mapping = Mapping::deserialize(is);
+  const auto deserialized_zero_count = read<double>(is);
+
+  const auto positive_store_serialized_size  = read<int>(is);
 
+  std::string pos_buf(positive_store_serialized_size, '\0');
+  is.read(pos_buf.data(), pos_buf.size());
+  std::stringstream pos_stream(pos_buf);
+  Store deserialized_positive_store = Store::deserialize(pos_stream);
+
+  const auto negative_store_serialized_size  = read<int>(is);
+  std::string neg_buf(negative_store_serialized_size, '\0');
+  is.read(neg_buf.data(), neg_buf.size());
+  std::stringstream neg_stream(neg_buf);
+  Store deserialized_negative_store = Store::deserialize(neg_stream);
+
+  DDSketch<Store, Mapping> ddsketch(deserialized_positive_store, deserialized_negative_store, deserialized_index_mapping);
+  ddsketch.zero_count = deserialized_zero_count;
+  return ddsketch;
+}
+
+template<store_concept Store, class Mapping>
+bool DDSketch<Store, Mapping>::operator==(const DDSketch<Store, Mapping>& other) const {
+  return positive_store == other.positive_store &&
+    negative_store == other.negative_store &&
+      index_mapping == other.index_mapping &&
+        zero_count == other.zero_count &&
+          min_indexed_value == other.min_indexed_value &&
+            max_indexed_value == other.max_indexed_value;
 }
 
 }

ddsketch/include/dense_store.hpp

@@ -136,6 +136,7 @@ class DenseStore {
 
   void serialize_common(std::ostream& os) const;
   static void deserialize_common(Derived& store, std::istream& is);
+  int get_serialized_size_bytes_common() const;
 
   Derived& derived();
   const Derived& derived() const;

ddsketch/include/dense_store_impl.hpp

@@ -374,7 +374,7 @@ void DenseStore<Derived, Allocator>::serialize_common(std::ostream& os) const {
 
   size_type non_empty_bins = 0;
   for (const double& count : bins) {
-    non_empty_bins += (count > 0.0);
+    non_empty_bins += (count > 1e-16);
   }
   write(os, non_empty_bins);
 
@@ -403,9 +403,46 @@ void DenseStore<Derived, Allocator>::deserialize_common(Derived& store, std::ist
   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);
+    store.bins[index-store.offset] = count;
   }
 }
+
+template<class Derived, typename Allocator>
+int DenseStore<Derived, Allocator>::get_serialized_size_bytes_common() const {
+  if (is_empty()) {
+    return 0;
+  }
+
+
+  // Keep the running total in size_t, cast to int at the end (the public API returns int)
+  size_t size_bytes = 0;
+
+  // Range metadata written by serialize_common
+  size_bytes += sizeof(this->min_index); // min_index
+  size_bytes += sizeof(this->max_index); // max_index
+  size_bytes += sizeof(this->offset);    // offset
+
+  // `serialize_common` writes the number of bins (the full allocated length)
+  size_type num_bins = static_cast<size_type>(this->bins.size());
+  (void)num_bins; // silence unused warning in templates
+  size_bytes += sizeof(num_bins);
+
+  // Count non-empty bins exactly as in serialize_common (threshold 1e-16)
+  size_type non_empty_bins = 0;
+  for (const double& count : this->bins) {
+    non_empty_bins += (count > 1e-16);
+  }
+
+  // It writes the non_empty_bins counter itself
+  size_bytes += sizeof(non_empty_bins);
+
+  // For each non-empty bin, serialize_common writes: index (int) + count (double)
+  size_bytes += static_cast<size_t>(non_empty_bins) * sizeof(int);
+  size_bytes += static_cast<size_t>(non_empty_bins) * sizeof(double);
+
+  // Final cast matches the serialized-size field type used elsewhere
+  return static_cast<int>(size_bytes);
+}
 }
 
 #endif //DENSE_STORE_IMPL_HPP

ddsketch/include/unbounded_size_dense_store.hpp

@@ -44,6 +44,7 @@ class UnboundedSizeDenseStore: public DenseStore<UnboundedSizeDenseStore<Allocat
   void serialize(std::ostream& os) const;
   static UnboundedSizeDenseStore deserialize(std::istream& is);
   int get_serialized_size_bytes() const;
+
 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

@@ -97,21 +97,7 @@ UnboundedSizeDenseStore<Allocator> UnboundedSizeDenseStore<Allocator>::deseriali
 
 template<typename Allocator>
 int UnboundedSizeDenseStore<Allocator>::get_serialized_size_bytes() const {
-  int size_bytes = 0;
-  size_bytes += sizeof(this->min_index);
-  size_bytes += sizeof(this->max_index);
-  size_bytes += sizeof(size_type);
-
-  size_type non_empty_bins = 0;
-  for (const double& count : this->bins) {
-    non_empty_bins += (count > 0.0);
-  }
-
-  size_bytes += sizeof(non_empty_bins);
-  size_bytes += non_empty_bins * sizeof(size_type);
-  size_bytes += non_empty_bins * sizeof(double);
-
-  return size_bytes;
+  return this->get_serialized_size_bytes_common();
 }
 
 

ddsketch/test/DDSketchTest.cpp

@@ -516,6 +516,56 @@ TEMPLATE_TEST_CASE("DDSketch merging random test", "[ddsketch][random]",
   }
 }
 
+TEMPLATE_TEST_CASE("DDSketch serialize - deserialize test", "[ddsketch][random]",
+  // DDSketchUnboundedStoreTestCase,
+  DDSketchCollapsingHighestStoreTestCase<4096>
+  // DDSketchCollapsingLowestStoreTestCase<4096>
+  // DDSketchSparseStoreTestCase
+) {
+  auto positive_store = *TestType::first_type::new_store();
+  auto negative_store = *TestType::first_type::new_store();
+  using StoreType = decltype(positive_store);
+  using MappingType = TestType::second_type;
+  constexpr double relative_accuracy = 0.01;
+  constexpr int num_tests = 100;
+  constexpr int max_num_values = 1000;
+
+  DDSketch<StoreType, MappingType> sketch(positive_store, negative_store, MappingType(relative_accuracy));
+  std::random_device rd;
+  std::mt19937_64 rng(rd());
+  std::uniform_int_distribution<int> size_dist(0, max_num_values - 1);
+  std::uniform_real_distribution<double> value_dist(-1000.0, 1000.0);
+
+  std::stringstream ss;
+  sketch.serialize(ss);
+  DDSketch<StoreType, MappingType> deserialized_empty_sketch = DDSketch<StoreType, MappingType>::deserialize(ss);
+  REQUIRE(sketch.is_empty());
+  REQUIRE(deserialized_empty_sketch.is_empty());
+  REQUIRE(ss.peek() == std::istream::traits_type::eof());
+  REQUIRE(sketch == deserialized_empty_sketch);
+  ss.clear();
+
+  for (int i = 0; i < num_tests; ++i) {
+    std::vector<double> values;
+    int num_values = size_dist(rng);
+
+    for (int j = 0; j < num_values; ++j) {
+      sketch.update(value_dist(rng));
+    }
+
+    sketch.serialize(ss);
+    auto deserialized_sketch = DDSketch<StoreType, MappingType>::deserialize(ss);
+    REQUIRE_FALSE(sketch.is_empty());
+    REQUIRE_FALSE(deserialized_sketch.is_empty());
+    REQUIRE(ss.peek() == std::istream::traits_type::eof());
+    REQUIRE(sketch == deserialized_sketch);
+    ss.clear();
+
+  }
+
+
+}
+
 TEST_CASE("quantile", "[ddsketch]") {
   std::random_device rd{};
   std::mt19937_64 gen{rd()};

ddsketch/test/StoreTest.cpp

@@ -524,7 +524,7 @@ TEMPLATE_TEST_CASE("test cross merge", "[storetest]",
   }
 }
 
-TEMPLATE_TEST_CASE("dense store serialization test", "[serialization]",
+TEMPLATE_TEST_CASE("test store serialize - deserialize", "[serialization]",
   CollapsingLowestStoreTestCase<8>,
   CollapsingLowestStoreTestCase<128>,
   CollapsingLowestStoreTestCase<1024>,