fluidB
/
fluidb
1
0
Fork 0
Code Pull Requests Activity

remove column family related stuff

Browse Source
main
Ziyang Hu 2 years ago
parent 57937b6f56
commit ece15f0940
6 changed files with 871 additions and 1042 deletions
Show Stats Download Patch File Download Diff File

96
cozorocks/bridge/cozorocks.cc
Unescape Escape View File

@ -19,50 +19,52 @@ BridgeStatus IteratorBridge::status() const {
return s;
}
unique_ptr<vector<PinnableSlice>> TransactionBridge::multiget_txn(
const ColumnFamilyHandle &cf,
rust::Slice<const rust::Slice<const uint8_t>> keys,
rust::Slice<BridgeStatus> statuses) const {
auto cpp_keys = vector<Slice>();
cpp_keys.reserve(keys.size());
for (auto key: keys) {
auto cpp_key = convert_slice(key);
cpp_keys.push_back(cpp_key);
}
auto ret = make_unique<vector<PinnableSlice>>(keys.size());
auto cpp_statuses = vector<Status>(keys.size());
inner->MultiGet(*r_ops,
const_cast<ColumnFamilyHandle *>(&cf),
keys.size(),
cpp_keys.data(),
ret->data(),
cpp_statuses.data());
for (size_t i = 0; i < cpp_statuses.size(); ++i) {
write_status(std::move(cpp_statuses[i]), statuses[i]);
}
return ret;
}
unique_ptr<vector<PinnableSlice>> TransactionBridge::multiget_raw(
const ColumnFamilyHandle &cf,
rust::Slice<const rust::Slice<const uint8_t>> keys,
rust::Slice<BridgeStatus> statuses) const {
auto cpp_keys = vector<Slice>();
cpp_keys.reserve(keys.size());
for (auto key: keys) {
auto cpp_key = convert_slice(key);
cpp_keys.push_back(cpp_key);
}
auto ret = make_unique<vector<PinnableSlice>>(keys.size());
auto cpp_statuses = vector<Status>(keys.size());
raw_db->MultiGet(*r_ops,
const_cast<ColumnFamilyHandle *>(&cf),
keys.size(),
cpp_keys.data(),
ret->data(),
cpp_statuses.data());
for (size_t i = 0; i < cpp_statuses.size(); ++i) {
write_status(std::move(cpp_statuses[i]), statuses[i]);
}
return ret;
}
//unique_ptr<vector<PinnableSlice>> TransactionBridge::multiget_txn(
//// const ColumnFamilyHandle &cf,
// rust::Slice<const rust::Slice<const uint8_t>> keys,
// rust::Slice<BridgeStatus> statuses) const {
// auto cpp_keys = vector<Slice>();
// cpp_keys.reserve(keys.size());
// for (auto key: keys) {
// auto cpp_key = convert_slice(key);
// cpp_keys.push_back(cpp_key);
// }
// auto ret = make_unique<vector<PinnableSlice>>(keys.size());
// auto cpp_statuses = vector<Status>(keys.size());
// inner->MultiGet(*r_ops,
// raw_db->DefaultColumnFamily(),
//// const_cast<ColumnFamilyHandle *>(&cf),
// keys.size(),
// cpp_keys.data(),
// ret->data(),
// cpp_statuses.data());
// for (size_t i = 0; i < cpp_statuses.size(); ++i) {
// write_status(std::move(cpp_statuses[i]), statuses[i]);
// }
// return ret;
//}
//
//unique_ptr<vector<PinnableSlice>> TransactionBridge::multiget_raw(
//// const ColumnFamilyHandle &cf,
// rust::Slice<const rust::Slice<const uint8_t>> keys,
// rust::Slice<BridgeStatus> statuses) const {
// auto cpp_keys = vector<Slice>();
// cpp_keys.reserve(keys.size());
// for (auto key: keys) {
// auto cpp_key = convert_slice(key);
// cpp_keys.push_back(cpp_key);
// }
// auto ret = make_unique<vector<PinnableSlice>>(keys.size());
// auto cpp_statuses = vector<Status>(keys.size());
// raw_db->MultiGet(*r_ops,
// raw_db->DefaultColumnFamily(),
//// const_cast<ColumnFamilyHandle *>(&cf),
// keys.size(),
// cpp_keys.data(),
// ret->data(),
// cpp_statuses.data());
// for (size_t i = 0; i < cpp_statuses.size(); ++i) {
// write_status(std::move(cpp_statuses[i]), statuses[i]);
// }
// return ret;
//}

