database/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00
Code Activity

Add in-memory data structures and (de)serialization support for subcompaction progress (#13928)

Browse Source 
Summary:
**Context**
Resuming compaction is designed to periodically record the progress of an ongoing compaction and can resume from that saved progress after interruptions such as cancellation, database shutdown, or crashes.

This PR introduces the data structures needed to store  subcompaction progress in memory, along with serialization and deserialization support to persist and parse this progress to/from "a manifest-like compaction progress file" (the actual creation of such file is in upcoming PRs).

Flow of resuming: DB::OpenAndCompact() -> Compaction progress file  -> SubcompactionProgress -> CompactionJob
Flow of persistence: CompactionJob -> SubcompactionProgress -> Compaction progress file  -> DB that is called with OpenAndCompact()

**Summary**
Progress represented by `SubcompactionProgress` will be tracked at the scope of a subcompaction, which is the smallest independent unit of compaction work.

The frequency of recording this progress is once every N compaction output files (to be detailed in future PRs).

When recording, all fields, except for the output files metadata in `SubcompactionProgress`, will directly overwrite the corresponding fields from the last saved progress (See `SubcompactionProgress` and `SubcompactionProgressBuilder` for more).

As a bonus, this PR refactors the file metadata encoding and decoding utilities into two static helper functions, EncodeToNewFile4() and DecodeNewFile4From(), to support subcompaction progress usage.

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

Test Plan:
- Added various `SubcompactionProgressTest` unit tests in version_edit_test.cc to verify basic serialization/deserialization and forward compatibility handling
- Existing UTs and stress/crash test

**Follow up:**
- Move output entry number and file verification to after each file creation so we can remove kNumProcessedOutputRecords persistence support and make resuming compaction work with `paranoid_file_checks=true` (by default false). Output verification will be done before persistence of progress. As long as this follow-up is done before the landing of the integration PR to create the progress file, we can change the manifest-like compaction progress file format freely.

Reviewed By: jaykorean

Differential Revision: D81986583

Pulled By: hx235

fbshipit-source-id: b42766da7d9c2e2f596c892d050c753238d1039f
This commit is contained in:
Hui Xiao
2025-09-22 15:03:46 -07:00
committed by Facebook GitHub Bot
parent eb1d924308
commit ab10ea0aac
3 changed files with 1120 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

657
db/version_edit.cc
View File

@@ -112,124 +112,9 @@ bool VersionEdit::EncodeTo(std::string* dst,
f.epoch_number == kUnknownEpochNumber) {
return false;
}
PutVarint32(dst, kNewFile4);
PutVarint32Varint64(dst, new_files_[i].first /* level */, f.fd.GetNumber());
PutVarint64(dst, f.fd.GetFileSize());
EncodeFileBoundaries(dst, f, ts_sz.value());
PutVarint64Varint64(dst, f.fd.smallest_seqno, f.fd.largest_seqno);
// Customized fields' format:
// +-----------------------------+
// | 1st field's tag (varint32) |
// +-----------------------------+
// | 1st field's size (varint32) |
// +-----------------------------+
// | bytes for 1st field |
// | (based on size decoded) |
// +-----------------------------+
// | |
// | ...... |
// | |
// +-----------------------------+
// | last field's size (varint32)|
// +-----------------------------+
// | bytes for last field |
// | (based on size decoded) |
// +-----------------------------+
// | terminating tag (varint32) |
// +-----------------------------+
//
// Customized encoding for fields:
// tag kPathId: 1 byte as path_id
// tag kNeedCompaction:
// now only can take one char value 1 indicating need-compaction
//
PutVarint32(dst, NewFileCustomTag::kOldestAncesterTime);
std::string varint_oldest_ancester_time;
PutVarint64(&varint_oldest_ancester_time, f.oldest_ancester_time);
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:VarintOldestAncesterTime",
&varint_oldest_ancester_time);
PutLengthPrefixedSlice(dst, Slice(varint_oldest_ancester_time));
PutVarint32(dst, NewFileCustomTag::kFileCreationTime);
std::string varint_file_creation_time;
PutVarint64(&varint_file_creation_time, f.file_creation_time);
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:VarintFileCreationTime",
&varint_file_creation_time);
PutLengthPrefixedSlice(dst, Slice(varint_file_creation_time));
PutVarint32(dst, NewFileCustomTag::kEpochNumber);
std::string varint_epoch_number;
PutVarint64(&varint_epoch_number, f.epoch_number);
PutLengthPrefixedSlice(dst, Slice(varint_epoch_number));
if (f.file_checksum_func_name != kUnknownFileChecksumFuncName) {
PutVarint32(dst, NewFileCustomTag::kFileChecksum);
PutLengthPrefixedSlice(dst, Slice(f.file_checksum));
PutVarint32(dst, NewFileCustomTag::kFileChecksumFuncName);
PutLengthPrefixedSlice(dst, Slice(f.file_checksum_func_name));
}
if (f.fd.GetPathId() != 0) {
PutVarint32(dst, NewFileCustomTag::kPathId);
char p = static_cast<char>(f.fd.GetPathId());
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (f.temperature != Temperature::kUnknown) {
PutVarint32(dst, NewFileCustomTag::kTemperature);
char p = static_cast<char>(f.temperature);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (f.marked_for_compaction) {
PutVarint32(dst, NewFileCustomTag::kNeedCompaction);
char p = static_cast<char>(1);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (has_min_log_number_to_keep_ && !min_log_num_written) {
PutVarint32(dst, NewFileCustomTag::kMinLogNumberToKeepHack);
std::string varint_log_number;
PutFixed64(&varint_log_number, min_log_number_to_keep_);
PutLengthPrefixedSlice(dst, Slice(varint_log_number));
min_log_num_written = true;
}
if (f.oldest_blob_file_number != kInvalidBlobFileNumber) {
PutVarint32(dst, NewFileCustomTag::kOldestBlobFileNumber);
std::string oldest_blob_file_number;
PutVarint64(&oldest_blob_file_number, f.oldest_blob_file_number);
PutLengthPrefixedSlice(dst, Slice(oldest_blob_file_number));
}
UniqueId64x2 unique_id = f.unique_id;
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:UniqueId", &unique_id);
if (unique_id != kNullUniqueId64x2) {
PutVarint32(dst, NewFileCustomTag::kUniqueId);
std::string unique_id_str = EncodeUniqueIdBytes(&unique_id);
PutLengthPrefixedSlice(dst, Slice(unique_id_str));
}
if (f.compensated_range_deletion_size) {
PutVarint32(dst, kCompensatedRangeDeletionSize);
std::string compensated_range_deletion_size;
PutVarint64(&compensated_range_deletion_size,
f.compensated_range_deletion_size);
PutLengthPrefixedSlice(dst, Slice(compensated_range_deletion_size));
}
if (f.tail_size) {
PutVarint32(dst, NewFileCustomTag::kTailSize);
std::string varint_tail_size;
PutVarint64(&varint_tail_size, f.tail_size);
PutLengthPrefixedSlice(dst, Slice(varint_tail_size));
}
if (!f.user_defined_timestamps_persisted) {
// The default value for the flag is true, it's only explicitly persisted
// when it's false. We are putting 0 as the value here to signal false
// (i.e. UDTS not persisted).
PutVarint32(dst, NewFileCustomTag::kUserDefinedTimestampsPersisted);
char p = static_cast<char>(0);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:NewFile4:CustomizeFields",
dst);
PutVarint32(dst, NewFileCustomTag::kTerminate);
EncodeToNewFile4(f, new_files_[i].first, ts_sz.value(),
has_min_log_number_to_keep_, min_log_number_to_keep_,
min_log_num_written, dst);
}
for (const auto& blob_file_addition : blob_file_additions_) {
@@ -288,9 +173,142 @@ bool VersionEdit::EncodeTo(std::string* dst,
char p = static_cast<char>(persist_user_defined_timestamps_);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (HasSubcompactionProgress()) {
PutVarint32(dst, kSubcompactionProgress);
std::string progress_data;
subcompaction_progress_.EncodeTo(&progress_data);
PutLengthPrefixedSlice(dst, progress_data);
}
return true;
}
void VersionEdit::EncodeToNewFile4(const FileMetaData& f, int level,
size_t ts_sz,
bool has_min_log_number_to_keep,
uint64_t min_log_number_to_keep,
bool& min_log_num_written,
std::string* dst) {
PutVarint32(dst, kNewFile4);
PutVarint32Varint64(dst, level, f.fd.GetNumber());
PutVarint64(dst, f.fd.GetFileSize());
EncodeFileBoundaries(dst, f, ts_sz);
PutVarint64Varint64(dst, f.fd.smallest_seqno, f.fd.largest_seqno);
// Customized fields' format:
// +-----------------------------+
// | 1st field's tag (varint32) |
// +-----------------------------+
// | 1st field's size (varint32) |
// +-----------------------------+
// | bytes for 1st field |
// | (based on size decoded) |
// +-----------------------------+
// | |
// | ...... |
// | |
// +-----------------------------+
// | last field's size (varint32)|
// +-----------------------------+
// | bytes for last field |
// | (based on size decoded) |
// +-----------------------------+
// | terminating tag (varint32) |
// +-----------------------------+
//
// Customized encoding for fields:
// tag kPathId: 1 byte as path_id
// tag kNeedCompaction:
// now only can take one char value 1 indicating need-compaction
//
PutVarint32(dst, NewFileCustomTag::kOldestAncesterTime);
std::string varint_oldest_ancester_time;
PutVarint64(&varint_oldest_ancester_time, f.oldest_ancester_time);
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:VarintOldestAncesterTime",
&varint_oldest_ancester_time);
PutLengthPrefixedSlice(dst, Slice(varint_oldest_ancester_time));
PutVarint32(dst, NewFileCustomTag::kFileCreationTime);
std::string varint_file_creation_time;
PutVarint64(&varint_file_creation_time, f.file_creation_time);
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:VarintFileCreationTime",
&varint_file_creation_time);
PutLengthPrefixedSlice(dst, Slice(varint_file_creation_time));
PutVarint32(dst, NewFileCustomTag::kEpochNumber);
std::string varint_epoch_number;
PutVarint64(&varint_epoch_number, f.epoch_number);
PutLengthPrefixedSlice(dst, Slice(varint_epoch_number));
if (f.file_checksum_func_name != kUnknownFileChecksumFuncName) {
PutVarint32(dst, NewFileCustomTag::kFileChecksum);
PutLengthPrefixedSlice(dst, Slice(f.file_checksum));
PutVarint32(dst, NewFileCustomTag::kFileChecksumFuncName);
PutLengthPrefixedSlice(dst, Slice(f.file_checksum_func_name));
}
if (f.fd.GetPathId() != 0) {
PutVarint32(dst, NewFileCustomTag::kPathId);
char p = static_cast<char>(f.fd.GetPathId());
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (f.temperature != Temperature::kUnknown) {
PutVarint32(dst, NewFileCustomTag::kTemperature);
char p = static_cast<char>(f.temperature);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (f.marked_for_compaction) {
PutVarint32(dst, NewFileCustomTag::kNeedCompaction);
char p = static_cast<char>(1);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
if (has_min_log_number_to_keep && !min_log_num_written) {
PutVarint32(dst, NewFileCustomTag::kMinLogNumberToKeepHack);
std::string varint_log_number;
PutFixed64(&varint_log_number, min_log_number_to_keep);
PutLengthPrefixedSlice(dst, Slice(varint_log_number));
min_log_num_written = true;
}
if (f.oldest_blob_file_number != kInvalidBlobFileNumber) {
PutVarint32(dst, NewFileCustomTag::kOldestBlobFileNumber);
std::string oldest_blob_file_number;
PutVarint64(&oldest_blob_file_number, f.oldest_blob_file_number);
PutLengthPrefixedSlice(dst, Slice(oldest_blob_file_number));
}
UniqueId64x2 unique_id = f.unique_id;
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:UniqueId", &unique_id);
if (unique_id != kNullUniqueId64x2) {
PutVarint32(dst, NewFileCustomTag::kUniqueId);
std::string unique_id_str = EncodeUniqueIdBytes(&unique_id);
PutLengthPrefixedSlice(dst, Slice(unique_id_str));
}
if (f.compensated_range_deletion_size) {
PutVarint32(dst, NewFileCustomTag::kCompensatedRangeDeletionSize);
std::string compensated_range_deletion_size;
PutVarint64(&compensated_range_deletion_size,
f.compensated_range_deletion_size);
PutLengthPrefixedSlice(dst, Slice(compensated_range_deletion_size));
}
if (f.tail_size) {
PutVarint32(dst, NewFileCustomTag::kTailSize);
std::string varint_tail_size;
PutVarint64(&varint_tail_size, f.tail_size);
PutLengthPrefixedSlice(dst, Slice(varint_tail_size));
}
if (!f.user_defined_timestamps_persisted) {
// The default value for the flag is true, it's only explicitly persisted
// when it's false. We are putting 0 as the value here to signal false
// (i.e. UDTS not persisted).
PutVarint32(dst, NewFileCustomTag::kUserDefinedTimestampsPersisted);
char p = static_cast<char>(0);
PutLengthPrefixedSlice(dst, Slice(&p, 1));
}
TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:NewFile4:CustomizeFields",
dst);
PutVarint32(dst, NewFileCustomTag::kTerminate);
}
static bool GetInternalKey(Slice* input, InternalKey* dst) {
Slice str;
if (GetLengthPrefixedSlice(input, &str)) {
@@ -301,12 +319,12 @@ static bool GetInternalKey(Slice* input, InternalKey* dst) {
}
}
bool VersionEdit::GetLevel(Slice* input, int* level, const char** /*msg*/) {
bool VersionEdit::GetLevel(Slice* input, int* level, int& max_level) {
uint32_t v = 0;
if (GetVarint32(input, &v)) {
*level = v;
if (max_level_ < *level) {
max_level_ = *level;
if (max_level < *level) {
max_level = *level;
}
return true;
} else {
@@ -314,16 +332,18 @@ bool VersionEdit::GetLevel(Slice* input, int* level, const char** /*msg*/) {
}
}
const char* VersionEdit::DecodeNewFile4From(Slice* input) {
const char* msg = nullptr;
const char* VersionEdit::DecodeNewFile4From(Slice* input, int& max_level,
uint64_t& min_log_number_to_keep,
bool& has_min_log_number_to_keep,
NewFiles& new_files,
FileMetaData& f) {
int level = 0;
FileMetaData f;
uint64_t number = 0;
uint32_t path_id = 0;
uint64_t file_size = 0;
SequenceNumber smallest_seqno = 0;
SequenceNumber largest_seqno = kMaxSequenceNumber;
if (GetLevel(input, &level, &msg) && GetVarint64(input, &number) &&
if (GetLevel(input, &level, max_level) && GetVarint64(input, &number) &&
GetVarint64(input, &file_size) && GetInternalKey(input, &f.smallest) &&
GetInternalKey(input, &f.largest) &&
GetVarint64(input, &smallest_seqno) &&
@@ -381,10 +401,10 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {
case kMinLogNumberToKeepHack:
// This is a hack to encode kMinLogNumberToKeep in a
// forward-compatible fashion.
if (!GetFixed64(&field, &min_log_number_to_keep_)) {
if (!GetFixed64(&field, &min_log_number_to_keep)) {
return "deleted log number malformatted";
}
has_min_log_number_to_keep_ = true;
has_min_log_number_to_keep = true;
break;
case kOldestBlobFileNumber:
if (!GetVarint64(&field, &f.oldest_blob_file_number)) {
@@ -436,13 +456,12 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {
}
f.fd =
FileDescriptor(number, path_id, file_size, smallest_seqno, largest_seqno);
new_files_.push_back(std::make_pair(level, f));
new_files.emplace_back(level, f);
return nullptr;
}
void VersionEdit::EncodeFileBoundaries(std::string* dst,
const FileMetaData& meta,
size_t ts_sz) const {
const FileMetaData& meta, size_t ts_sz) {
if (ts_sz == 0 || meta.user_defined_timestamps_persisted) {
PutLengthPrefixedSlice(dst, meta.smallest.Encode());
PutLengthPrefixedSlice(dst, meta.largest.Encode());
@@ -545,7 +564,8 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
break;
case kCompactCursor:
if (GetLevel(&input, &level, &msg) && GetInternalKey(&input, &key)) {
if (GetLevel(&input, &level, max_level_) &&
GetInternalKey(&input, &key)) {
// Here we re-use the output format of compact pointer in LevelDB
// to persist compact_cursors_
compact_cursors_.push_back(std::make_pair(level, key));
@@ -558,7 +578,8 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
case kDeletedFile: {
uint64_t number = 0;
if (GetLevel(&input, &level, &msg) && GetVarint64(&input, &number)) {
if (GetLevel(&input, &level, max_level_) &&
GetVarint64(&input, &number)) {
deleted_files_.insert(std::make_pair(level, number));
} else {
if (!msg) {
@@ -571,8 +592,8 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
case kNewFile: {
uint64_t number = 0;
uint64_t file_size = 0;
if (GetLevel(&input, &level, &msg) && GetVarint64(&input, &number) &&
GetVarint64(&input, &file_size) &&
if (GetLevel(&input, &level, max_level_) &&
GetVarint64(&input, &number) && GetVarint64(&input, &file_size) &&
GetInternalKey(&input, &f.smallest) &&
GetInternalKey(&input, &f.largest)) {
f.fd = FileDescriptor(number, 0, file_size);
@@ -589,8 +610,8 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
uint64_t file_size = 0;
SequenceNumber smallest_seqno = 0;
SequenceNumber largest_seqno = kMaxSequenceNumber;
if (GetLevel(&input, &level, &msg) && GetVarint64(&input, &number) &&
GetVarint64(&input, &file_size) &&
if (GetLevel(&input, &level, max_level_) &&
GetVarint64(&input, &number) && GetVarint64(&input, &file_size) &&
GetInternalKey(&input, &f.smallest) &&
GetInternalKey(&input, &f.largest) &&
GetVarint64(&input, &smallest_seqno) &&
@@ -612,8 +633,9 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
uint64_t file_size = 0;
SequenceNumber smallest_seqno = 0;
SequenceNumber largest_seqno = kMaxSequenceNumber;
if (GetLevel(&input, &level, &msg) && GetVarint64(&input, &number) &&
GetVarint32(&input, &path_id) && GetVarint64(&input, &file_size) &&
if (GetLevel(&input, &level, max_level_) &&
GetVarint64(&input, &number) && GetVarint32(&input, &path_id) &&
GetVarint64(&input, &file_size) &&
GetInternalKey(&input, &f.smallest) &&
GetInternalKey(&input, &f.largest) &&
GetVarint64(&input, &smallest_seqno) &&
@@ -630,7 +652,10 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}
case kNewFile4: {
msg = DecodeNewFile4From(&input);
FileMetaData ignored_file;
msg = DecodeNewFile4From(&input, max_level_, min_log_number_to_keep_,
has_min_log_number_to_keep_, new_files_,
ignored_file);
break;
}
@@ -767,6 +792,23 @@ Status VersionEdit::DecodeFrom(const Slice& src) {
}
break;
case kSubcompactionProgress: {
Slice encoded;
if (!GetLengthPrefixedSlice(&input, &encoded)) {
msg = "SubcompactionProgress not prefixed by length";
break;
}
SubcompactionProgress progress;
Status s = progress.DecodeFrom(&encoded);
if (!s.ok()) {
return s;
}
SetSubcompactionProgress(progress);
break;
}
default:
if (tag & kTagSafeIgnoreMask) {
// Tag from future which can be safely ignored.
@@ -1087,4 +1129,341 @@ std::string VersionEdit::DebugJSON(int edit_num, bool hex_key) const {
return jw.Get();
}
void SubcompactionProgressPerLevel::EncodeTo(std::string* dst) const {
if (num_processed_output_records_ > 0) {
PutVarint32(
dst,
SubcompactionProgressPerLevelCustomTag::kNumProcessedOutputRecords);
std::string varint_records;
PutVarint64(&varint_records, num_processed_output_records_);
PutLengthPrefixedSlice(dst, varint_records);
}
if (!output_files_.empty()) {
PutVarint32(dst, SubcompactionProgressPerLevelCustomTag::kOutputFilesDelta);
std::string files_data;
EncodeOutputFiles(&files_data);
PutLengthPrefixedSlice(dst, files_data);
} else if (!temp_output_files_allocation_.empty()) {
PutVarint32(dst, SubcompactionProgressPerLevelCustomTag::kOutputFilesDelta);
std::string files_data;
EncodeTemporaryOutputFilesAllocation(&files_data);
PutLengthPrefixedSlice(dst, files_data);
}
PutVarint32(dst, SubcompactionProgressPerLevelCustomTag::
kSubcompactionProgressPerLevelTerminate);
}
Status SubcompactionProgressPerLevel::DecodeFrom(Slice* input) {
Clear();
while (true) {
uint32_t tag = 0;
if (!GetVarint32(input, &tag)) {
return Status::Corruption("SubcompactionProgressPerLevel", "tag error");
}
if (tag == SubcompactionProgressPerLevelCustomTag::
kSubcompactionProgressPerLevelTerminate) {
break;
}
Slice field;
if (!GetLengthPrefixedSlice(input, &field)) {
return Status::Corruption("SubcompactionProgressPerLevel",
"field length prefixed slice error");
}
switch (tag) {
case SubcompactionProgressPerLevelCustomTag::kNumProcessedOutputRecords: {
if (!GetVarint64(&field, &num_processed_output_records_)) {
return Status::Corruption("SubcompactionProgressPerLevel",
"invalid num_processed_output_records_");
}
break;
}
case SubcompactionProgressPerLevelCustomTag::kOutputFilesDelta: {
Status s = DecodeOutputFiles(&field, temp_output_files_allocation_);
if (!s.ok()) {
return s;
}
break;
}
default:
// Forward compatibility: Handle unknown tags
if ((tag & SubcompactionProgressPerLevelCustomTag::
kSubcompactionProgressPerLevelCustomTagSafeIgnoreMask) !=
0) {
break;
} else {
return Status::NotSupported("SubcompactionProgress",
"unsupported critical custom field");
}
}
}
return Status::OK();
}
void SubcompactionProgressPerLevel::EncodeOutputFiles(std::string* dst) const {
size_t new_files_count =
output_files_.size() > last_persisted_output_files_count_
? output_files_.size() - last_persisted_output_files_count_
: 0;
assert(new_files_count > 0);
PutVarint32(dst, static_cast<uint32_t>(new_files_count));
for (size_t i = last_persisted_output_files_count_; i < output_files_.size();
++i) {
const FileMetaData* file_ptr = output_files_[i];
assert(file_ptr != nullptr);
std::string file_dst;
bool ignored_min_log_written = false;
VersionEdit::EncodeToNewFile4(*file_ptr, -1 /* level */, 0 /* ts_sz */,
false /* has_min_log_number_to_keep */,
0 /* min_log_number_to_keep */,
ignored_min_log_written, &file_dst);
PutLengthPrefixedSlice(dst, file_dst);
}
}
void SubcompactionProgressPerLevel::EncodeTemporaryOutputFilesAllocation(
std::string* dst) const {
size_t new_files_count =
temp_output_files_allocation_.size() > last_persisted_output_files_count_
? temp_output_files_allocation_.size() -
last_persisted_output_files_count_
: 0;
assert(new_files_count > 0);
PutVarint32(dst, static_cast<uint32_t>(new_files_count));
for (size_t i = last_persisted_output_files_count_;
i < temp_output_files_allocation_.size(); ++i) {
const FileMetaData& file = temp_output_files_allocation_[i];
std::string file_dst;
bool ignored_min_log_written = false;
VersionEdit::EncodeToNewFile4(file, -1 /* level */, 0 /* ts_sz */,
false /* has_min_log_number_to_keep */,
0 /* min_log_number_to_keep */,
ignored_min_log_written, &file_dst);
PutLengthPrefixedSlice(dst, file_dst);
}
}
Status SubcompactionProgressPerLevel::DecodeOutputFiles(
Slice* input, autovector<FileMetaData>& temporary_output_files_allocation) {
uint32_t new_file_count = 0;
if (!GetVarint32(input, &new_file_count)) {
return Status::Corruption("SubcompactionProgressPerLevel",
"new output file count");
}
assert(temporary_output_files_allocation.size() == 0);
temporary_output_files_allocation.reserve(new_file_count);
for (uint32_t i = 0; i < new_file_count; ++i) {
Slice file_input;
if (!GetLengthPrefixedSlice(input, &file_input)) {
return Status::Corruption("SubcompactionProgressPerLevel",
"output file metadata");
}
uint32_t tag = 0;
if (!GetVarint32(&file_input, &tag) || tag != kNewFile4) {
return Status::Corruption("SubcompactionProgressPerLevel",
"expected kNewFile4 tag");
}
int ignored_max_level = -1;
uint64_t ignored_min_log_number_to_keep = 0;
bool ignored_has_min_log_number_to_keep = false;
VersionEdit::NewFiles ignored_new_files;
FileMetaData file;
const char* err = VersionEdit::DecodeNewFile4From(
&file_input, ignored_max_level, ignored_min_log_number_to_keep,
ignored_has_min_log_number_to_keep, ignored_new_files, file);
if (err != nullptr) {
return Status::Corruption("SubcompactionProgressPerLevel", err);
}
temporary_output_files_allocation.push_back(std::move(file));
}
return Status::OK();
}
void SubcompactionProgress::EncodeTo(std::string* dst) const {
if (!next_internal_key_to_compact.empty()) {
PutVarint32(dst, SubcompactionProgressCustomTag::kNextInternalKeyToCompact);
PutLengthPrefixedSlice(dst, next_internal_key_to_compact);
}
if (num_processed_input_records > 0) {
PutVarint32(dst, SubcompactionProgressCustomTag::kNumProcessedInputRecords);
std::string varint_records;
PutVarint64(&varint_records, num_processed_input_records);
PutLengthPrefixedSlice(dst, varint_records);
}
if (output_level_progress.GetOutputFiles().size() >
output_level_progress.GetLastPersistedOutputFilesCount()) {
PutVarint32(dst, SubcompactionProgressCustomTag::kOutputLevelProgress);
std::string level_progress_data;
output_level_progress.EncodeTo(&level_progress_data);
PutLengthPrefixedSlice(dst, level_progress_data);
}
if (proximal_output_level_progress.GetOutputFiles().size() >
proximal_output_level_progress.GetLastPersistedOutputFilesCount()) {
PutVarint32(dst,
SubcompactionProgressCustomTag::kProximalOutputLevelProgress);
std::string level_progress_data;
proximal_output_level_progress.EncodeTo(&level_progress_data);
PutLengthPrefixedSlice(dst, level_progress_data);
}
PutVarint32(dst,
SubcompactionProgressCustomTag::kSubcompactionProgressTerminate);
}
Status SubcompactionProgress::DecodeFrom(Slice* input) {
Clear();
while (true) {
uint32_t custom_tag = 0;
if (!GetVarint32(input, &custom_tag)) {
return Status::Corruption("SubcompactionProgress",
"custom field tag error");
}
if (custom_tag ==
SubcompactionProgressCustomTag::kSubcompactionProgressTerminate) {
break;
}
Slice field;
if (!GetLengthPrefixedSlice(input, &field)) {
return Status::Corruption("SubcompactionProgress",
"custom field length prefixed slice error");
}
switch (custom_tag) {
case SubcompactionProgressCustomTag::kNextInternalKeyToCompact:
next_internal_key_to_compact = field.ToString();
break;
case SubcompactionProgressCustomTag::kNumProcessedInputRecords:
if (!GetVarint64(&field, &num_processed_input_records)) {
return Status::Corruption("SubcompactionProgress",
"invalid num_processed_input_records");
}
break;
case SubcompactionProgressCustomTag::kOutputLevelProgress: {
Status s = output_level_progress.DecodeFrom(&field);
if (!s.ok()) {
return s;
}
break;
}
case SubcompactionProgressCustomTag::kProximalOutputLevelProgress: {
Status s = proximal_output_level_progress.DecodeFrom(&field);
if (!s.ok()) {
return s;
}
break;
}
default:
if ((custom_tag & SubcompactionProgressCustomTag::
kSubcompactionProgressCustomTagSafeIgnoreMask) !=
0) {
break;
} else {
return Status::NotSupported("SubcompactionProgress",
"unsupported critical custom field");
}
}
}
return Status::OK();
}
bool SubcompactionProgressBuilder::ProcessVersionEdit(const VersionEdit& edit) {
if (!edit.HasSubcompactionProgress()) {
return false;
}
const SubcompactionProgress& progress = edit.GetSubcompactionProgress();
MergeDeltaProgress(progress);
has_subcompaction_progress_ = true;
return true;
}
void SubcompactionProgressBuilder::MergeDeltaProgress(
const SubcompactionProgress& delta_progress) {
accumulated_subcompaction_progress_.next_internal_key_to_compact =
delta_progress.next_internal_key_to_compact;
accumulated_subcompaction_progress_.num_processed_input_records =
delta_progress.num_processed_input_records;
MaybeMergeDeltaProgressPerLevel(
accumulated_subcompaction_progress_.output_level_progress,
delta_progress.output_level_progress);
MaybeMergeDeltaProgressPerLevel(
accumulated_subcompaction_progress_.proximal_output_level_progress,
delta_progress.proximal_output_level_progress);
}
void SubcompactionProgressBuilder::MaybeMergeDeltaProgressPerLevel(
SubcompactionProgressPerLevel& accumulated_level_progress,
const SubcompactionProgressPerLevel& delta_level_progress) {
assert(delta_level_progress.GetOutputFiles().empty());
if (delta_level_progress.GetTempOutputFilesAllocation().empty()) {
return;
}
accumulated_level_progress.SetNumProcessedOutputRecords(
delta_level_progress.GetNumProcessedOutputRecords());
auto& accumulated_temp_files =
accumulated_level_progress.TempOutputFilesAllocation();
const auto& delta_temp_files =
delta_level_progress.GetTempOutputFilesAllocation();
accumulated_temp_files.reserve(accumulated_temp_files.size() +
delta_temp_files.size());
for (const auto& file_allocation : delta_temp_files) {
accumulated_temp_files.push_back(file_allocation);
}
}
void SubcompactionProgressBuilder::Clear() {
accumulated_subcompaction_progress_.Clear();
has_subcompaction_progress_ = false;
}
} // namespace ROCKSDB_NAMESPACE

243
db/version_edit.h
View File

@@ -72,6 +72,23 @@ enum Tag : uint32_t {
kWalAddition2,
kWalDeletion2,
kPersistUserDefinedTimestamps,
kSubcompactionProgress,
};
enum SubcompactionProgressPerLevelCustomTag : uint32_t {
kSubcompactionProgressPerLevelTerminate = 1, // End of fields marker
kOutputFilesDelta = 2,
kNumProcessedOutputRecords = 3,
kSubcompactionProgressPerLevelCustomTagSafeIgnoreMask = 1 << 16,
};
enum SubcompactionProgressCustomTag : uint32_t {
kSubcompactionProgressTerminate = 1, // End of fields marker
kNextInternalKeyToCompact = 2,
kNumProcessedInputRecords = 3,
kOutputLevelProgress = 4,
kProximalOutputLevelProgress = 5,
kSubcompactionProgressCustomTagSafeIgnoreMask = 1 << 16,
};
enum NewFileCustomTag : uint32_t {
@@ -440,12 +457,198 @@ struct LevelFilesBrief {
}
};
struct SubcompactionProgressPerLevel {
uint64_t GetNumProcessedOutputRecords() const {
return num_processed_output_records_;
}
void SetNumProcessedOutputRecords(uint64_t num) {
num_processed_output_records_ = num;
}
const autovector<const FileMetaData*>& GetOutputFiles() const {
return output_files_;
}
void AddToOutputFiles(const FileMetaData* file) {
output_files_.push_back(file);
}
const autovector<FileMetaData>& GetTempOutputFilesAllocation() const {
return temp_output_files_allocation_;
}
autovector<FileMetaData>& TempOutputFilesAllocation() {
return temp_output_files_allocation_;
}
size_t GetLastPersistedOutputFilesCount() const {
return last_persisted_output_files_count_;
}
void UpdateLastPersistedOutputFilesCount() {
last_persisted_output_files_count_ =
std::max(output_files_.size(), temp_output_files_allocation_.size());
}
void EncodeTo(std::string* dst) const;
Status DecodeFrom(Slice* input);
void Clear() {
num_processed_output_records_ = 0;
output_files_.clear();
temp_output_files_allocation_.clear();
last_persisted_output_files_count_ = 0;
}
std::string ToString() const {
std::ostringstream oss;
oss << "SubcompactionProgressPerLevel{";
oss << " num_processed_output_records=" << num_processed_output_records_;
oss << ", output_files_count=" << output_files_.size();
oss << ", temp_output_files_allocation_count="
<< temp_output_files_allocation_.size();
oss << ", last_persisted_output_files_count="
<< last_persisted_output_files_count_;
oss << " }";
return oss.str();
}
void TEST_ClearOutputFiles() { output_files_.clear(); }
private:
uint64_t num_processed_output_records_ = 0;
// These pointers ONLY point to FileMetaData objects owned by compaction
// outputs. They are NEVER set to point to objects in
// `temp_output_files_allocation` This ensures stable pointers that don't get
// invalidated by copy/move operations on `SubcompactionProgress`
autovector<const FileMetaData*> output_files_ = {};
// These are ONLY used during deserialization from VersionEdit.
// They provide temporary storage before being moved to compaction outputs.
autovector<FileMetaData> temp_output_files_allocation_ = {};
// Number of files already persisted to help calculate the new output files to
// persist in the future. This is to prevent having to persist all the output
// files metadata so far every time of a "snapshot" of a progress is persisted
// which can lead to O(1+2+...+n) = O(n^2) file metadata being persisted. The
// current approach of persisting only the delta should always persist
// exactly the number (n) of output files in total.
size_t last_persisted_output_files_count_ = 0;
void EncodeOutputFiles(std::string* dst) const;
void EncodeTemporaryOutputFilesAllocation(std::string* dst) const;
Status DecodeOutputFiles(Slice* input,
autovector<FileMetaData>& temp_storage);
};
struct SubcompactionProgress {
static constexpr uint64_t kInaccurateNumProcessedInputRecords =
std::numeric_limits<uint64_t>::max();
std::string next_internal_key_to_compact;
uint64_t num_processed_input_records = 0;
SubcompactionProgressPerLevel output_level_progress;
SubcompactionProgressPerLevel proximal_output_level_progress;
SubcompactionProgress() = default;
void Clear() {
next_internal_key_to_compact.clear();
num_processed_input_records = 0;
output_level_progress.Clear();
proximal_output_level_progress.Clear();
}
void EncodeTo(std::string* dst) const;
Status DecodeFrom(Slice* input);
std::string ToString() const {
std::ostringstream oss;
oss << "SubcompactionProgress{";
oss << " next_internal_key_to_compact="
<< (next_internal_key_to_compact.empty()
? "NONE"
: next_internal_key_to_compact);
oss << ", num_processed_input_records=" << num_processed_input_records;
oss << ", output_level_progress" << output_level_progress.ToString();
oss << ", proximal_output_level_progress"
<< proximal_output_level_progress.ToString();
oss << " }";
return oss.str();
}
};
class VersionEdit;
// Builder class to reconstruct complete subcompaction progress object
// from multiple decoded VersionEdits containing delta output files information
// of the same subcompaction. See
// `SubcompactionProgressPerLevel::last_persisted_output_files_count_`'s comment
//
// WARNING: This class currently assumes all input VersionEdits contain progress
// information for the SAME subcompaction. It does not validate
// progress data from different subcompactions so mixing progress from
// multiple subcompactions can result in corrupted state silently. The caller is
// responsible for ensuring all VersionEdits processed by a single instance
// of this builder correspond to the same subcompaction.
class SubcompactionProgressBuilder {
public:
SubcompactionProgressBuilder() = default;
bool ProcessVersionEdit(const VersionEdit& edit);
const SubcompactionProgress& GetAccumulatedSubcompactionProgress() const {
return accumulated_subcompaction_progress_;
}
bool HasAccumulatedSubcompactionProgress() const {
return has_subcompaction_progress_;
}
void Clear();
private:
void MergeDeltaProgress(const SubcompactionProgress& delta_progress);
void MaybeMergeDeltaProgressPerLevel(
SubcompactionProgressPerLevel& accumulated_level_progress,
const SubcompactionProgressPerLevel& delta_level_progress);
SubcompactionProgress accumulated_subcompaction_progress_;
bool has_subcompaction_progress_ = false;
};
// Type alias for backward compatibility - vector of subcompaction progress
using CompactionProgress = std::vector<SubcompactionProgress>;
// The state of a DB at any given time is referred to as a Version.
// Any modification to the Version is considered a Version Edit. A Version is
// constructed by joining a sequence of Version Edits. Version Edits are written
// to the MANIFEST file.
class VersionEdit {
public:
// Retrieve the table files added as well as their associated levels.
using NewFiles = std::vector<std::pair<int, FileMetaData>>;
static void EncodeToNewFile4(const FileMetaData& f, int level, size_t ts_sz,
bool has_min_log_number_to_keep,
uint64_t min_log_number_to_keep,
bool& min_log_num_written, std::string* dst);
static const char* DecodeNewFile4From(Slice* input, int& max_level,
uint64_t& min_log_number_to_keep,
bool& has_min_log_number_to_keep,
NewFiles& new_files, FileMetaData& f);
void Clear();
void SetDBId(const std::string& db_id) {
@@ -564,8 +767,6 @@ class VersionEdit {
}
}
// Retrieve the table files added as well as their associated levels.
using NewFiles = std::vector<std::pair<int, FileMetaData>>;
const NewFiles& GetNewFiles() const { return new_files_; }
NewFiles& GetMutableNewFiles() { return new_files_; }
@@ -735,6 +936,22 @@ class VersionEdit {
full_history_ts_low_ = std::move(full_history_ts_low);
}
void SetSubcompactionProgress(const SubcompactionProgress& progress) {
has_subcompaction_progress_ = true;
subcompaction_progress_ = progress;
}
bool HasSubcompactionProgress() const { return has_subcompaction_progress_; }
const SubcompactionProgress& GetSubcompactionProgress() const {
return subcompaction_progress_;
}
void ClearSubcompactionProgress() {
has_subcompaction_progress_ = false;
subcompaction_progress_.Clear();
}
// return true on success.
// `ts_sz` is the size in bytes for the user-defined timestamp contained in
// a user key. This argument is optional because it's only required for
@@ -757,15 +974,22 @@ class VersionEdit {
std::string DebugJSON(int edit_num, bool hex_key = false) const;
private:
bool GetLevel(Slice* input, int* level, const char** msg);
const char* DecodeNewFile4From(Slice* input);
// Decode level information from serialized VersionEdit data and and track the
// maximum level seen.
//
// Parameters:
// input: Pointer to serialized data slice
// level: Output parameter for the decoded level value
// max_level: get updated if the decoded level is higher than passed in
// value
//
// Returns: true on successful decode, false on parse error
static bool GetLevel(Slice* input, int* level, int& max_level);
// Encode file boundaries `FileMetaData.smallest` and `FileMetaData.largest`.
// User-defined timestamps in the user key will be stripped if they shouldn't
// be persisted.
void EncodeFileBoundaries(std::string* dst, const FileMetaData& meta,
size_t ts_sz) const;
static void EncodeFileBoundaries(std::string* dst, const FileMetaData& meta,
size_t ts_sz);
int max_level_ = 0;
std::string db_id_;
@@ -816,6 +1040,9 @@ class VersionEdit {
std::string full_history_ts_low_;
bool persist_user_defined_timestamps_ = true;
bool has_subcompaction_progress_ = false;
SubcompactionProgress subcompaction_progress_;
// Newly created table files and blob files are eligible for deletion if they
// are not registered as live files after the background jobs creating them
// have finished. In case committing the VersionEdit containing such changes

367
db/version_edit_test.cc
View File

@@ -794,6 +794,373 @@ TEST(FileMetaDataTest, UpdateBoundariesBlobIndex) {
}
}
class SubcompactionProgressTest : public VersionEditTest {
protected:
static constexpr uint64_t kTestFileSize = 1024;
static constexpr SequenceNumber kTestSmallestSeq = 50;
static constexpr SequenceNumber kTestLargestSeq = 150;
static constexpr uint64_t kTestOldestAncesterTime = 12345;
static constexpr uint64_t kTestFileCreationTime = 67890;
static constexpr uint64_t kTestEpochNumber = 10;
static const std::string kTestChecksumFuncName;
FileMetaData CreateTestFile(uint64_t file_number, const std::string& prefix) {
FileMetaData file;
file.fd = FileDescriptor(file_number, 0, kTestFileSize, kTestSmallestSeq,
kTestLargestSeq);
file.smallest = InternalKey(prefix + "a", kTestSmallestSeq, kTypeValue);
file.largest = InternalKey(prefix + "z", kTestLargestSeq, kTypeValue);
file.oldest_ancester_time = kTestOldestAncesterTime;
file.file_creation_time = kTestFileCreationTime;
file.epoch_number = kTestEpochNumber;
file.file_checksum = "checksum_" + std::to_string(file_number);
file.file_checksum_func_name = kTestChecksumFuncName;
file.marked_for_compaction = false;
file.temperature = Temperature::kUnknown;
return file;
}
// Store external file metadata objects for testing
// These simulate files owned by CompactionOutputs
std::vector<FileMetaData> compaction_output_files_;
std::vector<FileMetaData> proximal_level_compaction_output_files_;
void SetupOutputFilePointers(
SubcompactionProgress& progress,
const std::vector<FileMetaData>& compaction_output_files,
const std::vector<FileMetaData>& proximal_level_compaction_output_files) {
if (!compaction_output_files.empty()) {
progress.output_level_progress.TEST_ClearOutputFiles();
}
for (const auto& file : compaction_output_files) {
progress.output_level_progress.AddToOutputFiles(&file);
}
if (!proximal_level_compaction_output_files.empty()) {
progress.proximal_output_level_progress.TEST_ClearOutputFiles();
}
for (const auto& file : proximal_level_compaction_output_files) {
progress.proximal_output_level_progress.AddToOutputFiles(&file);
}
}
SubcompactionProgress CreateSubcompactionProgress(
const std::string& next_key, uint64_t num_processed_input_records,
uint64_t num_processed_output_records,
uint64_t num_processed_proximal_level_output_records,
const std::vector<uint64_t>& output_file_numbers = {},
const std::vector<uint64_t>& proximal_file_numbers = {},
const std::string& file_prefix = "file_") {
SubcompactionProgress progress;
progress.next_internal_key_to_compact = next_key;
progress.num_processed_input_records = num_processed_input_records;
progress.output_level_progress.SetNumProcessedOutputRecords(
num_processed_output_records);
progress.proximal_output_level_progress.SetNumProcessedOutputRecords(
num_processed_proximal_level_output_records);
for (uint64_t file_num : output_file_numbers) {
compaction_output_files_.push_back(
CreateTestFile(file_num, file_prefix + "output_"));
}
for (uint64_t file_num : proximal_file_numbers) {
proximal_level_compaction_output_files_.push_back(
CreateTestFile(file_num, file_prefix + "proximal_"));
}
SetupOutputFilePointers(progress, compaction_output_files_,
proximal_level_compaction_output_files_);
return progress;
}
std::pair<const VersionEdit, const SubcompactionProgress>
EncodeDecodeProgress(const SubcompactionProgress& progress) {
VersionEdit edit;
edit.SetSubcompactionProgress(progress);
std::string encoded;
EXPECT_TRUE(edit.EncodeTo(&encoded, 0 /* ts_sz */));
VersionEdit decoded_edit;
EXPECT_OK(decoded_edit.DecodeFrom(encoded));
EXPECT_TRUE(decoded_edit.HasSubcompactionProgress());
SubcompactionProgress decoded_progress =
decoded_edit.GetSubcompactionProgress();
return {std::move(decoded_edit), std::move(decoded_progress)};
}
void VerifyFileMetaDataEquality(const FileMetaData& expected,
const FileMetaData& actual) {
// Verify the major fields only
ASSERT_EQ(actual.fd.GetNumber(), expected.fd.GetNumber());
ASSERT_EQ(actual.fd.GetFileSize(), expected.fd.GetFileSize());
ASSERT_EQ(actual.smallest.Encode(), expected.smallest.Encode());
ASSERT_EQ(actual.largest.Encode(), expected.largest.Encode());
ASSERT_EQ(actual.oldest_ancester_time, expected.oldest_ancester_time);
ASSERT_EQ(actual.file_creation_time, expected.file_creation_time);
ASSERT_EQ(actual.epoch_number, expected.epoch_number);
ASSERT_EQ(actual.file_checksum, expected.file_checksum);
ASSERT_EQ(actual.file_checksum_func_name, expected.file_checksum_func_name);
ASSERT_EQ(actual.marked_for_compaction, expected.marked_for_compaction);
ASSERT_EQ(actual.temperature, expected.temperature);
}
void VerifyProgressEquality(const SubcompactionProgress& expected,
const SubcompactionProgress& actual) {
ASSERT_EQ(actual.next_internal_key_to_compact,
expected.next_internal_key_to_compact);
ASSERT_EQ(actual.num_processed_input_records,
expected.num_processed_input_records);
for (const bool is_proximal_level : {false, true}) {
const SubcompactionProgressPerLevel&
actual_subcompaction_progress_by_level =
is_proximal_level ? actual.proximal_output_level_progress
: actual.output_level_progress;
const SubcompactionProgressPerLevel&
expected_subcompaction_progress_by_level =
is_proximal_level ? expected.proximal_output_level_progress
: expected.output_level_progress;
ASSERT_EQ(
actual_subcompaction_progress_by_level.GetNumProcessedOutputRecords(),
expected_subcompaction_progress_by_level
.GetNumProcessedOutputRecords());
ASSERT_EQ(
actual_subcompaction_progress_by_level.GetTempOutputFilesAllocation()
.size(),
expected_subcompaction_progress_by_level.GetOutputFiles().size());
for (size_t i = 0;
i < expected_subcompaction_progress_by_level.GetOutputFiles().size();
++i) {
VerifyFileMetaDataEquality(
*expected_subcompaction_progress_by_level.GetOutputFiles()[i],
actual_subcompaction_progress_by_level
.GetTempOutputFilesAllocation()[i]);
}
}
}
};
const std::string SubcompactionProgressTest::kTestChecksumFuncName = "crc32c";
TEST_F(SubcompactionProgressTest, BasicEncodeDecode) {
// Create progress with files for both levels
SubcompactionProgress progress = CreateSubcompactionProgress(
"key_100", // next_internal_key_to_compact
500, // num_processed_input_records
400, // num_processed_output_records
100, // num_processed_proximal_level_output_records
{1}, // output_file_numbers
{2}, // proximal_file_numbers
"test_" // file_prefix
);
auto [ignored, decoded_progress] = EncodeDecodeProgress(progress);
VerifyProgressEquality(progress, decoded_progress);
}
TEST_F(SubcompactionProgressTest, OutputFilesDeltaEncodeDecode) {
// Test Delta Encoding/Decoding
SubcompactionProgress initial_progress = CreateSubcompactionProgress(
"key_100", // next_internal_key_to_compact
100, // num_processed_input_records
40, // num_processed_output_records
60, // num_processed_proximal_level_output_records
{1}, // output_file_numbers
{2}, // proximal_file_numbers
"initial_" // file_prefix
);
auto [initial_decoded_edit, ignored_1] =
EncodeDecodeProgress(initial_progress);
initial_progress.output_level_progress.UpdateLastPersistedOutputFilesCount();
initial_progress.proximal_output_level_progress
.UpdateLastPersistedOutputFilesCount();
// Add one new output file to output and proximal level
SubcompactionProgress updated_progress = initial_progress;
updated_progress.next_internal_key_to_compact = "key_300";
updated_progress.num_processed_input_records = 1000;
updated_progress.output_level_progress.SetNumProcessedOutputRecords(400);
FileMetaData new_file = CreateTestFile(3, "new_");
compaction_output_files_.push_back(new_file);
updated_progress.proximal_output_level_progress.SetNumProcessedOutputRecords(
600);
FileMetaData new_file_proximal = CreateTestFile(4, "new_");
proximal_level_compaction_output_files_.push_back(new_file_proximal);
SetupOutputFilePointers(updated_progress, compaction_output_files_,
proximal_level_compaction_output_files_);
auto [delta_decoded_edit, delta_decoded_progress] =
EncodeDecodeProgress(updated_progress);
ASSERT_EQ(delta_decoded_progress.next_internal_key_to_compact,
updated_progress.next_internal_key_to_compact);
ASSERT_EQ(delta_decoded_progress.num_processed_input_records,
updated_progress.num_processed_input_records);
for (const bool& is_proximal_level : {false, true}) {
const SubcompactionProgressPerLevel& delta_progress_per_level =
is_proximal_level
? delta_decoded_progress.proximal_output_level_progress
: delta_decoded_progress.output_level_progress;
const SubcompactionProgressPerLevel& updated_progress_per_level =
is_proximal_level ? updated_progress.proximal_output_level_progress
: updated_progress.output_level_progress;
ASSERT_EQ(delta_progress_per_level.GetNumProcessedOutputRecords(),
updated_progress_per_level.GetNumProcessedOutputRecords());
// Delta encoding: only the one newly added file is present, not the
// previously persisted file
ASSERT_EQ(delta_progress_per_level.GetTempOutputFilesAllocation().size(),
1);
ASSERT_EQ(delta_progress_per_level.GetTempOutputFilesAllocation()[0]
.fd.GetNumber(),
is_proximal_level ? new_file_proximal.fd.GetNumber()
: new_file.fd.GetNumber());
}
// Test SubcompactionProgressBuilder
SubcompactionProgressBuilder builder;
ASSERT_FALSE(builder.HasAccumulatedSubcompactionProgress());
ASSERT_TRUE(builder.ProcessVersionEdit(initial_decoded_edit));
ASSERT_TRUE(builder.HasAccumulatedSubcompactionProgress());
ASSERT_TRUE(builder.ProcessVersionEdit(delta_decoded_edit));
const auto& accumulated_progress =
builder.GetAccumulatedSubcompactionProgress();
ASSERT_EQ(accumulated_progress.next_internal_key_to_compact,
updated_progress.next_internal_key_to_compact);
ASSERT_EQ(accumulated_progress.num_processed_input_records,
updated_progress.num_processed_input_records);
for (const bool& is_proximal_level : {false, true}) {
const SubcompactionProgressPerLevel& accumulated_progress_per_level =
is_proximal_level ? accumulated_progress.output_level_progress
: accumulated_progress.proximal_output_level_progress;
const SubcompactionProgressPerLevel& updated_progress_per_level =
is_proximal_level ? updated_progress.output_level_progress
: updated_progress.proximal_output_level_progress;
ASSERT_EQ(accumulated_progress_per_level.GetNumProcessedOutputRecords(),
updated_progress_per_level.GetNumProcessedOutputRecords());
ASSERT_EQ(
accumulated_progress_per_level.GetTempOutputFilesAllocation().size(),
updated_progress_per_level.GetOutputFiles().size());
std::set<uint64_t> accumulated_file_numbers;
for (const auto& file :
accumulated_progress_per_level.GetTempOutputFilesAllocation()) {
accumulated_file_numbers.insert(file.fd.GetNumber());
}
std::set<uint64_t> expected_file_numbers;
for (const auto& file : updated_progress_per_level.GetOutputFiles()) {
expected_file_numbers.insert(file->fd.GetNumber());
}
ASSERT_EQ(accumulated_file_numbers, expected_file_numbers);
}
// ===== PART 3: Test Builder Reset =====
builder.Clear();
ASSERT_FALSE(builder.HasAccumulatedSubcompactionProgress());
}
TEST_F(SubcompactionProgressTest, UnknownTags) {
SubcompactionProgress progress;
std::string encoded;
// 1. Test unknown ignorable tag
progress.next_internal_key_to_compact = "test_key";
progress.num_processed_input_records = 100;
PutVarint32(&encoded,
SubcompactionProgressCustomTag::kNextInternalKeyToCompact);
PutLengthPrefixedSlice(&encoded, progress.next_internal_key_to_compact);
PutVarint32(&encoded,
SubcompactionProgressCustomTag::kNumProcessedInputRecords);
std::string varint_records;
PutVarint64(&varint_records, progress.num_processed_input_records);
PutLengthPrefixedSlice(&encoded, varint_records);
// Manually encode with unknown ignorable tag (has
// SubcompactionProgressCustomTag::kSubcompactionProgressCustomTagSafeIgnoreMask
// bit set)
uint32_t unknown_ignorable_tag =
SubcompactionProgressCustomTag::
kSubcompactionProgressCustomTagSafeIgnoreMask +
1;
PutVarint32(&encoded, unknown_ignorable_tag);
PutLengthPrefixedSlice(&encoded, "future_data");
PutVarint32(&encoded,
SubcompactionProgressCustomTag::kSubcompactionProgressTerminate);
// Test decoding - should succeed and ignore unknown tag
Slice input(encoded);
SubcompactionProgress decoded_progress;
Status s = decoded_progress.DecodeFrom(&input);
ASSERT_OK(s);
// Verify known fields are preserved
ASSERT_EQ(decoded_progress.next_internal_key_to_compact,
progress.next_internal_key_to_compact);
ASSERT_EQ(decoded_progress.num_processed_input_records,
progress.num_processed_input_records);
// 2. Test unknown non-ignorable tag
encoded.clear();
PutVarint32(&encoded,
SubcompactionProgressCustomTag::kNextInternalKeyToCompact);
PutLengthPrefixedSlice(&encoded, "test_key");
// Manually encode with unknown non-ignorable tag (do not have
// SubcompactionProgressCustomTag::kSubcompactionProgressCustomTagSafeIgnoreMask
// bit set)
uint32_t unknown_critical_tag =
SubcompactionProgressCustomTag::
kSubcompactionProgressCustomTagSafeIgnoreMask -
1;
PutVarint32(&encoded, unknown_critical_tag);
PutLengthPrefixedSlice(&encoded, "critical_future_data");
PutVarint32(&encoded,
SubcompactionProgressCustomTag::kSubcompactionProgressTerminate);
// Test decoding - should fail on critical unknown tag
Slice critical_input(encoded);
SubcompactionProgress critical_progress;
Status critical_status = critical_progress.DecodeFrom(&critical_input);
ASSERT_NOK(critical_status);
ASSERT_TRUE(critical_status.IsNotSupported());
}
} // namespace ROCKSDB_NAMESPACE
int main(int argc, char** argv) {