Fixing non-unique key insert problem

Author: Ziqi Wang

src/include/index/bwtree.h

@@ -2986,7 +2986,8 @@ class BwTree : public BwTreeBase {
    * If both are nullptr then we just traverse and do not do anything
    */
   const KeyValuePair *Traverse(Context *context_p, const ValueType *value_p,
-                               std::pair<int, bool> *index_pair_p) {
+                               std::pair<int, bool> *index_pair_p,
+                               bool unique_key=false) {
     // For value collection it always returns nullptr
     const KeyValuePair *found_pair_p = nullptr;
 
@@ -3072,7 +3073,7 @@ class BwTree : public BwTreeBase {
     } else {
       // If a value is given then use this value to Traverse down leaf
       // page to find whether the value exists or not
-      found_pair_p = NavigateLeafNode(context_p, *value_p, index_pair_p);
+      found_pair_p = NavigateLeafNode(context_p, *value_p, index_pair_p, unique_key);
     }
 
     if (context_p->abort_flag == true) {
@@ -4172,7 +4173,8 @@ class BwTree : public BwTreeBase {
    */
   const KeyValuePair *NavigateLeafNode(Context *context_p,
                                        const ValueType &search_value,
-                                       std::pair<int, bool> *index_pair_p) {
+                                       std::pair<int, bool> *index_pair_p,
+                                       bool unique_key=false) {
     // This will go to the right sibling until we have seen
     // a node whose range match the search key
     NavigateSiblingChain(context_p);
@@ -4217,7 +4219,7 @@ class BwTree : public BwTreeBase {
             // We do not need to check any delete set here, since if the
             // value has been deleted earlier then this function would
             // already have returned
-            if (ValueCmpEqual(scan_start_it->second, search_value)) {
+            if (unique_key == true || ValueCmpEqual(scan_start_it->second, search_value)) {
               // Since only Delete() will use this piece of information
               // we set exist flag to false to indicate that the value
               // has been invalidated
@@ -4245,7 +4247,7 @@ class BwTree : public BwTreeBase {
               static_cast<const LeafInsertNode *>(node_p);
 
           if (KeyCmpEqual(search_key, insert_node_p->item.first)) {
-            if (ValueCmpEqual(insert_node_p->item.second, search_value)) {
+            if (unique_key == true || ValueCmpEqual(insert_node_p->item.second, search_value)) {
               // Only Delete() will use this
               // We just simply inherit from the first node
               *index_pair_p = insert_node_p->GetIndexPair();
@@ -4264,7 +4266,7 @@ class BwTree : public BwTreeBase {
 
           // If the value was deleted then return false
           if (KeyCmpEqual(search_key, delete_node_p->item.first)) {
-            if (ValueCmpEqual(delete_node_p->item.second, search_value)) {
+            if (unique_key == true || ValueCmpEqual(delete_node_p->item.second, search_value)) {
               // Only Insert() will use this
               // We just simply inherit from the first node
               *index_pair_p = delete_node_p->GetIndexPair();
@@ -6993,8 +6995,12 @@ class BwTree : public BwTreeBase {
    *
    * This function returns false if value already exists
    * If CAS fails this function retries until it succeeds
+   * 
+   * Note that this function also takes a unique_key argument, to indicate whether
+   * we allow the same key with different values. For a primary key index this
+   * should be set true. By default we allow non-unique key
    */
-  bool Insert(const KeyType &key, const ValueType &value) {
+  bool Insert(const KeyType &key, const ValueType &value, bool unique_key=false) {
     LOG_TRACE("Insert called");
 
 #ifdef BWTREE_DEBUG
@@ -7011,7 +7017,7 @@ class BwTree : public BwTreeBase {
       // Also if the key previously exists in the delta chain
       // then return the position of the node using next_key_p
       // if there is none then return nullptr
-      const KeyValuePair *item_p = Traverse(&context, &value, &index_pair);
+      const KeyValuePair *item_p = Traverse(&context, &value, &index_pair, unique_key);
 
       // If the key-value pair already exists then return false
       if (item_p != nullptr) {

src/index/bwtree_index.cpp

@@ -53,15 +53,20 @@ bool BWTREE_INDEX_TYPE::InsertEntry(const storage::Tuple *key,
   KeyType index_key;
   index_key.SetFromKey(key);
 
-  bool ret = container.Insert(index_key, value);
+  bool ret;
+  if(HasUniqueKeys() == true) {
+    ret = container.Insert(index_key, value, true);
+  } else {
+    ret = container.Insert(index_key, value, false);
+  }
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexInserts(metadata);
   }
 
-  LOG_TRACE("InsertEntry(key=%s, val=%s) [%s]",
-            index_key.GetInfo().c_str(),
+  // NOTE: If I use index_key.GetInfo() here, I always get an empty key?
+  LOG_DEBUG("InsertEntry(key=%s, val=%s) [%s]",
+            key->GetInfo().c_str(),
             IndexUtil::GetInfo(value).c_str(),
             (ret ? "SUCCESS" : "FAIL"));
 
@@ -85,14 +90,13 @@ bool BWTREE_INDEX_TYPE::DeleteEntry(const storage::Tuple *key,
   // it is unnecessary for us to allocate memory
   bool ret = container.Delete(index_key, value);
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexDeletes(
         delete_count, metadata);
   }
 
-  LOG_TRACE("DeleteEntry(key=%s, val=%s) [%s]",
-            index_key.GetInfo().c_str(),
+  LOG_DEBUG("DeleteEntry(key=%s, val=%s) [%s]",
+            key->GetInfo().c_str(),
             IndexUtil::GetInfo(value).c_str(),
             (ret ? "SUCCESS" : "FAIL"));
 
@@ -122,8 +126,7 @@ bool BWTREE_INDEX_TYPE::CondInsertEntry(
     assert(ret == false);
   }
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexInserts(metadata);
   }
 
@@ -197,8 +200,7 @@ void BWTREE_INDEX_TYPE::Scan(
     }
   }  // if is full scan
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexReads(
         result.size(), metadata);
   }
@@ -222,6 +224,7 @@ void BWTREE_INDEX_TYPE::ScanLimit(
     const std::vector<ExpressionType> &expr_list,
     ScanDirectionType scan_direction, std::vector<ValueType> &result,
     const ConjunctionScanPredicate *csp_p, uint64_t limit, uint64_t offset) {
+
   // Only work with limit == 1 and offset == 0
   // Because that gets translated to "min"
   // But still since we could not access tuples in the table
@@ -243,6 +246,7 @@ void BWTREE_INDEX_TYPE::ScanLimit(
     auto scan_itr = container.Begin(index_low_key);
     if ((scan_itr.IsEnd() == false) &&
         (container.KeyCmpLessEqual(scan_itr->first, index_high_key))) {
+
       result.push_back(scan_itr->second);
     }
   } else {
@@ -263,8 +267,7 @@ void BWTREE_INDEX_TYPE::ScanAllKeys(std::vector<ValueType> &result) {
     it++;
   }
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexReads(
         result.size(), metadata);
   }
@@ -280,8 +283,7 @@ void BWTREE_INDEX_TYPE::ScanKey(const storage::Tuple *key,
   // This function in BwTree fills a given vector
   container.GetValue(index_key, result);
 
-  if (static_cast<StatsType>(settings::SettingsManager::GetInt(
-          settings::SettingId::stats_mode)) != StatsType::INVALID) {
+  if (static_cast<StatsType>(settings::SettingsManager::GetInt(settings::SettingId::stats_mode)) != StatsType::INVALID) {
     stats::BackendStatsContext::GetInstance()->IncrementIndexReads(
         result.size(), metadata);
   }