Support Merge with wide-column entities in iterator (#10941)

Summary:
The patch adds `Merge` support for wide-column entities in `DBIter`. As before, the `Merge` operation is applied to the default column of the entity; any other columns are unchanged. As a small cleanup, the PR also changes the signature of `DBIter::Merge` to simply return a boolean instead of the `Merge` operation's `Status` since the actual `Status` is already stored in a member variable.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/10941

Test Plan: `make check`

Reviewed By: riversand963

Differential Revision: D41195471

Pulled By: ltamasi

fbshipit-source-id: 362cf555897296e252c3de5ddfbd569ef34f85ef

diff --git a/db/db_iter.cc b/db/db_iter.cc
index 9a130700515d9cc5472ee48e5827e7973000446c..e1375deb7d2130730fb33a092e79d1c826d243d7 100644 (file)
--- a/db/db_iter.cc
+++ b/db/db_iter.cc
@@ -546,8 +546,7 @@ bool DBIter::MergeValuesNewToOld() {
       // hit a put, merge the put value with operands and store the
       // final result in saved_value_. We are done!
       const Slice val = iter_.value();
-      Status s = Merge(&val, ikey.user_key);
-      if (!s.ok()) {
+      if (!Merge(&val, ikey.user_key)) {
         return false;
       }
       // iter_ is positioned after put
@@ -576,8 +575,7 @@ bool DBIter::MergeValuesNewToOld() {
         return false;
       }
       valid_ = true;
-      Status s = Merge(&blob_value_, ikey.user_key);
-      if (!s.ok()) {
+      if (!Merge(&blob_value_, ikey.user_key)) {
         return false;
       }
 
@@ -591,11 +589,18 @@ bool DBIter::MergeValuesNewToOld() {
       }
       return true;
     } else if (kTypeWideColumnEntity == ikey.type) {
-      // TODO: support wide-column entities
-      status_ = Status::NotSupported(
-          "Merge currently not supported for wide-column entities");
-      valid_ = false;
-      return false;
+      if (!MergeEntity(iter_.value(), ikey.user_key)) {
+        return false;
+      }
+
+      // iter_ is positioned after put
+      iter_.Next();
+      if (!iter_.status().ok()) {
+        valid_ = false;
+        return false;
+      }
+
+      return true;
     } else {
       valid_ = false;
       status_ = Status::Corruption(
@@ -614,8 +619,7 @@ bool DBIter::MergeValuesNewToOld() {
   // a deletion marker.
   // feed null as the existing value to the merge operator, such that
   // client can differentiate this scenario and do things accordingly.
-  Status s = Merge(nullptr, saved_key_.GetUserKey());
-  if (!s.ok()) {
+  if (!Merge(nullptr, saved_key_.GetUserKey())) {
     return false;
   }
   assert(status_.ok());
@@ -964,8 +968,7 @@ bool DBIter::FindValueForCurrentKey() {
       if (last_not_merge_type == kTypeDeletion ||
           last_not_merge_type == kTypeSingleDeletion ||
           last_not_merge_type == kTypeDeletionWithTimestamp) {
-        s = Merge(nullptr, saved_key_.GetUserKey());
-        if (!s.ok()) {
+        if (!Merge(nullptr, saved_key_.GetUserKey())) {
           return false;
         }
         return true;
@@ -980,8 +983,7 @@ bool DBIter::FindValueForCurrentKey() {
           return false;
         }
         valid_ = true;
-        s = Merge(&blob_value_, saved_key_.GetUserKey());
-        if (!s.ok()) {
+        if (!Merge(&blob_value_, saved_key_.GetUserKey())) {
           return false;
         }
 
@@ -989,15 +991,14 @@ bool DBIter::FindValueForCurrentKey() {
 
         return true;
       } else if (last_not_merge_type == kTypeWideColumnEntity) {
-        // TODO: support wide-column entities
-        status_ = Status::NotSupported(
-            "Merge currently not supported for wide-column entities");
-        valid_ = false;
-        return false;
+        if (!MergeEntity(pinned_value_, saved_key_.GetUserKey())) {
+          return false;
+        }
+
+        return true;
       } else {
         assert(last_not_merge_type == kTypeValue);
-        s = Merge(&pinned_value_, saved_key_.GetUserKey());
-        if (!s.ok()) {
+        if (!Merge(&pinned_value_, saved_key_.GetUserKey())) {
           return false;
         }
         return true;
@@ -1181,8 +1182,7 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
 
     if (ikey.type == kTypeValue) {
       const Slice val = iter_.value();
-      Status s = Merge(&val, saved_key_.GetUserKey());
-      if (!s.ok()) {
+      if (!Merge(&val, saved_key_.GetUserKey())) {
         return false;
       }
       return true;
@@ -1201,8 +1201,7 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
         return false;
       }
       valid_ = true;
-      Status s = Merge(&blob_value_, saved_key_.GetUserKey());
-      if (!s.ok()) {
+      if (!Merge(&blob_value_, saved_key_.GetUserKey())) {
         return false;
       }
 
@@ -1210,11 +1209,11 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
 
       return true;
     } else if (ikey.type == kTypeWideColumnEntity) {
-      // TODO: support wide-column entities
-      status_ = Status::NotSupported(
-          "Merge currently not supported for wide-column entities");
-      valid_ = false;
-      return false;
+      if (!MergeEntity(iter_.value(), saved_key_.GetUserKey())) {
+        return false;
+      }
+
+      return true;
     } else {
       valid_ = false;
       status_ = Status::Corruption(
@@ -1224,8 +1223,7 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
     }
   }
 
-  Status s = Merge(nullptr, saved_key_.GetUserKey());
-  if (!s.ok()) {
+  if (!Merge(nullptr, saved_key_.GetUserKey())) {
     return false;
   }
 
@@ -1248,7 +1246,7 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
   return true;
 }
 
-Status DBIter::Merge(const Slice* val, const Slice& user_key) {
+bool DBIter::Merge(const Slice* val, const Slice& user_key) {
   Status s = MergeHelper::TimedFullMerge(
       merge_operator_, user_key, val, merge_context_.GetOperands(),
       &saved_value_, logger_, statistics_, clock_, &pinned_value_,
@@ -1256,14 +1254,33 @@ Status DBIter::Merge(const Slice* val, const Slice& user_key) {
   if (!s.ok()) {
     valid_ = false;
     status_ = s;
-    return s;
+    return false;
   }
 
   SetValueAndColumnsFromPlain(pinned_value_.data() ? pinned_value_
                                                    : saved_value_);
 
   valid_ = true;
-  return s;
+  return true;
+}
+
+bool DBIter::MergeEntity(const Slice& entity, const Slice& user_key) {
+  Status s = MergeHelper::TimedFullMergeWithEntity(
+      merge_operator_, user_key, entity, merge_context_.GetOperands(),
+      &saved_value_, logger_, statistics_, clock_,
+      /* update_num_ops_stats */ true);
+  if (!s.ok()) {
+    valid_ = false;
+    status_ = s;
+    return false;
+  }
+
+  if (!SetValueAndColumnsFromEntity(saved_value_)) {
+    return false;
+  }
+
+  valid_ = true;
+  return true;
 }
 
 // Move backwards until the key smaller than saved_key_.

diff --git a/db/db_iter.h b/db/db_iter.h
index c1c7dd40468659ed8a987377371ad6879a9b6498..e87c2b4c90ecf015a700bf649973749e86a0b0cf 100644 (file)
--- a/db/db_iter.h
+++ b/db/db_iter.h
@@ -318,7 +318,8 @@ class DBIter final : public Iterator {
   }
 
   // If user-defined timestamp is enabled, `user_key` includes timestamp.
-  Status Merge(const Slice* val, const Slice& user_key);
+  bool Merge(const Slice* val, const Slice& user_key);
+  bool MergeEntity(const Slice& entity, const Slice& user_key);
 
   const SliceTransform* prefix_extractor_;
   Env* const env_;

diff --git a/db/wide/db_wide_basic_test.cc b/db/wide/db_wide_basic_test.cc
index cb3ec1af0c534c47e103345163efc3aef8d81a21..c6670f7f769804a29291ff16d20230f3fa6e0546 100644 (file)
--- a/db/wide/db_wide_basic_test.cc
+++ b/db/wide/db_wide_basic_test.cc
@@ -399,6 +399,16 @@ TEST_F(DBWideBasicTest, MergeEntity) {
   auto verify = [&]() {
     const std::string first_expected_default(
         first_columns[0].value().ToString() + delim + first_merge_operand);
+    WideColumns first_expected_columns{
+        {kDefaultWideColumnName, first_expected_default},
+        first_columns[1],
+        first_columns[2]};
+
+    const std::string second_expected_default(delim + second_merge_operand);
+    WideColumns second_expected_columns{
+        {kDefaultWideColumnName, second_expected_default},
+        second_columns[0],
+        second_columns[1]};
 
     {
       PinnableSlice result;
@@ -411,13 +421,7 @@ TEST_F(DBWideBasicTest, MergeEntity) {
       PinnableWideColumns result;
       ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
                                first_key, &result));
-
-      WideColumns expected_columns{
-          {kDefaultWideColumnName, first_expected_default},
-          first_columns[1],
-          first_columns[2]};
-
-      ASSERT_EQ(result.columns(), expected_columns);
+      ASSERT_EQ(result.columns(), first_expected_columns);
     }
 
     {
@@ -438,8 +442,6 @@ TEST_F(DBWideBasicTest, MergeEntity) {
       ASSERT_EQ(merge_operands[1], first_merge_operand);
     }
 
-    const std::string second_expected_default(delim + second_merge_operand);
-
     {
       PinnableSlice result;
       ASSERT_OK(db_->Get(ReadOptions(), db_->DefaultColumnFamily(), second_key,
@@ -451,13 +453,7 @@ TEST_F(DBWideBasicTest, MergeEntity) {
       PinnableWideColumns result;
       ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
                                second_key, &result));
-
-      WideColumns expected_columns{
-          {kDefaultWideColumnName, second_expected_default},
-          second_columns[0],
-          second_columns[1]};
-
-      ASSERT_EQ(result.columns(), expected_columns);
+      ASSERT_EQ(result.columns(), second_expected_columns);
     }
 
     {
@@ -495,18 +491,44 @@ TEST_F(DBWideBasicTest, MergeEntity) {
       ASSERT_OK(statuses[1]);
     }
 
-    // Note: Merge is currently not supported for wide-column entities in
-    // iterator
     {
       std::unique_ptr<Iterator> iter(db_->NewIterator(ReadOptions()));
 
       iter->SeekToFirst();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), first_key);
+      ASSERT_EQ(iter->value(), first_expected_default);
+      ASSERT_EQ(iter->columns(), first_expected_columns);
+
+      iter->Next();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), second_key);
+      ASSERT_EQ(iter->value(), second_expected_default);
+      ASSERT_EQ(iter->columns(), second_expected_columns);
+
+      iter->Next();
       ASSERT_FALSE(iter->Valid());
-      ASSERT_TRUE(iter->status().IsNotSupported());
+      ASSERT_OK(iter->status());
 
       iter->SeekToLast();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), second_key);
+      ASSERT_EQ(iter->value(), second_expected_default);
+      ASSERT_EQ(iter->columns(), second_expected_columns);
+
+      iter->Prev();
+      ASSERT_TRUE(iter->Valid());
+      ASSERT_OK(iter->status());
+      ASSERT_EQ(iter->key(), first_key);
+      ASSERT_EQ(iter->value(), first_expected_default);
+      ASSERT_EQ(iter->columns(), first_expected_columns);
+
+      iter->Prev();
       ASSERT_FALSE(iter->Valid());
-      ASSERT_TRUE(iter->status().IsNotSupported());
+      ASSERT_OK(iter->status());
     }
   };