frequent_items_sketch: add reset(), decide emptiness by preamble longs

fi/include/frequent_items_sketch.hpp

@@ -116,6 +116,15 @@ class frequent_items_sketch {
   void merge(frequent_items_sketch&& other);
 
   /**
+   * Resets this sketch to the empty state, as if newly constructed.
+   * The maximum map size, equality operator and allocator are retained.
+   * The internal hash map restarts at its minimum size.
+   */
+  void reset();
+
+  /**
+   * A sketch is empty if it has not been updated with any positive weight.
+   * A non-empty sketch may retain no items if a purge removed all of them.
    * @return true if this sketch is empty
    */
   bool is_empty() const;
@@ -311,13 +320,16 @@ class frequent_items_sketch {
   static const uint8_t PREAMBLE_LONGS_EMPTY = 1;
   static const uint8_t PREAMBLE_LONGS_NONEMPTY = 4;
   static constexpr double EPSILON_FACTOR = 3.5;
-  // due to a mistake different bits were used in C++ and Java to indicate empty sketch
-  // therefore both are set and checked for compatibility with historical binary format
+  // Emptiness of a serialized image is determined by preamble longs (1 for empty, 4 otherwise).
+  // The flags byte is only cross-checked against it. Due to a mistake different bits were used
+  // in C++ and Java to indicate an empty sketch, therefore both are set for compatibility with
+  // the historical binary format, and either one is accepted on read. No other flag bits are defined.
   enum flags { IS_EMPTY_1 = 0, IS_EMPTY_2 = 2 };
   W total_weight;
   W offset;
   reverse_purge_hash_map<T, W, H, E, A> map;
-  static void check_preamble_longs(uint8_t preamble_longs, bool is_empty);
+  static void check_preamble_longs(uint8_t preamble_longs, uint8_t flags_byte);
+  static void check_total_weight(W total_weight);
   static void check_serial_version(uint8_t serial_version);
   static void check_family_id(uint8_t family_id);
   static void check_size(uint8_t lg_cur_size, uint8_t lg_max_size);

fi/include/frequent_items_sketch_impl.hpp

@@ -86,6 +86,13 @@ void frequent_items_sketch<T, W, H, E, A>::merge(frequent_items_sketch&& other)
   total_weight = merged_total_weight;
 }
 
+template<typename T, typename W, typename H, typename E, typename A>
+void frequent_items_sketch<T, W, H, E, A>::reset() {
+  map = reverse_purge_hash_map<T, W, H, E, A>(LG_MIN_MAP_SIZE, map.get_lg_max_size(), map.get_equal(), map.get_allocator());
+  total_weight = 0;
+  offset = 0;
+}
+
 template<typename T, typename W, typename H, typename E, typename A>
 bool frequent_items_sketch<T, W, H, E, A>::is_empty() const {
   // a purge may clear all counters while offset and total_weight remain non-zero;
@@ -302,9 +309,8 @@ frequent_items_sketch<T, W, H, E, A> frequent_items_sketch<T, W, H, E, A>::deser
   const auto flags_byte = read<uint8_t>(is);
   read<uint16_t>(is); // unused
 
-  const bool is_empty = (flags_byte & (1 << flags::IS_EMPTY_1)) | (flags_byte & (1 << flags::IS_EMPTY_2));
-
-  check_preamble_longs(preamble_longs, is_empty);
+  check_preamble_longs(preamble_longs, flags_byte);
+  const bool is_empty = preamble_longs == PREAMBLE_LONGS_EMPTY;
   check_serial_version(serial_version);
   check_family_id(family_id);
   check_size(lg_cur_size, lg_max_size);
@@ -315,6 +321,7 @@ frequent_items_sketch<T, W, H, E, A> frequent_items_sketch<T, W, H, E, A>::deser
     read<uint32_t>(is); // unused
     const auto total_weight = read<W>(is);
     const auto offset = read<W>(is);
+    check_total_weight(total_weight);
 
     // batch deserialization with intermediate array of items and weights
     using AllocW = typename std::allocator_traits<A>::template rebind_alloc<W>;
@@ -355,9 +362,8 @@ frequent_items_sketch<T, W, H, E, A> frequent_items_sketch<T, W, H, E, A>::deser
   ptr += copy_from_mem(ptr, flags_byte);
   ptr += sizeof(uint16_t); // unused
 
-  const bool is_empty = (flags_byte & (1 << flags::IS_EMPTY_1)) | (flags_byte & (1 << flags::IS_EMPTY_2));
-
-  check_preamble_longs(preamble_longs, is_empty);
+  check_preamble_longs(preamble_longs, flags_byte);
+  const bool is_empty = preamble_longs == PREAMBLE_LONGS_EMPTY;
   check_serial_version(serial_version);
   check_family_id(family_id);
   check_size(lg_cur_size, lg_max_size);
@@ -372,6 +378,7 @@ frequent_items_sketch<T, W, H, E, A> frequent_items_sketch<T, W, H, E, A>::deser
     ptr += copy_from_mem(ptr, total_weight);
     W offset;
     ptr += copy_from_mem(ptr, offset);
+    check_total_weight(total_weight);
 
     ensure_minimum_memory(size, ptr - base + (sizeof(W) * num_items));
     // batch deserialization with intermediate array of items and weights
@@ -394,15 +401,23 @@ frequent_items_sketch<T, W, H, E, A> frequent_items_sketch<T, W, H, E, A>::deser
 }
 
 template<typename T, typename W, typename H, typename E, typename A>
-void frequent_items_sketch<T, W, H, E, A>::check_preamble_longs(uint8_t preamble_longs, bool is_empty) {
-  if (is_empty) {
-    if (preamble_longs != PREAMBLE_LONGS_EMPTY) {
-      throw std::invalid_argument("Possible corruption: preamble longs of an empty sketch must be " + std::to_string(PREAMBLE_LONGS_EMPTY) + ": " + std::to_string(preamble_longs));
-    }
-  } else {
-    if (preamble_longs != PREAMBLE_LONGS_NONEMPTY) {
-      throw std::invalid_argument("Possible corruption: preamble longs of an non-empty sketch must be " + std::to_string(PREAMBLE_LONGS_NONEMPTY) + ": " + std::to_string(preamble_longs));
-    }
+void frequent_items_sketch<T, W, H, E, A>::check_preamble_longs(uint8_t preamble_longs, uint8_t flags_byte) {
+  if (preamble_longs != PREAMBLE_LONGS_EMPTY && preamble_longs != PREAMBLE_LONGS_NONEMPTY) {
+    throw std::invalid_argument("Possible corruption: preamble longs must be " + std::to_string(PREAMBLE_LONGS_EMPTY)
+        + " or " + std::to_string(PREAMBLE_LONGS_NONEMPTY) + ": " + std::to_string(preamble_longs));
+  }
+  const bool empty_flag = (flags_byte & ((1 << flags::IS_EMPTY_1) | (1 << flags::IS_EMPTY_2))) != 0;
+  if (empty_flag != (preamble_longs == PREAMBLE_LONGS_EMPTY)) {
+    throw std::invalid_argument("Possible corruption: empty flag does not match preamble longs: flags "
+        + std::to_string(flags_byte) + ", preamble longs " + std::to_string(preamble_longs));
+  }
+}
+
+template<typename T, typename W, typename H, typename E, typename A>
+void frequent_items_sketch<T, W, H, E, A>::check_total_weight(W total_weight) {
+  // written as !(x > 0) to also reject NaN
+  if (!(total_weight > 0)) {
+    throw std::invalid_argument("Possible corruption: total weight of a non-empty sketch must be positive");
   }
 }
 

fi/include/reverse_purge_hash_map.hpp

@@ -62,6 +62,7 @@ class reverse_purge_hash_map {
   uint32_t get_capacity() const;
   uint32_t get_num_active() const;
   const A& get_allocator() const;
+  const E& get_equal() const;
 
   class iterator;
   iterator begin() const;

fi/include/reverse_purge_hash_map_impl.hpp

@@ -197,6 +197,11 @@ const A& reverse_purge_hash_map<K, V, H, E, A>::get_allocator() const {
   return allocator_;
 }
 
+template<typename K, typename V, typename H, typename E, typename A>
+const E& reverse_purge_hash_map<K, V, H, E, A>::get_equal() const {
+  return equal_;
+}
+
 template<typename K, typename V, typename H, typename E, typename A>
 typename reverse_purge_hash_map<K, V, H, E, A>::iterator reverse_purge_hash_map<K, V, H, E, A>::begin() const {
   const uint32_t size = 1 << lg_cur_size_;

fi/test/frequent_items_sketch_serialize_for_java.cpp

@@ -57,6 +57,30 @@ TEST_CASE("frequent strings sketch generate", "[serialize_for_java]") {
   }
 }
 
+// lg_max_map_size=8 -> capacity 192; the 193rd distinct item triggers a purge
+// whose median (1) removes every counter: non-empty with no retained items
+TEST_CASE("frequent longs sketch purged to zero items", "[serialize_for_java]") {
+  frequent_items_sketch<long> sketch(8);
+  for (long i = 1; i <= 193; ++i) sketch.update(i);
+  REQUIRE_FALSE(sketch.is_empty());
+  REQUIRE(sketch.get_num_active_items() == 0);
+  REQUIRE(sketch.get_total_weight() == 193);
+  REQUIRE(sketch.get_maximum_error() == 1);
+  std::ofstream os("frequent_long_purged_cpp.sk", std::ios::binary);
+  sketch.serialize(os);
+}
+
+TEST_CASE("frequent strings sketch purged to zero items", "[serialize_for_java]") {
+  frequent_items_sketch<std::string> sketch(8);
+  for (unsigned i = 1; i <= 193; ++i) sketch.update(std::to_string(i));
+  REQUIRE_FALSE(sketch.is_empty());
+  REQUIRE(sketch.get_num_active_items() == 0);
+  REQUIRE(sketch.get_total_weight() == 193);
+  REQUIRE(sketch.get_maximum_error() == 1);
+  std::ofstream os("frequent_string_purged_cpp.sk", std::ios::binary);
+  sketch.serialize(os);
+}
+
 TEST_CASE("frequent strings sketch ascii", "[serialize_for_java]") {
   frequent_items_sketch<std::string> sketch(6);
   sketch.update("aaaaaaaaaaaaaaaaaaaaaaaaaaaaa", 1);

fi/test/frequent_items_sketch_test.cpp

@@ -157,6 +157,96 @@ TEST_CASE("frequent items: purge clearing all counters is not empty", "[frequent
   REQUIRE(sketch3.get_maximum_error() == 1);
 }
 
+TEST_CASE("frequent items: purge clearing all counters serialized form", "[frequent_items_sketch]") {
+  frequent_items_sketch<uint64_t> sketch(8);
+  for (uint64_t i = 0; i < 193; ++i) sketch.update(i);
+  REQUIRE(sketch.get_num_active_items() == 0);
+
+  // full preamble with no items
+  auto bytes = sketch.serialize();
+  REQUIRE(bytes.size() == 32);
+  REQUIRE(bytes[0] == 4); // preamble longs
+  REQUIRE(bytes[5] == 0); // flags
+
+  std::stringstream s(std::ios::in | std::ios::out | std::ios::binary);
+  sketch.serialize(s);
+  auto sketch2 = frequent_items_sketch<uint64_t>::deserialize(s);
+  REQUIRE_FALSE(sketch2.is_empty());
+  REQUIRE(sketch2.get_num_active_items() == 0);
+  REQUIRE(sketch2.get_total_weight() == 193);
+  REQUIRE(sketch2.get_maximum_error() == 1);
+}
+
+TEST_CASE("frequent items: reset", "[frequent_items_sketch]") {
+  frequent_items_sketch<uint64_t> sketch(8);
+  for (uint64_t i = 0; i < 1000; ++i) sketch.update(i % 300, i % 7 + 1);
+  REQUIRE_FALSE(sketch.is_empty());
+  REQUIRE(sketch.get_maximum_error() > 0);
+  const double epsilon = sketch.get_epsilon();
+
+  sketch.reset();
+  REQUIRE(sketch.is_empty());
+  REQUIRE(sketch.get_num_active_items() == 0);
+  REQUIRE(sketch.get_total_weight() == 0);
+  REQUIRE(sketch.get_maximum_error() == 0);
+  REQUIRE(sketch.get_epsilon() == epsilon); // max map size retained
+  REQUIRE(sketch.get_serialized_size_bytes() == 8);
+
+  // same behavior as a newly constructed sketch
+  frequent_items_sketch<uint64_t> fresh(8);
+  for (uint64_t i = 0; i < 193; ++i) {
+    sketch.update(i);
+    fresh.update(i);
+  }
+  REQUIRE(sketch.serialize() == fresh.serialize());
+}
+
+TEST_CASE("frequent items: empty image with either legacy empty flag", "[frequent_items_sketch]") {
+  frequent_items_sketch<uint64_t> sketch(8);
+  auto bytes = sketch.serialize();
+  REQUIRE(bytes.size() == 8);
+  REQUIRE(bytes[5] == 5); // both empty bits written
+  for (uint8_t flags: {1, 4, 5}) {
+    bytes[5] = flags;
+    auto sketch2 = frequent_items_sketch<uint64_t>::deserialize(bytes.data(), bytes.size());
+    REQUIRE(sketch2.is_empty());
+  }
+}
+
+TEST_CASE("frequent items: corrupt preamble", "[frequent_items_sketch]") {
+  frequent_items_sketch<uint64_t> empty_sketch(8);
+  frequent_items_sketch<uint64_t> sketch(8);
+  sketch.update(1);
+
+  SECTION("invalid preamble longs") {
+    auto bytes = sketch.serialize();
+    bytes[0] = 2;
+    REQUIRE_THROWS_AS(frequent_items_sketch<uint64_t>::deserialize(bytes.data(), bytes.size()), std::invalid_argument);
+  }
+  SECTION("empty preamble longs, not empty flag") {
+    auto bytes = empty_sketch.serialize();
+    bytes[5] = 0;
+    REQUIRE_THROWS_AS(frequent_items_sketch<uint64_t>::deserialize(bytes.data(), bytes.size()), std::invalid_argument);
+  }
+  SECTION("full preamble longs, empty flag") {
+    auto bytes = sketch.serialize();
+    bytes[5] = 5;
+    REQUIRE_THROWS_AS(frequent_items_sketch<uint64_t>::deserialize(bytes.data(), bytes.size()), std::invalid_argument);
+  }
+  SECTION("full preamble longs, zero total weight, bytes") {
+    auto bytes = sketch.serialize();
+    for (size_t i = 16; i < 24; ++i) bytes[i] = 0;
+    REQUIRE_THROWS_AS(frequent_items_sketch<uint64_t>::deserialize(bytes.data(), bytes.size()), std::invalid_argument);
+  }
+  SECTION("full preamble longs, zero total weight, stream") {
+    auto bytes = sketch.serialize();
+    for (size_t i = 16; i < 24; ++i) bytes[i] = 0;
+    std::stringstream s(std::ios::in | std::ios::out | std::ios::binary);
+    s.write(reinterpret_cast<const char*>(bytes.data()), bytes.size());
+    REQUIRE_THROWS_AS(frequent_items_sketch<uint64_t>::deserialize(s), std::invalid_argument);
+  }
+}
+
 TEST_CASE("frequent items: merge exact mode", "[frequent_items_sketch]") {
   frequent_items_sketch<int> sketch1(3);
   sketch1.update(1);