mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a34ec68f26 | |||
| 9e09d493d0 | |||
| e510cecc1b | |||
| 8034cbc95d |
@@ -53,6 +53,7 @@ TESTS = \
|
||||
memenv_test \
|
||||
skiplist_test \
|
||||
table_test \
|
||||
ttl_test \
|
||||
block_test \
|
||||
version_edit_test \
|
||||
version_set_test \
|
||||
@@ -182,6 +183,9 @@ crc32c_test: util/crc32c_test.o $(LIBOBJECTS) $(TESTHARNESS)
|
||||
db_test: db/db_test.o $(LIBOBJECTS) $(TESTHARNESS)
|
||||
$(CXX) db/db_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
|
||||
|
||||
ttl_test: utilities/ttl/ttl_test.o $(LIBOBJECTS) $(TESTHARNESS)
|
||||
$(CXX) utilities/ttl/ttl_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
|
||||
|
||||
dbformat_test: db/dbformat_test.o $(LIBOBJECTS) $(TESTHARNESS)
|
||||
$(CXX) db/dbformat_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
|
||||
|
||||
|
||||
@@ -130,7 +130,7 @@ esac
|
||||
# except for the test and benchmark files. By default, find will output a list
|
||||
# of all files matching either rule, so we need to append -print to make the
|
||||
# prune take effect.
|
||||
DIRS="util db table"
|
||||
DIRS="util db table utilities"
|
||||
if test "$USE_THRIFT"; then
|
||||
DIRS="$DIRS thrift/server_utils.cpp thrift/gen-cpp "
|
||||
THRIFTSERVER=leveldb_server
|
||||
|
||||
+14
-9
@@ -1579,6 +1579,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
|
||||
SequenceNumber last_sequence_for_key __attribute__((unused)) =
|
||||
kMaxSequenceNumber;
|
||||
SequenceNumber visible_in_snapshot = kMaxSequenceNumber;
|
||||
std::string compaction_filter_value;
|
||||
for (; input->Valid() && !shutting_down_.Acquire_Load(); ) {
|
||||
// Prioritize immutable compaction work
|
||||
if (imm_.imm_flush_needed.NoBarrier_Load() != nullptr) {
|
||||
@@ -1594,7 +1595,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
|
||||
|
||||
Slice key = input->key();
|
||||
Slice value = input->value();
|
||||
Slice* compaction_filter_value = nullptr;
|
||||
|
||||
if (compact->compaction->ShouldStopBefore(key) &&
|
||||
compact->builder != nullptr) {
|
||||
status = FinishCompactionOutputFile(compact, input.get());
|
||||
@@ -1653,20 +1654,24 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
|
||||
ikey.type != kTypeDeletion &&
|
||||
ikey.sequence < earliest_snapshot) {
|
||||
// If the user has specified a compaction filter, then invoke
|
||||
// it. If this key is not visible via any snapshot and the
|
||||
// return value of the compaction filter is true and then
|
||||
// it. If the return value of the compaction filter is true,
|
||||
// drop this key from the output.
|
||||
bool value_changed = false;
|
||||
compaction_filter_value.clear();
|
||||
drop = options_.CompactionFilter(options_.compaction_filter_args,
|
||||
compact->compaction->level(),
|
||||
ikey.user_key, value, &compaction_filter_value);
|
||||
|
||||
compact->compaction->level(),
|
||||
ikey.user_key, value,
|
||||
&compaction_filter_value,
|
||||
&value_changed);
|
||||
// Another example of statistics update without holding the lock
|
||||
// TODO: clean it up
|
||||
if (drop) {
|
||||
RecordTick(options_.statistics, COMPACTION_KEY_DROP_USER);
|
||||
}
|
||||
|
||||
// If the application wants to change the value, then do so here.
|
||||
if (compaction_filter_value != nullptr) {
|
||||
value = *compaction_filter_value;
|
||||
delete compaction_filter_value;
|
||||
if (value_changed) {
|
||||
value = compaction_filter_value;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+8
-4
@@ -1351,22 +1351,26 @@ TEST(DBTest, RepeatedWritesToSameKey) {
|
||||
static int cfilter_count;
|
||||
static std::string NEW_VALUE = "NewValue";
|
||||
static bool keep_filter(void* arg, int level, const Slice& key,
|
||||
const Slice& value, Slice** new_value) {
|
||||
const Slice& value, std::string* new_value,
|
||||
bool* value_changed) {
|
||||
assert(arg == nullptr);
|
||||
cfilter_count++;
|
||||
return false;
|
||||
}
|
||||
static bool delete_filter(void*argv, int level, const Slice& key,
|
||||
const Slice& value, Slice** new_value) {
|
||||
const Slice& value, std::string* new_value,
|
||||
bool* value_changed) {
|
||||
assert(argv == nullptr);
|
||||
cfilter_count++;
|
||||
return true;
|
||||
}
|
||||
static bool change_filter(void*argv, int level, const Slice& key,
|
||||
const Slice& value, Slice** new_value) {
|
||||
const Slice& value, std::string* new_value,
|
||||
bool* value_changed) {
|
||||
assert(argv == (void*)100);
|
||||
assert(new_value != nullptr);
|
||||
*new_value = new Slice(NEW_VALUE);
|
||||
*new_value = NEW_VALUE;
|
||||
*value_changed = true;
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
@@ -356,22 +356,25 @@ struct Options {
|
||||
|
||||
// This method allows an application to modify/delete a key-value at
|
||||
// the time of compaction. The compaction process invokes this
|
||||
// method for every kv that is being compacted. A return value
|
||||
// method for kv that is being compacted. A return value
|
||||
// of false indicates that the kv should be preserved in the
|
||||
// output of this compaction run and a return value of true
|
||||
// indicates that this key-value should be removed from the
|
||||
// output of the compaction. The application can inspect
|
||||
// the existing value of the key, modify it if needed and
|
||||
// return back the new value for this key. The application
|
||||
// should allocate memory for the Slice object that is used to
|
||||
// return the new value and the leveldb framework will
|
||||
// free up that memory.
|
||||
// the existing value of the key and make decision based on it.
|
||||
|
||||
// When the value is to be preserved, the application has the option
|
||||
// to modify the existing_value and pass it back through new_value.
|
||||
// value_changed needs to be set to true in this case.
|
||||
|
||||
// The compaction_filter_args, if specified here, are passed
|
||||
// back to the invocation of the CompactionFilter.
|
||||
void* compaction_filter_args;
|
||||
bool (*CompactionFilter)(void* compaction_filter_args,
|
||||
int level, const Slice& key,
|
||||
const Slice& existing_value, Slice** new_value);
|
||||
int level, const Slice& key,
|
||||
const Slice& existing_value,
|
||||
std::string* new_value,
|
||||
bool* value_changed);
|
||||
|
||||
// Disable automatic compactions. Manual compactions can still
|
||||
// be issued on this database.
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
||||
|
||||
#ifndef LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_
|
||||
#define LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_
|
||||
|
||||
#include "include/leveldb/db.h"
|
||||
|
||||
namespace leveldb {
|
||||
|
||||
// This class contains APIs to open leveldb with specific support eg. TTL
|
||||
class UtilityDB {
|
||||
|
||||
public:
|
||||
// Open the database with TTL support.
|
||||
//
|
||||
// USE-CASES:
|
||||
// This API should be used to open the db when key-values inserted are
|
||||
// meant to be removed from the db in a non-strict 'ttl' amount of time
|
||||
// Therefore, this guarantees that key-values inserted will remain in the
|
||||
// db for >= ttl amount of time and the db will make efforts to remove the
|
||||
// key-values as soon as possible after ttl seconds of their insertion.
|
||||
//
|
||||
// BEHAVIOUR:
|
||||
// TTL is accepted in seconds
|
||||
// (int32_t)Timestamp(creation) is suffixed to values in Put internally
|
||||
// Expired TTL values deleted in compaction only:(Timestamp+ttl<time_now)
|
||||
// Get/Iterator may return expired entries(compaction not run on them yet)
|
||||
// Different TTL may be used during different Opens
|
||||
// Example: Open1 at t=0 with ttl=4 and insert k1,k2, close at t=2
|
||||
// Open2 at t=3 with ttl=5. Now k1,k2 should be deleted at t>=5
|
||||
//
|
||||
// CONSTRAINTS:
|
||||
// The caller must not specify any compaction-filter in options
|
||||
// Not specifying/passing or non-positive TTL behaves like TTL = infinity
|
||||
//
|
||||
// !!!WARNING!!!:
|
||||
// Calling DB::Open directly to re-open a db created by this API will get
|
||||
// corrupt values(timestamp suffixed) and no ttl effect will be there
|
||||
// during the second Open, so use this API consistently to open the db
|
||||
// Be careful when passing ttl with a small positive value because the
|
||||
// whole database may be deleted in a small amount of time
|
||||
static Status OpenTtlDB(const Options& options,
|
||||
const std::string& name,
|
||||
DB** dbptr,
|
||||
int32_t ttl = 0);
|
||||
};
|
||||
|
||||
} // namespace leveldb
|
||||
|
||||
#endif // LEVELDB_INCLUDE_UTILITIES_UTILITY_DB_H_
|
||||
+26
-3
@@ -21,7 +21,7 @@
|
||||
#include "db/version_set.h"
|
||||
#include "db/db_statistics.h"
|
||||
#include "leveldb/cache.h"
|
||||
#include "leveldb/db.h"
|
||||
#include "utilities/utility_db.h"
|
||||
#include "leveldb/env.h"
|
||||
#include "leveldb/write_batch.h"
|
||||
#include "leveldb/statistics.h"
|
||||
@@ -31,6 +31,8 @@
|
||||
#include "util/mutexlock.h"
|
||||
#include "util/random.h"
|
||||
#include "util/testutil.h"
|
||||
#include "util/logging.h"
|
||||
#include "utilities/ttl/db_ttl.h"
|
||||
#include "hdfs/env_hdfs.h"
|
||||
|
||||
static const long KB = 1024;
|
||||
@@ -54,6 +56,11 @@ static bool FLAGS_test_batches_snapshots = false;
|
||||
// Number of concurrent threads to run.
|
||||
static int FLAGS_threads = 32;
|
||||
|
||||
// Opens the db with this ttl value if this is not -1
|
||||
// Carefully specify a large value such that verifications on deleted
|
||||
// values don't fail
|
||||
static int FLAGS_ttl = -1;
|
||||
|
||||
// Size of each value will be this number times rand_int(1,3) bytes
|
||||
static int FLAGS_value_size_mult = 8;
|
||||
|
||||
@@ -870,6 +877,11 @@ class StressTest {
|
||||
kMajorVersion, kMinorVersion);
|
||||
fprintf(stdout, "Number of threads : %d\n", FLAGS_threads);
|
||||
fprintf(stdout, "Ops per thread : %d\n", FLAGS_ops_per_thread);
|
||||
std::string ttl_state("unused");
|
||||
if (FLAGS_ttl > 0) {
|
||||
ttl_state = NumberToString(FLAGS_ttl);
|
||||
}
|
||||
fprintf(stdout, "Time to live(sec) : %s\n", ttl_state.c_str());
|
||||
fprintf(stdout, "Read percentage : %d\n", FLAGS_readpercent);
|
||||
fprintf(stdout, "Write-buffer-size : %d\n", FLAGS_write_buffer_size);
|
||||
fprintf(stdout, "Delete percentage : %d\n", FLAGS_delpercent);
|
||||
@@ -936,7 +948,12 @@ class StressTest {
|
||||
if (purge_percent.Uniform(100) < FLAGS_purge_redundant_percent - 1) {
|
||||
options.purge_redundant_kvs_while_flush = false;
|
||||
}
|
||||
Status s = DB::Open(options, FLAGS_db, &db_);
|
||||
Status s;
|
||||
if (FLAGS_ttl == -1) {
|
||||
s = DB::Open(options, FLAGS_db, &db_);
|
||||
} else {
|
||||
s = UtilityDB::OpenTtlDB(options, FLAGS_db, &db_, FLAGS_ttl);
|
||||
}
|
||||
if (!s.ok()) {
|
||||
fprintf(stderr, "open error: %s\n", s.ToString().c_str());
|
||||
exit(1);
|
||||
@@ -946,7 +963,11 @@ class StressTest {
|
||||
void Reopen() {
|
||||
// do not close the db. Just delete the lock file. This
|
||||
// simulates a crash-recovery kind of situation.
|
||||
((DBImpl*) db_)->TEST_Destroy_DBImpl();
|
||||
if (FLAGS_ttl != -1) {
|
||||
((DBWithTTL*) db_)->TEST_Destroy_DBWithTtl();
|
||||
} else {
|
||||
((DBImpl*) db_)->TEST_Destroy_DBImpl();
|
||||
}
|
||||
db_ = nullptr;
|
||||
|
||||
num_times_reopened_++;
|
||||
@@ -1017,6 +1038,8 @@ int main(int argc, char** argv) {
|
||||
FLAGS_test_batches_snapshots = n;
|
||||
} else if (sscanf(argv[i], "--threads=%d%c", &n, &junk) == 1) {
|
||||
FLAGS_threads = n;
|
||||
} else if (sscanf(argv[i], "--ttl=%d%c", &n, &junk) == 1) {
|
||||
FLAGS_ttl = n;
|
||||
} else if (sscanf(argv[i], "--value_size_mult=%d%c", &n, &junk) == 1) {
|
||||
FLAGS_value_size_mult = n;
|
||||
} else if (sscanf(argv[i], "--write_buffer_size=%d%c", &n, &junk) == 1) {
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
||||
|
||||
#include "utilities/ttl/db_ttl.h"
|
||||
#include "include/utilities/utility_db.h"
|
||||
#include "db/filename.h"
|
||||
#include "util/coding.h"
|
||||
#include "include/leveldb/env.h"
|
||||
#include "include/leveldb/iterator.h"
|
||||
|
||||
namespace leveldb {
|
||||
|
||||
class TtlIterator : public Iterator {
|
||||
|
||||
public:
|
||||
TtlIterator(Iterator* iter, int32_t ts_len)
|
||||
: iter_(iter),
|
||||
ts_len_(ts_len) {
|
||||
assert(iter_);
|
||||
}
|
||||
|
||||
~TtlIterator() {
|
||||
delete iter_;
|
||||
}
|
||||
|
||||
bool Valid() const {
|
||||
return iter_->Valid();
|
||||
}
|
||||
|
||||
void SeekToFirst() {
|
||||
iter_->SeekToFirst();
|
||||
}
|
||||
|
||||
void SeekToLast() {
|
||||
iter_->SeekToLast();
|
||||
}
|
||||
|
||||
void Seek(const Slice& target) {
|
||||
iter_->Seek(target);
|
||||
}
|
||||
|
||||
void Next() {
|
||||
iter_->Next();
|
||||
}
|
||||
|
||||
void Prev() {
|
||||
iter_->Prev();
|
||||
}
|
||||
|
||||
Slice key() const {
|
||||
return iter_->key();
|
||||
}
|
||||
|
||||
Slice value() const {
|
||||
assert(DBWithTTL::SanityCheckTimestamp(iter_->value().ToString()).ok());
|
||||
Slice trimmed_value = iter_->value();
|
||||
trimmed_value.size_ -= ts_len_;
|
||||
return trimmed_value;
|
||||
}
|
||||
|
||||
Status status() const {
|
||||
return iter_->status();
|
||||
}
|
||||
|
||||
private:
|
||||
Iterator* iter_;
|
||||
int32_t ts_len_;
|
||||
};
|
||||
|
||||
// Open the db inside DBWithTTL because options needs pointer to its ttl
|
||||
DBWithTTL::DBWithTTL(const int32_t ttl,
|
||||
const Options& options,
|
||||
const std::string& dbname,
|
||||
Status& st)
|
||||
: ttl_(ttl) {
|
||||
assert(options.CompactionFilter == nullptr);
|
||||
Options options_to_open = options;
|
||||
options_to_open.compaction_filter_args = &ttl_;
|
||||
options_to_open.CompactionFilter = DeleteByTS;
|
||||
st = DB::Open(options_to_open, dbname, &db_);
|
||||
}
|
||||
|
||||
DBWithTTL::~DBWithTTL() {
|
||||
delete db_;
|
||||
}
|
||||
|
||||
Status UtilityDB::OpenTtlDB(
|
||||
const Options& options,
|
||||
const std::string& dbname,
|
||||
DB** dbptr,
|
||||
int32_t ttl) {
|
||||
Status st;
|
||||
*dbptr = new DBWithTTL(ttl, options, dbname, st);
|
||||
if (!st.ok()) {
|
||||
delete dbptr;
|
||||
}
|
||||
return st;
|
||||
}
|
||||
|
||||
// returns true(i.e. key-value to be deleted) if its TS has expired based on ttl
|
||||
bool DBWithTTL::DeleteByTS(
|
||||
void* args,
|
||||
int level,
|
||||
const Slice& key,
|
||||
const Slice& old_val,
|
||||
std::string* new_val,
|
||||
bool* value_changed) {
|
||||
return IsStale(old_val, *(int32_t*)args);
|
||||
}
|
||||
|
||||
// Gives back the current time
|
||||
Status DBWithTTL::GetCurrentTime(int32_t& curtime) {
|
||||
return Env::Default()->GetCurrentTime((int64_t*)&curtime);
|
||||
}
|
||||
|
||||
// Appends the current timestamp to the string.
|
||||
// Returns false if could not get the current_time, true if append succeeds
|
||||
Status DBWithTTL::AppendTS(const Slice& val, std::string& val_with_ts) {
|
||||
val_with_ts.reserve(kTSLength + val.size());
|
||||
char ts_string[kTSLength];
|
||||
int32_t curtime;
|
||||
Status st = GetCurrentTime(curtime);
|
||||
if (!st.ok()) {
|
||||
return st;
|
||||
}
|
||||
EncodeFixed32(ts_string, curtime);
|
||||
val_with_ts.append(val.data(), val.size());
|
||||
val_with_ts.append(ts_string, kTSLength);
|
||||
return st;
|
||||
}
|
||||
|
||||
// Returns corruption if the length of the string is lesser than timestamp, or
|
||||
// timestamp refers to a time lesser than ttl-feature release time
|
||||
Status DBWithTTL::SanityCheckTimestamp(const std::string& str) {
|
||||
if (str.length() < (unsigned)kTSLength) {
|
||||
return Status::Corruption("Error: value's length less than timestamp's\n");
|
||||
}
|
||||
// Checks that TS is not lesser than kMinTimestamp
|
||||
// Gaurds against corruption & normal database opened incorrectly in ttl mode
|
||||
int32_t timestamp_value =
|
||||
DecodeFixed32(str.data() + str.size() - kTSLength);
|
||||
if (timestamp_value < kMinTimestamp){
|
||||
return Status::Corruption("Error: Timestamp < ttl feature release time!\n");
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
// Checks if the string is stale or not according to TTl provided
|
||||
bool DBWithTTL::IsStale(const Slice& value, int32_t ttl) {
|
||||
if (ttl <= 0) { // Data is fresh if TTL is non-positive
|
||||
return false;
|
||||
}
|
||||
int32_t curtime;
|
||||
if (!GetCurrentTime(curtime).ok()) {
|
||||
return false; // Treat the data as fresh if could not get current time
|
||||
} else {
|
||||
int32_t timestamp_value =
|
||||
DecodeFixed32(value.data() + value.size() - kTSLength);
|
||||
if ((timestamp_value + ttl) < curtime) {
|
||||
return true; // Data is stale
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Strips the TS from the end of the string
|
||||
Status DBWithTTL::StripTS(std::string* str) {
|
||||
Status st;
|
||||
// Erasing characters which hold the TS
|
||||
str->erase(str->length() - kTSLength, kTSLength);
|
||||
return st;
|
||||
}
|
||||
|
||||
Status DBWithTTL::Put(
|
||||
const WriteOptions& o,
|
||||
const Slice& key,
|
||||
const Slice& val) {
|
||||
std::string value_with_ts;
|
||||
Status st = AppendTS(val, value_with_ts);
|
||||
if (!st.ok()) {
|
||||
return st;
|
||||
}
|
||||
return db_->Put(o, key, value_with_ts);
|
||||
}
|
||||
|
||||
Status DBWithTTL::Get(const ReadOptions& options,
|
||||
const Slice& key,
|
||||
std::string* value) {
|
||||
Status st = db_->Get(options, key, value);
|
||||
if (!st.ok()) {
|
||||
return st;
|
||||
}
|
||||
st = SanityCheckTimestamp(*value);
|
||||
if (!st.ok()) {
|
||||
return st;
|
||||
}
|
||||
return StripTS(value);
|
||||
}
|
||||
|
||||
Status DBWithTTL::Delete(const WriteOptions& wopts, const Slice& key) {
|
||||
return db_->Delete(wopts, key);
|
||||
}
|
||||
|
||||
Status DBWithTTL::Write(const WriteOptions& opts, WriteBatch* updates) {
|
||||
return db_->Write(opts, updates);
|
||||
}
|
||||
|
||||
Iterator* DBWithTTL::NewIterator(const ReadOptions& opts) {
|
||||
return new TtlIterator(db_->NewIterator(opts), kTSLength);
|
||||
}
|
||||
|
||||
const Snapshot* DBWithTTL::GetSnapshot() {
|
||||
return db_->GetSnapshot();
|
||||
}
|
||||
|
||||
void DBWithTTL::ReleaseSnapshot(const Snapshot* snapshot) {
|
||||
db_->ReleaseSnapshot(snapshot);
|
||||
}
|
||||
|
||||
bool DBWithTTL::GetProperty(const Slice& property, std::string* value) {
|
||||
return db_->GetProperty(property, value);
|
||||
}
|
||||
|
||||
void DBWithTTL::GetApproximateSizes(const Range* r, int n, uint64_t* sizes) {
|
||||
db_->GetApproximateSizes(r, n, sizes);
|
||||
}
|
||||
|
||||
void DBWithTTL::CompactRange(const Slice* begin, const Slice* end) {
|
||||
db_->CompactRange(begin, end);
|
||||
}
|
||||
|
||||
int DBWithTTL::NumberLevels() {
|
||||
return db_->NumberLevels();
|
||||
}
|
||||
|
||||
int DBWithTTL::MaxMemCompactionLevel() {
|
||||
return db_->MaxMemCompactionLevel();
|
||||
}
|
||||
|
||||
int DBWithTTL::Level0StopWriteTrigger() {
|
||||
return db_->Level0StopWriteTrigger();
|
||||
}
|
||||
|
||||
Status DBWithTTL::Flush(const FlushOptions& fopts) {
|
||||
return db_->Flush(fopts);
|
||||
}
|
||||
|
||||
Status DBWithTTL::DisableFileDeletions() {
|
||||
return db_->DisableFileDeletions();
|
||||
}
|
||||
|
||||
Status DBWithTTL::EnableFileDeletions() {
|
||||
return db_->EnableFileDeletions();
|
||||
}
|
||||
|
||||
Status DBWithTTL::GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs) {
|
||||
return db_->GetLiveFiles(vec, mfs);
|
||||
}
|
||||
|
||||
SequenceNumber DBWithTTL::GetLatestSequenceNumber() {
|
||||
return db_->GetLatestSequenceNumber();
|
||||
}
|
||||
|
||||
Status DBWithTTL::GetUpdatesSince(
|
||||
SequenceNumber seq_number,
|
||||
unique_ptr<TransactionLogIterator>* iter) {
|
||||
return db_->GetUpdatesSince(seq_number, iter);
|
||||
}
|
||||
|
||||
void DBWithTTL::TEST_Destroy_DBWithTtl() {
|
||||
((DBImpl*) db_)->TEST_Destroy_DBImpl();
|
||||
}
|
||||
|
||||
} // namespace leveldb
|
||||
@@ -0,0 +1,95 @@
|
||||
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
||||
|
||||
#ifndef LEVELDB_UTILITIES_TTL_DB_TTL_H_
|
||||
#define LEVELDB_UTILITIES_TTL_DB_TTL_H_
|
||||
|
||||
#include "include/leveldb/db.h"
|
||||
#include "db/db_impl.h"
|
||||
|
||||
namespace leveldb {
|
||||
|
||||
class DBWithTTL : public DB {
|
||||
public:
|
||||
DBWithTTL(const int32_t ttl,
|
||||
const Options& options,
|
||||
const std::string& dbname,
|
||||
Status& st);
|
||||
|
||||
virtual ~DBWithTTL();
|
||||
|
||||
virtual Status Put(const WriteOptions& o,
|
||||
const Slice& key,
|
||||
const Slice& val);
|
||||
|
||||
virtual Status Get(const ReadOptions& options,
|
||||
const Slice& key,
|
||||
std::string* value);
|
||||
|
||||
virtual Status Delete(const WriteOptions& wopts, const Slice& key);
|
||||
|
||||
virtual Status Write(const WriteOptions& opts, WriteBatch* updates);
|
||||
|
||||
virtual Iterator* NewIterator(const ReadOptions& opts);
|
||||
|
||||
virtual const Snapshot* GetSnapshot();
|
||||
|
||||
virtual void ReleaseSnapshot(const Snapshot* snapshot);
|
||||
|
||||
virtual bool GetProperty(const Slice& property, std::string* value);
|
||||
|
||||
virtual void GetApproximateSizes(const Range* r, int n, uint64_t* sizes);
|
||||
|
||||
virtual void CompactRange(const Slice* begin, const Slice* end);
|
||||
|
||||
virtual int NumberLevels();
|
||||
|
||||
virtual int MaxMemCompactionLevel();
|
||||
|
||||
virtual int Level0StopWriteTrigger();
|
||||
|
||||
virtual Status Flush(const FlushOptions& fopts);
|
||||
|
||||
virtual Status DisableFileDeletions();
|
||||
|
||||
virtual Status EnableFileDeletions();
|
||||
|
||||
virtual Status GetLiveFiles(std::vector<std::string>& vec, uint64_t* mfs);
|
||||
|
||||
virtual SequenceNumber GetLatestSequenceNumber();
|
||||
|
||||
virtual Status GetUpdatesSince(SequenceNumber seq_number,
|
||||
unique_ptr<TransactionLogIterator>* iter);
|
||||
|
||||
// Simulate a db crash, no elegant closing of database.
|
||||
void TEST_Destroy_DBWithTtl();
|
||||
|
||||
static bool DeleteByTS(void* args,
|
||||
int level,
|
||||
const Slice& key,
|
||||
const Slice& old_val,
|
||||
std::string* new_val,
|
||||
bool* value_changed);
|
||||
|
||||
static bool IsStale(const Slice& value, int32_t ttl);
|
||||
|
||||
static Status AppendTS(const Slice& val, std::string& val_with_ts);
|
||||
|
||||
static Status SanityCheckTimestamp(const std::string& str);
|
||||
|
||||
static Status StripTS(std::string* str);
|
||||
|
||||
static Status GetCurrentTime(int32_t& curtime);
|
||||
|
||||
static const int32_t kTSLength = sizeof(int32_t); // size of timestamp
|
||||
|
||||
static const int32_t kMinTimestamp = 1368146402; // 05/09/2013:5:40PM
|
||||
|
||||
private:
|
||||
DB* db_;
|
||||
int32_t ttl_;
|
||||
};
|
||||
|
||||
}
|
||||
#endif // LEVELDB_UTILITIES_TTL_DB_TTL_H_
|
||||
@@ -0,0 +1,281 @@
|
||||
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file. See the AUTHORS file for names of contributors.
|
||||
|
||||
#include "include/utilities/utility_db.h"
|
||||
#include "util/testharness.h"
|
||||
#include "util/logging.h"
|
||||
#include <map>
|
||||
#include <unistd.h>
|
||||
|
||||
namespace leveldb {
|
||||
|
||||
namespace {
|
||||
typedef std::map<std::string, std::string> KVMap;
|
||||
}
|
||||
|
||||
class TtlTest {
|
||||
public:
|
||||
TtlTest() {
|
||||
dbname_ = test::TmpDir() + "/db_ttl";
|
||||
options_.create_if_missing = true;
|
||||
// ensure that compaction is kicked in to always strip timestamp from kvs
|
||||
options_.max_grandparent_overlap_factor = 0;
|
||||
// compaction should take place always from level0 for determinism
|
||||
options_.max_mem_compaction_level = 0;
|
||||
db_ttl_ = nullptr;
|
||||
DestroyDB(dbname_, Options());
|
||||
}
|
||||
|
||||
~TtlTest() {
|
||||
CloseTtl();
|
||||
DestroyDB(dbname_, Options());
|
||||
}
|
||||
|
||||
// Open database with TTL support when TTL not provided with db_ttl_ pointer
|
||||
void OpenTtl() {
|
||||
assert(db_ttl_ == nullptr); // db should be closed before opening again
|
||||
ASSERT_OK(UtilityDB::OpenTtlDB(options_, dbname_, &db_ttl_));
|
||||
}
|
||||
|
||||
// Open database with TTL support when TTL provided with db_ttl_ pointer
|
||||
void OpenTtl(int32_t ttl) {
|
||||
assert(db_ttl_ == nullptr);
|
||||
ASSERT_OK(UtilityDB::OpenTtlDB(options_, dbname_, &db_ttl_, ttl));
|
||||
}
|
||||
|
||||
void CloseTtl() {
|
||||
delete db_ttl_;
|
||||
db_ttl_ = nullptr;
|
||||
}
|
||||
|
||||
// Populates and returns a kv-map
|
||||
void MakeKVMap(int64_t num_entries) {
|
||||
kvmap_.clear();
|
||||
|
||||
for (int64_t i = 0; i < num_entries; i++) {
|
||||
std::string key = "key";
|
||||
std::string value = "value";
|
||||
AppendNumberTo(&key, i);
|
||||
AppendNumberTo(&value, i);
|
||||
kvmap_[key] = value;
|
||||
}
|
||||
ASSERT_EQ((int)kvmap_.size(), num_entries);//check all insertions done
|
||||
}
|
||||
|
||||
// Puts num_entries starting from start_pos_map from kvmap_ into the database
|
||||
void PutValues(int start_pos_map, int num_entries) {
|
||||
assert(db_ttl_);
|
||||
ASSERT_LE(start_pos_map + num_entries, (int)kvmap_.size());
|
||||
static WriteOptions wopts;
|
||||
static FlushOptions flush_opts;
|
||||
kv_it_ = kvmap_.begin();
|
||||
advance(kv_it_, start_pos_map);
|
||||
for (int i = 0; kv_it_ != kvmap_.end(), i < num_entries; i++, kv_it_++) {
|
||||
ASSERT_OK(db_ttl_->Put(wopts, kv_it_->first, kv_it_->second));
|
||||
}
|
||||
// Put a mock kv at the end because CompactionFilter doesn't delete last key
|
||||
ASSERT_OK(db_ttl_->Put(wopts, "keymock", "valuemock"));
|
||||
db_ttl_->Flush(flush_opts);
|
||||
}
|
||||
|
||||
// Runs a manual compaction
|
||||
void ManualCompact() {
|
||||
db_ttl_->CompactRange(nullptr, nullptr);
|
||||
}
|
||||
|
||||
// Sleeps for slp_tim then runs a manual compaction
|
||||
// Checks span starting from st_pos from kvmap_ in the db and
|
||||
// Gets should return true if check is true and false otherwise
|
||||
// Also checks that value that we got is the same as inserted
|
||||
void SleepCompactCheck(int slp_tim, int st_pos, int span, bool check = true) {
|
||||
assert(db_ttl_);
|
||||
sleep(slp_tim);
|
||||
ManualCompact();
|
||||
static ReadOptions ropts;
|
||||
kv_it_ = kvmap_.begin();
|
||||
advance(kv_it_, st_pos);
|
||||
std::string v;
|
||||
for (int i = 0; kv_it_ != kvmap_.end(), i < span; i++, kv_it_++) {
|
||||
Status s = db_ttl_->Get(ropts, kv_it_->first, &v);
|
||||
if (s.ok() != check) {
|
||||
fprintf(stderr, "key=%s ",
|
||||
kv_it_->first.c_str());
|
||||
if (!s.ok()) {
|
||||
fprintf(stderr, "is absent from db but was expected to be present\n");
|
||||
} else {
|
||||
fprintf(stderr, "is present in db but was expected to be absent\n");
|
||||
}
|
||||
assert(false);
|
||||
} else if (s.ok() && (v.compare(kv_it_->second) != 0)) {
|
||||
fprintf(stderr, " value for key=%s present in database is %s but "
|
||||
" should be %s\n", kv_it_->first.c_str(), v.c_str(),
|
||||
kv_it_->second.c_str());
|
||||
assert(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Similar as SleepCompactCheck but uses TtlIterator to read from db
|
||||
void SleepCompactCheckIter(int slp, int st_pos, int span, bool check=true) {
|
||||
assert(db_ttl_);
|
||||
sleep(slp);
|
||||
ManualCompact();
|
||||
static ReadOptions ropts;
|
||||
Iterator *dbiter = db_ttl_->NewIterator(ropts);
|
||||
kv_it_ = kvmap_.begin();
|
||||
advance(kv_it_, st_pos);
|
||||
|
||||
dbiter->Seek(kv_it_->first);
|
||||
if (!check) {
|
||||
if (dbiter->Valid()) {
|
||||
ASSERT_NE(dbiter->value().compare(kv_it_->second), 0);
|
||||
}
|
||||
} else { // dbiter should have found out kvmap_[st_pos]
|
||||
for (int i = st_pos;
|
||||
kv_it_ != kvmap_.end() && i < st_pos + span;
|
||||
i++, kv_it_++) {
|
||||
ASSERT_TRUE(dbiter->Valid());
|
||||
ASSERT_EQ(dbiter->value().compare(kv_it_->second), 0);
|
||||
dbiter->Next();
|
||||
}
|
||||
}
|
||||
delete dbiter;
|
||||
}
|
||||
|
||||
// Choose carefully so that Put, Gets & Compaction complete in 1 second buffer
|
||||
const int64_t kSampleSize = 100;
|
||||
|
||||
private:
|
||||
std::string dbname_;
|
||||
DB* db_ttl_;
|
||||
Options options_;
|
||||
KVMap kvmap_;
|
||||
KVMap::iterator kv_it_;
|
||||
}; // class TtlTest
|
||||
|
||||
// If TTL is non positive or not provided, the behaviour is TTL = infinity
|
||||
// This test opens the db 3 times with such default behavior and inserts a
|
||||
// bunch of kvs each time. All kvs should accummulate in the db till the end
|
||||
// Partitions the sample-size provided into 3 sets over boundary1 and boundary2
|
||||
TEST(TtlTest, NoEffect) {
|
||||
MakeKVMap(kSampleSize);
|
||||
int boundary1 = kSampleSize / 3;
|
||||
int boundary2 = 2 * boundary1;
|
||||
|
||||
OpenTtl();
|
||||
PutValues(0, boundary1); //T=0: Set1 never deleted
|
||||
SleepCompactCheck(1, 0, boundary1); //T=1: Set1 still there
|
||||
CloseTtl();
|
||||
|
||||
OpenTtl(0);
|
||||
PutValues(boundary1, boundary2 - boundary1); //T=1: Set2 never deleted
|
||||
SleepCompactCheck(1, 0, boundary2); //T=2: Sets1 & 2 still there
|
||||
CloseTtl();
|
||||
|
||||
OpenTtl(-1);
|
||||
PutValues(boundary2, kSampleSize - boundary2); //T=3: Set3 never deleted
|
||||
SleepCompactCheck(1, 0, kSampleSize, true); //T=4: Sets 1,2,3 still there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Puts a set of values and checks its presence using Get during ttl
|
||||
TEST(TtlTest, PresentDuringTTL) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(2); // T=0:Open the db with ttl = 2
|
||||
PutValues(0, kSampleSize); // T=0:Insert Set1. Delete at t=2
|
||||
SleepCompactCheck(1, 0, kSampleSize, true); // T=1:Set1 should still be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Puts a set of values and checks its absence using Get after ttl
|
||||
TEST(TtlTest, AbsentAfterTTL) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(1); // T=0:Open the db with ttl = 2
|
||||
PutValues(0, kSampleSize); // T=0:Insert Set1. Delete at t=2
|
||||
SleepCompactCheck(2, 0, kSampleSize, false); // T=2:Set1 should not be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Resets the timestamp of a set of kvs by updating them and checks that they
|
||||
// are not deleted according to the old timestamp
|
||||
TEST(TtlTest, ResetTimestamp) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(3);
|
||||
PutValues(0, kSampleSize); // T=0: Insert Set1. Delete at t=3
|
||||
sleep(2); // T=2
|
||||
PutValues(0, kSampleSize); // T=2: Insert Set1. Delete at t=5
|
||||
SleepCompactCheck(2, 0, kSampleSize); // T=4: Set1 should still be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Similar to PresentDuringTTL but uses Iterator
|
||||
TEST(TtlTest, IterPresentDuringTTL) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(2);
|
||||
PutValues(0, kSampleSize); // T=0: Insert. Delete at t=2
|
||||
SleepCompactCheckIter(1, 0, kSampleSize); // T=1: Set should be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Similar to AbsentAfterTTL but uses Iterator
|
||||
TEST(TtlTest, IterAbsentAfterTTL) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(1);
|
||||
PutValues(0, kSampleSize); // T=0: Insert. Delete at t=1
|
||||
SleepCompactCheckIter(2, 0, kSampleSize, false); // T=2: Should not be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Checks presence while opening the same db more than once with the same ttl
|
||||
// Note: The second open will open the same db
|
||||
TEST(TtlTest, MultiOpenSamePresent) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(2);
|
||||
PutValues(0, kSampleSize); // T=0: Insert. Delete at t=2
|
||||
CloseTtl();
|
||||
|
||||
OpenTtl(2); // T=0. Delete at t=2
|
||||
SleepCompactCheck(1, 0, kSampleSize); // T=1: Set should be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Checks absence while opening the same db more than once with the same ttl
|
||||
// Note: The second open will open the same db
|
||||
TEST(TtlTest, MultiOpenSameAbsent) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(1);
|
||||
PutValues(0, kSampleSize); // T=0: Insert. Delete at t=1
|
||||
CloseTtl();
|
||||
|
||||
OpenTtl(1); // T=0.Delete at t=1
|
||||
SleepCompactCheck(2, 0, kSampleSize, false); // T=2: Set should not be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
// Checks presence while opening the same db more than once with bigger ttl
|
||||
TEST(TtlTest, MultiOpenDifferent) {
|
||||
MakeKVMap(kSampleSize);
|
||||
|
||||
OpenTtl(1);
|
||||
PutValues(0, kSampleSize); // T=0: Insert. Delete at t=1
|
||||
CloseTtl();
|
||||
|
||||
OpenTtl(3); // T=0: Set deleted at t=3
|
||||
SleepCompactCheck(2, 0, kSampleSize); // T=2: Set should be there
|
||||
CloseTtl();
|
||||
}
|
||||
|
||||
} // namespace leveldb
|
||||
|
||||
// A black-box test for the ttl wrapper around rocksdb
|
||||
int main(int argc, char** argv) {
|
||||
return leveldb::test::RunAllTests();
|
||||
}
|
||||
Reference in New Issue
Block a user