mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 19a1ebaa6d | |||
| 67dbbd79b9 | |||
| 8053b9414f | |||
| 5177a8e6c2 | |||
| 3883a8d05e | |||
| 768de6e50d | |||
| 62f05627be | |||
| 9ef369c606 | |||
| 30dba7f41a | |||
| 023fbb074a |
@@ -3,7 +3,7 @@ inputs:
|
||||
suite-run:
|
||||
description: Comma-separated list of test suites to run (empty to skip C++ tests)
|
||||
required: false
|
||||
default: arena_test,db_basic_test,db_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
|
||||
default: arena_test,db_basic_test,db_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
|
||||
run-java:
|
||||
description: Whether to run Java tests
|
||||
required: false
|
||||
|
||||
@@ -537,7 +537,7 @@ jobs:
|
||||
suite_run: db_test
|
||||
run_java: "false"
|
||||
- test_shard: other
|
||||
suite_run: arena_test,db_basic_test,db_test2,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
|
||||
suite_run: arena_test,db_basic_test,db_etc2_test,db_merge_operand_test,bloom_test,c_test,coding_test,crc32c_test,dynamic_bloom_test,env_basic_test,env_test,hash_test,random_test
|
||||
run_java: "false"
|
||||
- test_shard: java
|
||||
suite_run: ""
|
||||
|
||||
+1
-1
@@ -52,7 +52,7 @@ GRTAGS
|
||||
GTAGS
|
||||
rocksdb_dump
|
||||
rocksdb_undump
|
||||
db_test2
|
||||
db_etc2_test
|
||||
trace_analyzer
|
||||
block_cache_trace_analyzer
|
||||
io_tracer_parser
|
||||
|
||||
@@ -4868,6 +4868,12 @@ cpp_unittest_wrapper(name="db_encryption_test",
|
||||
extra_compiler_flags=[])
|
||||
|
||||
|
||||
cpp_unittest_wrapper(name="db_etc2_test",
|
||||
srcs=["db/db_etc2_test.cc"],
|
||||
deps=[":rocksdb_test_lib"],
|
||||
extra_compiler_flags=[])
|
||||
|
||||
|
||||
cpp_unittest_wrapper(name="db_etc3_test",
|
||||
srcs=["db/db_etc3_test.cc"],
|
||||
deps=[":rocksdb_test_lib"],
|
||||
@@ -5024,12 +5030,6 @@ cpp_unittest_wrapper(name="db_test",
|
||||
extra_compiler_flags=[])
|
||||
|
||||
|
||||
cpp_unittest_wrapper(name="db_test2",
|
||||
srcs=["db/db_test2.cc"],
|
||||
deps=[":rocksdb_test_lib"],
|
||||
extra_compiler_flags=[])
|
||||
|
||||
|
||||
cpp_unittest_wrapper(name="db_universal_compaction_test",
|
||||
srcs=["db/db_universal_compaction_test.cc"],
|
||||
deps=[":rocksdb_test_lib"],
|
||||
|
||||
@@ -209,12 +209,11 @@ The following patterns emerged as frequent sources of review feedback:
|
||||
## Important tips
|
||||
|
||||
### Build system
|
||||
* There are 3 build system. Make, CMake, BUCK(meta internal).
|
||||
* There are 3 build system. Make for git clones, BUCK (meta internal) for hg
|
||||
clones, and CMake for some special cases.
|
||||
* When a new .cc file is added, update Makefile, CMakeLists.txt, src.mk, BUCK.
|
||||
* Don't manually edit BUCK file, after updating src.mk, run
|
||||
/usr/local/bin/python3 buckifier/buckify_rocksdb.py to update it
|
||||
* Use make to build and run the test. CMake and BUCK are not used locally.
|
||||
* Use `make dbg` command to build all of the unit test in debug mode.
|
||||
* For -j in make command, use the number of CPU cores to decide it.
|
||||
|
||||
### When to run `make clean` (avoid mixing build modes)
|
||||
|
||||
+1
-1
@@ -1491,6 +1491,7 @@ if(WITH_TESTS)
|
||||
db/db_clip_test.cc
|
||||
db/db_dynamic_level_test.cc
|
||||
db/db_encryption_test.cc
|
||||
db/db_etc2_test.cc
|
||||
db/db_etc3_test.cc
|
||||
db/db_flush_test.cc
|
||||
db/db_inplace_update_test.cc
|
||||
@@ -1514,7 +1515,6 @@ if(WITH_TESTS)
|
||||
db/db_table_properties_test.cc
|
||||
db/db_tailing_iter_test.cc
|
||||
db/db_test.cc
|
||||
db/db_test2.cc
|
||||
db/db_logical_block_size_cache_test.cc
|
||||
db/db_universal_compaction_test.cc
|
||||
db/db_wal_test.cc
|
||||
|
||||
+22
@@ -1,6 +1,28 @@
|
||||
# Rocksdb Change Log
|
||||
> NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
|
||||
|
||||
## 11.4.0 (06/02/2026)
|
||||
### Public API Changes
|
||||
* Added `rocksdb_options_set_memtable_batch_lookup_optimization()` and `rocksdb_options_get_memtable_batch_lookup_optimization()` to the C API, exposing the existing `AdvancedColumnFamilyOptions::memtable_batch_lookup_optimization` field. This allows C API users (and downstream language bindings) to enable the skip-list memtable's batch-lookup optimization for `MultiGet`, which caches the search path between consecutive keys and reduces per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys.
|
||||
* Added `rocksdb_readoptions_set_optimize_multiget_for_io()` and `rocksdb_readoptions_get_optimize_multiget_for_io()` to the C API, exposing the existing `ReadOptions::optimize_multiget_for_io` field. This allows C API users (and downstream language bindings) to opt out of the multi-level parallel MultiGet path, which only takes effect when the library is built with `USE_COROUTINES`.
|
||||
* Added `rocksdb_block_based_table_index_block_search_type_auto` enum constant and the `rocksdb_block_based_options_set_uniform_cv_threshold()` setter to the C API. The constant exposes `BlockSearchType::kAuto`, and the setter exposes `BlockBasedTableOptions::uniform_cv_threshold`. Both are required for `kAuto` index-block search to take effect from the C API: without setting `uniform_cv_threshold >= 0`, the per-block `is_uniform` footer bit is never written, and `kAuto` falls back to binary search at read time.
|
||||
* Added `EventListener::OnDBShutdownBegin`, a callback that fires once when a DB begins shutdown, including when RocksDB cleans up a failed `DB::Open()` attempt.
|
||||
* Added a new PerfContext counter `blob_cache_read_byte` for blob cache bytes read.
|
||||
|
||||
### Behavior Changes
|
||||
* Remote compaction now falls back to local compaction when the primary cannot deserialize a successful `CompactionServiceResult` before installing any remote output files.
|
||||
* StringToMap (rocksdb/convenience.h) now preserves the outer braces of nested values so each map entry is in self-contained form and can be embedded directly into another `key=value;` string (e.g. SetOptions) without losing inner ';' delimiters. A new symmetric MapToString utility is provided.
|
||||
|
||||
### Bug Fixes
|
||||
* Fixed a bug where `AbortAllCompactions()` could leave automatic compaction work unscheduled when aborting before the background worker picked it, causing compaction and write stalls until DB restart.
|
||||
* Fix a bug where db had a false positive compaction corruption error due to remote compaction service result with `has_accurate_num_input_records=false` not being serialized.
|
||||
* Fixed WritePrepared TransactionDB cleanup after retryable commit or rollback write failures so prepared transactions remain rollbackable instead of leaving unresolved prepared state.
|
||||
* Fix a rare corruption bug for tiered compaction that incorrectly moved last level range tombstones into proximal level. This corruption error is surfaced when force_consistency_checks is enabled.
|
||||
|
||||
### Performance Improvements
|
||||
* Reduce the likelihood of user-facing metadata operations blocking on MANIFEST rotation when file creation is slow, such as on remote storage. MANIFEST write batches containing foreground edits from `CreateColumnFamily()`, `DropColumnFamily()`, `CreateColumnFamilyWithImport()`, `IngestExternalFile()`, and `DeleteFilesInRanges()` now get a relaxed file size threshold for triggering MANIFEST rotation; background-only MANIFEST write batches, such as compaction and flush, continue to use the normal threshold. This change should make auto-tuning manifest file size more attractive (see `max_manifest_file_size` and `max_manifest_space_amp_pct` options).
|
||||
|
||||
|
||||
## 11.3.0 (05/15/2026)
|
||||
### New Features
|
||||
* Add experimental DB option `async_wal_precreate` to precreate the next WAL file in a background thread and reduce foreground WAL rotation latency. The option is sanitized to false when WAL recycling is enabled.
|
||||
|
||||
@@ -967,8 +967,6 @@ gen_parallel_tests:
|
||||
# total time across many shards (need early queueing for fan-out).
|
||||
# Each alternative is wrapped in `^.*NAME.*$$` so the *whole input line* is
|
||||
# captured into $$1; then `s,(...),100 $$1,` prepends "100 " to the line.
|
||||
# Use a trailing dash on prefix-of-other-name binaries (e.g. `db_test-` to
|
||||
# avoid matching `db_test2-...`).
|
||||
#
|
||||
# Tiers below are based on observed timings (see suggest-slow-tests).
|
||||
# Tier 1: max single-shard time >= 30s (critical-path bottlenecks).
|
||||
@@ -976,7 +974,7 @@ gen_parallel_tests:
|
||||
# Tier 3: huge total time across many tiny shards; front-loading them keeps
|
||||
# the tail of the run busy while big shards finish.
|
||||
slow_test_regexp = \
|
||||
^.*point_lock_manager_stress_test.*$$|^.*db_test-.*$$|^.*external_sst_file_test.*$$|^.*compaction_service_test.*$$|^.*corruption_test.*$$|^.*comparator_db_test.*$$|^.*external_sst_file_basic_test.*$$|^.*rate_limiter_test.*$$|^.*db_compaction_test.*$$|^.*write_prepared_transaction_test.*$$|^.*db_merge_operator_test.*$$|\
|
||||
^.*point_lock_manager_stress_test.*$$|^.*db_test.*$$|^.*external_sst_file_test.*$$|^.*compaction_service_test.*$$|^.*corruption_test.*$$|^.*comparator_db_test.*$$|^.*external_sst_file_basic_test.*$$|^.*rate_limiter_test.*$$|^.*db_compaction_test.*$$|^.*write_prepared_transaction_test.*$$|^.*db_merge_operator_test.*$$|\
|
||||
^.*db_dynamic_level_test.*$$|^.*db_bloom_filter_test.*$$|^.*error_handler_fs_test.*$$|^.*merge_helper_test.*$$|^.*transaction_test.*$$|^.*db_kv_checksum_test.*$$|^.*inlineskiplist_test.*$$|\
|
||||
^.*db_with_timestamp_basic_test.*$$|^.*table_test.*$$|^.*db_wal_test.*$$|^.*block_based_table_reader_test.*$$|^.*block_test.*$$
|
||||
prioritize_long_running_tests = \
|
||||
@@ -1586,7 +1584,7 @@ db_open_with_config_test: $(OBJ_DIR)/db/db_open_with_config_test.o $(TEST_LIBRAR
|
||||
db_test: $(OBJ_DIR)/db/db_test.o $(TEST_LIBRARY) $(LIBRARY)
|
||||
$(AM_LINK)
|
||||
|
||||
db_test2: $(OBJ_DIR)/db/db_test2.o $(TEST_LIBRARY) $(LIBRARY)
|
||||
db_etc2_test: $(OBJ_DIR)/db/db_etc2_test.o $(TEST_LIBRARY) $(LIBRARY)
|
||||
$(AM_LINK)
|
||||
|
||||
db_etc3_test: $(OBJ_DIR)/db/db_etc3_test.o $(TEST_LIBRARY) $(LIBRARY)
|
||||
|
||||
@@ -651,6 +651,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
|
||||
{"cpu_micros",
|
||||
{offsetof(struct CompactionJobStats, cpu_micros), OptionType::kUInt64T,
|
||||
OptionVerificationType::kNormal, OptionTypeFlags::kNone}},
|
||||
{"has_accurate_num_input_records",
|
||||
{offsetof(struct CompactionJobStats, has_accurate_num_input_records),
|
||||
OptionType::kBoolean, OptionVerificationType::kNormal,
|
||||
OptionTypeFlags::kNone}},
|
||||
{"num_input_records",
|
||||
{offsetof(struct CompactionJobStats, num_input_records),
|
||||
OptionType::kUInt64T, OptionVerificationType::kNormal,
|
||||
|
||||
@@ -943,6 +943,37 @@ TEST_F(CompactionServiceTest, VerifyInputRecordCount) {
|
||||
SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||
}
|
||||
|
||||
TEST_F(CompactionServiceTest, InaccurateInputRecordCount) {
|
||||
Options options = CurrentOptions();
|
||||
options.disable_auto_compactions = true;
|
||||
ReopenWithCompactionService(&options);
|
||||
GenerateTestData();
|
||||
|
||||
auto my_cs = GetCompactionService();
|
||||
|
||||
std::string start_str = Key(15);
|
||||
std::string end_str = Key(45);
|
||||
Slice start(start_str);
|
||||
Slice end(end_str);
|
||||
uint64_t comp_num = my_cs->GetCompactionNum();
|
||||
|
||||
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
|
||||
"CompactionServiceCompactionJob::Run:0", [&](void* arg) {
|
||||
CompactionServiceResult* compaction_result =
|
||||
*(static_cast<CompactionServiceResult**>(arg));
|
||||
ASSERT_TRUE(compaction_result != nullptr);
|
||||
compaction_result->stats.has_accurate_num_input_records = false;
|
||||
compaction_result->stats.num_input_records = 0;
|
||||
});
|
||||
SyncPoint::GetInstance()->EnableProcessing();
|
||||
|
||||
ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &end));
|
||||
ASSERT_GE(my_cs->GetCompactionNum(), comp_num + 1);
|
||||
|
||||
SyncPoint::GetInstance()->DisableProcessing();
|
||||
SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||
}
|
||||
|
||||
TEST_F(CompactionServiceTest, EmptyResult) {
|
||||
Options options = CurrentOptions();
|
||||
options.disable_auto_compactions = true;
|
||||
@@ -1285,6 +1316,10 @@ TEST_F(CompactionServiceTest, TruncatedOutput) {
|
||||
|
||||
TEST_F(CompactionServiceTest, CustomFileChecksum) {
|
||||
Options options = CurrentOptions();
|
||||
// Pin compression so the auto-compacted LSM shape (and thus whether the
|
||||
// manual CompactRange below schedules a remote compaction) doesn't depend on
|
||||
// the default compression type. kNoCompression is always available.
|
||||
options.compression = kNoCompression;
|
||||
options.file_checksum_gen_factory = GetFileChecksumGenCrc32cFactory();
|
||||
ReopenWithCompactionService(&options);
|
||||
GenerateTestData();
|
||||
|
||||
@@ -2022,7 +2022,11 @@ TEST_F(DBBloomFilterTest, MutableFilterPolicy) {
|
||||
double expected_bpk = 10.0;
|
||||
// Other configs to try
|
||||
std::vector<std::pair<std::string, double>> configs = {
|
||||
{"ribbonfilter:10:-1", 7.0}, {"bloomfilter:5", 5.0}, {"nullptr", 0.0}};
|
||||
{"ribbonfilter:10:-1", 7.0},
|
||||
{"bloomfilter:5", 5.0},
|
||||
{"nullptr", 0.0},
|
||||
// As serialized in OPTIONS file
|
||||
{"{id=ribbonfilter:12:-1;bloom_before_level=-1;}", 8.4}};
|
||||
|
||||
table_options.cache_index_and_filter_blocks = true;
|
||||
options.table_factory.reset(NewBlockBasedTableFactory(table_options));
|
||||
|
||||
@@ -427,6 +427,38 @@ TEST_F(DBCompactionAbortTest, AbortAutomaticCompaction) {
|
||||
}
|
||||
}
|
||||
|
||||
TEST_F(DBCompactionAbortTest, AbortScheduledAutomaticCompactionBeforePick) {
|
||||
// The goal is to cover an automatic compaction that has been scheduled, but
|
||||
// aborts before the worker picks and removes a column family from the
|
||||
// compaction queue. This pins the scheduler bookkeeping invariant that
|
||||
// ResumeAllCompactions() must be able to schedule that still-queued work.
|
||||
constexpr int kCompactionTrigger = 4;
|
||||
constexpr int kNumL0Files = kCompactionTrigger + 1;
|
||||
Options options = GetOptionsWithStats();
|
||||
options.level0_file_num_compaction_trigger = kCompactionTrigger;
|
||||
options.max_background_compactions = 1;
|
||||
options.max_subcompactions = 1;
|
||||
options.disable_auto_compactions = false;
|
||||
Reopen(options);
|
||||
|
||||
SyncPointAbortHelper helper("BackgroundCallCompaction:0");
|
||||
helper.Setup(dbfull());
|
||||
|
||||
PopulateData(/*num_files=*/kNumL0Files, /*keys_per_file=*/100,
|
||||
/*value_size=*/1000);
|
||||
helper.CleanupAndWait();
|
||||
|
||||
const uint64_t compact_write_bytes_before_resume =
|
||||
stats_->getTickerCount(COMPACT_WRITE_BYTES);
|
||||
dbfull()->ResumeAllCompactions();
|
||||
ASSERT_OK(dbfull()->TEST_WaitForCompact());
|
||||
|
||||
ASSERT_GT(stats_->getTickerCount(COMPACT_WRITE_BYTES),
|
||||
compact_write_bytes_before_resume);
|
||||
ASSERT_LT(NumTableFilesAtLevel(0), kNumL0Files);
|
||||
VerifyDataIntegrity(/*num_keys=*/100);
|
||||
}
|
||||
|
||||
TEST_F(DBCompactionAbortTest, AbortAndVerifyNoOutputFiles) {
|
||||
Options options = CurrentOptions();
|
||||
options.level0_file_num_compaction_trigger = 4;
|
||||
|
||||
@@ -35,7 +35,7 @@ namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
class DBTest2 : public DBTestBase {
|
||||
public:
|
||||
DBTest2() : DBTestBase("db_test2", /*env_do_fsync=*/true) {}
|
||||
DBTest2() : DBTestBase("db_etc2_test", /*env_do_fsync=*/true) {}
|
||||
};
|
||||
|
||||
TEST_F(DBTest2, OpenForReadOnly) {
|
||||
+214
-50
@@ -4,6 +4,9 @@
|
||||
// (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
#include "db/db_test_util.h"
|
||||
#include "rocksdb/convenience.h"
|
||||
#include "rocksdb/metadata.h"
|
||||
#include "rocksdb/sst_file_writer.h"
|
||||
|
||||
namespace ROCKSDB_NAMESPACE {
|
||||
|
||||
@@ -43,50 +46,98 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
ASSERT_EQ(DBOptions{}.max_manifest_space_amp_pct, 500);
|
||||
|
||||
Options options = CurrentOptions();
|
||||
ASSERT_OK(db_->SetOptions({{"level0_file_num_compaction_trigger", "20"}}));
|
||||
// Test setup: create many flushed files. Keep level compaction semantics so
|
||||
// DeleteFilesInRanges can remove non-L0 files, but prevent automatic
|
||||
// compactions and write stalls from adding unrelated behavior.
|
||||
options.disable_auto_compactions = true;
|
||||
options.level0_slowdown_writes_trigger = 100;
|
||||
options.level0_stop_writes_trigger = 200;
|
||||
|
||||
// Use large column family names to essentially control the amount of payload
|
||||
// data needed for the manifest file. Drop manifest entries don't include the
|
||||
// CF name so are small.
|
||||
// Test strategy: use large column family names to control the rough amount
|
||||
// of payload added to the MANIFEST. Drop manifest entries do not include the
|
||||
// CF name, so they are small.
|
||||
//
|
||||
// Most CF helper calls piggy-back a background manifest write so the main
|
||||
// auto-tuning phases continue to test the unrelaxed background threshold
|
||||
// even though CF manipulation itself is foreground. Phase-specific foreground
|
||||
// checks disable that piggy-backed background write.
|
||||
uint64_t prev_manifest_num = 0, cur_manifest_num = 0;
|
||||
std::deque<ColumnFamilyHandle*> handles;
|
||||
int counter = 5;
|
||||
auto AddCfFn = [&]() {
|
||||
auto UpdateManifestNumsFrom = [&](uint64_t before_manifest_num) {
|
||||
prev_manifest_num = before_manifest_num;
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
};
|
||||
auto BackgroundManifestWriteFn = [&]() {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
ASSERT_OK(Put("x", std::to_string(counter++)));
|
||||
ASSERT_OK(Flush());
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
auto AddCfFn = [&](bool include_background_manifest_write = true) {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
std::string name = "cf" + std::to_string(counter++);
|
||||
name.resize(1000, 'a');
|
||||
ASSERT_OK(db_->CreateColumnFamily(options, name, &handles.emplace_back()));
|
||||
prev_manifest_num = cur_manifest_num;
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
if (include_background_manifest_write) {
|
||||
BackgroundManifestWriteFn();
|
||||
}
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
auto DropCfFn = [&]() {
|
||||
auto DropCfFn = [&](bool include_background_manifest_write = true) {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
ASSERT_OK(db_->DropColumnFamily(handles.front()));
|
||||
ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
|
||||
handles.pop_front();
|
||||
prev_manifest_num = cur_manifest_num;
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
};
|
||||
auto TrivialManifestWriteFn = [&]() {
|
||||
ASSERT_OK(Put("x", std::to_string(counter++)));
|
||||
ASSERT_OK(Flush());
|
||||
prev_manifest_num = cur_manifest_num;
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
if (include_background_manifest_write) {
|
||||
BackgroundManifestWriteFn();
|
||||
}
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
|
||||
// ---- Phase 1: foreground threshold relaxation is bounded ----
|
||||
//
|
||||
// Foreground operations should only get about 25% extra headroom, not an
|
||||
// unbounded threshold. With 1000-byte CF names and a 3000-byte normal limit,
|
||||
// the relaxed limit should allow the first four foreground-only CF additions
|
||||
// but require rotation on the fifth.
|
||||
options.max_manifest_file_size = 3000;
|
||||
options.max_manifest_space_amp_pct = 0;
|
||||
DestroyAndReopen(options);
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
prev_manifest_num = cur_manifest_num;
|
||||
for (int i = 1; i <= 4; ++i) {
|
||||
AddCfFn(/*include_background_manifest_write=*/false);
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
}
|
||||
AddCfFn(/*include_background_manifest_write=*/false);
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
while (!handles.empty()) {
|
||||
ASSERT_OK(db_->DestroyColumnFamilyHandle(handles.front()));
|
||||
handles.pop_front();
|
||||
}
|
||||
|
||||
// ---- Phase 2: no auto-tuning means frequent rotation ----
|
||||
//
|
||||
options.max_manifest_file_size = 1000000;
|
||||
options.max_manifest_space_amp_pct = 0; // no auto-tuning yet
|
||||
DestroyAndReopen(options);
|
||||
cur_manifest_num = dbfull()->TEST_Current_Manifest_FileNo();
|
||||
prev_manifest_num = cur_manifest_num;
|
||||
|
||||
// With the generous (minimum) maximum manifest size, should not be rotated
|
||||
// With the generous minimum manifest size, should not be rotated.
|
||||
AddCfFn();
|
||||
AddCfFn();
|
||||
AddCfFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// Change options for small max and (still) no auto-tuning
|
||||
// Lower the minimum while still disabling auto-tuning.
|
||||
ASSERT_OK(db_->SetDBOptions({{"max_manifest_file_size", "3000"}}));
|
||||
|
||||
// Takes effect on the next manifest write
|
||||
TrivialManifestWriteFn();
|
||||
BackgroundManifestWriteFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// Now we have to rewrite the whole manifest on each write because the
|
||||
@@ -97,28 +148,31 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
AddCfFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
TrivialManifestWriteFn();
|
||||
BackgroundManifestWriteFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// ---- Phase 3: auto-tuning raises the background threshold ----
|
||||
//
|
||||
// Enabling auto-tuning should fix this, immediately for next manifest writes.
|
||||
// This will allow up to double-ish the size of the compacted manifest,
|
||||
// which last should have been 4000 + some bytes.
|
||||
// This will allow up to roughly double the size of the compacted manifest,
|
||||
// which now includes CF entries plus the piggy-backed background writes.
|
||||
ASSERT_EQ(handles.size(), 4U);
|
||||
ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "105"}}));
|
||||
ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "95"}}));
|
||||
|
||||
// After 9 CF names should be enough to rotate the manifest
|
||||
for (int i = 1; i <= 5; ++i) {
|
||||
// Auto-tuning lets several more CF names accumulate in the MANIFEST before
|
||||
// the piggy-backed background write crosses the threshold.
|
||||
for (int i = 1; i <= 3; ++i) {
|
||||
if ((i % 2) == 1) {
|
||||
DropCfFn();
|
||||
}
|
||||
AddCfFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
}
|
||||
TrivialManifestWriteFn();
|
||||
AddCfFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// We now have a different last compacted manifest size, should be
|
||||
// able to go beyond 9 CFs named in manifest this time.
|
||||
// We now have a different last compacted manifest size, so the next
|
||||
// threshold should be based on the newly compacted MANIFEST.
|
||||
ASSERT_EQ(handles.size(), 6U);
|
||||
|
||||
DropCfFn();
|
||||
@@ -128,11 +182,10 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
AddCfFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
}
|
||||
// We've written 10 named CFs to the manifest. We should be able to
|
||||
// dynamically change the auto-tuning still based on the last "compacted"
|
||||
// manifest size of 7000 + some bytes.
|
||||
// We should be able to dynamically change the auto-tuning still based on
|
||||
// the last "compacted" manifest size.
|
||||
ASSERT_OK(db_->SetDBOptions({{"max_manifest_space_amp_pct", "51"}}));
|
||||
TrivialManifestWriteFn();
|
||||
BackgroundManifestWriteFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
// And the "compacted" manifest size has reset again, so should be changed
|
||||
// again sooner.
|
||||
@@ -141,16 +194,129 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
AddCfFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
}
|
||||
// Enough for manifest change
|
||||
AddCfFn();
|
||||
|
||||
// ---- Phase 4: foreground operations use relaxed threshold ----
|
||||
//
|
||||
// The current MANIFEST is now large enough for the next background manifest
|
||||
// write to rotate it, but still small enough for foreground operations to use
|
||||
// their 25% extra headroom. Assert that each foreground operation stays on
|
||||
// the same MANIFEST, then verify the next background write rotates.
|
||||
const std::string external_files_dir = dbname_ + "/external_files";
|
||||
ASSERT_OK(env_->CreateDirIfMissing(external_files_dir));
|
||||
auto WriteExternalSstFile = [&](const std::string& file_name,
|
||||
const std::string& key,
|
||||
const std::string& value) {
|
||||
const std::string file_path = external_files_dir + "/" + file_name;
|
||||
Status ignored = env_->DeleteFile(file_path);
|
||||
ignored.PermitUncheckedError();
|
||||
|
||||
SstFileWriter sst_file_writer(EnvOptions(), options);
|
||||
ASSERT_OK(sst_file_writer.Open(file_path));
|
||||
ASSERT_OK(sst_file_writer.Put(key, value));
|
||||
ASSERT_OK(sst_file_writer.Finish());
|
||||
};
|
||||
auto IngestExternalFileForegroundManifestWriteFn =
|
||||
[&](std::string* ingested_key) {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
const std::string key = "z" + std::to_string(counter++);
|
||||
const std::string value = "v" + std::to_string(counter++);
|
||||
const std::string file_name =
|
||||
"ingest_" + std::to_string(counter++) + ".sst";
|
||||
const std::string file_path = external_files_dir + "/" + file_name;
|
||||
WriteExternalSstFile(file_name, key, value);
|
||||
|
||||
ASSERT_OK(
|
||||
db_->IngestExternalFile({file_path}, IngestExternalFileOptions()));
|
||||
ASSERT_EQ(value, Get(key));
|
||||
*ingested_key = key;
|
||||
Status ignored = env_->DeleteFile(file_path);
|
||||
ignored.PermitUncheckedError();
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
auto CreateColumnFamilyWithImportForegroundManifestWriteFn = [&]() {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
const std::string cf_name = "import_cf" + std::to_string(counter++);
|
||||
const std::string key = "import" + std::to_string(counter++);
|
||||
const std::string value = "v" + std::to_string(counter++);
|
||||
const std::string file_name =
|
||||
"import_" + std::to_string(counter++) + ".sst";
|
||||
const std::string file_path = external_files_dir + "/" + file_name;
|
||||
WriteExternalSstFile(file_name, key, value);
|
||||
|
||||
LiveFileMetaData file_metadata;
|
||||
file_metadata.name = file_name;
|
||||
file_metadata.db_path = external_files_dir;
|
||||
file_metadata.smallest_seqno = 0;
|
||||
file_metadata.largest_seqno = 0;
|
||||
file_metadata.level = 0;
|
||||
ExportImportFilesMetaData metadata;
|
||||
metadata.files.push_back(file_metadata);
|
||||
metadata.db_comparator_name = options.comparator->Name();
|
||||
|
||||
ColumnFamilyHandle* import_handle = nullptr;
|
||||
ASSERT_OK(db_->CreateColumnFamilyWithImport(options, cf_name,
|
||||
ImportColumnFamilyOptions(),
|
||||
metadata, &import_handle));
|
||||
ASSERT_NE(import_handle, nullptr);
|
||||
handles.push_back(import_handle);
|
||||
|
||||
std::string result;
|
||||
ASSERT_OK(db_->Get(ReadOptions(), import_handle, key, &result));
|
||||
ASSERT_EQ(value, result);
|
||||
Status ignored = env_->DeleteFile(file_path);
|
||||
ignored.PermitUncheckedError();
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
auto DeleteFilesInRangesForegroundManifestWriteFn =
|
||||
[&](const std::string& key) {
|
||||
uint64_t before_manifest_num = cur_manifest_num;
|
||||
const std::string limit = key + "\xff";
|
||||
std::vector<RangeOpt> ranges;
|
||||
ranges.emplace_back(key, limit);
|
||||
ASSERT_OK(DeleteFilesInRanges(db_.get(), db_->DefaultColumnFamily(),
|
||||
ranges.data(), ranges.size(),
|
||||
/*include_end=*/false));
|
||||
|
||||
std::string result;
|
||||
ASSERT_TRUE(db_->Get(ReadOptions(), key, &result).IsNotFound());
|
||||
UpdateManifestNumsFrom(before_manifest_num);
|
||||
};
|
||||
|
||||
// Column family manipulation.
|
||||
AddCfFn(/*include_background_manifest_write=*/false);
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// SetOptions should not write to the MANIFEST. If that regresses, the write
|
||||
// should be treated as a background write and rotate here.
|
||||
{
|
||||
ASSERT_OK(db_->SetOptions({{"level0_slowdown_writes_trigger", "101"}}));
|
||||
UpdateManifestNumsFrom(cur_manifest_num);
|
||||
}
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// External file ingestion.
|
||||
std::string ingested_key;
|
||||
IngestExternalFileForegroundManifestWriteFn(&ingested_key);
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// Imported column family creation.
|
||||
CreateColumnFamilyWithImportForegroundManifestWriteFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// Physical file deletion by range.
|
||||
DeleteFilesInRangesForegroundManifestWriteFn(ingested_key);
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// Background flush.
|
||||
BackgroundManifestWriteFn();
|
||||
ASSERT_LT(prev_manifest_num, cur_manifest_num);
|
||||
|
||||
// ---- Verify persisted compacted manifest size survives close/reopen ----
|
||||
// ---- Phase 5: persisted compacted manifest size survives close/reopen ----
|
||||
// Close with CFs still live. Reopen with reuse_manifest_on_open so the
|
||||
// manifest is NOT rewritten from scratch. The persisted compacted size
|
||||
// should be loaded and used for auto-tuning.
|
||||
// At this point we have 7 CF handles plus default.
|
||||
ASSERT_EQ(handles.size(), 7U);
|
||||
// At this point we have 8 CF handles plus default.
|
||||
ASSERT_EQ(handles.size(), 8U);
|
||||
|
||||
// Collect CF names for reopen, then release handles (Close needs this)
|
||||
std::vector<std::string> cf_names = {"default"};
|
||||
@@ -163,18 +329,16 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
handles.clear();
|
||||
|
||||
Close();
|
||||
// Use a large max_manifest_file_size so the reused manifest (which is
|
||||
// already ~10KB) does NOT trigger rotation on the first few writes.
|
||||
// Auto-tuning with the persisted compacted size (~5KB) at 200% amp
|
||||
// gives a tuned threshold of ~15KB. Without persistence, the threshold
|
||||
// would be max(max_manifest_file_size, 0 * anything) =
|
||||
// max_manifest_file_size.
|
||||
// Use a max_manifest_file_size below the reused manifest size. Auto-tuning
|
||||
// with the persisted compacted size at 200% amp keeps the tuned threshold
|
||||
// high enough. Without persistence, the threshold would be
|
||||
// max(max_manifest_file_size, 0 * anything) = max_manifest_file_size.
|
||||
//
|
||||
// We set max_manifest_file_size to two values to distinguish:
|
||||
// - 3000: if persisted compacted size is NOT loaded, tuned = 3000,
|
||||
// and the first AddCf will rotate (manifest is already ~10KB > 3000)
|
||||
// - With persisted compacted size loaded, tuned = max(3000, 5000*3) = 15000,
|
||||
// so no rotation until we exceed 15KB
|
||||
// With max_manifest_file_size set to 3000, missing persisted compacted size
|
||||
// would keep tuned = 3000 and the first AddCf would rotate because the
|
||||
// reused manifest is already larger. With persisted compacted size loaded,
|
||||
// the tuned threshold is based on the compacted size, so no rotation until
|
||||
// the manifest grows well beyond the minimum.
|
||||
options.max_manifest_file_size = 3000;
|
||||
options.max_manifest_space_amp_pct = 200;
|
||||
options.reuse_manifest_on_open = true;
|
||||
@@ -186,11 +350,11 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
// during reopen, because the persisted compacted size keeps the tuned
|
||||
// threshold high enough. Without persistence, last_compacted = 0, so
|
||||
// tuned = max_manifest_file_size = 3000, and the first LogAndApply
|
||||
// during Open rotates the manifest because it's already ~10KB > 3000.
|
||||
// during Open rotates the manifest because it is already larger than 3000.
|
||||
ASSERT_EQ(manifest_num_before_reopen, cur_manifest_num);
|
||||
|
||||
// Adding CFs should still not trigger rotation because the tuned
|
||||
// threshold (~15KB) exceeds the current manifest size (~10KB + adds).
|
||||
// Adding CFs should still not trigger rotation because the tuned threshold
|
||||
// from the persisted compacted size exceeds the current manifest size.
|
||||
for (int i = 1; i <= 4; ++i) {
|
||||
AddCfFn();
|
||||
ASSERT_EQ(prev_manifest_num, cur_manifest_num);
|
||||
@@ -202,7 +366,7 @@ TEST_F(DBEtc3Test, AutoTuneManifestSize) {
|
||||
|
||||
// Wrap up
|
||||
while (!handles.empty()) {
|
||||
DropCfFn();
|
||||
DropCfFn(/*include_background_manifest_write=*/false);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+10
-5
@@ -5853,6 +5853,7 @@ Status DBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
|
||||
}
|
||||
|
||||
VersionEdit edit;
|
||||
edit.MarkForegroundOperation();
|
||||
std::set<FileMetaData*> deleted_files;
|
||||
JobContext job_context(next_job_id_.fetch_add(1), true);
|
||||
{
|
||||
@@ -6963,7 +6964,9 @@ Status DBImpl::IngestExternalFiles(
|
||||
assert(!cfd->IsDropped());
|
||||
cfds_to_commit.push_back(cfd);
|
||||
autovector<VersionEdit*> edit_list;
|
||||
edit_list.push_back(ingestion_jobs[i].edit());
|
||||
auto* edit = ingestion_jobs[i].edit();
|
||||
edit->MarkForegroundOperation();
|
||||
edit_list.push_back(edit);
|
||||
edit_lists.push_back(edit_list);
|
||||
++num_entries;
|
||||
}
|
||||
@@ -6978,7 +6981,6 @@ Status DBImpl::IngestExternalFiles(
|
||||
}
|
||||
status =
|
||||
versions_->LogAndApply(cfds_to_commit, read_options, write_options,
|
||||
|
||||
edit_lists, &mutex_, directories_.GetDbDir());
|
||||
// It is safe to update VersionSet last seqno here after LogAndApply since
|
||||
// LogAndApply persists last sequence number from VersionEdits,
|
||||
@@ -7111,6 +7113,7 @@ Status DBImpl::CreateColumnFamilyWithImport(
|
||||
|
||||
SuperVersionContext dummy_sv_ctx(/* create_superversion */ true);
|
||||
VersionEdit dummy_edit;
|
||||
dummy_edit.MarkForegroundOperation();
|
||||
uint64_t next_file_number = 0;
|
||||
std::unique_ptr<std::list<uint64_t>::iterator> pending_output_elem;
|
||||
{
|
||||
@@ -7168,9 +7171,10 @@ Status DBImpl::CreateColumnFamilyWithImport(
|
||||
|
||||
// Install job edit [Mutex will be unlocked here]
|
||||
if (status.ok()) {
|
||||
status = versions_->LogAndApply(cfd, read_options, write_options,
|
||||
import_job.edit(), &mutex_,
|
||||
directories_.GetDbDir());
|
||||
auto* edit = import_job.edit();
|
||||
edit->MarkForegroundOperation();
|
||||
status = versions_->LogAndApply(cfd, read_options, write_options, edit,
|
||||
&mutex_, directories_.GetDbDir());
|
||||
if (status.ok()) {
|
||||
InstallSuperVersionForConfigChange(cfd, &sv_context);
|
||||
}
|
||||
@@ -7601,6 +7605,7 @@ Status DBImpl::ReserveFileNumbersBeforeIngestion(
|
||||
// reuse the file number that has already assigned to the internal file,
|
||||
// and this will overwrite the external file. To protect the external
|
||||
// file, we have to make sure the file number will never being reused.
|
||||
dummy_edit.MarkForegroundOperation();
|
||||
s = versions_->LogAndApply(cfd, read_options, write_options, &dummy_edit,
|
||||
&mutex_, directories_.GetDbDir());
|
||||
if (s.ok()) {
|
||||
|
||||
@@ -4067,6 +4067,13 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
|
||||
status = Status::ShutdownInProgress();
|
||||
} else if (compaction_aborted_.load(std::memory_order_acquire) > 0) {
|
||||
status = Status::Incomplete(Status::SubCode::kCompactionAborted);
|
||||
if (!is_prepicked) {
|
||||
// This automatic compaction was scheduled from compaction_queue_, but
|
||||
// abort happened before PickCompactionFromQueue() could remove the
|
||||
// queued CF. Restore the unscheduled count so ResumeAllCompactions()
|
||||
// can schedule the still-queued work.
|
||||
unscheduled_compactions_++;
|
||||
}
|
||||
} else if (is_manual &&
|
||||
manual_compaction->canceled.load(std::memory_order_acquire)) {
|
||||
status = Status::Incomplete(Status::SubCode::kManualCompactionPaused);
|
||||
@@ -4074,10 +4081,14 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
|
||||
} else {
|
||||
status = error_handler_.GetBGError();
|
||||
// If we get here, it means a hard error happened after this compaction
|
||||
// was scheduled by MaybeScheduleFlushOrCompaction(), but before it got
|
||||
// a chance to execute. Since we didn't pop a cfd from the compaction
|
||||
// queue, increment unscheduled_compactions_
|
||||
unscheduled_compactions_++;
|
||||
// was scheduled by MaybeScheduleFlushOrCompaction(), but before it got a
|
||||
// chance to execute. Since non-prepicked work consumed an unscheduled
|
||||
// compaction credit without popping a cfd from the compaction queue,
|
||||
// restore the credit here. Prepicked work was scheduled directly and did
|
||||
// not consume unscheduled_compactions_.
|
||||
if (!is_prepicked) {
|
||||
unscheduled_compactions_++;
|
||||
}
|
||||
}
|
||||
|
||||
if (!status.ok()) {
|
||||
|
||||
@@ -216,12 +216,19 @@ class TestIterator : public InternalIterator {
|
||||
bool IsKeyPinned() const override { return true; }
|
||||
bool IsValuePinned() const override { return true; }
|
||||
|
||||
void Prepare(const MultiScanArgs* /*scan_opts*/) override {
|
||||
++prepare_call_count_;
|
||||
}
|
||||
|
||||
size_t prepare_call_count() const { return prepare_call_count_; }
|
||||
|
||||
private:
|
||||
bool initialized_;
|
||||
bool valid_;
|
||||
size_t sequence_number_;
|
||||
size_t iter_;
|
||||
size_t steps_ = 0;
|
||||
size_t prepare_call_count_ = 0;
|
||||
|
||||
InternalKeyComparator cmp;
|
||||
std::vector<std::pair<std::string, std::string>> data_;
|
||||
@@ -243,6 +250,53 @@ class DBIteratorTest : public testing::Test {
|
||||
DBIteratorTest() : env_(Env::Default()) {}
|
||||
};
|
||||
|
||||
TEST_F(DBIteratorTest, PrepareForwardsValidatedScanRanges) {
|
||||
Options options;
|
||||
ImmutableOptions ioptions = ImmutableOptions(options);
|
||||
MutableCFOptions mutable_cf_options = MutableCFOptions(options);
|
||||
TestIterator* internal_iter = new TestIterator(BytewiseComparator());
|
||||
internal_iter->AddPut("a", "val_a");
|
||||
internal_iter->AddPut("b", "val_b");
|
||||
internal_iter->Finish();
|
||||
|
||||
ReadOptions ro;
|
||||
std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
|
||||
env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
|
||||
internal_iter, nullptr /* version */, 10 /* sequence */,
|
||||
nullptr /* read_callback */, /*active_mem=*/nullptr));
|
||||
|
||||
MultiScanArgs scan_opts(BytewiseComparator());
|
||||
scan_opts.insert("a", "c");
|
||||
|
||||
db_iter->Prepare(scan_opts);
|
||||
ASSERT_OK(db_iter->status());
|
||||
EXPECT_EQ(internal_iter->prepare_call_count(), 1);
|
||||
}
|
||||
|
||||
TEST_F(DBIteratorTest, PrepareDoesNotForwardInvalidScanRanges) {
|
||||
Options options;
|
||||
ImmutableOptions ioptions = ImmutableOptions(options);
|
||||
MutableCFOptions mutable_cf_options = MutableCFOptions(options);
|
||||
TestIterator* internal_iter = new TestIterator(BytewiseComparator());
|
||||
internal_iter->AddPut("a", "val_a");
|
||||
internal_iter->AddPut("b", "val_b");
|
||||
internal_iter->Finish();
|
||||
|
||||
ReadOptions ro;
|
||||
std::unique_ptr<Iterator> db_iter(DBIter::NewIter(
|
||||
env_, ro, ioptions, mutable_cf_options, BytewiseComparator(),
|
||||
internal_iter, nullptr /* version */, 10 /* sequence */,
|
||||
nullptr /* read_callback */, /*active_mem=*/nullptr));
|
||||
|
||||
MultiScanArgs scan_opts(BytewiseComparator());
|
||||
scan_opts.insert("b", "d");
|
||||
scan_opts.insert("a", "c");
|
||||
|
||||
db_iter->Prepare(scan_opts);
|
||||
EXPECT_TRUE(db_iter->status().IsInvalidArgument());
|
||||
EXPECT_EQ(internal_iter->prepare_call_count(), 0);
|
||||
}
|
||||
|
||||
TEST_F(DBIteratorTest, DBIteratorPrevNext) {
|
||||
Options options;
|
||||
ImmutableOptions ioptions = ImmutableOptions(options);
|
||||
|
||||
@@ -2277,6 +2277,10 @@ TEST_P(DBIteratorTest, ReadAhead) {
|
||||
options.env = env_;
|
||||
options.disable_auto_compactions = true;
|
||||
options.write_buffer_size = 4 << 20;
|
||||
// Pin compression so the readahead byte thresholds below don't depend on the
|
||||
// default compression type. kNoCompression keeps the SST files large enough
|
||||
// to exercise readahead and is always available.
|
||||
options.compression = kNoCompression;
|
||||
options.statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();
|
||||
BlockBasedTableOptions table_options;
|
||||
table_options.block_size = 1024;
|
||||
@@ -4395,6 +4399,71 @@ TEST_P(DBMultiScanIteratorTest, RangeAcrossFiles) {
|
||||
iter.reset();
|
||||
}
|
||||
|
||||
TEST_P(DBMultiScanIteratorTest, SortedRangesSkipIODispatcherSort) {
|
||||
auto options = CurrentOptions();
|
||||
options.target_file_size_base = 100 << 10;
|
||||
options.compaction_style = kCompactionStyleUniversal;
|
||||
options.num_levels = 50;
|
||||
options.compression = kNoCompression;
|
||||
DestroyAndReopen(options);
|
||||
|
||||
auto key = [](int i) {
|
||||
std::stringstream ss;
|
||||
ss << "k" << std::setw(5) << std::setfill('0') << i;
|
||||
return ss.str();
|
||||
};
|
||||
|
||||
auto rnd = Random::GetTLSInstance();
|
||||
for (int i = 0; i < 100; ++i) {
|
||||
ASSERT_OK(Put(key(i), rnd->RandomString(2 << 10)));
|
||||
}
|
||||
ASSERT_OK(Flush());
|
||||
|
||||
ASSERT_OK(db_->CompactRange({}, nullptr, nullptr));
|
||||
ASSERT_EQ(2, NumTableFilesAtLevel(49));
|
||||
|
||||
int sort_count = 0;
|
||||
SyncPoint::GetInstance()->SetCallBack(
|
||||
"IODispatcherImpl::SubmitJob:SortBlockHandles",
|
||||
[&](void* /*arg*/) { ++sort_count; });
|
||||
SyncPoint::GetInstance()->EnableProcessing();
|
||||
|
||||
auto tracking_dispatcher = std::make_shared<TrackingIODispatcher>();
|
||||
std::vector<std::string> key_ranges({key(10), key(90)});
|
||||
MultiScanArgs scan_options(BytewiseComparator());
|
||||
scan_options.io_dispatcher = tracking_dispatcher;
|
||||
scan_options.use_async_io = false;
|
||||
scan_options.insert(key_ranges[0], key_ranges[1]);
|
||||
|
||||
ReadOptions ro;
|
||||
ro.fill_cache = GetParam();
|
||||
ColumnFamilyHandle* cfh = dbfull()->DefaultColumnFamily();
|
||||
std::unique_ptr<MultiScan> iter =
|
||||
dbfull()->NewMultiScan(ro, cfh, scan_options);
|
||||
|
||||
try {
|
||||
int i = 10;
|
||||
for (auto range : *iter) {
|
||||
for (auto it : range) {
|
||||
ASSERT_EQ(it.first.ToString(), key(i));
|
||||
++i;
|
||||
}
|
||||
}
|
||||
ASSERT_EQ(i, 90);
|
||||
} catch (MultiScanException& ex) {
|
||||
ASSERT_NOK(ex.status());
|
||||
std::cerr << "Iterator returned status " << ex.what();
|
||||
abort();
|
||||
} catch (std::logic_error& ex) {
|
||||
std::cerr << "Iterator returned logic error " << ex.what();
|
||||
abort();
|
||||
}
|
||||
|
||||
ASSERT_GT(tracking_dispatcher->GetReadSets().size(), 0);
|
||||
EXPECT_EQ(sort_count, 0);
|
||||
iter.reset();
|
||||
}
|
||||
|
||||
TEST_P(DBMultiScanIteratorTest, FailureTest) {
|
||||
auto options = CurrentOptions();
|
||||
options.compression = kNoCompression;
|
||||
|
||||
@@ -978,6 +978,11 @@ class VersionEdit {
|
||||
|
||||
bool IsColumnFamilyDrop() const { return is_column_family_drop_; }
|
||||
|
||||
void MarkForegroundOperation() { is_foreground_operation_ = true; }
|
||||
bool IsForegroundOperation() const {
|
||||
return is_foreground_operation_ || IsColumnFamilyManipulation();
|
||||
}
|
||||
|
||||
void MarkNoManifestWriteDummy() { is_no_manifest_write_dummy_ = true; }
|
||||
bool IsNoManifestWriteDummy() const { return is_no_manifest_write_dummy_; }
|
||||
|
||||
@@ -1111,6 +1116,7 @@ class VersionEdit {
|
||||
// it also includes column family name.
|
||||
bool is_column_family_drop_ = false;
|
||||
bool is_column_family_add_ = false;
|
||||
bool is_foreground_operation_ = false;
|
||||
std::string column_family_name_;
|
||||
|
||||
uint32_t remaining_entries_ = 0;
|
||||
|
||||
+25
-2
@@ -6111,9 +6111,32 @@ Status VersionSet::ProcessManifestWrites(
|
||||
|
||||
uint64_t prev_manifest_file_size = manifest_file_size_;
|
||||
assert(pending_manifest_file_number_ == 0);
|
||||
bool has_foreground_operation = false;
|
||||
for (const VersionEdit* e : batch_edits) {
|
||||
if (e->IsForegroundOperation()) {
|
||||
has_foreground_operation = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
// For MANIFEST write batches with any foreground operation (external file
|
||||
// ingestion/import, DeleteFilesInRange(s), and column family manipulations
|
||||
// like CreateColumnFamily and DropColumnFamily), relax the size limit by 25%
|
||||
// to reduce the likelihood of a user operation blocking on MANIFEST rotation.
|
||||
// Background-only batches (flush/compaction) still rotate at the normal
|
||||
// threshold.
|
||||
// TODO/future: for workloads like atomic-replace ingestion-only, with zero
|
||||
// or few flushes and compactions, it might be nice to trigger background
|
||||
// manifest rotation if we are beyond the soft limit. But the vast majority
|
||||
// of workloads should have plenty of background manifest ops to avoid
|
||||
// foreground rotation.
|
||||
uint64_t enforced_limit = tuned_max_manifest_file_size_;
|
||||
if (has_foreground_operation) {
|
||||
uint64_t new_limit = enforced_limit + enforced_limit / 4;
|
||||
// don't keep in case of overflow
|
||||
enforced_limit = std::max(enforced_limit, new_limit);
|
||||
}
|
||||
if (!skip_manifest_write &&
|
||||
(!descriptor_log_ ||
|
||||
prev_manifest_file_size >= tuned_max_manifest_file_size_)) {
|
||||
(!descriptor_log_ || prev_manifest_file_size >= enforced_limit)) {
|
||||
TEST_SYNC_POINT("VersionSet::ProcessManifestWrites:BeforeNewManifest");
|
||||
new_descriptor_log = true;
|
||||
} else {
|
||||
|
||||
@@ -432,9 +432,33 @@ Status GetOptionsFromString(const ConfigOptions& config_options,
|
||||
const Options& base_options,
|
||||
const std::string& opts_str, Options* new_options);
|
||||
|
||||
// StringToMap parses a serialized options string into a map. Each
|
||||
// resulting map value is in a self-contained form (it can be embedded
|
||||
// directly in a `key=value;` context -- e.g. SetOptions -- without further
|
||||
// escaping). Specifically: nested braced values from the input are
|
||||
// preserved with their outer braces. Permissive: accepts both braced and
|
||||
// unbraced forms; values not requiring braces are returned as-is.
|
||||
// Example:
|
||||
// "filter_policy={id=ribbonfilter:10;bloom_before_level=-1};block_size=4096"
|
||||
// produces:
|
||||
// {filter_policy -> "{id=ribbonfilter:10;bloom_before_level=-1}",
|
||||
// block_size -> "4096"}
|
||||
Status StringToMap(const std::string& opts_str,
|
||||
std::unordered_map<std::string, std::string>* opts_map);
|
||||
|
||||
// MapToString is the inverse of StringToMap: a naive `key=value;` join.
|
||||
// Each map value must already be in self-contained form (as returned by
|
||||
// StringToMap) -- i.e. simple text or a single balanced `{...}` block.
|
||||
// Values from StringToMap satisfy this property, so the round-trip
|
||||
//
|
||||
// StringToMap(MapToString(StringToMap(s))) == StringToMap(s)
|
||||
//
|
||||
// holds. Callers building a map by hand are responsible for ensuring
|
||||
// values are self-contained; raw values containing `;` or starting with
|
||||
// `{` without matching `}` won't round-trip.
|
||||
Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
|
||||
std::string* opts_str);
|
||||
|
||||
// Request stopping background work, if wait is true wait until it's done
|
||||
void CancelAllBackgroundWork(DB* db, bool wait = false);
|
||||
|
||||
|
||||
@@ -152,6 +152,11 @@ class BlockBasedTable;
|
||||
struct JobOptions {
|
||||
uint64_t io_coalesce_threshold = 16 * 1024;
|
||||
ReadOptions read_options;
|
||||
|
||||
// True when IOJob::block_handles is already sorted by file offset. Callers
|
||||
// that can guarantee sorted order may set this to let IODispatcher skip
|
||||
// defensive sorting before cache lookup and coalescing.
|
||||
bool block_handles_are_sorted = false;
|
||||
};
|
||||
|
||||
class IOJob {
|
||||
|
||||
@@ -191,18 +191,22 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
|
||||
|
||||
// Compress blocks using the specified compression algorithm.
|
||||
//
|
||||
// Default: kSnappyCompression, if it's supported. If snappy is not linked
|
||||
// with the library, the default is kNoCompression.
|
||||
// Default: kLZ4Compression if support is compiled in, else kSnappyCompression
|
||||
// if support is compiled in, else kNoCompression
|
||||
//
|
||||
// Typical speeds of kSnappyCompression on an Intel(R) Core(TM)2 2.4GHz:
|
||||
// ~200-500MB/s compression
|
||||
// ~400-800MB/s decompression
|
||||
// Typical single-core speed of kLZ4Compression on an AMD EPYC-Genoa
|
||||
// ~800 - 1200 MB/s compression
|
||||
// ~8000 - 16000 MB/s decompression
|
||||
// and with a 160 threads to saturate the cores:
|
||||
// ~60 - 90 GB/s compression
|
||||
// ~900 - 1000 GB/s decompression
|
||||
// using db_bench compress/uncompress benchmarks.
|
||||
//
|
||||
// Note that these speeds are significantly faster than most
|
||||
// persistent storage speeds, and therefore it is typically never
|
||||
// worth switching to kNoCompression. Even if the input data is
|
||||
// incompressible, the kSnappyCompression implementation will
|
||||
// efficiently detect that and will switch to uncompressed mode.
|
||||
// incompressible, the compression implementation will
|
||||
// efficiently detect that and fall back on no compression.
|
||||
//
|
||||
// If you do not set `compression_opts.level`, or set it to
|
||||
// `CompressionOptions::kDefaultCompressionLevel`, we will attempt to pick the
|
||||
@@ -1003,6 +1007,13 @@ struct DBOptions {
|
||||
//
|
||||
// This option is mutable with SetDBOptions(), taking effect on the next
|
||||
// manifest write (e.g. completed DB compaction or flush).
|
||||
//
|
||||
// For MANIFEST write batches containing foreground operations like external
|
||||
// file ingestion/import, DeleteFilesInRange, CreateColumnFamily, and
|
||||
// DropColumnFamily, the effective limit is relaxed by 25% to reduce the
|
||||
// likelihood of user operations blocking on MANIFEST rotation.
|
||||
// Background-only batches (flush and compaction) use the configured or
|
||||
// auto-tuned limit directly.
|
||||
uint64_t max_manifest_file_size = 1024 * 1024 * 1024;
|
||||
|
||||
// If true, on DB close, read back the entire MANIFEST file and validate
|
||||
|
||||
@@ -453,11 +453,13 @@ class OptionTypeInfo {
|
||||
const std::string& value, void* addr) {
|
||||
std::map<std::string, std::string> map;
|
||||
Status s;
|
||||
// Permit `{k1=v1;k2=v2}` (as serialized) or bare `k1=v1;k2=v2`.
|
||||
std::string stripped = OptionTypeInfo::StripOuterBraces(value);
|
||||
for (size_t start = 0, end = 0;
|
||||
s.ok() && start < value.size() && end != std::string::npos;
|
||||
s.ok() && start < stripped.size() && end != std::string::npos;
|
||||
start = end + 1) {
|
||||
std::string token;
|
||||
s = OptionTypeInfo::NextToken(value, item_separator, start, &end,
|
||||
s = OptionTypeInfo::NextToken(stripped, item_separator, start, &end,
|
||||
&token);
|
||||
if (s.ok() && !token.empty()) {
|
||||
size_t pos = token.find(kv_separator);
|
||||
@@ -947,6 +949,14 @@ class OptionTypeInfo {
|
||||
static Status NextToken(const std::string& opts, char delimiter, size_t start,
|
||||
size_t* end, std::string* token);
|
||||
|
||||
// Strips a single layer of outer "{" / "}" wrapping iff the leading '{'
|
||||
// pairs with the trailing '}'. This permissively unwraps values that
|
||||
// could legitimately be either braced or bare (e.g. `{a:b:c}` and
|
||||
// `a:b:c` are accepted equivalently by list-style parsers). The check
|
||||
// is done at brace depth: `{a}:{b}` is left as-is because the leading
|
||||
// '{' closes before the end of the string.
|
||||
static std::string StripOuterBraces(const std::string& value);
|
||||
|
||||
constexpr static const char* kIdPropName() { return "id"; }
|
||||
constexpr static const char* kIdPropSuffix() { return ".id"; }
|
||||
|
||||
@@ -995,12 +1005,16 @@ Status ParseArray(const ConfigOptions& config_options,
|
||||
|
||||
ConfigOptions copy = config_options;
|
||||
copy.ignore_unsupported_options = false;
|
||||
// Permit the caller to pass either bare `item1:item2` or wrapped
|
||||
// `{item1:item2}`. Either form parses identically.
|
||||
std::string stripped = OptionTypeInfo::StripOuterBraces(value);
|
||||
size_t i = 0, start = 0, end = 0;
|
||||
for (; status.ok() && i < kSize && start < value.size() &&
|
||||
for (; status.ok() && i < kSize && start < stripped.size() &&
|
||||
end != std::string::npos;
|
||||
i++, start = end + 1) {
|
||||
std::string token;
|
||||
status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
|
||||
status =
|
||||
OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
|
||||
if (status.ok()) {
|
||||
status = elem_info.Parse(copy, name, token, &((*result)[i]));
|
||||
if (config_options.ignore_unsupported_options &&
|
||||
@@ -1137,11 +1151,15 @@ Status ParseVector(const ConfigOptions& config_options,
|
||||
// object is valid or not.
|
||||
ConfigOptions copy = config_options;
|
||||
copy.ignore_unsupported_options = false;
|
||||
// Permit the caller to pass either bare `item1:item2` or wrapped
|
||||
// `{item1:item2}`. Either form parses identically.
|
||||
std::string stripped = OptionTypeInfo::StripOuterBraces(value);
|
||||
for (size_t start = 0, end = 0;
|
||||
status.ok() && start < value.size() && end != std::string::npos;
|
||||
status.ok() && start < stripped.size() && end != std::string::npos;
|
||||
start = end + 1) {
|
||||
std::string token;
|
||||
status = OptionTypeInfo::NextToken(value, separator, start, &end, &token);
|
||||
status =
|
||||
OptionTypeInfo::NextToken(stripped, separator, start, &end, &token);
|
||||
if (status.ok()) {
|
||||
T elem;
|
||||
status = elem_info.Parse(copy, name, token, &elem);
|
||||
|
||||
@@ -31,8 +31,9 @@ struct SortedRunBuilderOptions {
|
||||
uint64_t target_file_size_bytes = 64 * 1024 * 1024;
|
||||
|
||||
// Compression type for output SST files.
|
||||
// Default: kNoCompression (always available). Set to kSnappyCompression,
|
||||
// kZSTD, etc. if the desired library is linked with your binary.
|
||||
// Default: kNoCompression (always available). Set to kLZ4Compression,
|
||||
// kSnappyCompression, kZSTD, etc. if the desired library is linked with
|
||||
// your binary.
|
||||
CompressionType compression = kNoCompression;
|
||||
|
||||
// Max number of background compaction threads.
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// NOTE: in 'main' development branch, this should be the *next*
|
||||
// minor or major version number planned for release.
|
||||
#define ROCKSDB_MAJOR 11
|
||||
#define ROCKSDB_MINOR 4
|
||||
#define ROCKSDB_MINOR 5
|
||||
#define ROCKSDB_PATCH 0
|
||||
|
||||
// Make it easy to do conditional compilation based on version checks, i.e.
|
||||
|
||||
@@ -33,7 +33,7 @@ public class CompactionOptions extends RocksObject {
|
||||
/**
|
||||
* Set the compaction output compression type.
|
||||
*
|
||||
* Default: snappy
|
||||
* Default: lz4 if available, otherwise snappy if available, otherwise none
|
||||
*
|
||||
* If set to {@link CompressionType#DISABLE_COMPRESSION_OPTION},
|
||||
* RocksDB will choose compression type according to the
|
||||
|
||||
@@ -136,7 +136,8 @@ public interface MutableColumnFamilyOptionsInterface<
|
||||
* Compress blocks using the specified compression algorithm. This
|
||||
* parameter can be changed dynamically.
|
||||
* <p>
|
||||
* Default: SNAPPY_COMPRESSION, which gives lightweight but fast compression.
|
||||
* Default: LZ4_COMPRESSION if available, otherwise SNAPPY_COMPRESSION if
|
||||
* available, otherwise NO_COMPRESSION.
|
||||
*
|
||||
* @param compressionType Compression Type.
|
||||
* @return the reference to the current option.
|
||||
@@ -148,7 +149,8 @@ public interface MutableColumnFamilyOptionsInterface<
|
||||
* Compress blocks using the specified compression algorithm. This
|
||||
* parameter can be changed dynamically.
|
||||
* <p>
|
||||
* Default: SNAPPY_COMPRESSION, which gives lightweight but fast compression.
|
||||
* Default: LZ4_COMPRESSION if available, otherwise SNAPPY_COMPRESSION if
|
||||
* available, otherwise NO_COMPRESSION.
|
||||
*
|
||||
* @return Compression type.
|
||||
*/
|
||||
|
||||
@@ -253,7 +253,7 @@ struct MutableCFOptions {
|
||||
max_sequential_skip_in_iterations(0),
|
||||
paranoid_file_checks(false),
|
||||
report_bg_io_stats(false),
|
||||
compression(Snappy_Supported() ? kSnappyCompression : kNoCompression),
|
||||
compression(GetDefaultCompressionType()),
|
||||
bottommost_compression(kDisableCompressionOption),
|
||||
last_level_temperature(Temperature::kUnknown),
|
||||
default_write_temperature(Temperature::kUnknown),
|
||||
|
||||
@@ -557,11 +557,13 @@ static std::unordered_map<std::string, OptionTypeInfo>
|
||||
embedded.ignore_unsupported_options = true;
|
||||
std::vector<std::shared_ptr<EventListener>> listeners;
|
||||
Status s;
|
||||
// Permit `{item1:item2}` or bare `item1:item2`.
|
||||
std::string stripped = OptionTypeInfo::StripOuterBraces(value);
|
||||
for (size_t start = 0, end = 0;
|
||||
s.ok() && start < value.size() && end != std::string::npos;
|
||||
s.ok() && start < stripped.size() && end != std::string::npos;
|
||||
start = end + 1) {
|
||||
std::string token;
|
||||
s = OptionTypeInfo::NextToken(value, ':', start, &end, &token);
|
||||
s = OptionTypeInfo::NextToken(stripped, ':', start, &end, &token);
|
||||
if (s.ok() && !token.empty()) {
|
||||
std::shared_ptr<EventListener> listener;
|
||||
s = EventListener::CreateFromString(embedded, token, &listener);
|
||||
|
||||
+1
-1
@@ -129,7 +129,7 @@ AdvancedColumnFamilyOptions::AdvancedColumnFamilyOptions(const Options& options)
|
||||
}
|
||||
|
||||
ColumnFamilyOptions::ColumnFamilyOptions()
|
||||
: compression(Snappy_Supported() ? kSnappyCompression : kNoCompression),
|
||||
: compression(GetDefaultCompressionType()),
|
||||
table_factory(
|
||||
std::shared_ptr<TableFactory>(new BlockBasedTableFactory())) {}
|
||||
|
||||
|
||||
+102
-10
@@ -770,6 +770,11 @@ Status StringToMap(const std::string& opts_str,
|
||||
// Example:
|
||||
// opts_str = "write_buffer_size=1024;max_write_buffer_number=2;"
|
||||
// "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100"
|
||||
//
|
||||
// Each value in the resulting map can be substituted directly into a
|
||||
// `key=value;` context (e.g. SetOptions) and re-parsed without ambiguity:
|
||||
// values that were wrapped in `{...}` in the input keep their wrapping,
|
||||
// so embedded `;` characters stay escaped.
|
||||
size_t pos = 0;
|
||||
std::string opts = trim(opts_str);
|
||||
// If the input string starts and ends with "{...}", strip off the brackets
|
||||
@@ -790,23 +795,78 @@ Status StringToMap(const std::string& opts_str,
|
||||
return Status::InvalidArgument("Empty key found");
|
||||
}
|
||||
|
||||
// Locate value start (after '=' and any whitespace).
|
||||
size_t value_start = eq_pos + 1;
|
||||
while (value_start < opts.size() && isspace(opts[value_start])) {
|
||||
++value_start;
|
||||
}
|
||||
|
||||
std::string value;
|
||||
Status s = OptionTypeInfo::NextToken(opts, ';', eq_pos + 1, &pos, &value);
|
||||
if (!s.ok()) {
|
||||
return s;
|
||||
} else {
|
||||
(*opts_map)[key] = value;
|
||||
if (pos == std::string::npos) {
|
||||
break;
|
||||
} else {
|
||||
pos++;
|
||||
if (value_start < opts.size() && opts[value_start] == '{') {
|
||||
// Braced value: extract the entire `{...}` substring INCLUDING braces
|
||||
// so that the value can be re-emitted as-is in a `key=value;` context.
|
||||
int count = 1;
|
||||
size_t brace_pos = value_start + 1;
|
||||
while (brace_pos < opts.size() && count > 0) {
|
||||
if (opts[brace_pos] == '{') {
|
||||
++count;
|
||||
} else if (opts[brace_pos] == '}') {
|
||||
--count;
|
||||
}
|
||||
if (count > 0) {
|
||||
++brace_pos;
|
||||
}
|
||||
}
|
||||
if (count != 0) {
|
||||
return Status::InvalidArgument(
|
||||
"Mismatched curly braces for nested options");
|
||||
}
|
||||
value = opts.substr(value_start, brace_pos - value_start + 1);
|
||||
pos = brace_pos + 1;
|
||||
// Allow whitespace, then either delimiter or end.
|
||||
while (pos < opts.size() && isspace(opts[pos])) {
|
||||
++pos;
|
||||
}
|
||||
if (pos < opts.size() && opts[pos] != ';') {
|
||||
return Status::InvalidArgument("Unexpected chars after nested options");
|
||||
}
|
||||
} else {
|
||||
// Non-braced: scan to the next ';'. The value may contain '=' (only
|
||||
// the first '=' separates key from value).
|
||||
size_t end = opts.find(';', value_start);
|
||||
if (end == std::string::npos) {
|
||||
value = trim(opts.substr(value_start));
|
||||
pos = std::string::npos;
|
||||
} else {
|
||||
value = trim(opts.substr(value_start, end - value_start));
|
||||
pos = end;
|
||||
}
|
||||
}
|
||||
|
||||
(*opts_map)[key] = value;
|
||||
if (pos == std::string::npos) {
|
||||
break;
|
||||
} else {
|
||||
pos++;
|
||||
}
|
||||
}
|
||||
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status MapToString(const std::unordered_map<std::string, std::string>& opts_map,
|
||||
std::string* opts_str) {
|
||||
assert(opts_str);
|
||||
opts_str->clear();
|
||||
for (const auto& [key, value] : opts_map) {
|
||||
opts_str->append(key);
|
||||
opts_str->append("=");
|
||||
opts_str->append(value);
|
||||
opts_str->append(";");
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status GetStringFromDBOptions(std::string* opt_string,
|
||||
const DBOptions& db_options,
|
||||
const std::string& delimiter) {
|
||||
@@ -1053,6 +1113,32 @@ Status OptionTypeInfo::NextToken(const std::string& opts, char delimiter,
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
std::string OptionTypeInfo::StripOuterBraces(const std::string& value) {
|
||||
if (value.size() < 2 || value.front() != '{' || value.back() != '}') {
|
||||
return value;
|
||||
}
|
||||
// Verify the leading '{' actually pairs with the trailing '}'.
|
||||
// For `{a:b}` the leading brace closes only at the trailing `}` (one
|
||||
// matching pair, OK to strip). For `{a}:{b}` the leading brace closes
|
||||
// at the first inner `}`, so the leading and trailing braces are
|
||||
// independent -- leave the value alone. We strip at most one layer:
|
||||
// each layer of `{}` corresponds to one level of nesting that the
|
||||
// encoder added for that level, and the typed parser at this level
|
||||
// wants to peel exactly its own layer.
|
||||
int depth = 0;
|
||||
for (size_t i = 0; i + 1 < value.size(); ++i) {
|
||||
if (value[i] == '{') {
|
||||
++depth;
|
||||
} else if (value[i] == '}') {
|
||||
--depth;
|
||||
if (depth == 0) {
|
||||
return value; // outer braces are independent
|
||||
}
|
||||
}
|
||||
}
|
||||
return value.substr(1, value.size() - 2);
|
||||
}
|
||||
|
||||
Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
|
||||
const std::string& opt_name,
|
||||
const std::string& value, void* opt_ptr) const {
|
||||
@@ -1071,7 +1157,13 @@ Status OptionTypeInfo::Parse(const ConfigOptions& config_options,
|
||||
copy.invoke_prepare_options = false;
|
||||
void* opt_addr = GetOffset(opt_ptr);
|
||||
return parse_func_(copy, opt_name, opt_value, opt_addr);
|
||||
} else if (ParseOptionHelper(GetOffset(opt_ptr), type_, opt_value)) {
|
||||
} else if (ParseOptionHelper(GetOffset(opt_ptr), type_,
|
||||
StripOuterBraces(opt_value))) {
|
||||
// Scalar types (int, bool, enum, string, ...). StringToMap now
|
||||
// preserves outer braces in nested values, so a hand-crafted
|
||||
// `key={42}` (or `db_log_dir={/tmp}`) lands here with a wrap
|
||||
// that scalar parsers don't expect; permissively strip one
|
||||
// level so braced scalar input still parses as before.
|
||||
return Status::OK();
|
||||
} else if (IsConfigurable()) {
|
||||
// The option is <config>.<name>
|
||||
|
||||
+196
-11
@@ -1938,11 +1938,13 @@ TEST_F(OptionsTest, StringToMapTest) {
|
||||
ASSERT_EQ(opts_map["k2"], "");
|
||||
ASSERT_TRUE(opts_map.find("k3") != opts_map.end());
|
||||
ASSERT_EQ(opts_map["k3"], "");
|
||||
// Regular nested options
|
||||
// Regular nested options. Braces are preserved on values that came in
|
||||
// braced, so the value is self-contained for direct embedding via
|
||||
// `key=value;` and round-trips through StringToMap/MapToString.
|
||||
opts_map.clear();
|
||||
ASSERT_OK(StringToMap("k1=v1;k2={nk1=nv1;nk2=nv2};k3=v3", &opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2=nv2");
|
||||
ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2=nv2}");
|
||||
ASSERT_EQ(opts_map["k3"], "v3");
|
||||
// Multi-level nested options
|
||||
opts_map.clear();
|
||||
@@ -1951,34 +1953,35 @@ TEST_F(OptionsTest, StringToMapTest) {
|
||||
"k3={nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}};k4=v4",
|
||||
&opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "nk1=nv1;nk2={nnk1=nnk2}");
|
||||
ASSERT_EQ(opts_map["k3"], "nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}");
|
||||
ASSERT_EQ(opts_map["k2"], "{nk1=nv1;nk2={nnk1=nnk2}}");
|
||||
ASSERT_EQ(opts_map["k3"], "{nk1={nnk1={nnnk1=nnnv1;nnnk2;nnnv2}}}");
|
||||
ASSERT_EQ(opts_map["k4"], "v4");
|
||||
// Garbage inside curly braces
|
||||
opts_map.clear();
|
||||
ASSERT_OK(StringToMap("k1=v1;k2={dfad=};k3={=};k4=v4", &opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "dfad=");
|
||||
ASSERT_EQ(opts_map["k3"], "=");
|
||||
ASSERT_EQ(opts_map["k2"], "{dfad=}");
|
||||
ASSERT_EQ(opts_map["k3"], "{=}");
|
||||
ASSERT_EQ(opts_map["k4"], "v4");
|
||||
// Empty nested options
|
||||
opts_map.clear();
|
||||
ASSERT_OK(StringToMap("k1=v1;k2={};", &opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "");
|
||||
ASSERT_EQ(opts_map["k2"], "{}");
|
||||
opts_map.clear();
|
||||
ASSERT_OK(StringToMap("k1=v1;k2={{{{}}}{}{}};", &opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "{{{}}}{}{}");
|
||||
// With random spaces
|
||||
ASSERT_EQ(opts_map["k2"], "{{{{}}}{}{}}");
|
||||
// With random spaces. Internal whitespace is preserved verbatim now
|
||||
// (we keep the original substring of the value rather than trimming).
|
||||
opts_map.clear();
|
||||
ASSERT_OK(
|
||||
StringToMap(" k1 = v1 ; k2= {nk1=nv1; nk2={nnk1=nnk2}} ; "
|
||||
"k3={ { } }; k4= v4 ",
|
||||
&opts_map));
|
||||
ASSERT_EQ(opts_map["k1"], "v1");
|
||||
ASSERT_EQ(opts_map["k2"], "nk1=nv1; nk2={nnk1=nnk2}");
|
||||
ASSERT_EQ(opts_map["k3"], "{ }");
|
||||
ASSERT_EQ(opts_map["k2"], "{nk1=nv1; nk2={nnk1=nnk2}}");
|
||||
ASSERT_EQ(opts_map["k3"], "{ { } }");
|
||||
ASSERT_EQ(opts_map["k4"], "v4");
|
||||
|
||||
// Empty key
|
||||
@@ -2008,6 +2011,165 @@ TEST_F(OptionsTest, StringToMapTest) {
|
||||
ASSERT_NOK(StringToMap("k1=v1;k2={{dfdl}adfa}{}", &opts_map));
|
||||
}
|
||||
|
||||
TEST_F(OptionsTest, EmptyBracedVectorAndBracedScalarTest) {
|
||||
// An empty braced list value (`key={}`) must parse as an empty
|
||||
// collection, not as a vector with one empty element. StringToMap
|
||||
// preserves the braces, so the typed vector parser sees `{}`; its
|
||||
// outer-brace strip needs to handle the size-2 case.
|
||||
ColumnFamilyOptions cf_opts;
|
||||
ConfigOptions cfg;
|
||||
ASSERT_OK(GetColumnFamilyOptionsFromString(
|
||||
cfg, ColumnFamilyOptions(), "compression_per_level={};", &cf_opts));
|
||||
EXPECT_TRUE(cf_opts.compression_per_level.empty());
|
||||
|
||||
// A scalar value wrapped in `{}` (e.g. hand-crafted `key={42}`) must
|
||||
// still parse. StringToMap preserves the braces, so the scalar
|
||||
// dispatch in OptionTypeInfo::Parse permissively strips one outer
|
||||
// layer.
|
||||
DBOptions db_opts;
|
||||
ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(), "max_open_files={42};",
|
||||
&db_opts));
|
||||
EXPECT_EQ(db_opts.max_open_files, 42);
|
||||
|
||||
ASSERT_OK(GetDBOptionsFromString(cfg, DBOptions(),
|
||||
"create_if_missing={true};", &db_opts));
|
||||
EXPECT_TRUE(db_opts.create_if_missing);
|
||||
}
|
||||
|
||||
TEST_F(OptionsTest, MapToStringTest) {
|
||||
using Map = std::unordered_map<std::string, std::string>;
|
||||
std::string s;
|
||||
|
||||
// Simple values.
|
||||
Map simple{{"a", "1"}, {"b", "hello"}};
|
||||
ASSERT_OK(MapToString(simple, &s));
|
||||
Map reparsed;
|
||||
ASSERT_OK(StringToMap(s, &reparsed));
|
||||
ASSERT_EQ(reparsed, simple);
|
||||
|
||||
// A value already in self-contained braced form (as returned by
|
||||
// StringToMap) passes through unchanged. No extra wrapping is needed
|
||||
// for round-trip.
|
||||
Map already_braced{{"k", "{a=b;c=d}"}};
|
||||
ASSERT_OK(MapToString(already_braced, &s));
|
||||
ASSERT_EQ(s, "k={a=b;c=d};");
|
||||
Map reparsed_braced;
|
||||
ASSERT_OK(StringToMap(s, &reparsed_braced));
|
||||
ASSERT_EQ(reparsed_braced, already_braced);
|
||||
}
|
||||
|
||||
TEST_F(OptionsTest, StringToMapMapToStringRoundTripTest) {
|
||||
using Map = std::unordered_map<std::string, std::string>;
|
||||
// Pull a single entry out of a StringToMap-ed map and re-embed it
|
||||
// directly into a `key=value;` SetOptions-style string. The pulled
|
||||
// value must round-trip without any extra escaping by the caller --
|
||||
// this is the property that lets users compose option strings from
|
||||
// map entries safely.
|
||||
const std::string original =
|
||||
"filter_policy={id=ribbonfilter:10:-1;bloom_before_level=-1;};"
|
||||
"block_size=4096;cache_index_and_filter_blocks=true";
|
||||
|
||||
Map parsed;
|
||||
ASSERT_OK(StringToMap(original, &parsed));
|
||||
// The filter_policy value retains its braces -- it's self-contained.
|
||||
EXPECT_EQ(parsed["filter_policy"],
|
||||
"{id=ribbonfilter:10:-1;bloom_before_level=-1;}");
|
||||
EXPECT_EQ(parsed["block_size"], "4096");
|
||||
EXPECT_EQ(parsed["cache_index_and_filter_blocks"], "true");
|
||||
|
||||
// Pull a single entry out and embed it in a fresh `key=value;`
|
||||
// string. No extra wrapping required by the caller.
|
||||
const std::string embedded =
|
||||
"filter_policy=" + parsed["filter_policy"] + ";other_opt=42";
|
||||
Map reparsed;
|
||||
ASSERT_OK(StringToMap(embedded, &reparsed));
|
||||
EXPECT_EQ(reparsed["filter_policy"], parsed["filter_policy"]);
|
||||
EXPECT_EQ(reparsed["other_opt"], "42");
|
||||
|
||||
// Whole-map round-trip via MapToString.
|
||||
std::string regenerated;
|
||||
ASSERT_OK(MapToString(parsed, ®enerated));
|
||||
Map reparsed2;
|
||||
ASSERT_OK(StringToMap(regenerated, &reparsed2));
|
||||
EXPECT_EQ(reparsed2, parsed);
|
||||
}
|
||||
|
||||
TEST_F(OptionsTest, FullOptionsStringToMapRoundTripTest) {
|
||||
// Round-trip a populated DBOptions and CFOptions through the
|
||||
// StringToMap -> MapToString path. This catches any custom option
|
||||
// serializer that emits a value containing the StringToMap delimiter
|
||||
// (';') without enclosing it in '{}' -- such a value would survive
|
||||
// StringToMap as a self-contained entry but corrupt subsequent
|
||||
// re-serialization. Existing typed tests
|
||||
// (ColumnFamilyOptionsSerialization, DBOptionsSerialization) cover the
|
||||
// typed Configurable round-trip; this test specifically exercises the
|
||||
// public StringToMap + MapToString path that callers use to compose,
|
||||
// mutate, and re-emit option strings.
|
||||
Random rnd(607);
|
||||
ConfigOptions config_options;
|
||||
config_options.input_strings_escaped = false;
|
||||
|
||||
// ---- ColumnFamilyOptions ----
|
||||
Options options;
|
||||
ColumnFamilyOptions base_cf_opt;
|
||||
base_cf_opt.comparator = test::BytewiseComparatorWithU64TsWrapper();
|
||||
test::RandomInitCFOptions(&base_cf_opt, options, &rnd);
|
||||
std::string cf_str;
|
||||
ASSERT_OK(
|
||||
GetStringFromColumnFamilyOptions(config_options, base_cf_opt, &cf_str));
|
||||
|
||||
std::unordered_map<std::string, std::string> cf_map;
|
||||
ASSERT_OK(StringToMap(cf_str, &cf_map));
|
||||
// Each entry must be embeddable as-is in a `key=value;` context.
|
||||
for (const auto& [k, v] : cf_map) {
|
||||
std::string solo;
|
||||
solo.append(k).append("=").append(v).append(";");
|
||||
std::unordered_map<std::string, std::string> solo_map;
|
||||
ASSERT_OK(StringToMap(solo, &solo_map))
|
||||
<< "single-entry round-trip failed for " << k << ": " << v;
|
||||
ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
|
||||
ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
|
||||
}
|
||||
// Whole-map MapToString must reproduce a string parseable into a CF
|
||||
// options object equivalent to the original.
|
||||
std::string cf_round;
|
||||
ASSERT_OK(MapToString(cf_map, &cf_round));
|
||||
ColumnFamilyOptions cf_back;
|
||||
ASSERT_OK(GetColumnFamilyOptionsFromString(
|
||||
config_options, ColumnFamilyOptions(), cf_round, &cf_back));
|
||||
ASSERT_OK(RocksDBOptionsParser::VerifyCFOptions(config_options, base_cf_opt,
|
||||
cf_back));
|
||||
|
||||
// ---- DBOptions ----
|
||||
DBOptions base_db_opt;
|
||||
test::RandomInitDBOptions(&base_db_opt, &rnd);
|
||||
std::string db_str;
|
||||
ASSERT_OK(GetStringFromDBOptions(config_options, base_db_opt, &db_str));
|
||||
|
||||
std::unordered_map<std::string, std::string> db_map;
|
||||
ASSERT_OK(StringToMap(db_str, &db_map));
|
||||
for (const auto& [k, v] : db_map) {
|
||||
std::string solo;
|
||||
solo.append(k).append("=").append(v).append(";");
|
||||
std::unordered_map<std::string, std::string> solo_map;
|
||||
ASSERT_OK(StringToMap(solo, &solo_map))
|
||||
<< "single-entry round-trip failed for " << k << ": " << v;
|
||||
ASSERT_EQ(solo_map.size(), 1u) << k << "=" << v;
|
||||
ASSERT_EQ(solo_map[k], v) << "single-entry value mismatch for " << k;
|
||||
}
|
||||
std::string db_round;
|
||||
ASSERT_OK(MapToString(db_map, &db_round));
|
||||
DBOptions db_back;
|
||||
ASSERT_OK(
|
||||
GetDBOptionsFromString(config_options, DBOptions(), db_round, &db_back));
|
||||
ASSERT_OK(RocksDBOptionsParser::VerifyDBOptions(config_options, base_db_opt,
|
||||
db_back));
|
||||
|
||||
// RandomInitCFOptions allocates a raw CompactionFilter*; the
|
||||
// ColumnFamilyOptions destructor doesn't own it.
|
||||
delete base_cf_opt.compaction_filter;
|
||||
}
|
||||
|
||||
TEST_F(OptionsTest, StringToMapRandomTest) {
|
||||
std::unordered_map<std::string, std::string> opts_map;
|
||||
// Make sure segfault is not hit by semi-random strings
|
||||
@@ -2028,6 +2190,15 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
|
||||
str[pos] = ' ';
|
||||
Status s = StringToMap(str, &opts_map);
|
||||
ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
|
||||
if (s.ok()) {
|
||||
// Round-trip: StringToMap entries are self-contained, so
|
||||
// MapToString of them must parse back to the same map.
|
||||
std::string regen;
|
||||
ASSERT_OK(MapToString(opts_map, ®en));
|
||||
std::unordered_map<std::string, std::string> reparsed;
|
||||
ASSERT_OK(StringToMap(regen, &reparsed));
|
||||
ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
|
||||
}
|
||||
opts_map.clear();
|
||||
}
|
||||
}
|
||||
@@ -2047,8 +2218,22 @@ TEST_F(OptionsTest, StringToMapRandomTest) {
|
||||
}
|
||||
Status s = StringToMap(str, &opts_map);
|
||||
ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
|
||||
if (s.ok()) {
|
||||
std::string regen;
|
||||
ASSERT_OK(MapToString(opts_map, ®en));
|
||||
std::unordered_map<std::string, std::string> reparsed;
|
||||
ASSERT_OK(StringToMap(regen, &reparsed));
|
||||
ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
|
||||
}
|
||||
s = StringToMap("name=" + str, &opts_map);
|
||||
ASSERT_TRUE(s.ok() || s.IsInvalidArgument());
|
||||
if (s.ok()) {
|
||||
std::string regen;
|
||||
ASSERT_OK(MapToString(opts_map, ®en));
|
||||
std::unordered_map<std::string, std::string> reparsed;
|
||||
ASSERT_OK(StringToMap(regen, &reparsed));
|
||||
ASSERT_EQ(reparsed, opts_map) << "regen=" << regen;
|
||||
}
|
||||
opts_map.clear();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -506,6 +506,7 @@ TEST_MAIN_SOURCES = \
|
||||
db/db_clip_test.cc \
|
||||
db/db_dynamic_level_test.cc \
|
||||
db/db_encryption_test.cc \
|
||||
db/db_etc2_test.cc \
|
||||
db/db_etc3_test.cc \
|
||||
db/db_flush_test.cc \
|
||||
db/db_follower_test.cc \
|
||||
@@ -533,7 +534,6 @@ TEST_MAIN_SOURCES = \
|
||||
db/db_table_properties_test.cc \
|
||||
db/db_tailing_iter_test.cc \
|
||||
db/db_test.cc \
|
||||
db/db_test2.cc \
|
||||
db/db_logical_block_size_cache_test.cc \
|
||||
db/db_universal_compaction_test.cc \
|
||||
db/db_wal_test.cc \
|
||||
|
||||
@@ -1077,6 +1077,9 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
|
||||
multiscan_opts->io_coalesce_threshold;
|
||||
job->job_options.read_options = read_options_;
|
||||
job->job_options.read_options.async_io = multiscan_opts->use_async_io;
|
||||
// CollectBlockHandles walks sorted scan ranges through the table index, whose
|
||||
// key order matches data block offset order.
|
||||
job->job_options.block_handles_are_sorted = true;
|
||||
|
||||
std::shared_ptr<ReadSet> read_set;
|
||||
// IODispatcher should be provided by DBIter::Prepare() to enable sharing
|
||||
|
||||
@@ -67,6 +67,8 @@ scriptpath=`dirname ${BASH_SOURCE[0]}`
|
||||
|
||||
test_dir=${TEST_TMPDIR:-"/tmp"}"/rocksdb_format_compatible_$USER"
|
||||
rm -rf ${test_dir:?}
|
||||
# Also clean the noexec fallback dir from any prior run (see cs_test_dir below).
|
||||
rm -rf "$PWD/tmp/cs"
|
||||
|
||||
# Prevent 'make clean' etc. from wiping out test_dir
|
||||
export TEST_TMPDIR=$test_dir"/misc"
|
||||
@@ -88,6 +90,17 @@ mkdir -p $db_test_dir
|
||||
# For backup/restore test (uses DB test)
|
||||
bak_test_dir=$test_dir"/bak"
|
||||
mkdir -p $bak_test_dir
|
||||
# For remote compaction test. The saved ldb binary must be executable.
|
||||
# Detect noexec early and fall back to $PWD/tmp/cs (git-ignored).
|
||||
cs_test_dir=$test_dir"/cs"
|
||||
mkdir -p $cs_test_dir
|
||||
if cp /bin/true $cs_test_dir/_exec_test 2>/dev/null && \
|
||||
! $cs_test_dir/_exec_test 2>/dev/null; then
|
||||
echo " $cs_test_dir is noexec, using $PWD/tmp/cs instead"
|
||||
cs_test_dir="$PWD/tmp/cs"
|
||||
mkdir -p $cs_test_dir
|
||||
fi
|
||||
rm -f $cs_test_dir/_exec_test
|
||||
|
||||
python_bin=$(which python3 || which python || echo python3)
|
||||
|
||||
@@ -148,7 +161,7 @@ EOF
|
||||
|
||||
# To check for DB forward compatibility with loading options (old version
|
||||
# reading data from new), as well as backward compatibility
|
||||
declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb")
|
||||
declare -a db_forward_with_options_refs=("10.4.fb" "10.5.fb" "10.6.fb" "10.7.fb" "10.8.fb" "10.9.fb" "10.10.fb" "10.11.fb" "11.0.fb" "11.1.fb" "11.2.fb" "11.3.fb" "11.4.fb")
|
||||
# To check for DB forward compatibility without loading options (in addition
|
||||
# to the "with loading options" set), as well as backward compatibility
|
||||
declare -a db_forward_no_options_refs=() # N/A at the moment
|
||||
@@ -238,7 +251,7 @@ invoke_make()
|
||||
generate_db()
|
||||
{
|
||||
set +e
|
||||
[ "$SANITY_CHECK" ] || bash "$script_copy_dir"/generate_random_db.sh "$1" "$2"
|
||||
[ "$SANITY_CHECK" ] || bash "$script_copy_dir"/generate_random_db.sh "$1" "$2" "$3"
|
||||
if [ $? -ne 0 ]; then
|
||||
echo ==== Error loading data from $2 to $1 ====
|
||||
exit 1
|
||||
@@ -320,6 +333,40 @@ member_of_array()
|
||||
return 1
|
||||
}
|
||||
|
||||
# Run one cross-version remote compaction compatibility test.
|
||||
# Generates a DB using the primary's ldb (so its OPTIONS file matches the
|
||||
# primary's version), then runs primary and worker from potentially
|
||||
# different versions. They coordinate via files in test_dir (see
|
||||
# remote_compaction_primary/worker ldb commands for protocol details).
|
||||
# Tests the wire format of CompactionServiceInput/Result between versions.
|
||||
run_cs_compat_test()
|
||||
{
|
||||
local test_label="$1"
|
||||
local primary_ldb="$2"
|
||||
local worker_ldb="$3"
|
||||
local cs_run_dir="$4"
|
||||
|
||||
echo "== $test_label"
|
||||
rm -rf "$cs_run_dir" && mkdir -p "$cs_run_dir"
|
||||
generate_db $input_data_path "$cs_run_dir/db" "$primary_ldb"
|
||||
# Write keys spanning the full key range to create an L0 file that
|
||||
# overlaps with existing L0 files, ensuring CompactRange triggers a
|
||||
# real compaction (not a trivial move).
|
||||
printf "a ==> overlap_start\nzzzzzzz ==> overlap_end\n" | \
|
||||
$primary_ldb load --db="$cs_run_dir/db" --auto_compaction=false
|
||||
$worker_ldb remote_compaction_worker --db="$cs_run_dir/db" --job_dir="$cs_run_dir" &
|
||||
local worker_pid=$!
|
||||
local primary_exit=0
|
||||
$primary_ldb remote_compaction_primary --db="$cs_run_dir/db" --job_dir="$cs_run_dir" || primary_exit=$?
|
||||
local worker_exit=0
|
||||
wait $worker_pid || worker_exit=$?
|
||||
if [ $primary_exit -ne 0 ] || [ $worker_exit -ne 0 ]; then
|
||||
echo "==== Error running remote compaction: $test_label (primary=$primary_exit worker=$worker_exit) ===="
|
||||
kill $worker_pid 2>/dev/null || true
|
||||
exit 1
|
||||
fi
|
||||
}
|
||||
|
||||
force_no_fbcode()
|
||||
{
|
||||
# Not all branches recognize ROCKSDB_NO_FBCODE and we should not need
|
||||
@@ -339,8 +386,10 @@ force_no_fbcode()
|
||||
# * (Again) check out, build, and do (other) stuff with the "current"
|
||||
# branch, potentially using data from older branches.
|
||||
#
|
||||
# This way, we only do at most n+1 checkout+build steps, without the
|
||||
# need to stash away executables.
|
||||
# This way, we only do at most n+1 checkout+build steps. The one
|
||||
# exception is the remote compaction test, which saves a copy of the
|
||||
# current ldb binary before old-ref checkouts overwrite ./ldb, so both
|
||||
# versions can run simultaneously as primary and worker.
|
||||
|
||||
# Decorate name
|
||||
current_checkout_name="$current_checkout_name ($current_checkout_hash)"
|
||||
@@ -351,6 +400,33 @@ force_no_fbcode
|
||||
invoke_make clean
|
||||
DISABLE_WARNING_AS_ERROR=1 invoke_make ldb -j$J
|
||||
|
||||
# Save current ldb for cross-version remote compaction tests. The old-ref
|
||||
# checkout will overwrite ./ldb, so we save a copy now.
|
||||
save_cs_current_ldb()
|
||||
{
|
||||
cs_current_ldb=$cs_test_dir/current_ldb
|
||||
cp -f ./ldb $cs_current_ldb
|
||||
# Copy shared libs next to the binary for LD_LIBRARY_PATH.
|
||||
# Static builds have no .so files - cp fails harmlessly.
|
||||
cp -f ./librocksdb*.so* $cs_test_dir/ 2>/dev/null || true
|
||||
cs_current_ldb_cmd="env LD_LIBRARY_PATH=$cs_test_dir $cs_current_ldb"
|
||||
cs_ldb_output=$($cs_current_ldb_cmd --version 2>&1)
|
||||
cs_ldb_exit=$?
|
||||
return $cs_ldb_exit
|
||||
}
|
||||
echo "== Saving current ldb for remote compaction cross-version tests"
|
||||
cs_current_ldb_cmd=""
|
||||
if [ ! "$SANITY_CHECK" ]; then
|
||||
if ! save_cs_current_ldb; then
|
||||
echo "==== Error: saved current ldb cannot run from $cs_test_dir (exit=$cs_ldb_exit): $cs_ldb_output ===="
|
||||
exit 1
|
||||
fi
|
||||
# Smoke test before cross-version tests to catch integration failures early.
|
||||
run_cs_compat_test \
|
||||
"Remote compaction smoke test: current primary + current worker" \
|
||||
"$cs_current_ldb_cmd" "$cs_current_ldb_cmd" "$cs_test_dir/smoke"
|
||||
fi
|
||||
|
||||
echo "== Using $current_checkout_name, generate DB with extern SST and ingest"
|
||||
current_ext_test_dir=$ext_test_dir"/current"
|
||||
write_external_sst $input_data_path ${current_ext_test_dir}_pointless $current_ext_test_dir
|
||||
@@ -429,6 +505,30 @@ do
|
||||
restore_db $current_bak_test_dir $db_test_dir/$checkout_ref
|
||||
compare_db $db_test_dir/$checkout_ref $current_db_test_dir forward_${checkout_ref}_dump.txt 0
|
||||
fi
|
||||
|
||||
# Remote compaction format compatibility: test that primary and worker from
|
||||
# different versions can exchange CompactionServiceInput/Result.
|
||||
# Requires db_forward_with_options_refs (the old worker must be able to
|
||||
# read OPTIONS written by the current primary) and the remote compaction
|
||||
# related ldb commands.
|
||||
# Skipped in SANITY_CHECK mode (no ldb binary to test).
|
||||
if [ ! "$SANITY_CHECK" ] &&
|
||||
[ -n "$cs_current_ldb_cmd" ] &&
|
||||
member_of_array "$checkout_ref" "${db_forward_with_options_refs[@]}"
|
||||
then
|
||||
if ./ldb --help 2>&1 | grep -q remote_compaction_primary; then
|
||||
cs_old_ldb_cmd="./ldb"
|
||||
ref_dir=$cs_test_dir/$checkout_ref
|
||||
run_cs_compat_test \
|
||||
"Remote compaction compatibility: current primary + $checkout_ref worker" \
|
||||
"$cs_current_ldb_cmd" "$cs_old_ldb_cmd" "$ref_dir/test1"
|
||||
run_cs_compat_test \
|
||||
"Remote compaction compatibility: $checkout_ref primary + current worker" \
|
||||
"$cs_old_ldb_cmd" "$cs_current_ldb_cmd" "$ref_dir/test2"
|
||||
else
|
||||
echo " remote_compaction commands not available at $checkout_ref, skipping"
|
||||
fi
|
||||
fi
|
||||
done
|
||||
|
||||
echo "== Building $current_checkout_name debug (again, final)"
|
||||
@@ -466,6 +566,7 @@ do
|
||||
restore_db $bak_test_dir/$checkout_ref $db_test_dir/$checkout_ref
|
||||
compare_db $db_test_dir/$checkout_ref $current_db_test_dir db_dump.txt 1 0
|
||||
fi
|
||||
|
||||
done
|
||||
|
||||
if [ "$SANITY_CHECK" ]; then
|
||||
|
||||
+73
-4
@@ -219,6 +219,10 @@ DEFINE_string(
|
||||
"\txxhash -- repeated xxHash of <block size> data\n"
|
||||
"\txxhash64 -- repeated xxHash64 of <block size> data\n"
|
||||
"\txxh3 -- repeated XXH3 of <block size> data\n"
|
||||
"\tcompressreject -- repeated compression of <block size> data forced to "
|
||||
"be rejected (output buffer ~10% of input smaller than the predicted "
|
||||
"compressed size), to measure the cost of attempting then declining "
|
||||
"compression\n"
|
||||
"\tacquireload -- load N*1000 times\n"
|
||||
"\tfillseekseq -- write N values in sequential key, then read "
|
||||
"them by seeking to each key\n"
|
||||
@@ -3994,6 +3998,8 @@ class Benchmark {
|
||||
method = &Benchmark::AcquireLoad;
|
||||
} else if (name == "compress") {
|
||||
method = &Benchmark::Compress;
|
||||
} else if (name == "compressreject") {
|
||||
method = &Benchmark::CompressReject;
|
||||
} else if (name == "uncompress") {
|
||||
method = &Benchmark::Uncompress;
|
||||
} else if (name == "randomtransaction") {
|
||||
@@ -4506,8 +4512,8 @@ class Benchmark {
|
||||
auto working_area = compressor->ObtainWorkingArea();
|
||||
|
||||
GrowableBuffer compressed;
|
||||
// Compress 1G
|
||||
while (bytes < int64_t(1) << 30) {
|
||||
// Compress 3G
|
||||
while (bytes < int64_t(3) << 30) {
|
||||
compressed.ResetForSize(input.size());
|
||||
CompressionType actual_type = kNoCompression;
|
||||
s = compressor->CompressBlock(input, compressed.data(),
|
||||
@@ -4536,6 +4542,68 @@ class Benchmark {
|
||||
}
|
||||
}
|
||||
|
||||
// Like Compress, but forces every block's compression to be rejected by
|
||||
// giving the compressor an output buffer slightly smaller than the data is
|
||||
// expected to compress to. This exercises (and measures the speed of) the
|
||||
// path that attempts compression but falls back to storing data uncompressed,
|
||||
// as happens for incompressible data.
|
||||
void CompressReject(ThreadState* thread) {
|
||||
RandomGenerator gen;
|
||||
Slice input = gen.Generate(FLAGS_block_size);
|
||||
int64_t bytes = 0;
|
||||
Status s;
|
||||
|
||||
auto compressor = GetCompressor();
|
||||
if (!compressor) {
|
||||
thread->stats.AddMessage("(compression type not supported)");
|
||||
return;
|
||||
}
|
||||
auto working_area = compressor->ObtainWorkingArea();
|
||||
|
||||
// RandomGenerator produces data that compresses to about
|
||||
// FLAGS_compression_ratio of its original size (see CompressibleString),
|
||||
// which should be an accurate/optimistic predictor of the compressed size.
|
||||
// Make the output buffer 10% of the uncompressed size smaller than that
|
||||
// predicted compressed size, so the real output won't fit and compression
|
||||
// is rejected on every call. Clamp to at least 1 byte, which should always
|
||||
// reject.
|
||||
double reject_fraction = std::max(0.0, FLAGS_compression_ratio - 0.10);
|
||||
size_t reject_size = std::max<size_t>(
|
||||
1, static_cast<size_t>(reject_fraction * input.size()));
|
||||
|
||||
GrowableBuffer compressed;
|
||||
|
||||
// Compress 3G
|
||||
while (bytes < int64_t(3) << 30) {
|
||||
compressed.ResetForSize(reject_size);
|
||||
CompressionType actual_type = kNoCompression;
|
||||
s = compressor->CompressBlock(input, compressed.data(),
|
||||
&compressed.MutableSize(), &actual_type,
|
||||
&working_area);
|
||||
if (UNLIKELY(!s.ok())) {
|
||||
break;
|
||||
}
|
||||
if (UNLIKELY(actual_type != kNoCompression)) {
|
||||
s = Status::Aborted(
|
||||
"Compression unexpectedly not rejected; try a smaller "
|
||||
"compression_ratio");
|
||||
break;
|
||||
}
|
||||
bytes += input.size();
|
||||
thread->stats.FinishedOps(nullptr, nullptr, 1, kCompress);
|
||||
}
|
||||
|
||||
if (!s.ok()) {
|
||||
thread->stats.AddMessage("(compression failure: " + s.ToString() + ")");
|
||||
} else {
|
||||
char buf[340];
|
||||
snprintf(buf, sizeof(buf), "(rejected; output buffer %.1f%% of input)",
|
||||
(reject_size * 100.0) / input.size());
|
||||
thread->stats.AddMessage(buf);
|
||||
thread->stats.AddBytes(bytes);
|
||||
}
|
||||
}
|
||||
|
||||
void Uncompress(ThreadState* thread) {
|
||||
RandomGenerator gen;
|
||||
Slice input = gen.Generate(FLAGS_block_size);
|
||||
@@ -4568,8 +4636,9 @@ class Benchmark {
|
||||
actual_type);
|
||||
auto decomp_working_area = decompressor->ObtainWorkingArea(actual_type);
|
||||
|
||||
int64_t bytes = 0;
|
||||
while (bytes < 1024 * 1048576) {
|
||||
uint64_t bytes = 0;
|
||||
// Decompress 20GB
|
||||
while (bytes < uint64_t{20} * 1024U * 1048576U) {
|
||||
Decompressor::Args args;
|
||||
args.compression_type = actual_type;
|
||||
args.compressed_data = compressed.AsSlice();
|
||||
|
||||
+11
-10
@@ -1,28 +1,29 @@
|
||||
#!/usr/bin/env bash
|
||||
# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
|
||||
#
|
||||
# A shell script to load some pre generated data file to a DB using ldb tool
|
||||
# ./ldb needs to be avaible to be executed.
|
||||
# A shell script to load some pre generated data file to a DB using ldb tool.
|
||||
# <ldb_command> must be available to execute (defaults to ./ldb).
|
||||
#
|
||||
# Usage: <SCRIPT> <input_data_path> <DB Path>
|
||||
# Usage: <SCRIPT> <input_data_path> <DB Path> [<ldb_command>]
|
||||
|
||||
if [ "$#" -lt 2 ]; then
|
||||
echo "usage: $BASH_SOURCE <input_data_path> <DB Path>"
|
||||
echo "usage: $BASH_SOURCE <input_data_path> <DB Path> [<ldb_command>]"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
input_data_dir=$1
|
||||
db_dir=$2
|
||||
ldb_cmd=${3:-./ldb}
|
||||
rm -rf $db_dir
|
||||
|
||||
second_gen_compression_support=
|
||||
mixed_compression_support=
|
||||
# Support for `ldb --version` is a crude under-approximation for versions
|
||||
# supporting dictionary compression and algorithms including zstd and lz4
|
||||
if ./ldb --version 2>/dev/null >/dev/null; then
|
||||
if $ldb_cmd --version 2>/dev/null >/dev/null; then
|
||||
second_gen_compression_support=1
|
||||
|
||||
if ./ldb load --db=$db_dir --compression_type=mixed --create_if_missing \
|
||||
if $ldb_cmd load --db=$db_dir --compression_type=mixed --create_if_missing \
|
||||
< /dev/null 2>/dev/null >/dev/null; then
|
||||
mixed_compression_support=1
|
||||
fi
|
||||
@@ -31,7 +32,7 @@ fi
|
||||
|
||||
# Check if deleterange command is supported by grepping ldb --help
|
||||
deleterange_support=
|
||||
if ./ldb --help 2>&1 | grep -q deleterange; then
|
||||
if $ldb_cmd --help 2>&1 | grep -q deleterange; then
|
||||
deleterange_support=1
|
||||
fi
|
||||
|
||||
@@ -69,7 +70,7 @@ do
|
||||
else
|
||||
d_arg=""
|
||||
fi
|
||||
./ldb load --db=$db_dir --compression_type=$c $d_arg --bloom_bits=10 \
|
||||
$ldb_cmd load --db=$db_dir --compression_type=$c $d_arg --bloom_bits=10 \
|
||||
--auto_compaction=false --create_if_missing < $input_data_dir/$f
|
||||
|
||||
# Use md5sum of file to deterministically decide whether to add a range
|
||||
@@ -89,11 +90,11 @@ do
|
||||
end_key="${key}0"
|
||||
if [ "$deleterange_support" == "1" ]; then
|
||||
echo "== Deleting range [$key, $end_key) from $f"
|
||||
./ldb deleterange --db=$db_dir "$key" "$end_key"
|
||||
$ldb_cmd deleterange --db=$db_dir "$key" "$end_key"
|
||||
else
|
||||
# Fall back to point delete for equivalent logical contents
|
||||
echo "== Deleting key $key from $f"
|
||||
./ldb delete --db=$db_dir "$key"
|
||||
$ldb_cmd delete --db=$db_dir "$key"
|
||||
fi
|
||||
fi
|
||||
fi
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
#include "rocksdb/utilities/ldb_cmd.h"
|
||||
|
||||
#include <cstddef>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <ctime>
|
||||
#include <fstream>
|
||||
@@ -16,6 +17,7 @@
|
||||
#include <sstream>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
|
||||
#include "db/blob/blob_index.h"
|
||||
#include "db/db_impl/db_impl.h"
|
||||
@@ -434,6 +436,14 @@ LDBCommand* LDBCommand::SelectCommand(const ParsedParams& parsed_params) {
|
||||
return new CompactionProgressDumpCommand(parsed_params.cmd_params,
|
||||
parsed_params.option_map,
|
||||
parsed_params.flags);
|
||||
} else if (parsed_params.cmd == RemoteCompactionPrimaryCommand::Name()) {
|
||||
return new RemoteCompactionPrimaryCommand(parsed_params.cmd_params,
|
||||
parsed_params.option_map,
|
||||
parsed_params.flags);
|
||||
} else if (parsed_params.cmd == RemoteCompactionWorkerCommand::Name()) {
|
||||
return new RemoteCompactionWorkerCommand(parsed_params.cmd_params,
|
||||
parsed_params.option_map,
|
||||
parsed_params.flags);
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
@@ -5454,4 +5464,191 @@ void CompactionProgressDumpCommand::DoCommand() {
|
||||
DumpCompactionProgressFile(path_);
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
const std::string kInputFile = "input.bin";
|
||||
const std::string kResultFile = "result.bin";
|
||||
constexpr int kPollIntervalMs = 100;
|
||||
constexpr int kPollTimeoutMs = 120000;
|
||||
constexpr int kPollMaxAttempts = kPollTimeoutMs / kPollIntervalMs;
|
||||
|
||||
// Atomic write via tmp+rename so the polling reader never sees partial data.
|
||||
bool AtomicWriteStringToFile(const std::string& path, const std::string& data) {
|
||||
std::string tmp_path = path + ".tmp";
|
||||
std::ofstream ofs(tmp_path, std::ios::binary);
|
||||
if (!ofs) {
|
||||
return false;
|
||||
}
|
||||
ofs.write(data.data(), data.size());
|
||||
if (!ofs.good()) {
|
||||
return false;
|
||||
}
|
||||
ofs.close();
|
||||
return std::rename(tmp_path.c_str(), path.c_str()) == 0;
|
||||
}
|
||||
|
||||
bool PollForFile(Env* env, const std::string& path, std::string* data) {
|
||||
for (int i = 0; i < kPollMaxAttempts; i++) {
|
||||
Status s = ROCKSDB_NAMESPACE::ReadFileToString(env, path, data);
|
||||
if (s.ok() && !data->empty()) {
|
||||
return true;
|
||||
}
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kPollIntervalMs));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
class LocalFileCompactionService : public CompactionService {
|
||||
public:
|
||||
explicit LocalFileCompactionService(const std::string& job_dir)
|
||||
: job_dir_(job_dir) {}
|
||||
|
||||
const char* Name() const override { return "LocalFileCompactionService"; }
|
||||
|
||||
CompactionServiceScheduleResponse Schedule(
|
||||
const CompactionServiceJobInfo& /*info*/,
|
||||
const std::string& compaction_service_input) override {
|
||||
assert(!scheduled_);
|
||||
scheduled_ = true;
|
||||
if (!AtomicWriteStringToFile(job_dir_ + "/" + kInputFile,
|
||||
compaction_service_input)) {
|
||||
return CompactionServiceScheduleResponse(
|
||||
CompactionServiceJobStatus::kFailure);
|
||||
}
|
||||
return CompactionServiceScheduleResponse(
|
||||
"job_0001", CompactionServiceJobStatus::kSuccess);
|
||||
}
|
||||
|
||||
CompactionServiceJobStatus Wait(const std::string& /*scheduled_job_id*/,
|
||||
std::string* result) override {
|
||||
if (!PollForFile(Env::Default(), job_dir_ + "/" + kResultFile, result)) {
|
||||
return CompactionServiceJobStatus::kFailure;
|
||||
}
|
||||
return CompactionServiceJobStatus::kSuccess;
|
||||
}
|
||||
|
||||
private:
|
||||
std::string job_dir_;
|
||||
bool scheduled_ = false;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
const std::string RemoteCompactionPrimaryCommand::ARG_JOB_DIR =
|
||||
"job_dir"; // NOLINT(cert-err58-cpp)
|
||||
|
||||
RemoteCompactionPrimaryCommand::RemoteCompactionPrimaryCommand(
|
||||
const std::vector<std::string>& /*params*/,
|
||||
const std::map<std::string, std::string>& options,
|
||||
const std::vector<std::string>& flags)
|
||||
: LDBCommand(options, flags, false, BuildCmdLineOptions({ARG_JOB_DIR})) {
|
||||
auto it = options.find(ARG_JOB_DIR);
|
||||
if (it != options.end()) {
|
||||
job_dir_ = it->second;
|
||||
} else {
|
||||
exec_state_ = LDBCommandExecuteResult::Failed("--job_dir is required");
|
||||
}
|
||||
}
|
||||
|
||||
void RemoteCompactionPrimaryCommand::Help(std::string& ret) {
|
||||
ret.append(" ");
|
||||
ret.append(Name());
|
||||
ret.append(" --db=<db_path> --job_dir=<dir>");
|
||||
ret.append("\n Open an existing DB and run CompactRange() through ");
|
||||
ret.append("LocalFileCompactionService.\n");
|
||||
ret.append(" Schedule() writes job_dir/input.bin (once). ");
|
||||
ret.append("Wait() polls for job_dir/result.bin.\n");
|
||||
ret.append(" Must run with remote_compaction_worker in background on ");
|
||||
ret.append("the same --db and --job_dir,\n");
|
||||
ret.append(" otherwise Wait() will hang until timeout.\n");
|
||||
}
|
||||
|
||||
void RemoteCompactionPrimaryCommand::DoCommand() {
|
||||
auto cs = std::make_shared<LocalFileCompactionService>(job_dir_);
|
||||
|
||||
Options options;
|
||||
options.compaction_service = cs;
|
||||
options.disable_auto_compactions = true;
|
||||
|
||||
std::unique_ptr<DB> db;
|
||||
Status s = DB::Open(options, db_path_, &db);
|
||||
if (!s.ok()) {
|
||||
exec_state_ = LDBCommandExecuteResult::Failed("Open: " + s.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
s = db->CompactRange(CompactRangeOptions(), nullptr, nullptr);
|
||||
if (!s.ok()) {
|
||||
exec_state_ =
|
||||
LDBCommandExecuteResult::Failed("CompactRange: " + s.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
fprintf(stdout, "remote_compaction_primary: OK\n");
|
||||
}
|
||||
|
||||
const std::string RemoteCompactionWorkerCommand::ARG_JOB_DIR =
|
||||
"job_dir"; // NOLINT(cert-err58-cpp)
|
||||
|
||||
RemoteCompactionWorkerCommand::RemoteCompactionWorkerCommand(
|
||||
const std::vector<std::string>& /*params*/,
|
||||
const std::map<std::string, std::string>& options,
|
||||
const std::vector<std::string>& flags)
|
||||
: LDBCommand(options, flags, false, BuildCmdLineOptions({ARG_JOB_DIR})) {
|
||||
auto it = options.find(ARG_JOB_DIR);
|
||||
if (it != options.end()) {
|
||||
job_dir_ = it->second;
|
||||
} else {
|
||||
exec_state_ = LDBCommandExecuteResult::Failed("--job_dir is required");
|
||||
}
|
||||
}
|
||||
|
||||
void RemoteCompactionWorkerCommand::Help(std::string& ret) {
|
||||
ret.append(" ");
|
||||
ret.append(Name());
|
||||
ret.append(" --db=<db_path> --job_dir=<dir>");
|
||||
ret.append("\n Read job_dir/input.bin, run DB::OpenAndCompact(), ");
|
||||
ret.append("write job_dir/result.bin.\n");
|
||||
}
|
||||
|
||||
void RemoteCompactionWorkerCommand::DoCommand() {
|
||||
std::string input_path = job_dir_ + "/" + kInputFile;
|
||||
std::string input;
|
||||
if (!PollForFile(Env::Default(), input_path, &input)) {
|
||||
exec_state_ = LDBCommandExecuteResult::Failed(
|
||||
"Timed out (" + std::to_string(kPollTimeoutMs / 1000) +
|
||||
"s) waiting for " + input_path);
|
||||
return;
|
||||
}
|
||||
|
||||
std::string output_dir = job_dir_ + "/output";
|
||||
Status dir_s = Env::Default()->CreateDirIfMissing(output_dir);
|
||||
if (!dir_s.ok()) {
|
||||
exec_state_ = LDBCommandExecuteResult::Failed("CreateDirIfMissing: " +
|
||||
dir_s.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
std::string result;
|
||||
CompactionServiceOptionsOverride override_options;
|
||||
override_options.table_factory.reset(NewBlockBasedTableFactory());
|
||||
Status s = DB::OpenAndCompact(OpenAndCompactOptions(), db_path_, output_dir,
|
||||
input, &result, override_options);
|
||||
if (!s.ok()) {
|
||||
exec_state_ =
|
||||
LDBCommandExecuteResult::Failed("OpenAndCompact: " + s.ToString());
|
||||
return;
|
||||
}
|
||||
|
||||
std::string result_path = job_dir_ + "/" + kResultFile;
|
||||
if (!AtomicWriteStringToFile(result_path, result)) {
|
||||
exec_state_ =
|
||||
LDBCommandExecuteResult::Failed("Failed to write " + result_path);
|
||||
return;
|
||||
}
|
||||
|
||||
fprintf(stdout, "remote_compaction_worker: OK (%zu bytes result)\n",
|
||||
result.size());
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -835,4 +835,36 @@ class CompactionProgressDumpCommand : public LDBCommand {
|
||||
static const std::string ARG_PATH;
|
||||
};
|
||||
|
||||
class RemoteCompactionPrimaryCommand : public LDBCommand {
|
||||
public:
|
||||
static std::string Name() { return "remote_compaction_primary"; }
|
||||
RemoteCompactionPrimaryCommand(
|
||||
const std::vector<std::string>& params,
|
||||
const std::map<std::string, std::string>& options,
|
||||
const std::vector<std::string>& flags);
|
||||
void DoCommand() override;
|
||||
bool NoDBOpen() override { return true; }
|
||||
static void Help(std::string& ret);
|
||||
|
||||
private:
|
||||
std::string job_dir_;
|
||||
static const std::string ARG_JOB_DIR;
|
||||
};
|
||||
|
||||
class RemoteCompactionWorkerCommand : public LDBCommand {
|
||||
public:
|
||||
static std::string Name() { return "remote_compaction_worker"; }
|
||||
RemoteCompactionWorkerCommand(
|
||||
const std::vector<std::string>& params,
|
||||
const std::map<std::string, std::string>& options,
|
||||
const std::vector<std::string>& flags);
|
||||
void DoCommand() override;
|
||||
bool NoDBOpen() override { return true; }
|
||||
static void Help(std::string& ret);
|
||||
|
||||
private:
|
||||
std::string job_dir_;
|
||||
static const std::string ARG_JOB_DIR;
|
||||
};
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
@@ -145,6 +145,8 @@ void LDBCommandRunner::PrintHelp(const LDBOptions& ldb_options,
|
||||
WriteExternalSstFilesCommand::Help(ret);
|
||||
IngestExternalSstFilesCommand::Help(ret);
|
||||
UnsafeRemoveSstFileCommand::Help(ret);
|
||||
RemoteCompactionPrimaryCommand::Help(ret);
|
||||
RemoteCompactionWorkerCommand::Help(ret);
|
||||
|
||||
fprintf(to_stderr ? stderr : stdout, "%s\n", ret.c_str());
|
||||
}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
The default `compression` for column families is now `kLZ4Compression` rather than `kSnappyCompression`. This affects only column families that do not explicitly set `compression`, and only newly-written SST files. The change is fully compatible: existing data remains readable with no migration needed, as RocksDB selects the decompressor per block. LZ4 offers slightly better compression ratios and decompression CPU efficiency vs. Snappy and similar compression CPU, tested across various server CPUs. When support is not compiled in, the fallback path for default compression is LZ4 -> Snappy -> NoCompression.
|
||||
-1
@@ -1 +0,0 @@
|
||||
Remote compaction now falls back to local compaction when the primary cannot deserialize a successful `CompactionServiceResult` before installing any remote output files.
|
||||
@@ -1 +0,0 @@
|
||||
Fixed WritePrepared TransactionDB cleanup after retryable commit or rollback write failures so prepared transactions remain rollbackable instead of leaving unresolved prepared state.
|
||||
@@ -1 +0,0 @@
|
||||
Fix a rare corruption bug for tiered compaction that incorrectly moved last level range tombstones into proximal level. This corruption error is surfaced when force_consistency_checks is enabled.
|
||||
@@ -1 +0,0 @@
|
||||
Added `rocksdb_options_set_memtable_batch_lookup_optimization()` and `rocksdb_options_get_memtable_batch_lookup_optimization()` to the C API, exposing the existing `AdvancedColumnFamilyOptions::memtable_batch_lookup_optimization` field. This allows C API users (and downstream language bindings) to enable the skip-list memtable's batch-lookup optimization for `MultiGet`, which caches the search path between consecutive keys and reduces per-key cost from O(log N) to O(log d) where d is the distance between consecutive keys.
|
||||
@@ -1 +0,0 @@
|
||||
Added `rocksdb_readoptions_set_optimize_multiget_for_io()` and `rocksdb_readoptions_get_optimize_multiget_for_io()` to the C API, exposing the existing `ReadOptions::optimize_multiget_for_io` field. This allows C API users (and downstream language bindings) to opt out of the multi-level parallel MultiGet path, which only takes effect when the library is built with `USE_COROUTINES`.
|
||||
@@ -1 +0,0 @@
|
||||
Added `rocksdb_block_based_table_index_block_search_type_auto` enum constant and the `rocksdb_block_based_options_set_uniform_cv_threshold()` setter to the C API. The constant exposes `BlockSearchType::kAuto`, and the setter exposes `BlockBasedTableOptions::uniform_cv_threshold`. Both are required for `kAuto` index-block search to take effect from the C API: without setting `uniform_cv_threshold >= 0`, the per-block `is_uniform` footer bit is never written, and `kAuto` falls back to binary search at read time.
|
||||
@@ -1 +0,0 @@
|
||||
Added `EventListener::OnDBShutdownBegin`, a callback that fires once when a DB begins shutdown, including when RocksDB cleans up a failed `DB::Open()` attempt.
|
||||
@@ -1 +0,0 @@
|
||||
Added a new PerfContext counter `blob_cache_read_byte` for blob cache bytes read.
|
||||
@@ -421,6 +421,12 @@ inline bool LZ4_Supported() {
|
||||
#endif
|
||||
}
|
||||
|
||||
inline CompressionType GetDefaultCompressionType() {
|
||||
return LZ4_Supported()
|
||||
? kLZ4Compression
|
||||
: (Snappy_Supported() ? kSnappyCompression : kNoCompression);
|
||||
}
|
||||
|
||||
inline bool XPRESS_Supported() {
|
||||
#ifdef XPRESS
|
||||
return true;
|
||||
|
||||
@@ -521,7 +521,7 @@ class PresetCompressionDictTest
|
||||
public testing::WithParamInterface<std::tuple<CompressionType, bool>> {
|
||||
public:
|
||||
PresetCompressionDictTest()
|
||||
: DBTestBase("db_test2", false /* env_do_fsync */),
|
||||
: DBTestBase("compression_test_preset_dict", false /* env_do_fsync */),
|
||||
compression_type_(std::get<0>(GetParam())),
|
||||
bottommost_(std::get<1>(GetParam())) {}
|
||||
|
||||
|
||||
+38
-23
@@ -44,6 +44,25 @@ struct IODispatcherImplData {
|
||||
virtual void ReleaseMemory(size_t bytes) = 0;
|
||||
};
|
||||
|
||||
#ifndef NDEBUG
|
||||
static bool BlockIndicesAreSortedByOffset(
|
||||
const std::vector<BlockHandle>& block_handles,
|
||||
const std::vector<size_t>& block_indices) {
|
||||
uint64_t prev_offset = 0;
|
||||
for (size_t i = 0; i < block_indices.size(); ++i) {
|
||||
if (block_indices[i] >= block_handles.size()) {
|
||||
return false;
|
||||
}
|
||||
const uint64_t current_offset = block_handles[block_indices[i]].offset();
|
||||
if (i > 0 && current_offset < prev_offset) {
|
||||
return false;
|
||||
}
|
||||
prev_offset = current_offset;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
#endif // NDEBUG
|
||||
|
||||
// Helper function to create and pin a block from a buffer
|
||||
// Used by both ReadSet::PollAndProcessAsyncIO and IODispatcherImpl::Impl
|
||||
static Status CreateAndPinBlockFromBuffer(
|
||||
@@ -479,7 +498,8 @@ struct IODispatcherImpl::Impl : public IODispatcherImplData,
|
||||
const std::vector<size_t>& block_indices);
|
||||
|
||||
// Pre-coalesce blocks into groups, respecting max_group_bytes size limit.
|
||||
// Returns groups ordered by first block index (earlier blocks first).
|
||||
// block_indices must be sorted by block offset.
|
||||
// Returns groups ordered by block offset.
|
||||
std::vector<CoalescedPrefetchGroup> PreCoalesceBlocks(
|
||||
const std::shared_ptr<IOJob>& job, const std::shared_ptr<ReadSet>& rs,
|
||||
const std::vector<size_t>& block_indices, size_t max_group_bytes);
|
||||
@@ -697,16 +717,22 @@ Status IODispatcherImpl::Impl::SubmitJob(const std::shared_ptr<IOJob>& job,
|
||||
for (size_t i = 0; i < job->block_handles.size(); ++i) {
|
||||
rs->sorted_block_indices_[i] = i;
|
||||
}
|
||||
std::sort(rs->sorted_block_indices_.begin(), rs->sorted_block_indices_.end(),
|
||||
[&job](size_t a, size_t b) {
|
||||
return job->block_handles[a].offset() <
|
||||
job->block_handles[b].offset();
|
||||
});
|
||||
if (!job->job_options.block_handles_are_sorted) {
|
||||
TEST_SYNC_POINT("IODispatcherImpl::SubmitJob:SortBlockHandles");
|
||||
std::sort(rs->sorted_block_indices_.begin(),
|
||||
rs->sorted_block_indices_.end(), [&job](size_t a, size_t b) {
|
||||
return job->block_handles[a].offset() <
|
||||
job->block_handles[b].offset();
|
||||
});
|
||||
}
|
||||
|
||||
// Step 1: Check cache and pin cached blocks
|
||||
// Step 1: Check cache and pin cached blocks. Iterate in offset order so all
|
||||
// downstream private helpers can assume block index vectors are already
|
||||
// sorted.
|
||||
std::vector<size_t> block_indices_to_read;
|
||||
block_indices_to_read.reserve(job->block_handles.size());
|
||||
|
||||
for (size_t i = 0; i < job->block_handles.size(); ++i) {
|
||||
for (size_t i : rs->sorted_block_indices_) {
|
||||
const auto& data_block_handle = job->block_handles[i];
|
||||
|
||||
// Lookup and pin block in cache
|
||||
@@ -816,17 +842,11 @@ void IODispatcherImpl::Impl::PrepareIORequests(
|
||||
const std::vector<BlockHandle>& block_handles,
|
||||
std::vector<FSReadRequest>* read_reqs,
|
||||
std::vector<std::vector<size_t>>* coalesced_block_indices) {
|
||||
// This is necessary because block handles may not be in sorted order
|
||||
std::vector<size_t> sorted_block_indices = block_indices_to_read;
|
||||
std::sort(sorted_block_indices.begin(), sorted_block_indices.end(),
|
||||
[&block_handles](size_t a, size_t b) {
|
||||
return block_handles[a].offset() < block_handles[b].offset();
|
||||
});
|
||||
|
||||
assert(BlockIndicesAreSortedByOffset(block_handles, block_indices_to_read));
|
||||
assert(coalesced_block_indices->empty());
|
||||
coalesced_block_indices->resize(1);
|
||||
|
||||
for (const auto& block_idx : sorted_block_indices) {
|
||||
for (const auto& block_idx : block_indices_to_read) {
|
||||
if (!coalesced_block_indices->back().empty()) {
|
||||
// Check if we can coalesce with previous block
|
||||
const auto& last_block_handle =
|
||||
@@ -883,17 +903,12 @@ std::vector<CoalescedPrefetchGroup> IODispatcherImpl::Impl::PreCoalesceBlocks(
|
||||
const auto& block_handles = job->block_handles;
|
||||
const uint64_t coalesce_threshold = job->job_options.io_coalesce_threshold;
|
||||
|
||||
// Sort block indices by offset for coalescing
|
||||
std::vector<size_t> sorted_indices = block_indices;
|
||||
std::sort(sorted_indices.begin(), sorted_indices.end(),
|
||||
[&block_handles](size_t a, size_t b) {
|
||||
return block_handles[a].offset() < block_handles[b].offset();
|
||||
});
|
||||
assert(BlockIndicesAreSortedByOffset(block_handles, block_indices));
|
||||
|
||||
// Build coalesced groups respecting max_group_bytes
|
||||
groups.emplace_back();
|
||||
|
||||
for (size_t idx : sorted_indices) {
|
||||
for (size_t idx : block_indices) {
|
||||
size_t block_size = rs->block_sizes_[idx];
|
||||
|
||||
// Skip blocks that are individually larger than the memory budget
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
|
||||
#include "rocksdb/io_dispatcher.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
@@ -836,6 +837,85 @@ TEST_F(IODispatcherTest, VerifyCoalescing) {
|
||||
}
|
||||
}
|
||||
|
||||
TEST_F(IODispatcherTest, SortedBlockHandlesSkipDispatcherSorts) {
|
||||
struct SortCounters {
|
||||
int submit_job = 0;
|
||||
};
|
||||
|
||||
auto install_callbacks = [](SortCounters* counters) {
|
||||
SyncPoint::GetInstance()->SetCallBack(
|
||||
"IODispatcherImpl::SubmitJob:SortBlockHandles",
|
||||
[counters](void*) { ++counters->submit_job; });
|
||||
SyncPoint::GetInstance()->EnableProcessing();
|
||||
};
|
||||
|
||||
auto submit_job = [](IODispatcher* dispatcher, BlockBasedTable* table,
|
||||
const std::vector<BlockHandle>& block_handles,
|
||||
bool sorted) {
|
||||
auto job = std::make_shared<IOJob>();
|
||||
job->block_handles = block_handles;
|
||||
job->table = table;
|
||||
job->job_options.block_handles_are_sorted = sorted;
|
||||
job->job_options.read_options.async_io = false;
|
||||
job->job_options.io_coalesce_threshold = 1024 * 1024;
|
||||
|
||||
std::shared_ptr<ReadSet> read_set;
|
||||
ASSERT_OK(dispatcher->SubmitJob(job, &read_set));
|
||||
ASSERT_NE(read_set, nullptr);
|
||||
for (size_t i = 0; i < block_handles.size(); ++i) {
|
||||
CachableEntry<Block> block;
|
||||
ASSERT_OK(read_set->ReadIndex(i, &block));
|
||||
ASSERT_NE(block.GetValue(), nullptr);
|
||||
ASSERT_GT(block.GetValue()->size(), 0);
|
||||
|
||||
InternalKeyComparator internal_comparator(BytewiseComparator());
|
||||
std::unique_ptr<DataBlockIter> iter(block.GetValue()->NewDataIterator(
|
||||
internal_comparator.user_comparator(), kDisableGlobalSequenceNumber));
|
||||
iter->SeekToFirst();
|
||||
ASSERT_TRUE(iter->Valid());
|
||||
|
||||
ParsedInternalKey parsed_key;
|
||||
ASSERT_OK(
|
||||
ParseInternalKey(iter->key(), &parsed_key, true /* log_err_key */));
|
||||
ASSERT_TRUE(parsed_key.user_key.starts_with("key"));
|
||||
ASSERT_OK(iter->status());
|
||||
}
|
||||
};
|
||||
|
||||
std::unique_ptr<BlockBasedTable> sorted_table;
|
||||
std::vector<BlockHandle> sorted_handles;
|
||||
ASSERT_OK(CreateAndOpenSST(10, &sorted_table, &sorted_handles));
|
||||
ASSERT_GE(sorted_handles.size(), 5);
|
||||
sorted_handles.resize(5);
|
||||
|
||||
SortCounters sorted_counters;
|
||||
install_callbacks(&sorted_counters);
|
||||
std::unique_ptr<IODispatcher> sorted_dispatcher(NewIODispatcher());
|
||||
submit_job(sorted_dispatcher.get(), sorted_table.get(), sorted_handles,
|
||||
true /* sorted */);
|
||||
SyncPoint::GetInstance()->DisableProcessing();
|
||||
SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||
|
||||
EXPECT_EQ(sorted_counters.submit_job, 0);
|
||||
|
||||
std::unique_ptr<BlockBasedTable> unsorted_table;
|
||||
std::vector<BlockHandle> unsorted_handles;
|
||||
ASSERT_OK(CreateAndOpenSST(10, &unsorted_table, &unsorted_handles));
|
||||
ASSERT_GE(unsorted_handles.size(), 5);
|
||||
unsorted_handles.resize(5);
|
||||
std::reverse(unsorted_handles.begin(), unsorted_handles.end());
|
||||
|
||||
SortCounters unsorted_counters;
|
||||
install_callbacks(&unsorted_counters);
|
||||
std::unique_ptr<IODispatcher> unsorted_dispatcher(NewIODispatcher());
|
||||
submit_job(unsorted_dispatcher.get(), unsorted_table.get(), unsorted_handles,
|
||||
false /* sorted */);
|
||||
SyncPoint::GetInstance()->DisableProcessing();
|
||||
SyncPoint::GetInstance()->ClearAllCallBacks();
|
||||
|
||||
EXPECT_EQ(unsorted_counters.submit_job, 1);
|
||||
}
|
||||
|
||||
// Test that verifies the read request offsets and lengths match the
|
||||
// expected block handles.
|
||||
TEST_F(IODispatcherTest, VerifyReadRequestDetails) {
|
||||
|
||||
+11
-10
@@ -251,21 +251,22 @@ std::string UnescapeOptionString(const std::string& escaped_string) {
|
||||
}
|
||||
|
||||
std::string trim(const std::string& str) {
|
||||
if (str.empty()) {
|
||||
return std::string();
|
||||
}
|
||||
size_t start = 0;
|
||||
size_t end = str.size() - 1;
|
||||
while (isspace(str[start]) != 0 && start < end) {
|
||||
size_t end = str.size();
|
||||
for (;;) {
|
||||
if (start >= end) {
|
||||
return {}; // all whitespace
|
||||
}
|
||||
if (!isspace(str[start])) {
|
||||
break; // first non-whitespace found
|
||||
}
|
||||
++start;
|
||||
}
|
||||
while (isspace(str[end]) != 0 && start < end) {
|
||||
while (isspace(str[end - 1]) != 0) {
|
||||
--end;
|
||||
assert(end > start);
|
||||
}
|
||||
if (start <= end) {
|
||||
return str.substr(start, end - start + 1);
|
||||
}
|
||||
return std::string();
|
||||
return str.substr(start, end - start);
|
||||
}
|
||||
|
||||
bool EndsWith(const std::string& string, const std::string& pattern) {
|
||||
|
||||
@@ -27,6 +27,31 @@ TEST(StringUtilTest, NumberToHumanString) {
|
||||
ASSERT_EQ("-10G", NumberToHumanString(-10000000000));
|
||||
}
|
||||
|
||||
TEST(StringUtilTest, Trim) {
|
||||
// Empty input.
|
||||
EXPECT_EQ("", trim(""));
|
||||
// No whitespace to strip.
|
||||
EXPECT_EQ("a", trim("a"));
|
||||
EXPECT_EQ("abc", trim("abc"));
|
||||
// Leading whitespace only.
|
||||
EXPECT_EQ("a", trim(" a"));
|
||||
EXPECT_EQ("abc", trim(" abc"));
|
||||
// Trailing whitespace only.
|
||||
EXPECT_EQ("a", trim("a "));
|
||||
EXPECT_EQ("abc", trim("abc "));
|
||||
// Both ends.
|
||||
EXPECT_EQ("a", trim(" a "));
|
||||
EXPECT_EQ("abc", trim(" abc "));
|
||||
EXPECT_EQ("a b c", trim(" a b c ")); // interior whitespace preserved
|
||||
// All-whitespace inputs of various lengths must trim to empty.
|
||||
EXPECT_EQ("", trim(" "));
|
||||
EXPECT_EQ("", trim(" "));
|
||||
EXPECT_EQ("", trim(" "));
|
||||
EXPECT_EQ("", trim("\t"));
|
||||
EXPECT_EQ("", trim("\n"));
|
||||
EXPECT_EQ("", trim(" \t\n\r"));
|
||||
}
|
||||
|
||||
} // namespace ROCKSDB_NAMESPACE
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
||||
Reference in New Issue
Block a user