Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Make it able to ignore WAL related VersionEdits in older versions #7873

Closed
wants to merge 5 commits into from
Closed

Make it able to ignore WAL related VersionEdits in older versions #7873

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c009190
be able to ignore blob and WAL related VersionEdits in older versions
Jan 16, 2021
1a2f02e
remove changes to blob related edits
Jan 16, 2021
65f6c2e
use new tags
Jan 19, 2021
b76199d
remove printf
Jan 19, 2021
160f99f
rename tags
Jan 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 49 additions & 8 deletions db/version_edit.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,22 +217,30 @@ bool VersionEdit::EncodeTo(std::string* dst) const {

for (const auto& blob_file_addition : blob_file_additions_) {
PutVarint32(dst, kBlobFileAddition);
blob_file_addition.EncodeTo(dst);
std::string encoded;
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ltamasi This breaks backward compatibility. But without doing so, users of BlobDB cannot downgrade. What do you think?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll remove changes related to BlobDB from this PR to make it focus on WAL edits.

Copy link
Contributor

@ltamasi ltamasi Jan 19, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ltamasi This breaks backward compatibility. But without doing so, users of BlobDB cannot downgrade. What do you think?

Nice catch @Cheng-Chang, and thanks for letting me know! In hindsight, the blob file related tags shouldn't have been put in the "ignorable" range since if you have blob files and downgrade to a version that cannot read these, you won't be able to read your data. Fortunately, since these are related to the integrated BlobDB project (which is WIP), we can still move these to the "non-ignorable" range. I'll take care of that.

blob_file_addition.EncodeTo(&encoded);
PutLengthPrefixedSlice(dst, encoded);
}

for (const auto& blob_file_garbage : blob_file_garbages_) {
PutVarint32(dst, kBlobFileGarbage);
blob_file_garbage.EncodeTo(dst);
std::string encoded;
blob_file_garbage.EncodeTo(&encoded);
PutLengthPrefixedSlice(dst, encoded);
}

for (const auto& wal_addition : wal_additions_) {
PutVarint32(dst, kWalAddition);
wal_addition.EncodeTo(dst);
std::string encoded;
wal_addition.EncodeTo(&encoded);
PutLengthPrefixedSlice(dst, encoded);
}

if (!wal_deletion_.IsEmpty()) {
PutVarint32(dst, kWalDeletion);
wal_deletion_.EncodeTo(dst);
std::string encoded;
wal_deletion_.EncodeTo(&encoded);
PutLengthPrefixedSlice(dst, encoded);
}

// 0 is default and does not need to be explicitly written
Expand Down Expand Up @@ -375,6 +383,11 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {

Status VersionEdit::DecodeFrom(const Slice& src) {
Clear();
#ifndef NDEBUG
bool ignore_ignorable_tags = false;
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:IgnoreIgnorableTags",
&ignore_ignorable_tags);
#endif
Slice input = src;
const char* msg = nullptr;
uint32_t tag = 0;
Expand All @@ -385,6 +398,12 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
Slice str;
InternalKey key;
while (msg == nullptr && GetVarint32(&input, &tag)) {
#ifndef NDEBUG
if (ignore_ignorable_tags && tag > kTagSafeIgnoreMask) {
printf("tag = %d\n", tag);
tag = kTagSafeIgnoreMask;
}
#endif
switch (tag) {
case kDbId:
if (GetLengthPrefixedSlice(&input, &str)) {
Expand Down Expand Up @@ -543,8 +562,13 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}

case kBlobFileAddition: {
Slice encoded;
if (!GetLengthPrefixedSlice(&input, &encoded)) {
msg = "BlobFileAddition not prefixed by length";
break;
}
BlobFileAddition blob_file_addition;
const Status s = blob_file_addition.DecodeFrom(&input);
const Status s = blob_file_addition.DecodeFrom(&encoded);
if (!s.ok()) {
return s;
}
Expand All @@ -554,8 +578,13 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}

case kBlobFileGarbage: {
Slice encoded;
if (!GetLengthPrefixedSlice(&input, &encoded)) {
msg = "BlobFileGarbage not prefixed by length";
break;
}
BlobFileGarbage blob_file_garbage;
const Status s = blob_file_garbage.DecodeFrom(&input);
const Status s = blob_file_garbage.DecodeFrom(&encoded);
if (!s.ok()) {
return s;
}
Expand All @@ -565,8 +594,14 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}

case kWalAddition: {
Slice encoded;
if (!GetLengthPrefixedSlice(&input, &encoded)) {
msg = "WalAddition not prefixed by length";
break;
}

WalAddition wal_addition;
const Status s = wal_addition.DecodeFrom(&input);
const Status s = wal_addition.DecodeFrom(&encoded);
if (!s.ok()) {
return s;
}
Expand All @@ -576,8 +611,14 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}

case kWalDeletion: {
Slice encoded;
if (!GetLengthPrefixedSlice(&input, &encoded)) {
msg = "WalDeletion not prefixed by length";
break;
}

WalDeletion wal_deletion;
const Status s = wal_deletion.DecodeFrom(&input);
const Status s = wal_deletion.DecodeFrom(&encoded);
if (!s.ok()) {
return s;
}
Expand Down
110 changes: 96 additions & 14 deletions db/version_edit_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,14 +324,22 @@ TEST_F(VersionEditTest, AddWalEncodeDecode) {
TestEncodeDecode(edit);
}

static std::string PrefixEncodedWalAdditionWithLength(
const std::string& encoded) {
std::string ret;
PutVarint32(&ret, Tag::kWalAddition);
PutLengthPrefixedSlice(&ret, encoded);
return ret;
}

TEST_F(VersionEditTest, AddWalDecodeBadLogNumber) {
std::string encoded;
PutVarint32(&encoded, Tag::kWalAddition);

{
// No log number.
std::string encoded_edit = PrefixEncodedWalAdditionWithLength(encoded);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
ASSERT_TRUE(s.ToString().find("Error decoding WAL log number") !=
std::string::npos)
Expand All @@ -345,8 +353,10 @@ TEST_F(VersionEditTest, AddWalDecodeBadLogNumber) {
unsigned char* ptr = reinterpret_cast<unsigned char*>(&c);
*ptr = 128;
encoded.append(1, c);

std::string encoded_edit = PrefixEncodedWalAdditionWithLength(encoded);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
ASSERT_TRUE(s.ToString().find("Error decoding WAL log number") !=
std::string::npos)
Expand All @@ -358,39 +368,45 @@ TEST_F(VersionEditTest, AddWalDecodeBadTag) {
constexpr WalNumber kLogNumber = 100;
constexpr uint64_t kSizeInBytes = 100;

std::string encoded_without_tag;
PutVarint32(&encoded_without_tag, Tag::kWalAddition);
PutVarint64(&encoded_without_tag, kLogNumber);
std::string encoded;
PutVarint64(&encoded, kLogNumber);

{
// No tag.
std::string encoded_edit = PrefixEncodedWalAdditionWithLength(encoded);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded_without_tag);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
ASSERT_TRUE(s.ToString().find("Error decoding tag") != std::string::npos)
<< s.ToString();
}

{
// Only has size tag, no terminate tag.
std::string encoded_with_size = encoded_without_tag;
std::string encoded_with_size = encoded;
PutVarint32(&encoded_with_size,
static_cast<uint32_t>(WalAdditionTag::kSyncedSize));
PutVarint64(&encoded_with_size, kSizeInBytes);

std::string encoded_edit =
PrefixEncodedWalAdditionWithLength(encoded_with_size);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded_with_size);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
ASSERT_TRUE(s.ToString().find("Error decoding tag") != std::string::npos)
<< s.ToString();
}

{
// Only has terminate tag.
std::string encoded_with_terminate = encoded_without_tag;
std::string encoded_with_terminate = encoded;
PutVarint32(&encoded_with_terminate,
static_cast<uint32_t>(WalAdditionTag::kTerminate));

std::string encoded_edit =
PrefixEncodedWalAdditionWithLength(encoded_with_terminate);
VersionEdit edit;
ASSERT_OK(edit.DecodeFrom(encoded_with_terminate));
ASSERT_OK(edit.DecodeFrom(encoded_edit));
auto& wal_addition = edit.GetWalAdditions()[0];
ASSERT_EQ(wal_addition.GetLogNumber(), kLogNumber);
ASSERT_FALSE(wal_addition.GetMetadata().HasSyncedSize());
Expand All @@ -401,15 +417,15 @@ TEST_F(VersionEditTest, AddWalDecodeNoSize) {
constexpr WalNumber kLogNumber = 100;

std::string encoded;
PutVarint32(&encoded, Tag::kWalAddition);
PutVarint64(&encoded, kLogNumber);
PutVarint32(&encoded, static_cast<uint32_t>(WalAdditionTag::kSyncedSize));
// No real size after the size tag.

{
// Without terminate tag.
std::string encoded_edit = PrefixEncodedWalAdditionWithLength(encoded);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
ASSERT_TRUE(s.ToString().find("Error decoding WAL file size") !=
std::string::npos)
Expand All @@ -419,8 +435,10 @@ TEST_F(VersionEditTest, AddWalDecodeNoSize) {
{
// With terminate tag.
PutVarint32(&encoded, static_cast<uint32_t>(WalAdditionTag::kTerminate));

std::string encoded_edit = PrefixEncodedWalAdditionWithLength(encoded);
VersionEdit edit;
Status s = edit.DecodeFrom(encoded);
Status s = edit.DecodeFrom(encoded_edit);
ASSERT_TRUE(s.IsCorruption());
// The terminate tag is misunderstood as the size.
ASSERT_TRUE(s.ToString().find("Error decoding tag") != std::string::npos)
Expand Down Expand Up @@ -515,6 +533,70 @@ TEST_F(VersionEditTest, FullHistoryTsLow) {
TestEncodeDecode(edit);
}

// Tests that if RocksDB is downgraded, the new types of VersionEdits
// that have a tag larger than kTagSafeIgnoreMask can be safely ignored.
TEST_F(VersionEditTest, IgnorableTags) {
SyncPoint::GetInstance()->SetCallBack(
"VersionEdit::EncodeTo:IgnoreIgnorableTags", [&](void* arg) {
bool* ignore = static_cast<bool*>(arg);
*ignore = true;
});
SyncPoint::GetInstance()->EnableProcessing();

constexpr uint64_t kPrevLogNumber = 100;
constexpr uint64_t kLogNumber = 200;
constexpr uint64_t kNextFileNumber = 300;
constexpr uint64_t kColumnFamilyId = 400;

VersionEdit edit;
// Add some ignorable entries.
for (int i = 0; i < 2; i++) {
edit.AddWal(i + 1, WalMetadata(i + 2));
}
edit.SetDBId("db_id");
// Add unignorable entries.
edit.SetPrevLogNumber(kPrevLogNumber);
edit.SetLogNumber(kLogNumber);
// Add more ignorable entries.
edit.DeleteWalsBefore(100);
for (int i = 0; i < 2; i++) {
edit.AddBlobFile(i + 1, i + 2, i + 3, "checksum_method", "checksum_value");
}
// Add unignorable entry.
edit.SetNextFile(kNextFileNumber);
// Add more ignorable entries.
for (int i = 0; i < 2; i++) {
edit.AddBlobFileGarbage(i + 1, i + 2, i + 3);
}
edit.SetFullHistoryTsLow("ts");
// Add unignorable entry.
edit.SetColumnFamily(kColumnFamilyId);

std::string encoded;
ASSERT_TRUE(edit.EncodeTo(&encoded));

VersionEdit decoded;
ASSERT_OK(decoded.DecodeFrom(encoded));

// Check that all ignorable entries are ignored.
ASSERT_FALSE(decoded.HasDbId());
ASSERT_FALSE(decoded.HasFullHistoryTsLow());
ASSERT_TRUE(decoded.GetBlobFileAdditions().empty());
ASSERT_TRUE(decoded.GetBlobFileGarbages().empty());
ASSERT_FALSE(decoded.IsWalAddition());
ASSERT_FALSE(decoded.IsWalDeletion());
ASSERT_TRUE(decoded.GetWalAdditions().empty());
ASSERT_TRUE(decoded.GetWalDeletion().IsEmpty());

// Check that unignorable entries are still present.
ASSERT_EQ(edit.GetPrevLogNumber(), kPrevLogNumber);
ASSERT_EQ(edit.GetLogNumber(), kLogNumber);
ASSERT_EQ(edit.GetNextFile(), kNextFileNumber);
ASSERT_EQ(edit.GetColumnFamily(), kColumnFamilyId);

SyncPoint::GetInstance()->DisableProcessing();
}

} // namespace ROCKSDB_NAMESPACE

int main(int argc, char** argv) {
Expand Down