mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1266de2a27 | |||
| 9402a77a37 | |||
| 9d9014d4d0 | |||
| 412b7fd2bd | |||
| 8a201ce68a | |||
| a53094983b | |||
| cc20520d5f | |||
| 36797a43fe | |||
| 7c06a1e503 | |||
| d9e7f6a3b9 | |||
| bf61a59a5b | |||
| ec2c2921e6 |
+37
@@ -1,6 +1,43 @@
|
|||||||
# Rocksdb Change Log
|
# Rocksdb Change Log
|
||||||
> NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
|
> NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
|
||||||
|
|
||||||
|
## 8.11.99 (02/21/2024)
|
||||||
|
### Behavior Changes
|
||||||
|
* Merge writes will only keep merge operand count within `ColumnFamilyOptions::max_successive_merges` when the key's merge operands are all found in memory, unless `strict_max_successive_merges` is explicitly set.
|
||||||
|
|
||||||
|
## 8.11.2 (02/16/2024)
|
||||||
|
* Update zlib to 1.3.1 for Java builds
|
||||||
|
|
||||||
|
## 8.11.1 (01/25/2024)
|
||||||
|
### Bug Fixes
|
||||||
|
* Fix a bug where older data of an ingested key can be returned for read when universal compaction is used
|
||||||
|
* Apply appropriate rate limiting and priorities in more places.
|
||||||
|
|
||||||
|
## 8.11.0 (01/19/2024)
|
||||||
|
### New Features
|
||||||
|
* Add new statistics: `rocksdb.sst.write.micros` measures time of each write to SST file; `rocksdb.file.write.{flush|compaction|db.open}.micros` measure time of each write to SST table (currently only block-based table format) and blob file for flush, compaction and db open.
|
||||||
|
|
||||||
|
### Public API Changes
|
||||||
|
* Added another enumerator `kVerify` to enum class `FileOperationType` in listener.h. Update your `switch` statements as needed.
|
||||||
|
* Add CompressionOptions to the CompressedSecondaryCacheOptions structure to allow users to specify library specific options when creating the compressed secondary cache.
|
||||||
|
* Deprecated several options: `level_compaction_dynamic_file_size`, `ignore_max_compaction_bytes_for_input`, `check_flush_compaction_key_order`, `flush_verify_memtable_count`, `compaction_verify_record_count`, `fail_if_options_file_error`, and `enforce_single_del_contracts`
|
||||||
|
* Exposed options ttl via c api.
|
||||||
|
|
||||||
|
### Behavior Changes
|
||||||
|
* `rocksdb.blobdb.blob.file.write.micros` expands to also measure time writing the header and footer. Therefore the COUNT may be higher and values may be smaller than before. For stacked BlobDB, it no longer measures the time of explictly flushing blob file.
|
||||||
|
* Files will be compacted to the next level if the data age exceeds periodic_compaction_seconds except for the last level.
|
||||||
|
* Reduced the compaction debt ratio trigger for scheduling parallel compactions
|
||||||
|
* For leveled compaction with default compaction pri (kMinOverlappingRatio), files marked for compaction will be prioritized over files not marked when picking a file from a level for compaction.
|
||||||
|
|
||||||
|
### Bug Fixes
|
||||||
|
* Fix bug in auto_readahead_size that combined with IndexType::kBinarySearchWithFirstKey + fails or iterator lands at a wrong key
|
||||||
|
* Fixed some cases in which DB file corruption was detected but ignored on creating a backup with BackupEngine.
|
||||||
|
* Fix bugs where `rocksdb.blobdb.blob.file.synced` includes blob files failed to get synced and `rocksdb.blobdb.blob.file.bytes.written` includes blob bytes failed to get written.
|
||||||
|
* Fixed a possible memory leak or crash on a failure (such as I/O error) in automatic atomic flush of multiple column families.
|
||||||
|
* Fixed some cases of in-memory data corruption using mmap reads with `BackupEngine`, `sst_dump`, or `ldb`.
|
||||||
|
* Fixed issues with experimental `preclude_last_level_data_seconds` option that could interfere with expected data tiering.
|
||||||
|
* Fixed the handling of the edge case when all existing blob files become unreferenced. Such files are now correctly deleted.
|
||||||
|
|
||||||
## 8.10.0 (12/15/2023)
|
## 8.10.0 (12/15/2023)
|
||||||
### New Features
|
### New Features
|
||||||
* Provide support for async_io to trim readahead_size by doing block cache lookup
|
* Provide support for async_io to trim readahead_size by doing block cache lookup
|
||||||
|
|||||||
@@ -2097,8 +2097,8 @@ ROCKSDB_JAVADOCS_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-javadoc.jar
|
|||||||
ROCKSDB_SOURCES_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-sources.jar
|
ROCKSDB_SOURCES_JAR = rocksdbjni-$(ROCKSDB_JAVA_VERSION)-sources.jar
|
||||||
SHA256_CMD = sha256sum
|
SHA256_CMD = sha256sum
|
||||||
|
|
||||||
ZLIB_VER ?= 1.3
|
ZLIB_VER ?= 1.3.1
|
||||||
ZLIB_SHA256 ?= ff0ba4c292013dbc27530b3a81e1f9a813cd39de01ca5e0f8bf355702efa593e
|
ZLIB_SHA256 ?= 9a93b2b7dfdac77ceba5a558a580e74667dd6fede4585b91eefb60f03b72df23
|
||||||
ZLIB_DOWNLOAD_BASE ?= http://zlib.net
|
ZLIB_DOWNLOAD_BASE ?= http://zlib.net
|
||||||
BZIP2_VER ?= 1.0.8
|
BZIP2_VER ?= 1.0.8
|
||||||
BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269
|
BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269
|
||||||
|
|||||||
@@ -484,7 +484,8 @@ void CompactionJob::GenSubcompactionBoundaries() {
|
|||||||
// overlap with N-1 other ranges. Since we requested a relatively large number
|
// overlap with N-1 other ranges. Since we requested a relatively large number
|
||||||
// (128) of ranges from each input files, even N range overlapping would
|
// (128) of ranges from each input files, even N range overlapping would
|
||||||
// cause relatively small inaccuracy.
|
// cause relatively small inaccuracy.
|
||||||
const ReadOptions read_options(Env::IOActivity::kCompaction);
|
ReadOptions read_options(Env::IOActivity::kCompaction);
|
||||||
|
read_options.rate_limiter_priority = GetRateLimiterPriority();
|
||||||
auto* c = compact_->compaction;
|
auto* c = compact_->compaction;
|
||||||
if (c->max_subcompactions() <= 1 &&
|
if (c->max_subcompactions() <= 1 &&
|
||||||
!(c->immutable_options()->compaction_pri == kRoundRobin &&
|
!(c->immutable_options()->compaction_pri == kRoundRobin &&
|
||||||
@@ -736,8 +737,9 @@ Status CompactionJob::Run() {
|
|||||||
// use_direct_io_for_flush_and_compaction is true, we will regard this
|
// use_direct_io_for_flush_and_compaction is true, we will regard this
|
||||||
// verification as user reads since the goal is to cache it here for
|
// verification as user reads since the goal is to cache it here for
|
||||||
// further user reads
|
// further user reads
|
||||||
const ReadOptions verify_table_read_options(
|
ReadOptions verify_table_read_options(Env::IOActivity::kCompaction);
|
||||||
Env::IOActivity::kCompaction);
|
verify_table_read_options.rate_limiter_priority =
|
||||||
|
GetRateLimiterPriority();
|
||||||
InternalIterator* iter = cfd->table_cache()->NewIterator(
|
InternalIterator* iter = cfd->table_cache()->NewIterator(
|
||||||
verify_table_read_options, file_options_,
|
verify_table_read_options, file_options_,
|
||||||
cfd->internal_comparator(), files_output[file_idx]->meta,
|
cfd->internal_comparator(), files_output[file_idx]->meta,
|
||||||
|
|||||||
+19
-3
@@ -3947,6 +3947,15 @@ TEST_F(DBTest2, RateLimitedCompactionReads) {
|
|||||||
options.table_factory.reset(NewBlockBasedTableFactory(bbto));
|
options.table_factory.reset(NewBlockBasedTableFactory(bbto));
|
||||||
DestroyAndReopen(options);
|
DestroyAndReopen(options);
|
||||||
|
|
||||||
|
// To precisely control when to start bg compaction for excluding previous
|
||||||
|
// rate-limited bytes of flush read for table verification
|
||||||
|
std::shared_ptr<test::SleepingBackgroundTask> sleeping_task(
|
||||||
|
new test::SleepingBackgroundTask());
|
||||||
|
env_->SetBackgroundThreads(1, Env::LOW);
|
||||||
|
env_->Schedule(&test::SleepingBackgroundTask::DoSleepTask,
|
||||||
|
sleeping_task.get(), Env::Priority::LOW);
|
||||||
|
sleeping_task->WaitUntilSleeping();
|
||||||
|
|
||||||
for (int i = 0; i < kNumL0Files; ++i) {
|
for (int i = 0; i < kNumL0Files; ++i) {
|
||||||
for (int j = 0; j <= kNumKeysPerFile; ++j) {
|
for (int j = 0; j <= kNumKeysPerFile; ++j) {
|
||||||
ASSERT_OK(Put(Key(j), DummyString(kBytesPerKey)));
|
ASSERT_OK(Put(Key(j), DummyString(kBytesPerKey)));
|
||||||
@@ -3956,13 +3965,20 @@ TEST_F(DBTest2, RateLimitedCompactionReads) {
|
|||||||
ASSERT_EQ(i + 1, NumTableFilesAtLevel(0));
|
ASSERT_EQ(i + 1, NumTableFilesAtLevel(0));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
size_t rate_limited_bytes_start_bytes =
|
||||||
|
options.rate_limiter->GetTotalBytesThrough(Env::IO_TOTAL);
|
||||||
|
|
||||||
|
sleeping_task->WakeUp();
|
||||||
|
sleeping_task->WaitUntilDone();
|
||||||
ASSERT_OK(dbfull()->TEST_WaitForCompact());
|
ASSERT_OK(dbfull()->TEST_WaitForCompact());
|
||||||
ASSERT_EQ(0, NumTableFilesAtLevel(0));
|
ASSERT_EQ(0, NumTableFilesAtLevel(0));
|
||||||
|
|
||||||
// should be slightly above 512KB due to non-data blocks read. Arbitrarily
|
// should be slightly above 512KB due to non-data blocks read. Arbitrarily
|
||||||
// chose 1MB as the upper bound on the total bytes read.
|
// chose 1MB as the upper bound on the total bytes read.
|
||||||
size_t rate_limited_bytes = static_cast<size_t>(
|
size_t rate_limited_bytes =
|
||||||
options.rate_limiter->GetTotalBytesThrough(Env::IO_TOTAL));
|
static_cast<size_t>(
|
||||||
|
options.rate_limiter->GetTotalBytesThrough(Env::IO_TOTAL)) -
|
||||||
|
rate_limited_bytes_start_bytes;
|
||||||
// The charges can exist for `IO_LOW` and `IO_USER` priorities.
|
// The charges can exist for `IO_LOW` and `IO_USER` priorities.
|
||||||
size_t rate_limited_bytes_by_pri =
|
size_t rate_limited_bytes_by_pri =
|
||||||
options.rate_limiter->GetTotalBytesThrough(Env::IO_LOW) +
|
options.rate_limiter->GetTotalBytesThrough(Env::IO_LOW) +
|
||||||
|
|||||||
@@ -9,6 +9,8 @@
|
|||||||
#include "db/version_edit.h"
|
#include "db/version_edit.h"
|
||||||
#include "port/port.h"
|
#include "port/port.h"
|
||||||
#include "port/stack_trace.h"
|
#include "port/stack_trace.h"
|
||||||
|
#include "rocksdb/advanced_options.h"
|
||||||
|
#include "rocksdb/options.h"
|
||||||
#include "rocksdb/sst_file_writer.h"
|
#include "rocksdb/sst_file_writer.h"
|
||||||
#include "test_util/testharness.h"
|
#include "test_util/testharness.h"
|
||||||
#include "test_util/testutil.h"
|
#include "test_util/testutil.h"
|
||||||
@@ -1292,6 +1294,80 @@ TEST_F(ExternalSSTFileBasicTest, VerifyChecksumReadahead) {
|
|||||||
Destroy(options);
|
Destroy(options);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST_F(ExternalSSTFileBasicTest, ReadOldValueOfIngestedKeyBug) {
|
||||||
|
Options options = CurrentOptions();
|
||||||
|
options.compaction_style = kCompactionStyleUniversal;
|
||||||
|
options.disable_auto_compactions = true;
|
||||||
|
options.num_levels = 3;
|
||||||
|
options.preserve_internal_time_seconds = 36000;
|
||||||
|
DestroyAndReopen(options);
|
||||||
|
|
||||||
|
// To create the following LSM tree to trigger the bug:
|
||||||
|
// L0
|
||||||
|
// L1 with seqno [1, 2]
|
||||||
|
// L2 with seqno [3, 4]
|
||||||
|
|
||||||
|
// To create L1 shape
|
||||||
|
ASSERT_OK(
|
||||||
|
db_->Put(WriteOptions(), db_->DefaultColumnFamily(), "k1", "seqno1"));
|
||||||
|
ASSERT_OK(db_->Flush(FlushOptions()));
|
||||||
|
ASSERT_OK(
|
||||||
|
db_->Put(WriteOptions(), db_->DefaultColumnFamily(), "k1", "seqno2"));
|
||||||
|
ASSERT_OK(db_->Flush(FlushOptions()));
|
||||||
|
ColumnFamilyMetaData meta_1;
|
||||||
|
db_->GetColumnFamilyMetaData(&meta_1);
|
||||||
|
auto& files_1 = meta_1.levels[0].files;
|
||||||
|
ASSERT_EQ(files_1.size(), 2);
|
||||||
|
std::string file1 = files_1[0].db_path + files_1[0].name;
|
||||||
|
std::string file2 = files_1[1].db_path + files_1[1].name;
|
||||||
|
ASSERT_OK(db_->CompactFiles(CompactionOptions(), {file1, file2}, 1));
|
||||||
|
// To confirm L1 shape
|
||||||
|
ColumnFamilyMetaData meta_2;
|
||||||
|
db_->GetColumnFamilyMetaData(&meta_2);
|
||||||
|
ASSERT_EQ(meta_2.levels[0].files.size(), 0);
|
||||||
|
ASSERT_EQ(meta_2.levels[1].files.size(), 1);
|
||||||
|
// Seqno starts from non-zero due to seqno reservation for
|
||||||
|
// preserve_internal_time_seconds greater than 0;
|
||||||
|
ASSERT_EQ(meta_2.levels[1].files[0].largest_seqno, 102);
|
||||||
|
ASSERT_EQ(meta_2.levels[2].files.size(), 0);
|
||||||
|
// To create L2 shape
|
||||||
|
ASSERT_OK(db_->Put(WriteOptions(), db_->DefaultColumnFamily(), "k2overlap",
|
||||||
|
"old_value"));
|
||||||
|
ASSERT_OK(db_->Flush(FlushOptions()));
|
||||||
|
ASSERT_OK(db_->Put(WriteOptions(), db_->DefaultColumnFamily(), "k2overlap",
|
||||||
|
"old_value"));
|
||||||
|
ASSERT_OK(db_->Flush(FlushOptions()));
|
||||||
|
ColumnFamilyMetaData meta_3;
|
||||||
|
db_->GetColumnFamilyMetaData(&meta_3);
|
||||||
|
auto& files_3 = meta_3.levels[0].files;
|
||||||
|
std::string file3 = files_3[0].db_path + files_3[0].name;
|
||||||
|
std::string file4 = files_3[1].db_path + files_3[1].name;
|
||||||
|
ASSERT_OK(db_->CompactFiles(CompactionOptions(), {file3, file4}, 2));
|
||||||
|
// To confirm L2 shape
|
||||||
|
ColumnFamilyMetaData meta_4;
|
||||||
|
db_->GetColumnFamilyMetaData(&meta_4);
|
||||||
|
ASSERT_EQ(meta_4.levels[0].files.size(), 0);
|
||||||
|
ASSERT_EQ(meta_4.levels[1].files.size(), 1);
|
||||||
|
ASSERT_EQ(meta_4.levels[2].files.size(), 1);
|
||||||
|
ASSERT_EQ(meta_4.levels[2].files[0].largest_seqno, 104);
|
||||||
|
|
||||||
|
// Ingest a file with new value of the key "k2overlap"
|
||||||
|
SstFileWriter sst_file_writer(EnvOptions(), options);
|
||||||
|
std::string f = sst_files_dir_ + "f.sst";
|
||||||
|
ASSERT_OK(sst_file_writer.Open(f));
|
||||||
|
ASSERT_OK(sst_file_writer.Put("k2overlap", "new_value"));
|
||||||
|
ExternalSstFileInfo f_info;
|
||||||
|
ASSERT_OK(sst_file_writer.Finish(&f_info));
|
||||||
|
ASSERT_OK(db_->IngestExternalFile({f}, IngestExternalFileOptions()));
|
||||||
|
|
||||||
|
// To verify new value of the key "k2overlap" is correctly returned
|
||||||
|
ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
|
||||||
|
std::string value;
|
||||||
|
ASSERT_OK(db_->Get(ReadOptions(), "k2overlap", &value));
|
||||||
|
// Before the fix, the value would be "old_value" and assertion failed
|
||||||
|
ASSERT_EQ(value, "new_value");
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(ExternalSSTFileBasicTest, IngestRangeDeletionTombstoneWithGlobalSeqno) {
|
TEST_F(ExternalSSTFileBasicTest, IngestRangeDeletionTombstoneWithGlobalSeqno) {
|
||||||
for (int i = 5; i < 25; i++) {
|
for (int i = 5; i < 25; i++) {
|
||||||
ASSERT_OK(db_->Put(WriteOptions(), db_->DefaultColumnFamily(), Key(i),
|
ASSERT_OK(db_->Put(WriteOptions(), db_->DefaultColumnFamily(), Key(i),
|
||||||
|
|||||||
@@ -937,26 +937,6 @@ Status ExternalSstFileIngestionJob::AssignLevelAndSeqnoForIngestedFile(
|
|||||||
overlap_with_db = true;
|
overlap_with_db = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (compaction_style == kCompactionStyleUniversal && lvl != 0) {
|
|
||||||
const std::vector<FileMetaData*>& level_files =
|
|
||||||
vstorage->LevelFiles(lvl);
|
|
||||||
const SequenceNumber level_largest_seqno =
|
|
||||||
(*std::max_element(level_files.begin(), level_files.end(),
|
|
||||||
[](FileMetaData* f1, FileMetaData* f2) {
|
|
||||||
return f1->fd.largest_seqno <
|
|
||||||
f2->fd.largest_seqno;
|
|
||||||
}))
|
|
||||||
->fd.largest_seqno;
|
|
||||||
// should only assign seqno to current level's largest seqno when
|
|
||||||
// the file fits
|
|
||||||
if (level_largest_seqno != 0 &&
|
|
||||||
IngestedFileFitInLevel(file_to_ingest, lvl)) {
|
|
||||||
*assigned_seqno = level_largest_seqno;
|
|
||||||
} else {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else if (compaction_style == kCompactionStyleUniversal) {
|
} else if (compaction_style == kCompactionStyleUniversal) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-3
@@ -861,7 +861,7 @@ Status FlushJob::WriteLevel0Table() {
|
|||||||
|
|
||||||
{
|
{
|
||||||
auto write_hint = cfd_->CalculateSSTWriteHint(0);
|
auto write_hint = cfd_->CalculateSSTWriteHint(0);
|
||||||
Env::IOPriority io_priority = GetRateLimiterPriorityForWrite();
|
Env::IOPriority io_priority = GetRateLimiterPriority();
|
||||||
db_mutex_->Unlock();
|
db_mutex_->Unlock();
|
||||||
if (log_buffer_) {
|
if (log_buffer_) {
|
||||||
log_buffer_->FlushBufferToLog();
|
log_buffer_->FlushBufferToLog();
|
||||||
@@ -960,7 +960,8 @@ Status FlushJob::WriteLevel0Table() {
|
|||||||
|
|
||||||
const std::string* const full_history_ts_low =
|
const std::string* const full_history_ts_low =
|
||||||
(full_history_ts_low_.empty()) ? nullptr : &full_history_ts_low_;
|
(full_history_ts_low_.empty()) ? nullptr : &full_history_ts_low_;
|
||||||
const ReadOptions read_options(Env::IOActivity::kFlush);
|
ReadOptions read_options(Env::IOActivity::kFlush);
|
||||||
|
read_options.rate_limiter_priority = io_priority;
|
||||||
const WriteOptions write_options(io_priority, Env::IOActivity::kFlush);
|
const WriteOptions write_options(io_priority, Env::IOActivity::kFlush);
|
||||||
TableBuilderOptions tboptions(
|
TableBuilderOptions tboptions(
|
||||||
*cfd_->ioptions(), mutable_cf_options_, read_options, write_options,
|
*cfd_->ioptions(), mutable_cf_options_, read_options, write_options,
|
||||||
@@ -1087,7 +1088,7 @@ Status FlushJob::WriteLevel0Table() {
|
|||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|
||||||
Env::IOPriority FlushJob::GetRateLimiterPriorityForWrite() {
|
Env::IOPriority FlushJob::GetRateLimiterPriority() {
|
||||||
if (versions_ && versions_->GetColumnFamilySet() &&
|
if (versions_ && versions_->GetColumnFamilySet() &&
|
||||||
versions_->GetColumnFamilySet()->write_controller()) {
|
versions_->GetColumnFamilySet()->write_controller()) {
|
||||||
WriteController* write_controller =
|
WriteController* write_controller =
|
||||||
|
|||||||
+1
-1
@@ -129,7 +129,7 @@ class FlushJob {
|
|||||||
Status MemPurge();
|
Status MemPurge();
|
||||||
bool MemPurgeDecider(double threshold);
|
bool MemPurgeDecider(double threshold);
|
||||||
// The rate limiter priority (io_priority) is determined dynamically here.
|
// The rate limiter priority (io_priority) is determined dynamically here.
|
||||||
Env::IOPriority GetRateLimiterPriorityForWrite();
|
Env::IOPriority GetRateLimiterPriority();
|
||||||
std::unique_ptr<FlushJobInfo> GetFlushJobInfo() const;
|
std::unique_ptr<FlushJobInfo> GetFlushJobInfo() const;
|
||||||
|
|
||||||
// Require db_mutex held.
|
// Require db_mutex held.
|
||||||
|
|||||||
@@ -588,7 +588,7 @@ TEST_F(FlushJobTest, GetRateLimiterPriorityForWrite) {
|
|||||||
Env::Priority::USER, nullptr /*IOTracer*/, empty_seqno_to_time_mapping_);
|
Env::Priority::USER, nullptr /*IOTracer*/, empty_seqno_to_time_mapping_);
|
||||||
|
|
||||||
// When the state from WriteController is normal.
|
// When the state from WriteController is normal.
|
||||||
ASSERT_EQ(flush_job.GetRateLimiterPriorityForWrite(), Env::IO_HIGH);
|
ASSERT_EQ(flush_job.GetRateLimiterPriority(), Env::IO_HIGH);
|
||||||
|
|
||||||
WriteController* write_controller =
|
WriteController* write_controller =
|
||||||
flush_job.versions_->GetColumnFamilySet()->write_controller();
|
flush_job.versions_->GetColumnFamilySet()->write_controller();
|
||||||
@@ -597,14 +597,14 @@ TEST_F(FlushJobTest, GetRateLimiterPriorityForWrite) {
|
|||||||
// When the state from WriteController is Delayed.
|
// When the state from WriteController is Delayed.
|
||||||
std::unique_ptr<WriteControllerToken> delay_token =
|
std::unique_ptr<WriteControllerToken> delay_token =
|
||||||
write_controller->GetDelayToken(1000000);
|
write_controller->GetDelayToken(1000000);
|
||||||
ASSERT_EQ(flush_job.GetRateLimiterPriorityForWrite(), Env::IO_USER);
|
ASSERT_EQ(flush_job.GetRateLimiterPriority(), Env::IO_USER);
|
||||||
}
|
}
|
||||||
|
|
||||||
{
|
{
|
||||||
// When the state from WriteController is Stopped.
|
// When the state from WriteController is Stopped.
|
||||||
std::unique_ptr<WriteControllerToken> stop_token =
|
std::unique_ptr<WriteControllerToken> stop_token =
|
||||||
write_controller->GetStopToken();
|
write_controller->GetStopToken();
|
||||||
ASSERT_EQ(flush_job.GetRateLimiterPriorityForWrite(), Env::IO_USER);
|
ASSERT_EQ(flush_job.GetRateLimiterPriority(), Env::IO_USER);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -61,6 +61,8 @@ ImmutableMemTableOptions::ImmutableMemTableOptions(
|
|||||||
inplace_update_num_locks(mutable_cf_options.inplace_update_num_locks),
|
inplace_update_num_locks(mutable_cf_options.inplace_update_num_locks),
|
||||||
inplace_callback(ioptions.inplace_callback),
|
inplace_callback(ioptions.inplace_callback),
|
||||||
max_successive_merges(mutable_cf_options.max_successive_merges),
|
max_successive_merges(mutable_cf_options.max_successive_merges),
|
||||||
|
strict_max_successive_merges(
|
||||||
|
mutable_cf_options.strict_max_successive_merges),
|
||||||
statistics(ioptions.stats),
|
statistics(ioptions.stats),
|
||||||
merge_operator(ioptions.merge_operator.get()),
|
merge_operator(ioptions.merge_operator.get()),
|
||||||
info_log(ioptions.logger),
|
info_log(ioptions.logger),
|
||||||
|
|||||||
@@ -54,6 +54,7 @@ struct ImmutableMemTableOptions {
|
|||||||
Slice delta_value,
|
Slice delta_value,
|
||||||
std::string* merged_value);
|
std::string* merged_value);
|
||||||
size_t max_successive_merges;
|
size_t max_successive_merges;
|
||||||
|
bool strict_max_successive_merges;
|
||||||
Statistics* statistics;
|
Statistics* statistics;
|
||||||
MergeOperator* merge_operator;
|
MergeOperator* merge_operator;
|
||||||
Logger* info_log;
|
Logger* info_log;
|
||||||
|
|||||||
@@ -1052,6 +1052,114 @@ TEST_F(SeqnoTimeTest, MappingAppend) {
|
|||||||
ASSERT_EQ(test.TEST_GetLastEntry(), P({11, 250}));
|
ASSERT_EQ(test.TEST_GetLastEntry(), P({11, 250}));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST_F(SeqnoTimeTest, CapacityLimits) {
|
||||||
|
using P = SeqnoToTimeMapping::SeqnoTimePair;
|
||||||
|
SeqnoToTimeMapping test;
|
||||||
|
|
||||||
|
test.SetCapacity(3);
|
||||||
|
EXPECT_TRUE(test.Append(10, 300));
|
||||||
|
EXPECT_TRUE(test.Append(20, 400));
|
||||||
|
EXPECT_TRUE(test.Append(30, 500));
|
||||||
|
EXPECT_TRUE(test.Append(40, 600));
|
||||||
|
// Capacity 3 is small enough that the non-strict limit is
|
||||||
|
// equal to the strict limit.
|
||||||
|
EXPECT_EQ(3U, test.Size());
|
||||||
|
EXPECT_EQ(test.TEST_GetLastEntry(), P({40, 600}));
|
||||||
|
|
||||||
|
// Same for Capacity 2
|
||||||
|
test.SetCapacity(2);
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
EXPECT_EQ(test.TEST_GetLastEntry(), P({40, 600}));
|
||||||
|
|
||||||
|
EXPECT_TRUE(test.Append(50, 700));
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
EXPECT_EQ(test.TEST_GetLastEntry(), P({50, 700}));
|
||||||
|
|
||||||
|
// Capacity 1 is difficult to work with internally, so is
|
||||||
|
// coerced to 2.
|
||||||
|
test.SetCapacity(1);
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
EXPECT_EQ(test.TEST_GetLastEntry(), P({50, 700}));
|
||||||
|
|
||||||
|
EXPECT_TRUE(test.Append(60, 800));
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
EXPECT_EQ(test.TEST_GetLastEntry(), P({60, 800}));
|
||||||
|
|
||||||
|
// Capacity 0 means throw everything away
|
||||||
|
test.SetCapacity(0);
|
||||||
|
EXPECT_EQ(0U, test.Size());
|
||||||
|
|
||||||
|
EXPECT_FALSE(test.Append(70, 900));
|
||||||
|
EXPECT_EQ(0U, test.Size());
|
||||||
|
|
||||||
|
// Unlimited capacity
|
||||||
|
test.SetCapacity(UINT64_MAX);
|
||||||
|
for (unsigned i = 1; i <= 10101U; i++) {
|
||||||
|
EXPECT_TRUE(test.Append(i, 11U * i));
|
||||||
|
}
|
||||||
|
EXPECT_EQ(10101U, test.Size());
|
||||||
|
}
|
||||||
|
|
||||||
|
TEST_F(SeqnoTimeTest, TimeSpanLimits) {
|
||||||
|
SeqnoToTimeMapping test;
|
||||||
|
|
||||||
|
// Default: no limit
|
||||||
|
for (unsigned i = 1; i <= 63U; i++) {
|
||||||
|
EXPECT_TRUE(test.Append(1000 + i, uint64_t{1} << i));
|
||||||
|
}
|
||||||
|
// None dropped.
|
||||||
|
EXPECT_EQ(63U, test.Size());
|
||||||
|
|
||||||
|
test.Clear();
|
||||||
|
|
||||||
|
// Explicit no limit
|
||||||
|
test.SetMaxTimeSpan(UINT64_MAX);
|
||||||
|
for (unsigned i = 1; i <= 63U; i++) {
|
||||||
|
EXPECT_TRUE(test.Append(1000 + i, uint64_t{1} << i));
|
||||||
|
}
|
||||||
|
// None dropped.
|
||||||
|
EXPECT_EQ(63U, test.Size());
|
||||||
|
|
||||||
|
// We generally keep 2 entries as long as the configured max time span
|
||||||
|
// is non-zero
|
||||||
|
test.SetMaxTimeSpan(10);
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
|
||||||
|
test.SetMaxTimeSpan(1);
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
|
||||||
|
// But go down to 1 entry if the max time span is zero
|
||||||
|
test.SetMaxTimeSpan(0);
|
||||||
|
EXPECT_EQ(1U, test.Size());
|
||||||
|
|
||||||
|
EXPECT_TRUE(test.Append(2000, (uint64_t{1} << 63) + 42U));
|
||||||
|
EXPECT_EQ(1U, test.Size());
|
||||||
|
|
||||||
|
test.Clear();
|
||||||
|
|
||||||
|
// Test more typical behavior. Note that one entry at or beyond the max span
|
||||||
|
// is kept.
|
||||||
|
test.SetMaxTimeSpan(100);
|
||||||
|
EXPECT_TRUE(test.Append(1001, 123));
|
||||||
|
EXPECT_TRUE(test.Append(1002, 134));
|
||||||
|
EXPECT_TRUE(test.Append(1003, 150));
|
||||||
|
EXPECT_TRUE(test.Append(1004, 189));
|
||||||
|
EXPECT_TRUE(test.Append(1005, 220));
|
||||||
|
EXPECT_EQ(5U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1006, 233));
|
||||||
|
EXPECT_EQ(6U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1007, 234));
|
||||||
|
EXPECT_EQ(6U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1008, 235));
|
||||||
|
EXPECT_EQ(7U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1009, 300));
|
||||||
|
EXPECT_EQ(6U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1010, 350));
|
||||||
|
EXPECT_EQ(3U, test.Size());
|
||||||
|
EXPECT_TRUE(test.Append(1011, 470));
|
||||||
|
EXPECT_EQ(2U, test.Size());
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(SeqnoTimeTest, ProximalFunctions) {
|
TEST_F(SeqnoTimeTest, ProximalFunctions) {
|
||||||
SeqnoToTimeMapping test;
|
SeqnoToTimeMapping test;
|
||||||
test.SetCapacity(10);
|
test.SetCapacity(10);
|
||||||
@@ -1201,6 +1309,93 @@ TEST_F(SeqnoTimeTest, PrePopulate) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
TEST_F(SeqnoTimeTest, CopyFromSeqnoRange) {
|
||||||
|
SeqnoToTimeMapping test_from;
|
||||||
|
SeqnoToTimeMapping test_to;
|
||||||
|
|
||||||
|
// With zero to draw from
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 1000000);
|
||||||
|
EXPECT_EQ(test_to.Size(), 0U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 100, 100);
|
||||||
|
EXPECT_EQ(test_to.Size(), 0U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, kMaxSequenceNumber, 0);
|
||||||
|
EXPECT_EQ(test_to.Size(), 0U);
|
||||||
|
|
||||||
|
// With one to draw from
|
||||||
|
EXPECT_TRUE(test_from.Append(10, 500));
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 1000000);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
// Includes one entry before range
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 100, 100);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
// Includes one entry before range (even if somewhat nonsensical)
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, kMaxSequenceNumber, 0);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 9);
|
||||||
|
EXPECT_EQ(test_to.Size(), 0U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 10);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
// With more to draw from
|
||||||
|
EXPECT_TRUE(test_from.Append(20, 600));
|
||||||
|
EXPECT_TRUE(test_from.Append(30, 700));
|
||||||
|
EXPECT_TRUE(test_from.Append(40, 800));
|
||||||
|
EXPECT_TRUE(test_from.Append(50, 900));
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 1000000);
|
||||||
|
EXPECT_EQ(test_to.Size(), 5U);
|
||||||
|
|
||||||
|
// Includes one entry before range
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 100, 100);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 19, 19);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
// Includes one entry before range (even if somewhat nonsensical)
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, kMaxSequenceNumber, 0);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 9);
|
||||||
|
EXPECT_EQ(test_to.Size(), 0U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 0, 10);
|
||||||
|
EXPECT_EQ(test_to.Size(), 1U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 20, 20);
|
||||||
|
EXPECT_EQ(test_to.Size(), 2U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 20, 29);
|
||||||
|
EXPECT_EQ(test_to.Size(), 2U);
|
||||||
|
|
||||||
|
test_to.Clear();
|
||||||
|
test_to.CopyFromSeqnoRange(test_from, 20, 30);
|
||||||
|
EXPECT_EQ(test_to.Size(), 3U);
|
||||||
|
}
|
||||||
|
|
||||||
TEST_F(SeqnoTimeTest, EnforceWithNow) {
|
TEST_F(SeqnoTimeTest, EnforceWithNow) {
|
||||||
constexpr uint64_t kMaxTimeSpan = 420;
|
constexpr uint64_t kMaxTimeSpan = 420;
|
||||||
SeqnoToTimeMapping test;
|
SeqnoToTimeMapping test;
|
||||||
|
|||||||
@@ -72,20 +72,16 @@ SequenceNumber SeqnoToTimeMapping::GetProximalSeqnoBeforeTime(
|
|||||||
void SeqnoToTimeMapping::EnforceMaxTimeSpan(uint64_t now) {
|
void SeqnoToTimeMapping::EnforceMaxTimeSpan(uint64_t now) {
|
||||||
assert(enforced_); // at least sorted
|
assert(enforced_); // at least sorted
|
||||||
uint64_t cutoff_time;
|
uint64_t cutoff_time;
|
||||||
|
if (pairs_.size() <= 1) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
if (now > 0) {
|
if (now > 0) {
|
||||||
if (pairs_.size() <= 1) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (now < max_time_span_) {
|
if (now < max_time_span_) {
|
||||||
// Nothing eligible to prune / avoid underflow
|
// Nothing eligible to prune / avoid underflow
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
cutoff_time = now - max_time_span_;
|
cutoff_time = now - max_time_span_;
|
||||||
} else {
|
} else {
|
||||||
if (pairs_.size() <= 2) {
|
|
||||||
// Need to keep at least two if we don't know the current time
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
const auto& last = pairs_.back();
|
const auto& last = pairs_.back();
|
||||||
if (last.time < max_time_span_) {
|
if (last.time < max_time_span_) {
|
||||||
// Nothing eligible to prune / avoid underflow
|
// Nothing eligible to prune / avoid underflow
|
||||||
@@ -108,7 +104,6 @@ void SeqnoToTimeMapping::EnforceCapacity(bool strict) {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// Treat cap of 1 as 2 to work with the below algorithm (etc.)
|
// Treat cap of 1 as 2 to work with the below algorithm (etc.)
|
||||||
// TODO: unit test
|
|
||||||
if (strict_cap == 1) {
|
if (strict_cap == 1) {
|
||||||
strict_cap = 2;
|
strict_cap = 2;
|
||||||
}
|
}
|
||||||
@@ -400,12 +395,16 @@ void SeqnoToTimeMapping::CopyFromSeqnoRange(const SeqnoToTimeMapping& src,
|
|||||||
SequenceNumber to_seqno) {
|
SequenceNumber to_seqno) {
|
||||||
bool orig_empty = Empty();
|
bool orig_empty = Empty();
|
||||||
auto src_it = src.FindGreaterEqSeqno(from_seqno);
|
auto src_it = src.FindGreaterEqSeqno(from_seqno);
|
||||||
|
// Allow nonsensical ranges like [1000, 0] which might show up e.g. for
|
||||||
|
// an SST file with no entries.
|
||||||
|
auto src_it_end =
|
||||||
|
to_seqno < from_seqno ? src_it : src.FindGreaterSeqno(to_seqno);
|
||||||
// To best answer GetProximalTimeBeforeSeqno(from_seqno) we need an entry
|
// To best answer GetProximalTimeBeforeSeqno(from_seqno) we need an entry
|
||||||
// with a seqno before that (if available)
|
// with a seqno before that (if available)
|
||||||
if (src_it != src.pairs_.begin()) {
|
if (src_it != src.pairs_.begin()) {
|
||||||
--src_it;
|
--src_it;
|
||||||
}
|
}
|
||||||
auto src_it_end = src.FindGreaterSeqno(to_seqno);
|
assert(src_it <= src_it_end);
|
||||||
std::copy(src_it, src_it_end, std::back_inserter(pairs_));
|
std::copy(src_it, src_it_end, std::back_inserter(pairs_));
|
||||||
|
|
||||||
if (!orig_empty || max_time_span_ < UINT64_MAX || capacity_ < UINT64_MAX) {
|
if (!orig_empty || max_time_span_ < UINT64_MAX || capacity_ < UINT64_MAX) {
|
||||||
@@ -420,6 +419,7 @@ bool SeqnoToTimeMapping::Append(SequenceNumber seqno, uint64_t time) {
|
|||||||
bool added = false;
|
bool added = false;
|
||||||
if (seqno == 0) {
|
if (seqno == 0) {
|
||||||
// skip seq number 0, which may have special meaning, like zeroed out data
|
// skip seq number 0, which may have special meaning, like zeroed out data
|
||||||
|
// TODO: consider changing?
|
||||||
} else if (pairs_.empty()) {
|
} else if (pairs_.empty()) {
|
||||||
enforced_ = true;
|
enforced_ = true;
|
||||||
pairs_.push_back({seqno, time});
|
pairs_.push_back({seqno, time});
|
||||||
|
|||||||
@@ -125,12 +125,13 @@ class SeqnoToTimeMapping {
|
|||||||
// under enforcement mode, the structure will maintian only one entry older
|
// under enforcement mode, the structure will maintian only one entry older
|
||||||
// than the newest entry time minus max_time_span, so that
|
// than the newest entry time minus max_time_span, so that
|
||||||
// GetProximalSeqnoBeforeTime queries back to that time return a good result.
|
// GetProximalSeqnoBeforeTime queries back to that time return a good result.
|
||||||
// UINT64_MAX == unlimited. Returns *this.
|
// UINT64_MAX == unlimited. 0 == retain just one latest entry. Returns *this.
|
||||||
SeqnoToTimeMapping& SetMaxTimeSpan(uint64_t max_time_span);
|
SeqnoToTimeMapping& SetMaxTimeSpan(uint64_t max_time_span);
|
||||||
|
|
||||||
// Set the nominal capacity under enforcement mode. The structure is allowed
|
// Set the nominal capacity under enforcement mode. The structure is allowed
|
||||||
// to grow some reasonable fraction larger but will automatically compact
|
// to grow some reasonable fraction larger but will automatically compact
|
||||||
// down to this size. UINT64_MAX == unlimited. Returns *this.
|
// down to this size. UINT64_MAX == unlimited. 0 == retain nothing.
|
||||||
|
// Returns *this.
|
||||||
SeqnoToTimeMapping& SetCapacity(uint64_t capacity);
|
SeqnoToTimeMapping& SetCapacity(uint64_t capacity);
|
||||||
|
|
||||||
// ==== Modifiers, enforced ==== //
|
// ==== Modifiers, enforced ==== //
|
||||||
@@ -243,6 +244,14 @@ class SeqnoToTimeMapping {
|
|||||||
|
|
||||||
std::deque<SeqnoTimePair> pairs_;
|
std::deque<SeqnoTimePair> pairs_;
|
||||||
|
|
||||||
|
// Whether this object is in the "enforced" state. Between calls to public
|
||||||
|
// functions, enforced_==true means that
|
||||||
|
// * `pairs_` is sorted
|
||||||
|
// * The capacity limit (non-strict) is met
|
||||||
|
// * The time span limit is met
|
||||||
|
// However, some places within the implementation (Append()) will temporarily
|
||||||
|
// violate those last two conditions while enforced_==true. See also the
|
||||||
|
// Enforce*() and Sort*() private functions below.
|
||||||
bool enforced_ = true;
|
bool enforced_ = true;
|
||||||
|
|
||||||
void EnforceMaxTimeSpan(uint64_t now = 0);
|
void EnforceMaxTimeSpan(uint64_t now = 0);
|
||||||
|
|||||||
@@ -2513,6 +2513,11 @@ class MemTableInserter : public WriteBatch::Handler {
|
|||||||
|
|
||||||
// TODO: plumb Env::IOActivity, Env::IOPriority
|
// TODO: plumb Env::IOActivity, Env::IOPriority
|
||||||
ReadOptions read_options;
|
ReadOptions read_options;
|
||||||
|
if (!moptions->strict_max_successive_merges) {
|
||||||
|
// Blocking the write path with read I/O is typically unacceptable, so
|
||||||
|
// only do this merge when the operands are all found in memory.
|
||||||
|
read_options.read_tier = kBlockCacheTier;
|
||||||
|
}
|
||||||
read_options.snapshot = &read_from_snapshot;
|
read_options.snapshot = &read_from_snapshot;
|
||||||
|
|
||||||
auto cf_handle = cf_mems_->GetColumnFamilyHandle();
|
auto cf_handle = cf_mems_->GetColumnFamilyHandle();
|
||||||
|
|||||||
@@ -247,6 +247,7 @@ bool StressTest::BuildOptionsTable() {
|
|||||||
}},
|
}},
|
||||||
{"memtable_huge_page_size", {"0", std::to_string(2 * 1024 * 1024)}},
|
{"memtable_huge_page_size", {"0", std::to_string(2 * 1024 * 1024)}},
|
||||||
{"max_successive_merges", {"0", "2", "4"}},
|
{"max_successive_merges", {"0", "2", "4"}},
|
||||||
|
{"strict_max_successive_merges", {"false", "true"}},
|
||||||
{"inplace_update_num_locks", {"100", "200", "300"}},
|
{"inplace_update_num_locks", {"100", "200", "300"}},
|
||||||
// TODO: re-enable once internal task T124324915 is fixed.
|
// TODO: re-enable once internal task T124324915 is fixed.
|
||||||
// {"experimental_mempurge_threshold", {"0.0", "1.0"}},
|
// {"experimental_mempurge_threshold", {"0.0", "1.0"}},
|
||||||
|
|||||||
@@ -16,6 +16,7 @@
|
|||||||
#include "monitoring/histogram.h"
|
#include "monitoring/histogram.h"
|
||||||
#include "monitoring/iostats_context_imp.h"
|
#include "monitoring/iostats_context_imp.h"
|
||||||
#include "port/port.h"
|
#include "port/port.h"
|
||||||
|
#include "rocksdb/file_system.h"
|
||||||
#include "test_util/sync_point.h"
|
#include "test_util/sync_point.h"
|
||||||
#include "util/aligned_buffer.h"
|
#include "util/aligned_buffer.h"
|
||||||
#include "util/random.h"
|
#include "util/random.h"
|
||||||
@@ -38,6 +39,8 @@ IOStatus SequentialFileReader::Create(
|
|||||||
IOStatus SequentialFileReader::Read(size_t n, Slice* result, char* scratch,
|
IOStatus SequentialFileReader::Read(size_t n, Slice* result, char* scratch,
|
||||||
Env::IOPriority rate_limiter_priority) {
|
Env::IOPriority rate_limiter_priority) {
|
||||||
IOStatus io_s;
|
IOStatus io_s;
|
||||||
|
IOOptions io_opts;
|
||||||
|
io_opts.rate_limiter_priority = rate_limiter_priority;
|
||||||
if (use_direct_io()) {
|
if (use_direct_io()) {
|
||||||
//
|
//
|
||||||
// |-offset_advance-|---bytes returned--|
|
// |-offset_advance-|---bytes returned--|
|
||||||
@@ -76,7 +79,7 @@ IOStatus SequentialFileReader::Read(size_t n, Slice* result, char* scratch,
|
|||||||
start_ts = FileOperationInfo::StartNow();
|
start_ts = FileOperationInfo::StartNow();
|
||||||
}
|
}
|
||||||
io_s = file_->PositionedRead(aligned_offset + buf.CurrentSize(), allowed,
|
io_s = file_->PositionedRead(aligned_offset + buf.CurrentSize(), allowed,
|
||||||
IOOptions(), &tmp, buf.Destination(),
|
io_opts, &tmp, buf.Destination(),
|
||||||
nullptr /* dbg */);
|
nullptr /* dbg */);
|
||||||
if (ShouldNotifyListeners()) {
|
if (ShouldNotifyListeners()) {
|
||||||
auto finish_ts = FileOperationInfo::FinishNow();
|
auto finish_ts = FileOperationInfo::FinishNow();
|
||||||
@@ -119,7 +122,7 @@ IOStatus SequentialFileReader::Read(size_t n, Slice* result, char* scratch,
|
|||||||
start_ts = FileOperationInfo::StartNow();
|
start_ts = FileOperationInfo::StartNow();
|
||||||
}
|
}
|
||||||
Slice tmp;
|
Slice tmp;
|
||||||
io_s = file_->Read(allowed, IOOptions(), &tmp, scratch + read,
|
io_s = file_->Read(allowed, io_opts, &tmp, scratch + read,
|
||||||
nullptr /* dbg */);
|
nullptr /* dbg */);
|
||||||
if (ShouldNotifyListeners()) {
|
if (ShouldNotifyListeners()) {
|
||||||
auto finish_ts = FileOperationInfo::FinishNow();
|
auto finish_ts = FileOperationInfo::FinishNow();
|
||||||
|
|||||||
@@ -673,18 +673,29 @@ struct AdvancedColumnFamilyOptions {
|
|||||||
TablePropertiesCollectorFactories table_properties_collector_factories;
|
TablePropertiesCollectorFactories table_properties_collector_factories;
|
||||||
|
|
||||||
// Maximum number of successive merge operations on a key in the memtable.
|
// Maximum number of successive merge operations on a key in the memtable.
|
||||||
|
// It may be violated when filesystem reads would be needed to stay under the
|
||||||
|
// limit, unless `strict_max_successive_merges` is explicitly set.
|
||||||
//
|
//
|
||||||
// When a merge operation is added to the memtable and the maximum number of
|
// When a merge operation is added to the memtable and the maximum number of
|
||||||
// successive merges is reached, the value of the key will be calculated and
|
// successive merges is reached, RocksDB will attempt to read the value. Upon
|
||||||
// inserted into the memtable instead of the merge operation. This will
|
// success, the value will be inserted into the memtable instead of the merge
|
||||||
// ensure that there are never more than max_successive_merges merge
|
// operation.
|
||||||
// operations in the memtable.
|
|
||||||
//
|
//
|
||||||
// Default: 0 (disabled)
|
// Default: 0 (disabled)
|
||||||
//
|
//
|
||||||
// Dynamically changeable through SetOptions() API
|
// Dynamically changeable through SetOptions() API
|
||||||
size_t max_successive_merges = 0;
|
size_t max_successive_merges = 0;
|
||||||
|
|
||||||
|
// Whether to allow filesystem reads to stay under the `max_successive_merges`
|
||||||
|
// limit. When true, this can lead to merge writes blocking the write path
|
||||||
|
// waiting on filesystem reads.
|
||||||
|
//
|
||||||
|
// This option is temporary in case the recent change to disallow filesystem
|
||||||
|
// reads during merge writes has a problem and users need to undo it quickly.
|
||||||
|
//
|
||||||
|
// Default: false
|
||||||
|
bool strict_max_successive_merges = false;
|
||||||
|
|
||||||
// This flag specifies that the implementation should optimize the filters
|
// This flag specifies that the implementation should optimize the filters
|
||||||
// mainly for cases where keys are found rather than also optimize for keys
|
// mainly for cases where keys are found rather than also optimize for keys
|
||||||
// missed. This would be used in cases where the application knows that
|
// missed. This would be used in cases where the application knows that
|
||||||
|
|||||||
@@ -102,7 +102,7 @@ class OptimisticTransactionDB : public StackableDB {
|
|||||||
std::vector<ColumnFamilyHandle*>* handles,
|
std::vector<ColumnFamilyHandle*>* handles,
|
||||||
OptimisticTransactionDB** dbptr);
|
OptimisticTransactionDB** dbptr);
|
||||||
|
|
||||||
virtual ~OptimisticTransactionDB() {}
|
~OptimisticTransactionDB() override {}
|
||||||
|
|
||||||
// Starts a new Transaction.
|
// Starts a new Transaction.
|
||||||
//
|
//
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ class StackableDB : public DB {
|
|||||||
explicit StackableDB(std::shared_ptr<DB> db)
|
explicit StackableDB(std::shared_ptr<DB> db)
|
||||||
: db_(db.get()), shared_db_ptr_(db) {}
|
: db_(db.get()), shared_db_ptr_(db) {}
|
||||||
|
|
||||||
~StackableDB() {
|
~StackableDB() override {
|
||||||
if (shared_db_ptr_ == nullptr) {
|
if (shared_db_ptr_ == nullptr) {
|
||||||
delete db_;
|
delete db_;
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -135,7 +135,7 @@ class RangeLockManagerHandle : public LockManagerHandle {
|
|||||||
virtual std::vector<RangeDeadlockPath> GetRangeDeadlockInfoBuffer() = 0;
|
virtual std::vector<RangeDeadlockPath> GetRangeDeadlockInfoBuffer() = 0;
|
||||||
virtual void SetRangeDeadlockInfoBufferSize(uint32_t target_size) = 0;
|
virtual void SetRangeDeadlockInfoBufferSize(uint32_t target_size) = 0;
|
||||||
|
|
||||||
virtual ~RangeLockManagerHandle() {}
|
~RangeLockManagerHandle() override {}
|
||||||
};
|
};
|
||||||
|
|
||||||
// A factory function to create a Range Lock Manager. The created object should
|
// A factory function to create a Range Lock Manager. The created object should
|
||||||
@@ -503,4 +503,3 @@ class TransactionDB : public StackableDB {
|
|||||||
};
|
};
|
||||||
|
|
||||||
} // namespace ROCKSDB_NAMESPACE
|
} // namespace ROCKSDB_NAMESPACE
|
||||||
|
|
||||||
|
|||||||
@@ -13,7 +13,7 @@
|
|||||||
// minor or major version number planned for release.
|
// minor or major version number planned for release.
|
||||||
#define ROCKSDB_MAJOR 8
|
#define ROCKSDB_MAJOR 8
|
||||||
#define ROCKSDB_MINOR 11
|
#define ROCKSDB_MINOR 11
|
||||||
#define ROCKSDB_PATCH 0
|
#define ROCKSDB_PATCH 99
|
||||||
|
|
||||||
// Do not use these. We made the mistake of declaring macros starting with
|
// Do not use these. We made the mistake of declaring macros starting with
|
||||||
// double underscore. Now we have to live with our choice. We'll deprecate these
|
// double underscore. Now we have to live with our choice. We'll deprecate these
|
||||||
|
|||||||
@@ -342,6 +342,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
|
|||||||
{offsetof(struct MutableCFOptions, max_successive_merges),
|
{offsetof(struct MutableCFOptions, max_successive_merges),
|
||||||
OptionType::kSizeT, OptionVerificationType::kNormal,
|
OptionType::kSizeT, OptionVerificationType::kNormal,
|
||||||
OptionTypeFlags::kMutable}},
|
OptionTypeFlags::kMutable}},
|
||||||
|
{"strict_max_successive_merges",
|
||||||
|
{offsetof(struct MutableCFOptions, strict_max_successive_merges),
|
||||||
|
OptionType::kBoolean, OptionVerificationType::kNormal,
|
||||||
|
OptionTypeFlags::kMutable}},
|
||||||
{"memtable_huge_page_size",
|
{"memtable_huge_page_size",
|
||||||
{offsetof(struct MutableCFOptions, memtable_huge_page_size),
|
{offsetof(struct MutableCFOptions, memtable_huge_page_size),
|
||||||
OptionType::kSizeT, OptionVerificationType::kNormal,
|
OptionType::kSizeT, OptionVerificationType::kNormal,
|
||||||
@@ -1060,6 +1064,8 @@ void MutableCFOptions::Dump(Logger* log) const {
|
|||||||
ROCKS_LOG_INFO(log,
|
ROCKS_LOG_INFO(log,
|
||||||
" max_successive_merges: %" ROCKSDB_PRIszt,
|
" max_successive_merges: %" ROCKSDB_PRIszt,
|
||||||
max_successive_merges);
|
max_successive_merges);
|
||||||
|
ROCKS_LOG_INFO(log, " strict_max_successive_merges: %d",
|
||||||
|
strict_max_successive_merges);
|
||||||
ROCKS_LOG_INFO(log,
|
ROCKS_LOG_INFO(log,
|
||||||
" inplace_update_num_locks: %" ROCKSDB_PRIszt,
|
" inplace_update_num_locks: %" ROCKSDB_PRIszt,
|
||||||
inplace_update_num_locks);
|
inplace_update_num_locks);
|
||||||
|
|||||||
@@ -120,6 +120,7 @@ struct MutableCFOptions {
|
|||||||
memtable_whole_key_filtering(options.memtable_whole_key_filtering),
|
memtable_whole_key_filtering(options.memtable_whole_key_filtering),
|
||||||
memtable_huge_page_size(options.memtable_huge_page_size),
|
memtable_huge_page_size(options.memtable_huge_page_size),
|
||||||
max_successive_merges(options.max_successive_merges),
|
max_successive_merges(options.max_successive_merges),
|
||||||
|
strict_max_successive_merges(options.strict_max_successive_merges),
|
||||||
inplace_update_num_locks(options.inplace_update_num_locks),
|
inplace_update_num_locks(options.inplace_update_num_locks),
|
||||||
prefix_extractor(options.prefix_extractor),
|
prefix_extractor(options.prefix_extractor),
|
||||||
experimental_mempurge_threshold(
|
experimental_mempurge_threshold(
|
||||||
@@ -192,6 +193,7 @@ struct MutableCFOptions {
|
|||||||
memtable_whole_key_filtering(false),
|
memtable_whole_key_filtering(false),
|
||||||
memtable_huge_page_size(0),
|
memtable_huge_page_size(0),
|
||||||
max_successive_merges(0),
|
max_successive_merges(0),
|
||||||
|
strict_max_successive_merges(false),
|
||||||
inplace_update_num_locks(0),
|
inplace_update_num_locks(0),
|
||||||
prefix_extractor(nullptr),
|
prefix_extractor(nullptr),
|
||||||
experimental_mempurge_threshold(0.0),
|
experimental_mempurge_threshold(0.0),
|
||||||
@@ -259,6 +261,7 @@ struct MutableCFOptions {
|
|||||||
bool memtable_whole_key_filtering;
|
bool memtable_whole_key_filtering;
|
||||||
size_t memtable_huge_page_size;
|
size_t memtable_huge_page_size;
|
||||||
size_t max_successive_merges;
|
size_t max_successive_merges;
|
||||||
|
bool strict_max_successive_merges;
|
||||||
size_t inplace_update_num_locks;
|
size_t inplace_update_num_locks;
|
||||||
std::shared_ptr<const SliceTransform> prefix_extractor;
|
std::shared_ptr<const SliceTransform> prefix_extractor;
|
||||||
// [experimental]
|
// [experimental]
|
||||||
|
|||||||
@@ -87,6 +87,7 @@ AdvancedColumnFamilyOptions::AdvancedColumnFamilyOptions(const Options& options)
|
|||||||
table_properties_collector_factories(
|
table_properties_collector_factories(
|
||||||
options.table_properties_collector_factories),
|
options.table_properties_collector_factories),
|
||||||
max_successive_merges(options.max_successive_merges),
|
max_successive_merges(options.max_successive_merges),
|
||||||
|
strict_max_successive_merges(options.strict_max_successive_merges),
|
||||||
optimize_filters_for_hits(options.optimize_filters_for_hits),
|
optimize_filters_for_hits(options.optimize_filters_for_hits),
|
||||||
paranoid_file_checks(options.paranoid_file_checks),
|
paranoid_file_checks(options.paranoid_file_checks),
|
||||||
force_consistency_checks(options.force_consistency_checks),
|
force_consistency_checks(options.force_consistency_checks),
|
||||||
@@ -399,6 +400,9 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
|
|||||||
log,
|
log,
|
||||||
" Options.max_successive_merges: %" ROCKSDB_PRIszt,
|
" Options.max_successive_merges: %" ROCKSDB_PRIszt,
|
||||||
max_successive_merges);
|
max_successive_merges);
|
||||||
|
ROCKS_LOG_HEADER(log,
|
||||||
|
" Options.strict_max_successive_merges: %d",
|
||||||
|
strict_max_successive_merges);
|
||||||
ROCKS_LOG_HEADER(log,
|
ROCKS_LOG_HEADER(log,
|
||||||
" Options.optimize_filters_for_hits: %d",
|
" Options.optimize_filters_for_hits: %d",
|
||||||
optimize_filters_for_hits);
|
optimize_filters_for_hits);
|
||||||
|
|||||||
@@ -206,6 +206,7 @@ void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
|
|||||||
cf_opts->memtable_whole_key_filtering = moptions.memtable_whole_key_filtering;
|
cf_opts->memtable_whole_key_filtering = moptions.memtable_whole_key_filtering;
|
||||||
cf_opts->memtable_huge_page_size = moptions.memtable_huge_page_size;
|
cf_opts->memtable_huge_page_size = moptions.memtable_huge_page_size;
|
||||||
cf_opts->max_successive_merges = moptions.max_successive_merges;
|
cf_opts->max_successive_merges = moptions.max_successive_merges;
|
||||||
|
cf_opts->strict_max_successive_merges = moptions.strict_max_successive_merges;
|
||||||
cf_opts->inplace_update_num_locks = moptions.inplace_update_num_locks;
|
cf_opts->inplace_update_num_locks = moptions.inplace_update_num_locks;
|
||||||
cf_opts->prefix_extractor = moptions.prefix_extractor;
|
cf_opts->prefix_extractor = moptions.prefix_extractor;
|
||||||
cf_opts->experimental_mempurge_threshold =
|
cf_opts->experimental_mempurge_threshold =
|
||||||
|
|||||||
@@ -493,6 +493,7 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
|
|||||||
"target_file_size_base=4294976376;"
|
"target_file_size_base=4294976376;"
|
||||||
"memtable_huge_page_size=2557;"
|
"memtable_huge_page_size=2557;"
|
||||||
"max_successive_merges=5497;"
|
"max_successive_merges=5497;"
|
||||||
|
"strict_max_successive_merges=true;"
|
||||||
"max_sequential_skip_in_iterations=4294971408;"
|
"max_sequential_skip_in_iterations=4294971408;"
|
||||||
"arena_block_size=1893;"
|
"arena_block_size=1893;"
|
||||||
"target_file_size_multiplier=35;"
|
"target_file_size_multiplier=35;"
|
||||||
|
|||||||
@@ -115,6 +115,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
|
|||||||
{"memtable_huge_page_size", "28"},
|
{"memtable_huge_page_size", "28"},
|
||||||
{"bloom_locality", "29"},
|
{"bloom_locality", "29"},
|
||||||
{"max_successive_merges", "30"},
|
{"max_successive_merges", "30"},
|
||||||
|
{"strict_max_successive_merges", "true"},
|
||||||
{"min_partial_merge_operands", "31"},
|
{"min_partial_merge_operands", "31"},
|
||||||
{"prefix_extractor", "fixed:31"},
|
{"prefix_extractor", "fixed:31"},
|
||||||
{"experimental_mempurge_threshold", "0.003"},
|
{"experimental_mempurge_threshold", "0.003"},
|
||||||
@@ -270,6 +271,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
|
|||||||
ASSERT_EQ(new_cf_opt.memtable_huge_page_size, 28U);
|
ASSERT_EQ(new_cf_opt.memtable_huge_page_size, 28U);
|
||||||
ASSERT_EQ(new_cf_opt.bloom_locality, 29U);
|
ASSERT_EQ(new_cf_opt.bloom_locality, 29U);
|
||||||
ASSERT_EQ(new_cf_opt.max_successive_merges, 30U);
|
ASSERT_EQ(new_cf_opt.max_successive_merges, 30U);
|
||||||
|
ASSERT_EQ(new_cf_opt.strict_max_successive_merges, true);
|
||||||
ASSERT_TRUE(new_cf_opt.prefix_extractor != nullptr);
|
ASSERT_TRUE(new_cf_opt.prefix_extractor != nullptr);
|
||||||
ASSERT_EQ(new_cf_opt.optimize_filters_for_hits, true);
|
ASSERT_EQ(new_cf_opt.optimize_filters_for_hits, true);
|
||||||
ASSERT_EQ(new_cf_opt.prefix_extractor->AsString(), "rocksdb.FixedPrefix.31");
|
ASSERT_EQ(new_cf_opt.prefix_extractor->AsString(), "rocksdb.FixedPrefix.31");
|
||||||
@@ -2333,6 +2335,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
|
|||||||
{"memtable_huge_page_size", "28"},
|
{"memtable_huge_page_size", "28"},
|
||||||
{"bloom_locality", "29"},
|
{"bloom_locality", "29"},
|
||||||
{"max_successive_merges", "30"},
|
{"max_successive_merges", "30"},
|
||||||
|
{"strict_max_successive_merges", "true"},
|
||||||
{"min_partial_merge_operands", "31"},
|
{"min_partial_merge_operands", "31"},
|
||||||
{"prefix_extractor", "fixed:31"},
|
{"prefix_extractor", "fixed:31"},
|
||||||
{"experimental_mempurge_threshold", "0.003"},
|
{"experimental_mempurge_threshold", "0.003"},
|
||||||
@@ -2485,6 +2488,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
|
|||||||
ASSERT_EQ(new_cf_opt.memtable_huge_page_size, 28U);
|
ASSERT_EQ(new_cf_opt.memtable_huge_page_size, 28U);
|
||||||
ASSERT_EQ(new_cf_opt.bloom_locality, 29U);
|
ASSERT_EQ(new_cf_opt.bloom_locality, 29U);
|
||||||
ASSERT_EQ(new_cf_opt.max_successive_merges, 30U);
|
ASSERT_EQ(new_cf_opt.max_successive_merges, 30U);
|
||||||
|
ASSERT_EQ(new_cf_opt.strict_max_successive_merges, true);
|
||||||
ASSERT_TRUE(new_cf_opt.prefix_extractor != nullptr);
|
ASSERT_TRUE(new_cf_opt.prefix_extractor != nullptr);
|
||||||
ASSERT_EQ(new_cf_opt.optimize_filters_for_hits, true);
|
ASSERT_EQ(new_cf_opt.optimize_filters_for_hits, true);
|
||||||
ASSERT_EQ(new_cf_opt.prefix_extractor->AsString(), "rocksdb.FixedPrefix.31");
|
ASSERT_EQ(new_cf_opt.prefix_extractor->AsString(), "rocksdb.FixedPrefix.31");
|
||||||
|
|||||||
@@ -370,6 +370,7 @@ void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, DBOptions& db_options,
|
|||||||
cf_opt->memtable_whole_key_filtering = rnd->Uniform(2);
|
cf_opt->memtable_whole_key_filtering = rnd->Uniform(2);
|
||||||
cf_opt->enable_blob_files = rnd->Uniform(2);
|
cf_opt->enable_blob_files = rnd->Uniform(2);
|
||||||
cf_opt->enable_blob_garbage_collection = rnd->Uniform(2);
|
cf_opt->enable_blob_garbage_collection = rnd->Uniform(2);
|
||||||
|
cf_opt->strict_max_successive_merges = rnd->Uniform(2);
|
||||||
|
|
||||||
// double options
|
// double options
|
||||||
cf_opt->memtable_prefix_bloom_size_ratio =
|
cf_opt->memtable_prefix_bloom_size_ratio =
|
||||||
|
|||||||
@@ -1648,6 +1648,10 @@ DEFINE_int32(max_successive_merges, 0,
|
|||||||
"Maximum number of successive merge operations on a key in the "
|
"Maximum number of successive merge operations on a key in the "
|
||||||
"memtable");
|
"memtable");
|
||||||
|
|
||||||
|
DEFINE_bool(strict_max_successive_merges, false,
|
||||||
|
"Whether to issue filesystem reads to keep within "
|
||||||
|
"`max_successive_merges` limit");
|
||||||
|
|
||||||
static bool ValidatePrefixSize(const char* flagname, int32_t value) {
|
static bool ValidatePrefixSize(const char* flagname, int32_t value) {
|
||||||
if (value < 0 || value >= 2000000000) {
|
if (value < 0 || value >= 2000000000) {
|
||||||
fprintf(stderr, "Invalid value for --%s: %d. 0<= PrefixSize <=2000000000\n",
|
fprintf(stderr, "Invalid value for --%s: %d. 0<= PrefixSize <=2000000000\n",
|
||||||
@@ -4639,6 +4643,7 @@ class Benchmark {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
options.max_successive_merges = FLAGS_max_successive_merges;
|
options.max_successive_merges = FLAGS_max_successive_merges;
|
||||||
|
options.strict_max_successive_merges = FLAGS_strict_max_successive_merges;
|
||||||
options.report_bg_io_stats = FLAGS_report_bg_io_stats;
|
options.report_bg_io_stats = FLAGS_report_bg_io_stats;
|
||||||
|
|
||||||
// set universal style compaction configurations, if applicable
|
// set universal style compaction configurations, if applicable
|
||||||
|
|||||||
@@ -1 +0,0 @@
|
|||||||
`rocksdb.blobdb.blob.file.write.micros` expands to also measure time writing the header and footer. Therefore the COUNT may be higher and values may be smaller than before. For stacked BlobDB, it no longer measures the time of explictly flushing blob file.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Files will be compacted to the next level if the data age exceeds periodic_compaction_seconds except for the last level.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Reduced the compaction debt ratio trigger for scheduling parallel compactions
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
* For leveled compaction with default compaction pri (kMinOverlappingRatio), files marked for compaction will be prioritized over files not marked when picking a file from a level for compaction.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fix bug in auto_readahead_size that combined with IndexType::kBinarySearchWithFirstKey + fails or iterator lands at a wrong key
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fixed some cases in which DB file corruption was detected but ignored on creating a backup with BackupEngine.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fix bugs where `rocksdb.blobdb.blob.file.synced` includes blob files failed to get synced and `rocksdb.blobdb.blob.file.bytes.written` includes blob bytes failed to get written.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fixed a possible memory leak or crash on a failure (such as I/O error) in automatic atomic flush of multiple column families.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fixed some cases of in-memory data corruption using mmap reads with `BackupEngine`, `sst_dump`, or `ldb`.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fixed issues with experimental `preclude_last_level_data_seconds` option that could interfere with expected data tiering.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Fixed the handling of the edge case when all existing blob files become unreferenced. Such files are now correctly deleted.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Add new statistics: `rocksdb.sst.write.micros` measures time of each write to SST file; `rocksdb.file.write.{flush|compaction|db.open}.micros` measure time of each write to SST table (currently only block-based table format) and blob file for flush, compaction and db open.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Add CompressionOptions to the CompressedSecondaryCacheOptions structure to allow users to specify library specific options when creating the compressed secondary cache.
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Deprecated several options: `level_compaction_dynamic_file_size`, `ignore_max_compaction_bytes_for_input`, `check_flush_compaction_key_order`, `flush_verify_memtable_count`, `compaction_verify_record_count`, `fail_if_options_file_error`, and `enforce_single_del_contracts`
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
Exposed options ttl via c api.
|
|
||||||
@@ -181,6 +181,40 @@ class TestFs : public FileSystemWrapper {
|
|||||||
bool fail_reads_;
|
bool fail_reads_;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
class CheckIOOptsSequentialFile : public FSSequentialFileOwnerWrapper {
|
||||||
|
public:
|
||||||
|
CheckIOOptsSequentialFile(std::unique_ptr<FSSequentialFile>&& f,
|
||||||
|
const std::string& file_name)
|
||||||
|
: FSSequentialFileOwnerWrapper(std::move(f)) {
|
||||||
|
is_sst_file_ = file_name.find(".sst") != std::string::npos;
|
||||||
|
}
|
||||||
|
|
||||||
|
IOStatus Read(size_t n, const IOOptions& options, Slice* result,
|
||||||
|
char* scratch, IODebugContext* dbg) override {
|
||||||
|
// Backup currently associates only SST read with rate limiter priority
|
||||||
|
assert(!is_sst_file_ || options.rate_limiter_priority ==
|
||||||
|
kExpectedBackupReadRateLimiterPri);
|
||||||
|
IOStatus rv = target()->Read(n, options, result, scratch, dbg);
|
||||||
|
return rv;
|
||||||
|
}
|
||||||
|
|
||||||
|
IOStatus PositionedRead(uint64_t offset, size_t n, const IOOptions& options,
|
||||||
|
Slice* result, char* scratch,
|
||||||
|
IODebugContext* dbg) override {
|
||||||
|
// Backup currently associates only SST read with rate limiter priority
|
||||||
|
assert(!is_sst_file_ || options.rate_limiter_priority ==
|
||||||
|
kExpectedBackupReadRateLimiterPri);
|
||||||
|
IOStatus rv =
|
||||||
|
target()->PositionedRead(offset, n, options, result, scratch, dbg);
|
||||||
|
return rv;
|
||||||
|
}
|
||||||
|
|
||||||
|
private:
|
||||||
|
static const Env::IOPriority kExpectedBackupReadRateLimiterPri =
|
||||||
|
Env::IO_LOW;
|
||||||
|
bool is_sst_file_;
|
||||||
|
};
|
||||||
|
|
||||||
IOStatus NewSequentialFile(const std::string& f, const FileOptions& file_opts,
|
IOStatus NewSequentialFile(const std::string& f, const FileOptions& file_opts,
|
||||||
std::unique_ptr<FSSequentialFile>* r,
|
std::unique_ptr<FSSequentialFile>* r,
|
||||||
IODebugContext* dbg) override {
|
IODebugContext* dbg) override {
|
||||||
@@ -189,6 +223,14 @@ class TestFs : public FileSystemWrapper {
|
|||||||
r->reset(
|
r->reset(
|
||||||
new TestFs::DummySequentialFile(dummy_sequential_file_fail_reads_));
|
new TestFs::DummySequentialFile(dummy_sequential_file_fail_reads_));
|
||||||
return IOStatus::OK();
|
return IOStatus::OK();
|
||||||
|
} else if (check_iooptions_sequential_file_) {
|
||||||
|
std::unique_ptr<FSSequentialFile> file;
|
||||||
|
IOStatus s =
|
||||||
|
FileSystemWrapper::NewSequentialFile(f, file_opts, &file, dbg);
|
||||||
|
if (s.ok()) {
|
||||||
|
r->reset(new TestFs::CheckIOOptsSequentialFile(std::move(file), f));
|
||||||
|
}
|
||||||
|
return s;
|
||||||
} else {
|
} else {
|
||||||
IOStatus s = FileSystemWrapper::NewSequentialFile(f, file_opts, r, dbg);
|
IOStatus s = FileSystemWrapper::NewSequentialFile(f, file_opts, r, dbg);
|
||||||
if (s.ok()) {
|
if (s.ok()) {
|
||||||
@@ -292,6 +334,11 @@ class TestFs : public FileSystemWrapper {
|
|||||||
dummy_sequential_file_fail_reads_ = dummy_sequential_file_fail_reads;
|
dummy_sequential_file_fail_reads_ = dummy_sequential_file_fail_reads;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void SetCheckIOOptionsSequentialFile(bool check_iooptions_sequential_file) {
|
||||||
|
MutexLock l(&mutex_);
|
||||||
|
check_iooptions_sequential_file_ = check_iooptions_sequential_file;
|
||||||
|
}
|
||||||
|
|
||||||
void SetGetChildrenFailure(bool fail) { get_children_failure_ = fail; }
|
void SetGetChildrenFailure(bool fail) { get_children_failure_ = fail; }
|
||||||
IOStatus GetChildren(const std::string& dir, const IOOptions& io_opts,
|
IOStatus GetChildren(const std::string& dir, const IOOptions& io_opts,
|
||||||
std::vector<std::string>* r,
|
std::vector<std::string>* r,
|
||||||
@@ -387,6 +434,7 @@ class TestFs : public FileSystemWrapper {
|
|||||||
port::Mutex mutex_;
|
port::Mutex mutex_;
|
||||||
bool dummy_sequential_file_ = false;
|
bool dummy_sequential_file_ = false;
|
||||||
bool dummy_sequential_file_fail_reads_ = false;
|
bool dummy_sequential_file_fail_reads_ = false;
|
||||||
|
bool check_iooptions_sequential_file_ = false;
|
||||||
std::vector<std::string> written_files_;
|
std::vector<std::string> written_files_;
|
||||||
std::vector<std::string> filenames_for_mocked_attrs_;
|
std::vector<std::string> filenames_for_mocked_attrs_;
|
||||||
uint64_t limit_written_files_ = 1000000;
|
uint64_t limit_written_files_ = 1000000;
|
||||||
@@ -1184,7 +1232,8 @@ TEST_P(BackupEngineTestWithParam, OnlineIntegrationTest) {
|
|||||||
// restore)
|
// restore)
|
||||||
// options_.db_paths.emplace_back(dbname_, 500 * 1024);
|
// options_.db_paths.emplace_back(dbname_, 500 * 1024);
|
||||||
// options_.db_paths.emplace_back(dbname_ + "_2", 1024 * 1024 * 1024);
|
// options_.db_paths.emplace_back(dbname_ + "_2", 1024 * 1024 * 1024);
|
||||||
|
test_db_fs_->SetCheckIOOptionsSequentialFile(true);
|
||||||
|
test_backup_fs_->SetCheckIOOptionsSequentialFile(true);
|
||||||
OpenDBAndBackupEngine(true);
|
OpenDBAndBackupEngine(true);
|
||||||
// write some data, backup, repeat
|
// write some data, backup, repeat
|
||||||
for (int i = 0; i < 5; ++i) {
|
for (int i = 0; i < 5; ++i) {
|
||||||
@@ -1241,6 +1290,8 @@ TEST_P(BackupEngineTestWithParam, OnlineIntegrationTest) {
|
|||||||
AssertBackupConsistency(0, 0, 3 * keys_iteration, max_key);
|
AssertBackupConsistency(0, 0, 3 * keys_iteration, max_key);
|
||||||
|
|
||||||
CloseBackupEngine();
|
CloseBackupEngine();
|
||||||
|
test_db_fs_->SetCheckIOOptionsSequentialFile(false);
|
||||||
|
test_backup_fs_->SetCheckIOOptionsSequentialFile(false);
|
||||||
}
|
}
|
||||||
#endif // !defined(ROCKSDB_VALGRIND_RUN) || defined(ROCKSDB_FULL_VALGRIND_RUN)
|
#endif // !defined(ROCKSDB_VALGRIND_RUN) || defined(ROCKSDB_FULL_VALGRIND_RUN)
|
||||||
|
|
||||||
|
|||||||
@@ -250,7 +250,7 @@ class BlobDB : public StackableDB {
|
|||||||
|
|
||||||
virtual Status SyncBlobFiles(const WriteOptions& write_options) = 0;
|
virtual Status SyncBlobFiles(const WriteOptions& write_options) = 0;
|
||||||
|
|
||||||
virtual ~BlobDB() {}
|
~BlobDB() override {}
|
||||||
|
|
||||||
protected:
|
protected:
|
||||||
explicit BlobDB();
|
explicit BlobDB();
|
||||||
|
|||||||
Reference in New Issue
Block a user