test cases for column sketches

Author: Gillgamesh

CMakeLists.txt

@@ -37,8 +37,8 @@ if(COMPILER_SUPPORTS_MARCH_NATIVE)
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -march=native")
 endif()
 
-#add_compile_options(-fsanitize=address)
-#add_link_options(-fsanitize=address)
+# add_compile_options(-fsanitize=address)
+# add_link_options(-fsanitize=address)
 #add_compile_options(-fsanitize=undefined)
 #add_link_options(-fsanitize=undefined)
 

include/bucket.h

@@ -23,6 +23,10 @@ struct Bucket {
   bool operator!=(const Bucket &rhs) const {
     return alpha != rhs.alpha || gamma != rhs.gamma;
   };
+  friend std::ostream &operator<<(std::ostream &os, const Bucket &b) {
+    os << "(a: " << b.alpha << " g: " << b.gamma << ")";
+    return os;
+  }
 };
 #pragma pack(pop)
 

include/sketch/sketch_columns.h

@@ -12,38 +12,55 @@
 */
 class FixedSizeSketchColumn {
 private:
-  static uint64_t seed;
 
-  Bucket *buckets;
+  std::unique_ptr<Bucket[]> buckets;
   Bucket deterministic_bucket = {0, 0};
+  uint16_t col_idx; // determines column seeding
   uint8_t capacity;
-  uint8_t col_idx; // determines column seeding
 public:
+  static uint64_t seed;
   static void set_seed(uint64_t new_seed) {
     seed = new_seed;
   };  
   static const uint64_t get_seed() {
     return seed;
   };
 
-  FixedSizeSketchColumn(uint8_t capacity, uint8_t col_idx);
+  FixedSizeSketchColumn(uint8_t capacity, uint16_t col_idx);
+  FixedSizeSketchColumn(const FixedSizeSketchColumn &other);
   ~FixedSizeSketchColumn();
   SketchSample<vec_t> sample() const;
   void clear();
   void update(const vec_t update);
   void merge(FixedSizeSketchColumn &other);
   uint8_t get_depth() const;
   void serialize(std::ostream &binary_out) const;
+  friend std::ostream& operator<<(std::ostream &os, const FixedSizeSketchColumn &sketch) {
+    os << "FixedSizeSketchColumn: " << std::endl;
+    os << "Capacity: " << (int)sketch.capacity << std::endl;
+    os << "Column Index: " << (int)sketch.col_idx << std::endl;
+    os << "Deterministic Bucket: " << sketch.deterministic_bucket << std::endl;
+    for (size_t i = 0; i < sketch.capacity; ++i) {
+      os << "Bucket[" << i << "]: " << sketch.buckets[i] << std::endl;
+    }
+    return os;
+  }
 };
 
