Record format generalization

Currently, tombstone counting is bugged. But the rest of it appears to
be working.

2 files changed, 86 insertions, 26 deletions

diff --git a/include/util/Cursor.h b/include/util/Cursor.h
index 2339800..2609ae5 100644
--- a/include/util/Cursor.h
+++ b/include/util/Cursor.h
@@ -14,10 +14,10 @@
 #include "io/PagedFile.h"
 
 namespace de {
-template<typename K, typename V, typename W=void>
+template<typename R>
 struct Cursor {
-    Record<K,V,W> *ptr;
-    const Record<K,V,W> *end;
+    R *ptr;
+    R *end;
     size_t cur_rec_idx;
     size_t rec_cnt;
 
@@ -36,8 +36,8 @@ struct Cursor {
  * be updated to be equal to end, and false will be returned. Iterators will
  * not be closed.
  */
-template<typename K, typename V, typename W>
-inline static bool advance_cursor(Cursor<K,V,W> &cur, PagedFileIterator *iter = nullptr) {
+template<typename R>
+inline static bool advance_cursor(Cursor<R> &cur, PagedFileIterator *iter = nullptr) {
     cur.ptr++;
     cur.cur_rec_idx++;
 
@@ -45,8 +45,8 @@ inline static bool advance_cursor(Cursor<K,V,W> &cur, PagedFileIterator *iter =
 
     if (cur.ptr >= cur.end) {
         if (iter && iter->next()) {
-            cur.ptr = (Record<K,V,W>*)iter->get_item();
-            cur.end = cur.ptr + (PAGE_SIZE / sizeof(Record<K,V,W>));
+            cur.ptr = (R*)iter->get_item();
+            cur.end = cur.ptr + (PAGE_SIZE / sizeof(R));
             return true;
         }
 
@@ -62,14 +62,14 @@ inline static bool advance_cursor(Cursor<K,V,W> &cur, PagedFileIterator *iter =
  *   This allows for "peaking" at the next largest element after the current 
  *   largest is processed.
  */
-template <typename K, typename V, typename W>
-inline static Cursor<K,V,W> *get_next(std::vector<Cursor<K,V,W>> &cursors, Cursor<K,V,W> *current=nullptr) {
-    const Record<K,V,W> *min_rec = nullptr;
-    Cursor<K,V,W> *result = nullptr;
+template <typename R>
+inline static Cursor<R> *get_next(std::vector<Cursor<R>> &cursors, Cursor<R> *current=nullptr) {
+    const R *min_rec = nullptr;
+    Cursor<R> *result = nullptr;
     for (size_t i=0; i< cursors.size(); i++) {
-        if (cursors[i] == (Cursor<K,V,W>) {0} ) continue;
+        if (cursors[i] == (Cursor<R>) {0} ) continue;
 
-        const Record<K,V,W> *rec = (&cursors[i] == current) ? cursors[i].ptr + 1 : cursors[i].ptr;
+        const R *rec = (&cursors[i] == current) ? cursors[i].ptr + 1 : cursors[i].ptr;
         if (rec >= cursors[i].end) continue;
 
         if (min_rec == nullptr) {
diff --git a/include/util/Record.h b/include/util/Record.h
index 687e745..ce101f4 100644
--- a/include/util/Record.h
+++ b/include/util/Record.h
@@ -10,50 +10,110 @@
 #pragma once
 
 #include <cstring>
+#include <concepts>
 
 #include "util/base.h"
 
 namespace de {
 
-template <typename K, typename V, typename W=void>
+template<typename R>
+concept RecordInterface = requires(R r, R s) {
+    r.key;
+    r.value;
+    r.header;
+
+    {r.is_tombstone()} -> std::convertible_to<bool>;
+    {r.is_deleted()} -> std::convertible_to<bool>;
+    r.set_delete();
+    r.set_tombstone(std::declval<bool>);
+    { r < s } ->std::convertible_to<bool>;
+    { r == s } ->std::convertible_to<bool>;
+    { r.header < s.header } -> std::convertible_to<bool>;
+};
+
+template <typename R>
+concept WeightedRecordInterface = RecordInterface<R> && requires(R r) {
+    {r.weight} -> std::convertible_to<double>;
+};
+
+template <typename K, typename V>
 struct Record {
     K key;
     V value;
-    typename std::conditional<!std::is_same<W, void>::value, W, std::false_type>::type weight;
-    uint32_t header;
+    uint32_t header = 0;
+
+    inline void set_delete() {
+        header |= 2;
+    }
+
+    inline bool is_deleted() const {
+        return header & 2;
+    }
+
+    inline void set_tombstone(bool val=true) {
+        if (val) {
+            header |= val;
+        } else {
+            header &= 0;
+        }
+    }
+
+    inline bool is_tombstone() const {
+        return header & 1;
+    }
+
+    inline bool operator<(const Record& other) const {
+        return key < other.key || (key == other.key && value < other.value);
+    }
 
-    inline bool match(K k, V v, bool is_tombstone) const {
-        return (key == k) && (value == v) && ((header & 1) == is_tombstone);
+    inline bool operator==(const Record& other) const {
+        return key == other.key && value == other.value;
     }
+};
+
+template <typename K, typename V, typename W>
+struct WeightedRecord {
+    K key;
+    V value;
+    W weight = 1;
+    uint32_t header = 0;
 
-    inline void set_delete_status() {
+    inline void set_delete() {
         header |= 2;
     }
 
-    inline bool get_delete_status() const {
+    inline bool is_deleted() const {
         return header & 2;
     }
 
+    inline void set_tombstone(bool val=true) {
+        if (val) {
+            header |= val;
+        } else {
+            header &= 0;
+        }
+    }
+
     inline bool is_tombstone() const {
         return header & 1;
     }
 
-    inline int match(const Record* other) const {
+    inline int match(const WeightedRecord* other) const {
         return key == other->key && value == other->value;
     }
 
-    inline bool operator<(const Record& other) const {
+    inline bool operator<(const WeightedRecord& other) const {
         return key < other.key || (key == other.key && value < other.value);
     }
 
-    inline bool lt(const K& k, const V& v) const {
-        return key < k || (key == k && value < v);
+    inline bool operator==(const WeightedRecord& other) const {
+        return key == other.key && value == other.value;
     }
 };
 
 
-template <typename K, typename V, typename W=void>
-static bool memtable_record_cmp(const Record<K, V, W>& a, const Record<K, V, W>& b) {
+template <RecordInterface R>
+static bool memtable_record_cmp(const R& a, const R& b) {
     return (a.key < b.key) || (a.key == b.key && a.value < b.value)
         || (a.key == b.key && a.value == b.value && a.header < b.header);
 }