jabberwocky238/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-07-07 14:47:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Remove compression support (#14266)

Browse Source 
Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/14266

Compression is unused in production, making this dead code that adds unnecessary complexity.

Core changes:
- Remove `compression` field from `BlobDBOptions`
- Remove compression/decompression methods (`GetCompressedSlice`,
  `DecompressSlice`, `BlobDecompressor`)
- Simplify `ReadBlobFromOldFile` and `GetBlobValue` to handle only
  uncompressed blobs
- Update compaction filter to skip compression/decompression

CLI tool cleanup:
- Remove `--blob_db_compression_type` flag from db_bench (Stacked BlobDB)
- Remove `--show_uncompressed_blob` from blob_dump tool
- Remove `--dump_uncompressed_blobs` from ldb dump/file_dump commands
- blob_dump_tool now fails fast with NotSupported for compressed files

Tests:
- Remove compression-related tests (`Compression`, `DecompressAfterReopen`,
  `EnableDisableCompressionGC`, `ChangeCompressionGC`)

Reviewed By: xingbowang

Differential Revision: D91088957

fbshipit-source-id: 496ee41dcbd0023b794aa8a6d7dcc9c2451b7470
This commit is contained in:
Maciej Szeszko
2026-01-28 11:30:37 -08:00
committed by meta-codesync[bot]
parent acfea34c91
commit 2366f63e4f
17 changed files with 54 additions and 520 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/rocksdb/utilities/ldb_cmd.h
-1
View File
@@ -71,7 +71,6 @@ class LDBCommand {
static const std::string ARG_BLOB_FILE_STARTING_LEVEL;
static const std::string ARG_PREPOPULATE_BLOB_CACHE;
static const std::string ARG_DECODE_BLOB_INDEX;
static const std::string ARG_DUMP_UNCOMPRESSED_BLOBS;
static const std::string ARG_READ_TIMESTAMP;
static const std::string ARG_GET_WRITE_UNIX_TIME;
tools/blob_dump.cc
+1 -16
View File
@@ -27,12 +27,10 @@ int main(int argc, char** argv) {
{"file", required_argument, nullptr, 'f'},
{"show_key", optional_argument, nullptr, 'k'},
{"show_blob", optional_argument, nullptr, 'b'},
{"show_uncompressed_blob", optional_argument, nullptr, 'r'},
{"show_summary", optional_argument, nullptr, 's'},
};
DisplayType show_key = DisplayType::kRaw;
DisplayType show_blob = DisplayType::kNone;
DisplayType show_uncompressed_blob = DisplayType::kNone;
bool show_summary = false;
std::string file;
while (true) {
@@ -47,7 +45,6 @@ int main(int argc, char** argv) {
"Usage: blob_dump --file=filename "
"[--show_key[=none|raw|hex|detail]] "
"[--show_blob[=none|raw|hex|detail]] "
"[--show_uncompressed_blob[=none|raw|hex|detail]] "
"[--show_summary]\n");
return 0;
case 'f':
@@ -73,17 +70,6 @@ int main(int argc, char** argv) {
show_blob = DisplayType::kHex;
}
break;
case 'r':
if (optarg) {
if (display_types.count(arg_str) == 0) {
fprintf(stderr, "Unrecognized blob display type.\n");
return -1;
}
show_uncompressed_blob = display_types.at(arg_str);
} else {
show_uncompressed_blob = DisplayType::kHex;
}
break;
case 's':
show_summary = true;
break;
@@ -93,8 +79,7 @@ int main(int argc, char** argv) {
}
}
BlobDumpTool tool;
Status s =
tool.Run(file, show_key, show_blob, show_uncompressed_blob, show_summary);
Status s = tool.Run(file, show_key, show_blob, show_summary);
if (!s.ok()) {
fprintf(stderr, "Failed: %s\n", s.ToString().c_str());
return -1;
tools/db_bench_tool.cc
-11
View File
@@ -1087,12 +1087,6 @@ DEFINE_uint64(blob_db_file_size,
ROCKSDB_NAMESPACE::blob_db::BlobDBOptions().blob_file_size,
"[Stacked BlobDB] Target size of each blob file.");
DEFINE_string(
blob_db_compression_type, "snappy",
"[Stacked BlobDB] Algorithm to use to compress blobs in blob files.");
static enum ROCKSDB_NAMESPACE::CompressionType
FLAGS_blob_db_compression_type_e = ROCKSDB_NAMESPACE::kSnappyCompression;
// Integrated BlobDB options
DEFINE_bool(
enable_blob_files,
@@ -5204,7 +5198,6 @@ class Benchmark {
blob_db_options.min_blob_size = FLAGS_blob_db_min_blob_size;
blob_db_options.bytes_per_sync = FLAGS_blob_db_bytes_per_sync;
blob_db_options.blob_file_size = FLAGS_blob_db_file_size;
blob_db_options.compression = FLAGS_blob_db_compression_type_e;
blob_db::BlobDB* ptr = nullptr;
s = hooks.Open(options, blob_db_options, db_name, &ptr);
if (s.ok()) {
@@ -9191,10 +9184,6 @@ int db_bench_tool(int argc, char** argv, ToolHooks& hooks) {
FLAGS_compressed_secondary_cache_compression_type_e = StringToCompressionType(
FLAGS_compressed_secondary_cache_compression_type.c_str());
// Stacked BlobDB
FLAGS_blob_db_compression_type_e =
StringToCompressionType(FLAGS_blob_db_compression_type.c_str());
int env_opts = !FLAGS_env_uri.empty() + !FLAGS_fs_uri.empty();
if (env_opts > 1) {
fprintf(stderr, "Error: --env_uri and --fs_uri are mutually exclusive\n");
tools/ldb_cmd.cc
+15 -32
View File
@@ -110,8 +110,6 @@ const std::string LDBCommand::ARG_BLOB_FILE_STARTING_LEVEL =
const std::string LDBCommand::ARG_PREPOPULATE_BLOB_CACHE =
"prepopulate_blob_cache";
const std::string LDBCommand::ARG_DECODE_BLOB_INDEX = "decode_blob_index";
const std::string LDBCommand::ARG_DUMP_UNCOMPRESSED_BLOBS =
"dump_uncompressed_blobs";
const std::string LDBCommand::ARG_READ_TIMESTAMP = "read_timestamp";
const std::string LDBCommand::ARG_GET_WRITE_UNIX_TIME = "get_write_unix_time";
@@ -201,7 +199,7 @@ void DumpSstFile(Options options, std::string filename, bool output_hex,
std::string from_key = "", std::string to_key = "");
void DumpBlobFile(const std::string& filename, bool is_key_hex,
bool is_value_hex, bool dump_uncompressed_blobs);
bool is_value_hex);
Status EncodeUserProvidedTimestamp(const std::string& user_timestamp,
std::string* ts_buf);
@@ -2288,13 +2286,12 @@ DBDumperCommand::DBDumperCommand(
const std::vector<std::string>& /*params*/,
const std::map<std::string, std::string>& options,
const std::vector<std::string>& flags)
: LDBCommand(
options, flags, true /* is_read_only */,
BuildCmdLineOptions(
{ARG_TTL, ARG_HEX, ARG_KEY_HEX, ARG_VALUE_HEX, ARG_FROM, ARG_TO,
ARG_MAX_KEYS, ARG_COUNT_ONLY, ARG_COUNT_DELIM, ARG_STATS,
ARG_TTL_START, ARG_TTL_END, ARG_TTL_BUCKET, ARG_TIMESTAMP,
ARG_PATH, ARG_DECODE_BLOB_INDEX, ARG_DUMP_UNCOMPRESSED_BLOBS})),
: LDBCommand(options, flags, true /* is_read_only */,
BuildCmdLineOptions(
{ARG_TTL, ARG_HEX, ARG_KEY_HEX, ARG_VALUE_HEX, ARG_FROM,
ARG_TO, ARG_MAX_KEYS, ARG_COUNT_ONLY, ARG_COUNT_DELIM,
ARG_STATS, ARG_TTL_START, ARG_TTL_END, ARG_TTL_BUCKET,
ARG_TIMESTAMP, ARG_PATH, ARG_DECODE_BLOB_INDEX})),
null_from_(true),
null_to_(true),
max_keys_(-1),
@@ -2342,7 +2339,6 @@ DBDumperCommand::DBDumperCommand(
print_stats_ = IsFlagPresent(flags, ARG_STATS);
count_only_ = IsFlagPresent(flags, ARG_COUNT_ONLY);
decode_blob_index_ = IsFlagPresent(flags, ARG_DECODE_BLOB_INDEX);
dump_uncompressed_blobs_ = IsFlagPresent(flags, ARG_DUMP_UNCOMPRESSED_BLOBS);
if (is_key_hex_) {
if (!null_from_) {
@@ -2377,7 +2373,6 @@ void DBDumperCommand::Help(std::string& ret) {
ret.append(" [--" + ARG_TTL_END + "=<N>:- is exclusive]");
ret.append(" [--" + ARG_PATH + "=<path_to_a_file>]");
ret.append(" [--" + ARG_DECODE_BLOB_INDEX + "]");
ret.append(" [--" + ARG_DUMP_UNCOMPRESSED_BLOBS + "]");
ret.append("\n");
}
@@ -2424,8 +2419,7 @@ void DBDumperCommand::DoCommand() {
/* json_ */ false, column_families_);
break;
case kBlobFile:
DumpBlobFile(path_, is_key_hex_, is_value_hex_,
dump_uncompressed_blobs_);
DumpBlobFile(path_, is_key_hex_, is_value_hex_);
break;
default:
exec_state_ = LDBCommandExecuteResult::Failed(
@@ -4718,22 +4712,16 @@ void DumpSstFile(Options options, std::string filename, bool output_hex,
}
void DumpBlobFile(const std::string& filename, bool is_key_hex,
bool is_value_hex, bool dump_uncompressed_blobs) {
bool is_value_hex) {
using ROCKSDB_NAMESPACE::blob_db::BlobDumpTool;
BlobDumpTool tool;
BlobDumpTool::DisplayType blob_type = is_value_hex
BlobDumpTool::DisplayType show_blob = is_value_hex
? BlobDumpTool::DisplayType::kHex
: BlobDumpTool::DisplayType::kRaw;
BlobDumpTool::DisplayType show_uncompressed_blob =
dump_uncompressed_blobs ? blob_type : BlobDumpTool::DisplayType::kNone;
BlobDumpTool::DisplayType show_blob =
dump_uncompressed_blobs ? BlobDumpTool::DisplayType::kNone : blob_type;
BlobDumpTool::DisplayType show_key = is_key_hex
? BlobDumpTool::DisplayType::kHex
: BlobDumpTool::DisplayType::kRaw;
Status s = tool.Run(filename, show_key, show_blob, show_uncompressed_blob,
/* show_summary */ true);
Status s = tool.Run(filename, show_key, show_blob, /* show_summary */ true);
if (!s.ok()) {
fprintf(stderr, "Failed: %s\n", s.ToString().c_str());
}
@@ -4757,17 +4745,13 @@ DBFileDumperCommand::DBFileDumperCommand(
const std::map<std::string, std::string>& options,
const std::vector<std::string>& flags)
: LDBCommand(options, flags, true /* is_read_only */,
BuildCmdLineOptions(
{ARG_DECODE_BLOB_INDEX, ARG_DUMP_UNCOMPRESSED_BLOBS})),
decode_blob_index_(IsFlagPresent(flags, ARG_DECODE_BLOB_INDEX)),
dump_uncompressed_blobs_(
IsFlagPresent(flags, ARG_DUMP_UNCOMPRESSED_BLOBS)) {}
BuildCmdLineOptions({ARG_DECODE_BLOB_INDEX})),
decode_blob_index_(IsFlagPresent(flags, ARG_DECODE_BLOB_INDEX)) {}
void DBFileDumperCommand::Help(std::string& ret) {
ret.append(" ");
ret.append(DBFileDumperCommand::Name());
ret.append(" [--" + ARG_DECODE_BLOB_INDEX + "] ");
ret.append(" [--" + ARG_DUMP_UNCOMPRESSED_BLOBS + "] ");
ret.append(" [--" + ARG_DECODE_BLOB_INDEX + "]");
ret.append("\n");
}
@@ -4835,8 +4819,7 @@ void DBFileDumperCommand::DoCommand() {
filename = NormalizePath(filename);
std::cout << filename << std::endl;
std::cout << "------------------------------" << std::endl;
DumpBlobFile(filename, /* is_key_hex */ false, /* is_value_hex */ false,
dump_uncompressed_blobs_);
DumpBlobFile(filename, /* is_key_hex */ false, /* is_value_hex */ false);
std::cout << std::endl;
}
}
tools/ldb_cmd_impl.h
-2
View File
@@ -47,7 +47,6 @@ class DBFileDumperCommand : public LDBCommand {
private:
bool decode_blob_index_;
bool dump_uncompressed_blobs_;
};
class DBLiveFilesMetadataDumperCommand : public LDBCommand {
@@ -109,7 +108,6 @@ class DBDumperCommand : public LDBCommand {
bool print_stats_;
std::string path_;
bool decode_blob_index_;
bool dump_uncompressed_blobs_;
static const std::string ARG_COUNT_ONLY;
static const std::string ARG_COUNT_DELIM;
tools/ldb_test.py
+2 -2
View File
@@ -613,7 +613,7 @@ class LDBTestCase(unittest.TestCase):
# Call the dump_live_files function with the edited dbPath name.
self.assertTrue(
self.dumpLiveFiles(
"--db=%s --decode_blob_index --dump_uncompressed_blobs" % dbPath,
"--db=%s --decode_blob_index" % dbPath,
dumpFilePath,
)
)
@@ -881,7 +881,7 @@ class LDBTestCase(unittest.TestCase):
expected_pattern = re.compile(regex)
blob_files = self.getBlobFiles(dbPath)
self.assertTrue(len(blob_files) >= 1)
cmd = "dump --path=%s --dump_uncompressed_blobs"
cmd = "dump --path=%s"
self.assertRunOKFull(
(cmd) % (blob_files[0]), expected_pattern, unexpected=False, isPattern=True
)
utilities/blob_db/blob_compaction_filter.cc
+8 -59
View File
@@ -94,10 +94,7 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::FilterV2(
}
// Read value from blob file.
PinnableSlice blob;
CompressionType compression_type = kNoCompression;
constexpr bool need_decompress = true;
if (!ReadBlobFromOldFile(ikey.user_key, blob_index, &blob, need_decompress,
&compression_type)) {
if (!ReadBlobFromOldFile(ikey.user_key, blob_index, &blob)) {
return Decision::kIOError;
}
CompactionFilter::Decision decision = ucf->FilterV2(
@@ -123,15 +120,6 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::HandleValueChange(
return Decision::kIOError;
}
Slice new_blob_value(*new_value);
GrowableBuffer compressed_output;
if (blob_db_impl->bdb_options_.compression != kNoCompression) {
Status s = blob_db_impl->CompressBlob(new_blob_value, &compressed_output);
if (!s.ok()) {
// Best approximation
return Decision::kIOError;
}
new_blob_value = compressed_output.AsSlice();
}
uint64_t new_blob_file_number = 0;
uint64_t new_blob_offset = 0;
if (!WriteBlobToNewFile(key, new_blob_value, &new_blob_file_number,
@@ -142,8 +130,7 @@ CompactionFilter::Decision BlobIndexCompactionFilterBase::HandleValueChange(
return Decision::kIOError;
}
BlobIndex::EncodeBlob(new_value, new_blob_file_number, new_blob_offset,
new_blob_value.size(),
blob_db_impl->bdb_options_.compression);
new_blob_value.size(), kNoCompression);
return Decision::kChangeBlobIndex;
}
@@ -205,14 +192,13 @@ bool BlobIndexCompactionFilterBase::OpenNewBlobFileIfNeeded() const {
}
bool BlobIndexCompactionFilterBase::ReadBlobFromOldFile(
const Slice& key, const BlobIndex& blob_index, PinnableSlice* blob,
bool need_decompress, CompressionType* compression_type) const {
const Slice& key, const BlobIndex& blob_index, PinnableSlice* blob) const {
BlobDBImpl* const blob_db_impl = context_.blob_db_impl;
assert(blob_db_impl);
Status s = blob_db_impl->GetRawBlobFromFile(
key, blob_index.file_number(), blob_index.offset(), blob_index.size(),
blob, compression_type);
Status s = blob_db_impl->GetRawBlobFromFile(key, blob_index.file_number(),
blob_index.offset(),
blob_index.size(), blob);
if (!s.ok()) {
ROCKS_LOG_ERROR(
@@ -225,21 +211,6 @@ bool BlobIndexCompactionFilterBase::ReadBlobFromOldFile(
return false;
}
if (need_decompress && *compression_type != kNoCompression) {
s = blob_db_impl->DecompressSlice(*blob, *compression_type, blob);
if (!s.ok()) {
ROCKS_LOG_ERROR(
blob_db_impl->db_options_.info_log,
"Uncompression error during blob read from file: %" PRIu64
" blob_offset: %" PRIu64 " blob_size: %" PRIu64
" key: %s status: '%s'",
blob_index.file_number(), blob_index.offset(), blob_index.size(),
key.ToString(/* output_hex */ true).c_str(), s.ToString().c_str());
return false;
}
}
return true;
}
@@ -372,33 +343,11 @@ CompactionFilter::BlobDecision BlobIndexCompactionFilterGC::PrepareBlobOutput(
}
PinnableSlice blob;
CompressionType compression_type = kNoCompression;
GrowableBuffer compressed_output;
if (!ReadBlobFromOldFile(key, blob_index, &blob, false, &compression_type)) {
if (!ReadBlobFromOldFile(key, blob_index, &blob)) {
gc_stats_.SetError();
return BlobDecision::kIOError;
}
// If the compression_type is changed, re-compress it with the new compression
// type.
if (compression_type != blob_db_impl->bdb_options_.compression) {
if (compression_type != kNoCompression) {
const Status status =
blob_db_impl->DecompressSlice(blob, compression_type, &blob);
if (!status.ok()) {
gc_stats_.SetError();
return BlobDecision::kCorruption;
}
}
if (blob_db_impl->bdb_options_.compression != kNoCompression) {
s = blob_db_impl->CompressBlob(blob, &compressed_output);
if (!s.ok()) {
return BlobDecision::kCorruption;
}
blob.PinSelf(compressed_output.AsSlice());
}
}
uint64_t new_blob_file_number = 0;
uint64_t new_blob_offset = 0;
if (!WriteBlobToNewFile(key, blob, &new_blob_file_number, &new_blob_offset)) {
@@ -412,7 +361,7 @@ CompactionFilter::BlobDecision BlobIndexCompactionFilterGC::PrepareBlobOutput(
}
BlobIndex::EncodeBlob(new_value, new_blob_file_number, new_blob_offset,
blob.size(), compression_type);
blob.size(), kNoCompression);
gc_stats_.AddRelocatedBlob(blob_index.size());
utilities/blob_db/blob_compaction_filter.h
+1 -2
View File
@@ -59,8 +59,7 @@ class BlobIndexCompactionFilterBase : public LayeredCompactionFilterBase {
bool IsBlobFileOpened() const;
virtual bool OpenNewBlobFileIfNeeded() const;
bool ReadBlobFromOldFile(const Slice& key, const BlobIndex& blob_index,
PinnableSlice* blob, bool need_decompress,
CompressionType* compression_type) const;
PinnableSlice* blob) const;
bool WriteBlobToNewFile(const Slice& key, const Slice& blob,
uint64_t* new_blob_file_number,
uint64_t* new_blob_offset) const;
utilities/blob_db/blob_db.cc
-3
View File
@@ -92,9 +92,6 @@ void BlobDBOptions::Dump(Logger* log) const {
ROCKS_LOG_HEADER(
log, " BlobDBOptions.blob_file_size: %" PRIu64,
blob_file_size);
ROCKS_LOG_HEADER(
log, " BlobDBOptions.compression: %d",
static_cast<int>(compression));
ROCKS_LOG_HEADER(
log, " BlobDBOptions.enable_garbage_collection: %d",
enable_garbage_collection);
utilities/blob_db/blob_db.h
-3
View File
@@ -66,9 +66,6 @@ struct BlobDBOptions {
// after it exceeds that size
uint64_t blob_file_size = 256 * 1024 * 1024;
// what compression to use for Blob's
CompressionType compression = kNoCompression;
// If enabled, BlobDB cleans up stale blobs in non-TTL files during compaction
// by rewriting the remaining live blobs to new files.
bool enable_garbage_collection = false;
utilities/blob_db/blob_db_impl.cc
+11 -99
View File
@@ -83,10 +83,7 @@ BlobDBImpl::BlobDBImpl(const std::string& dbname,
live_sst_size_(0),
fifo_eviction_seq_(0),
evict_expiration_up_to_(0),
debug_level_(0),
// NOTE: returns nullptr for kNoCompression
blob_compressor_(GetBuiltinV2CompressionManager()->GetCompressor(
CompressionOptions{}, bdb_options_.compression)) {
debug_level_(0) {
clock_ = env_->GetSystemClock().get();
blob_dir_ = (bdb_options_.path_relative)
? dbname + "/" + bdb_options_.blob_dir
@@ -708,7 +705,7 @@ std::shared_ptr<BlobFile> BlobDBImpl::NewBlobFile(
static_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily())->GetID();
auto blob_file = std::make_shared<BlobFile>(
this, blob_dir_, file_num, db_options_.info_log.get(), column_family_id,
bdb_options_.compression, has_ttl, expiration_range);
has_ttl, expiration_range);
ROCKS_LOG_DEBUG(db_options_.info_log, "New blob file created: %s reason='%s'",
blob_file->PathName().c_str(), reason.c_str());
@@ -1095,32 +1092,14 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
RecordTick(statistics_, BLOB_DB_WRITE_INLINED_TTL);
}
} else {
GrowableBuffer compression_output;
Slice value_maybe_compressed;
if (blob_compressor_) {
assert(bdb_options_.compression != kNoCompression);
assert(bdb_options_.compression ==
blob_compressor_->GetPreferredCompressionType());
s = CompressBlob(value, &compression_output);
if (!s.ok()) {
return s;
}
value_maybe_compressed = compression_output.AsSlice();
} else {
assert(bdb_options_.compression == kNoCompression);
value_maybe_compressed = value;
}
std::string headerbuf;
BlobLogWriter::ConstructBlobHeader(&headerbuf, key, value_maybe_compressed,
expiration);
BlobLogWriter::ConstructBlobHeader(&headerbuf, key, value, expiration);
// Check DB size limit before selecting blob file to
// Since CheckSizeAndEvictBlobFiles() can close blob files, it needs to be
// done before calling SelectBlobFile().
s = CheckSizeAndEvictBlobFiles(
write_options,
headerbuf.size() + key.size() + value_maybe_compressed.size());
write_options, headerbuf.size() + key.size() + value.size());
if (!s.ok()) {
return s;
}
@@ -1133,9 +1112,8 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
}
if (s.ok()) {
assert(blob_file != nullptr);
assert(blob_file->GetCompressionType() == bdb_options_.compression);
s = AppendBlob(write_options, blob_file, headerbuf, key,
value_maybe_compressed, expiration, &index_entry);
s = AppendBlob(write_options, blob_file, headerbuf, key, value,
expiration, &index_entry);
}
if (s.ok()) {
if (expiration != kNoExpiration) {
@@ -1172,44 +1150,6 @@ Status BlobDBImpl::PutBlobValue(const WriteOptions& write_options,
return s;
}
Status BlobDBImpl::CompressBlob(const Slice& raw,
GrowableBuffer* compression_output) const {
StopWatch compression_sw(clock_, statistics_, BLOB_DB_COMPRESSION_MICROS);
return LegacyForceBuiltinCompression(
*blob_compressor_, /*working_area=*/nullptr, raw, compression_output);
}
Decompressor& BlobDecompressor() {
static auto decompressor =
GetBuiltinV2CompressionManager()->GetDecompressor();
return *decompressor;
}
Status BlobDBImpl::DecompressSlice(const Slice& compressed_value,
CompressionType compression_type,
PinnableSlice* value_output) const {
assert(compression_type != kNoCompression);
BlockContents contents;
auto cfh = static_cast<ColumnFamilyHandleImpl*>(DefaultColumnFamily());
{
StopWatch decompression_sw(clock_, statistics_,
BLOB_DB_DECOMPRESSION_MICROS);
Status s = DecompressBlockData(
compressed_value.data(), compressed_value.size(), compression_type,
BlobDecompressor(), &contents, cfh->cfd()->ioptions());
if (!s.ok()) {
return Status::Corruption("Unable to decompress blob.");
}
}
value_output->PinSelf(contents.data);
return Status::OK();
}
Status BlobDBImpl::CompactFiles(
const CompactionOptions& compact_options,
const std::vector<std::string>& input_file_names, const int output_level,
@@ -1409,11 +1349,10 @@ Status BlobDBImpl::AppendBlob(const WriteOptions& write_options,
if (expiration == kNoExpiration) {
BlobIndex::EncodeBlob(index_entry, bfile->BlobFileNumber(), blob_offset,
value.size(), bdb_options_.compression);
value.size(), kNoCompression);
} else {
BlobIndex::EncodeBlobTTL(index_entry, expiration, bfile->BlobFileNumber(),
blob_offset, value.size(),
bdb_options_.compression);
blob_offset, value.size(), kNoCompression);
}
return s;
@@ -1511,39 +1450,14 @@ Status BlobDBImpl::GetBlobValue(const Slice& key, const Slice& index_entry,
return Status::OK();
}
CompressionType compression_type = kNoCompression;
s = GetRawBlobFromFile(key, blob_index.file_number(), blob_index.offset(),
blob_index.size(), value, &compression_type);
if (!s.ok()) {
return s;
}
if (compression_type != kNoCompression) {
s = DecompressSlice(*value, compression_type, value);
if (!s.ok()) {
if (debug_level_ >= 2) {
ROCKS_LOG_ERROR(
db_options_.info_log,
"Uncompression error during blob read from file: %" PRIu64
" blob_offset: %" PRIu64 " blob_size: %" PRIu64
" key: %s status: '%s'",
blob_index.file_number(), blob_index.offset(), blob_index.size(),
key.ToString(/* output_hex */ true).c_str(), s.ToString().c_str());
}
return s;
}
}
return Status::OK();
return GetRawBlobFromFile(key, blob_index.file_number(), blob_index.offset(),
blob_index.size(), value);
}
Status BlobDBImpl::GetRawBlobFromFile(const Slice& key, uint64_t file_number,
uint64_t offset, uint64_t size,
PinnableSlice* value,
CompressionType* compression_type) {
PinnableSlice* value) {
assert(value);
assert(compression_type);
assert(*compression_type == kNoCompression);
if (!size) {
value->PinSelf("");
@@ -1581,8 +1495,6 @@ Status BlobDBImpl::GetRawBlobFromFile(const Slice& key, uint64_t file_number,
blob_file = it->second;
}
*compression_type = blob_file->GetCompressionType();
// takes locks when called
std::shared_ptr<RandomAccessFileReader> reader;
Status s = GetBlobFileReader(blob_file, &reader);
utilities/blob_db/blob_db_impl.h
+1 -13
View File
@@ -227,15 +227,7 @@ class BlobDBImpl : public BlobDB {
Status GetRawBlobFromFile(const Slice& key, uint64_t file_number,
uint64_t offset, uint64_t size,
PinnableSlice* value,
CompressionType* compression_type);
Status CompressBlob(const Slice& raw,
GrowableBuffer* compression_output) const;
Status DecompressSlice(const Slice& compressed_value,
CompressionType compression_type,
PinnableSlice* value_output) const;
PinnableSlice* value);
// Close a file by appending a footer, and removes file from open files list.
// REQUIRES: lock held on write_mutex_, write lock held on both the db mutex_
@@ -503,11 +495,7 @@ class BlobDBImpl : public BlobDB {
int disable_file_deletions_ = 0;
uint32_t debug_level_;
std::unique_ptr<Compressor> blob_compressor_;
};
Decompressor& BlobDecompressor();
} // namespace blob_db
} // namespace ROCKSDB_NAMESPACE
utilities/blob_db/blob_db_test.cc
-206
View File
@@ -481,208 +481,6 @@ TEST_F(BlobDBTest, Override) {
VerifyDB(data);
}
#ifdef SNAPPY
TEST_F(BlobDBTest, Compression) {
Random rnd(301);
BlobDBOptions bdb_options;
bdb_options.min_blob_size = 0;
bdb_options.disable_background_tasks = true;
bdb_options.compression = CompressionType::kSnappyCompression;
Open(bdb_options);
std::map<std::string, std::string> data;
for (size_t i = 0; i < 100; i++) {
PutRandom("put-key" + std::to_string(i), &rnd, &data);
}
for (int i = 0; i < 100; i++) {
WriteBatch batch;
for (size_t j = 0; j < 10; j++) {
PutRandomToWriteBatch("write-batch-key" + std::to_string(j * 100 + i),
&rnd, &batch, &data);
}
ASSERT_OK(blob_db_->Write(WriteOptions(), &batch));
}
VerifyDB(data);
}
TEST_F(BlobDBTest, DecompressAfterReopen) {
Random rnd(301);
BlobDBOptions bdb_options;
bdb_options.min_blob_size = 0;
bdb_options.disable_background_tasks = true;
bdb_options.compression = CompressionType::kSnappyCompression;
Open(bdb_options);
std::map<std::string, std::string> data;
for (size_t i = 0; i < 100; i++) {
PutRandom("put-key" + std::to_string(i), &rnd, &data);
}
VerifyDB(data);
bdb_options.compression = CompressionType::kNoCompression;
Reopen(bdb_options);
VerifyDB(data);
}
TEST_F(BlobDBTest, EnableDisableCompressionGC) {
Random rnd(301);
BlobDBOptions bdb_options;
bdb_options.min_blob_size = 0;
bdb_options.garbage_collection_cutoff = 1.0;
bdb_options.disable_background_tasks = true;
bdb_options.compression = kSnappyCompression;
Open(bdb_options);
std::map<std::string, std::string> data;
size_t data_idx = 0;
for (; data_idx < 100; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
auto blob_files = blob_db_impl()->TEST_GetBlobFiles();
ASSERT_EQ(1, blob_files.size());
ASSERT_EQ(kSnappyCompression, blob_files[0]->GetCompressionType());
// disable compression
bdb_options.compression = kNoCompression;
Reopen(bdb_options);
// Add more data with new compression type
for (; data_idx < 200; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
blob_files = blob_db_impl()->TEST_GetBlobFiles();
ASSERT_EQ(2, blob_files.size());
ASSERT_EQ(kNoCompression, blob_files[1]->GetCompressionType());
// Enable GC. If we do it earlier the snapshot release triggered compaction
// may compact files and trigger GC before we can verify there are two files.
bdb_options.enable_garbage_collection = true;
Reopen(bdb_options);
// Trigger compaction
ASSERT_OK(blob_db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
blob_db_impl()->TEST_DeleteObsoleteFiles();
VerifyDB(data);
blob_files = blob_db_impl()->TEST_GetBlobFiles();
for (const auto& bfile : blob_files) {
ASSERT_EQ(kNoCompression, bfile->GetCompressionType());
}
// enabling the compression again
bdb_options.compression = kSnappyCompression;
Reopen(bdb_options);
// Add more data with new compression type
for (; data_idx < 300; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
// Trigger compaction
ASSERT_OK(blob_db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
blob_db_impl()->TEST_DeleteObsoleteFiles();
VerifyDB(data);
blob_files = blob_db_impl()->TEST_GetBlobFiles();
for (const auto& bfile : blob_files) {
ASSERT_EQ(kSnappyCompression, bfile->GetCompressionType());
}
}
#ifdef LZ4
// Test switch compression types and run GC, it needs both Snappy and LZ4
// support.
TEST_F(BlobDBTest, ChangeCompressionGC) {
Random rnd(301);
BlobDBOptions bdb_options;
bdb_options.min_blob_size = 0;
bdb_options.garbage_collection_cutoff = 1.0;
bdb_options.disable_background_tasks = true;
bdb_options.compression = kLZ4Compression;
Open(bdb_options);
std::map<std::string, std::string> data;
size_t data_idx = 0;
for (; data_idx < 100; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
auto blob_files = blob_db_impl()->TEST_GetBlobFiles();
ASSERT_EQ(1, blob_files.size());
ASSERT_EQ(kLZ4Compression, blob_files[0]->GetCompressionType());
// Change compression type
bdb_options.compression = kSnappyCompression;
Reopen(bdb_options);
// Add more data with Snappy compression type
for (; data_idx < 200; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
// Verify blob file compression type
blob_files = blob_db_impl()->TEST_GetBlobFiles();
ASSERT_EQ(2, blob_files.size());
ASSERT_EQ(kSnappyCompression, blob_files[1]->GetCompressionType());
// Enable GC. If we do it earlier the snapshot release triggered compaction
// may compact files and trigger GC before we can verify there are two files.
bdb_options.enable_garbage_collection = true;
Reopen(bdb_options);
ASSERT_OK(blob_db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
VerifyDB(data);
blob_db_impl()->TEST_DeleteObsoleteFiles();
blob_files = blob_db_impl()->TEST_GetBlobFiles();
for (const auto& bfile : blob_files) {
ASSERT_EQ(kSnappyCompression, bfile->GetCompressionType());
}
// Disable compression
bdb_options.compression = kNoCompression;
Reopen(bdb_options);
for (; data_idx < 300; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
ASSERT_OK(blob_db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
VerifyDB(data);
blob_db_impl()->TEST_DeleteObsoleteFiles();
blob_files = blob_db_impl()->TEST_GetBlobFiles();
for (const auto& bfile : blob_files) {
ASSERT_EQ(kNoCompression, bfile->GetCompressionType());
}
// switching different compression types to generate mixed compression types
bdb_options.compression = kSnappyCompression;
Reopen(bdb_options);
for (; data_idx < 400; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
bdb_options.compression = kLZ4Compression;
Reopen(bdb_options);
for (; data_idx < 500; data_idx++) {
PutRandom("put-key" + std::to_string(data_idx), &rnd, &data);
}
VerifyDB(data);
ASSERT_OK(blob_db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
VerifyDB(data);
blob_db_impl()->TEST_DeleteObsoleteFiles();
blob_files = blob_db_impl()->TEST_GetBlobFiles();
for (const auto& bfile : blob_files) {
ASSERT_EQ(kLZ4Compression, bfile->GetCompressionType());
}
}
#endif // LZ4
#endif // SNAPPY
TEST_F(BlobDBTest, MultipleWriters) {
Open(BlobDBOptions());
@@ -1358,9 +1156,6 @@ TEST_F(BlobDBTest, UserCompactionFilter) {
bdb_options.min_blob_size = kMinBlobSize;
bdb_options.blob_file_size = kMaxValueSize * 10;
bdb_options.disable_background_tasks = true;
if (Snappy_Supported()) {
bdb_options.compression = CompressionType::kSnappyCompression;
}
// case_num == 0: Test user defined compaction filter
// case_num == 1: Test user defined compaction filter factory
for (int case_num = 0; case_num < 2; case_num++) {
@@ -1440,7 +1235,6 @@ TEST_F(BlobDBTest, UserCompactionFilter_BlobIOError) {
bdb_options.min_blob_size = 0;
bdb_options.blob_file_size = kValueSize * 10;
bdb_options.disable_background_tasks = true;
bdb_options.compression = CompressionType::kNoCompression;
std::vector<std::string> io_failure_cases = {
"BlobDBImpl::CreateBlobFileAndWriter",
utilities/blob_db/blob_dump_tool.cc
+8 -51
View File
@@ -16,10 +16,8 @@
#include "port/port.h"
#include "rocksdb/convenience.h"
#include "rocksdb/file_system.h"
#include "table/format.h"
#include "util/coding.h"
#include "util/string_util.h"
#include "utilities/blob_db/blob_db_impl.h"
namespace ROCKSDB_NAMESPACE::blob_db {
@@ -27,9 +25,7 @@ BlobDumpTool::BlobDumpTool()
: reader_(nullptr), buffer_(nullptr), buffer_size_(0) {}
Status BlobDumpTool::Run(const std::string& filename, DisplayType show_key,
DisplayType show_blob,
DisplayType show_uncompressed_blob,
bool show_summary) {
DisplayType show_blob, bool show_summary) {
constexpr size_t kReadaheadSize = 2 * 1024 * 1024;
Status s;
const auto fs = FileSystem::Default();
@@ -55,8 +51,7 @@ Status BlobDumpTool::Run(const std::string& filename, DisplayType show_key,
reader_.reset(new RandomAccessFileReader(std::move(file), filename));
uint64_t offset = 0;
uint64_t footer_offset = 0;
CompressionType compression = kNoCompression;
s = DumpBlobLogHeader(&offset, &compression);
s = DumpBlobLogHeader(&offset);
if (!s.ok()) {
return s;
}
@@ -67,12 +62,10 @@ Status BlobDumpTool::Run(const std::string& filename, DisplayType show_key,
uint64_t total_records = 0;
uint64_t total_key_size = 0;
uint64_t total_blob_size = 0;
uint64_t total_uncompressed_blob_size = 0;
if (show_key != DisplayType::kNone || show_summary) {
if (show_key != DisplayType::kNone) {
while (offset < footer_offset) {
s = DumpRecord(show_key, show_blob, show_uncompressed_blob, show_summary,
compression, &offset, &total_records, &total_key_size,
&total_blob_size, &total_uncompressed_blob_size);
s = DumpRecord(show_key, show_blob, &offset, &total_records,
&total_key_size, &total_blob_size);
if (!s.ok()) {
break;
}
@@ -83,10 +76,6 @@ Status BlobDumpTool::Run(const std::string& filename, DisplayType show_key,
fprintf(stdout, " total records: %" PRIu64 "\n", total_records);
fprintf(stdout, " total key size: %" PRIu64 "\n", total_key_size);
fprintf(stdout, " total blob size: %" PRIu64 "\n", total_blob_size);
if (compression != kNoCompression) {
fprintf(stdout, " total raw blob size: %" PRIu64 "\n",
total_uncompressed_blob_size);
}
}
return s;
}
@@ -112,8 +101,7 @@ Status BlobDumpTool::Read(uint64_t offset, size_t size, Slice* result) {
return s;
}
Status BlobDumpTool::DumpBlobLogHeader(uint64_t* offset,
CompressionType* compression) {
Status BlobDumpTool::DumpBlobLogHeader(uint64_t* offset) {
Slice slice;
Status s = Read(0, BlobLogHeader::kSize, &slice);
if (!s.ok()) {
@@ -128,17 +116,10 @@ Status BlobDumpTool::DumpBlobLogHeader(uint64_t* offset,
fprintf(stdout, " Version : %" PRIu32 "\n", header.version);
fprintf(stdout, " Column Family ID : %" PRIu32 "\n",
header.column_family_id);
std::string compression_str;
if (!GetStringFromCompressionType(&compression_str, header.compression)
.ok()) {
compression_str = "Unrecongnized compression type (" +
std::to_string((int)header.compression) + ")";
}
fprintf(stdout, " Compression : %s\n", compression_str.c_str());
fprintf(stdout, " Compression : kNoCompression\n");
fprintf(stdout, " Expiration range : %s\n",
GetString(header.expiration_range).c_str());
*offset = BlobLogHeader::kSize;
*compression = header.compression;
return s;
}
@@ -171,12 +152,9 @@ Status BlobDumpTool::DumpBlobLogFooter(uint64_t file_size,
}
Status BlobDumpTool::DumpRecord(DisplayType show_key, DisplayType show_blob,
DisplayType show_uncompressed_blob,
bool show_summary, CompressionType compression,
uint64_t* offset, uint64_t* total_records,
uint64_t* total_key_size,
uint64_t* total_blob_size,
uint64_t* total_uncompressed_blob_size) {
uint64_t* total_blob_size) {
if (show_key != DisplayType::kNone) {
fprintf(stdout, "Read record with offset 0x%" PRIx64 " (%" PRIu64 "):\n",
*offset, *offset);
@@ -203,22 +181,6 @@ Status BlobDumpTool::DumpRecord(DisplayType show_key, DisplayType show_blob,
if (!s.ok()) {
return s;
}
// Decompress value
std::string uncompressed_value;
if (compression != kNoCompression &&
(show_uncompressed_blob != DisplayType::kNone || show_summary)) {
BlockContents contents;
UncompressionContext context(compression);
UncompressionInfo info(context, UncompressionDict::GetEmptyDict(),
compression);
s = DecompressBlockData(
slice.data() + key_size, static_cast<size_t>(value_size), compression,
BlobDecompressor(), &contents, ImmutableOptions(Options()));
if (!s.ok()) {
return s;
}
uncompressed_value = contents.data.ToString();
}
if (show_key != DisplayType::kNone) {
fprintf(stdout, " key : ");
DumpSlice(Slice(slice.data(), static_cast<size_t>(key_size)), show_key);
@@ -228,16 +190,11 @@ Status BlobDumpTool::DumpRecord(DisplayType show_key, DisplayType show_blob,
static_cast<size_t>(value_size)),
show_blob);
}
if (show_uncompressed_blob != DisplayType::kNone) {
fprintf(stdout, " raw blob : ");
DumpSlice(Slice(uncompressed_value), show_uncompressed_blob);
}
}
*offset += key_size + value_size;
*total_records += 1;
*total_key_size += key_size;
*total_blob_size += value_size;
*total_uncompressed_blob_size += uncompressed_value.size();
return s;
}
utilities/blob_db/blob_dump_tool.h
+4 -8
View File
@@ -28,8 +28,7 @@ class BlobDumpTool {
BlobDumpTool();
Status Run(const std::string& filename, DisplayType show_key,
DisplayType show_blob, DisplayType show_uncompressed_blob,
bool show_summary);
DisplayType show_blob, bool show_summary);
private:
std::unique_ptr<RandomAccessFileReader> reader_;
@@ -37,14 +36,11 @@ class BlobDumpTool {
size_t buffer_size_;
Status Read(uint64_t offset, size_t size, Slice* result);
Status DumpBlobLogHeader(uint64_t* offset, CompressionType* compression);
Status DumpBlobLogHeader(uint64_t* offset);
Status DumpBlobLogFooter(uint64_t file_size, uint64_t* footer_offset);
Status DumpRecord(DisplayType show_key, DisplayType show_blob,
DisplayType show_uncompressed_blob, bool show_summary,
CompressionType compression, uint64_t* offset,
uint64_t* total_records, uint64_t* total_key_size,
uint64_t* total_blob_size,
uint64_t* total_uncompressed_blob_size);
uint64_t* offset, uint64_t* total_records,
uint64_t* total_key_size, uint64_t* total_blob_size);
void DumpSlice(const Slice s, DisplayType type);
template <class T>
utilities/blob_db/blob_file.cc
+2 -5
View File
@@ -25,18 +25,16 @@ BlobFile::BlobFile(const BlobDBImpl* p, const std::string& bdir, uint64_t fn,
: parent_(p), path_to_dir_(bdir), file_number_(fn), info_log_(info_log) {}
BlobFile::BlobFile(const BlobDBImpl* p, const std::string& bdir, uint64_t fn,
Logger* info_log, uint32_t column_family_id,
CompressionType compression, bool has_ttl,
Logger* info_log, uint32_t column_family_id, bool has_ttl,
const ExpirationRange& expiration_range)
: parent_(p),
path_to_dir_(bdir),
file_number_(fn),
info_log_(info_log),
column_family_id_(column_family_id),
compression_(compression),
has_ttl_(has_ttl),
expiration_range_(expiration_range),
header_(column_family_id, compression, has_ttl, expiration_range),
header_(column_family_id, kNoCompression, has_ttl, expiration_range),
header_valid_(true) {}
BlobFile::~BlobFile() {
@@ -259,7 +257,6 @@ Status BlobFile::ReadMetadata(const std::shared_ptr<FileSystem>& fs,
return s;
}
column_family_id_ = header.column_family_id;
compression_ = header.compression;
has_ttl_ = header.has_ttl;
if (has_ttl_) {
expiration_range_ = header.expiration_range;
utilities/blob_db/blob_file.h
+1 -7
View File
@@ -51,9 +51,6 @@ class BlobFile {
// Column family id.
uint32_t column_family_id_{std::numeric_limits<uint32_t>::max()};
// Compression type of blobs in the file
CompressionType compression_{kNoCompression};
// If true, the keys in this file all has TTL. Otherwise all keys don't
// have TTL.
bool has_ttl_{false};
@@ -108,8 +105,7 @@ class BlobFile {
Logger* info_log);
BlobFile(const BlobDBImpl* parent, const std::string& bdir, uint64_t fnum,
Logger* info_log, uint32_t column_family_id,
CompressionType compression, bool has_ttl,
Logger* info_log, uint32_t column_family_id, bool has_ttl,
const ExpirationRange& expiration_range);
~BlobFile();
@@ -201,8 +197,6 @@ class BlobFile {
void SetHasTTL(bool has_ttl) { has_ttl_ = has_ttl; }
CompressionType GetCompressionType() const { return compression_; }
std::shared_ptr<BlobLogWriter> GetWriter() const { return log_writer_; }
// Read blob file header and footer. Return corruption if file header is