Merge pull request #1276 from wangziqi2013/non-unique-key-fix

BwTree non-unique key insert fix

Author: pervazea

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

@@ -38,7 +38,7 @@ BWTREE_INDEX_TYPE::BWTreeIndex(IndexMetadata *metadata)
       container{false, comparator, equals, hash_func} {
   return;
 }
-
+        
 BWTREE_TEMPLATE_ARGUMENTS
 BWTREE_INDEX_TYPE::~BWTreeIndex() {}
 
@@ -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);
   }
 
+  // NOTE: If I use index_key.GetInfo() here, I always get an empty key?
   LOG_TRACE("InsertEntry(key=%s, val=%s) [%s]",
-            index_key.GetInfo().c_str(),
+            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(),
+            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);
   }

test/include/index/testing_index_util.h

@@ -37,9 +37,9 @@ TEST_F(BwTreeIndexTests, UniqueKeyInsertTest) {
   TestingIndexUtil::UniqueKeyInsertTest(IndexType::BWTREE);
 }
 
-//TEST_F(BwTreeIndexTests, UniqueKeyDeleteTest) {
-//  TestingIndexUtil::UniqueKeyDeleteTest(IndexType::BWTREE);
-//}
+TEST_F(BwTreeIndexTests, UniqueKeyDeleteTest) {
+  TestingIndexUtil::UniqueKeyDeleteTest(IndexType::BWTREE);
+}
 
 TEST_F(BwTreeIndexTests, NonUniqueKeyDeleteTest) {
   TestingIndexUtil::NonUniqueKeyDeleteTest(IndexType::BWTREE);

test/index/bwtree_index_test.cpp

@@ -112,14 +112,27 @@ void TestingIndexUtil::UniqueKeyInsertTest(const IndexType index_type) {
   LaunchParallelTest(1, TestingIndexUtil::InsertHelper, index.get(), pool,
                      scale_factor);
 
-  // Checks
+  // Check basic insert (key0 only has 1 value, so it is not sufficient)
   std::unique_ptr<storage::Tuple> key0(new storage::Tuple(key_schema, true));
   key0->SetValue(0, type::ValueFactory::GetIntegerValue(100), pool);
   key0->SetValue(1, type::ValueFactory::GetVarcharValue("a"), pool);
 
   index->ScanKey(key0.get(), location_ptrs);
+  LOG_DEBUG("Checking key0's value: %lu", location_ptrs.size());
+  EXPECT_EQ(1, location_ptrs.size());
+  location_ptrs.clear();
+
+  // This tests non-unqie key. If scan returns more than 1 then it should fail
+  std::unique_ptr<storage::Tuple> key1(new storage::Tuple(key_schema, true));
+  key1->SetValue(0, type::ValueFactory::GetIntegerValue(100), pool);
+  key1->SetValue(1, type::ValueFactory::GetVarcharValue("b"), pool);
+
+  index->ScanKey(key1.get(), location_ptrs);
+  LOG_DEBUG("Checking key1's value: %lu", location_ptrs.size());
   EXPECT_EQ(1, location_ptrs.size());
   location_ptrs.clear();
+
+  return;
 }
 
 void TestingIndexUtil::UniqueKeyDeleteTest(const IndexType index_type) {
@@ -163,32 +176,37 @@ void TestingIndexUtil::UniqueKeyDeleteTest(const IndexType index_type) {
   index->ScanKey(key0.get(), location_ptrs);
 #ifdef LOG_DEBUG_ENABLED
   for (auto ptr : location_ptrs) {
-    LOG_DEBUG(" FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
+    LOG_DEBUG("key0 FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
   }
 #endif
+  LOG_DEBUG("key0 size: %lu", location_ptrs.size());
   EXPECT_EQ(0, location_ptrs.size());
   location_ptrs.clear();
 
   LOG_DEBUG("ScanKey(key1=%s)", key1->GetInfo().c_str());
   index->ScanKey(key1.get(), location_ptrs);
 #ifdef LOG_DEBUG_ENABLED
   for (auto ptr : location_ptrs) {
-    LOG_DEBUG(" FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
+    LOG_DEBUG("key1 FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
   }
 #endif
+  LOG_DEBUG("key1 size: %lu", location_ptrs.size());
   EXPECT_EQ(0, location_ptrs.size());
   location_ptrs.clear();
 
   LOG_DEBUG("ScanKey(key2=%s)", key2->GetInfo().c_str());
   index->ScanKey(key2.get(), location_ptrs);
 #ifdef LOG_DEBUG_ENABLED
   for (auto ptr : location_ptrs) {
-    LOG_DEBUG(" FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
+    LOG_DEBUG("key2 FOUND: %s", index::IndexUtil::GetInfo(ptr).c_str());
   }
 #endif
+  LOG_DEBUG("key2 size: %lu", location_ptrs.size());
   EXPECT_EQ(1, location_ptrs.size());
   EXPECT_EQ(TestingIndexUtil::item1->block, location_ptrs[0]->block);
   location_ptrs.clear();
+
+  return;
 }
 
 void TestingIndexUtil::NonUniqueKeyDeleteTest(const IndexType index_type) {
@@ -843,11 +861,13 @@ void TestingIndexUtil::DeleteHelper(index::Index *index,
     index->DeleteEntry(key3.get(), TestingIndexUtil::item1.get());
     index->DeleteEntry(key4.get(), TestingIndexUtil::item1.get());
 
-    // should be no key0
+    // For non-unique key should be no key0
     // key1 item 0 1 2
     // key2 item 1
     // no key3
     // no key4
+
+    // For unique key should be only key2
   }
 }