andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add a tool to change number of levels

Browse Source 
Summary: as subject.

Test Plan: manually test it, will add a testcase

Reviewers: dhruba, MarkCallaghan

Differential Revision: https://reviews.facebook.net/D6345
This commit is contained in:
heyongqiang 2012-10-31 11:47:18 -07:00
parent a1bd5b7752
commit d55c2ba305
8 changed files with 910 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile
View File

@ -51,6 +51,7 @@ TESTS = \
table_test \
version_edit_test \
version_set_test \
reduce_levels_test \
write_batch_test \
filelock_test
@ -173,6 +174,9 @@ version_edit_test: db/version_edit_test.o $(LIBOBJECTS) $(TESTHARNESS)
version_set_test: db/version_set_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(CXX) db/version_set_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LDFLAGS)
reduce_levels_test: tools/reduce_levels_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(CXX) tools/reduce_levels_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LDFLAGS)
write_batch_test: db/write_batch_test.o $(LIBOBJECTS) $(TESTHARNESS)
$(CXX) db/write_batch_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LDFLAGS)

41
db/version_set.cc
View File

@ -714,21 +714,10 @@ VersionSet::VersionSet(const std::string& dbname,
descriptor_file_(NULL),
descriptor_log_(NULL),
dummy_versions_(this),
current_(NULL) {
current_(NULL),
num_levels_(options_->num_levels) {
compact_pointer_ = new std::string[options_->num_levels];
max_file_size_ = new uint64_t[options_->num_levels];
level_max_bytes_ = new uint64_t[options->num_levels];
int target_file_size_multiplier = options_->target_file_size_multiplier;
int max_bytes_multiplier = options_->max_bytes_for_level_multiplier;
for (int i = 0; i < options_->num_levels; i++) {
if (i > 1) {
max_file_size_[i] = max_file_size_[i-1] * target_file_size_multiplier;
level_max_bytes_[i] = level_max_bytes_[i-1] * max_bytes_multiplier;
} else {
max_file_size_[i] = options_->target_file_size_base;
level_max_bytes_[i] = options_->max_bytes_for_level_base;
}
}
Init(options_->num_levels);
AppendVersion(new Version(this));
}
@ -742,6 +731,22 @@ VersionSet::~VersionSet() {
delete descriptor_file_;
}
void VersionSet::Init(int num_levels) {
max_file_size_ = new uint64_t[num_levels];
level_max_bytes_ = new uint64_t[num_levels];
int target_file_size_multiplier = options_->target_file_size_multiplier;
int max_bytes_multiplier = options_->max_bytes_for_level_multiplier;
for (int i = 0; i < num_levels; i++) {
if (i > 1) {
max_file_size_[i] = max_file_size_[i-1] * target_file_size_multiplier;
level_max_bytes_[i] = level_max_bytes_[i-1] * max_bytes_multiplier;
} else {
max_file_size_[i] = options_->target_file_size_base;
level_max_bytes_[i] = options_->max_bytes_for_level_base;
}
}
}
void VersionSet::AppendVersion(Version* v) {
// Make "v" current
assert(v->refs_ == 0);
@ -759,7 +764,8 @@ void VersionSet::AppendVersion(Version* v) {
v->next_->prev_ = v;
}
Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu,
bool new_descriptor_log) {
if (edit->has_log_number_) {
assert(edit->log_number_ >= log_number_);
assert(edit->log_number_ < next_file_number_);
@ -787,10 +793,10 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
std::string new_manifest_file;
uint64_t new_manifest_file_size = 0;
Status s;
if (descriptor_log_ == NULL) {
if (descriptor_log_ == NULL || new_descriptor_log) {
// No reason to unlock *mu here since we only hit this path in the
// first call to LogAndApply (when opening the database).
assert(descriptor_file_ == NULL);
assert(descriptor_file_ == NULL || new_descriptor_log)
new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
edit->SetNextFile(next_file_number_);
s = env_->NewWritableFile(new_manifest_file, &descriptor_file_);
@ -1090,7 +1096,6 @@ Status VersionSet::DumpManifest(Options& options, std::string& dscname) {
printf("%s \n", v->DebugString().c_str());
}
return s;
}

18
db/version_set.h
View File

@ -21,6 +21,7 @@
#include "db/dbformat.h"
#include "db/version_edit.h"
#include "port/port.h"
#include "db/table_cache.h"
namespace leveldb {
@ -156,11 +157,20 @@ class VersionSet {
// current version. Will release *mu while actually writing to the file.
// REQUIRES: *mu is held on entry.
// REQUIRES: no other thread concurrently calls LogAndApply()
Status LogAndApply(VersionEdit* edit, port::Mutex* mu);
Status LogAndApply(VersionEdit* edit, port::Mutex* mu,
bool new_descriptor_log = false);
// Recover the last saved descriptor from persistent storage.
Status Recover();
// Try to reduce the number of levels. This call is valid when
// only one level from the new max level to the old
// max level containing files.
// For example, a db currently has 7 levels [0-6], and a call to
// to reduce to 5 [0-4] can only be executed when only one level
// among [4-6] contains files.
Status ReduceNumberOfLevels(int new_levels, port::Mutex* mu);
// Return the current version.
Version* current() const { return current_; }
@ -204,7 +214,7 @@ class VersionSet {
// being compacted, or zero if there is no such log file.
uint64_t PrevLogNumber() const { return prev_log_number_; }
int NumberLevels() const { return options_->num_levels; }
int NumberLevels() const { return num_levels_; }
// Pick level and inputs for a new compaction.
// Returns NULL if there is no compaction to be done.
@ -274,6 +284,8 @@ class VersionSet {
friend class Compaction;
friend class Version;
void Init(int num_levels);
void Finalize(Version* v);
void GetRange(const std::vector<FileMetaData*>& inputs,
@ -311,6 +323,8 @@ class VersionSet {
uint64_t log_number_;
uint64_t prev_log_number_; // 0 or backing store for memtable being compacted
int num_levels_;
// Opened lazily
WritableFile* descriptor_file_;
log::Writer* descriptor_log_;

70
db/version_set_reduce_num_levels.cc Normal file
View File

@ -0,0 +1,70 @@
// Copyright (c) 2012 Facebook. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "db/version_set.h"
#include <algorithm>
#include <stdio.h>
#include "db/log_reader.h"
#include "db/log_writer.h"
#include "util/logging.h"
namespace leveldb {
Status VersionSet::ReduceNumberOfLevels(int new_levels, port::Mutex* mu) {
if(new_levels <= 1) {
return Status::InvalidArgument(
"Number of levels needs to be bigger than 1");
}
Version* current_version = current_;
int current_levels = NumberLevels();
// Make sure there are file only on one level from
// (new_levels-1) to (current_levels-1)
int first_nonempty_level = -1;
int first_nonempty_level_filenum = 0;
for (int i = new_levels - 1; i < current_levels; i++) {
int file_num = NumLevelFiles(i);
if (file_num != 0) {
if (first_nonempty_level < 0) {
first_nonempty_level = i;
first_nonempty_level_filenum = file_num;
} else {
char msg[255];
sprintf(msg, "Found at least two levels containing files: "
"[%d:%d],[%d:%d].\n",
first_nonempty_level, first_nonempty_level_filenum, i, file_num);
return Status::InvalidArgument(msg);
}
}
}
Status st;
std::vector<FileMetaData*>* old_files_list = current_version->files_;
std::vector<FileMetaData*>* new_files_list =
new std::vector<FileMetaData*>[new_levels];
for (int i = 0; i < new_levels - 1; i++) {
new_files_list[i] = old_files_list[i];
}
if (first_nonempty_level > 0) {
new_files_list[new_levels - 1] = old_files_list[first_nonempty_level];
}
delete[] current_version->files_;
current_version->files_ = new_files_list;
delete[] compact_pointer_;
delete[] max_file_size_;
delete[] level_max_bytes_;
num_levels_ = new_levels;
compact_pointer_ = new std::string[new_levels];
Init(new_levels);
st = LogAndApply(new VersionEdit(new_levels), mu, true);
return st;
}
}

268
tools/ldb.cc
View File

@ -2,234 +2,74 @@
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include <string>
#include <iostream>
#include <sstream>
#include <stdlib.h>
#include "util/ldb_cmd.h"
#include "leveldb/db.h"
#include "leveldb/options.h"
#include "leveldb/iterator.h"
#include "leveldb/slice.h"
namespace leveldb {
std::string HexToString(const std::string& str) {
std::string parsed;
for (int i = 0; i < str.length(); ) {
int c;
sscanf(str.c_str() + i, "%2X", &c);
parsed.push_back(c);
i += 2;
}
return parsed;
class LDBCommandRunner {
public:
static void PrintHelp(const char* exec_name) {
std::string ret;
ret.append("--- compact ----:\n");
ret.append(exec_name);
ret.append(" compact ");
Compactor::Help(ret);
ret.append("\n--- dump ----:\n");
ret.append(exec_name);
ret.append(" dump ");
DBDumper::Help(ret);
ret.append("\n---reduce_levels ----:\n");
ret.append(exec_name);
ret.append(" reduce_levels ");
ReduceDBLevels::Help(ret);
fprintf(stderr, "%s\n", ret.c_str());
}
static void print_usage() {
fprintf(stderr,
"ldb [compact|dump] "
"--db-path=database_path "
"[--from=START KEY] "
"[--to=END KEY ] "
"[--max_keys=[NUM] (only for dump)] "
"[--hex ] "
"[--count_only (only for dump)] "
"[--stats (only for dump) ] \n");
}
static void safe_open_db(const std::string& dbname, leveldb::DB** db) {
leveldb::Options options;
options.create_if_missing = false;
leveldb::Status status = leveldb::DB::Open(options, dbname, db);
if(!status.ok()) {
fprintf(
stderr,
"Could not open db at %s\nERROR: %s",
dbname.data(),
status.ToString().data()
);
static void RunCommand(int argc, char** argv) {
if (argc <= 2) {
PrintHelp(argv[0]);
exit(1);
}
}
static void dump_db(
const std::string& db_path,
std::string& start,
std::string& end,
int64_t max_keys,
const bool hex,
const bool print_stats,
const bool count_only
) {
// Parse command line args
uint64_t count = 0;
if (hex) {
start = HexToString(start);
end = HexToString(end);
}
leveldb::DB *db;
safe_open_db(db_path, &db);
if (print_stats) {
std::string stats;
if (db->GetProperty("leveldb.stats", &stats)) {
fprintf(stdout, "%s\n", stats.c_str());
}
}
// Setup key iterator
leveldb::Iterator* iter = db->NewIterator(leveldb::ReadOptions());
leveldb::Status status = iter->status();
if (!status.ok()) {
fprintf(stderr, "%s\n", status.ToString().c_str());
delete db;
exit(1);
}
for (iter->Seek(start); iter->Valid(); iter->Next()) {
// If end marker was specified, we stop before it
if (!end.empty() && (iter->key().ToString() >= end))
break;
// Terminate if maximum number of keys have been dumped
if (max_keys == 0)
break;
--max_keys;
++count;
if (!count_only) {
if (hex) {
std::string str = iter->key().ToString();
for (int i = 0; i < str.length(); ++i) {
fprintf(stdout, "%X", str[i]);
}
fprintf(stdout, " ==> ");
str = iter->value().ToString();
for (int i = 0; i < str.length(); ++i) {
fprintf(stdout, "%X", str[i]);
}
fprintf(stdout, "\n");
const char* cmd = argv[1];
std::string db_name;
std::vector<std::string> args;
for (int i = 2; i < argc; i++) {
if (strncmp(argv[i], "--db=", strlen("--db=")) == 0) {
db_name = argv[i] + strlen("--db=");
} else {
fprintf(stdout, "%s ==> %s\n",
iter->key().ToString().c_str(),
iter->value().ToString().c_str());
args.push_back(argv[i]);
}
}
LDBCommand* cmdObj = NULL;
if (strncmp(cmd, "compact", strlen("compact")) == 0) {
// run compactor
cmdObj = new Compactor(db_name, args);
} else if (strncmp(cmd, "dump", strlen("dump")) == 0) {
// run dump
cmdObj = new DBDumper(db_name, args);
} else if (strncmp(cmd, "reduce_levels", strlen("reduce_levels")) == 0) {
// reduce db levels
cmdObj = new ReduceDBLevels(db_name, args);
} else {
fprintf(stderr, "Unknown command: %s\n", cmd);
PrintHelp(argv[0]);
exit(1);
}
fprintf(stdout, "Keys in range: %lld\n", (long long) count);
// Clean up
delete iter;
delete db;
cmdObj->Run();
LDBCommandExecuteResult ret = cmdObj->GetExecuteState();
fprintf(stderr, "%s\n", ret.ToString().c_str());
delete cmdObj;
}
};
static void compact(
const std::string dbname,
std::string from,
std::string to,
const bool hex
) {
leveldb::DB* db;
safe_open_db(dbname, &db);
if(hex) {
from = HexToString(from);
to = HexToString(to);
}
leveldb::Slice* begin = from.empty() ? NULL : new leveldb::Slice(from);
leveldb::Slice* end = to.empty() ? NULL : new leveldb::Slice(to);
db->CompactRange(begin, end);
delete db;
}
int main(int argc, char** argv) {
enum {
DUMP, COMPACT
} command;
if (argc < 2) {
print_usage();
exit(1);
}
size_t n;
const std::string dbnameKey = "--db-path=";
const std::string toKey = "--to=";
const std::string fromKey = "--from=";
std::string dbname;
bool dbnameFound = false;
std::string from;
std::string to;
int64_t temp;
int64_t max_keys = -1;
bool print_stats = false;
bool count_only = false;
bool hex = false;
char junk;
std::string commandString = argv[1];
if (commandString == "dump") {
command = DUMP;
} else if (commandString == "compact") {
command = COMPACT;
} else {
print_usage();
exit(1);
}
for (int i = 2; i < argc; i++) {
std::string param(argv[i]);
if ((n = param.find(dbnameKey)) != std::string::npos) {
dbname = param.substr(dbnameKey.size());
dbnameFound = true;
} else if ((n = param.find(fromKey)) != std::string::npos) {
from = param.substr(fromKey.size());
} else if ((n = param.find(toKey)) != std::string::npos) {
to = param.substr(toKey.size());
} else if (sscanf(argv[i], "--max_keys=%ld%c", &temp, &junk) == 1) {
max_keys = temp;
} else if (strncmp(argv[i], "--stats", 7) == 0) {
print_stats = true;
} else if (strncmp(argv[i], "--count_only", 12) == 0) {
count_only = true;
} else if (strncmp(argv[i], "--hex", 5) == 0) {
hex = true;
} else {
print_usage();
exit(1);
}
}
if (!dbnameFound || dbname.empty()) {
fprintf(stderr, "DB path required. See help\n");
print_usage();
exit(1);
}
switch(command) {
case DUMP:
dump_db(dbname, from, to, max_keys, hex, print_stats, count_only);
break;
case COMPACT:
compact(dbname, from, to, hex);
break;
default:
print_usage();
exit(1);
}
return 0;
leveldb::LDBCommandRunner::RunCommand(argc, argv);
}

200
tools/reduce_levels_test.cc Normal file
View File

@ -0,0 +1,200 @@
// Copyright (c) 2012 Facebook. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "leveldb/db.h"
#include "db/db_impl.h"
#include "db/version_set.h"
#include "util/logging.h"
#include "util/testutil.h"
#include "util/testharness.h"
#include "util/ldb_cmd.h"
namespace leveldb {
class ReduceLevelTest {
public:
ReduceLevelTest() {
dbname_ = test::TmpDir() + "/db_reduce_levels_test";
DestroyDB(dbname_, Options());
db_ = NULL;
}
Status OpenDB(bool create_if_missing, int levels,
int mem_table_compact_level);
Status Put(const std::string& k, const std::string& v) {
return db_->Put(WriteOptions(), k, v);
}
std::string Get(const std::string& k) {
ReadOptions options;
std::string result;
Status s = db_->Get(options, k, &result);
if (s.IsNotFound()) {
result = "NOT_FOUND";
} else if (!s.ok()) {
result = s.ToString();
}
return result;
}
Status CompactMemTable() {
if (db_ == NULL) {
return Status::InvalidArgument("DB not opened.");
}
DBImpl* db_impl = reinterpret_cast<DBImpl*>(db_);
return db_impl->TEST_CompactMemTable();
}
void CloseDB() {
if (db_ != NULL) {
delete db_;
db_ = NULL;
}
}
bool ReduceLevels(int target_level);
int FilesOnLevel(int level) {
std::string property;
ASSERT_TRUE(
db_->GetProperty("leveldb.num-files-at-level" + NumberToString(level),
&property));
return atoi(property.c_str());
}
private:
std::string dbname_;
DB* db_;
};
Status ReduceLevelTest::OpenDB(bool create_if_missing, int num_levels,
int mem_table_compact_level) {
leveldb::Options opt;
opt.num_levels = num_levels;
opt.create_if_missing = create_if_missing;
opt.max_mem_compaction_level = mem_table_compact_level;
leveldb::Status st = leveldb::DB::Open(opt, dbname_, &db_);
if (!st.ok()) {
fprintf(stderr, "Can't open the db:%s\n", st.ToString().c_str());
}
return st;
}
bool ReduceLevelTest::ReduceLevels(int target_level) {
std::vector<std::string> args = leveldb::ReduceDBLevels::PrepareArgs(
target_level, false);
ReduceDBLevels level_reducer(dbname_, args);
level_reducer.Run();
return level_reducer.GetExecuteState().IsSucceed();
}
TEST(ReduceLevelTest, Last_Level) {
// create files on all levels;
ASSERT_OK(OpenDB(true, 4, 3));
ASSERT_OK(Put("aaaa", "11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(3), 1);
CloseDB();
ASSERT_TRUE(ReduceLevels(3));
ASSERT_OK(OpenDB(true, 3, 1));
ASSERT_EQ(FilesOnLevel(2), 1);
CloseDB();
ASSERT_TRUE(ReduceLevels(2));
ASSERT_OK(OpenDB(true, 2, 1));
ASSERT_EQ(FilesOnLevel(1), 1);
CloseDB();
}
TEST(ReduceLevelTest, Top_Level) {
// create files on all levels;
ASSERT_OK(OpenDB(true, 5, 0));
ASSERT_OK(Put("aaaa", "11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(0), 1);
CloseDB();
// The CompactRange(NULL, NULL) call in ReduceLevels
// will push this file to level-1
ASSERT_TRUE(ReduceLevels(4));
ASSERT_OK(OpenDB(true, 4, 0));
ASSERT_EQ(FilesOnLevel(1), 1);
CloseDB();
ASSERT_TRUE(ReduceLevels(3));
ASSERT_OK(OpenDB(true, 3, 0));
ASSERT_EQ(FilesOnLevel(1), 1);
CloseDB();
ASSERT_TRUE(ReduceLevels(2));
ASSERT_OK(OpenDB(true, 2, 0));
ASSERT_EQ(FilesOnLevel(1), 1);
CloseDB();
}
TEST(ReduceLevelTest, All_Levels) {
// create files on all levels;
ASSERT_OK(OpenDB(true, 5, 1));
ASSERT_OK(Put("a", "a11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(1), 1);
CloseDB();
ASSERT_OK(OpenDB(true, 5, 2));
ASSERT_OK(Put("b", "b11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(1), 1);
ASSERT_EQ(FilesOnLevel(2), 1);
CloseDB();
ASSERT_OK(OpenDB(true, 5, 3));
ASSERT_OK(Put("c", "c11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(1), 1);
ASSERT_EQ(FilesOnLevel(2), 1);
ASSERT_EQ(FilesOnLevel(3), 1);
CloseDB();
ASSERT_OK(OpenDB(true, 5, 4));
ASSERT_OK(Put("d", "d11111"));
ASSERT_OK(CompactMemTable());
ASSERT_EQ(FilesOnLevel(1), 1);
ASSERT_EQ(FilesOnLevel(2), 1);
ASSERT_EQ(FilesOnLevel(3), 1);
ASSERT_EQ(FilesOnLevel(4), 1);
CloseDB();
ASSERT_TRUE(ReduceLevels(4));
ASSERT_OK(OpenDB(true, 4, 0));
ASSERT_EQ("a11111", Get("a"));
ASSERT_EQ("b11111", Get("b"));
ASSERT_EQ("c11111", Get("c"));
ASSERT_EQ("d11111", Get("d"));
CloseDB();
ASSERT_TRUE(ReduceLevels(3));
ASSERT_OK(OpenDB(true, 3, 0));
ASSERT_EQ("a11111", Get("a"));
ASSERT_EQ("b11111", Get("b"));
ASSERT_EQ("c11111", Get("c"));
ASSERT_EQ("d11111", Get("d"));
CloseDB();
ASSERT_TRUE(ReduceLevels(2));
ASSERT_OK(OpenDB(true, 2, 0));
ASSERT_EQ("a11111", Get("a"));
ASSERT_EQ("b11111", Get("b"));
ASSERT_EQ("c11111", Get("c"));
ASSERT_EQ("d11111", Get("d"));
CloseDB();
}
}
int main(int argc, char** argv) {
return leveldb::test::RunAllTests();
}

294
util/ldb_cmd.cc Normal file
View File

@ -0,0 +1,294 @@
// Copyright (c) 2012 Facebook. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "util/ldb_cmd.h"
namespace leveldb {
const char* LDBCommand::FROM_ARG = "--from=";
const char* LDBCommand::END_ARG = "--to=";
const char* LDBCommand::HEX_ARG = "--hex";
Compactor::Compactor(std::string& db_name, std::vector<std::string>& args) :
LDBCommand(db_name, args), null_from_(true), null_to_(true), hex_(false) {
for (int i = 0; i < args.size(); i++) {
std::string& arg = args.at(i);
if (arg.find(FROM_ARG) == 0) {
null_from_ = false;
from_ = arg.substr(strlen(FROM_ARG));
} else if (arg.find(END_ARG) == 0) {
null_to_ = false;
to_ = arg.substr(strlen(END_ARG));
} else if (arg.find(HEX_ARG) == 0) {
hex_ = true;
} else {
exec_state_ = LDBCommandExecuteResult::FAILED("Unknown argument." + arg);
}
}
if (hex_) {
if (!null_from_) {
from_ = HexToString(from_);
}
if (!null_to_) {
to_ = HexToString(to_);
}
}
}
void Compactor::Help(std::string& ret) {
LDBCommand::Help(ret);
ret.append("[--from=START KEY] ");
ret.append("[--to=START KEY] ");
ret.append("[--hex] ");
}
void Compactor::DoCommand() {
leveldb::Slice* begin = NULL;
leveldb::Slice* end = NULL;
if (!null_from_) {
begin = new leveldb::Slice(from_);
}
if (!null_to_) {
end = new leveldb::Slice(to_);
}
db_->CompactRange(begin, end);
exec_state_ = LDBCommandExecuteResult::SUCCEED("");
delete begin;
delete end;
}
const char* DBDumper::MAX_KEYS_ARG = "--max_keys=";
const char* DBDumper::COUNT_ONLY_ARG = "--count_only";
const char* DBDumper::STATS_ARG = "--stats";
const char* DBDumper::HEX_OUTPUT_ARG = "--output_hex";
DBDumper::DBDumper(std::string& db_name, std::vector<std::string>& args) :
LDBCommand(db_name, args), null_from_(true), null_to_(true), hex_(false),
count_only_(false), print_stats_(false), max_keys_(-1),
hex_output_(false) {
for (int i = 0; i < args.size(); i++) {
std::string& arg = args.at(i);
if (arg.find(FROM_ARG) == 0) {
null_from_ = false;
from_ = arg.substr(strlen(FROM_ARG));
} else if (arg.find(END_ARG) == 0) {
null_to_ = false;
to_ = arg.substr(strlen(END_ARG));
} else if (arg.find(HEX_ARG) == 0) {
hex_ = true;
} else if (arg.find(MAX_KEYS_ARG) == 0) {
max_keys_ = atoi(arg.substr(strlen(MAX_KEYS_ARG)).c_str());
} else if (arg.find(STATS_ARG) == 0) {
print_stats_ = true;
} else if (arg.find(COUNT_ONLY_ARG) == 0) {
count_only_ = true;
} else if (arg.find(HEX_OUTPUT_ARG) == 0) {
hex_output_ = true;
} else {
exec_state_ = LDBCommandExecuteResult::FAILED("Unknown argument:" + arg);
}
}
if (hex_) {
if (!null_from_) {
from_ = HexToString(from_);
}
if (!null_to_) {
to_ = HexToString(to_);
}
}
}
void DBDumper::Help(std::string& ret) {
LDBCommand::Help(ret);
ret.append("[--from=START KEY] ");
ret.append("[--to=END Key] ");
ret.append("[--hex] ");
ret.append("[--output_hex] ");
ret.append("[--max_keys=NUM] ");
ret.append("[--count_only] ");
ret.append("[--stats] ");
}
void DBDumper::DoCommand() {
// Parse command line args
uint64_t count = 0;
if (print_stats_) {
std::string stats;
if (db_->GetProperty("leveldb.stats", &stats)) {
fprintf(stdout, "%s\n", stats.c_str());
}
}
// Setup key iterator
leveldb::Iterator* iter = db_->NewIterator(leveldb::ReadOptions());
leveldb::Status st = iter->status();
if (!st.ok()) {
exec_state_ = LDBCommandExecuteResult::FAILED("Iterator error."
+ st.ToString());
}
if (!null_from_) {
iter->Seek(from_);
} else {
iter->SeekToFirst();
}
int max_keys = max_keys_;
for (; iter->Valid(); iter->Next()) {
// If end marker was specified, we stop before it
if (!null_to_ && (iter->key().ToString() >= to_))
break;
// Terminate if maximum number of keys have been dumped
if (max_keys == 0)
break;
if (max_keys > 0) {
--max_keys;
}
++count;
if (!count_only_) {
if (hex_output_) {
std::string str = iter->key().ToString();
for (int i = 0; i < str.length(); ++i) {
fprintf(stdout, "%X", str[i]);
}
fprintf(stdout, " ==> ");
str = iter->value().ToString();
for (int i = 0; i < str.length(); ++i) {
fprintf(stdout, "%X", str[i]);
}
fprintf(stdout, "\n");
} else {
fprintf(stdout, "%s ==> %s\n", iter->key().ToString().c_str(),
iter->value().ToString().c_str());
}
}
}
fprintf(stdout, "Keys in range: %lld\n", (long long) count);
// Clean up
delete iter;
}
const char* ReduceDBLevels::NEW_LEVLES_ARG = "--new_levels=";
const char* ReduceDBLevels::PRINT_OLD_LEVELS_ARG = "--print_old_levels";
ReduceDBLevels::ReduceDBLevels(std::string& db_name,
std::vector<std::string>& args)
: LDBCommand(db_name, args),
new_levels_(-1),
print_old_levels_(false) {
for (int i = 0; i < args.size(); i++) {
std::string& arg = args.at(i);
if (arg.find(NEW_LEVLES_ARG) == 0) {
new_levels_ = atoi(arg.substr(strlen(NEW_LEVLES_ARG)).c_str());
} else if (arg.find(PRINT_OLD_LEVELS_ARG) == 0) {
print_old_levels_ = true;
} else {
exec_state_ = LDBCommandExecuteResult::FAILED(
"Unknown argument." + arg);
}
}
if(new_levels_ <= 0) {
exec_state_ = LDBCommandExecuteResult::FAILED(
" Use --new_levels to specify a new level number\n");
}
}
std::vector<std::string> ReduceDBLevels::PrepareArgs(int new_levels,
bool print_old_level) {
std::vector<std::string> ret;
char arg[100];
sprintf(arg, "%s%d", NEW_LEVLES_ARG, new_levels);
ret.push_back(arg);
if(print_old_level) {
sprintf(arg, "%s", PRINT_OLD_LEVELS_ARG);
ret.push_back(arg);
}
return ret;
}
void ReduceDBLevels::Help(std::string& msg) {
LDBCommand::Help(msg);
msg.append("[--new_levels=New number of levels] ");
msg.append("[--print_old_levels] ");
}
leveldb::Options ReduceDBLevels::PrepareOptionsForOpenDB() {
leveldb::Options opt = LDBCommand::PrepareOptionsForOpenDB();
// Set to a big value to make sure we can open the db
opt.num_levels = 1 << 16;
return opt;
}
void ReduceDBLevels::DoCommand() {
if (new_levels_ <= 1) {
exec_state_ = LDBCommandExecuteResult::FAILED(
"Invalid number of levels.\n");
return;
}
leveldb::Status st;
leveldb::Options opt = PrepareOptionsForOpenDB();
if (print_old_levels_) {
TableCache* tc = new TableCache(db_path_, &opt, 10);
const InternalKeyComparator* cmp = new InternalKeyComparator(
opt.comparator);
VersionSet* versions = new VersionSet(db_path_, &opt,
tc, cmp);
// We rely the VersionSet::Recover to tell us the internal data structures
// in the db. And the Recover() should never do any change
// (like LogAndApply) to the manifest file.
st = versions->Recover();
int max = -1;
for(int i = 0; i<versions->NumberLevels(); i++) {
if (versions->NumLevelFiles(i)) {
max = i;
}
}
fprintf(stdout, "The old number of levels in use is %d\n", max + 1);
delete versions;
if (!st.ok()) {
exec_state_ = LDBCommandExecuteResult::FAILED(st.ToString());
return;
}
}
// Compact the whole DB to put all files to the highest level.
db_->CompactRange(NULL, NULL);
CloseDB();
TableCache* tc = new TableCache(db_path_, &opt, 10);
const InternalKeyComparator* cmp = new InternalKeyComparator(
opt.comparator);
VersionSet* versions = new VersionSet(db_path_, &opt,
tc, cmp);
// We rely the VersionSet::Recover to tell us the internal data structures
// in the db. And the Recover() should never do any change (like LogAndApply)
// to the manifest file.
st = versions->Recover();
if (!st.ok()) {
exec_state_ = LDBCommandExecuteResult::FAILED(st.ToString());
return;
}
port::Mutex mu;
mu.Lock();
st = versions->ReduceNumberOfLevels(new_levels_, &mu);
mu.Unlock();
if (!st.ok()) {
exec_state_ = LDBCommandExecuteResult::FAILED(st.ToString());
return;
}
}
}

248
util/ldb_cmd.h Normal file
View File

@ -0,0 +1,248 @@
// Copyright (c) 2012 Facebook. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifndef LEVELDB_UTIL_LDB_H_
#define LEVELDB_UTIL_LDB_H_
#include <string>
#include <iostream>
#include <sstream>
#include <stdlib.h>
#include <algorithm>
#include <stdio.h>
#include "leveldb/db.h"
#include "leveldb/options.h"
#include "leveldb/iterator.h"
#include "leveldb/slice.h"
#include "db/version_set.h"
#include "util/logging.h"
namespace leveldb {
class LDBCommandExecuteResult {
public:
enum State {
EXEC_NOT_STARTED = 0, EXEC_SUCCEED = 1, EXEC_FAILED = 2,
};
LDBCommandExecuteResult() {
state_ = EXEC_NOT_STARTED;
message_ = "";
}
LDBCommandExecuteResult(State state, std::string& msg) {
state_ = state;
message_ = msg;
}
std::string ToString() {
std::string ret;
switch (state_) {
case EXEC_SUCCEED:
ret.append("Succeeded.");
break;
case EXEC_FAILED:
ret.append("Failed.");
break;
case EXEC_NOT_STARTED:
ret.append("Not started.");
}
if (!message_.empty()) {
ret.append(message_);
}
return ret;
}
void Reset() {
state_ = EXEC_NOT_STARTED;
message_ = "";
}
bool IsSucceed() {
return state_ == EXEC_SUCCEED;
}
bool IsNotStarted() {
return state_ == EXEC_NOT_STARTED;
}
bool IsFailed() {
return state_ == EXEC_FAILED;
}
static LDBCommandExecuteResult SUCCEED(std::string msg) {
return LDBCommandExecuteResult(EXEC_SUCCEED, msg);
}
static LDBCommandExecuteResult FAILED(std::string msg) {
return LDBCommandExecuteResult(EXEC_FAILED, msg);
}
private:
State state_;
std::string message_;
bool operator==(const LDBCommandExecuteResult&);
bool operator!=(const LDBCommandExecuteResult&);
};
class LDBCommand {
public:
/* Constructor */
LDBCommand(std::string& db_name, std::vector<std::string>& args) :
db_path_(db_name),
db_(NULL) {
}
virtual leveldb::Options PrepareOptionsForOpenDB() {
leveldb::Options opt;
opt.create_if_missing = false;
return opt;
}
virtual ~LDBCommand() {
if (db_ != NULL) {
delete db_;
db_ = NULL;
}
}
/* Print the help message */
static void Help(std::string& ret) {
ret.append("--db=DB_PATH ");
}
/* Run the command, and return the execute result. */
void Run() {
if (!exec_state_.IsNotStarted()) {
return;
}
if (db_ == NULL) {
OpenDB();
}
DoCommand();
if (exec_state_.IsNotStarted()) {
exec_state_ = LDBCommandExecuteResult::SUCCEED("");
}
CloseDB ();
}
virtual void DoCommand() = 0;
LDBCommandExecuteResult GetExecuteState() {
return exec_state_;
}
void ClearPreviousRunState() {
exec_state_.Reset();
}
static std::string HexToString(const std::string& str) {
std::string parsed;
for (int i = 0; i < str.length();) {
int c;
sscanf(str.c_str() + i, "%2X", &c);
parsed.push_back(c);
i += 2;
}
return parsed;
}
protected:
void OpenDB() {
leveldb::Options opt = PrepareOptionsForOpenDB();
// Open the DB.
leveldb::Status st = leveldb::DB::Open(opt, db_path_, &db_);
if (!st.ok()) {
std::string msg = st.ToString();
exec_state_ = LDBCommandExecuteResult::FAILED(msg);
}
}
void CloseDB () {
if (db_ != NULL) {
delete db_;
db_ = NULL;
}
}
static const char* FROM_ARG;
static const char* END_ARG;
static const char* HEX_ARG;
LDBCommandExecuteResult exec_state_;
std::string db_path_;
leveldb::DB* db_;
};
class Compactor: public LDBCommand {
public:
Compactor(std::string& db_name, std::vector<std::string>& args);
virtual ~Compactor() {}
static void Help(std::string& ret);
virtual void DoCommand();
private:
bool null_from_;
std::string from_;
bool null_to_;
std::string to_;
bool hex_;
};
class DBDumper: public LDBCommand {
public:
DBDumper(std::string& db_name, std::vector<std::string>& args);
virtual ~DBDumper() {}
static void Help(std::string& ret);
virtual void DoCommand();
private:
bool null_from_;
std::string from_;
bool null_to_;
std::string to_;
int max_keys_;
bool count_only_;
bool print_stats_;
bool hex_;
bool hex_output_;
static const char* MAX_KEYS_ARG;
static const char* COUNT_ONLY_ARG;
static const char* STATS_ARG;
static const char* HEX_OUTPUT_ARG;
};
class ReduceDBLevels : public LDBCommand {
public:
ReduceDBLevels (std::string& db_name, std::vector<std::string>& args);
~ReduceDBLevels() {}
virtual leveldb::Options PrepareOptionsForOpenDB();
virtual void DoCommand();
static void Help(std::string& msg);
static std::vector<std::string> PrepareArgs(int new_levels,
bool print_old_level = false);
private:
int new_levels_;
bool print_old_levels_;
static const char* NEW_LEVLES_ARG;
static const char* PRINT_OLD_LEVELS_ARG;
};
}
#endif