-FixedSizeSketchColumn::FixedSizeSketchColumn(uint8_t capacity, uint8_t col_idx) :
+FixedSizeSketchColumn::FixedSizeSketchColumn(uint8_t capacity, uint16_t col_idx) :
     capacity(capacity), col_idx(col_idx) {
-  buckets = new Bucket[capacity];
-  // std::memset(buckets, 0, capacity * sizeof(Bucket));
+  buckets = std::make_unique<Bucket[]>(capacity);
+  // std::memset(buckets.get(), 0, capacity * sizeof(Bucket));
+}
+
+FixedSizeSketchColumn::FixedSizeSketchColumn(const FixedSizeSketchColumn &other) :
+    capacity(other.capacity), col_idx(other.col_idx), deterministic_bucket(other.deterministic_bucket) {
+  buckets = std::make_unique<Bucket[]>(capacity);
+  std::memcpy(buckets.get(), other.buckets.get(), capacity * sizeof(Bucket));
 }
 
 FixedSizeSketchColumn::~FixedSizeSketchColumn() {
-  delete[] buckets;
+  // delete[] buckets;
 }
 
 uint8_t FixedSizeSketchColumn::get_depth() const {
@@ -56,7 +73,7 @@ uint8_t FixedSizeSketchColumn::get_depth() const {
 
 // TODO - implement actual deserialization
 void FixedSizeSketchColumn::serialize(std::ostream &binary_out) const {
-  binary_out.write((char *) buckets, capacity * sizeof(Bucket));
+  binary_out.write((char *) buckets.get(), capacity * sizeof(Bucket));
   binary_out.write((char *) &deterministic_bucket, sizeof(Bucket));
   binary_out.write((char *) &capacity, sizeof(uint8_t));
   binary_out.write((char *) &col_idx, sizeof(uint8_t));
@@ -75,7 +92,7 @@ SketchSample<vec_t> FixedSizeSketchColumn::sample() const {
 }
 
 void FixedSizeSketchColumn::clear() {
-  std::memset(buckets, 0, capacity * sizeof(Bucket));
+  std::memset(buckets.get(), 0, capacity * sizeof(Bucket));
   deterministic_bucket = {0, 0};
 }
 
@@ -89,6 +106,7 @@ void FixedSizeSketchColumn::merge(FixedSizeSketchColumn &other) {
 void FixedSizeSketchColumn::update(const vec_t update) {
   vec_hash_t checksum = Bucket_Boruvka::get_index_hash(update, seed);
   col_hash_t depth = Bucket_Boruvka::get_index_depth(update, seed, col_idx, capacity);
+  assert(depth < capacity);
   buckets[depth] ^= {update, checksum};
   deterministic_bucket ^= {update, checksum};
 }
@@ -98,16 +116,16 @@ void FixedSizeSketchColumn::update(const vec_t update) {
 class ResizeableSketchColumn {
 private:
   static uint64_t seed;
-
   hwy::AlignedFreeUniquePtr<Bucket[]> aligned_buckets;
   Bucket deterministic_bucket = {0, 0};
+  uint16_t col_idx; // determines column seeding
   uint8_t capacity;
-  uint8_t col_idx; // determines column seeding
 public:
   static void set_seed(uint64_t new_seed) { seed = new_seed; };
   static const uint64_t get_seed() { return seed; };
 
-  ResizeableSketchColumn(uint8_t start_capacity, uint8_t col_idx);
+  ResizeableSketchColumn(uint8_t start_capacity, uint16_t col_idx);
+  ResizeableSketchColumn(const ResizeableSketchColumn &other);
   ~ResizeableSketchColumn();
   SketchSample<vec_t> sample() const;
   void clear();
@@ -119,15 +137,27 @@ class ResizeableSketchColumn {
   void reallocate(uint8_t new_capacity);
 };
 
+uint64_t ResizeableSketchColumn::seed = 0;
+uint64_t FixedSizeSketchColumn::seed = 0;
+
 
-ResizeableSketchColumn::ResizeableSketchColumn(uint8_t start_capacity, uint8_t col_idx) :
+ResizeableSketchColumn::ResizeableSketchColumn(uint8_t start_capacity, uint16_t col_idx) :
     capacity(start_capacity), col_idx(col_idx) {
 
     // auto aligned_memptr = hwy::MakeUniqueAlignedArray<Bucket>(start_capacity);
     aligned_buckets = hwy::AllocateAligned<Bucket>(start_capacity);
     std::memset(aligned_buckets.get(), 0, capacity * sizeof(Bucket));
 }
 
+ResizeableSketchColumn::ResizeableSketchColumn(const ResizeableSketchColumn &other) :
+    capacity(other.capacity), col_idx(other.col_idx), deterministic_bucket(other.deterministic_bucket) {
+  aligned_buckets = hwy::AllocateAligned<Bucket>(capacity);
+  std::memcpy(aligned_buckets.get(), other.aligned_buckets.get(), capacity * sizeof(Bucket));
+}
+
+ResizeableSketchColumn::~ResizeableSketchColumn() {
+}
+
 /*
   Note this DROPS the contents if allocated down too much.
 */
@@ -154,13 +184,29 @@ void ResizeableSketchColumn::serialize(std::ostream &binary_out) const {
   binary_out.write((char *) &col_idx, sizeof(uint8_t));
 }
 
+SketchSample<vec_t> ResizeableSketchColumn::sample() const {
+  if (Bucket_Boruvka::is_empty(deterministic_bucket)) {
+    return {0, ZERO};  // the "first" bucket is deterministic so if all zero then no edges to return
+  }
+  for (size_t i = capacity; i > 0; --i) {
+    if (Bucket_Boruvka::is_good(aligned_buckets[i - 1], seed)) {
+      return {aligned_buckets[i - 1].alpha, GOOD};
+    }
+  }
+  return {0, FAIL};
+}
+
 void ResizeableSketchColumn::update(const vec_t update) {
   vec_hash_t checksum = Bucket_Boruvka::get_index_hash(update, seed);
-  col_hash_t depth = Bucket_Boruvka::get_index_depth(update, seed, col_idx, capacity);
+  // TODO - remove magic number
+  // TODO - get_index_depth needs to be fixed. hashes need to be longer
+  // than 32 bits if we're not using the deep bucket buffer idea.
+  col_hash_t depth = Bucket_Boruvka::get_index_depth(update, seed, col_idx, 32);
   deterministic_bucket ^= {update, checksum};
 
-  if (depth >= capacity) {
-    reallocate(depth + 4);
+  while (depth >= capacity) {
+    // first multple of 4 larger than or equal to depth
+    reallocate(capacity + 4);
   }
   aligned_buckets[depth] ^= {update, checksum};
 }
@@ -187,6 +233,7 @@ uint8_t ResizeableSketchColumn::get_depth() const {
 }
 
 
+
 static_assert(SketchColumnConcept<FixedSizeSketchColumn, vec_t>,
               "FixedSizeSketchColumn does not satisfy SketchColumnConcept");
 

test/sketch_test.cpp

@@ -479,7 +479,62 @@ TEST(SketchTestSuite, TestRawBucketUpdate) {
 }
 
 
-// TEST(SketchColumnTestSuite, TestSketchColumnZero) {
-//   ResizeableSketchColumn::set_seed(get_seed());
-//   ResizeableSketchColumn column(4, 0);
-// }
+TEST(SketchColumnTestSuite, TestSketchColumnSampling) {
+  ResizeableSketchColumn::set_seed(get_seed());
+  // ResizeableSketchColumn::seed = get_seed();
+  ResizeableSketchColumn column(18, 0);
+  column.update(10);
+  auto sample = column.sample();
+  ASSERT_EQ(sample.result, GOOD);
+  ASSERT_EQ(sample.idx, 10);
+  column.update(20);
+  sample = column.sample();
+  ASSERT_NE(sample.result, ZERO);
+  column.update(10);
+  sample = column.sample();
+  ASSERT_EQ(sample.result, GOOD);
+  ASSERT_EQ(sample.idx, 20);
+  column.update(20);
+  sample = column.sample();
+  ASSERT_EQ(sample.result, ZERO);
+}
+
+TEST(SketchColumnTestSuite, TestSketchColumnMerging) {
+  ResizeableSketchColumn::set_seed(get_seed());
+  for (size_t col_idx =0; col_idx < 16; col_idx++) {
+    ResizeableSketchColumn column1(18, col_idx);
+    ResizeableSketchColumn column2(18, col_idx);
+    for (vec_t i = 0; i < (1 << 11); i++) {
+      column1.update(i);
+      column2.update(i + 128);
+    }
+    column1.merge(column2);
+    // at this point, the value should be [0,127] or [1 << 11]
+    auto sample = column1.sample();
+    if (sample.result == GOOD) {
+      // std::cout << "sample.idx: " << sample.idx << std::endl;
+      ASSERT_TRUE(
+        sample.idx < 128 || (
+          sample.idx >= (1 << 11) && sample.idx < (1 << 11) + 128
+        )
+      );
+    }
+  }
+}
+
+TEST(SketchColumnTestSuite, TestSketchColumnMergeMany) {
+  ResizeableSketchColumn::set_seed(get_seed());
+  std::vector<ResizeableSketchColumn> columns(1 << 13, ResizeableSketchColumn(4, 0));
+  for (size_t i = 0; i < columns.size(); i++) {
+    columns[i].update(i);
+  }
+  for (size_t i = 1; i < columns.size(); i++) {
+    columns[0].merge(columns[i]);
+  }
+  for (size_t i = 1; i < columns.size(); i++) {
+    columns[0].update(i);
+  }
+  auto sample = columns[0].sample();
+  ASSERT_EQ(sample.result, GOOD);
+  ASSERT_EQ(sample.idx, 0);
+}