From dbc4101b89a4d864c021cb7a68fdc5b6202458b4 Mon Sep 17 00:00:00 2001 From: Levi Tamasi Date: Thu, 10 Nov 2022 18:00:08 -0800 Subject: [PATCH] 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 --- db/db_iter.cc | 89 +++++++++++++++++++++-------------- db/db_iter.h | 3 +- db/wide/db_wide_basic_test.cc | 62 ++++++++++++++++-------- 3 files changed, 97 insertions(+), 57 deletions(-) diff --git a/db/db_iter.cc b/db/db_iter.cc index 9a1307005..e1375deb7 100644 --- 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 c1c7dd404..e87c2b4c9 100644 --- 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 cb3ec1af0..c6670f7f7 100644 --- 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 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()); } };