249
cozorocks/bridge/cozorocks.h
Unescape Escape View File

@ -215,13 +215,12 @@ struct IteratorBridge {
inner->SeekForPrev(k);
}
inline std::shared_ptr<Slice> key_raw() const {
// std::cout << "c++ get " << inner->key().size() << std::endl;
return std::make_shared<Slice>(inner->key());
inline std::unique_ptr<Slice> key_raw() const {
return std::make_unique<Slice>(inner->key());
}
inline std::shared_ptr<Slice> value_raw() const {
return std::make_shared<Slice>(inner->value());
inline std::unique_ptr<Slice> value_raw() const {
return std::make_unique<Slice>(inner->value());
}
inline void refresh(BridgeStatus &status) const {
@ -246,6 +245,14 @@ inline unique_ptr<OptimisticTransactionOptions> new_optimistic_transaction_optio
return ret;
}
inline void reset_pinnable_slice(PinnableSlice &slice) {
slice.Reset();
}
unique_ptr<PinnableSlice> new_pinnable_slice() {
return make_unique<PinnableSlice>();
}
struct TransactionBridge {
DB *raw_db;
unique_ptr<Transaction> inner;
@ -282,193 +289,123 @@ struct TransactionBridge {
write_status(inner->PopSavePoint(), status);
}
unique_ptr<vector<PinnableSlice>> multiget_txn(
const ColumnFamilyHandle &cf,
rust::Slice<const rust::Slice<const uint8_t>> keys,
rust::Slice<BridgeStatus> statuses) const;
unique_ptr<vector<PinnableSlice>> multiget_raw(
const ColumnFamilyHandle &cf,
rust::Slice<const rust::Slice<const uint8_t>> keys,
rust::Slice<BridgeStatus> statuses) const;
inline shared_ptr<PinnableSlice> get_txn(
const ColumnFamilyHandle &cf,
inline void get_txn(
rust::Slice<const uint8_t> key,
PinnableSlice &pinnable_val,
BridgeStatus &status
) const {
auto pinnable_val = std::make_shared<PinnableSlice>();
write_status(
inner->Get(*r_ops,
const_cast<ColumnFamilyHandle *>(&cf),
convert_slice(key),
&*pinnable_val),
status
);
return pinnable_val;
write_status(inner->Get(*r_ops, convert_slice(key), &pinnable_val), status);
}
inline shared_ptr<PinnableSlice> get_for_update_txn(
const ColumnFamilyHandle &cf,
inline void get_for_update_txn(
rust::Slice<const uint8_t> key,
PinnableSlice &pinnable_val,
BridgeStatus &status
) const {
auto pinnable_val = std::make_shared<PinnableSlice>();
write_status(
inner->GetForUpdate(*r_ops,
const_cast<ColumnFamilyHandle *>(&cf),
raw_db->DefaultColumnFamily(),
convert_slice(key),
&*pinnable_val),
&pinnable_val),
status
);
return pinnable_val;
}
inline std::shared_ptr<PinnableSlice> get_raw(
const ColumnFamilyHandle &cf,
inline void get_raw(
rust::Slice<const uint8_t> key,
PinnableSlice &pinnable_val,
BridgeStatus &status
) const {
auto pinnable_val = std::make_shared<PinnableSlice>();
write_status(
raw_db->Get(*r_ops,
const_cast<ColumnFamilyHandle *>(&cf),
raw_db->DefaultColumnFamily(),
convert_slice(key),
&*pinnable_val),
&pinnable_val),
status
);
return pinnable_val;
}
inline void put_txn(
const ColumnFamilyHandle &cf,
rust::Slice<const uint8_t> key,
rust::Slice<const uint8_t> val,
BridgeStatus &status
) const {
write_status(
inner->Put(const_cast<ColumnFamilyHandle *>(&cf),
convert_slice(key),
convert_slice(val)),
status
);
write_status(inner->Put(convert_slice(key), convert_slice(val)), status);
}
inline void put_raw(
const ColumnFamilyHandle &cf,
rust::Slice<const uint8_t> key,
rust::Slice<const uint8_t> val,
BridgeStatus &status
) const {
auto k = convert_slice(key);
auto v = convert_slice(val);
// std::cout << "c++ put " << key.size() << " " << k.size() << std::endl;
write_status(
raw_db->Put(
*raw_w_ops,
const_cast<ColumnFamilyHandle *>(&cf),
k,
v),
status
);
write_status(raw_db->Put(*raw_w_ops, k, v), status);
}
inline void del_txn(
const ColumnFamilyHandle &cf,
rust::Slice<const uint8_t> key,
BridgeStatus &status
) const {
write_status(
inner->Delete(const_cast<ColumnFamilyHandle *>(&cf),
convert_slice(key)),
status
);
write_status(inner->Delete(convert_slice(key)), status);
}
inline void del_raw(
const ColumnFamilyHandle &cf,
rust::Slice<const uint8_t> key,
BridgeStatus &status
) const {
write_status(
raw_db->Delete(
*raw_w_ops,
const_cast<ColumnFamilyHandle *>(&cf),
convert_slice(key)),
status
);
write_status(raw_db->Delete(*raw_w_ops, convert_slice(key)), status);
}
inline void del_range_raw(
const ColumnFamilyHandle &cf,
rust::Slice<const uint8_t> start_key,
rust::Slice<const uint8_t> end_key,
BridgeStatus &status
) const {
write_status(
raw_db->GetRootDB()->DeleteRange(
*raw_w_ops, const_cast<ColumnFamilyHandle *>(&cf),
*raw_w_ops,
raw_db->DefaultColumnFamily(),
convert_slice(start_key), convert_slice(end_key)),
status);
}
inline void flush_raw(const ColumnFamilyHandle &cf, const FlushOptions &options, BridgeStatus &status) const {
write_status(raw_db->Flush(options, const_cast<ColumnFamilyHandle *>(&cf)), status);
inline void flush_raw(const FlushOptions &options, BridgeStatus &status) const {
write_status(raw_db->Flush(options), status);
}
inline void compact_all_raw(const ColumnFamilyHandle &cf, BridgeStatus &status) const {
inline void compact_all_raw(BridgeStatus &status) const {
auto options = CompactRangeOptions();
options.change_level = true;
options.target_level = 0;
options.exclusive_manual_compaction = false;
write_status(raw_db->CompactRange(options, const_cast<ColumnFamilyHandle *>(&cf), nullptr, nullptr), status);
write_status(raw_db->CompactRange(options,
raw_db->DefaultColumnFamily(),
nullptr, nullptr), status);
}
inline std::unique_ptr<IteratorBridge> iterator_txn(
const ColumnFamilyHandle &cf) const {
inline std::unique_ptr<IteratorBridge> iterator_txn() const {
return std::make_unique<IteratorBridge>(
inner->GetIterator(*r_ops, const_cast<ColumnFamilyHandle *>(&cf)));
inner->GetIterator(*r_ops));
}
inline std::unique_ptr<IteratorBridge> iterator_raw(
const ColumnFamilyHandle &cf) const {
inline std::unique_ptr<IteratorBridge> iterator_raw() const {
return std::make_unique<IteratorBridge>(
raw_db->NewIterator(*raw_r_ops, const_cast<ColumnFamilyHandle *>(&cf)));
raw_db->NewIterator(*raw_r_ops));
}
};
inline tuple<vector<string>, vector<ColumnFamilyDescriptor>>
get_cf_data(const Options &options,
const string &path) {
auto cf_names = std::vector<std::string>();
DB::ListColumnFamilies(options, path, &cf_names);
if (cf_names.empty()) {
cf_names.push_back(kDefaultColumnFamilyName);
}
std::vector<ColumnFamilyDescriptor> column_families;
for (auto el: cf_names) {
column_families.push_back(ColumnFamilyDescriptor(
el, options));
}
return std::make_tuple(cf_names, column_families);
}
struct TDBBridge {
mutable unique_ptr<StackableDB> db;
mutable TransactionDB *tdb;
mutable OptimisticTransactionDB *odb;
mutable unordered_map<string, shared_ptr<ColumnFamilyHandle>> handles;
mutable Lock handle_lock;
bool is_odb;
TDBBridge(StackableDB *db_,
TransactionDB *tdb_,
OptimisticTransactionDB *odb_,
unordered_map<string, shared_ptr<ColumnFamilyHandle>> &&handles_) :
db(db_), tdb(tdb_), odb(odb_), handles(handles_), handle_lock() {
OptimisticTransactionDB *odb_) :
db(db_), tdb(tdb_), odb(odb_) {
is_odb = (tdb_ == nullptr);
}
@ -513,64 +450,6 @@ struct TDBBridge {
ret->inner = unique_ptr<Transaction>(txn);
return ret;
}
inline shared_ptr<ColumnFamilyHandle> get_cf_handle_raw(const string &name) const {
ReadLock r_lock(handle_lock);
try {
return handles.at(name);
} catch (const std::out_of_range &) {
return shared_ptr<ColumnFamilyHandle>(nullptr);
}
}
inline shared_ptr<ColumnFamilyHandle> get_default_cf_handle_raw() const {
return handles.at("default");
}
inline shared_ptr<ColumnFamilyHandle>
create_column_family_raw(const Options &options, const string &name, BridgeStatus &status) const {
{
ReadLock r_lock(handle_lock);
if (handles.find(name) != handles.end()) {
write_status_impl(status, StatusCode::kMaxCode, StatusSubCode::kMaxSubCode,
StatusSeverity::kSoftError,
2);
return shared_ptr<ColumnFamilyHandle>(nullptr);
}
}
{
WriteLock w_lock(handle_lock);
ColumnFamilyHandle *handle;
auto s = db->CreateColumnFamily(options, name, &handle);
write_status(std::move(s), status);
auto ret = shared_ptr<ColumnFamilyHandle>(handle);
handles[name] = ret;
return ret;
}
}
inline void drop_column_family_raw(const string &name, BridgeStatus &status) const {
WriteLock w_lock(handle_lock);
auto cf_it = handles.find(name);
if (cf_it != handles.end()) {
auto s = db->DropColumnFamily(cf_it->second.get());
handles.erase(cf_it);
write_status(std::move(s), status);
} else {
write_status_impl(status, StatusCode::kMaxCode, StatusSubCode::kMaxSubCode, StatusSeverity::kSoftError,
3);
}
// When should we call DestroyColumnFamilyHandle?
}
inline unique_ptr<vector<string>> get_column_family_names_raw() const {
ReadLock r_lock(handle_lock);
auto ret = make_unique<vector<string >>();
for (auto entry: handles) {
ret->push_back(entry.first);
}
return ret;
}
};
inline unique_ptr<TransactionDBOptions> new_tdb_options() {
@ -593,64 +472,34 @@ void set_allow_write_stall(FlushOptions &options, bool v) {
options.allow_write_stall = v;
}
inline unique_ptr<TDBBridge>
inline shared_ptr<TDBBridge>
open_tdb_raw(const Options &options,
const TransactionDBOptions &txn_db_options,
const string &path,
BridgeStatus &status) {
auto cf_info = get_cf_data(options, path);
auto cf_names = std::get<0>(cf_info);
auto column_families = std::get<1>(cf_info);
std::vector<ColumnFamilyHandle *> handles;
TransactionDB *txn_db = nullptr;
Status s = TransactionDB::Open(options, txn_db_options, path,
column_families, &handles,
&txn_db);
Status s = TransactionDB::Open(options, txn_db_options, path, &txn_db);
auto ok = s.ok();
write_status(std::move(s), status);
unordered_map<string, shared_ptr<ColumnFamilyHandle>> handle_map;
if (ok) {
assert(handles.size() == cf_names.size());
for (size_t i = 0; i < handles.size(); ++i) {
handle_map[cf_names[i]] = shared_ptr<ColumnFamilyHandle>(handles[i]);
}
}
return make_unique<TDBBridge>(txn_db, txn_db, nullptr, std::move(handle_map));
return make_shared<TDBBridge>(txn_db, txn_db, nullptr);
}
inline unique_ptr<TDBBridge>
open_odb_raw(const Options &options,
const OptimisticTransactionDBOptions &txn_db_options,
const string &path,
BridgeStatus &status) {
auto cf_info = get_cf_data(options, path);
auto cf_names = std::get<0>(cf_info);
auto column_families = std::get<1>(cf_info);
std::vector<ColumnFamilyHandle *> handles;
inline shared_ptr<TDBBridge>
open_odb_raw(const Options &options, const string &path, BridgeStatus &status) {
OptimisticTransactionDB *txn_db = nullptr;
Status s = OptimisticTransactionDB::Open(options, txn_db_options, path,
column_families, &handles,
Status s = OptimisticTransactionDB::Open(options,
path,
&txn_db);
auto ok = s.ok();
write_status(std::move(s), status);
unordered_map<string, shared_ptr<ColumnFamilyHandle>> handle_map;
if (ok) {
assert(handles.size() == cf_names.size());
for (size_t i = 0; i < handles.size(); ++i) {
handle_map[cf_names[i]] = shared_ptr<ColumnFamilyHandle>(handles[i]);
}
}
return make_unique<TDBBridge>(txn_db, nullptr, txn_db, std::move(handle_map));
return make_shared<TDBBridge>(txn_db, nullptr, txn_db);
}

60
cozorocks/src/bridge.rs
Unescape Escape View File

@ -76,6 +76,8 @@ mod ffi {
type Slice;
type PinnableSlice;
fn reset_pinnable_slice(o: Pin<&mut PinnableSlice>);
fn new_pinnable_slice() -> UniquePtr<PinnableSlice>;
fn convert_slice_back(s: &Slice) -> &[u8];
fn convert_pinnable_slice_back(s: &PinnableSlice) -> &[u8];
@ -131,8 +133,8 @@ mod ffi {
fn is_valid(self: &IteratorBridge) -> bool;
fn do_seek(self: &IteratorBridge, key: &[u8]);
fn do_seek_for_prev(self: &IteratorBridge, key: &[u8]);
fn key_raw(self: &IteratorBridge) -> SharedPtr<Slice>;
fn value_raw(self: &IteratorBridge) -> SharedPtr<Slice>;
fn key_raw(self: &IteratorBridge) -> UniquePtr<Slice>;
fn value_raw(self: &IteratorBridge) -> UniquePtr<Slice>;
fn status(self: &IteratorBridge) -> BridgeStatus;
fn refresh(self: &IteratorBridge, status: &mut BridgeStatus);
@ -145,78 +147,58 @@ mod ffi {
fn pop_savepoint(self: &TransactionBridge, status: &mut BridgeStatus);
fn get_txn(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
slice: Pin<&mut PinnableSlice>,
status: &mut BridgeStatus,
) -> SharedPtr<PinnableSlice>;
);
fn get_for_update_txn(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
slice: Pin<&mut PinnableSlice>,
status: &mut BridgeStatus,
) -> SharedPtr<PinnableSlice>;
);
fn get_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
slice: Pin<&mut PinnableSlice>,
status: &mut BridgeStatus,
) -> SharedPtr<PinnableSlice>;
);
fn put_txn(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
val: &[u8],
status: &mut BridgeStatus,
);
fn put_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
val: &[u8],
status: &mut BridgeStatus,
);
fn del_txn(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
status: &mut BridgeStatus,
);
fn del_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
key: &[u8],
status: &mut BridgeStatus,
);
fn del_range_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
start_key: &[u8],
end_key: &[u8],
status: &mut BridgeStatus,
);
fn flush_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
options: &FlushOptions,
status: &mut BridgeStatus,
);
fn compact_all_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
status: &mut BridgeStatus,
);
fn iterator_txn(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
) -> UniquePtr<IteratorBridge>;
fn iterator_raw(
self: &TransactionBridge,
cf: &ColumnFamilyHandle,
) -> UniquePtr<IteratorBridge>;
// fn multiget_txn(self: &TransactionBridge, cf: &ColumnFamilyHandle,
// keys: &[&[u8]], statuses: &mut [BridgeStatus]) -> UniquePtr<CxxVector<PinnableSlice>>;
// fn multiget_raw(self: &TransactionBridge, cf: &ColumnFamilyHandle,
// keys: &[&[u8]], statuses: &mut [BridgeStatus]) -> UniquePtr<CxxVector<PinnableSlice>>;
fn compact_all_raw(self: &TransactionBridge, status: &mut BridgeStatus);
fn iterator_txn(self: &TransactionBridge) -> UniquePtr<IteratorBridge>;
fn iterator_raw(self: &TransactionBridge) -> UniquePtr<IteratorBridge>;
pub type ColumnFamilyHandle;
@ -237,28 +219,18 @@ mod ffi {
raw_r_ops: UniquePtr<ReadOptions>,
txn_options: UniquePtr<OptimisticTransactionOptions>,
) -> UniquePtr<TransactionBridge>;
fn get_cf_handle_raw(self: &TDBBridge, name: &CxxString) -> SharedPtr<ColumnFamilyHandle>;
fn get_default_cf_handle_raw(self: &TDBBridge) -> SharedPtr<ColumnFamilyHandle>;
fn create_column_family_raw(
self: &TDBBridge,
options: &Options,
name: &CxxString,
status: &mut BridgeStatus,
) -> SharedPtr<ColumnFamilyHandle>;
fn drop_column_family_raw(self: &TDBBridge, name: &CxxString, status: &mut BridgeStatus);
fn get_column_family_names_raw(self: &TDBBridge) -> UniquePtr<CxxVector<CxxString>>;
fn open_tdb_raw(
options: &Options,
txn_options: &TransactionDBOptions,
path: &CxxString,
status: &mut BridgeStatus,
) -> UniquePtr<TDBBridge>;
) -> SharedPtr<TDBBridge>;
fn open_odb_raw(
options: &Options,
txn_options: &OptimisticTransactionDBOptions,
// txn_options: &OptimisticTransactionDBOptions,
path: &CxxString,
status: &mut BridgeStatus,
) -> UniquePtr<TDBBridge>;
) -> SharedPtr<TDBBridge>;
}
}

1481
cozorocks/src/lib.rs
View File

File diff suppressed because it is too large Load Diff

26
src/db.rs
Unescape Escape View File

@ -1,10 +1,16 @@
pub mod cnf_transform;
pub mod ddl;
pub mod engine;
pub mod env;
pub mod eval;
pub mod iterator;
pub mod mutation;
pub mod plan;
pub mod query;
pub mod table;
// pub mod cnf_transform;
// pub mod ddl;
// pub mod engine;
// pub mod env;
// pub mod eval;
// pub mod iterator;
// pub mod mutation;
// pub mod plan;
// pub mod query;
// pub mod table;
pub struct DBInstance {
pub(crate) main_db: (),
session_handles: (),
}

1
src/lib.rs
Unescape Escape View File

@ -5,3 +5,4 @@
pub(crate) mod data;
pub(crate) mod logger;
pub(crate) mod parser;
pub(crate) mod db;

Write Preview
Loading…
Cancel
Save