❌ Enhanced deleteByKey functionality

src/PTable.hpp

@@ -20,6 +20,7 @@
 #ifndef PTable_hpp_
 #define PTable_hpp_
 
+#include <algorithm>
 #include <bitset>
 #include "mpark/variant.hpp"
 
@@ -54,8 +55,8 @@ using ColumnRangeMap = std::unordered_map<uint16_t, std::pair<IntDoubleString, I
 template<class Tuple, typename KeyType>
 class PTable {
   static const auto BRANCHKEYS = ((TARGET_INDEX_NODE_SIZE - 28) / (sizeof(KeyType) + 24)) & ~1;
-  static const auto
-    LEAFKEYS = ((TARGET_INDEX_NODE_SIZE - 36) / (sizeof(KeyType) + sizeof(PTuple<Tuple, KeyType>))) & ~1;
+  static const auto LEAFKEYS = ((TARGET_INDEX_NODE_SIZE - 36) /
+    (sizeof(KeyType) + sizeof(PTuple<Tuple, KeyType>))) & ~1;
 
   using ColumnIntMap = std::map<uint16_t, uint16_t>;
   using IndexType = PBPTree<KeyType, PTuple<Tuple, KeyType>, BRANCHKEYS, LEAFKEYS>;
@@ -404,13 +405,32 @@ class PTable {
     std::size_t nres;
     auto pop = pool_by_vptr(this);
     transaction::exec_tx(pop, [&] {
-      //TODO: Also delete data?
-      //delete_persistent<PTuple<Tuple>>(getByKey(key));
-      nres = static_cast<std::size_t>(root->index->erase(key));
-      //TODO: Delete from Historgram
+      auto ptp = getByKey(key);
+
+      // indicate the deleted tuple
+      const auto &dataPos = reinterpret_cast<const uint16_t &>(ptp.getNode()->block.get_ro()[gDataOffsetPos]);
+      uint16_t pos = 0u;
+      switch (root->tInfo->columnInfo(0).getType()) {
+        case Int_Type: pos = (ptp.getOffsetAt(0) - ptp.getNode()->block.get_ro()[dataPos]) / sizeof(int) + 1;
+          break;
+        case Double_Type: pos = (ptp.getOffsetAt(0) - ptp.getNode()->block.get_ro()[dataPos]) / sizeof(double) + 1;
+          break;
+        case String_Type: pos = (ptp.getOffsetAt(0) - ptp.getNode()->block.get_ro()[dataPos]) / gOffsetSize + 1;
+          break;
+        default:;
+      }
+      ptp.getNode()->deleted.get_rw().emplace_back(pos);
+
+      //TODO: Delete from histogram?
 //      const auto &ptp = getByKey(key);
 //      const auto xtr = static_cast<uint32_t>(getBDCCFromTuple(*ptp.createTuple()).to_ulong());
+//      ptp.getNode()->histogram[xtr]--;
 
+      // delete from index
+      nres = static_cast<std::size_t>(root->index->erase(key));
+
+      // free space of PTuple
+//      delete_persistent<PTuple<Tuple, KeyType>>(ptp);
     });
     return nres;
   }
@@ -459,7 +479,8 @@ class PTable {
     auto cnt = 0ul;
     auto targetNode = root->dataNodes;
     do {
-      cnt += reinterpret_cast<const uint32_t &>(targetNode->block.get_ro()[gCountPos]);
+      cnt += reinterpret_cast<const uint32_t &>(targetNode->block.get_ro()[gCountPos])
+        - targetNode->deleted.get_ro().size();
       targetNode = targetNode->next;
     } while (targetNode != nullptr);
     return cnt;
@@ -934,6 +955,11 @@ class PTable {
     // TODO: what about deleted records? Extra Bitmap?
     const auto &cnt = reinterpret_cast<const uint16_t &>(block0[gCountPos]);
     for (auto tuplePos = 0u; tuplePos < cnt; tuplePos++) {
+      // Skip if marked as deleted
+      auto &deleted = oldNode->deleted.get_ro();
+      if (std::find(deleted.begin(), deleted.end(), tuplePos) != deleted.end())
+        continue;
+
       std::array<uint16_t, PTuple<Tuple, KeyType>::NUM_ATTRIBUTES> pTupleOffsets;
       const auto &key = oldNode->keys.get_ro()[tuplePos];
 

src/core/DataNode.hpp

@@ -81,6 +81,7 @@ using BDCC_Block = typename std::array<uint8_t, gBlockSize>;
 
 template<typename KeyType>
 struct DataNode {
+  using DeletedVector = std::vector<uint16_t, allocator<uint16_t>>;
   using KeyVector = std::array<KeyType, 8192>;
   using HistogramType = std::unordered_map<uint32_t,
                                           std::size_t,
@@ -89,12 +90,12 @@ struct DataNode {
                                           allocator<uint32_t>>;
 
   DataNode() : next(nullptr) {}
-
   DataNode(BDCC_Block _block) : next(nullptr), block(_block) {}
 
   persistent_ptr<struct DataNode> next;
   p<BDCC_Block> block;
   p<KeyVector> keys;
+  p<DeletedVector> deleted;
   p<HistogramType> histogram;
 
   const uint32_t calcAverageBDCC() const {

src/test/PTableTest.cpp

@@ -84,7 +84,7 @@ TEST_CASE("Storing tuples in PTable", "[PTable]") {
       c += pTable->deleteByKey(i + 1);
     }
     REQUIRE(c == 5);
-    //TODO: REQUIRE(pTable->count() == 5);
+    REQUIRE(pTable->count() == 5);
 
     for (auto i = 5u; i < 10; i++) {
       REQUIRE(get<0>(pTable->getByKey(i + 1)) == i + 1);