Skip to content

T
titan
Unverified Commit 07aa0655 authored by Connor's avatar Connor Committed by GitHub
Browse files
Options

Rename version_set to blob_file_set (#69) 

* rename version_set to blob_set
Signed-off-by: 's avatarConnor1996 <zbk602423539@gmail.com>
parent f963b880
Hide whitespace changes
Inline Side-by-side
Showing with 127 additions and 118 deletions
src/version_set.cc → src/blob_file_set.cc
View file @ 07aa0655
#include "version_set.h" #include "blob_file_set.h"
#include <inttypes.h> #include <inttypes.h>
...@@ -11,7 +11,7 @@ namespace titandb { ...@@ -11,7 +11,7 @@ namespace titandb {
const size_t kMaxFileCacheSize = 1024 * 1024; const size_t kMaxFileCacheSize = 1024 * 1024;
VersionSet::VersionSet(const TitanDBOptions& options, TitanStats* stats) BlobFileSet::BlobFileSet(const TitanDBOptions& options, TitanStats* stats)
: dirname_(options.dirname), : dirname_(options.dirname),
env_(options.env), env_(options.env),
env_options_(options), env_options_(options),
...@@ -24,7 +24,7 @@ VersionSet::VersionSet(const TitanDBOptions& options, TitanStats* stats) ...@@ -24,7 +24,7 @@ VersionSet::VersionSet(const TitanDBOptions& options, TitanStats* stats)
file_cache_ = NewLRUCache(file_cache_size); file_cache_ = NewLRUCache(file_cache_size);
} }
Status VersionSet::Open( Status BlobFileSet::Open(
const std::map<uint32_t, TitanCFOptions>& column_families) { const std::map<uint32_t, TitanCFOptions>& column_families) {
// Sets up initial column families. // Sets up initial column families.
AddColumnFamilies(column_families); AddColumnFamilies(column_families);
...@@ -39,7 +39,7 @@ Status VersionSet::Open( ...@@ -39,7 +39,7 @@ Status VersionSet::Open(
return OpenManifest(NewFileNumber()); return OpenManifest(NewFileNumber());
} }
Status VersionSet::Recover() { Status BlobFileSet::Recover() {
struct LogReporter : public log::Reader::Reporter { struct LogReporter : public log::Reader::Reporter {
Status* status; Status* status;
void Corruption(size_t, const Status& s) override { void Corruption(size_t, const Status& s) override {
...@@ -145,7 +145,7 @@ Status VersionSet::Recover() { ...@@ -145,7 +145,7 @@ Status VersionSet::Recover() {
return Status::OK(); return Status::OK();
} }
Status VersionSet::OpenManifest(uint64_t file_number) { Status BlobFileSet::OpenManifest(uint64_t file_number) {
Status s; Status s;
auto file_name = DescriptorFileName(dirname_, file_number); auto file_name = DescriptorFileName(dirname_, file_number);
...@@ -177,7 +177,7 @@ Status VersionSet::OpenManifest(uint64_t file_number) { ...@@ -177,7 +177,7 @@ Status VersionSet::OpenManifest(uint64_t file_number) {
return s; return s;
} }
Status VersionSet::WriteSnapshot(log::Writer* log) { Status BlobFileSet::WriteSnapshot(log::Writer* log) {
Status s; Status s;
// Saves global information // Saves global information
{ {
...@@ -207,8 +207,8 @@ Status VersionSet::WriteSnapshot(log::Writer* log) { ...@@ -207,8 +207,8 @@ Status VersionSet::WriteSnapshot(log::Writer* log) {
return s; return s;
} }
Status VersionSet::LogAndApply(VersionEdit& edit) { Status BlobFileSet::LogAndApply(VersionEdit& edit) {
TEST_SYNC_POINT("VersionSet::LogAndApply"); TEST_SYNC_POINT("BlobFileSet::LogAndApply");
// TODO(@huachao): write manifest file unlocked // TODO(@huachao): write manifest file unlocked
std::string record; std::string record;
edit.SetNextFileNumber(next_file_number_.load()); edit.SetNextFileNumber(next_file_number_.load());
...@@ -228,7 +228,7 @@ Status VersionSet::LogAndApply(VersionEdit& edit) { ...@@ -228,7 +228,7 @@ Status VersionSet::LogAndApply(VersionEdit& edit) {
return collector.Apply(*this); return collector.Apply(*this);
} }
void VersionSet::AddColumnFamilies( void BlobFileSet::AddColumnFamilies(
const std::map<uint32_t, TitanCFOptions>& column_families) { const std::map<uint32_t, TitanCFOptions>& column_families) {
for (auto& cf : column_families) { for (auto& cf : column_families) {
auto file_cache = std::make_shared<BlobFileCache>(db_options_, cf.second, auto file_cache = std::make_shared<BlobFileCache>(db_options_, cf.second,
...@@ -239,7 +239,7 @@ void VersionSet::AddColumnFamilies( ...@@ -239,7 +239,7 @@ void VersionSet::AddColumnFamilies(
} }
} }
Status VersionSet::DropColumnFamilies( Status BlobFileSet::DropColumnFamilies(
const std::vector<uint32_t>& column_families, const std::vector<uint32_t>& column_families,
SequenceNumber obsolete_sequence) { SequenceNumber obsolete_sequence) {
Status s; Status s;
...@@ -266,7 +266,7 @@ Status VersionSet::DropColumnFamilies( ...@@ -266,7 +266,7 @@ Status VersionSet::DropColumnFamilies(
return s; return s;
} }
Status VersionSet::MaybeDestroyColumnFamily(uint32_t cf_id) { Status BlobFileSet::MaybeDestroyColumnFamily(uint32_t cf_id) {
obsolete_columns_.erase(cf_id); obsolete_columns_.erase(cf_id);
auto it = column_families_.find(cf_id); auto it = column_families_.find(cf_id);
if (it != column_families_.end()) { if (it != column_families_.end()) {
...@@ -281,8 +281,8 @@ Status VersionSet::MaybeDestroyColumnFamily(uint32_t cf_id) { ...@@ -281,8 +281,8 @@ Status VersionSet::MaybeDestroyColumnFamily(uint32_t cf_id) {
return Status::NotFound("invalid column family"); return Status::NotFound("invalid column family");
} }
void VersionSet::GetObsoleteFiles(std::vector<std::string>* obsolete_files, void BlobFileSet::GetObsoleteFiles(std::vector<std::string>* obsolete_files,
SequenceNumber oldest_sequence) { SequenceNumber oldest_sequence) {
for (auto it = column_families_.begin(); it != column_families_.end();) { for (auto it = column_families_.begin(); it != column_families_.end();) {
auto& cf_id = it->first; auto& cf_id = it->first;
auto& blob_storage = it->second; auto& blob_storage = it->second;
......
src/version_set.h → src/blob_file_set.h
View file @ 07aa0655
...@@ -20,9 +20,11 @@ ...@@ -20,9 +20,11 @@
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
class VersionSet { // BlobFileSet is the set of all the blobs file generated by Titan.
// It records blob file meta in terms of column family.
class BlobFileSet {
public: public:
explicit VersionSet(const TitanDBOptions& options, TitanStats* stats); explicit BlobFileSet(const TitanDBOptions& options, TitanStats* stats);
// Sets up the storage specified in "options.dirname". // Sets up the storage specified in "options.dirname".
// If the manifest doesn't exist, it will create one. // If the manifest doesn't exist, it will create one.
...@@ -31,8 +33,7 @@ class VersionSet { ...@@ -31,8 +33,7 @@ class VersionSet {
// outside of the provided column families. // outside of the provided column families.
Status Open(const std::map<uint32_t, TitanCFOptions>& column_families); Status Open(const std::map<uint32_t, TitanCFOptions>& column_families);
// Applies *edit on the current version to form a new version that is // Applies *edit and saved to the manifest.
// both saved to the manifest and installed as the new current version.
// REQUIRES: mutex is held // REQUIRES: mutex is held
Status LogAndApply(VersionEdit& edit); Status LogAndApply(VersionEdit& edit);
......
src/blob_file_size_collector.h
View file @ 07aa0655
#pragma once #pragma once
#include "blob_file_set.h"
#include "db_impl.h" #include "db_impl.h"
#include "rocksdb/listener.h" #include "rocksdb/listener.h"
#include "rocksdb/table_properties.h" #include "rocksdb/table_properties.h"
#include "util/coding.h" #include "util/coding.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
......
src/blob_gc_job.cc
View file @ 07aa0655
...@@ -76,7 +76,7 @@ BlobGCJob::BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex, ...@@ -76,7 +76,7 @@ BlobGCJob::BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex,
const TitanDBOptions& titan_db_options, Env* env, const TitanDBOptions& titan_db_options, Env* env,
const EnvOptions& env_options, const EnvOptions& env_options,
BlobFileManager* blob_file_manager, BlobFileManager* blob_file_manager,
VersionSet* version_set, LogBuffer* log_buffer, BlobFileSet* blob_file_set, LogBuffer* log_buffer,
std::atomic_bool* shuting_down, TitanStats* stats) std::atomic_bool* shuting_down, TitanStats* stats)
: blob_gc_(blob_gc), : blob_gc_(blob_gc),
base_db_(db), base_db_(db),
...@@ -86,7 +86,7 @@ BlobGCJob::BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex, ...@@ -86,7 +86,7 @@ BlobGCJob::BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex,
env_(env), env_(env),
env_options_(env_options), env_options_(env_options),
blob_file_manager_(blob_file_manager), blob_file_manager_(blob_file_manager),
version_set_(version_set), blob_file_set_(blob_file_set),
log_buffer_(log_buffer), log_buffer_(log_buffer),
shuting_down_(shuting_down), shuting_down_(shuting_down),
stats_(stats) {} stats_(stats) {}
...@@ -586,7 +586,7 @@ Status BlobGCJob::DeleteInputBlobFiles() { ...@@ -586,7 +586,7 @@ Status BlobGCJob::DeleteInputBlobFiles() {
metrics_.blob_db_gc_num_files++; metrics_.blob_db_gc_num_files++;
edit.DeleteBlobFile(file->file_number(), obsolete_sequence); edit.DeleteBlobFile(file->file_number(), obsolete_sequence);
} }
s = version_set_->LogAndApply(edit); s = blob_file_set_->LogAndApply(edit);
// TODO(@DorianZheng) Purge pending outputs // TODO(@DorianZheng) Purge pending outputs
// base_db_->pending_outputs_.erase(handle->GetNumber()); // base_db_->pending_outputs_.erase(handle->GetNumber());
return s; return s;
......
src/blob_gc_job.h
View file @ 07aa0655
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
#include "blob_file_builder.h" #include "blob_file_builder.h"
#include "blob_file_iterator.h" #include "blob_file_iterator.h"
#include "blob_file_manager.h" #include "blob_file_manager.h"
#include "blob_file_set.h"
#include "blob_gc.h" #include "blob_gc.h"
#include "db/db_impl/db_impl.h" #include "db/db_impl/db_impl.h"
#include "rocksdb/statistics.h" #include "rocksdb/statistics.h"
...@@ -10,7 +11,6 @@ ...@@ -10,7 +11,6 @@
#include "titan/options.h" #include "titan/options.h"
#include "titan_stats.h" #include "titan_stats.h"
#include "version_edit.h" #include "version_edit.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
...@@ -20,7 +20,7 @@ class BlobGCJob { ...@@ -20,7 +20,7 @@ class BlobGCJob {
BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex, BlobGCJob(BlobGC* blob_gc, DB* db, port::Mutex* mutex,
const TitanDBOptions& titan_db_options, Env* env, const TitanDBOptions& titan_db_options, Env* env,
const EnvOptions& env_options, BlobFileManager* blob_file_manager, const EnvOptions& env_options, BlobFileManager* blob_file_manager,
VersionSet* version_set, LogBuffer* log_buffer, BlobFileSet* blob_file_set, LogBuffer* log_buffer,
std::atomic_bool* shuting_down, TitanStats* stats); std::atomic_bool* shuting_down, TitanStats* stats);
// No copying allowed // No copying allowed
...@@ -50,7 +50,7 @@ class BlobGCJob { ...@@ -50,7 +50,7 @@ class BlobGCJob {
Env* env_; Env* env_;
EnvOptions env_options_; EnvOptions env_options_;
BlobFileManager* blob_file_manager_; BlobFileManager* blob_file_manager_;
VersionSet* version_set_; BlobFileSet* blob_file_set_;
LogBuffer* log_buffer_{nullptr}; LogBuffer* log_buffer_{nullptr};
std::vector<std::pair<std::unique_ptr<BlobFileHandle>, std::vector<std::pair<std::unique_ptr<BlobFileHandle>,
......
src/blob_gc_job_test.cc
View file @ 07aa0655
...@@ -28,7 +28,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -28,7 +28,7 @@ class BlobGCJobTest : public testing::Test {
TitanDB* db_; TitanDB* db_;
DBImpl* base_db_; DBImpl* base_db_;
TitanDBImpl* tdb_; TitanDBImpl* tdb_;
VersionSet* version_set_; BlobFileSet* blob_file_set_;
TitanOptions options_; TitanOptions options_;
port::Mutex* mutex_; port::Mutex* mutex_;
...@@ -45,7 +45,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -45,7 +45,7 @@ class BlobGCJobTest : public testing::Test {
std::weak_ptr<BlobStorage> GetBlobStorage(uint32_t cf_id) { std::weak_ptr<BlobStorage> GetBlobStorage(uint32_t cf_id) {
MutexLock l(mutex_); MutexLock l(mutex_);
return version_set_->GetBlobStorage(cf_id); return blob_file_set_->GetBlobStorage(cf_id);
} }
void CheckBlobNumber(int expected) { void CheckBlobNumber(int expected) {
...@@ -75,7 +75,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -75,7 +75,7 @@ class BlobGCJobTest : public testing::Test {
ClearDir(); ClearDir();
ASSERT_OK(TitanDB::Open(options_, dbname_, &db_)); ASSERT_OK(TitanDB::Open(options_, dbname_, &db_));
tdb_ = reinterpret_cast<TitanDBImpl*>(db_); tdb_ = reinterpret_cast<TitanDBImpl*>(db_);
version_set_ = tdb_->vset_.get(); blob_file_set_ = tdb_->blob_file_set_.get();
mutex_ = &tdb_->mutex_; mutex_ = &tdb_->mutex_;
base_db_ = reinterpret_cast<DBImpl*>(tdb_->GetRootDB()); base_db_ = reinterpret_cast<DBImpl*>(tdb_->GetRootDB());
} }
...@@ -120,7 +120,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -120,7 +120,7 @@ class BlobGCJobTest : public testing::Test {
std::shared_ptr<BlobGCPicker> blob_gc_picker = std::shared_ptr<BlobGCPicker> blob_gc_picker =
std::make_shared<BasicBlobGCPicker>(db_options, cf_options); std::make_shared<BasicBlobGCPicker>(db_options, cf_options);
blob_gc = blob_gc_picker->PickBlobGC( blob_gc = blob_gc_picker->PickBlobGC(
version_set_->GetBlobStorage(cfh->GetID()).lock().get()); blob_file_set_->GetBlobStorage(cfh->GetID()).lock().get());
} }
if (expected) { if (expected) {
...@@ -132,7 +132,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -132,7 +132,7 @@ class BlobGCJobTest : public testing::Test {
BlobGCJob blob_gc_job(blob_gc.get(), base_db_, mutex_, tdb_->db_options_, BlobGCJob blob_gc_job(blob_gc.get(), base_db_, mutex_, tdb_->db_options_,
tdb_->env_, EnvOptions(options_), tdb_->env_, EnvOptions(options_),
tdb_->blob_manager_.get(), version_set_, tdb_->blob_manager_.get(), blob_file_set_,
&log_buffer, nullptr, nullptr); &log_buffer, nullptr, nullptr);
s = blob_gc_job.Prepare(); s = blob_gc_job.Prepare();
...@@ -189,7 +189,7 @@ class BlobGCJobTest : public testing::Test { ...@@ -189,7 +189,7 @@ class BlobGCJobTest : public testing::Test {
BlobGC blob_gc(std::move(tmp), TitanCFOptions(), false /*trigger_next*/); BlobGC blob_gc(std::move(tmp), TitanCFOptions(), false /*trigger_next*/);
blob_gc.SetColumnFamily(cfh); blob_gc.SetColumnFamily(cfh);
BlobGCJob blob_gc_job(&blob_gc, base_db_, mutex_, TitanDBOptions(), BlobGCJob blob_gc_job(&blob_gc, base_db_, mutex_, TitanDBOptions(),
Env::Default(), EnvOptions(), nullptr, version_set_, Env::Default(), EnvOptions(), nullptr, blob_file_set_,
nullptr, nullptr, nullptr); nullptr, nullptr, nullptr);
bool discardable = false; bool discardable = false;
ASSERT_OK(blob_gc_job.DiscardEntry(key, blob_index, &discardable)); ASSERT_OK(blob_gc_job.DiscardEntry(key, blob_index, &discardable));
......
src/blob_storage.cc
View file @ 07aa0655
#include "blob_storage.h" #include "blob_storage.h"
#include "version_set.h" #include "blob_file_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
......
src/blob_storage.h
View file @ 07aa0655
...@@ -13,7 +13,7 @@ namespace rocksdb { ...@@ -13,7 +13,7 @@ namespace rocksdb {
namespace titandb { namespace titandb {
// Provides methods to access the blob storage for a specific // Provides methods to access the blob storage for a specific
// column family. The version must be valid when this storage is used. // column family.
class BlobStorage { class BlobStorage {
public: public:
BlobStorage(const BlobStorage& bs) : destroyed_(false) { BlobStorage(const BlobStorage& bs) : destroyed_(false) {
...@@ -101,7 +101,7 @@ class BlobStorage { ...@@ -101,7 +101,7 @@ class BlobStorage {
SequenceNumber obsolete_sequence); SequenceNumber obsolete_sequence);
private: private:
friend class VersionSet; friend class BlobFileSet;
friend class VersionTest; friend class VersionTest;
friend class BlobGCPickerTest; friend class BlobGCPickerTest;
friend class BlobGCJobTest; friend class BlobGCJobTest;
......
src/db_impl.cc
View file @ 07aa0655
...@@ -27,7 +27,7 @@ class TitanDBImpl::FileManager : public BlobFileManager { ...@@ -27,7 +27,7 @@ class TitanDBImpl::FileManager : public BlobFileManager {
FileManager(TitanDBImpl* db) : db_(db) {} FileManager(TitanDBImpl* db) : db_(db) {}
Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override { Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override {
auto number = db_->vset_->NewFileNumber(); auto number = db_->blob_file_set_->NewFileNumber();
auto name = BlobFileName(db_->dirname_, number); auto name = BlobFileName(db_->dirname_, number);
Status s; Status s;
...@@ -75,7 +75,7 @@ class TitanDBImpl::FileManager : public BlobFileManager { ...@@ -75,7 +75,7 @@ class TitanDBImpl::FileManager : public BlobFileManager {
{ {
MutexLock l(&db_->mutex_); MutexLock l(&db_->mutex_);
s = db_->vset_->LogAndApply(edit); s = db_->blob_file_set_->LogAndApply(edit);
if (!s.ok()) { if (!s.ok()) {
db_->SetBGError(s); db_->SetBGError(s);
} }
...@@ -220,9 +220,9 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs, ...@@ -220,9 +220,9 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs,
// Add EventListener to collect statistics for GC // Add EventListener to collect statistics for GC
db_options_.listeners.emplace_back(std::make_shared<BaseDbListener>(this)); db_options_.listeners.emplace_back(std::make_shared<BaseDbListener>(this));
// Note that info log is initialized after `CreateLoggerFromOptions`, // Note that info log is initialized after `CreateLoggerFromOptions`,
// so new `VersionSet` here but not in constructor is to get a proper info // so new `BlobFileSet` here but not in constructor is to get a proper info
// log. // log.
vset_.reset(new VersionSet(db_options_, stats_.get())); blob_file_set_.reset(new BlobFileSet(db_options_, stats_.get()));
s = DB::Open(db_options_, dbname_, init_descs, handles, &db_); s = DB::Open(db_options_, dbname_, init_descs, handles, &db_);
if (s.ok()) { if (s.ok()) {
...@@ -237,8 +237,8 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs, ...@@ -237,8 +237,8 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs,
auto& base_table_factory = base_descs[i].options.table_factory; auto& base_table_factory = base_descs[i].options.table_factory;
assert(base_table_factory != nullptr); assert(base_table_factory != nullptr);
auto titan_table_factory = std::make_shared<TitanTableFactory>( auto titan_table_factory = std::make_shared<TitanTableFactory>(
db_options_, descs[i].options, blob_manager_, &mutex_, vset_.get(), db_options_, descs[i].options, blob_manager_, &mutex_,
stats_.get()); blob_file_set_.get(), stats_.get());
cf_info_.emplace(cf_id, cf_info_.emplace(cf_id,
TitanColumnFamilyInfo( TitanColumnFamilyInfo(
{cf_name, ImmutableTitanCFOptions(descs[i].options), {cf_name, ImmutableTitanCFOptions(descs[i].options),
...@@ -256,7 +256,7 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs, ...@@ -256,7 +256,7 @@ Status TitanDBImpl::Open(const std::vector<TitanCFDescriptor>& descs,
} }
if (!s.ok()) return s; if (!s.ok()) return s;
s = vset_->Open(column_families); s = blob_file_set_->Open(column_families);
if (!s.ok()) return s; if (!s.ok()) return s;
static bool has_init_background_threads = false; static bool has_init_background_threads = false;
...@@ -355,7 +355,7 @@ Status TitanDBImpl::CreateColumnFamilies( ...@@ -355,7 +355,7 @@ Status TitanDBImpl::CreateColumnFamilies(
// Replaces the provided table factory with TitanTableFactory. // Replaces the provided table factory with TitanTableFactory.
base_table_factory.emplace_back(options.table_factory); base_table_factory.emplace_back(options.table_factory);
titan_table_factory.emplace_back(std::make_shared<TitanTableFactory>( titan_table_factory.emplace_back(std::make_shared<TitanTableFactory>(
db_options_, desc.options, blob_manager_, &mutex_, vset_.get(), db_options_, desc.options, blob_manager_, &mutex_, blob_file_set_.get(),
stats_.get())); stats_.get()));
options.table_factory = titan_table_factory.back(); options.table_factory = titan_table_factory.back();
base_descs.emplace_back(desc.name, options); base_descs.emplace_back(desc.name, options);
...@@ -379,7 +379,7 @@ Status TitanDBImpl::CreateColumnFamilies( ...@@ -379,7 +379,7 @@ Status TitanDBImpl::CreateColumnFamilies(
MutableTitanCFOptions(descs[i].options), base_table_factory[i], MutableTitanCFOptions(descs[i].options), base_table_factory[i],
titan_table_factory[i]})); titan_table_factory[i]}));
} }
vset_->AddColumnFamilies(column_families); blob_file_set_->AddColumnFamilies(column_families);
} }
} }
if (s.ok()) { if (s.ok()) {
...@@ -412,7 +412,7 @@ Status TitanDBImpl::DropColumnFamilies( ...@@ -412,7 +412,7 @@ Status TitanDBImpl::DropColumnFamilies(
if (s.ok()) { if (s.ok()) {
MutexLock l(&mutex_); MutexLock l(&mutex_);
SequenceNumber obsolete_sequence = db_impl_->GetLatestSequenceNumber(); SequenceNumber obsolete_sequence = db_impl_->GetLatestSequenceNumber();
s = vset_->DropColumnFamilies(column_families, obsolete_sequence); s = blob_file_set_->DropColumnFamilies(column_families, obsolete_sequence);
} }
if (s.ok()) { if (s.ok()) {
ROCKS_LOG_INFO(db_options_.info_log, "Dropped column families: %s", ROCKS_LOG_INFO(db_options_.info_log, "Dropped column families: %s",
...@@ -437,8 +437,8 @@ Status TitanDBImpl::DestroyColumnFamilyHandle( ...@@ -437,8 +437,8 @@ Status TitanDBImpl::DestroyColumnFamilyHandle(
if (s.ok()) { if (s.ok()) {
MutexLock l(&mutex_); MutexLock l(&mutex_);
// it just changes some marks and doesn't delete blob files physically. // it just changes some marks and doesn't delete blob files physically.
Status destroy_status = vset_->MaybeDestroyColumnFamily(cf_id); Status destroy_status = blob_file_set_->MaybeDestroyColumnFamily(cf_id);
// VersionSet will return NotFound status if the cf is not destroyed. // BlobFileSet will return NotFound status if the cf is not destroyed.
if (destroy_status.ok()) { if (destroy_status.ok()) {
assert(cf_info_.count(cf_id) > 0); assert(cf_info_.count(cf_id) > 0);
cf_info_.erase(cf_id); cf_info_.erase(cf_id);
...@@ -548,7 +548,7 @@ Status TitanDBImpl::GetImpl(const ReadOptions& options, ...@@ -548,7 +548,7 @@ Status TitanDBImpl::GetImpl(const ReadOptions& options,
PinnableSlice buffer; PinnableSlice buffer;
mutex_.Lock(); mutex_.Lock();
auto storage = vset_->GetBlobStorage(handle->GetID()).lock(); auto storage = blob_file_set_->GetBlobStorage(handle->GetID()).lock();
mutex_.Unlock(); mutex_.Unlock();
{ {
...@@ -624,7 +624,7 @@ Iterator* TitanDBImpl::NewIteratorImpl( ...@@ -624,7 +624,7 @@ Iterator* TitanDBImpl::NewIteratorImpl(
auto cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(handle)->cfd(); auto cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(handle)->cfd();
mutex_.Lock(); mutex_.Lock();
auto storage = vset_->GetBlobStorage(handle->GetID()); auto storage = blob_file_set_->GetBlobStorage(handle->GetID());
mutex_.Unlock(); mutex_.Unlock();
std::unique_ptr<ArenaWrappedDBIter> iter(db_impl_->NewIteratorImpl( std::unique_ptr<ArenaWrappedDBIter> iter(db_impl_->NewIteratorImpl(
...@@ -777,7 +777,7 @@ Status TitanDBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family, ...@@ -777,7 +777,7 @@ Status TitanDBImpl::DeleteFilesInRanges(ColumnFamilyHandle* column_family,
if (!s.ok()) return s; if (!s.ok()) return s;
MutexLock l(&mutex_); MutexLock l(&mutex_);
auto bs = vset_->GetBlobStorage(cf_id).lock(); auto bs = blob_file_set_->GetBlobStorage(cf_id).lock();
if (!bs) { if (!bs) {
// TODO: Should treat it as background error and make DB read-only. // TODO: Should treat it as background error and make DB read-only.
ROCKS_LOG_ERROR(db_options_.info_log, ROCKS_LOG_ERROR(db_options_.info_log,
...@@ -954,7 +954,8 @@ void TitanDBImpl::OnFlushCompleted(const FlushJobInfo& flush_job_info) { ...@@ -954,7 +954,8 @@ void TitanDBImpl::OnFlushCompleted(const FlushJobInfo& flush_job_info) {
{ {
MutexLock l(&mutex_); MutexLock l(&mutex_);
auto blob_storage = vset_->GetBlobStorage(flush_job_info.cf_id).lock(); auto blob_storage =
blob_file_set_->GetBlobStorage(flush_job_info.cf_id).lock();
if (!blob_storage) { if (!blob_storage) {
// TODO: Should treat it as background error and make DB read-only. // TODO: Should treat it as background error and make DB read-only.
ROCKS_LOG_ERROR(db_options_.info_log, ROCKS_LOG_ERROR(db_options_.info_log,
...@@ -1042,7 +1043,7 @@ void TitanDBImpl::OnCompactionCompleted( ...@@ -1042,7 +1043,7 @@ void TitanDBImpl::OnCompactionCompleted(
{ {
MutexLock l(&mutex_); MutexLock l(&mutex_);
auto bs = vset_->GetBlobStorage(compaction_job_info.cf_id).lock(); auto bs = blob_file_set_->GetBlobStorage(compaction_job_info.cf_id).lock();
if (!bs) { if (!bs) {
// TODO: Should treat it as background error and make DB read-only. // TODO: Should treat it as background error and make DB read-only.
ROCKS_LOG_ERROR(db_options_.info_log, ROCKS_LOG_ERROR(db_options_.info_log,
......
src/db_impl.h
View file @ 07aa0655
...@@ -5,10 +5,10 @@ ...@@ -5,10 +5,10 @@
#include "util/repeatable_thread.h" #include "util/repeatable_thread.h"
#include "blob_file_manager.h" #include "blob_file_manager.h"
#include "blob_file_set.h"
#include "table_factory.h" #include "table_factory.h"
#include "titan/db.h" #include "titan/db.h"
#include "titan_stats.h" #include "titan_stats.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
...@@ -245,7 +245,7 @@ class TitanDBImpl : public TitanDB { ...@@ -245,7 +245,7 @@ class TitanDBImpl : public TitanDB {
// handle for dump internal stats at fixed intervals. // handle for dump internal stats at fixed intervals.
std::unique_ptr<RepeatableThread> thread_dump_stats_; std::unique_ptr<RepeatableThread> thread_dump_stats_;
std::unique_ptr<VersionSet> vset_; std::unique_ptr<BlobFileSet> blob_file_set_;
std::set<uint64_t> pending_outputs_; std::set<uint64_t> pending_outputs_;
std::shared_ptr<BlobFileManager> blob_manager_; std::shared_ptr<BlobFileManager> blob_manager_;
......
src/db_impl_files.cc
View file @ 07aa0655
...@@ -9,7 +9,7 @@ Status TitanDBImpl::PurgeObsoleteFilesImpl() { ...@@ -9,7 +9,7 @@ Status TitanDBImpl::PurgeObsoleteFilesImpl() {
auto oldest_sequence = GetOldestSnapshotSequence(); auto oldest_sequence = GetOldestSnapshotSequence();
{ {
MutexLock l(&mutex_); MutexLock l(&mutex_);
vset_->GetObsoleteFiles(&candidate_files, oldest_sequence); blob_file_set_->GetObsoleteFiles(&candidate_files, oldest_sequence);
} }
// dedup state.inputs so we don't try to delete the same // dedup state.inputs so we don't try to delete the same
......
src/db_impl_gc.cc
View file @ 07aa0655
...@@ -67,7 +67,7 @@ Status TitanDBImpl::BackgroundGC(LogBuffer* log_buffer) { ...@@ -67,7 +67,7 @@ Status TitanDBImpl::BackgroundGC(LogBuffer* log_buffer) {
Status s; Status s;
if (!gc_queue_.empty()) { if (!gc_queue_.empty()) {
uint32_t column_family_id = PopFirstFromGCQueue(); uint32_t column_family_id = PopFirstFromGCQueue();
auto bs = vset_->GetBlobStorage(column_family_id).lock().get(); auto bs = blob_file_set_->GetBlobStorage(column_family_id).lock().get();
if (bs) { if (bs) {
const auto& cf_options = bs->cf_options(); const auto& cf_options = bs->cf_options();
...@@ -90,8 +90,9 @@ Status TitanDBImpl::BackgroundGC(LogBuffer* log_buffer) { ...@@ -90,8 +90,9 @@ Status TitanDBImpl::BackgroundGC(LogBuffer* log_buffer) {
ROCKS_LOG_BUFFER(log_buffer, "Titan GC nothing to do"); ROCKS_LOG_BUFFER(log_buffer, "Titan GC nothing to do");
} else { } else {
BlobGCJob blob_gc_job(blob_gc.get(), db_, &mutex_, db_options_, env_, BlobGCJob blob_gc_job(blob_gc.get(), db_, &mutex_, db_options_, env_,
env_options_, blob_manager_.get(), vset_.get(), env_options_, blob_manager_.get(),
log_buffer, &shuting_down_, stats_.get()); blob_file_set_.get(), log_buffer, &shuting_down_,
stats_.get());
s = blob_gc_job.Prepare(); s = blob_gc_job.Prepare();
if (s.ok()) { if (s.ok()) {
mutex_.Unlock(); mutex_.Unlock();
...@@ -142,7 +143,7 @@ Status TitanDBImpl::TEST_StartGC(uint32_t column_family_id) { ...@@ -142,7 +143,7 @@ Status TitanDBImpl::TEST_StartGC(uint32_t column_family_id) {
std::unique_ptr<BlobGC> blob_gc; std::unique_ptr<BlobGC> blob_gc;
std::unique_ptr<ColumnFamilyHandle> cfh; std::unique_ptr<ColumnFamilyHandle> cfh;
auto bs = vset_->GetBlobStorage(column_family_id).lock().get(); auto bs = blob_file_set_->GetBlobStorage(column_family_id).lock().get();
const auto& cf_options = bs->cf_options(); const auto& cf_options = bs->cf_options();
std::shared_ptr<BlobGCPicker> blob_gc_picker = std::shared_ptr<BlobGCPicker> blob_gc_picker =
std::make_shared<BasicBlobGCPicker>(db_options_, cf_options); std::make_shared<BasicBlobGCPicker>(db_options_, cf_options);
...@@ -158,8 +159,9 @@ Status TitanDBImpl::TEST_StartGC(uint32_t column_family_id) { ...@@ -158,8 +159,9 @@ Status TitanDBImpl::TEST_StartGC(uint32_t column_family_id) {
ROCKS_LOG_BUFFER(&log_buffer, "Titan GC nothing to do"); ROCKS_LOG_BUFFER(&log_buffer, "Titan GC nothing to do");
} else { } else {
BlobGCJob blob_gc_job(blob_gc.get(), db_, &mutex_, db_options_, env_, BlobGCJob blob_gc_job(blob_gc.get(), db_, &mutex_, db_options_, env_,
env_options_, blob_manager_.get(), vset_.get(), env_options_, blob_manager_.get(),
&log_buffer, &shuting_down_, stats_.get()); blob_file_set_.get(), &log_buffer, &shuting_down_,
stats_.get());
s = blob_gc_job.Prepare(); s = blob_gc_job.Prepare();
if (s.ok()) { if (s.ok()) {
......
src/edit_collector.h
View file @ 07aa0655
...@@ -2,9 +2,9 @@ ...@@ -2,9 +2,9 @@
#include <unordered_map> #include <unordered_map>
#include "blob_file_set.h"
#include "util/string_util.h" #include "util/string_util.h"
#include "version_edit.h" #include "version_edit.h"
#include "version_set.h"
#include <inttypes.h> #include <inttypes.h>
...@@ -49,12 +49,12 @@ class EditCollector { ...@@ -49,12 +49,12 @@ class EditCollector {
} }
// Seal the batch and check the validation of the edits. // Seal the batch and check the validation of the edits.
Status Seal(VersionSet& vset) { Status Seal(BlobFileSet& blob_file_set) {
if (!status_.ok()) return status_; if (!status_.ok()) return status_;
for (auto& cf : column_families_) { for (auto& cf : column_families_) {
auto cf_id = cf.first; auto cf_id = cf.first;
auto storage = vset.GetBlobStorage(cf_id).lock(); auto storage = blob_file_set.GetBlobStorage(cf_id).lock();
if (!storage) { if (!storage) {
// TODO: support OpenForReadOnly which doesn't open DB with all column // TODO: support OpenForReadOnly which doesn't open DB with all column
// family so there are maybe some invalid column family, but we can't // family so there are maybe some invalid column family, but we can't
...@@ -71,7 +71,7 @@ class EditCollector { ...@@ -71,7 +71,7 @@ class EditCollector {
} }
// Apply the edits of the batch. // Apply the edits of the batch.
Status Apply(VersionSet& vset) { Status Apply(BlobFileSet& blob_file_set) {
if (!status_.ok()) return status_; if (!status_.ok()) return status_;
if (!sealed_) if (!sealed_)
return Status::Incomplete( return Status::Incomplete(
...@@ -79,7 +79,7 @@ class EditCollector { ...@@ -79,7 +79,7 @@ class EditCollector {
for (auto& cf : column_families_) { for (auto& cf : column_families_) {
auto cf_id = cf.first; auto cf_id = cf.first;
auto storage = vset.GetBlobStorage(cf_id).lock(); auto storage = blob_file_set.GetBlobStorage(cf_id).lock();
if (!storage) { if (!storage) {
// TODO: support OpenForReadOnly which doesn't open DB with all column // TODO: support OpenForReadOnly which doesn't open DB with all column
// family so there are maybe some invalid column family, but we can't // family so there are maybe some invalid column family, but we can't
......
src/table_builder.h
View file @ 07aa0655
...@@ -2,10 +2,10 @@ ...@@ -2,10 +2,10 @@
#include "blob_file_builder.h" #include "blob_file_builder.h"
#include "blob_file_manager.h" #include "blob_file_manager.h"
#include "blob_file_set.h"
#include "table/table_builder.h" #include "table/table_builder.h"
#include "titan/options.h" #include "titan/options.h"
#include "titan_stats.h" #include "titan_stats.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
......
src/table_builder_test.cc
View file @ 07aa0655
...@@ -5,9 +5,9 @@ ...@@ -5,9 +5,9 @@
#include "blob_file_manager.h" #include "blob_file_manager.h"
#include "blob_file_reader.h" #include "blob_file_reader.h"
#include "blob_file_set.h"
#include "table_builder.h" #include "table_builder.h"
#include "table_factory.h" #include "table_factory.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
...@@ -18,8 +18,10 @@ const uint64_t kTargetBlobFileSize = 4096; ...@@ -18,8 +18,10 @@ const uint64_t kTargetBlobFileSize = 4096;
class FileManager : public BlobFileManager { class FileManager : public BlobFileManager {
public: public:
FileManager(const TitanDBOptions& db_options, VersionSet* vset) FileManager(const TitanDBOptions& db_options, BlobFileSet* blob_file_set)
: db_options_(db_options), number_(kTestFileNumber), vset_(vset) {} : db_options_(db_options),
number_(kTestFileNumber),
blob_file_set_(blob_file_set) {}
Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override { Status NewFile(std::unique_ptr<BlobFileHandle>* handle) override {
auto number = number_.fetch_add(1); auto number = number_.fetch_add(1);
...@@ -41,7 +43,7 @@ class FileManager : public BlobFileManager { ...@@ -41,7 +43,7 @@ class FileManager : public BlobFileManager {
Status s = handle->GetFile()->Sync(true); Status s = handle->GetFile()->Sync(true);
if (s.ok()) { if (s.ok()) {
s = handle->GetFile()->Close(); s = handle->GetFile()->Close();
auto storage = vset_->GetBlobStorage(0).lock(); auto storage = blob_file_set_->GetBlobStorage(0).lock();
storage->AddBlobFile(file); storage->AddBlobFile(file);
} }
return s; return s;
...@@ -93,7 +95,7 @@ class FileManager : public BlobFileManager { ...@@ -93,7 +95,7 @@ class FileManager : public BlobFileManager {
EnvOptions env_options_; EnvOptions env_options_;
TitanDBOptions db_options_; TitanDBOptions db_options_;
std::atomic<uint64_t> number_{0}; std::atomic<uint64_t> number_{0};
VersionSet* vset_; BlobFileSet* blob_file_set_;
}; };
class TableBuilderTest : public testing::Test { class TableBuilderTest : public testing::Test {
...@@ -106,13 +108,13 @@ class TableBuilderTest : public testing::Test { ...@@ -106,13 +108,13 @@ class TableBuilderTest : public testing::Test {
blob_name_(BlobFileName(tmpdir_, kTestFileNumber)) { blob_name_(BlobFileName(tmpdir_, kTestFileNumber)) {
db_options_.dirname = tmpdir_; db_options_.dirname = tmpdir_;
cf_options_.min_blob_size = kMinBlobSize; cf_options_.min_blob_size = kMinBlobSize;
vset_.reset(new VersionSet(db_options_, nullptr)); blob_file_set_.reset(new BlobFileSet(db_options_, nullptr));
std::map<uint32_t, TitanCFOptions> cfs{{0, cf_options_}}; std::map<uint32_t, TitanCFOptions> cfs{{0, cf_options_}};
vset_->AddColumnFamilies(cfs); blob_file_set_->AddColumnFamilies(cfs);
blob_manager_.reset(new FileManager(db_options_, vset_.get())); blob_manager_.reset(new FileManager(db_options_, blob_file_set_.get()));
table_factory_.reset(new TitanTableFactory(db_options_, cf_options_, table_factory_.reset(new TitanTableFactory(db_options_, cf_options_,
blob_manager_, &mutex_, blob_manager_, &mutex_,
vset_.get(), nullptr)); blob_file_set_.get(), nullptr));
} }
~TableBuilderTest() { ~TableBuilderTest() {
...@@ -205,7 +207,7 @@ class TableBuilderTest : public testing::Test { ...@@ -205,7 +207,7 @@ class TableBuilderTest : public testing::Test {
std::string blob_name_; std::string blob_name_;
std::unique_ptr<TableFactory> table_factory_; std::unique_ptr<TableFactory> table_factory_;
std::shared_ptr<BlobFileManager> blob_manager_; std::shared_ptr<BlobFileManager> blob_manager_;
std::unique_ptr<VersionSet> vset_; std::unique_ptr<BlobFileSet> blob_file_set_;
}; };
TEST_F(TableBuilderTest, Basic) { TEST_F(TableBuilderTest, Basic) {
...@@ -354,8 +356,9 @@ TEST_F(TableBuilderTest, NumEntries) { ...@@ -354,8 +356,9 @@ TEST_F(TableBuilderTest, NumEntries) {
// To test size of each blob file is around blob_file_target_size after building // To test size of each blob file is around blob_file_target_size after building
TEST_F(TableBuilderTest, TargetSize) { TEST_F(TableBuilderTest, TargetSize) {
cf_options_.blob_file_target_size = kTargetBlobFileSize; cf_options_.blob_file_target_size = kTargetBlobFileSize;
table_factory_.reset(new TitanTableFactory( table_factory_.reset(new TitanTableFactory(db_options_, cf_options_,
db_options_, cf_options_, blob_manager_, &mutex_, vset_.get(), nullptr)); blob_manager_, &mutex_,
blob_file_set_.get(), nullptr));
std::unique_ptr<WritableFileWriter> base_file; std::unique_ptr<WritableFileWriter> base_file;
NewBaseFileWriter(&base_file); NewBaseFileWriter(&base_file);
std::unique_ptr<TableBuilder> table_builder; std::unique_ptr<TableBuilder> table_builder;
...@@ -384,8 +387,9 @@ TEST_F(TableBuilderTest, TargetSize) { ...@@ -384,8 +387,9 @@ TEST_F(TableBuilderTest, TargetSize) {
// correct // correct
TEST_F(TableBuilderTest, LevelMerge) { TEST_F(TableBuilderTest, LevelMerge) {
cf_options_.level_merge = true; cf_options_.level_merge = true;
table_factory_.reset(new TitanTableFactory( table_factory_.reset(new TitanTableFactory(db_options_, cf_options_,
db_options_, cf_options_, blob_manager_, &mutex_, vset_.get(), nullptr)); blob_manager_, &mutex_,
blob_file_set_.get(), nullptr));
std::unique_ptr<WritableFileWriter> base_file; std::unique_ptr<WritableFileWriter> base_file;
NewBaseFileWriter(&base_file); NewBaseFileWriter(&base_file);
std::unique_ptr<TableBuilder> table_builder; std::unique_ptr<TableBuilder> table_builder;
...@@ -437,7 +441,7 @@ TEST_F(TableBuilderTest, LevelMerge) { ...@@ -437,7 +441,7 @@ TEST_F(TableBuilderTest, LevelMerge) {
// Compare key, index and blob records after level merge // Compare key, index and blob records after level merge
first_iter->SeekToFirst(); first_iter->SeekToFirst();
second_iter->SeekToFirst(); second_iter->SeekToFirst();
auto storage = vset_->GetBlobStorage(0).lock(); auto storage = blob_file_set_->GetBlobStorage(0).lock();
for (unsigned char i = 0; i < n; i++) { for (unsigned char i = 0; i < n; i++) {
ASSERT_TRUE(first_iter->Valid()); ASSERT_TRUE(first_iter->Valid());
ASSERT_TRUE(second_iter->Valid()); ASSERT_TRUE(second_iter->Valid());
......
src/table_factory.cc
View file @ 07aa0655
...@@ -30,7 +30,7 @@ TableBuilder* TitanTableFactory::NewTableBuilder( ...@@ -30,7 +30,7 @@ TableBuilder* TitanTableFactory::NewTableBuilder(
{ {
MutexLock l(db_mutex_); MutexLock l(db_mutex_);
blob_storage = vset_->GetBlobStorage(column_family_id); blob_storage = blob_file_set_->GetBlobStorage(column_family_id);
} }
return new TitanTableBuilder( return new TitanTableBuilder(
......
src/table_factory.h
View file @ 07aa0655
...@@ -3,10 +3,10 @@ ...@@ -3,10 +3,10 @@
#include <atomic> #include <atomic>
#include "blob_file_manager.h" #include "blob_file_manager.h"
#include "blob_file_set.h"
#include "rocksdb/table.h" #include "rocksdb/table.h"
#include "titan/options.h" #include "titan/options.h"
#include "titan_stats.h" #include "titan_stats.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
...@@ -16,14 +16,15 @@ class TitanTableFactory : public TableFactory { ...@@ -16,14 +16,15 @@ class TitanTableFactory : public TableFactory {
TitanTableFactory(const TitanDBOptions& db_options, TitanTableFactory(const TitanDBOptions& db_options,
const TitanCFOptions& cf_options, const TitanCFOptions& cf_options,
std::shared_ptr<BlobFileManager> blob_manager, std::shared_ptr<BlobFileManager> blob_manager,
port::Mutex* db_mutex, VersionSet* vset, TitanStats* stats) port::Mutex* db_mutex, BlobFileSet* blob_file_set,
TitanStats* stats)
: db_options_(db_options), : db_options_(db_options),
cf_options_(cf_options), cf_options_(cf_options),
blob_run_mode_(cf_options.blob_run_mode), blob_run_mode_(cf_options.blob_run_mode),
base_factory_(cf_options.table_factory), base_factory_(cf_options.table_factory),
blob_manager_(blob_manager), blob_manager_(blob_manager),
db_mutex_(db_mutex), db_mutex_(db_mutex),
vset_(vset), blob_file_set_(blob_file_set),
stats_(stats) {} stats_(stats) {}
const char* Name() const override { return "TitanTable"; } const char* Name() const override { return "TitanTable"; }
...@@ -67,7 +68,7 @@ class TitanTableFactory : public TableFactory { ...@@ -67,7 +68,7 @@ class TitanTableFactory : public TableFactory {
std::shared_ptr<TableFactory> base_factory_; std::shared_ptr<TableFactory> base_factory_;
std::shared_ptr<BlobFileManager> blob_manager_; std::shared_ptr<BlobFileManager> blob_manager_;
port::Mutex* db_mutex_; port::Mutex* db_mutex_;
VersionSet* vset_; BlobFileSet* blob_file_set_;
TitanStats* stats_; TitanStats* stats_;
}; };
......
src/titan_db_test.cc
View file @ 07aa0655
...@@ -102,7 +102,7 @@ class TitanDBTest : public testing::Test { ...@@ -102,7 +102,7 @@ class TitanDBTest : public testing::Test {
} }
Status LogAndApply(VersionEdit& edit) { Status LogAndApply(VersionEdit& edit) {
return db_impl_->vset_->LogAndApply(edit); return db_impl_->blob_file_set_->LogAndApply(edit);
} }
void Put(uint64_t k, std::map<std::string, std::string>* data = nullptr) { void Put(uint64_t k, std::map<std::string, std::string>* data = nullptr) {
...@@ -132,7 +132,7 @@ class TitanDBTest : public testing::Test { ...@@ -132,7 +132,7 @@ class TitanDBTest : public testing::Test {
cf_handle = db_->DefaultColumnFamily(); cf_handle = db_->DefaultColumnFamily();
} }
MutexLock l(&db_impl_->mutex_); MutexLock l(&db_impl_->mutex_);
return db_impl_->vset_->GetBlobStorage(cf_handle->GetID()); return db_impl_->blob_file_set_->GetBlobStorage(cf_handle->GetID());
} }
void VerifyDB(const std::map<std::string, std::string>& data, void VerifyDB(const std::map<std::string, std::string>& data,
...@@ -924,7 +924,7 @@ TEST_F(TitanDBTest, BackgroundErrorTrigger) { ...@@ -924,7 +924,7 @@ TEST_F(TitanDBTest, BackgroundErrorTrigger) {
} }
Flush(); Flush();
ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr)); ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
SyncPoint::GetInstance()->SetCallBack("VersionSet::LogAndApply", [&](void*) { SyncPoint::GetInstance()->SetCallBack("BlobFileSet::LogAndApply", [&](void*) {
mock_env->SetFilesystemActive(false, Status::IOError("Injected error")); mock_env->SetFilesystemActive(false, Status::IOError("Injected error"));
}); });
SyncPoint::GetInstance()->EnableProcessing(); SyncPoint::GetInstance()->EnableProcessing();
......
src/version_edit.h
View file @ 07aa0655
...@@ -43,7 +43,7 @@ class VersionEdit { ...@@ -43,7 +43,7 @@ class VersionEdit {
friend bool operator==(const VersionEdit& lhs, const VersionEdit& rhs); friend bool operator==(const VersionEdit& lhs, const VersionEdit& rhs);
private: private:
friend class VersionSet; friend class BlobFileSet;
friend class VersionTest; friend class VersionTest;
friend class EditCollector; friend class EditCollector;
......
src/version_test.cc
View file @ 07aa0655
#include "file/filename.h" #include "file/filename.h"
#include "test_util/testharness.h" #include "test_util/testharness.h"
#include "blob_file_set.h"
#include "blob_format.h" #include "blob_format.h"
#include "edit_collector.h" #include "edit_collector.h"
#include "testutil.h" #include "testutil.h"
#include "util.h" #include "util.h"
#include "version_edit.h" #include "version_edit.h"
#include "version_set.h"
namespace rocksdb { namespace rocksdb {
namespace titandb { namespace titandb {
...@@ -30,7 +30,7 @@ class VersionTest : public testing::Test { ...@@ -30,7 +30,7 @@ class VersionTest : public testing::Test {
TitanCFOptions cf_options_; TitanCFOptions cf_options_;
std::shared_ptr<BlobFileCache> file_cache_; std::shared_ptr<BlobFileCache> file_cache_;
std::map<uint32_t, std::shared_ptr<BlobStorage>> column_families_; std::map<uint32_t, std::shared_ptr<BlobStorage>> column_families_;
std::unique_ptr<VersionSet> vset_; std::unique_ptr<BlobFileSet> blob_file_set_;
port::Mutex mutex_; port::Mutex mutex_;
std::string dbname_; std::string dbname_;
Env* env_; Env* env_;
...@@ -50,8 +50,8 @@ class VersionTest : public testing::Test { ...@@ -50,8 +50,8 @@ class VersionTest : public testing::Test {
DeleteDir(env_, db_options_.dirname); DeleteDir(env_, db_options_.dirname);
env_->CreateDirIfMissing(db_options_.dirname); env_->CreateDirIfMissing(db_options_.dirname);
vset_.reset(new VersionSet(db_options_, nullptr)); blob_file_set_.reset(new BlobFileSet(db_options_, nullptr));
ASSERT_OK(vset_->Open({})); ASSERT_OK(blob_file_set_->Open({}));
column_families_.clear(); column_families_.clear();
// Sets up some column families. // Sets up some column families.
for (uint32_t id = 0; id < 10; id++) { for (uint32_t id = 0; id < 10; id++) {
...@@ -61,7 +61,7 @@ class VersionTest : public testing::Test { ...@@ -61,7 +61,7 @@ class VersionTest : public testing::Test {
column_families_.emplace(id, storage); column_families_.emplace(id, storage);
storage.reset( storage.reset(
new BlobStorage(db_options_, cf_options_, id, file_cache_, nullptr)); new BlobStorage(db_options_, cf_options_, id, file_cache_, nullptr));
vset_->column_families_.emplace(id, storage); blob_file_set_->column_families_.emplace(id, storage);
} }
} }
...@@ -85,9 +85,9 @@ class VersionTest : public testing::Test { ...@@ -85,9 +85,9 @@ class VersionTest : public testing::Test {
for (auto& edit : edits) { for (auto& edit : edits) {
ASSERT_OK(collector.AddEdit(edit)); ASSERT_OK(collector.AddEdit(edit));
} }
ASSERT_OK(collector.Seal(*vset_.get())); ASSERT_OK(collector.Seal(*blob_file_set_.get()));
ASSERT_OK(collector.Apply(*vset_.get())); ASSERT_OK(collector.Apply(*blob_file_set_.get()));
for (auto& it : vset_->column_families_) { for (auto& it : blob_file_set_->column_families_) {
auto& storage = column_families_[it.first]; auto& storage = column_families_[it.first];
// ignore obsolete file // ignore obsolete file
auto size = 0; auto size = 0;
...@@ -106,7 +106,7 @@ class VersionTest : public testing::Test { ...@@ -106,7 +106,7 @@ class VersionTest : public testing::Test {
} }
void CheckColumnFamiliesSize(uint64_t size) { void CheckColumnFamiliesSize(uint64_t size) {
ASSERT_EQ(vset_->column_families_.size(), size); ASSERT_EQ(blob_file_set_->column_families_.size(), size);
} }
void LegacyEncode(const VersionEdit& edit, std::string* dst) { void LegacyEncode(const VersionEdit& edit, std::string* dst) {
...@@ -164,8 +164,8 @@ TEST_F(VersionTest, InvalidEdit) { ...@@ -164,8 +164,8 @@ TEST_F(VersionTest, InvalidEdit) {
auto add1_0_4 = AddBlobFilesEdit(1, 0, 4); auto add1_0_4 = AddBlobFilesEdit(1, 0, 4);
EditCollector collector; EditCollector collector;
ASSERT_OK(collector.AddEdit(add1_0_4)); ASSERT_OK(collector.AddEdit(add1_0_4));
ASSERT_OK(collector.Seal(*vset_.get())); ASSERT_OK(collector.Seal(*blob_file_set_.get()));
ASSERT_OK(collector.Apply(*vset_.get())); ASSERT_OK(collector.Apply(*blob_file_set_.get()));
} }
// delete nonexistent blobs // delete nonexistent blobs
...@@ -173,8 +173,8 @@ TEST_F(VersionTest, InvalidEdit) { ...@@ -173,8 +173,8 @@ TEST_F(VersionTest, InvalidEdit) {
auto del1_4_6 = DeleteBlobFilesEdit(1, 4, 6); auto del1_4_6 = DeleteBlobFilesEdit(1, 4, 6);
EditCollector collector; EditCollector collector;
ASSERT_OK(collector.AddEdit(del1_4_6)); ASSERT_OK(collector.AddEdit(del1_4_6));
ASSERT_NOK(collector.Seal(*vset_.get())); ASSERT_NOK(collector.Seal(*blob_file_set_.get()));
ASSERT_NOK(collector.Apply(*vset_.get())); ASSERT_NOK(collector.Apply(*blob_file_set_.get()));
} }
// add already existing blobs // add already existing blobs
...@@ -182,8 +182,8 @@ TEST_F(VersionTest, InvalidEdit) { ...@@ -182,8 +182,8 @@ TEST_F(VersionTest, InvalidEdit) {
auto add1_1_3 = AddBlobFilesEdit(1, 1, 3); auto add1_1_3 = AddBlobFilesEdit(1, 1, 3);
EditCollector collector; EditCollector collector;
ASSERT_OK(collector.AddEdit(add1_1_3)); ASSERT_OK(collector.AddEdit(add1_1_3));
ASSERT_NOK(collector.Seal(*vset_.get())); ASSERT_NOK(collector.Seal(*blob_file_set_.get()));
ASSERT_NOK(collector.Apply(*vset_.get())); ASSERT_NOK(collector.Apply(*blob_file_set_.get()));
} }
// add same blobs // add same blobs
...@@ -193,8 +193,8 @@ TEST_F(VersionTest, InvalidEdit) { ...@@ -193,8 +193,8 @@ TEST_F(VersionTest, InvalidEdit) {
EditCollector collector; EditCollector collector;
ASSERT_OK(collector.AddEdit(add1_4_5_1)); ASSERT_OK(collector.AddEdit(add1_4_5_1));
ASSERT_NOK(collector.AddEdit(add1_4_5_2)); ASSERT_NOK(collector.AddEdit(add1_4_5_2));
ASSERT_NOK(collector.Seal(*vset_.get())); ASSERT_NOK(collector.Seal(*blob_file_set_.get()));
ASSERT_NOK(collector.Apply(*vset_.get())); ASSERT_NOK(collector.Apply(*blob_file_set_.get()));
} }
// delete same blobs // delete same blobs
...@@ -204,8 +204,8 @@ TEST_F(VersionTest, InvalidEdit) { ...@@ -204,8 +204,8 @@ TEST_F(VersionTest, InvalidEdit) {
EditCollector collector; EditCollector collector;
ASSERT_OK(collector.AddEdit(del1_3_4_1)); ASSERT_OK(collector.AddEdit(del1_3_4_1));
ASSERT_NOK(collector.AddEdit(del1_3_4_2)); ASSERT_NOK(collector.AddEdit(del1_3_4_2));
ASSERT_NOK(collector.Seal(*vset_.get())); ASSERT_NOK(collector.Seal(*blob_file_set_.get()));
ASSERT_NOK(collector.Apply(*vset_.get())); ASSERT_NOK(collector.Apply(*blob_file_set_.get()));
} }
} }
...@@ -252,35 +252,35 @@ TEST_F(VersionTest, ObsoleteFiles) { ...@@ -252,35 +252,35 @@ TEST_F(VersionTest, ObsoleteFiles) {
std::map<uint32_t, TitanCFOptions> m; std::map<uint32_t, TitanCFOptions> m;
m.insert({1, TitanCFOptions()}); m.insert({1, TitanCFOptions()});
m.insert({2, TitanCFOptions()}); m.insert({2, TitanCFOptions()});
vset_->AddColumnFamilies(m); blob_file_set_->AddColumnFamilies(m);
{ {
auto add1_1_5 = AddBlobFilesEdit(1, 1, 5); auto add1_1_5 = AddBlobFilesEdit(1, 1, 5);
MutexLock l(&mutex_); MutexLock l(&mutex_);
vset_->LogAndApply(add1_1_5); blob_file_set_->LogAndApply(add1_1_5);
} }
std::vector<std::string> of; std::vector<std::string> of;
vset_->GetObsoleteFiles(&of, kMaxSequenceNumber); blob_file_set_->GetObsoleteFiles(&of, kMaxSequenceNumber);
ASSERT_EQ(of.size(), 0); ASSERT_EQ(of.size(), 0);
{ {
auto del1_4_5 = DeleteBlobFilesEdit(1, 4, 5); auto del1_4_5 = DeleteBlobFilesEdit(1, 4, 5);
MutexLock l(&mutex_); MutexLock l(&mutex_);
vset_->LogAndApply(del1_4_5); blob_file_set_->LogAndApply(del1_4_5);
} }
vset_->GetObsoleteFiles(&of, kMaxSequenceNumber); blob_file_set_->GetObsoleteFiles(&of, kMaxSequenceNumber);
ASSERT_EQ(of.size(), 1); ASSERT_EQ(of.size(), 1);
std::vector<uint32_t> cfs = {1}; std::vector<uint32_t> cfs = {1};
ASSERT_OK(vset_->DropColumnFamilies(cfs, 0)); ASSERT_OK(blob_file_set_->DropColumnFamilies(cfs, 0));
vset_->GetObsoleteFiles(&of, kMaxSequenceNumber); blob_file_set_->GetObsoleteFiles(&of, kMaxSequenceNumber);
ASSERT_EQ(of.size(), 1); ASSERT_EQ(of.size(), 1);
CheckColumnFamiliesSize(10); CheckColumnFamiliesSize(10);
ASSERT_OK(vset_->MaybeDestroyColumnFamily(1)); ASSERT_OK(blob_file_set_->MaybeDestroyColumnFamily(1));
vset_->GetObsoleteFiles(&of, kMaxSequenceNumber); blob_file_set_->GetObsoleteFiles(&of, kMaxSequenceNumber);
ASSERT_EQ(of.size(), 4); ASSERT_EQ(of.size(), 4);
CheckColumnFamiliesSize(9); CheckColumnFamiliesSize(9);
ASSERT_OK(vset_->MaybeDestroyColumnFamily(2)); ASSERT_OK(blob_file_set_->MaybeDestroyColumnFamily(2));
CheckColumnFamiliesSize(8); CheckColumnFamiliesSize(8);
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment