andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix a bug in MANIFEST group commit (#4157)

Browse Source 
Summary:
PR #3944 introduces group commit of `VersionEdit` in MANIFEST. The
implementation has a bug. When updating the log file number of each column
family, we must consider only `VersionEdit`s that operate on the same column
family. Otherwise, a column family may accidentally set its log file number
higher than actual value, indicating that log files with smaller file number
will be ignored, thus causing some updates to be lost.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/4157

Differential Revision: D8916650

Pulled By: riversand963

fbshipit-source-id: 8f456cf688f17bf35ad87b38e30e899aa162f201
This commit is contained in:
Yanqin Jin 2018-07-19 17:26:25 -07:00
parent b79a965f7c
commit 87f1325b2b
1 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
db/version_set.cc
View File

@ -2994,25 +2994,31 @@ Status VersionSet::ProcessManifestWrites(
delete first_writer.cfd;
}
} else {
// Each version in versions corresponds to a column family.
// For each column family, update its log number indicating that logs
// with number smaller than this should be ignored.
for (const auto version : versions) {
uint64_t max_log_number_in_batch = 0;
uint64_t last_min_log_number_to_keep = 0;
for (auto& e : batch_edits) {
if (e->has_log_number_) {
uint32_t cf_id = version->cfd_->GetID();
for (const auto& e : batch_edits) {
if (e->has_log_number_ && e->column_family_ == cf_id) {
max_log_number_in_batch =
std::max(max_log_number_in_batch, e->log_number_);
}
}
if (max_log_number_in_batch != 0) {
assert(version->cfd_->GetLogNumber() <= max_log_number_in_batch);
version->cfd_->SetLogNumber(max_log_number_in_batch);
}
}
uint64_t last_min_log_number_to_keep = 0;
for (auto& e : batch_edits) {
if (e->has_min_log_number_to_keep_) {
last_min_log_number_to_keep =
std::max(last_min_log_number_to_keep, e->min_log_number_to_keep_);
}
}
if (max_log_number_in_batch != 0) {
for (int i = 0; i < static_cast<int>(versions.size()); ++i) {
ColumnFamilyData* cfd = versions[i]->cfd_;
assert(cfd->GetLogNumber() <= max_log_number_in_batch);
cfd->SetLogNumber(max_log_number_in_batch);
}
}
if (last_min_log_number_to_keep != 0) {
// Should only be set in 2PC mode.