mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-08 07:05:23 +08:00
Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2888fa95b5 | |||
| b0e20194ea | |||
| c377c2ba15 | |||
| 531a5f88a1 | |||
| 10196d7edc | |||
| 90e245697f | |||
| c871142988 | |||
| 4c41e51c07 | |||
| 00803d619c | |||
| 83031e7343 | |||
| 296b47db25 | |||
| b1f62be10e | |||
| d89483098f | |||
| c861fb390d | |||
| fd00f39f97 | |||
| bb75092574 | |||
| 8972dd1ffa | |||
| 139778dfb3 | |||
| 728e5f5750 | |||
| 48cd7a3aae | |||
| 0be89e87fd |
@@ -326,12 +326,23 @@ CHECK_CXX_SOURCE_COMPILES("
|
||||
#endif
|
||||
int main() {
|
||||
static __thread int tls;
|
||||
(void)tls;
|
||||
}
|
||||
" HAVE_THREAD_LOCAL)
|
||||
if(HAVE_THREAD_LOCAL)
|
||||
add_definitions(-DROCKSDB_SUPPORT_THREAD_LOCAL)
|
||||
endif()
|
||||
|
||||
option(WITH_IOSTATS_CONTEXT "Enable IO stats context" ON)
|
||||
if (NOT WITH_IOSTATS_CONTEXT)
|
||||
add_definitions(-DNIOSTATS_CONTEXT)
|
||||
endif()
|
||||
|
||||
option(WITH_PERF_CONTEXT "Enable perf context" ON)
|
||||
if (NOT WITH_PERF_CONTEXT)
|
||||
add_definitions(-DNPERF_CONTEXT)
|
||||
endif()
|
||||
|
||||
option(FAIL_ON_WARNINGS "Treat compile warnings as errors" ON)
|
||||
if(FAIL_ON_WARNINGS)
|
||||
if(MSVC)
|
||||
|
||||
+11
-2
@@ -1,9 +1,10 @@
|
||||
# Rocksdb Change Log
|
||||
## Unreleased
|
||||
## 6.20.0 (04/16/2021)
|
||||
### Behavior Changes
|
||||
* `ColumnFamilyOptions::sample_for_compression` now takes effect for creation of all block-based tables. Previously it only took effect for block-based tables created by flush.
|
||||
* `CompactFiles()` can no longer compact files from lower level to up level, which has the risk to corrupt DB (details: #8063). The validation is also added to all compactions.
|
||||
* Fixed some cases in which DB::OpenForReadOnly() could write to the filesystem. If you want a Logger with a read-only DB, you must now set DBOptions::info_log yourself, such as using CreateLoggerFromOptions().
|
||||
* get_iostats_context() will never return nullptr. If thread-local support is not available, and user does not opt-out iostats context, then compilation will fail. The same applies to perf context as well.
|
||||
|
||||
### Bug Fixes
|
||||
* Use thread-safe `strerror_r()` to get error messages.
|
||||
@@ -11,6 +12,8 @@
|
||||
* Made BackupEngine thread-safe and added documentation comments to clarify what is safe for multiple BackupEngine objects accessing the same backup directory.
|
||||
* Fixed crash (divide by zero) when compression dictionary is applied to a file containing only range tombstones.
|
||||
* Fixed a backward iteration bug with partitioned filter enabled: not including the prefix of the last key of the previous filter partition in current filter partition can cause wrong iteration result.
|
||||
* Fixed a bug that allowed `DBOptions::max_open_files` to be set with a non-negative integer with `ColumnFamilyOptions::compaction_style = kCompactionStyleFIFO`.
|
||||
* Fix a bug that prevents newly ingested files from being written with incorrect smallest/largest key metadata. Reading from a DB with pre-existing ingested files could still be problematic.
|
||||
|
||||
### Performance Improvements
|
||||
* On ARM platform, use `yield` instead of `wfe` to relax cpu to gain better performance.
|
||||
@@ -18,10 +21,16 @@
|
||||
### Public API change
|
||||
* Added `TableProperties::slow_compression_estimated_data_size` and `TableProperties::fast_compression_estimated_data_size`. When `ColumnFamilyOptions::sample_for_compression > 0`, they estimate what `TableProperties::data_size` would have been if the "fast" or "slow" (see `ColumnFamilyOptions::sample_for_compression` API doc for definitions) compression had been used instead.
|
||||
* Update DB::StartIOTrace and remove Env object from the arguments as its redundant and DB already has Env object that is passed down to IOTracer::StartIOTrace
|
||||
* For new integrated BlobDB, add support for blob files for backup/restore like table files. Because of current limitations, blob files always use the kLegacyCrc32cAndFileSize naming scheme, and incremental backups must read and checksum all blob files in a DB, even for files that are already backed up.
|
||||
* Added `FlushReason::kWalFull`, which is reported when a memtable is flushed due to the WAL reaching its size limit; those flushes were previously reported as `FlushReason::kWriteBufferManager`. Also, changed the reason for flushes triggered by the write buffer manager to `FlushReason::kWriteBufferManager`; they were previously reported as `FlushReason::kWriteBufferFull`.
|
||||
* Extend file_checksum_dump ldb command and DB::GetLiveFilesChecksumInfo API for IntegratedBlobDB and get checksum of blob files along with SST files.
|
||||
|
||||
### New Features
|
||||
* Added the ability to open BackupEngine backups as read-only DBs, using BackupInfo::name_for_open and env_for_open provided by BackupEngine::GetBackupInfo() with include_file_details=true.
|
||||
* Added BackupEngine support for integrated BlobDB, with blob files shared between backups when table files are shared. Because of current limitations, blob files always use the kLegacyCrc32cAndFileSize naming scheme, and incremental backups must read and checksum all blob files in a DB, even for files that are already backed up.
|
||||
* Added an optional output parameter to BackupEngine::CreateNewBackup(WithMetadata) to return the BackupID of the new backup.
|
||||
* Added BackupEngine::GetBackupInfo / GetLatestBackupInfo for querying individual backups.
|
||||
* Made the Ribbon filter a long-term supported feature in terms of the SST schema(compatible with version >= 6.15.0) though the API for enabling it is expected to change.
|
||||
* Added hybrid configuration of Ribbon filter and Bloom filter where some LSM levels use Ribbon for memory space efficiency and some use Bloom for speed. See NewExperimentalRibbonFilterPolicy. This also changes the default behavior of NewExperimentalRibbonFilterPolicy to use Bloom on level 0 and Ribbon on later levels.
|
||||
|
||||
## 6.19.0 (03/21/2021)
|
||||
### Bug Fixes
|
||||
|
||||
@@ -55,50 +55,25 @@ DEBUG_LEVEL?=1
|
||||
# Set the default LIB_MODE to static
|
||||
LIB_MODE?=static
|
||||
|
||||
ifeq ($(MAKECMDGOALS),dbg)
|
||||
# OBJ_DIR is where the object files reside. Default to the current directory
|
||||
OBJ_DIR?=.
|
||||
|
||||
# Check the MAKECMDGOALS to set the DEBUG_LEVEL and LIB_MODE appropriately
|
||||
|
||||
ifneq ($(filter clean release install, $(MAKECMDGOALS)),)
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
ifneq ($(filter dbg, $(MAKECMDGOALS)),)
|
||||
DEBUG_LEVEL=2
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),clean)
|
||||
else ifneq ($(filter shared_lib install-shared, $(MAKECMDGOALS)),)
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),release)
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),shared_lib)
|
||||
LIB_MODE=shared
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),install-shared)
|
||||
LIB_MODE=shared
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),static_lib)
|
||||
else ifneq ($(filter static_lib install-static, $(MAKECMDGOALS)),)
|
||||
DEBUG_LEVEL=0
|
||||
LIB_MODE=static
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),install-static)
|
||||
DEBUG_LEVEL=0
|
||||
LIB_MODE=static
|
||||
endif
|
||||
|
||||
ifeq ($(MAKECMDGOALS),install)
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
|
||||
|
||||
ifneq ($(findstring jtest, $(MAKECMDGOALS)),)
|
||||
else ifneq ($(filter jtest rocksdbjava%, $(MAKECMDGOALS)),)
|
||||
OBJ_DIR=jl
|
||||
LIB_MODE=shared
|
||||
endif
|
||||
|
||||
ifneq ($(findstring rocksdbjava, $(MAKECMDGOALS)),)
|
||||
LIB_MODE=shared
|
||||
ifneq ($(findstring rocksdbjavastatic, $(MAKECMDGOALS)),)
|
||||
OBJ_DIR=jls
|
||||
ifneq ($(DEBUG_LEVEL),2)
|
||||
@@ -107,8 +82,6 @@ ifneq ($(findstring rocksdbjava, $(MAKECMDGOALS)),)
|
||||
ifeq ($(MAKECMDGOALS),rocksdbjavastaticpublish)
|
||||
DEBUG_LEVEL=0
|
||||
endif
|
||||
else
|
||||
OBJ_DIR=jl
|
||||
endif
|
||||
endif
|
||||
|
||||
@@ -491,7 +464,6 @@ CXXFLAGS += $(WARNING_FLAGS) -I. -I./include $(PLATFORM_CXXFLAGS) $(OPT) -Woverl
|
||||
|
||||
LDFLAGS += $(PLATFORM_LDFLAGS)
|
||||
|
||||
OBJ_DIR?=.
|
||||
LIB_OBJECTS = $(patsubst %.cc, $(OBJ_DIR)/%.o, $(LIB_SOURCES))
|
||||
LIB_OBJECTS += $(patsubst %.cc, $(OBJ_DIR)/%.o, $(ROCKSDB_PLUGIN_SOURCES))
|
||||
ifeq ($(HAVE_POWER8),1)
|
||||
@@ -651,8 +623,8 @@ else
|
||||
LIBRARY=$(STATIC_LIBRARY)
|
||||
TEST_LIBRARY=$(STATIC_TEST_LIBRARY)
|
||||
TOOLS_LIBRARY=$(STATIC_TOOLS_LIBRARY)
|
||||
STRESS_LIBRARY=$(STATIC_STRESS_LIBRARY)
|
||||
endif
|
||||
STRESS_LIBRARY=$(STATIC_STRESS_LIBRARY)
|
||||
|
||||
ROCKSDB_MAJOR = $(shell egrep "ROCKSDB_MAJOR.[0-9]" include/rocksdb/version.h | cut -d ' ' -f 3)
|
||||
ROCKSDB_MINOR = $(shell egrep "ROCKSDB_MINOR.[0-9]" include/rocksdb/version.h | cut -d ' ' -f 3)
|
||||
@@ -1246,7 +1218,7 @@ $(STATIC_TOOLS_LIBRARY): $(TOOL_OBJECTS)
|
||||
$(AM_V_AR)rm -f $@ $(SHARED_TOOLS_LIBRARY)
|
||||
$(AM_V_at)$(AR) $(ARFLAGS) $@ $^
|
||||
|
||||
$(STATIC_STRESS_LIBRARY): $(ANALYZE_OBJECTS) $(STRESS_OBJECTS)
|
||||
$(STATIC_STRESS_LIBRARY): $(ANALYZE_OBJECTS) $(STRESS_OBJECTS) $(TESTUTIL)
|
||||
$(AM_V_AR)rm -f $@ $(SHARED_STRESS_LIBRARY)
|
||||
$(AM_V_at)$(AR) $(ARFLAGS) $@ $^
|
||||
|
||||
@@ -1258,7 +1230,7 @@ $(SHARED_TOOLS_LIBRARY): $(TOOL_OBJECTS) $(SHARED1)
|
||||
$(AM_V_AR)rm -f $@ $(STATIC_TOOLS_LIBRARY)
|
||||
$(AM_SHARE)
|
||||
|
||||
$(SHARED_STRESS_LIBRARY): $(ANALYZE_OBJECTS) $(STRESS_OBJECTS) $(SHARED_TOOLS_LIBRARY) $(SHARED1)
|
||||
$(SHARED_STRESS_LIBRARY): $(ANALYZE_OBJECTS) $(STRESS_OBJECTS) $(TESTUTIL) $(SHARED_TOOLS_LIBRARY) $(SHARED1)
|
||||
$(AM_V_AR)rm -f $@ $(STATIC_STRESS_LIBRARY)
|
||||
$(AM_SHARE)
|
||||
|
||||
@@ -1292,7 +1264,7 @@ memtablerep_bench: $(OBJ_DIR)/memtable/memtablerep_bench.o $(LIBRARY)
|
||||
filter_bench: $(OBJ_DIR)/util/filter_bench.o $(LIBRARY)
|
||||
$(AM_LINK)
|
||||
|
||||
db_stress: $(OBJ_DIR)/db_stress_tool/db_stress.o $(STRESS_LIBRARY) $(TOOLS_LIBRARY) $(TESTUTIL) $(LIBRARY)
|
||||
db_stress: $(OBJ_DIR)/db_stress_tool/db_stress.o $(STRESS_LIBRARY) $(TOOLS_LIBRARY) $(LIBRARY)
|
||||
$(AM_LINK)
|
||||
|
||||
write_stress: $(OBJ_DIR)/tools/write_stress.o $(LIBRARY)
|
||||
|
||||
@@ -340,6 +340,7 @@ cpp_library(
|
||||
"util/compression_context_cache.cc",
|
||||
"util/concurrent_task_limiter_impl.cc",
|
||||
"util/crc32c.cc",
|
||||
"util/crc32c_arm64.cc",
|
||||
"util/dynamic_bloom.cc",
|
||||
"util/file_checksum_helper.cc",
|
||||
"util/hash.cc",
|
||||
@@ -649,6 +650,7 @@ cpp_library(
|
||||
"util/compression_context_cache.cc",
|
||||
"util/concurrent_task_limiter_impl.cc",
|
||||
"util/crc32c.cc",
|
||||
"util/crc32c_arm64.cc",
|
||||
"util/dynamic_bloom.cc",
|
||||
"util/file_checksum_helper.cc",
|
||||
"util/hash.cc",
|
||||
|
||||
@@ -306,7 +306,6 @@ Status BlobFileBuilder::CloseBlobFile() {
|
||||
" total blobs, %" PRIu64 " total bytes",
|
||||
column_family_name_.c_str(), job_id_, blob_file_number,
|
||||
blob_count_, blob_bytes_);
|
||||
|
||||
if (blob_callback_) {
|
||||
s = blob_callback_->OnBlobFileCompleted(blob_file_paths_->back());
|
||||
}
|
||||
|
||||
@@ -16,6 +16,14 @@ namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
class BlobFileCompletionCallback {
|
||||
public:
|
||||
#ifdef ROCKSDB_LITE
|
||||
BlobFileCompletionCallback(SstFileManager* /*sst_file_manager*/,
|
||||
InstrumentedMutex* /*mutex*/,
|
||||
ErrorHandler* /*error_handler*/) {}
|
||||
Status OnBlobFileCompleted(const std::string& /*file_name*/) {
|
||||
return Status::OK();
|
||||
}
|
||||
#else
|
||||
BlobFileCompletionCallback(SstFileManager* sst_file_manager,
|
||||
InstrumentedMutex* mutex,
|
||||
ErrorHandler* error_handler)
|
||||
@@ -25,8 +33,6 @@ class BlobFileCompletionCallback {
|
||||
|
||||
Status OnBlobFileCompleted(const std::string& file_name) {
|
||||
Status s;
|
||||
|
||||
#ifndef ROCKSDB_LITE
|
||||
auto sfm = static_cast<SstFileManagerImpl*>(sst_file_manager_);
|
||||
if (sfm) {
|
||||
// Report new blob files to SstFileManagerImpl
|
||||
@@ -39,9 +45,6 @@ class BlobFileCompletionCallback {
|
||||
error_handler_->SetBGError(s, BackgroundErrorReason::kFlush);
|
||||
}
|
||||
}
|
||||
#else
|
||||
(void)file_name;
|
||||
#endif // ROCKSDB_LITE
|
||||
return s;
|
||||
}
|
||||
|
||||
@@ -49,5 +52,6 @@ class BlobFileCompletionCallback {
|
||||
SstFileManager* sst_file_manager_;
|
||||
InstrumentedMutex* mutex_;
|
||||
ErrorHandler* error_handler_;
|
||||
#endif // ROCKSDB_LITE
|
||||
};
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -267,6 +267,49 @@ TEST_F(DBBlobBasicTest, GenerateIOTracing) {
|
||||
}
|
||||
#endif // !ROCKSDB_LITE
|
||||
|
||||
TEST_F(DBBlobBasicTest, BestEffortsRecovery_MissingNewestBlobFile) {
|
||||
Options options = GetDefaultOptions();
|
||||
options.enable_blob_files = true;
|
||||
options.min_blob_size = 0;
|
||||
options.create_if_missing = true;
|
||||
Reopen(options);
|
||||
|
||||
ASSERT_OK(dbfull()->DisableFileDeletions());
|
||||
constexpr int kNumTableFiles = 2;
|
||||
for (int i = 0; i < kNumTableFiles; ++i) {
|
||||
for (char ch = 'a'; ch != 'c'; ++ch) {
|
||||
std::string key(1, ch);
|
||||
ASSERT_OK(Put(key, "value" + std::to_string(i)));
|
||||
}
|
||||
ASSERT_OK(Flush());
|
||||
}
|
||||
|
||||
Close();
|
||||
|
||||
std::vector<std::string> files;
|
||||
ASSERT_OK(env_->GetChildren(dbname_, &files));
|
||||
std::string blob_file_path;
|
||||
uint64_t max_blob_file_num = kInvalidBlobFileNumber;
|
||||
for (const auto& fname : files) {
|
||||
uint64_t file_num = 0;
|
||||
FileType type;
|
||||
if (ParseFileName(fname, &file_num, /*info_log_name_prefix=*/"", &type) &&
|
||||
type == kBlobFile) {
|
||||
if (file_num > max_blob_file_num) {
|
||||
max_blob_file_num = file_num;
|
||||
blob_file_path = dbname_ + "/" + fname;
|
||||
}
|
||||
}
|
||||
}
|
||||
ASSERT_OK(env_->DeleteFile(blob_file_path));
|
||||
|
||||
options.best_efforts_recovery = true;
|
||||
Reopen(options);
|
||||
std::string value;
|
||||
ASSERT_OK(db_->Get(ReadOptions(), "a", &value));
|
||||
ASSERT_EQ("value" + std::to_string(kNumTableFiles - 2), value);
|
||||
}
|
||||
|
||||
class DBBlobBasicIOErrorTest : public DBBlobBasicTest,
|
||||
public testing::WithParamInterface<std::string> {
|
||||
protected:
|
||||
|
||||
@@ -2682,6 +2682,59 @@ unsigned char rocksdb_options_get_skip_checking_sst_file_sizes_on_db_open(
|
||||
return opt->rep.skip_checking_sst_file_sizes_on_db_open;
|
||||
}
|
||||
|
||||
/* Blob Options Settings */
|
||||
void rocksdb_options_set_enable_blob_files(rocksdb_options_t* opt,
|
||||
unsigned char val) {
|
||||
opt->rep.enable_blob_files = val;
|
||||
}
|
||||
extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_enable_blob_files(
|
||||
rocksdb_options_t* opt) {
|
||||
return opt->rep.enable_blob_files;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_min_blob_size(rocksdb_options_t* opt, uint64_t val) {
|
||||
opt->rep.min_blob_size = val;
|
||||
}
|
||||
|
||||
uint64_t rocksdb_options_get_min_blob_size(rocksdb_options_t* opt) {
|
||||
return opt->rep.min_blob_size;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_blob_file_size(rocksdb_options_t* opt, uint64_t val) {
|
||||
opt->rep.blob_file_size = val;
|
||||
}
|
||||
|
||||
uint64_t rocksdb_options_get_blob_file_size(rocksdb_options_t* opt) {
|
||||
return opt->rep.blob_file_size;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_blob_compression_type(rocksdb_options_t* opt,
|
||||
int val) {
|
||||
opt->rep.blob_compression_type = static_cast<CompressionType>(val);
|
||||
}
|
||||
|
||||
int rocksdb_options_get_blob_compression_type(rocksdb_options_t* opt) {
|
||||
return opt->rep.blob_compression_type;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_enable_blob_gc(rocksdb_options_t* opt,
|
||||
unsigned char val) {
|
||||
opt->rep.enable_blob_garbage_collection = val;
|
||||
}
|
||||
|
||||
unsigned char rocksdb_options_get_enable_blob_gc(rocksdb_options_t* opt) {
|
||||
return opt->rep.enable_blob_garbage_collection;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_blob_gc_age_cutoff(rocksdb_options_t* opt,
|
||||
double val) {
|
||||
opt->rep.blob_garbage_collection_age_cutoff = val;
|
||||
}
|
||||
|
||||
double rocksdb_options_get_blob_gc_age_cutoff(rocksdb_options_t* opt) {
|
||||
return opt->rep.blob_garbage_collection_age_cutoff;
|
||||
}
|
||||
|
||||
void rocksdb_options_set_num_levels(rocksdb_options_t* opt, int n) {
|
||||
opt->rep.num_levels = n;
|
||||
}
|
||||
@@ -4112,6 +4165,10 @@ void rocksdb_cache_destroy(rocksdb_cache_t* cache) {
|
||||
delete cache;
|
||||
}
|
||||
|
||||
void rocksdb_cache_disown_data(rocksdb_cache_t* cache) {
|
||||
cache->rep->DisownData();
|
||||
}
|
||||
|
||||
void rocksdb_cache_set_capacity(rocksdb_cache_t* cache, size_t capacity) {
|
||||
cache->rep->SetCapacity(capacity);
|
||||
}
|
||||
|
||||
+20
@@ -1759,6 +1759,25 @@ int main(int argc, char** argv) {
|
||||
rocksdb_options_set_atomic_flush(o, 1);
|
||||
CheckCondition(1 == rocksdb_options_get_atomic_flush(o));
|
||||
|
||||
/* Blob Options */
|
||||
rocksdb_options_set_enable_blob_files(o, 1);
|
||||
CheckCondition(1 == rocksdb_options_get_enable_blob_files(o));
|
||||
|
||||
rocksdb_options_set_min_blob_size(o, 29);
|
||||
CheckCondition(29 == rocksdb_options_get_min_blob_size(o));
|
||||
|
||||
rocksdb_options_set_blob_file_size(o, 30);
|
||||
CheckCondition(30 == rocksdb_options_get_blob_file_size(o));
|
||||
|
||||
rocksdb_options_set_blob_compression_type(o, 4);
|
||||
CheckCondition(4 == rocksdb_options_get_blob_compression_type(o));
|
||||
|
||||
rocksdb_options_set_enable_blob_gc(o, 1);
|
||||
CheckCondition(1 == rocksdb_options_get_enable_blob_gc(o));
|
||||
|
||||
rocksdb_options_set_blob_gc_age_cutoff(o, 0.75);
|
||||
CheckCondition(0.75 == rocksdb_options_get_blob_gc_age_cutoff(o));
|
||||
|
||||
// Create a copy that should be equal to the original.
|
||||
rocksdb_options_t* copy;
|
||||
copy = rocksdb_options_create_copy(o);
|
||||
@@ -2869,6 +2888,7 @@ int main(int argc, char** argv) {
|
||||
rocksdb_readoptions_destroy(roptions);
|
||||
rocksdb_writeoptions_destroy(woptions);
|
||||
rocksdb_compactoptions_destroy(coptions);
|
||||
rocksdb_cache_disown_data(cache);
|
||||
rocksdb_cache_destroy(cache);
|
||||
rocksdb_comparator_destroy(cmp);
|
||||
rocksdb_dbpath_destroy(dbpath);
|
||||
|
||||
@@ -1355,6 +1355,12 @@ Status ColumnFamilyData::ValidateOptions(
|
||||
"[0.0, 1.0].");
|
||||
}
|
||||
|
||||
if (cf_options.compaction_style == kCompactionStyleFIFO &&
|
||||
db_options.max_open_files != -1 && cf_options.ttl > 0) {
|
||||
return Status::NotSupported(
|
||||
"FIFO compaction only supported with max_open_files = -1.");
|
||||
}
|
||||
|
||||
return s;
|
||||
}
|
||||
|
||||
|
||||
@@ -810,6 +810,7 @@ class TestingContextCustomFilterPolicy
|
||||
TEST_F(DBBloomFilterTest, ContextCustomFilterPolicy) {
|
||||
for (bool fifo : {true, false}) {
|
||||
Options options = CurrentOptions();
|
||||
options.max_open_files = fifo ? -1 : options.max_open_files;
|
||||
options.statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();
|
||||
options.compaction_style =
|
||||
fifo ? kCompactionStyleFIFO : kCompactionStyleLevel;
|
||||
@@ -820,6 +821,7 @@ TEST_F(DBBloomFilterTest, ContextCustomFilterPolicy) {
|
||||
table_options.format_version = 5;
|
||||
options.table_factory.reset(NewBlockBasedTableFactory(table_options));
|
||||
|
||||
TryReopen(options);
|
||||
CreateAndReopenWithCF({fifo ? "abe" : "bob"}, options);
|
||||
|
||||
const int maxKey = 10000;
|
||||
|
||||
@@ -1605,7 +1605,7 @@ class DBImpl : public DB {
|
||||
Status SwitchWAL(WriteContext* write_context);
|
||||
|
||||
// REQUIRES: mutex locked and in write thread.
|
||||
Status HandleWriteBufferFull(WriteContext* write_context);
|
||||
Status HandleWriteBufferManagerFlush(WriteContext* write_context);
|
||||
|
||||
// REQUIRES: mutex locked
|
||||
Status PreprocessWrite(const WriteOptions& write_options, bool* need_log_sync,
|
||||
|
||||
@@ -1383,7 +1383,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
|
||||
mutable_cf_options.compression_opts, paranoid_file_checks,
|
||||
cfd->internal_stats(), TableFileCreationReason::kRecovery, &io_s,
|
||||
io_tracer_, &event_logger_, job_id, Env::IO_HIGH,
|
||||
nullptr /* table_properties */, -1 /* level */, current_time,
|
||||
nullptr /* table_properties */, 0 /* level */, current_time,
|
||||
0 /* oldest_key_time */, write_hint, 0 /* file_creation_time */,
|
||||
db_id_, db_session_id_, nullptr /*full_history_ts_low*/,
|
||||
&blob_callback_);
|
||||
|
||||
@@ -937,7 +937,7 @@ Status DBImpl::PreprocessWrite(const WriteOptions& write_options,
|
||||
// be flushed. We may end up with flushing much more DBs than needed. It's
|
||||
// suboptimal but still correct.
|
||||
WaitForPendingWrites();
|
||||
status = HandleWriteBufferFull(write_context);
|
||||
status = HandleWriteBufferManagerFlush(write_context);
|
||||
}
|
||||
|
||||
if (UNLIKELY(status.ok() && !trim_history_scheduler_.Empty())) {
|
||||
@@ -1348,20 +1348,20 @@ Status DBImpl::SwitchWAL(WriteContext* write_context) {
|
||||
if (!immutable_db_options_.atomic_flush) {
|
||||
FlushRequest flush_req;
|
||||
GenerateFlushRequest({cfd}, &flush_req);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferManager);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWalFull);
|
||||
}
|
||||
}
|
||||
if (immutable_db_options_.atomic_flush) {
|
||||
FlushRequest flush_req;
|
||||
GenerateFlushRequest(cfds, &flush_req);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferManager);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWalFull);
|
||||
}
|
||||
MaybeScheduleFlushOrCompaction();
|
||||
}
|
||||
return status;
|
||||
}
|
||||
|
||||
Status DBImpl::HandleWriteBufferFull(WriteContext* write_context) {
|
||||
Status DBImpl::HandleWriteBufferManagerFlush(WriteContext* write_context) {
|
||||
mutex_.AssertHeld();
|
||||
assert(write_context != nullptr);
|
||||
Status status;
|
||||
@@ -1373,7 +1373,7 @@ Status DBImpl::HandleWriteBufferFull(WriteContext* write_context) {
|
||||
// suboptimal but still correct.
|
||||
ROCKS_LOG_INFO(
|
||||
immutable_db_options_.info_log,
|
||||
"Flushing column family with oldest memtable entry. Write buffer is "
|
||||
"Flushing column family with oldest memtable entry. Write buffers are "
|
||||
"using %" ROCKSDB_PRIszt " bytes out of a total of %" ROCKSDB_PRIszt ".",
|
||||
write_buffer_manager_->memory_usage(),
|
||||
write_buffer_manager_->buffer_size());
|
||||
@@ -1434,13 +1434,13 @@ Status DBImpl::HandleWriteBufferFull(WriteContext* write_context) {
|
||||
if (!immutable_db_options_.atomic_flush) {
|
||||
FlushRequest flush_req;
|
||||
GenerateFlushRequest({cfd}, &flush_req);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferFull);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferManager);
|
||||
}
|
||||
}
|
||||
if (immutable_db_options_.atomic_flush) {
|
||||
FlushRequest flush_req;
|
||||
GenerateFlushRequest(cfds, &flush_req);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferFull);
|
||||
SchedulePendingFlush(flush_req, FlushReason::kWriteBufferManager);
|
||||
}
|
||||
MaybeScheduleFlushOrCompaction();
|
||||
}
|
||||
|
||||
@@ -1626,6 +1626,7 @@ TEST_F(DBPropertiesTest, EstimateOldestKeyTime) {
|
||||
|
||||
options.compaction_style = kCompactionStyleFIFO;
|
||||
options.ttl = 300;
|
||||
options.max_open_files = -1;
|
||||
options.compaction_options_fifo.allow_compaction = false;
|
||||
DestroyAndReopen(options);
|
||||
|
||||
|
||||
@@ -561,6 +561,7 @@ TEST_F(DBSSTTest, DBWithSstFileManagerForBlobFilesWithGC) {
|
||||
constexpr Slice* end = nullptr;
|
||||
|
||||
ASSERT_OK(db_->CompactRange(CompactRangeOptions(), begin, end));
|
||||
sfm->WaitForEmptyTrash();
|
||||
|
||||
ASSERT_EQ(Get(first_key), first_value);
|
||||
ASSERT_EQ(Get(second_key), second_value);
|
||||
@@ -593,6 +594,7 @@ TEST_F(DBSSTTest, DBWithSstFileManagerForBlobFilesWithGC) {
|
||||
|
||||
Close();
|
||||
ASSERT_OK(DestroyDB(dbname_, options));
|
||||
sfm->WaitForEmptyTrash();
|
||||
ASSERT_EQ(files_deleted, 5);
|
||||
ASSERT_EQ(files_scheduled_to_delete, 5);
|
||||
|
||||
|
||||
+12
-5
@@ -908,6 +908,9 @@ TEST_F(DBTest, FlushSchedule) {
|
||||
static_cast<int64_t>(options.write_buffer_size);
|
||||
options.max_write_buffer_number = 2;
|
||||
options.write_buffer_size = 120 * 1024;
|
||||
auto flush_listener = std::make_shared<FlushCounterListener>();
|
||||
flush_listener->expected_flush_reason = FlushReason::kWriteBufferFull;
|
||||
options.listeners.push_back(flush_listener);
|
||||
CreateAndReopenWithCF({"pikachu"}, options);
|
||||
std::vector<port::Thread> threads;
|
||||
|
||||
@@ -3504,17 +3507,21 @@ TEST_F(DBTest, FIFOCompactionStyleWithCompactionAndDelete) {
|
||||
}
|
||||
|
||||
// Check that FIFO-with-TTL is not supported with max_open_files != -1.
|
||||
// Github issue #8014
|
||||
TEST_F(DBTest, FIFOCompactionWithTTLAndMaxOpenFilesTest) {
|
||||
Options options;
|
||||
Options options = CurrentOptions();
|
||||
options.compaction_style = kCompactionStyleFIFO;
|
||||
options.create_if_missing = true;
|
||||
options.ttl = 600; // seconds
|
||||
|
||||
// TTL is now supported with max_open_files != -1.
|
||||
options.max_open_files = 100;
|
||||
options = CurrentOptions(options);
|
||||
ASSERT_OK(TryReopen(options));
|
||||
// TTL is not supported with max_open_files != -1.
|
||||
options.max_open_files = 0;
|
||||
ASSERT_TRUE(TryReopen(options).IsNotSupported());
|
||||
|
||||
options.max_open_files = 100;
|
||||
ASSERT_TRUE(TryReopen(options).IsNotSupported());
|
||||
|
||||
// TTL is supported with unlimited max_open_files
|
||||
options.max_open_files = -1;
|
||||
ASSERT_OK(TryReopen(options));
|
||||
}
|
||||
|
||||
@@ -344,6 +344,10 @@ class DBTestSharedWriteBufferAcrossCFs
|
||||
TEST_P(DBTestSharedWriteBufferAcrossCFs, SharedWriteBufferAcrossCFs) {
|
||||
Options options = CurrentOptions();
|
||||
options.arena_block_size = 4096;
|
||||
auto flush_listener = std::make_shared<FlushCounterListener>();
|
||||
options.listeners.push_back(flush_listener);
|
||||
// Don't trip the listener at shutdown.
|
||||
options.avoid_flush_during_shutdown = true;
|
||||
|
||||
// Avoid undeterministic value by malloc_usable_size();
|
||||
// Force arena block size to 1
|
||||
@@ -387,6 +391,7 @@ TEST_P(DBTestSharedWriteBufferAcrossCFs, SharedWriteBufferAcrossCFs) {
|
||||
|
||||
// Create some data and flush "default" and "nikitich" so that they
|
||||
// are newer CFs created.
|
||||
flush_listener->expected_flush_reason = FlushReason::kManualFlush;
|
||||
ASSERT_OK(Put(3, Key(1), DummyString(1), wo));
|
||||
Flush(3);
|
||||
ASSERT_OK(Put(3, Key(1), DummyString(1), wo));
|
||||
@@ -397,6 +402,7 @@ TEST_P(DBTestSharedWriteBufferAcrossCFs, SharedWriteBufferAcrossCFs) {
|
||||
ASSERT_EQ(GetNumberOfSstFilesForColumnFamily(db_, "nikitich"),
|
||||
static_cast<uint64_t>(1));
|
||||
|
||||
flush_listener->expected_flush_reason = FlushReason::kWriteBufferManager;
|
||||
ASSERT_OK(Put(3, Key(1), DummyString(30000), wo));
|
||||
if (cost_cache_) {
|
||||
ASSERT_GE(cache->GetUsage(), 256 * 1024);
|
||||
@@ -521,6 +527,10 @@ TEST_F(DBTest2, SharedWriteBufferLimitAcrossDB) {
|
||||
std::string dbname2 = test::PerThreadDBPath("db_shared_wb_db2");
|
||||
Options options = CurrentOptions();
|
||||
options.arena_block_size = 4096;
|
||||
auto flush_listener = std::make_shared<FlushCounterListener>();
|
||||
options.listeners.push_back(flush_listener);
|
||||
// Don't trip the listener at shutdown.
|
||||
options.avoid_flush_during_shutdown = true;
|
||||
// Avoid undeterministic value by malloc_usable_size();
|
||||
// Force arena block size to 1
|
||||
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
|
||||
@@ -558,6 +568,7 @@ TEST_F(DBTest2, SharedWriteBufferLimitAcrossDB) {
|
||||
};
|
||||
|
||||
// Trigger a flush on cf2
|
||||
flush_listener->expected_flush_reason = FlushReason::kWriteBufferManager;
|
||||
ASSERT_OK(Put(2, Key(1), DummyString(70000), wo));
|
||||
wait_flush();
|
||||
ASSERT_OK(Put(0, Key(1), DummyString(20000), wo));
|
||||
|
||||
@@ -487,6 +487,7 @@ Options DBTestBase::GetOptions(
|
||||
}
|
||||
case kFIFOCompaction: {
|
||||
options.compaction_style = kCompactionStyleFIFO;
|
||||
options.max_open_files = -1;
|
||||
break;
|
||||
}
|
||||
case kBlockBasedTableWithPrefixHashIndex: {
|
||||
|
||||
@@ -761,6 +761,17 @@ class OnFileDeletionListener : public EventListener {
|
||||
size_t matched_count_;
|
||||
std::string expected_file_name_;
|
||||
};
|
||||
|
||||
class FlushCounterListener : public EventListener {
|
||||
public:
|
||||
std::atomic<int> count{0};
|
||||
std::atomic<FlushReason> expected_flush_reason{FlushReason::kOthers};
|
||||
|
||||
void OnFlushBegin(DB* /*db*/, const FlushJobInfo& flush_job_info) override {
|
||||
count++;
|
||||
ASSERT_EQ(expected_flush_reason.load(), flush_job_info.flush_reason);
|
||||
}
|
||||
};
|
||||
#endif
|
||||
|
||||
// A test merge operator mimics put but also fails if one of merge operands is
|
||||
|
||||
+2
-13
@@ -1801,19 +1801,8 @@ TEST_P(DBWALTestWithParamsVaryingRecoveryMode,
|
||||
// avoid_flush_during_recovery=true.
|
||||
// Flush should trigger if max_total_wal_size is reached.
|
||||
TEST_F(DBWALTest, RestoreTotalLogSizeAfterRecoverWithoutFlush) {
|
||||
class TestFlushListener : public EventListener {
|
||||
public:
|
||||
std::atomic<int> count{0};
|
||||
|
||||
TestFlushListener() = default;
|
||||
|
||||
void OnFlushBegin(DB* /*db*/, const FlushJobInfo& flush_job_info) override {
|
||||
count++;
|
||||
ASSERT_EQ(FlushReason::kWriteBufferManager, flush_job_info.flush_reason);
|
||||
}
|
||||
};
|
||||
std::shared_ptr<TestFlushListener> test_listener =
|
||||
std::make_shared<TestFlushListener>();
|
||||
auto test_listener = std::make_shared<FlushCounterListener>();
|
||||
test_listener->expected_flush_reason = FlushReason::kWalFull;
|
||||
|
||||
constexpr size_t kKB = 1024;
|
||||
constexpr size_t kMB = 1024 * 1024;
|
||||
|
||||
@@ -75,6 +75,8 @@ const char* GetFlushReasonString (FlushReason flush_reason) {
|
||||
return "Manual Flush";
|
||||
case FlushReason::kErrorRecovery:
|
||||
return "Error Recovery";
|
||||
case FlushReason::kWalFull:
|
||||
return "WAL Full";
|
||||
default:
|
||||
return "Invalid";
|
||||
}
|
||||
|
||||
+27
-1
@@ -517,6 +517,28 @@ class VersionBuilder::Rep {
|
||||
return meta->oldest_blob_file_number;
|
||||
}
|
||||
|
||||
uint64_t GetMinOldestBlobFileNumber() const {
|
||||
uint64_t min_oldest_blob_file_num = std::numeric_limits<uint64_t>::max();
|
||||
for (int level = 0; level < num_levels_; ++level) {
|
||||
const auto& base_files = base_vstorage_->LevelFiles(level);
|
||||
for (const auto* fmeta : base_files) {
|
||||
assert(fmeta);
|
||||
min_oldest_blob_file_num =
|
||||
std::min(min_oldest_blob_file_num, fmeta->oldest_blob_file_number);
|
||||
}
|
||||
const auto& added_files = levels_[level].added_files;
|
||||
for (const auto& elem : added_files) {
|
||||
assert(elem.second);
|
||||
min_oldest_blob_file_num = std::min(
|
||||
min_oldest_blob_file_num, elem.second->oldest_blob_file_number);
|
||||
}
|
||||
}
|
||||
if (min_oldest_blob_file_num == std::numeric_limits<uint64_t>::max()) {
|
||||
min_oldest_blob_file_num = kInvalidBlobFileNumber;
|
||||
}
|
||||
return min_oldest_blob_file_num;
|
||||
}
|
||||
|
||||
Status ApplyFileDeletion(int level, uint64_t file_number) {
|
||||
assert(level != VersionStorageInfo::FileLocation::Invalid().GetLevel());
|
||||
|
||||
@@ -834,7 +856,7 @@ class VersionBuilder::Rep {
|
||||
}
|
||||
}
|
||||
|
||||
// Save the current state in *v.
|
||||
// Save the current state in *vstorage.
|
||||
Status SaveTo(VersionStorageInfo* vstorage) {
|
||||
Status s = CheckConsistency(base_vstorage_);
|
||||
if (!s.ok()) {
|
||||
@@ -1052,6 +1074,10 @@ Status VersionBuilder::LoadTableHandlers(
|
||||
is_initial_load, prefix_extractor, max_file_size_for_l0_meta_pin);
|
||||
}
|
||||
|
||||
uint64_t VersionBuilder::GetMinOldestBlobFileNumber() const {
|
||||
return rep_->GetMinOldestBlobFileNumber();
|
||||
}
|
||||
|
||||
BaseReferencedVersionBuilder::BaseReferencedVersionBuilder(
|
||||
ColumnFamilyData* cfd)
|
||||
: version_builder_(new VersionBuilder(
|
||||
|
||||
@@ -44,6 +44,7 @@ class VersionBuilder {
|
||||
bool is_initial_load,
|
||||
const SliceTransform* prefix_extractor,
|
||||
size_t max_file_size_for_l0_meta_pin);
|
||||
uint64_t GetMinOldestBlobFileNumber() const;
|
||||
|
||||
private:
|
||||
class Rep;
|
||||
|
||||
+113
-13
@@ -11,6 +11,8 @@
|
||||
|
||||
#include <cinttypes>
|
||||
|
||||
#include "db/blob/blob_file_cache.h"
|
||||
#include "db/blob/blob_file_reader.h"
|
||||
#include "monitoring/persistent_stats_history.h"
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
@@ -129,14 +131,14 @@ Status FileChecksumRetriever::ApplyVersionEdit(VersionEdit& edit,
|
||||
VersionEditHandler::VersionEditHandler(
|
||||
bool read_only, std::vector<ColumnFamilyDescriptor> column_families,
|
||||
VersionSet* version_set, bool track_missing_files,
|
||||
bool no_error_if_table_files_missing,
|
||||
const std::shared_ptr<IOTracer>& io_tracer, bool skip_load_table_files)
|
||||
bool no_error_if_files_missing, const std::shared_ptr<IOTracer>& io_tracer,
|
||||
bool skip_load_table_files)
|
||||
: VersionEditHandlerBase(),
|
||||
read_only_(read_only),
|
||||
column_families_(std::move(column_families)),
|
||||
version_set_(version_set),
|
||||
track_missing_files_(track_missing_files),
|
||||
no_error_if_table_files_missing_(no_error_if_table_files_missing),
|
||||
no_error_if_files_missing_(no_error_if_files_missing),
|
||||
io_tracer_(io_tracer),
|
||||
skip_load_table_files_(skip_load_table_files),
|
||||
initialized_(false) {
|
||||
@@ -301,6 +303,14 @@ bool VersionEditHandler::HasMissingFiles() const {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!ret) {
|
||||
for (const auto& elem : cf_to_missing_blob_files_high_) {
|
||||
if (elem.second != kInvalidBlobFileNumber) {
|
||||
ret = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
@@ -437,6 +447,8 @@ ColumnFamilyData* VersionEditHandler::CreateCfAndInit(
|
||||
if (track_missing_files_) {
|
||||
cf_to_missing_files_.emplace(edit.column_family_,
|
||||
std::unordered_set<uint64_t>());
|
||||
cf_to_missing_blob_files_high_.emplace(edit.column_family_,
|
||||
kInvalidBlobFileNumber);
|
||||
}
|
||||
return cfd;
|
||||
}
|
||||
@@ -450,6 +462,12 @@ ColumnFamilyData* VersionEditHandler::DestroyCfAndCleanup(
|
||||
auto missing_files_iter = cf_to_missing_files_.find(edit.column_family_);
|
||||
assert(missing_files_iter != cf_to_missing_files_.end());
|
||||
cf_to_missing_files_.erase(missing_files_iter);
|
||||
|
||||
auto missing_blob_files_high_iter =
|
||||
cf_to_missing_blob_files_high_.find(edit.column_family_);
|
||||
assert(missing_blob_files_high_iter !=
|
||||
cf_to_missing_blob_files_high_.end());
|
||||
cf_to_missing_blob_files_high_.erase(missing_blob_files_high_iter);
|
||||
}
|
||||
ColumnFamilyData* ret =
|
||||
version_set_->GetColumnFamilySet()->GetColumnFamily(edit.column_family_);
|
||||
@@ -505,8 +523,7 @@ Status VersionEditHandler::LoadTables(ColumnFamilyData* cfd,
|
||||
prefetch_index_and_filter_in_cache, is_initial_load,
|
||||
cfd->GetLatestMutableCFOptions()->prefix_extractor.get(),
|
||||
MaxFileSizeForL0MetaPin(*cfd->GetLatestMutableCFOptions()));
|
||||
if ((s.IsPathNotFound() || s.IsCorruption()) &&
|
||||
no_error_if_table_files_missing_) {
|
||||
if ((s.IsPathNotFound() || s.IsCorruption()) && no_error_if_files_missing_) {
|
||||
s = Status::OK();
|
||||
}
|
||||
if (!s.ok() && !version_set_->db_options_->paranoid_checks) {
|
||||
@@ -576,7 +593,7 @@ VersionEditHandlerPointInTime::VersionEditHandlerPointInTime(
|
||||
VersionSet* version_set, const std::shared_ptr<IOTracer>& io_tracer)
|
||||
: VersionEditHandler(read_only, column_families, version_set,
|
||||
/*track_missing_files=*/true,
|
||||
/*no_error_if_table_files_missing=*/true, io_tracer) {}
|
||||
/*no_error_if_files_missing=*/true, io_tracer) {}
|
||||
|
||||
VersionEditHandlerPointInTime::~VersionEditHandlerPointInTime() {
|
||||
for (const auto& elem : versions_) {
|
||||
@@ -626,7 +643,29 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
|
||||
auto missing_files_iter = cf_to_missing_files_.find(cfd->GetID());
|
||||
assert(missing_files_iter != cf_to_missing_files_.end());
|
||||
std::unordered_set<uint64_t>& missing_files = missing_files_iter->second;
|
||||
const bool prev_has_missing_files = !missing_files.empty();
|
||||
|
||||
auto missing_blob_files_high_iter =
|
||||
cf_to_missing_blob_files_high_.find(cfd->GetID());
|
||||
assert(missing_blob_files_high_iter != cf_to_missing_blob_files_high_.end());
|
||||
const uint64_t prev_missing_blob_file_high =
|
||||
missing_blob_files_high_iter->second;
|
||||
|
||||
VersionBuilder* builder = nullptr;
|
||||
|
||||
if (prev_missing_blob_file_high != kInvalidBlobFileNumber) {
|
||||
auto builder_iter = builders_.find(cfd->GetID());
|
||||
assert(builder_iter != builders_.end());
|
||||
builder = builder_iter->second->version_builder();
|
||||
assert(builder != nullptr);
|
||||
}
|
||||
|
||||
// At this point, we have not yet applied the new version edits read from the
|
||||
// MANIFEST. We check whether we have any missing table and blob files.
|
||||
const bool prev_has_missing_files =
|
||||
!missing_files.empty() ||
|
||||
(prev_missing_blob_file_high != kInvalidBlobFileNumber &&
|
||||
prev_missing_blob_file_high >= builder->GetMinOldestBlobFileNumber());
|
||||
|
||||
for (const auto& file : edit.GetDeletedFiles()) {
|
||||
uint64_t file_num = file.second;
|
||||
auto fiter = missing_files.find(file_num);
|
||||
@@ -634,6 +673,8 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
|
||||
missing_files.erase(fiter);
|
||||
}
|
||||
}
|
||||
|
||||
assert(!cfd->ioptions()->cf_paths.empty());
|
||||
Status s;
|
||||
for (const auto& elem : edit.GetNewFiles()) {
|
||||
const FileMetaData& meta = elem.second;
|
||||
@@ -649,17 +690,60 @@ Status VersionEditHandlerPointInTime::MaybeCreateVersion(
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
uint64_t missing_blob_file_num = prev_missing_blob_file_high;
|
||||
for (const auto& elem : edit.GetBlobFileAdditions()) {
|
||||
uint64_t file_num = elem.GetBlobFileNumber();
|
||||
s = VerifyBlobFile(cfd, file_num, elem);
|
||||
if (s.IsPathNotFound() || s.IsNotFound() || s.IsCorruption()) {
|
||||
missing_blob_file_num = std::max(missing_blob_file_num, file_num);
|
||||
s = Status::OK();
|
||||
} else if (!s.ok()) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
bool has_missing_blob_files = false;
|
||||
if (missing_blob_file_num != kInvalidBlobFileNumber &&
|
||||
missing_blob_file_num >= prev_missing_blob_file_high) {
|
||||
missing_blob_files_high_iter->second = missing_blob_file_num;
|
||||
has_missing_blob_files = true;
|
||||
} else if (missing_blob_file_num < prev_missing_blob_file_high) {
|
||||
assert(false);
|
||||
}
|
||||
|
||||
// We still have not applied the new version edit, but have tried to add new
|
||||
// table and blob files after verifying their presence and consistency.
|
||||
// Therefore, we know whether we will see new missing table and blob files
|
||||
// later after actually applying the version edit. We perform the check here
|
||||
// and record the result.
|
||||
const bool has_missing_files =
|
||||
!missing_files.empty() || has_missing_blob_files;
|
||||
|
||||
bool missing_info = !version_edit_params_.has_log_number_ ||
|
||||
!version_edit_params_.has_next_file_number_ ||
|
||||
!version_edit_params_.has_last_sequence_;
|
||||
|
||||
// Create version before apply edit
|
||||
// Create version before apply edit. The version will represent the state
|
||||
// before applying the version edit.
|
||||
// A new version will created if:
|
||||
// 1) no error has occurred so far, and
|
||||
// 2) log_number_, next_file_number_ and last_sequence_ are known, and
|
||||
// 3) any of the following:
|
||||
// a) no missing file before, but will have missing file(s) after applying
|
||||
// this version edit.
|
||||
// b) no missing file after applying the version edit, and the caller
|
||||
// explicitly request that a new version be created.
|
||||
if (s.ok() && !missing_info &&
|
||||
((!missing_files.empty() && !prev_has_missing_files) ||
|
||||
(missing_files.empty() && force_create_version))) {
|
||||
auto builder_iter = builders_.find(cfd->GetID());
|
||||
assert(builder_iter != builders_.end());
|
||||
auto* builder = builder_iter->second->version_builder();
|
||||
((has_missing_files && !prev_has_missing_files) ||
|
||||
(!has_missing_files && force_create_version))) {
|
||||
if (!builder) {
|
||||
auto builder_iter = builders_.find(cfd->GetID());
|
||||
assert(builder_iter != builders_.end());
|
||||
builder = builder_iter->second->version_builder();
|
||||
assert(builder);
|
||||
}
|
||||
|
||||
auto* version = new Version(cfd, version_set_, version_set_->file_options_,
|
||||
*cfd->GetLatestMutableCFOptions(), io_tracer_,
|
||||
version_set_->current_version_number_++);
|
||||
@@ -687,6 +771,22 @@ Status VersionEditHandlerPointInTime::VerifyFile(const std::string& fpath,
|
||||
return version_set_->VerifyFileMetadata(fpath, fmeta);
|
||||
}
|
||||
|
||||
Status VersionEditHandlerPointInTime::VerifyBlobFile(
|
||||
ColumnFamilyData* cfd, uint64_t blob_file_num,
|
||||
const BlobFileAddition& blob_addition) {
|
||||
BlobFileCache* blob_file_cache = cfd->blob_file_cache();
|
||||
assert(blob_file_cache);
|
||||
CacheHandleGuard<BlobFileReader> blob_file_reader;
|
||||
Status s =
|
||||
blob_file_cache->GetBlobFileReader(blob_file_num, &blob_file_reader);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
}
|
||||
// TODO: verify checksum
|
||||
(void)blob_addition;
|
||||
return s;
|
||||
}
|
||||
|
||||
Status ManifestTailer::Initialize() {
|
||||
if (Mode::kRecovery == mode_) {
|
||||
return VersionEditHandler::Initialize();
|
||||
|
||||
@@ -97,7 +97,7 @@ using VersionBuilderUPtr = std::unique_ptr<BaseReferencedVersionBuilder>;
|
||||
// 1. Create an object of VersionEditHandler or its subclasses.
|
||||
// VersionEditHandler handler(read_only, column_families, version_set,
|
||||
// track_missing_files,
|
||||
// no_error_if_table_files_missing);
|
||||
// no_error_if_files_missing);
|
||||
// 2. Status s = handler.Iterate(reader, &db_id);
|
||||
// 3. Check s and handle possible errors.
|
||||
//
|
||||
@@ -109,10 +109,10 @@ class VersionEditHandler : public VersionEditHandlerBase {
|
||||
bool read_only,
|
||||
const std::vector<ColumnFamilyDescriptor>& column_families,
|
||||
VersionSet* version_set, bool track_missing_files,
|
||||
bool no_error_if_table_files_missing,
|
||||
bool no_error_if_files_missing,
|
||||
const std::shared_ptr<IOTracer>& io_tracer)
|
||||
: VersionEditHandler(read_only, column_families, version_set,
|
||||
track_missing_files, no_error_if_table_files_missing,
|
||||
track_missing_files, no_error_if_files_missing,
|
||||
io_tracer, /*skip_load_table_files=*/false) {}
|
||||
|
||||
~VersionEditHandler() override {}
|
||||
@@ -133,7 +133,7 @@ class VersionEditHandler : public VersionEditHandlerBase {
|
||||
explicit VersionEditHandler(
|
||||
bool read_only, std::vector<ColumnFamilyDescriptor> column_families,
|
||||
VersionSet* version_set, bool track_missing_files,
|
||||
bool no_error_if_table_files_missing,
|
||||
bool no_error_if_files_missing,
|
||||
const std::shared_ptr<IOTracer>& io_tracer, bool skip_load_table_files);
|
||||
|
||||
Status ApplyVersionEdit(VersionEdit& edit, ColumnFamilyData** cfd) override;
|
||||
@@ -183,7 +183,8 @@ class VersionEditHandler : public VersionEditHandlerBase {
|
||||
const bool track_missing_files_;
|
||||
std::unordered_map<uint32_t, std::unordered_set<uint64_t>>
|
||||
cf_to_missing_files_;
|
||||
bool no_error_if_table_files_missing_;
|
||||
std::unordered_map<uint32_t, uint64_t> cf_to_missing_blob_files_high_;
|
||||
bool no_error_if_files_missing_;
|
||||
std::shared_ptr<IOTracer> io_tracer_;
|
||||
bool skip_load_table_files_;
|
||||
bool initialized_;
|
||||
@@ -213,6 +214,8 @@ class VersionEditHandlerPointInTime : public VersionEditHandler {
|
||||
bool force_create_version) override;
|
||||
virtual Status VerifyFile(const std::string& fpath,
|
||||
const FileMetaData& fmeta);
|
||||
virtual Status VerifyBlobFile(ColumnFamilyData* cfd, uint64_t blob_file_num,
|
||||
const BlobFileAddition& blob_addition);
|
||||
|
||||
std::unordered_map<uint32_t, Version*> versions_;
|
||||
};
|
||||
@@ -267,7 +270,7 @@ class DumpManifestHandler : public VersionEditHandler {
|
||||
: VersionEditHandler(
|
||||
/*read_only=*/true, column_families, version_set,
|
||||
/*track_missing_files=*/false,
|
||||
/*no_error_if_table_files_missing=*/false, io_tracer,
|
||||
/*no_error_if_files_missing=*/false, io_tracer,
|
||||
/*skip_load_table_files=*/true),
|
||||
verbose_(verbose),
|
||||
hex_(hex),
|
||||
|
||||
+30
-64
@@ -4490,60 +4490,6 @@ Status VersionSet::LogAndApplyHelper(ColumnFamilyData* cfd,
|
||||
return builder ? builder->Apply(edit) : Status::OK();
|
||||
}
|
||||
|
||||
Status VersionSet::ExtractInfoFromVersionEdit(
|
||||
ColumnFamilyData* cfd, const VersionEdit& from_edit,
|
||||
VersionEditParams* version_edit_params) {
|
||||
if (cfd != nullptr) {
|
||||
if (from_edit.has_db_id_) {
|
||||
version_edit_params->SetDBId(from_edit.db_id_);
|
||||
}
|
||||
if (from_edit.has_log_number_) {
|
||||
if (cfd->GetLogNumber() > from_edit.log_number_) {
|
||||
ROCKS_LOG_WARN(
|
||||
db_options_->info_log,
|
||||
"MANIFEST corruption detected, but ignored - Log numbers in "
|
||||
"records NOT monotonically increasing");
|
||||
} else {
|
||||
cfd->SetLogNumber(from_edit.log_number_);
|
||||
version_edit_params->SetLogNumber(from_edit.log_number_);
|
||||
}
|
||||
}
|
||||
if (from_edit.has_comparator_ &&
|
||||
from_edit.comparator_ != cfd->user_comparator()->Name()) {
|
||||
return Status::InvalidArgument(
|
||||
cfd->user_comparator()->Name(),
|
||||
"does not match existing comparator " + from_edit.comparator_);
|
||||
}
|
||||
if (from_edit.HasFullHistoryTsLow()) {
|
||||
const std::string& new_ts = from_edit.GetFullHistoryTsLow();
|
||||
cfd->SetFullHistoryTsLow(new_ts);
|
||||
}
|
||||
}
|
||||
|
||||
if (from_edit.has_prev_log_number_) {
|
||||
version_edit_params->SetPrevLogNumber(from_edit.prev_log_number_);
|
||||
}
|
||||
|
||||
if (from_edit.has_next_file_number_) {
|
||||
version_edit_params->SetNextFile(from_edit.next_file_number_);
|
||||
}
|
||||
|
||||
if (from_edit.has_max_column_family_) {
|
||||
version_edit_params->SetMaxColumnFamily(from_edit.max_column_family_);
|
||||
}
|
||||
|
||||
if (from_edit.has_min_log_number_to_keep_) {
|
||||
version_edit_params->min_log_number_to_keep_ =
|
||||
std::max(version_edit_params->min_log_number_to_keep_,
|
||||
from_edit.min_log_number_to_keep_);
|
||||
}
|
||||
|
||||
if (from_edit.has_last_sequence_) {
|
||||
version_edit_params->SetLastSequence(from_edit.last_sequence_);
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status VersionSet::GetCurrentManifestPath(const std::string& dbname,
|
||||
FileSystem* fs,
|
||||
std::string* manifest_path,
|
||||
@@ -4610,10 +4556,10 @@ Status VersionSet::Recover(
|
||||
reporter.status = &log_read_status;
|
||||
log::Reader reader(nullptr, std::move(manifest_file_reader), &reporter,
|
||||
true /* checksum */, 0 /* log_number */);
|
||||
VersionEditHandler handler(
|
||||
read_only, column_families, const_cast<VersionSet*>(this),
|
||||
/*track_missing_files=*/false,
|
||||
/*no_error_if_table_files_missing=*/false, io_tracer_);
|
||||
VersionEditHandler handler(read_only, column_families,
|
||||
const_cast<VersionSet*>(this),
|
||||
/*track_missing_files=*/false,
|
||||
/*no_error_if_files_missing=*/false, io_tracer_);
|
||||
handler.Iterate(reader, &log_read_status);
|
||||
s = handler.status();
|
||||
if (s.ok()) {
|
||||
@@ -4937,7 +4883,7 @@ Status VersionSet::ReduceNumberOfLevels(const std::string& dbname,
|
||||
}
|
||||
|
||||
// Get the checksum information including the checksum and checksum function
|
||||
// name of all SST files in VersionSet. Store the information in
|
||||
// name of all SST and blob files in VersionSet. Store the information in
|
||||
// FileChecksumList which contains a map from file number to its checksum info.
|
||||
// If DB is not running, make sure call VersionSet::Recover() to load the file
|
||||
// metadata from Manifest to VersionSet before calling this function.
|
||||
@@ -4954,6 +4900,7 @@ Status VersionSet::GetLiveFilesChecksumInfo(FileChecksumList* checksum_list) {
|
||||
if (cfd->IsDropped() || !cfd->initialized()) {
|
||||
continue;
|
||||
}
|
||||
/* SST files */
|
||||
for (int level = 0; level < cfd->NumberLevels(); level++) {
|
||||
for (const auto& file :
|
||||
cfd->current()->storage_info()->LevelFiles(level)) {
|
||||
@@ -4961,17 +4908,36 @@ Status VersionSet::GetLiveFilesChecksumInfo(FileChecksumList* checksum_list) {
|
||||
file->file_checksum,
|
||||
file->file_checksum_func_name);
|
||||
if (!s.ok()) {
|
||||
break;
|
||||
return s;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Blob files */
|
||||
const auto& blob_files = cfd->current()->storage_info()->GetBlobFiles();
|
||||
for (const auto& pair : blob_files) {
|
||||
const uint64_t blob_file_number = pair.first;
|
||||
const auto& meta = pair.second;
|
||||
|
||||
assert(meta);
|
||||
assert(blob_file_number == meta->GetBlobFileNumber());
|
||||
|
||||
std::string checksum_value = meta->GetChecksumValue();
|
||||
std::string checksum_method = meta->GetChecksumMethod();
|
||||
assert(checksum_value.empty() == checksum_method.empty());
|
||||
if (meta->GetChecksumMethod().empty()) {
|
||||
checksum_value = kUnknownFileChecksum;
|
||||
checksum_method = kUnknownFileChecksumFuncName;
|
||||
}
|
||||
|
||||
s = checksum_list->InsertOneFileChecksum(blob_file_number, checksum_value,
|
||||
checksum_method);
|
||||
if (!s.ok()) {
|
||||
break;
|
||||
return s;
|
||||
}
|
||||
}
|
||||
if (!s.ok()) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return s;
|
||||
}
|
||||
|
||||
|
||||
@@ -1331,10 +1331,6 @@ class VersionSet {
|
||||
ColumnFamilyData* CreateColumnFamily(const ColumnFamilyOptions& cf_options,
|
||||
const VersionEdit* edit);
|
||||
|
||||
Status ExtractInfoFromVersionEdit(ColumnFamilyData* cfd,
|
||||
const VersionEdit& from_edit,
|
||||
VersionEditParams* version_edit_params);
|
||||
|
||||
Status VerifyFileMetadata(const std::string& fpath,
|
||||
const FileMetaData& meta) const;
|
||||
|
||||
|
||||
@@ -144,7 +144,7 @@ DECLARE_bool(enable_write_thread_adaptive_yield);
|
||||
DECLARE_int32(reopen);
|
||||
DECLARE_double(bloom_bits);
|
||||
DECLARE_bool(use_block_based_filter);
|
||||
DECLARE_bool(use_ribbon_filter);
|
||||
DECLARE_int32(ribbon_starting_level);
|
||||
DECLARE_bool(partition_filters);
|
||||
DECLARE_bool(optimize_filters_for_memory);
|
||||
DECLARE_int32(index_type);
|
||||
|
||||
@@ -410,8 +410,8 @@ DEFINE_bool(use_block_based_filter, false,
|
||||
"use block based filter"
|
||||
"instead of full filter for block based table");
|
||||
|
||||
DEFINE_bool(use_ribbon_filter, false,
|
||||
"Use Ribbon filter instead of Bloom filter");
|
||||
DEFINE_int32(ribbon_starting_level, false,
|
||||
"First level to use Ribbon filter instead of Bloom");
|
||||
|
||||
DEFINE_bool(partition_filters, false,
|
||||
"use partitioned filters "
|
||||
|
||||
@@ -26,11 +26,12 @@ StressTest::StressTest()
|
||||
compressed_cache_(NewLRUCache(FLAGS_compressed_cache_size)),
|
||||
filter_policy_(
|
||||
FLAGS_bloom_bits >= 0
|
||||
? FLAGS_use_ribbon_filter
|
||||
? NewExperimentalRibbonFilterPolicy(FLAGS_bloom_bits)
|
||||
? FLAGS_ribbon_starting_level < FLAGS_num_levels
|
||||
? NewExperimentalRibbonFilterPolicy(
|
||||
FLAGS_bloom_bits, FLAGS_ribbon_starting_level)
|
||||
: FLAGS_use_block_based_filter
|
||||
? NewBloomFilterPolicy(FLAGS_bloom_bits, true)
|
||||
: NewBloomFilterPolicy(FLAGS_bloom_bits, false)
|
||||
? NewBloomFilterPolicy(FLAGS_bloom_bits, true)
|
||||
: NewBloomFilterPolicy(FLAGS_bloom_bits, false)
|
||||
: nullptr),
|
||||
db_(nullptr),
|
||||
#ifndef ROCKSDB_LITE
|
||||
@@ -2328,11 +2329,10 @@ void StressTest::Open() {
|
||||
|
||||
if ((options_.enable_blob_files || options_.enable_blob_garbage_collection ||
|
||||
FLAGS_allow_setting_blob_options_dynamically) &&
|
||||
(FLAGS_use_merge || FLAGS_backup_one_in > 0 ||
|
||||
FLAGS_best_efforts_recovery)) {
|
||||
(FLAGS_use_merge || FLAGS_best_efforts_recovery)) {
|
||||
fprintf(stderr,
|
||||
"Integrated BlobDB is currently incompatible with Merge, "
|
||||
"backup/restore, and best-effort recovery\n");
|
||||
"and best-effort recovery\n");
|
||||
exit(1);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
---
|
||||
title: (Call For Contribution) Make Universal Compaction More Incremental
|
||||
layout: post
|
||||
author: sdong
|
||||
category: blog
|
||||
---
|
||||
|
||||
### Motivation
|
||||
|
||||
Universal Compaction is an important compaction style, but few changes were made after we made the structure multi-leveled. Yet the major restriction of always compacting full sorted run is not relaxed. Compared to Leveled Compaction, where we usually only compile several SST files together, in universal compaction, we frequently compact GBs of data. Two issues with this gap: 1. it makes it harder to unify universal and leveled compaction; 2. periodically data is fully compacted, and in the mean time space is doubled. To ease the problem, we can break the restriction and do similar as leveled compaction, and bring it closer to unified compaction.
|
||||
|
||||
We call for help for making following improvements.
|
||||
|
||||
|
||||
### How Universal Compaction Works
|
||||
|
||||
In universal, whole levels are compacted together to satisfy two conditions (See [wiki page](https://github.com/facebook/rocksdb/wiki/Universal-Compaction) for more details):
|
||||
|
||||
1. total size / bottommost level size > a threshold, or
|
||||
2. total number of sorted runs (non-0 levels + L0 files) is within a threshold
|
||||
|
||||
1 is to limit extra space overhead used for dead data and 2 is for read performance.
|
||||
|
||||
If 1 is triggered, likely a full compaction will be triggered. If 2 is triggered, RocksDB compact some sorted runs to bring the number down. It does it by using a simple heuristic so that less writes needed for that purpose over time: it starts from compacting smaller files, but if total size to compact is similar to or larger than size of the next level, it will take that level together, as soon on (whether it is the best heuristic is another question and we’ve never seriously looked at it).
|
||||
|
||||
### How We Can Improve?
|
||||
|
||||
Let’s start from condition 1. Here we do full compaction but is not necessary. A simple optimization would be to compact so that just enough files are merged into the bottommost level (Lmax) to satisfy condition 1. It would work if we only need to pick some files from Lmax-1, or if it is cheaper over time, we can pick some files from other levels too.
|
||||
|
||||
Then condition 2. If we finish condition 1, there might be holes in some ranges in older levels. These holes might make it possible that only by compacting some sub ranges, we can fix the LSM-tree for condition 2. RocksDB can take single files into consideration and apply more sophisticated heuristic.
|
||||
|
||||
This new approach makes universal compaction closer to leveled compaction. The operation for 1 is closer to how Leveled compaction triggeres Lmax-1 to Lmax compaction. And 2 can potentially be implemented as something similar to level picking in Leveled Compaction. In fact, all those file picking can co-existing in one single compaction style and there isn’t fundamental conflicts to that.
|
||||
|
||||
### Limitation
|
||||
|
||||
There are two limitations:
|
||||
|
||||
* Periodic automatic full compaction is unpleasant but at the same time is pleasant in another way. Some users might uses it to reason that everything is periodically collapsed so dead data is gone and old data is rewritten. We need to make sure periodic compaction works to continue with that.
|
||||
* L0 to the first non-L0 level compaction is the first time data is partitioned in LSM-tree so that incremental compaction by range is possible. We might need to do more of these compactions in order to make incremental possible, which will increase compaction slightly.
|
||||
* Compacting subset of a level would introduce some extra overhead for unaligned files, just as in leveled compaction. More SST boundary cutting heuristic can reduce this overhead but it will be there.
|
||||
|
||||
But I believe the benefits would outweight the limitations. Reducing temporary space doubling and moving towards to unified compaction would be important achievements.
|
||||
|
||||
### Interested in Help?
|
||||
|
||||
Compaction is the core of LSM-tree, but its improvements are far overdue. If you are a user of universal compaction and would be able to benefit from those improvements, we will be happy to work with you on speeding up the project and bring them to RocksDB sooner. Feel free to communicate with us in [this issue](https://github.com/facebook/rocksdb/issues/8181).
|
||||
@@ -1068,6 +1068,37 @@ extern ROCKSDB_LIBRARY_API unsigned char
|
||||
rocksdb_options_get_skip_checking_sst_file_sizes_on_db_open(
|
||||
rocksdb_options_t* opt);
|
||||
|
||||
/* Blob Options Settings */
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_enable_blob_files(
|
||||
rocksdb_options_t* opt, unsigned char val);
|
||||
extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_enable_blob_files(
|
||||
rocksdb_options_t* opt);
|
||||
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_min_blob_size(
|
||||
rocksdb_options_t* opt, uint64_t val);
|
||||
extern ROCKSDB_LIBRARY_API uint64_t
|
||||
rocksdb_options_get_min_blob_size(rocksdb_options_t* opt);
|
||||
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_blob_file_size(
|
||||
rocksdb_options_t* opt, uint64_t val);
|
||||
extern ROCKSDB_LIBRARY_API uint64_t
|
||||
rocksdb_options_get_blob_file_size(rocksdb_options_t* opt);
|
||||
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_blob_compression_type(
|
||||
rocksdb_options_t* opt, int val);
|
||||
extern ROCKSDB_LIBRARY_API int rocksdb_options_get_blob_compression_type(
|
||||
rocksdb_options_t* opt);
|
||||
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_enable_blob_gc(
|
||||
rocksdb_options_t* opt, unsigned char val);
|
||||
extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_enable_blob_gc(
|
||||
rocksdb_options_t* opt);
|
||||
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_options_set_blob_gc_age_cutoff(
|
||||
rocksdb_options_t* opt, double val);
|
||||
extern ROCKSDB_LIBRARY_API double rocksdb_options_get_blob_gc_age_cutoff(
|
||||
rocksdb_options_t* opt);
|
||||
|
||||
/* returns a pointer to a malloc()-ed, null terminated string */
|
||||
extern ROCKSDB_LIBRARY_API char* rocksdb_options_statistics_get_string(
|
||||
rocksdb_options_t* opt);
|
||||
@@ -1712,6 +1743,8 @@ extern ROCKSDB_LIBRARY_API unsigned char rocksdb_flushoptions_get_wait(
|
||||
extern ROCKSDB_LIBRARY_API rocksdb_cache_t* rocksdb_cache_create_lru(
|
||||
size_t capacity);
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_cache_destroy(rocksdb_cache_t* cache);
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_cache_disown_data(
|
||||
rocksdb_cache_t* cache);
|
||||
extern ROCKSDB_LIBRARY_API void rocksdb_cache_set_capacity(
|
||||
rocksdb_cache_t* cache, size_t capacity);
|
||||
extern ROCKSDB_LIBRARY_API size_t
|
||||
|
||||
@@ -1368,7 +1368,7 @@ class DB {
|
||||
virtual void GetLiveFilesMetaData(
|
||||
std::vector<LiveFileMetaData>* /*metadata*/) {}
|
||||
|
||||
// Return a list of all table file checksum info.
|
||||
// Return a list of all table and blob files checksum info.
|
||||
// Note: This function might be of limited use because it cannot be
|
||||
// synchronized with GetLiveFiles.
|
||||
virtual Status GetLiveFilesChecksumInfo(FileChecksumList* checksum_list) = 0;
|
||||
|
||||
@@ -216,23 +216,33 @@ class FilterPolicy {
|
||||
extern const FilterPolicy* NewBloomFilterPolicy(
|
||||
double bits_per_key, bool use_block_based_builder = false);
|
||||
|
||||
// An EXPERIMENTAL new Bloom alternative that saves about 30% space
|
||||
// compared to Bloom filters, with about 3-4x construction time and
|
||||
// similar query times. For example, if you pass in 10 for
|
||||
// An new Bloom alternative that saves about 30% space compared to
|
||||
// Bloom filters, with about 3-4x construction time and similar
|
||||
// query times. For example, if you pass in 10 for
|
||||
// bloom_equivalent_bits_per_key, you'll get the same 0.95% FP rate
|
||||
// as Bloom filter but only using about 7 bits per key. (This
|
||||
// way of configuring the new filter is considered experimental
|
||||
// and/or transitional, so is expected to go away.)
|
||||
// and/or transitional, so is expected to be replaced with a new API.
|
||||
// The constructed filters will be given long-term support.)
|
||||
//
|
||||
// Ribbon filters are ignored by previous versions of RocksDB, as if
|
||||
// no filter was used.
|
||||
// The space savings of Ribbon filters makes sense for lower (higher
|
||||
// numbered; larger; longer-lived) levels of LSM, whereas the speed of
|
||||
// Bloom filters make sense for highest levels of LSM. Setting
|
||||
// ribbon_starting_level allows for this design. For example,
|
||||
// ribbon_starting_level=1 means that Bloom filters will be used in
|
||||
// level 0, including flushes, and Ribbon filters elsewhere.
|
||||
// ribbon_starting_level=0 means (almost) always use Ribbon.
|
||||
//
|
||||
// Note: this policy can generate Bloom filters in some cases.
|
||||
// For very small filters (well under 1KB), Bloom fallback is by
|
||||
// design, as the current Ribbon schema is not optimized to save vs.
|
||||
// Bloom for such small filters. Other cases of Bloom fallback should
|
||||
// be exceptional and log an appropriate warning.
|
||||
// Ribbon filters are compatible with RocksDB >= 6.15.0. Earlier
|
||||
// versions reading the data will behave as if no filter was used
|
||||
// (degraded performance until compaction rebuilds filters).
|
||||
//
|
||||
// Note: even with ribbon_starting_level=0, this policy can generate
|
||||
// Bloom filters in some cases. For very small filters (well under 1KB),
|
||||
// Bloom fallback is by design, as the current Ribbon schema is not
|
||||
// optimized to save vs. Bloom for such small filters. Other cases of
|
||||
// Bloom fallback should be exceptional and log an appropriate warning.
|
||||
extern const FilterPolicy* NewExperimentalRibbonFilterPolicy(
|
||||
double bloom_equivalent_bits_per_key);
|
||||
double bloom_equivalent_bits_per_key, int ribbon_starting_level = 1);
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -50,7 +50,15 @@ struct IOStatsContext {
|
||||
uint64_t cpu_read_nanos;
|
||||
};
|
||||
|
||||
// Get Thread-local IOStatsContext object pointer
|
||||
// If RocksDB is compiled with -DNIOSTATS_CONTEXT, then a pointer to a global,
|
||||
// non-thread-local IOStatsContext object will be returned. Attempts to update
|
||||
// this object will be ignored, and reading from it will also be no-op.
|
||||
// Otherwise,
|
||||
// a) if thread-local is supported on the platform, then a pointer to
|
||||
// a thread-local IOStatsContext object will be returned.
|
||||
// b) if thread-local is NOT supported, then compilation will fail.
|
||||
//
|
||||
// This function never returns nullptr.
|
||||
IOStatsContext* get_iostats_context();
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -118,6 +118,7 @@ enum class FlushReason : int {
|
||||
// When set the flush reason to kErrorRecoveryRetryFlush, SwitchMemtable
|
||||
// will not be called to avoid many small immutable memtables.
|
||||
kErrorRecoveryRetryFlush = 0xc,
|
||||
kWalFull = 0xd,
|
||||
};
|
||||
|
||||
// TODO: In the future, BackgroundErrorReason will only be used to indicate
|
||||
|
||||
@@ -230,8 +230,15 @@ struct PerfContext {
|
||||
bool per_level_perf_context_enabled = false;
|
||||
};
|
||||
|
||||
// Get Thread-local PerfContext object pointer
|
||||
// if defined(NPERF_CONTEXT), then the pointer is not thread-local
|
||||
// If RocksDB is compiled with -DNPERF_CONTEXT, then a pointer to a global,
|
||||
// non-thread-local PerfContext object will be returned. Attempts to update
|
||||
// this object will be ignored, and reading from it will also be no-op.
|
||||
// Otherwise,
|
||||
// a) if thread-local is supported on the platform, then a pointer to
|
||||
// a thread-local PerfContext object will be returned.
|
||||
// b) if thread-local is NOT supported, then compilation will fail.
|
||||
//
|
||||
// This function never returns nullptr.
|
||||
PerfContext* get_perf_context();
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -371,6 +371,21 @@ class BackupEngineReadOnlyBase {
|
||||
public:
|
||||
virtual ~BackupEngineReadOnlyBase() {}
|
||||
|
||||
// Returns info about the latest good backup in backup_info, or NotFound
|
||||
// no good backup exists.
|
||||
// Setting include_file_details=true provides information about each
|
||||
// backed-up file in BackupInfo::file_details and more.
|
||||
virtual Status GetLatestBackupInfo(
|
||||
BackupInfo* backup_info, bool include_file_details = false) const = 0;
|
||||
|
||||
// Returns info about a specific backup in backup_info, or NotFound
|
||||
// or Corruption status if the requested backup id does not exist or is
|
||||
// known corrupt.
|
||||
// Setting include_file_details=true provides information about each
|
||||
// backed-up file in BackupInfo::file_details and more.
|
||||
virtual Status GetBackupInfo(BackupID backup_id, BackupInfo* backup_info,
|
||||
bool include_file_details = false) const = 0;
|
||||
|
||||
// Returns info about backups in backup_info
|
||||
// Setting include_file_details=true provides information about each
|
||||
// backed-up file in BackupInfo::file_details and more.
|
||||
@@ -439,7 +454,7 @@ class BackupEngineAppendOnlyBase {
|
||||
// same as CreateNewBackup, but stores extra application metadata.
|
||||
virtual Status CreateNewBackupWithMetadata(
|
||||
const CreateBackupOptions& options, DB* db,
|
||||
const std::string& app_metadata) = 0;
|
||||
const std::string& app_metadata, BackupID* new_backup_id = nullptr) = 0;
|
||||
|
||||
// keep here for backward compatibility.
|
||||
virtual Status CreateNewBackupWithMetadata(
|
||||
@@ -451,9 +466,12 @@ class BackupEngineAppendOnlyBase {
|
||||
return CreateNewBackupWithMetadata(options, db, app_metadata);
|
||||
}
|
||||
|
||||
// Captures the state of the database by creating a new (latest) backup
|
||||
virtual Status CreateNewBackup(const CreateBackupOptions& options, DB* db) {
|
||||
return CreateNewBackupWithMetadata(options, db, "");
|
||||
// Captures the state of the database by creating a new (latest) backup.
|
||||
// On success (OK status), the BackupID of the new backup is saved to
|
||||
// *new_backup_id when not nullptr.
|
||||
virtual Status CreateNewBackup(const CreateBackupOptions& options, DB* db,
|
||||
BackupID* new_backup_id = nullptr) {
|
||||
return CreateNewBackupWithMetadata(options, db, "", new_backup_id);
|
||||
}
|
||||
|
||||
// keep here for backward compatibility.
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
#include "rocksdb/rocksdb_namespace.h"
|
||||
|
||||
#define ROCKSDB_MAJOR 6
|
||||
#define ROCKSDB_MINOR 19
|
||||
#define ROCKSDB_MINOR 20
|
||||
#define ROCKSDB_PATCH 0
|
||||
|
||||
// Do not use these. We made the mistake of declaring macros starting with
|
||||
|
||||
@@ -9,19 +9,23 @@
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
#ifdef ROCKSDB_SUPPORT_THREAD_LOCAL
|
||||
#ifdef NIOSTATS_CONTEXT
|
||||
// Should not be used because the counters are not thread-safe.
|
||||
// Put here just to make get_iostats_context() simple without ifdef.
|
||||
static IOStatsContext iostats_context;
|
||||
#elif defined(ROCKSDB_SUPPORT_THREAD_LOCAL)
|
||||
__thread IOStatsContext iostats_context;
|
||||
#else
|
||||
#error \
|
||||
"No thread-local support. Disable iostats context with -DNIOSTATS_CONTEXT."
|
||||
#endif
|
||||
|
||||
IOStatsContext* get_iostats_context() {
|
||||
#ifdef ROCKSDB_SUPPORT_THREAD_LOCAL
|
||||
return &iostats_context;
|
||||
#else
|
||||
return nullptr;
|
||||
#endif
|
||||
}
|
||||
|
||||
void IOStatsContext::Reset() {
|
||||
#ifndef NIOSTATS_CONTEXT
|
||||
thread_pool_id = Env::Priority::TOTAL;
|
||||
bytes_read = 0;
|
||||
bytes_written = 0;
|
||||
@@ -35,6 +39,7 @@ void IOStatsContext::Reset() {
|
||||
logger_nanos = 0;
|
||||
cpu_write_nanos = 0;
|
||||
cpu_read_nanos = 0;
|
||||
#endif //! NIOSTATS_CONTEXT
|
||||
}
|
||||
|
||||
#define IOSTATS_CONTEXT_OUTPUT(counter) \
|
||||
@@ -43,6 +48,10 @@ void IOStatsContext::Reset() {
|
||||
}
|
||||
|
||||
std::string IOStatsContext::ToString(bool exclude_zero_counters) const {
|
||||
#ifdef NIOSTATS_CONTEXT
|
||||
(void)exclude_zero_counters;
|
||||
return "";
|
||||
#else
|
||||
std::ostringstream ss;
|
||||
IOSTATS_CONTEXT_OUTPUT(thread_pool_id);
|
||||
IOSTATS_CONTEXT_OUTPUT(bytes_read);
|
||||
@@ -61,6 +70,7 @@ std::string IOStatsContext::ToString(bool exclude_zero_counters) const {
|
||||
std::string str = ss.str();
|
||||
str.erase(str.find_last_not_of(", ") + 1);
|
||||
return str;
|
||||
#endif //! NIOSTATS_CONTEXT
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
#include "monitoring/perf_step_timer.h"
|
||||
#include "rocksdb/iostats_context.h"
|
||||
|
||||
#ifdef ROCKSDB_SUPPORT_THREAD_LOCAL
|
||||
#if defined(ROCKSDB_SUPPORT_THREAD_LOCAL) && !defined(NIOSTATS_CONTEXT)
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
extern __thread IOStatsContext iostats_context;
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
@@ -44,7 +44,7 @@ extern __thread IOStatsContext iostats_context;
|
||||
PerfLevel::kEnableTimeAndCPUTimeExceptForMutex); \
|
||||
iostats_step_timer_##metric.Start();
|
||||
|
||||
#else // ROCKSDB_SUPPORT_THREAD_LOCAL
|
||||
#else // ROCKSDB_SUPPORT_THREAD_LOCAL && !NIOSTATS_CONTEXT
|
||||
|
||||
#define IOSTATS_ADD(metric, value)
|
||||
#define IOSTATS_ADD_IF_POSITIVE(metric, value)
|
||||
@@ -57,4 +57,4 @@ extern __thread IOStatsContext iostats_context;
|
||||
#define IOSTATS_TIMER_GUARD(metric)
|
||||
#define IOSTATS_CPU_TIMER_GUARD(metric, clock) static_cast<void>(clock)
|
||||
|
||||
#endif // ROCKSDB_SUPPORT_THREAD_LOCAL
|
||||
#endif // ROCKSDB_SUPPORT_THREAD_LOCAL && !NIOSTATS_CONTEXT
|
||||
|
||||
@@ -9,26 +9,22 @@
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
#if defined(NPERF_CONTEXT) || !defined(ROCKSDB_SUPPORT_THREAD_LOCAL)
|
||||
#if defined(NPERF_CONTEXT)
|
||||
// Should not be used because the counters are not thread-safe.
|
||||
// Put here just to make get_perf_context() simple without ifdef.
|
||||
PerfContext perf_context;
|
||||
#else
|
||||
#elif defined(ROCKSDB_SUPPORT_THREAD_LOCAL)
|
||||
#if defined(OS_SOLARIS)
|
||||
__thread PerfContext perf_context_;
|
||||
#else
|
||||
__thread PerfContext perf_context;
|
||||
#else // OS_SOLARIS
|
||||
thread_local PerfContext perf_context;
|
||||
#endif
|
||||
#endif // OS_SOLARIS
|
||||
#else
|
||||
#error "No thread-local support. Disable perf context with -DNPERF_CONTEXT."
|
||||
#endif
|
||||
|
||||
PerfContext* get_perf_context() {
|
||||
#if defined(NPERF_CONTEXT) || !defined(ROCKSDB_SUPPORT_THREAD_LOCAL)
|
||||
return &perf_context;
|
||||
#else
|
||||
#if defined(OS_SOLARIS)
|
||||
return &perf_context_;
|
||||
#else
|
||||
return &perf_context;
|
||||
#endif
|
||||
#endif
|
||||
}
|
||||
|
||||
PerfContext::~PerfContext() {
|
||||
|
||||
@@ -940,6 +940,15 @@ TEST_F(OptionsTest, GetBlockBasedTableOptionsFromString) {
|
||||
&new_opt));
|
||||
ASSERT_TRUE(new_opt.filter_policy != nullptr);
|
||||
bfp = dynamic_cast<const BloomFilterPolicy*>(new_opt.filter_policy.get());
|
||||
// Not a BloomFilterPolicy
|
||||
EXPECT_FALSE(bfp);
|
||||
|
||||
ASSERT_OK(GetBlockBasedTableOptionsFromString(
|
||||
config_options, table_opt, "filter_policy=experimental_ribbon:5.678:0;",
|
||||
&new_opt));
|
||||
ASSERT_TRUE(new_opt.filter_policy != nullptr);
|
||||
bfp = dynamic_cast<const BloomFilterPolicy*>(new_opt.filter_policy.get());
|
||||
// Pure Ribbon configuration is (oddly) BloomFilterPolicy
|
||||
EXPECT_EQ(bfp->GetMillibitsPerKey(), 5678);
|
||||
EXPECT_EQ(bfp->GetMode(), BloomFilterPolicy::kStandard128Ribbon);
|
||||
|
||||
|
||||
+2
-1
@@ -682,7 +682,8 @@ IOStatus WinFileSystem::GetChildren(const std::string& dir,
|
||||
// which appear only on some platforms
|
||||
const bool ignore =
|
||||
((data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY) != 0) &&
|
||||
(strcmp(data.cFileName, ".") == 0 || strcmp(data.cFileName, "..") == 0);
|
||||
(RX_FNCMP(data.cFileName, ".") == 0 ||
|
||||
RX_FNCMP(data.cFileName, "..") == 0);
|
||||
if (!ignore) {
|
||||
auto x = RX_FILESTRING(data.cFileName, RX_FNLEN(data.cFileName));
|
||||
result->push_back(FN_TO_RX(x));
|
||||
|
||||
@@ -355,6 +355,7 @@ extern void SetCpuPriority(ThreadId id, CpuPriority priority);
|
||||
#define RX_FILESTRING std::wstring
|
||||
#define RX_FN(a) ROCKSDB_NAMESPACE::port::utf8_to_utf16(a)
|
||||
#define FN_TO_RX(a) ROCKSDB_NAMESPACE::port::utf16_to_utf8(a)
|
||||
#define RX_FNCMP(a, b) ::wcscmp(a, RX_FN(b).c_str())
|
||||
#define RX_FNLEN(a) ::wcslen(a)
|
||||
|
||||
#define RX_DeleteFile DeleteFileW
|
||||
@@ -379,6 +380,7 @@ extern void SetCpuPriority(ThreadId id, CpuPriority priority);
|
||||
#define RX_FILESTRING std::string
|
||||
#define RX_FN(a) a
|
||||
#define FN_TO_RX(a) a
|
||||
#define RX_FNCMP(a, b) strcmp(a, b)
|
||||
#define RX_FNLEN(a) strlen(a)
|
||||
|
||||
#define RX_DeleteFile DeleteFileA
|
||||
|
||||
@@ -205,6 +205,7 @@ LIB_SOURCES = \
|
||||
util/compression_context_cache.cc \
|
||||
util/concurrent_task_limiter_impl.cc \
|
||||
util/crc32c.cc \
|
||||
util/crc32c_arm64.cc \
|
||||
util/dynamic_bloom.cc \
|
||||
util/hash.cc \
|
||||
util/murmurhash.cc \
|
||||
@@ -275,11 +276,6 @@ LIB_SOURCES = \
|
||||
utilities/write_batch_with_index/write_batch_with_index.cc \
|
||||
utilities/write_batch_with_index/write_batch_with_index_internal.cc \
|
||||
|
||||
ifeq ($(ARMCRC_SOURCE),1)
|
||||
LIB_SOURCES +=\
|
||||
util/crc32c_arm64.cc
|
||||
endif
|
||||
|
||||
ifeq (,$(shell $(CXX) -fsyntax-only -maltivec -xc /dev/null 2>&1))
|
||||
LIB_SOURCES_ASM =\
|
||||
util/crc32c_ppc_asm.S
|
||||
|
||||
@@ -23,6 +23,7 @@
|
||||
#include "util/hash.h"
|
||||
#include "util/ribbon_config.h"
|
||||
#include "util/ribbon_impl.h"
|
||||
#include "util/string_util.h"
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
@@ -1054,7 +1055,7 @@ BloomFilterPolicy::BloomFilterPolicy(double bits_per_key, Mode mode)
|
||||
|
||||
BloomFilterPolicy::~BloomFilterPolicy() {}
|
||||
|
||||
const char* BloomFilterPolicy::Name() const {
|
||||
const char* BuiltinFilterPolicy::Name() const {
|
||||
return "rocksdb.BuiltinBloomFilter";
|
||||
}
|
||||
|
||||
@@ -1087,8 +1088,8 @@ void BloomFilterPolicy::CreateFilter(const Slice* keys, int n,
|
||||
}
|
||||
}
|
||||
|
||||
bool BloomFilterPolicy::KeyMayMatch(const Slice& key,
|
||||
const Slice& bloom_filter) const {
|
||||
bool BuiltinFilterPolicy::KeyMayMatch(const Slice& key,
|
||||
const Slice& bloom_filter) const {
|
||||
const size_t len = bloom_filter.size();
|
||||
if (len < 2 || len > 0xffffffffU) {
|
||||
return false;
|
||||
@@ -1110,7 +1111,7 @@ bool BloomFilterPolicy::KeyMayMatch(const Slice& key,
|
||||
array);
|
||||
}
|
||||
|
||||
FilterBitsBuilder* BloomFilterPolicy::GetFilterBitsBuilder() const {
|
||||
FilterBitsBuilder* BuiltinFilterPolicy::GetFilterBitsBuilder() const {
|
||||
// This code path should no longer be used, for the built-in
|
||||
// BloomFilterPolicy. Internal to RocksDB and outside
|
||||
// BloomFilterPolicy, only get a FilterBitsBuilder with
|
||||
@@ -1184,7 +1185,7 @@ FilterBitsBuilder* BloomFilterPolicy::GetBuilderFromContext(
|
||||
|
||||
// Read metadata to determine what kind of FilterBitsReader is needed
|
||||
// and return a new one.
|
||||
FilterBitsReader* BloomFilterPolicy::GetFilterBitsReader(
|
||||
FilterBitsReader* BuiltinFilterPolicy::GetFilterBitsReader(
|
||||
const Slice& contents) const {
|
||||
uint32_t len_with_meta = static_cast<uint32_t>(contents.size());
|
||||
if (len_with_meta <= kMetadataLen) {
|
||||
@@ -1265,7 +1266,7 @@ FilterBitsReader* BloomFilterPolicy::GetFilterBitsReader(
|
||||
log2_cache_line_size);
|
||||
}
|
||||
|
||||
FilterBitsReader* BloomFilterPolicy::GetRibbonBitsReader(
|
||||
FilterBitsReader* BuiltinFilterPolicy::GetRibbonBitsReader(
|
||||
const Slice& contents) const {
|
||||
uint32_t len_with_meta = static_cast<uint32_t>(contents.size());
|
||||
uint32_t len = len_with_meta - kMetadataLen;
|
||||
@@ -1289,7 +1290,7 @@ FilterBitsReader* BloomFilterPolicy::GetRibbonBitsReader(
|
||||
}
|
||||
|
||||
// For newer Bloom filter implementations
|
||||
FilterBitsReader* BloomFilterPolicy::GetBloomBitsReader(
|
||||
FilterBitsReader* BuiltinFilterPolicy::GetBloomBitsReader(
|
||||
const Slice& contents) const {
|
||||
uint32_t len_with_meta = static_cast<uint32_t>(contents.size());
|
||||
uint32_t len = len_with_meta - kMetadataLen;
|
||||
@@ -1362,10 +1363,50 @@ const FilterPolicy* NewBloomFilterPolicy(double bits_per_key,
|
||||
return new BloomFilterPolicy(bits_per_key, m);
|
||||
}
|
||||
|
||||
// Chooses between two filter policies based on LSM level
|
||||
class LevelThresholdFilterPolicy : public BuiltinFilterPolicy {
|
||||
public:
|
||||
LevelThresholdFilterPolicy(std::unique_ptr<const FilterPolicy>&& a,
|
||||
std::unique_ptr<const FilterPolicy>&& b,
|
||||
int starting_level_for_b)
|
||||
: policy_a_(std::move(a)),
|
||||
policy_b_(std::move(b)),
|
||||
starting_level_for_b_(starting_level_for_b) {
|
||||
assert(starting_level_for_b_ >= 0);
|
||||
}
|
||||
|
||||
// Deprecated block-based filter only
|
||||
void CreateFilter(const Slice* keys, int n, std::string* dst) const override {
|
||||
policy_a_->CreateFilter(keys, n, dst);
|
||||
}
|
||||
|
||||
FilterBitsBuilder* GetBuilderWithContext(
|
||||
const FilterBuildingContext& context) const override {
|
||||
if (context.level_at_creation >= starting_level_for_b_) {
|
||||
return policy_b_->GetBuilderWithContext(context);
|
||||
} else {
|
||||
return policy_a_->GetBuilderWithContext(context);
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
const std::unique_ptr<const FilterPolicy> policy_a_;
|
||||
const std::unique_ptr<const FilterPolicy> policy_b_;
|
||||
int starting_level_for_b_;
|
||||
};
|
||||
|
||||
extern const FilterPolicy* NewExperimentalRibbonFilterPolicy(
|
||||
double bloom_equivalent_bits_per_key) {
|
||||
return new BloomFilterPolicy(bloom_equivalent_bits_per_key,
|
||||
BloomFilterPolicy::kStandard128Ribbon);
|
||||
double bloom_equivalent_bits_per_key, int ribbon_starting_level) {
|
||||
std::unique_ptr<const FilterPolicy> ribbon_only{new BloomFilterPolicy(
|
||||
bloom_equivalent_bits_per_key, BloomFilterPolicy::kStandard128Ribbon)};
|
||||
if (ribbon_starting_level > 0) {
|
||||
std::unique_ptr<const FilterPolicy> bloom_only{new BloomFilterPolicy(
|
||||
bloom_equivalent_bits_per_key, BloomFilterPolicy::kFastLocalBloom)};
|
||||
return new LevelThresholdFilterPolicy(
|
||||
std::move(bloom_only), std::move(ribbon_only), ribbon_starting_level);
|
||||
} else {
|
||||
return ribbon_only.release();
|
||||
}
|
||||
}
|
||||
|
||||
FilterBuildingContext::FilterBuildingContext(
|
||||
@@ -1396,10 +1437,18 @@ Status FilterPolicy::CreateFromString(
|
||||
NewBloomFilterPolicy(bits_per_key, use_block_based_builder));
|
||||
}
|
||||
} else if (value.compare(0, kExpRibbonName.size(), kExpRibbonName) == 0) {
|
||||
size_t pos = value.find(':', kExpRibbonName.size());
|
||||
int ribbon_starting_level;
|
||||
if (pos == std::string::npos) {
|
||||
pos = value.size();
|
||||
ribbon_starting_level = 1;
|
||||
} else {
|
||||
ribbon_starting_level = ParseInt(trim(value.substr(pos + 1)));
|
||||
}
|
||||
double bloom_equivalent_bits_per_key =
|
||||
ParseDouble(trim(value.substr(kExpRibbonName.size())));
|
||||
policy->reset(
|
||||
NewExperimentalRibbonFilterPolicy(bloom_equivalent_bits_per_key));
|
||||
ParseDouble(trim(value.substr(kExpRibbonName.size(), pos)));
|
||||
policy->reset(NewExperimentalRibbonFilterPolicy(
|
||||
bloom_equivalent_bits_per_key, ribbon_starting_level));
|
||||
} else {
|
||||
return Status::NotFound("Invalid filter policy name ", value);
|
||||
#else
|
||||
|
||||
@@ -38,10 +38,38 @@ class BuiltinFilterBitsBuilder : public FilterBitsBuilder {
|
||||
virtual double EstimatedFpRate(size_t num_entries, size_t bytes) = 0;
|
||||
};
|
||||
|
||||
// Abstract base class for RocksDB built-in filter policies.
|
||||
// This class is considered internal API and subject to change.
|
||||
class BuiltinFilterPolicy : public FilterPolicy {
|
||||
public:
|
||||
// Shared name because any built-in policy can read filters from
|
||||
// any other
|
||||
const char* Name() const override;
|
||||
|
||||
// Deprecated block-based filter only
|
||||
bool KeyMayMatch(const Slice& key, const Slice& bloom_filter) const override;
|
||||
|
||||
// Old API
|
||||
FilterBitsBuilder* GetFilterBitsBuilder() const override;
|
||||
|
||||
// Read metadata to determine what kind of FilterBitsReader is needed
|
||||
// and return a new one. This must successfully process any filter data
|
||||
// generated by a built-in FilterBitsBuilder, regardless of the impl
|
||||
// chosen for this BloomFilterPolicy. Not compatible with CreateFilter.
|
||||
FilterBitsReader* GetFilterBitsReader(const Slice& contents) const override;
|
||||
|
||||
private:
|
||||
// For newer Bloom filter implementation(s)
|
||||
FilterBitsReader* GetBloomBitsReader(const Slice& contents) const;
|
||||
|
||||
// For Ribbon filter implementation(s)
|
||||
FilterBitsReader* GetRibbonBitsReader(const Slice& contents) const;
|
||||
};
|
||||
|
||||
// RocksDB built-in filter policy for Bloom or Bloom-like filters.
|
||||
// This class is considered internal API and subject to change.
|
||||
// See NewBloomFilterPolicy.
|
||||
class BloomFilterPolicy : public FilterPolicy {
|
||||
class BloomFilterPolicy : public BuiltinFilterPolicy {
|
||||
public:
|
||||
// An internal marker for operating modes of BloomFilterPolicy, in terms
|
||||
// of selecting an implementation. This makes it easier for tests to track
|
||||
@@ -88,16 +116,9 @@ class BloomFilterPolicy : public FilterPolicy {
|
||||
|
||||
~BloomFilterPolicy() override;
|
||||
|
||||
const char* Name() const override;
|
||||
|
||||
// Deprecated block-based filter only
|
||||
void CreateFilter(const Slice* keys, int n, std::string* dst) const override;
|
||||
|
||||
// Deprecated block-based filter only
|
||||
bool KeyMayMatch(const Slice& key, const Slice& bloom_filter) const override;
|
||||
|
||||
FilterBitsBuilder* GetFilterBitsBuilder() const override;
|
||||
|
||||
// To use this function, call GetBuilderFromContext().
|
||||
//
|
||||
// Neither the context nor any objects therein should be saved beyond
|
||||
@@ -110,12 +131,6 @@ class BloomFilterPolicy : public FilterPolicy {
|
||||
// (An internal convenience function to save boilerplate.)
|
||||
static FilterBitsBuilder* GetBuilderFromContext(const FilterBuildingContext&);
|
||||
|
||||
// Read metadata to determine what kind of FilterBitsReader is needed
|
||||
// and return a new one. This must successfully process any filter data
|
||||
// generated by a built-in FilterBitsBuilder, regardless of the impl
|
||||
// chosen for this BloomFilterPolicy. Not compatible with CreateFilter.
|
||||
FilterBitsReader* GetFilterBitsReader(const Slice& contents) const override;
|
||||
|
||||
// Essentially for testing only: configured millibits/key
|
||||
int GetMillibitsPerKey() const { return millibits_per_key_; }
|
||||
// Essentially for testing only: legacy whole bits/key
|
||||
@@ -157,12 +172,6 @@ class BloomFilterPolicy : public FilterPolicy {
|
||||
// Sum over all generated filters f:
|
||||
// (predicted_fp_rate(f) - predicted_fp_rate(f|o_f_f_m=false)) * 2^32
|
||||
mutable std::atomic<int64_t> aggregate_rounding_balance_;
|
||||
|
||||
// For newer Bloom filter implementation(s)
|
||||
FilterBitsReader* GetBloomBitsReader(const Slice& contents) const;
|
||||
|
||||
// For Ribbon filter implementation(s)
|
||||
FilterBitsReader* GetRibbonBitsReader(const Slice& contents) const;
|
||||
};
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
+17
-1
@@ -23,6 +23,9 @@
|
||||
#include <mach/mach_host.h>
|
||||
#include <sys/sysctl.h>
|
||||
#endif
|
||||
#ifdef __FreeBSD__
|
||||
#include <sys/sysctl.h>
|
||||
#endif
|
||||
#include <atomic>
|
||||
#include <cinttypes>
|
||||
#include <condition_variable>
|
||||
@@ -2638,7 +2641,7 @@ class Benchmark {
|
||||
fprintf(stderr, "RocksDB: version %d.%d\n",
|
||||
kMajorVersion, kMinorVersion);
|
||||
|
||||
#if defined(__linux) || defined(__APPLE__)
|
||||
#if defined(__linux) || defined(__APPLE__) || defined(__FreeBSD__)
|
||||
time_t now = time(nullptr);
|
||||
char buf[52];
|
||||
// Lint complains about ctime() usage, so replace it with ctime_r(). The
|
||||
@@ -2696,6 +2699,19 @@ class Benchmark {
|
||||
fprintf(stderr, "CPU: %d * %s\n", h.max_cpus, cpu_type.c_str());
|
||||
fprintf(stderr, "CPUCache: %s\n", cache_size.c_str());
|
||||
}
|
||||
#elif defined(__FreeBSD__)
|
||||
int ncpus;
|
||||
size_t len = sizeof(ncpus);
|
||||
int mib[2] = {CTL_HW, HW_NCPU};
|
||||
if (sysctl(mib, 2, &ncpus, &len, nullptr, 0) == 0) {
|
||||
char cpu_type[16];
|
||||
len = sizeof(cpu_type) - 1;
|
||||
mib[1] = HW_MACHINE;
|
||||
if (sysctl(mib, 2, cpu_type, &len, nullptr, 0) == 0) cpu_type[len] = 0;
|
||||
|
||||
fprintf(stderr, "CPU: %d * %s\n", ncpus, cpu_type);
|
||||
// no programmatic way to get the cache line size except on PPC
|
||||
}
|
||||
#endif
|
||||
#endif
|
||||
}
|
||||
|
||||
@@ -102,7 +102,7 @@ default_params = {
|
||||
"mock_direct_io": False,
|
||||
"use_full_merge_v1": lambda: random.randint(0, 1),
|
||||
"use_merge": lambda: random.randint(0, 1),
|
||||
"use_ribbon_filter": lambda: random.randint(0, 1),
|
||||
"ribbon_starting_level": lambda: random.randint(0, 10),
|
||||
"verify_checksum": 1,
|
||||
"write_buffer_size": 4 * 1024 * 1024,
|
||||
"writepercent": 35,
|
||||
@@ -279,7 +279,6 @@ blob_params = {
|
||||
"blob_garbage_collection_age_cutoff": lambda: random.choice([0.0, 0.25, 0.5, 0.75, 1.0]),
|
||||
# The following are currently incompatible with the integrated BlobDB
|
||||
"use_merge": 0,
|
||||
"backup_one_in": 0,
|
||||
}
|
||||
|
||||
ts_params = {
|
||||
|
||||
@@ -349,6 +349,85 @@ TEST_F(LdbCmdTest, DumpFileChecksumNoChecksum) {
|
||||
ASSERT_OK(fct_helper_ac.VerifyChecksumInManifest(live_files));
|
||||
}
|
||||
|
||||
TEST_F(LdbCmdTest, BlobDBDumpFileChecksumNoChecksum) {
|
||||
Env* base_env = TryLoadCustomOrDefaultEnv();
|
||||
std::unique_ptr<Env> env(NewMemEnv(base_env));
|
||||
Options opts;
|
||||
opts.env = env.get();
|
||||
opts.create_if_missing = true;
|
||||
opts.enable_blob_files = true;
|
||||
|
||||
DB* db = nullptr;
|
||||
std::string dbname = test::PerThreadDBPath(env.get(), "ldb_cmd_test");
|
||||
ASSERT_OK(DB::Open(opts, dbname, &db));
|
||||
|
||||
WriteOptions wopts;
|
||||
FlushOptions fopts;
|
||||
fopts.wait = true;
|
||||
Random rnd(test::RandomSeed());
|
||||
for (int i = 0; i < 200; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 100; i < 300; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 200; i < 400; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 300; i < 400; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
|
||||
char arg1[] = "./ldb";
|
||||
std::string arg2_str = "--db=" + dbname;
|
||||
char arg3[] = "file_checksum_dump";
|
||||
char* argv[] = {arg1, const_cast<char*>(arg2_str.c_str()), arg3};
|
||||
|
||||
ASSERT_EQ(0,
|
||||
LDBCommandRunner::RunCommand(3, argv, opts, LDBOptions(), nullptr));
|
||||
|
||||
// Verify each sst and blob file checksum value and checksum name
|
||||
FileChecksumTestHelper fct_helper(opts, db, dbname);
|
||||
ASSERT_OK(fct_helper.VerifyEachFileChecksum());
|
||||
|
||||
// Manually trigger compaction
|
||||
std::ostringstream oss_b_buf;
|
||||
oss_b_buf << std::setfill('0') << std::setw(8) << std::fixed << 0;
|
||||
std::ostringstream oss_e_buf;
|
||||
oss_e_buf << std::setfill('0') << std::setw(8) << std::fixed << 399;
|
||||
std::string b_buf = oss_b_buf.str();
|
||||
std::string e_buf = oss_e_buf.str();
|
||||
Slice begin(b_buf);
|
||||
Slice end(e_buf);
|
||||
|
||||
CompactRangeOptions options;
|
||||
ASSERT_OK(db->CompactRange(options, &begin, &end));
|
||||
// Verify each sst file checksum after compaction
|
||||
FileChecksumTestHelper fct_helper_ac(opts, db, dbname);
|
||||
ASSERT_OK(fct_helper_ac.VerifyEachFileChecksum());
|
||||
|
||||
ASSERT_EQ(0,
|
||||
LDBCommandRunner::RunCommand(3, argv, opts, LDBOptions(), nullptr));
|
||||
|
||||
delete db;
|
||||
}
|
||||
|
||||
TEST_F(LdbCmdTest, DumpFileChecksumCRC32) {
|
||||
Env* base_env = TryLoadCustomOrDefaultEnv();
|
||||
std::unique_ptr<Env> env(NewMemEnv(base_env));
|
||||
@@ -430,6 +509,85 @@ TEST_F(LdbCmdTest, DumpFileChecksumCRC32) {
|
||||
ASSERT_OK(fct_helper_ac.VerifyChecksumInManifest(live_files));
|
||||
}
|
||||
|
||||
TEST_F(LdbCmdTest, BlobDBDumpFileChecksumCRC32) {
|
||||
Env* base_env = TryLoadCustomOrDefaultEnv();
|
||||
std::unique_ptr<Env> env(NewMemEnv(base_env));
|
||||
Options opts;
|
||||
opts.env = env.get();
|
||||
opts.create_if_missing = true;
|
||||
opts.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
|
||||
opts.enable_blob_files = true;
|
||||
|
||||
DB* db = nullptr;
|
||||
std::string dbname = test::PerThreadDBPath(env.get(), "ldb_cmd_test");
|
||||
ASSERT_OK(DB::Open(opts, dbname, &db));
|
||||
|
||||
WriteOptions wopts;
|
||||
FlushOptions fopts;
|
||||
fopts.wait = true;
|
||||
Random rnd(test::RandomSeed());
|
||||
for (int i = 0; i < 100; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 50; i < 150; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 100; i < 200; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
for (int i = 150; i < 250; i++) {
|
||||
std::ostringstream oss;
|
||||
oss << std::setfill('0') << std::setw(8) << std::fixed << i;
|
||||
std::string v = rnd.RandomString(100);
|
||||
ASSERT_OK(db->Put(wopts, oss.str(), v));
|
||||
}
|
||||
ASSERT_OK(db->Flush(fopts));
|
||||
|
||||
char arg1[] = "./ldb";
|
||||
std::string arg2_str = "--db=" + dbname;
|
||||
char arg3[] = "file_checksum_dump";
|
||||
char* argv[] = {arg1, const_cast<char*>(arg2_str.c_str()), arg3};
|
||||
|
||||
ASSERT_EQ(0,
|
||||
LDBCommandRunner::RunCommand(3, argv, opts, LDBOptions(), nullptr));
|
||||
|
||||
// Verify each sst and blob file checksum value and checksum name
|
||||
FileChecksumTestHelper fct_helper(opts, db, dbname);
|
||||
ASSERT_OK(fct_helper.VerifyEachFileChecksum());
|
||||
|
||||
// Manually trigger compaction
|
||||
std::ostringstream oss_b_buf;
|
||||
oss_b_buf << std::setfill('0') << std::setw(8) << std::fixed << 0;
|
||||
std::ostringstream oss_e_buf;
|
||||
oss_e_buf << std::setfill('0') << std::setw(8) << std::fixed << 249;
|
||||
std::string b_buf = oss_b_buf.str();
|
||||
std::string e_buf = oss_e_buf.str();
|
||||
Slice begin(b_buf);
|
||||
Slice end(e_buf);
|
||||
|
||||
CompactRangeOptions options;
|
||||
ASSERT_OK(db->CompactRange(options, &begin, &end));
|
||||
// Verify each sst file checksum after compaction
|
||||
FileChecksumTestHelper fct_helper_ac(opts, db, dbname);
|
||||
ASSERT_OK(fct_helper_ac.VerifyEachFileChecksum());
|
||||
|
||||
ASSERT_EQ(0,
|
||||
LDBCommandRunner::RunCommand(3, argv, opts, LDBOptions(), nullptr));
|
||||
delete db;
|
||||
}
|
||||
|
||||
TEST_F(LdbCmdTest, OptionParsing) {
|
||||
// test parsing flags
|
||||
Options opts;
|
||||
|
||||
@@ -1195,6 +1195,51 @@ INSTANTIATE_TEST_CASE_P(Full, FullBloomTest,
|
||||
BloomFilterPolicy::kFastLocalBloom,
|
||||
BloomFilterPolicy::kStandard128Ribbon));
|
||||
|
||||
static double GetEffectiveBitsPerKey(FilterBitsBuilder* builder) {
|
||||
union {
|
||||
uint64_t key_value;
|
||||
char key_bytes[8];
|
||||
};
|
||||
|
||||
const unsigned kNumKeys = 1000;
|
||||
|
||||
Slice key_slice{key_bytes, 8};
|
||||
for (key_value = 0; key_value < kNumKeys; ++key_value) {
|
||||
builder->AddKey(key_slice);
|
||||
}
|
||||
|
||||
std::unique_ptr<const char[]> buf;
|
||||
auto filter = builder->Finish(&buf);
|
||||
return filter.size() * /*bits per byte*/ 8 / (1.0 * kNumKeys);
|
||||
}
|
||||
|
||||
TEST(RibbonTest, RibbonTestLevelThreshold) {
|
||||
BlockBasedTableOptions opts;
|
||||
FilterBuildingContext ctx(opts);
|
||||
// A few settings
|
||||
for (int ribbon_starting_level : {0, 1, 10}) {
|
||||
std::unique_ptr<const FilterPolicy> policy{
|
||||
NewExperimentalRibbonFilterPolicy(8, ribbon_starting_level)};
|
||||
|
||||
// Claim to be generating filter for this level
|
||||
ctx.level_at_creation = ribbon_starting_level;
|
||||
std::unique_ptr<FilterBitsBuilder> builder{
|
||||
policy->GetBuilderWithContext(ctx)};
|
||||
|
||||
// Must be Ribbon (more space efficient than 8 bits per key)
|
||||
ASSERT_LT(GetEffectiveBitsPerKey(builder.get()), 7.5);
|
||||
|
||||
if (ribbon_starting_level > 0) {
|
||||
// Claim to be generating filter for this level
|
||||
ctx.level_at_creation = ribbon_starting_level - 1;
|
||||
builder.reset(policy->GetBuilderWithContext(ctx));
|
||||
|
||||
// Must be Bloom (~ 8 bits per key)
|
||||
ASSERT_GT(GetEffectiveBitsPerKey(builder.get()), 7.5);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
||||
@@ -129,7 +129,8 @@ class BackupEngineImpl {
|
||||
~BackupEngineImpl();
|
||||
|
||||
Status CreateNewBackupWithMetadata(const CreateBackupOptions& options, DB* db,
|
||||
const std::string& app_metadata);
|
||||
const std::string& app_metadata,
|
||||
BackupID* new_backup_id_ptr);
|
||||
|
||||
Status PurgeOldBackups(uint32_t num_backups_to_keep);
|
||||
|
||||
@@ -144,6 +145,9 @@ class BackupEngineImpl {
|
||||
void GetBackupInfo(std::vector<BackupInfo>* backup_info,
|
||||
bool include_file_details) const;
|
||||
|
||||
Status GetBackupInfo(BackupID backup_id, BackupInfo* backup_info,
|
||||
bool include_file_details = false) const;
|
||||
|
||||
void GetCorruptedBackups(std::vector<BackupID>* corrupt_backup_ids) const;
|
||||
|
||||
Status RestoreDBFromBackup(const RestoreOptions& options, BackupID backup_id,
|
||||
@@ -460,6 +464,10 @@ class BackupEngineImpl {
|
||||
mutable std::shared_ptr<Env> env_for_open_;
|
||||
}; // BackupMeta
|
||||
|
||||
void SetBackupInfoFromBackupMeta(BackupID id, const BackupMeta& meta,
|
||||
BackupInfo* backup_info,
|
||||
bool include_file_details) const;
|
||||
|
||||
inline std::string GetAbsolutePath(
|
||||
const std::string &relative_path = "") const {
|
||||
assert(relative_path.size() == 0 || relative_path[0] != '/');
|
||||
@@ -802,9 +810,11 @@ class BackupEngineImplThreadSafe : public BackupEngine,
|
||||
|
||||
using BackupEngine::CreateNewBackupWithMetadata;
|
||||
Status CreateNewBackupWithMetadata(const CreateBackupOptions& options, DB* db,
|
||||
const std::string& app_metadata) override {
|
||||
const std::string& app_metadata,
|
||||
BackupID* new_backup_id) override {
|
||||
WriteLock lock(&mutex_);
|
||||
return impl_.CreateNewBackupWithMetadata(options, db, app_metadata);
|
||||
return impl_.CreateNewBackupWithMetadata(options, db, app_metadata,
|
||||
new_backup_id);
|
||||
}
|
||||
|
||||
Status PurgeOldBackups(uint32_t num_backups_to_keep) override {
|
||||
@@ -827,6 +837,19 @@ class BackupEngineImplThreadSafe : public BackupEngine,
|
||||
return impl_.GarbageCollect();
|
||||
}
|
||||
|
||||
Status GetLatestBackupInfo(BackupInfo* backup_info,
|
||||
bool include_file_details = false) const override {
|
||||
ReadLock lock(&mutex_);
|
||||
return impl_.GetBackupInfo(kLatestBackupIDMarker, backup_info,
|
||||
include_file_details);
|
||||
}
|
||||
|
||||
Status GetBackupInfo(BackupID backup_id, BackupInfo* backup_info,
|
||||
bool include_file_details = false) const override {
|
||||
ReadLock lock(&mutex_);
|
||||
return impl_.GetBackupInfo(backup_id, backup_info, include_file_details);
|
||||
}
|
||||
|
||||
void GetBackupInfo(std::vector<BackupInfo>* backup_info,
|
||||
bool include_file_details) const override {
|
||||
ReadLock lock(&mutex_);
|
||||
@@ -1181,8 +1204,8 @@ Status BackupEngineImpl::Initialize() {
|
||||
}
|
||||
|
||||
Status BackupEngineImpl::CreateNewBackupWithMetadata(
|
||||
const CreateBackupOptions& options, DB* db,
|
||||
const std::string& app_metadata) {
|
||||
const CreateBackupOptions& options, DB* db, const std::string& app_metadata,
|
||||
BackupID* new_backup_id_ptr) {
|
||||
assert(initialized_);
|
||||
assert(!read_only_);
|
||||
if (app_metadata.size() > kMaxAppMetaSize) {
|
||||
@@ -1417,6 +1440,9 @@ Status BackupEngineImpl::CreateNewBackupWithMetadata(
|
||||
// in the LATEST_BACKUP file
|
||||
latest_backup_id_ = new_backup_id;
|
||||
latest_valid_backup_id_ = new_backup_id;
|
||||
if (new_backup_id_ptr) {
|
||||
*new_backup_id_ptr = new_backup_id;
|
||||
}
|
||||
ROCKS_LOG_INFO(options_.info_log, "Backup DONE. All is good");
|
||||
|
||||
// backup_speed is in byte/second
|
||||
@@ -1546,30 +1572,61 @@ Status BackupEngineImpl::DeleteBackupNoGC(BackupID backup_id) {
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
void BackupEngineImpl::SetBackupInfoFromBackupMeta(
|
||||
BackupID id, const BackupMeta& meta, BackupInfo* backup_info,
|
||||
bool include_file_details) const {
|
||||
*backup_info = BackupInfo(id, meta.GetTimestamp(), meta.GetSize(),
|
||||
meta.GetNumberFiles(), meta.GetAppMetadata());
|
||||
if (include_file_details) {
|
||||
auto& file_details = backup_info->file_details;
|
||||
file_details.reserve(meta.GetFiles().size());
|
||||
for (auto& file_ptr : meta.GetFiles()) {
|
||||
BackupFileInfo& finfo = *file_details.emplace(file_details.end());
|
||||
finfo.relative_filename = file_ptr->filename;
|
||||
finfo.size = file_ptr->size;
|
||||
}
|
||||
backup_info->name_for_open = GetAbsolutePath(GetPrivateFileRel(id));
|
||||
backup_info->name_for_open.pop_back(); // remove trailing '/'
|
||||
backup_info->env_for_open = meta.GetEnvForOpen();
|
||||
}
|
||||
}
|
||||
|
||||
Status BackupEngineImpl::GetBackupInfo(BackupID backup_id,
|
||||
BackupInfo* backup_info,
|
||||
bool include_file_details) const {
|
||||
assert(initialized_);
|
||||
if (backup_id == kLatestBackupIDMarker) {
|
||||
// Note: Read latest_valid_backup_id_ inside of lock
|
||||
backup_id = latest_valid_backup_id_;
|
||||
}
|
||||
auto corrupt_itr = corrupt_backups_.find(backup_id);
|
||||
if (corrupt_itr != corrupt_backups_.end()) {
|
||||
return Status::Corruption(corrupt_itr->second.first.ToString());
|
||||
}
|
||||
auto backup_itr = backups_.find(backup_id);
|
||||
if (backup_itr == backups_.end()) {
|
||||
return Status::NotFound("Backup not found");
|
||||
}
|
||||
auto& backup = backup_itr->second;
|
||||
if (backup->Empty()) {
|
||||
return Status::NotFound("Backup not found");
|
||||
}
|
||||
|
||||
SetBackupInfoFromBackupMeta(backup_id, *backup, backup_info,
|
||||
include_file_details);
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
void BackupEngineImpl::GetBackupInfo(std::vector<BackupInfo>* backup_info,
|
||||
bool include_file_details) const {
|
||||
assert(initialized_);
|
||||
backup_info->reserve(backups_.size());
|
||||
backup_info->resize(backups_.size());
|
||||
size_t i = 0;
|
||||
for (auto& backup : backups_) {
|
||||
const BackupMeta& meta = *backup.second;
|
||||
if (!meta.Empty()) {
|
||||
backup_info->push_back(BackupInfo(backup.first, meta.GetTimestamp(),
|
||||
meta.GetSize(), meta.GetNumberFiles(),
|
||||
meta.GetAppMetadata()));
|
||||
BackupInfo& binfo = backup_info->back();
|
||||
if (include_file_details) {
|
||||
auto& file_details = binfo.file_details;
|
||||
file_details.reserve(meta.GetFiles().size());
|
||||
for (auto& file_ptr : meta.GetFiles()) {
|
||||
BackupFileInfo& finfo = *file_details.emplace(file_details.end());
|
||||
finfo.relative_filename = file_ptr->filename;
|
||||
finfo.size = file_ptr->size;
|
||||
}
|
||||
binfo.name_for_open =
|
||||
GetAbsolutePath(GetPrivateFileRel(binfo.backup_id));
|
||||
binfo.name_for_open.pop_back(); // remove trailing '/'
|
||||
binfo.env_for_open = meta.GetEnvForOpen();
|
||||
}
|
||||
SetBackupInfoFromBackupMeta(backup.first, meta, &backup_info->at(i++),
|
||||
include_file_details);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2168,6 +2168,8 @@ TEST_F(BackupableDBTest, TableFileCorruptionBeforeIncremental) {
|
||||
// between incremental backups
|
||||
TEST_F(BackupableDBTest, FileSizeForIncremental) {
|
||||
const auto share_no_checksum = static_cast<ShareFilesNaming>(0);
|
||||
// TODO: enable blob files once Integrated BlobDB supports DB session id.
|
||||
options_.enable_blob_files = false;
|
||||
|
||||
for (ShareFilesNaming option : {share_no_checksum, kLegacyCrc32cAndFileSize,
|
||||
kNamingDefault, kUseDbSessionId}) {
|
||||
@@ -2192,9 +2194,9 @@ TEST_F(BackupableDBTest, FileSizeForIncremental) {
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true /*flush*/));
|
||||
CloseDBAndBackupEngine();
|
||||
|
||||
// Corrupt backup SST and blob file
|
||||
// Corrupt backup SST file
|
||||
ASSERT_OK(file_manager_->GetChildrenFileAttributes(shared_dir, &children));
|
||||
ASSERT_EQ(children.size(), 2U); // one sst and one blob file
|
||||
ASSERT_EQ(children.size(), 1U); // one sst
|
||||
for (const auto& child : children) {
|
||||
if (child.name.size() > 4 && child.size_bytes > 0) {
|
||||
ASSERT_OK(
|
||||
@@ -2251,10 +2253,10 @@ TEST_F(BackupableDBTest, FileSizeForIncremental) {
|
||||
OpenDBAndBackupEngine(false, false, share);
|
||||
ASSERT_OK(db_->Put(WriteOptions(), "y", Random(42).RandomString(500)));
|
||||
|
||||
// Count backup SSTs and blob files.
|
||||
// Count backup SSTs files.
|
||||
children.clear();
|
||||
ASSERT_OK(file_manager_->GetChildrenFileAttributes(shared_dir, &children));
|
||||
ASSERT_EQ(children.size(), 4U); // two sst and two blob files
|
||||
ASSERT_EQ(children.size(), 2U); // two sst files
|
||||
|
||||
// Try create backup 3
|
||||
s = backup_engine_->CreateNewBackup(db_.get(), true /*flush*/);
|
||||
@@ -2267,18 +2269,18 @@ TEST_F(BackupableDBTest, FileSizeForIncremental) {
|
||||
// Acceptable to call it corruption if size is not in name and
|
||||
// db session id collision is practically impossible.
|
||||
EXPECT_TRUE(s.IsCorruption());
|
||||
EXPECT_EQ(children.size(), 4U); // no SST/Blob file added
|
||||
EXPECT_EQ(children.size(), 2U); // no SST file added
|
||||
} else if (option == share_no_checksum) {
|
||||
// Good to call it corruption if both backups cannot be
|
||||
// accommodated.
|
||||
EXPECT_TRUE(s.IsCorruption());
|
||||
EXPECT_EQ(children.size(), 4U); // no SST/Blob file added
|
||||
EXPECT_EQ(children.size(), 2U); // no SST file added
|
||||
} else {
|
||||
// Since opening a DB seems sufficient for detecting size corruption
|
||||
// on the DB side, this should be a good thing, ...
|
||||
EXPECT_OK(s);
|
||||
// ... as long as we did actually treat it as a distinct SST file.
|
||||
EXPECT_EQ(children.size(), 6U); // Another SST and blob added
|
||||
EXPECT_EQ(children.size(), 3U); // Another SST added
|
||||
}
|
||||
CloseDBAndBackupEngine();
|
||||
ASSERT_OK(DestroyDB(dbname_, options_));
|
||||
@@ -2646,17 +2648,17 @@ TEST_F(BackupableDBTest, OpenBackupAsReadOnlyDB) {
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), /*flush*/ false));
|
||||
db_.reset(); // CloseDB
|
||||
DestroyDB(dbname_, options_);
|
||||
std::vector<BackupInfo> backup_info;
|
||||
BackupInfo backup_info;
|
||||
// First, check that we get empty fields without include_file_details
|
||||
backup_engine_->GetBackupInfo(&backup_info, /*with file details*/ false);
|
||||
ASSERT_EQ(backup_info.size(), 2);
|
||||
ASSERT_EQ(backup_info[0].name_for_open, "");
|
||||
ASSERT_FALSE(backup_info[0].env_for_open);
|
||||
ASSERT_OK(backup_engine_->GetBackupInfo(/*id*/ 1U, &backup_info,
|
||||
/*with file details*/ false));
|
||||
ASSERT_EQ(backup_info.name_for_open, "");
|
||||
ASSERT_FALSE(backup_info.env_for_open);
|
||||
|
||||
// Now for the real test
|
||||
backup_info.clear();
|
||||
backup_engine_->GetBackupInfo(&backup_info, /*with file details*/ true);
|
||||
ASSERT_EQ(backup_info.size(), 2);
|
||||
backup_info = BackupInfo();
|
||||
ASSERT_OK(backup_engine_->GetBackupInfo(/*id*/ 1U, &backup_info,
|
||||
/*with file details*/ true));
|
||||
|
||||
// Caution: DBOptions only holds a raw pointer to Env, so something else
|
||||
// must keep it alive.
|
||||
@@ -2668,9 +2670,9 @@ TEST_F(BackupableDBTest, OpenBackupAsReadOnlyDB) {
|
||||
opts.create_if_missing = false;
|
||||
opts.info_log.reset();
|
||||
|
||||
opts.env = backup_info[0].env_for_open.get();
|
||||
std::string name = backup_info[0].name_for_open;
|
||||
backup_info.clear();
|
||||
opts.env = backup_info.env_for_open.get();
|
||||
std::string name = backup_info.name_for_open;
|
||||
backup_info = BackupInfo();
|
||||
ASSERT_OK(DB::OpenForReadOnly(opts, name, &db));
|
||||
|
||||
AssertExists(db, 0, 100);
|
||||
@@ -2680,13 +2682,13 @@ TEST_F(BackupableDBTest, OpenBackupAsReadOnlyDB) {
|
||||
db = nullptr;
|
||||
|
||||
// Case 2: Keeping BackupInfo alive rather than BackupEngine also suffices
|
||||
backup_engine_->GetBackupInfo(&backup_info, /*with file details*/ true);
|
||||
ASSERT_EQ(backup_info.size(), 2);
|
||||
ASSERT_OK(backup_engine_->GetBackupInfo(/*id*/ 2U, &backup_info,
|
||||
/*with file details*/ true));
|
||||
CloseBackupEngine();
|
||||
opts.create_if_missing = true; // check also OK (though pointless)
|
||||
opts.env = backup_info[1].env_for_open.get();
|
||||
name = backup_info[1].name_for_open;
|
||||
// Note: keeping backup_info[1] alive
|
||||
opts.env = backup_info.env_for_open.get();
|
||||
name = backup_info.name_for_open;
|
||||
// Note: keeping backup_info alive
|
||||
ASSERT_OK(DB::OpenForReadOnly(opts, name, &db));
|
||||
|
||||
AssertExists(db, 0, 200);
|
||||
@@ -2848,17 +2850,32 @@ TEST_F(BackupableDBTest, BackupWithMetadata) {
|
||||
for (int i = 0; i < 5; ++i) {
|
||||
const std::string metadata = std::to_string(i);
|
||||
FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
|
||||
ASSERT_OK(
|
||||
backup_engine_->CreateNewBackupWithMetadata(db_.get(), metadata, true));
|
||||
// Here also test CreateNewBackupWithMetadata with CreateBackupOptions
|
||||
// and outputting saved BackupID.
|
||||
CreateBackupOptions opts;
|
||||
opts.flush_before_backup = true;
|
||||
BackupID new_id = 0;
|
||||
ASSERT_OK(backup_engine_->CreateNewBackupWithMetadata(opts, db_.get(),
|
||||
metadata, &new_id));
|
||||
ASSERT_EQ(new_id, static_cast<BackupID>(i + 1));
|
||||
}
|
||||
CloseDBAndBackupEngine();
|
||||
|
||||
OpenDBAndBackupEngine();
|
||||
std::vector<BackupInfo> backup_infos;
|
||||
backup_engine_->GetBackupInfo(&backup_infos);
|
||||
ASSERT_EQ(5, backup_infos.size());
|
||||
{ // Verify in bulk BackupInfo
|
||||
std::vector<BackupInfo> backup_infos;
|
||||
backup_engine_->GetBackupInfo(&backup_infos);
|
||||
ASSERT_EQ(5, backup_infos.size());
|
||||
for (int i = 0; i < 5; i++) {
|
||||
ASSERT_EQ(std::to_string(i), backup_infos[i].app_metadata);
|
||||
}
|
||||
}
|
||||
// Also verify in individual BackupInfo
|
||||
for (int i = 0; i < 5; i++) {
|
||||
ASSERT_EQ(std::to_string(i), backup_infos[i].app_metadata);
|
||||
BackupInfo backup_info;
|
||||
ASSERT_OK(backup_engine_->GetBackupInfo(static_cast<BackupID>(i + 1),
|
||||
&backup_info));
|
||||
ASSERT_EQ(std::to_string(i), backup_info.app_metadata);
|
||||
}
|
||||
CloseDBAndBackupEngine();
|
||||
DestroyDB(dbname_, options_);
|
||||
@@ -3285,6 +3302,8 @@ TEST_F(BackupableDBTest, CreateWhenLatestBackupCorrupted) {
|
||||
// succeed even when latest backup is corrupted.
|
||||
const int kNumKeys = 5000;
|
||||
OpenDBAndBackupEngine(true /* destroy_old_data */);
|
||||
BackupInfo backup_info;
|
||||
ASSERT_TRUE(backup_engine_->GetLatestBackupInfo(&backup_info).IsNotFound());
|
||||
FillDB(db_.get(), 0 /* from */, kNumKeys);
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(),
|
||||
true /* flush_before_backup */));
|
||||
@@ -3293,12 +3312,26 @@ TEST_F(BackupableDBTest, CreateWhenLatestBackupCorrupted) {
|
||||
CloseDBAndBackupEngine();
|
||||
|
||||
OpenDBAndBackupEngine();
|
||||
ASSERT_TRUE(backup_engine_->GetLatestBackupInfo(&backup_info).IsNotFound());
|
||||
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(),
|
||||
true /* flush_before_backup */));
|
||||
|
||||
ASSERT_TRUE(backup_engine_->GetLatestBackupInfo(&backup_info).ok());
|
||||
ASSERT_EQ(2, backup_info.backup_id);
|
||||
|
||||
std::vector<BackupInfo> backup_infos;
|
||||
backup_engine_->GetBackupInfo(&backup_infos);
|
||||
ASSERT_EQ(1, backup_infos.size());
|
||||
ASSERT_EQ(2, backup_infos[0].backup_id);
|
||||
|
||||
// Verify individual GetBackupInfo by ID
|
||||
ASSERT_TRUE(backup_engine_->GetBackupInfo(0U, &backup_info).IsNotFound());
|
||||
ASSERT_TRUE(backup_engine_->GetBackupInfo(1U, &backup_info).IsCorruption());
|
||||
ASSERT_TRUE(backup_engine_->GetBackupInfo(2U, &backup_info).ok());
|
||||
ASSERT_TRUE(backup_engine_->GetBackupInfo(3U, &backup_info).IsNotFound());
|
||||
ASSERT_TRUE(
|
||||
backup_engine_->GetBackupInfo(999999U, &backup_info).IsNotFound());
|
||||
}
|
||||
|
||||
TEST_F(BackupableDBTest, WriteOnlyEngineNoSharedFileDeletion) {
|
||||
@@ -3456,7 +3489,11 @@ TEST_F(BackupableDBTest, BackgroundThreadCpuPriority) {
|
||||
CreateBackupOptions options;
|
||||
options.decrease_background_thread_cpu_priority = true;
|
||||
options.background_thread_cpu_priority = CpuPriority::kIdle;
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(options, db_.get()));
|
||||
|
||||
// Also check output backup_id with CreateNewBackup
|
||||
BackupID new_id = 0;
|
||||
ASSERT_OK(backup_engine_->CreateNewBackup(options, db_.get(), &new_id));
|
||||
ASSERT_EQ(new_id, 5U);
|
||||
|
||||
ASSERT_EQ(priority, CpuPriority::kNormal);
|
||||
}
|
||||
|
||||
@@ -54,7 +54,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate1) {
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
old_options.level_compaction_dynamic_level_bytes = is_dynamic1_;
|
||||
}
|
||||
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
old_options.max_open_files = -1;
|
||||
}
|
||||
old_options.level0_file_num_compaction_trigger = 3;
|
||||
old_options.write_buffer_size = 64 * 1024;
|
||||
old_options.target_file_size_base = 128 * 1024;
|
||||
@@ -92,6 +94,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate1) {
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
new_options.level_compaction_dynamic_level_bytes = is_dynamic2_;
|
||||
}
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
new_options.max_open_files = -1;
|
||||
}
|
||||
new_options.target_file_size_base = 256 * 1024;
|
||||
new_options.num_levels = level2_;
|
||||
new_options.max_bytes_for_level_base = 150 * 1024;
|
||||
@@ -123,6 +128,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate2) {
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
old_options.level_compaction_dynamic_level_bytes = is_dynamic2_;
|
||||
}
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
old_options.max_open_files = -1;
|
||||
}
|
||||
old_options.level0_file_num_compaction_trigger = 3;
|
||||
old_options.write_buffer_size = 64 * 1024;
|
||||
old_options.target_file_size_base = 128 * 1024;
|
||||
@@ -161,6 +169,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate2) {
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
new_options.level_compaction_dynamic_level_bytes = is_dynamic1_;
|
||||
}
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
new_options.max_open_files = -1;
|
||||
}
|
||||
new_options.target_file_size_base = 256 * 1024;
|
||||
new_options.num_levels = level1_;
|
||||
new_options.max_bytes_for_level_base = 150 * 1024;
|
||||
@@ -191,7 +202,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate3) {
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
old_options.level_compaction_dynamic_level_bytes = is_dynamic1_;
|
||||
}
|
||||
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
old_options.max_open_files = -1;
|
||||
}
|
||||
old_options.level0_file_num_compaction_trigger = 3;
|
||||
old_options.write_buffer_size = 64 * 1024;
|
||||
old_options.target_file_size_base = 128 * 1024;
|
||||
@@ -235,6 +248,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate3) {
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
new_options.level_compaction_dynamic_level_bytes = is_dynamic2_;
|
||||
}
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
new_options.max_open_files = -1;
|
||||
}
|
||||
new_options.target_file_size_base = 256 * 1024;
|
||||
new_options.num_levels = level2_;
|
||||
new_options.max_bytes_for_level_base = 150 * 1024;
|
||||
@@ -266,6 +282,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate4) {
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
old_options.level_compaction_dynamic_level_bytes = is_dynamic2_;
|
||||
}
|
||||
if (old_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
old_options.max_open_files = -1;
|
||||
}
|
||||
old_options.level0_file_num_compaction_trigger = 3;
|
||||
old_options.write_buffer_size = 64 * 1024;
|
||||
old_options.target_file_size_base = 128 * 1024;
|
||||
@@ -310,6 +329,9 @@ TEST_P(DBOptionChangeMigrationTests, Migrate4) {
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleLevel) {
|
||||
new_options.level_compaction_dynamic_level_bytes = is_dynamic1_;
|
||||
}
|
||||
if (new_options.compaction_style == CompactionStyle::kCompactionStyleFIFO) {
|
||||
new_options.max_open_files = -1;
|
||||
}
|
||||
new_options.target_file_size_base = 256 * 1024;
|
||||
new_options.num_levels = level1_;
|
||||
new_options.max_bytes_for_level_base = 150 * 1024;
|
||||
|
||||
Reference in New Issue
Block a user