mirrors/scylladb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
28ef8feffa2bfc62083fb96fe97a883e418e15ce
scylladb/database.cc
Avi Kivity f4706c7050 Merge "initial support to leveled compaction" from Raphael 
"This patchset introduces leveled compaction to Scylla.
We don't handle all corner cases yet, but we already have the strategy
and compaction working as expected. Test cases were written and I also
tested the stability with a load of cassandra-stress.

Leveled compaction may output more than one sstable because there is
a limit on the size of sstables. 160M by default.
Related to handling of partial compaction, it's still something to be
worked on.

Anyway, it will not be a big problem. Why? Suppose that a leveled
compaction will generate 2 sstables, and scylla is interrupted after
the first sstable is completely written but before the second one is
completely written. The next boot will delete the second sstable,
because it was partially written, but will not do anything with the
first one as it was completely written.
As a result, we will have two sstables with redundant data."
2015-10-19 16:17:45 +03:00

1835 lines
70 KiB
C++
Raw Blame History

/*
* Copyright (C) 2014 Cloudius Systems, Ltd.
*/
/*
* This file is part of Scylla.
*
* Scylla is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* Scylla is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with Scylla. If not, see <http://www.gnu.org/licenses/>.
*/
#include "log.hh"
#include "database.hh"
#include "unimplemented.hh"
#include "core/future-util.hh"
#include "db/system_keyspace.hh"
#include "db/consistency_level.hh"
#include "db/serializer.hh"
#include "db/commitlog/commitlog.hh"
#include "db/config.hh"
#include "to_string.hh"
#include "query-result-writer.hh"
#include "nway_merger.hh"
#include "cql3/column_identifier.hh"
#include "core/seastar.hh"
#include <seastar/core/sleep.hh>
#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/string/split.hpp>
#include "sstables/sstables.hh"
#include "sstables/compaction.hh"
#include <boost/range/adaptor/transformed.hpp>
#include <boost/range/adaptor/map.hpp>
#include "locator/simple_snitch.hh"
#include <boost/algorithm/cxx11/all_of.hpp>
#include <boost/function_output_iterator.hpp>
#include <boost/range/algorithm/heap_algorithm.hpp>
#include <boost/range/algorithm/find.hpp>
#include "frozen_mutation.hh"
#include "mutation_partition_applier.hh"
#include "core/do_with.hh"
#include "service/migration_manager.hh"
#include "service/storage_service.hh"
#include "mutation_query.hh"
#include "sstable_mutation_readers.hh"
#include <core/fstream.hh>
#include "utils/latency.hh"
#include "utils/flush_queue.hh"
using namespace std::chrono_literals;
logging::logger dblog("database");
// Do this to avoid having to include or forward-declare template in our header.
class column_family::memtable_flush_queue : public utils::flush_queue<db::replay_position> {};
column_family::column_family(schema_ptr schema, config config, db::commitlog& cl, compaction_manager& compaction_manager)
: _schema(std::move(schema))
, _config(std::move(config))
, _memtables(make_lw_shared(memtable_list{}))
, _sstables(make_lw_shared<sstable_list>())
, _cache(_schema, sstables_as_mutation_source(), global_cache_tracker())
, _commitlog(&cl)
, _compaction_manager(compaction_manager)
, _flush_queue(std::make_unique<memtable_flush_queue>())
{
add_memtable();
if (!_config.enable_disk_writes) {
dblog.warn("Writes disabled, column family no durable.");
}
}
column_family::column_family(schema_ptr schema, config config, no_commitlog cl, compaction_manager& compaction_manager)
: _schema(std::move(schema))
, _config(std::move(config))
, _memtables(make_lw_shared(memtable_list{}))
, _sstables(make_lw_shared<sstable_list>())
, _cache(_schema, sstables_as_mutation_source(), global_cache_tracker())
, _commitlog(nullptr)
, _compaction_manager(compaction_manager)
, _flush_queue(std::make_unique<memtable_flush_queue>())
{
add_memtable();
if (!_config.enable_disk_writes) {
dblog.warn("Writes disabled, column family no durable.");
}
}
partition_presence_checker
column_family::make_partition_presence_checker(lw_shared_ptr<sstable_list> old_sstables) {
return [this, old_sstables = std::move(old_sstables)] (const partition_key& key) {
for (auto&& s : *old_sstables) {
if (s.second->filter_has_key(*_schema, key)) {
return partition_presence_checker_result::maybe_exists;
}
}
return partition_presence_checker_result::definitely_doesnt_exist;
};
}
mutation_source
column_family::sstables_as_mutation_source() {
return [this] (const query::partition_range& r) {
return make_sstable_reader(r);
};
}
// define in .cc, since sstable is forward-declared in .hh
column_family::~column_family() {
}
logalloc::occupancy_stats column_family::occupancy() const {
logalloc::occupancy_stats res;
for (auto m : *_memtables.get()) {
res += m->region().occupancy();
}
return res;
}
static
bool belongs_to_current_shard(const mutation& m) {
return dht::shard_of(m.token()) == engine().cpu_id();
}
class range_sstable_reader final : public mutation_reader::impl {
const query::partition_range& _pr;
lw_shared_ptr<sstable_list> _sstables;
mutation_reader _reader;
public:
range_sstable_reader(schema_ptr s, lw_shared_ptr<sstable_list> sstables, const query::partition_range& pr)
: _pr(pr)
, _sstables(std::move(sstables))
{
std::vector<mutation_reader> readers;
for (const lw_shared_ptr<sstables::sstable>& sst : *_sstables | boost::adaptors::map_values) {
// FIXME: make sstable::read_range_rows() return ::mutation_reader so that we can drop this wrapper.
mutation_reader reader = make_mutation_reader<sstable_range_wrapping_reader>(sst, s, pr);
if (sst->is_shared()) {
reader = make_filtering_reader(std::move(reader), belongs_to_current_shard);
}
readers.emplace_back(std::move(reader));
}
_reader = make_combined_reader(std::move(readers));
}
range_sstable_reader(range_sstable_reader&&) = delete; // reader takes reference to member fields
virtual future<mutation_opt> operator()() override {
return _reader();
}
};
class single_key_sstable_reader final : public mutation_reader::impl {
schema_ptr _schema;
sstables::key _key;
mutation_opt _m;
bool _done = false;
lw_shared_ptr<sstable_list> _sstables;
public:
single_key_sstable_reader(schema_ptr schema, lw_shared_ptr<sstable_list> sstables, const partition_key& key)
: _schema(std::move(schema))
, _key(sstables::key::from_partition_key(*_schema, key))
, _sstables(std::move(sstables))
{ }
virtual future<mutation_opt> operator()() override {
if (_done) {
return make_ready_future<mutation_opt>();
}
return parallel_for_each(*_sstables | boost::adaptors::map_values, [this](const lw_shared_ptr<sstables::sstable>& sstable) {
return sstable->read_row(_schema, _key).then([this](mutation_opt mo) {
apply(_m, std::move(mo));
});
}).then([this] {
_done = true;
return std::move(_m);
});
}
};
mutation_reader
column_family::make_sstable_reader(const query::partition_range& pr) const {
if (pr.is_singular() && pr.start()->value().has_key()) {
const dht::ring_position& pos = pr.start()->value();
if (dht::shard_of(pos.token()) != engine().cpu_id()) {
return make_empty_reader(); // range doesn't belong to this shard
}
return make_mutation_reader<single_key_sstable_reader>(_schema, _sstables, *pos.key());
} else {
// range_sstable_reader is not movable so we need to wrap it
return make_mutation_reader<range_sstable_reader>(_schema, _sstables, pr);
}
}
// Exposed for testing, not performance critical.
future<column_family::const_mutation_partition_ptr>
column_family::find_partition(const dht::decorated_key& key) const {
return do_with(query::partition_range::make_singular(key), [this] (auto& range) {
return do_with(this->make_reader(range), [] (mutation_reader& reader) {
return reader().then([] (mutation_opt&& mo) -> std::unique_ptr<const mutation_partition> {
if (!mo) {
return {};
}
return std::make_unique<const mutation_partition>(std::move(mo->partition()));
});
});
});
}
future<column_family::const_mutation_partition_ptr>
column_family::find_partition_slow(const partition_key& key) const {
return find_partition(dht::global_partitioner().decorate_key(*_schema, key));
}
future<column_family::const_row_ptr>
column_family::find_row(const dht::decorated_key& partition_key, clustering_key clustering_key) const {
return find_partition(partition_key).then([clustering_key = std::move(clustering_key)] (const_mutation_partition_ptr p) {
if (!p) {
return make_ready_future<const_row_ptr>();
}
auto r = p->find_row(clustering_key);
if (r) {
// FIXME: remove copy if only one data source
return make_ready_future<const_row_ptr>(std::make_unique<row>(*r));
} else {
return make_ready_future<const_row_ptr>();
}
});
}
mutation_reader
column_family::make_reader(const query::partition_range& range) const {
if (query::is_wrap_around(range, *_schema)) {
// make_combined_reader() can't handle streams that wrap around yet.
fail(unimplemented::cause::WRAP_AROUND);
}
std::vector<mutation_reader> readers;
readers.reserve(_memtables->size() + _sstables->size());
// We're assuming that cache and memtables are both read atomically
// for single-key queries, so we don't need to special case memtable
// undergoing a move to cache. At any given point in time between
// deferring points the sum of data in memtable and cache is coherent. If
// single-key queries for each data source were performed across deferring
// points, it would be possible that partitions which are ahead of the
// memtable cursor would be placed behind the cache cursor, resulting in
// those partitions being missing in the combined reader.
//
// We need to handle this in range queries though, as they are always
// deferring. scanning_reader from memtable.cc is falling back to reading
// the sstable when memtable is flushed. After memtable is moved to cache,
// new readers will no longer use the old memtable, but until then
// performance may suffer. We should fix this when we add support for
// range queries in cache, so that scans can always be satisfied form
// memtable and cache only, as long as data is not evicted.
//
// https://github.com/scylladb/scylla/issues/309
// https://github.com/scylladb/scylla/issues/185
for (auto&& mt : *_memtables) {
readers.emplace_back(mt->make_reader(range));
}
if (_config.enable_cache) {
readers.emplace_back(_cache.make_reader(range));
} else {
readers.emplace_back(make_sstable_reader(range));
}
return make_combined_reader(std::move(readers));
}
template <typename Func>
future<bool>
column_family::for_all_partitions(Func&& func) const {
static_assert(std::is_same<bool, std::result_of_t<Func(const dht::decorated_key&, const mutation_partition&)>>::value,
"bad Func signature");
struct iteration_state {
mutation_reader reader;
Func func;
bool ok = true;
bool empty = false;
public:
bool done() const { return !ok || empty; }
iteration_state(const column_family& cf, Func&& func)
: reader(cf.make_reader())
, func(std::move(func))
{ }
};
return do_with(iteration_state(*this, std::move(func)), [] (iteration_state& is) {
return do_until([&is] { return is.done(); }, [&is] {
return is.reader().then([&is](mutation_opt&& mo) {
if (!mo) {
is.empty = true;
} else {
is.ok = is.func(mo->decorated_key(), mo->partition());
}
});
}).then([&is] {
return is.ok;
});
});
}
future<bool>
column_family::for_all_partitions_slow(std::function<bool (const dht::decorated_key&, const mutation_partition&)> func) const {
return for_all_partitions(std::move(func));
}
class lister {
file _f;
std::function<future<> (directory_entry de)> _walker;
directory_entry_type _expected_type;
subscription<directory_entry> _listing;
sstring _dirname;
public:
lister(file f, directory_entry_type type, std::function<future<> (directory_entry)> walker, sstring dirname)
: _f(std::move(f))
, _walker(std::move(walker))
, _expected_type(type)
, _listing(_f.list_directory([this] (directory_entry de) { return _visit(de); }))
, _dirname(dirname) {
}
static future<> scan_dir(sstring name, directory_entry_type type, std::function<future<> (directory_entry)> walker);
protected:
future<> _visit(directory_entry de) {
return guarantee_type(std::move(de)).then([this] (directory_entry de) {
// Hide all synthetic directories and hidden files.
if ((de.type != _expected_type) || (de.name[0] == '.')) {
return make_ready_future<>();
}
return _walker(de);
});
}
future<> done() { return _listing.done(); }
private:
future<directory_entry> guarantee_type(directory_entry de) {
if (de.type) {
return make_ready_future<directory_entry>(std::move(de));
} else {
auto f = engine().file_type(_dirname + "/" + de.name);
return f.then([de = std::move(de)] (std::experimental::optional<directory_entry_type> t) mutable {
de.type = t;
return make_ready_future<directory_entry>(std::move(de));
});
}
}
};
future<> lister::scan_dir(sstring name, directory_entry_type type, std::function<future<> (directory_entry)> walker) {
return engine().open_directory(name).then([type, walker = std::move(walker), name] (file f) {
auto l = make_lw_shared<lister>(std::move(f), type, walker, name);
return l->done().then([l] { });
});
}
static std::vector<sstring> parse_fname(sstring filename) {
std::vector<sstring> comps;
boost::split(comps , filename ,boost::is_any_of(".-"));
return comps;
}
future<sstables::entry_descriptor> column_family::probe_file(sstring sstdir, sstring fname) {
using namespace sstables;
entry_descriptor comps = entry_descriptor::make_descriptor(fname);
// Every table will have a TOC. Using a specific file as a criteria, as
// opposed to, say verifying _sstables.count() to be zero is more robust
// against parallel loading of the directory contents.
if (comps.component != sstable::component_type::TOC) {
return make_ready_future<entry_descriptor>(std::move(comps));
}
// Make sure new sstables don't overwrite this one.
_sstable_generation = std::max<uint64_t>(_sstable_generation, comps.generation / smp::count + 1);
assert(_sstables->count(comps.generation) == 0);
auto sst = std::make_unique<sstables::sstable>(_schema->ks_name(), _schema->cf_name(), sstdir, comps.generation, comps.version, comps.format);
auto fut = sst->load();
return std::move(fut).then([this, sst = std::move(sst)] () mutable {
add_sstable(std::move(*sst));
return make_ready_future<>();
}).then_wrapped([fname, comps = std::move(comps)] (future<> f) {
try {
f.get();
} catch (malformed_sstable_exception& e) {
dblog.error("malformed sstable {}: {}. Refusing to boot", fname, e.what());
throw;
} catch(...) {
dblog.error("Unrecognized error while processing {}: Refusing to boot", fname);
throw;
}
return make_ready_future<entry_descriptor>(std::move(comps));
});
}
void column_family::update_stats_for_new_sstable(uint64_t new_sstable_data_size) {
_stats.live_disk_space_used += new_sstable_data_size;
_stats.total_disk_space_used += new_sstable_data_size;
_stats.live_sstable_count++;
}
void column_family::add_sstable(sstables::sstable&& sstable) {
add_sstable(make_lw_shared(std::move(sstable)));
}
void column_family::add_sstable(lw_shared_ptr<sstables::sstable> sstable) {
auto key_shard = [this] (const partition_key& pk) {
auto token = dht::global_partitioner().get_token(*_schema, pk);
return dht::shard_of(token);
};
auto s1 = key_shard(sstable->get_first_partition_key(*_schema));
auto s2 = key_shard(sstable->get_last_partition_key(*_schema));
auto me = engine().cpu_id();
auto included = (s1 <= me) && (me <= s2);
if (!included) {
dblog.info("sstable {} not relevant for this shard, ignoring", sstable->get_filename());
sstable->mark_for_deletion();
return;
}
auto generation = sstable->generation();
// allow in-progress reads to continue using old list
_sstables = make_lw_shared<sstable_list>(*_sstables);
update_stats_for_new_sstable(sstable->data_size());
_sstables->emplace(generation, std::move(sstable));
}
void column_family::add_memtable() {
// allow in-progress reads to continue using old list
_memtables = make_lw_shared(memtable_list(*_memtables));
_memtables->emplace_back(make_lw_shared<memtable>(_schema, _config.dirty_memory_region_group));
}
future<>
column_family::update_cache(memtable& m, lw_shared_ptr<sstable_list> old_sstables) {
if (_config.enable_cache) {
// be careful to use the old sstable list, since the new one will hit every
// mutation in m.
return _cache.update(m, make_partition_presence_checker(std::move(old_sstables)));
} else {
return make_ready_future<>();
}
}
future<>
column_family::seal_active_memtable() {
auto old = _memtables->back();
dblog.debug("Sealing active memtable, partitions: {}, occupancy: {}", old->partition_count(), old->occupancy());
if (!_config.enable_disk_writes) {
return make_ready_future<>();
}
if (old->empty()) {
dblog.debug("Memtable is empty");
return make_ready_future<>();
}
add_memtable();
assert(_highest_flushed_rp < old->replay_position()
|| (_highest_flushed_rp == db::replay_position() && old->replay_position() == db::replay_position())
);
_highest_flushed_rp = old->replay_position();
return _flush_queue->run_with_ordered_post_op(old->replay_position(), [old, this] {
return repeat([this, old] {
_flush_queue->check_open_gate();
return try_flush_memtable_to_sstable(old);
});
}, [old, this] {
if (_commitlog) {
_commitlog->discard_completed_segments(_schema->id(), old->replay_position());
}
});
// FIXME: release commit log
// FIXME: provide back-pressure to upper layers
}
future<stop_iteration>
column_family::try_flush_memtable_to_sstable(lw_shared_ptr<memtable> old) {
// FIXME: better way of ensuring we don't attempt to
// overwrite an existing table.
auto gen = _sstable_generation++ * smp::count + engine().cpu_id();
auto newtab = make_lw_shared<sstables::sstable>(_schema->ks_name(), _schema->cf_name(),
_config.datadir, gen,
sstables::sstable::version_types::ka,
sstables::sstable::format_types::big);
newtab->set_unshared();
dblog.debug("Flushing to {}", newtab->get_filename());
return newtab->write_components(*old).then([this, newtab, old] {
return newtab->open_data().then([this, newtab] {
// Note that due to our sharded architecture, it is possible that
// in the face of a value change some shards will backup sstables
// while others won't.
//
// This is, in theory, possible to mitigate through a rwlock.
// However, this doesn't differ from the situation where all tables
// are coming from a single shard and the toggle happens in the
// middle of them.
//
// The code as is guarantees that we'll never partially backup a
// single sstable, so that is enough of a guarantee.
if (!incremental_backups_enabled()) {
return make_ready_future<>();
}
auto dir = newtab->get_dir() + "/backups/";
return touch_directory(dir).then([dir, newtab] {
return newtab->create_links(dir);
});
});
}).then([this, old, newtab] {
dblog.debug("Flushing done");
// We must add sstable before we call update_cache(), because
// memtable's data after moving to cache can be evicted at any time.
auto old_sstables = _sstables;
add_sstable(newtab);
old->mark_flushed(newtab);
return update_cache(*old, std::move(old_sstables));
}).then_wrapped([this, old] (future<> ret) {
try {
ret.get();
_memtables->erase(boost::range::find(*_memtables, old));
dblog.debug("Memtable replaced");
trigger_compaction();
return make_ready_future<stop_iteration>(stop_iteration::yes);
} catch (...) {
dblog.error("failed to write sstable: {}", std::current_exception());
}
return sleep(10s).then([] {
return make_ready_future<stop_iteration>(stop_iteration::no);
});
});
}
void
column_family::start() {
// FIXME: add option to disable automatic compaction.
start_compaction();
}
future<>
column_family::stop() {
seal_active_memtable();
return _compaction_manager.remove(this).then([this] {
return _flush_queue->close();
});
}
future<>
column_family::compact_sstables(sstables::compaction_descriptor descriptor) {
if (!descriptor.sstables.size()) {
// if there is nothing to compact, just return.
return make_ready_future<>();
}
auto sstables_to_compact = make_lw_shared<std::vector<sstables::shared_sstable>>(std::move(descriptor.sstables));
auto new_tables = make_lw_shared<std::vector<
std::pair<unsigned, sstables::shared_sstable>>>();
auto create_sstable = [this, new_tables] {
// FIXME: this generation calculation should be in a function.
auto gen = _sstable_generation++ * smp::count + engine().cpu_id();
// FIXME: use "tmp" marker in names of incomplete sstable
auto sst = make_lw_shared<sstables::sstable>(_schema->ks_name(), _schema->cf_name(), _config.datadir, gen,
sstables::sstable::version_types::ka,
sstables::sstable::format_types::big);
sst->set_unshared();
new_tables->emplace_back(gen, sst);
return sst;
};
return sstables::compact_sstables(*sstables_to_compact, *this,
create_sstable, descriptor.max_sstable_bytes, descriptor.level).then([this, new_tables, sstables_to_compact] {
// Build a new list of _sstables: We remove from the existing list the
// tables we compacted (by now, there might be more sstables flushed
// later), and we add the new tables generated by the compaction.
// We create a new list rather than modifying it in-place, so that
// on-going reads can continue to use the old list.
auto current_sstables = _sstables;
_sstables = make_lw_shared<sstable_list>();
// zeroing live_disk_space_used and live_sstable_count because the
// sstable list is re-created below.
_stats.live_disk_space_used = 0;
_stats.live_sstable_count = 0;
std::unordered_set<sstables::shared_sstable> s(
sstables_to_compact->begin(), sstables_to_compact->end());
for (const auto& oldtab : *current_sstables) {
if (!s.count(oldtab.second)) {
update_stats_for_new_sstable(oldtab.second->data_size());
_sstables->emplace(oldtab.first, oldtab.second);
}
}
for (const auto& newtab : *new_tables) {
// FIXME: rename the new sstable(s). Verify a rename doesn't cause
// problems for the sstable object.
update_stats_for_new_sstable(newtab.second->data_size());
_sstables->emplace(newtab.first, newtab.second);
}
for (const auto& oldtab : *sstables_to_compact) {
oldtab->mark_for_deletion();
}
});
}
// FIXME: this is just an example, should be changed to something more general
// Note: We assume that the column_family does not get destroyed during compaction.
future<>
column_family::compact_all_sstables() {
std::vector<sstables::shared_sstable> sstables;
sstables.reserve(_sstables->size());
for (auto&& entry : *_sstables) {
sstables.push_back(entry.second);
}
// FIXME: check if the lower bound min_compaction_threshold() from schema
// should be taken into account before proceeding with compaction.
return compact_sstables(sstables::compaction_descriptor(std::move(sstables)));
}
void column_family::start_compaction() {
set_compaction_strategy(_schema->compaction_strategy());
}
void column_family::trigger_compaction() {
// Submitting compaction job to compaction manager.
if (!_compaction_disabled) {
_stats.pending_compactions++;
_compaction_manager.submit(this);
}
}
future<> column_family::run_compaction() {
sstables::compaction_strategy strategy = _compaction_strategy;
return do_with(std::move(strategy), [this] (sstables::compaction_strategy& cs) {
return cs.compact(*this).then([this] {
_stats.pending_compactions--;
});
});
}
void column_family::set_compaction_strategy(sstables::compaction_strategy_type strategy) {
_compaction_strategy = make_compaction_strategy(strategy, _schema->compaction_strategy_options());
}
bool column_family::compaction_manager_queued() const {
return _compaction_manager_queued;
}
void column_family::set_compaction_manager_queued(bool compaction_manager_queued) {
_compaction_manager_queued = compaction_manager_queued;
}
bool column_family::pending_compactions() const {
return _stats.pending_compactions > 0;
}
size_t column_family::sstables_count() {
return _sstables->size();
}
int64_t column_family::get_unleveled_sstables() const {
// TODO: when we support leveled compaction, we should return the number of
// SSTables in L0. If leveled compaction is enabled in this column family,
// then we should return zero, as we currently do.
return 0;
}
lw_shared_ptr<sstable_list> column_family::get_sstables() {
return _sstables;
}
future<> column_family::populate(sstring sstdir) {
// We can catch most errors when we try to load an sstable. But if the TOC
// file is the one missing, we won't try to load the sstable at all. This
// case is still an invalid case, but it is way easier for us to treat it
// by waiting for all files to be loaded, and then checking if we saw a
// file during scan_dir, without its corresponding TOC.
enum class status {
has_some_file,
has_toc_file,
has_temporary_toc_file,
};
struct sstable_descriptor {
std::experimental::optional<sstables::sstable::version_types> version;
std::experimental::optional<sstables::sstable::format_types> format;
};
auto verifier = make_lw_shared<std::unordered_map<unsigned long, status>>();
auto descriptor = make_lw_shared<sstable_descriptor>();
return lister::scan_dir(sstdir, directory_entry_type::regular, [this, sstdir, verifier, descriptor] (directory_entry de) {
// FIXME: The secondary indexes are in this level, but with a directory type, (starting with ".")
return probe_file(sstdir, de.name).then([verifier, descriptor] (auto entry) {
if (verifier->count(entry.generation)) {
if (verifier->at(entry.generation) == status::has_toc_file) {
if (entry.component == sstables::sstable::component_type::TOC) {
throw sstables::malformed_sstable_exception("Invalid State encountered. TOC file already processed");
} else if (entry.component == sstables::sstable::component_type::TemporaryTOC) {
throw sstables::malformed_sstable_exception("Invalid State encountered. Temporary TOC file found after TOC file was processed");
}
} else if (entry.component == sstables::sstable::component_type::TOC) {
verifier->at(entry.generation) = status::has_toc_file;
} else if (entry.component == sstables::sstable::component_type::TemporaryTOC) {
verifier->at(entry.generation) = status::has_temporary_toc_file;
}
} else {
if (entry.component == sstables::sstable::component_type::TOC) {
verifier->emplace(entry.generation, status::has_toc_file);
} else if (entry.component == sstables::sstable::component_type::TemporaryTOC) {
verifier->emplace(entry.generation, status::has_temporary_toc_file);
} else {
verifier->emplace(entry.generation, status::has_some_file);
}
}
// Retrieve both version and format used for this column family.
if (!descriptor->version) {
descriptor->version = entry.version;
}
if (!descriptor->format) {
descriptor->format = entry.format;
}
});
}).then([verifier, sstdir, descriptor, this] {
return parallel_for_each(*verifier, [sstdir = std::move(sstdir), descriptor, this] (auto v) {
if (v.second == status::has_temporary_toc_file) {
unsigned long gen = v.first;
assert(descriptor->version);
sstables::sstable::version_types version = descriptor->version.value();
assert(descriptor->format);
sstables::sstable::format_types format = descriptor->format.value();
if (engine().cpu_id() != 0) {
dblog.info("At directory: {}, partial SSTable with generation {} not relevant for this shard, ignoring", sstdir, v.first);
return make_ready_future<>();
}
// shard 0 is the responsible for removing a partial sstable.
return sstables::sstable::remove_sstable_with_temp_toc(_schema->ks_name(), _schema->cf_name(), sstdir, gen, version, format);
} else if (v.second != status::has_toc_file) {
throw sstables::malformed_sstable_exception(sprint("At directory: %s: no TOC found for SSTable with generation %d!. Refusing to boot", sstdir, v.first));
}
return make_ready_future<>();
});
});
}
utils::UUID database::empty_version = utils::UUID_gen::get_name_UUID(bytes{});
database::database() : database(db::config())
{}
database::database(const db::config& cfg)
: _cfg(std::make_unique<db::config>(cfg))
, _version(empty_version)
{
_memtable_total_space = size_t(_cfg->memtable_total_space_in_mb()) << 20;
if (!_memtable_total_space) {
_memtable_total_space = memory::stats().total_memory() / 2;
}
bool durable = cfg.data_file_directories().size() > 0;
db::system_keyspace::make(*this, durable, _cfg->volatile_system_keyspace_for_testing());
// Start compaction manager with two tasks for handling compaction jobs.
_compaction_manager.start(2);
setup_collectd();
dblog.info("Row: max_vector_size: {}, internal_count: {}", size_t(row::max_vector_size), size_t(row::internal_count));
}
void
database::setup_collectd() {
_collectd.push_back(
scollectd::add_polled_metric(scollectd::type_instance_id("memory"
, scollectd::per_cpu_plugin_instance
, "bytes", "dirty")
, scollectd::make_typed(scollectd::data_type::GAUGE, [this] {
return _dirty_memory_region_group.memory_used();
})));
}
database::~database() {
}
void database::update_version(const utils::UUID& version) {
_version = version;
}
const utils::UUID& database::get_version() const {
return _version;
}
future<> database::populate_keyspace(sstring datadir, sstring ks_name) {
auto ksdir = datadir + "/" + ks_name;
auto i = _keyspaces.find(ks_name);
if (i == _keyspaces.end()) {
dblog.warn("Skipping undefined keyspace: {}", ks_name);
} else {
dblog.info("Populating Keyspace {}", ks_name);
return lister::scan_dir(ksdir, directory_entry_type::directory, [this, ksdir, ks_name] (directory_entry de) {
auto comps = parse_fname(de.name);
if (comps.size() < 2) {
dblog.error("Keyspace {}: Skipping malformed CF {} ", ksdir, de.name);
return make_ready_future<>();
}
sstring cfname = comps[0];
auto sstdir = ksdir + "/" + de.name;
try {
auto& cf = find_column_family(ks_name, cfname);
dblog.info("Keyspace {}: Reading CF {} ", ksdir, cfname);
// FIXME: Increase parallelism.
return cf.populate(sstdir);
} catch (no_such_column_family&) {
dblog.warn("{}, CF {}: schema not loaded!", ksdir, comps[0]);
return make_ready_future<>();
}
});
}
return make_ready_future<>();
}
future<> database::populate(sstring datadir) {
return lister::scan_dir(datadir, directory_entry_type::directory, [this, datadir] (directory_entry de) {
auto& ks_name = de.name;
if (ks_name == "system") {
return make_ready_future<>();
}
return populate_keyspace(datadir, ks_name);
});
}
template <typename Func>
static future<>
do_parse_system_tables(distributed<service::storage_proxy>& proxy, const sstring& _cf_name, Func&& func) {
using namespace db::schema_tables;
static_assert(std::is_same<future<>, std::result_of_t<Func(schema_result::value_type&)>>::value,
"bad Func signature");
auto cf_name = make_lw_shared<sstring>(_cf_name);
return db::system_keyspace::query(proxy, *cf_name).then([] (auto rs) {
auto names = std::set<sstring>();
for (auto& r : rs->rows()) {
auto keyspace_name = r.template get_nonnull<sstring>("keyspace_name");
names.emplace(keyspace_name);
}
return std::move(names);
}).then([&proxy, cf_name, func = std::forward<Func>(func)] (std::set<sstring>&& names) mutable {
return parallel_for_each(names.begin(), names.end(), [&proxy, cf_name, func = std::forward<Func>(func)] (sstring name) mutable {
if (name == "system") {
return make_ready_future<>();
}
return read_schema_partition_for_keyspace(proxy, *cf_name, name).then([func, cf_name] (auto&& v) mutable {
return do_with(std::move(v), [func = std::forward<Func>(func), cf_name] (auto& v) {
return func(v).then_wrapped([cf_name, &v] (future<> f) {
try {
f.get();
} catch (std::exception& e) {
dblog.error("Skipping: {}. Exception occurred when loading system table {}: {}", v.first, *cf_name, e.what());
}
});
});
});
});
});
}
future<> database::parse_system_tables(distributed<service::storage_proxy>& proxy) {
using namespace db::schema_tables;
return do_parse_system_tables(proxy, db::schema_tables::KEYSPACES, [this] (schema_result::value_type &v) {
auto ksm = create_keyspace_from_schema_partition(v);
return create_keyspace(ksm);
}).then([&proxy, this] {
return do_parse_system_tables(proxy, db::schema_tables::COLUMNFAMILIES, [this, &proxy] (schema_result::value_type &v) {
return create_tables_from_tables_partition(proxy, v.second).then([this] (std::map<sstring, schema_ptr> tables) {
for (auto& t: tables) {
auto s = t.second;
auto& ks = this->find_keyspace(s->ks_name());
auto cfg = ks.make_column_family_config(*s);
this->add_column_family(std::move(s), std::move(cfg));
}
});
});
});
}
future<>
database::init_system_keyspace() {
// FIXME support multiple directories
return touch_directory(_cfg->data_file_directories()[0] + "/" + db::system_keyspace::NAME).then([this] {
return populate_keyspace(_cfg->data_file_directories()[0], db::system_keyspace::NAME).then([this]() {
return init_commitlog();
});
});
}
future<>
database::load_sstables(distributed<service::storage_proxy>& proxy) {
return parse_system_tables(proxy).then([this] {
return populate(_cfg->data_file_directories()[0]);
});
}
future<>
database::init_commitlog() {
return db::commitlog::create_commitlog(*_cfg).then([this](db::commitlog&& log) {
_commitlog = std::make_unique<db::commitlog>(std::move(log));
_commitlog->add_flush_handler([this](db::cf_id_type id, db::replay_position pos) {
if (_column_families.count(id) == 0) {
// the CF has been removed.
_commitlog->discard_completed_segments(id, pos);
return;
}
_column_families[id]->flush(pos);
}).release(); // we have longer life time than CL. Ignore reg anchor
});
}
unsigned
database::shard_of(const dht::token& t) {
return dht::shard_of(t);
}
unsigned
database::shard_of(const mutation& m) {
return shard_of(m.token());
}
unsigned
database::shard_of(const frozen_mutation& m) {
// FIXME: This lookup wouldn't be necessary if we
// sent the partition key in legacy form or together
// with token.
schema_ptr schema = find_schema(m.column_family_id());
return shard_of(dht::global_partitioner().get_token(*schema, m.key(*schema)));
}
void database::add_keyspace(sstring name, keyspace k) {
if (_keyspaces.count(name) != 0) {
throw std::invalid_argument("Keyspace " + name + " already exists");
}
_keyspaces.emplace(std::move(name), std::move(k));
}
void database::update_keyspace(const sstring& name) {
throw std::runtime_error("not implemented");
}
void database::drop_keyspace(const sstring& name) {
_keyspaces.erase(name);
}
void database::add_column_family(schema_ptr schema, column_family::config cfg) {
auto uuid = schema->id();
lw_shared_ptr<column_family> cf;
if (cfg.enable_commitlog && _commitlog) {
cf = make_lw_shared<column_family>(schema, std::move(cfg), *_commitlog, _compaction_manager);
} else {
cf = make_lw_shared<column_family>(schema, std::move(cfg), column_family::no_commitlog(), _compaction_manager);
}
auto ks = _keyspaces.find(schema->ks_name());
if (ks == _keyspaces.end()) {
throw std::invalid_argument("Keyspace " + schema->ks_name() + " not defined");
}
if (_column_families.count(uuid) != 0) {
throw std::invalid_argument("UUID " + uuid.to_sstring() + " already mapped");
}
auto kscf = std::make_pair(schema->ks_name(), schema->cf_name());
if (_ks_cf_to_uuid.count(kscf) != 0) {
throw std::invalid_argument("Column family " + schema->cf_name() + " exists");
}
ks->second.add_column_family(schema);
cf->start();
_column_families.emplace(uuid, std::move(cf));
_ks_cf_to_uuid.emplace(std::move(kscf), uuid);
}
future<> database::update_column_family(const sstring& ks_name, const sstring& cf_name) {
auto& proxy = service::get_storage_proxy();
auto old_cfm = find_schema(ks_name, cf_name);
return db::schema_tables::create_table_from_name(proxy, ks_name, cf_name).then([old_cfm] (auto&& new_cfm) {
if (old_cfm->id() != new_cfm->id()) {
return make_exception_future<>(exceptions::configuration_exception(sprint("Column family ID mismatch (found %s; expected %s)", new_cfm->id(), old_cfm->id())));
}
return make_exception_future<>(std::runtime_error("update column family not implemented"));
});
}
future<> database::drop_column_family(db_clock::time_point dropped_at, const sstring& ks_name, const sstring& cf_name) {
auto uuid = find_uuid(ks_name, cf_name);
auto& ks = find_keyspace(ks_name);
auto cf = _column_families.at(uuid);
_column_families.erase(uuid);
ks.metadata()->remove_column_family(cf->schema());
_ks_cf_to_uuid.erase(std::make_pair(ks_name, cf_name));
return truncate(dropped_at, ks, *cf).then([this, cf] {
return cf->stop();
}).then([this, cf] {
return make_ready_future<>();
});
}
const utils::UUID& database::find_uuid(const sstring& ks, const sstring& cf) const throw (std::out_of_range) {
return _ks_cf_to_uuid.at(std::make_pair(ks, cf));
}
const utils::UUID& database::find_uuid(const schema_ptr& schema) const throw (std::out_of_range) {
return find_uuid(schema->ks_name(), schema->cf_name());
}
keyspace& database::find_keyspace(const sstring& name) throw (no_such_keyspace) {
try {
return _keyspaces.at(name);
} catch (...) {
std::throw_with_nested(no_such_keyspace(name));
}
}
const keyspace& database::find_keyspace(const sstring& name) const throw (no_such_keyspace) {
try {
return _keyspaces.at(name);
} catch (...) {
std::throw_with_nested(no_such_keyspace(name));
}
}
bool database::has_keyspace(const sstring& name) const {
return _keyspaces.count(name) != 0;
}
std::vector<sstring> database::get_non_system_keyspaces() const {
std::vector<sstring> res;
for (auto const &i : _keyspaces) {
if (i.first != db::system_keyspace::NAME) {
res.push_back(i.first);
}
}
return res;
}
column_family& database::find_column_family(const sstring& ks_name, const sstring& cf_name) throw (no_such_column_family) {
try {
return find_column_family(find_uuid(ks_name, cf_name));
} catch (...) {
std::throw_with_nested(no_such_column_family(ks_name, cf_name));
}
}
const column_family& database::find_column_family(const sstring& ks_name, const sstring& cf_name) const throw (no_such_column_family) {
try {
return find_column_family(find_uuid(ks_name, cf_name));
} catch (...) {
std::throw_with_nested(no_such_column_family(ks_name, cf_name));
}
}
column_family& database::find_column_family(const utils::UUID& uuid) throw (no_such_column_family) {
try {
return *_column_families.at(uuid);
} catch (...) {
std::throw_with_nested(no_such_column_family(uuid));
}
}
const column_family& database::find_column_family(const utils::UUID& uuid) const throw (no_such_column_family) {
try {
return *_column_families.at(uuid);
} catch (...) {
std::throw_with_nested(no_such_column_family(uuid));
}
}
void
keyspace::create_replication_strategy(const std::map<sstring, sstring>& options) {
using namespace locator;
auto& ss = service::get_local_storage_service();
_replication_strategy =
abstract_replication_strategy::create_replication_strategy(
_metadata->name(), _metadata->strategy_name(),
ss.get_token_metadata(), options);
}
locator::abstract_replication_strategy&
keyspace::get_replication_strategy() {
return *_replication_strategy;
}
const locator::abstract_replication_strategy&
keyspace::get_replication_strategy() const {
return *_replication_strategy;
}
void
keyspace::set_replication_strategy(std::unique_ptr<locator::abstract_replication_strategy> replication_strategy) {
_replication_strategy = std::move(replication_strategy);
}
column_family::config
keyspace::make_column_family_config(const schema& s) const {
column_family::config cfg;
cfg.datadir = column_family_directory(s.cf_name(), s.id());
cfg.enable_disk_reads = _config.enable_disk_reads;
cfg.enable_disk_writes = _config.enable_disk_writes;
cfg.enable_commitlog = _config.enable_commitlog;
cfg.enable_cache = _config.enable_cache;
cfg.max_memtable_size = _config.max_memtable_size;
cfg.dirty_memory_region_group = _config.dirty_memory_region_group;
cfg.enable_incremental_backups = _config.enable_incremental_backups;
return cfg;
}
sstring
keyspace::column_family_directory(const sstring& name, utils::UUID uuid) const {
auto uuid_sstring = uuid.to_sstring();
boost::erase_all(uuid_sstring, "-");
return sprint("%s/%s-%s", _config.datadir, name, uuid_sstring);
}
future<>
keyspace::make_directory_for_column_family(const sstring& name, utils::UUID uuid) {
return touch_directory(column_family_directory(name, uuid));
}
no_such_keyspace::no_such_keyspace(const sstring& ks_name)
: runtime_error{sprint("Can't find a keyspace %s", ks_name)}
{
}
no_such_column_family::no_such_column_family(const utils::UUID& uuid)
: runtime_error{sprint("Can't find a column family with UUID %s", uuid)}
{
}
no_such_column_family::no_such_column_family(const sstring& ks_name, const sstring& cf_name)
: runtime_error{sprint("Can't find a column family %s in keyspace %s", cf_name, ks_name)}
{
}
column_family& database::find_column_family(const schema_ptr& schema) throw (no_such_column_family) {
return find_column_family(schema->id());
}
const column_family& database::find_column_family(const schema_ptr& schema) const throw (no_such_column_family) {
return find_column_family(schema->id());
}
void keyspace_metadata::validate() const {
using namespace locator;
auto& ss = service::get_local_storage_service();
abstract_replication_strategy::validate_replication_strategy(name(), strategy_name(), ss.get_token_metadata(), strategy_options());
}
schema_ptr database::find_schema(const sstring& ks_name, const sstring& cf_name) const throw (no_such_column_family) {
try {
return find_schema(find_uuid(ks_name, cf_name));
} catch (std::out_of_range&) {
std::throw_with_nested(no_such_column_family(ks_name, cf_name));
}
}
schema_ptr database::find_schema(const utils::UUID& uuid) const throw (no_such_column_family) {
return find_column_family(uuid).schema();
}
bool database::has_schema(const sstring& ks_name, const sstring& cf_name) const {
return _ks_cf_to_uuid.count(std::make_pair(ks_name, cf_name)) > 0;
}
void database::create_in_memory_keyspace(const lw_shared_ptr<keyspace_metadata>& ksm) {
keyspace ks(ksm, std::move(make_keyspace_config(*ksm)));
ks.create_replication_strategy(ksm->strategy_options());
_keyspaces.emplace(ksm->name(), std::move(ks));
}
future<>
database::create_keyspace(const lw_shared_ptr<keyspace_metadata>& ksm) {
auto i = _keyspaces.find(ksm->name());
if (i != _keyspaces.end()) {
return make_ready_future<>();
}
create_in_memory_keyspace(ksm);
auto& datadir = _keyspaces.at(ksm->name()).datadir();
if (datadir != "") {
return touch_directory(datadir);
} else {
return make_ready_future<>();
}
}
std::set<sstring>
database::existing_index_names(const sstring& cf_to_exclude) const {
std::set<sstring> names;
for (auto& p : _column_families) {
auto& cf = *p.second;
if (!cf_to_exclude.empty() && cf.schema()->cf_name() == cf_to_exclude) {
continue;
}
for (auto& cd : cf.schema()->all_columns_in_select_order()) {
if (cd.idx_info.index_name) {
names.emplace(*cd.idx_info.index_name);
}
}
}
return names;
}
// Based on:
// - org.apache.cassandra.db.AbstractCell#reconcile()
// - org.apache.cassandra.db.BufferExpiringCell#reconcile()
// - org.apache.cassandra.db.BufferDeletedCell#reconcile()
int
compare_atomic_cell_for_merge(atomic_cell_view left, atomic_cell_view right) {
if (left.timestamp() != right.timestamp()) {
return left.timestamp() > right.timestamp() ? 1 : -1;
}
if (left.is_live() != right.is_live()) {
return left.is_live() ? -1 : 1;
}
if (left.is_live()) {
auto c = compare_unsigned(left.value(), right.value());
if (c != 0) {
return c;
}
if (left.is_live_and_has_ttl()
&& right.is_live_and_has_ttl()
&& left.expiry() != right.expiry())
{
return left.expiry() < right.expiry() ? -1 : 1;
}
} else {
// Both are deleted
if (left.deletion_time() != right.deletion_time()) {
// Origin compares big-endian serialized deletion time. That's because it
// delegates to AbstractCell.reconcile() which compares values after
// comparing timestamps, which in case of deleted cells will hold
// serialized expiry.
return (uint32_t) left.deletion_time().time_since_epoch().count()
< (uint32_t) right.deletion_time().time_since_epoch().count() ? -1 : 1;
}
}
return 0;
}
struct query_state {
explicit query_state(const query::read_command& cmd, const std::vector<query::partition_range>& ranges)
: cmd(cmd)
, builder(cmd.slice)
, limit(cmd.row_limit)
, current_partition_range(ranges.begin())
, range_end(ranges.end()){
}
const query::read_command& cmd;
query::result::builder builder;
uint32_t limit;
bool range_empty = false; // Avoid ubsan false-positive when moving after construction
std::vector<query::partition_range>::const_iterator current_partition_range;
std::vector<query::partition_range>::const_iterator range_end;
mutation_reader reader;
bool done() const {
return !limit || current_partition_range == range_end;
}
};
future<lw_shared_ptr<query::result>>
column_family::query(const query::read_command& cmd, const std::vector<query::partition_range>& partition_ranges) {
utils::latency_counter lc;
_stats.reads.set_latency(lc);
return do_with(query_state(cmd, partition_ranges), [this] (query_state& qs) {
return do_until(std::bind(&query_state::done, &qs), [this, &qs] {
auto&& range = *qs.current_partition_range++;
qs.reader = make_reader(range);
qs.range_empty = false;
return do_until([&qs] { return !qs.limit || qs.range_empty; }, [this, &qs] {
return qs.reader().then([this, &qs](mutation_opt mo) {
if (mo) {
auto p_builder = qs.builder.add_partition(mo->key());
auto is_distinct = qs.cmd.slice.options.contains(query::partition_slice::option::distinct);
auto limit = !is_distinct ? qs.limit : 1;
mo->partition().query(p_builder, *_schema, qs.cmd.timestamp, limit);
qs.limit -= p_builder.row_count();
} else {
qs.range_empty = true;
}
});
});
}).then([&qs] {
return make_ready_future<lw_shared_ptr<query::result>>(
make_lw_shared<query::result>(qs.builder.build()));
});
}).finally([lc, this]() mutable {
_stats.reads.mark(lc);
});
}
mutation_source
column_family::as_mutation_source() const {
return [this] (const query::partition_range& range) {
return this->make_reader(range);
};
}
future<lw_shared_ptr<query::result>>
database::query(const query::read_command& cmd, const std::vector<query::partition_range>& ranges) {
static auto make_empty = [] {
return make_ready_future<lw_shared_ptr<query::result>>(make_lw_shared(query::result()));
};
try {
column_family& cf = find_column_family(cmd.cf_id);
return cf.query(cmd, ranges);
} catch (const no_such_column_family&) {
// FIXME: load from sstables
return make_empty();
}
}
future<reconcilable_result>
database::query_mutations(const query::read_command& cmd, const query::partition_range& range) {
try {
column_family& cf = find_column_family(cmd.cf_id);
return mutation_query(cf.as_mutation_source(), range, cmd.slice, cmd.row_limit, cmd.timestamp);
} catch (const no_such_column_family&) {
// FIXME: load from sstables
return make_ready_future<reconcilable_result>(reconcilable_result());
}
}
std::ostream& operator<<(std::ostream& out, const atomic_cell_or_collection& c) {
return out << to_hex(c._data);
}
std::ostream& operator<<(std::ostream& os, const mutation& m) {
fprint(os, "{mutation: schema %p key %s data ", m.schema().get(), m.decorated_key());
os << m.partition() << "}";
return os;
}
std::ostream& operator<<(std::ostream& out, const column_family& cf) {
return fprint(out, "{column_family: %s/%s}", cf._schema->ks_name(), cf._schema->cf_name());
}
std::ostream& operator<<(std::ostream& out, const database& db) {
out << "{\n";
for (auto&& e : db._column_families) {
auto&& cf = *e.second;
out << "(" << e.first.to_sstring() << ", " << cf.schema()->cf_name() << ", " << cf.schema()->ks_name() << "): " << cf << "\n";
}
out << "}";
return out;
}
future<> database::apply_in_memory(const frozen_mutation& m, const db::replay_position& rp) {
try {
auto& cf = find_column_family(m.column_family_id());
cf.apply(m, rp);
} catch (no_such_column_family&) {
// TODO: log a warning
// FIXME: load keyspace meta-data from storage
}
return make_ready_future<>();
}
future<> database::do_apply(const frozen_mutation& m) {
// I'm doing a nullcheck here since the init code path for db etc
// is a little in flux and commitlog is created only when db is
// initied from datadir.
auto& cf = find_column_family(m.column_family_id());
if (cf.commitlog() != nullptr) {
auto uuid = m.column_family_id();
bytes_view repr = m.representation();
auto write_repr = [repr] (data_output& out) { out.write(repr.begin(), repr.end()); };
return cf.commitlog()->add_mutation(uuid, repr.size(), write_repr).then([&m, this](auto rp) {
try {
return this->apply_in_memory(m, rp);
} catch (replay_position_reordered_exception&) {
// expensive, but we're assuming this is super rare.
// if we failed to apply the mutation due to future re-ordering
// (which should be the ever only reason for rp mismatch in CF)
// let's just try again, add the mutation to the CL once more,
// and assume success in inevitable eventually.
dblog.debug("replay_position reordering detected");
return this->apply(m);
}
});
}
return apply_in_memory(m, db::replay_position());
}
future<> database::throttle() {
if (_dirty_memory_region_group.memory_used() < _memtable_total_space
&& _throttled_requests.empty()) {
// All is well, go ahead
return make_ready_future<>();
}
// We must throttle, wait a bit
if (_throttled_requests.empty()) {
_throttling_timer.arm_periodic(10ms);
}
_throttled_requests.emplace_back();
return _throttled_requests.back().get_future();
}
void database::unthrottle() {
// Release one request per free 1MB we have
// FIXME: improve this
if (_dirty_memory_region_group.memory_used() >= _memtable_total_space) {
return;
}
size_t avail = (_memtable_total_space - _dirty_memory_region_group.memory_used()) >> 20;
avail = std::min(_throttled_requests.size(), avail);
for (size_t i = 0; i < avail; ++i) {
_throttled_requests.front().set_value();
_throttled_requests.pop_front();
}
if (_throttled_requests.empty()) {
_throttling_timer.cancel();
}
}
future<> database::apply(const frozen_mutation& m) {
return throttle().then([this, &m] {
return do_apply(m);
});
}
keyspace::config
database::make_keyspace_config(const keyspace_metadata& ksm) {
// FIXME support multiple directories
keyspace::config cfg;
if (_cfg->data_file_directories().size() > 0) {
cfg.datadir = sprint("%s/%s", _cfg->data_file_directories()[0], ksm.name());
cfg.enable_disk_writes = !_cfg->enable_in_memory_data_store();
cfg.enable_disk_reads = true; // we allways read from disk
cfg.enable_commitlog = ksm.durable_writes() && _cfg->enable_commitlog() && !_cfg->enable_in_memory_data_store();
cfg.enable_cache = _cfg->enable_cache();
cfg.max_memtable_size = _memtable_total_space * _cfg->memtable_cleanup_threshold();
} else {
cfg.datadir = "";
cfg.enable_disk_writes = false;
cfg.enable_disk_reads = false;
cfg.enable_commitlog = false;
cfg.enable_cache = false;
cfg.max_memtable_size = std::numeric_limits<size_t>::max();
}
cfg.dirty_memory_region_group = &_dirty_memory_region_group;
cfg.enable_incremental_backups = _cfg->incremental_backups();
return cfg;
}
namespace db {
std::ostream& operator<<(std::ostream& os, db::consistency_level cl) {
switch (cl) {
case db::consistency_level::ANY: return os << "ANY";
case db::consistency_level::ONE: return os << "ONE";
case db::consistency_level::TWO: return os << "TWO";
case db::consistency_level::THREE: return os << "THREE";
case db::consistency_level::QUORUM: return os << "QUORUM";
case db::consistency_level::ALL: return os << "ALL";
case db::consistency_level::LOCAL_QUORUM: return os << "LOCAL_QUORUM";
case db::consistency_level::EACH_QUORUM: return os << "EACH_QUORUM";
case db::consistency_level::SERIAL: return os << "SERIAL";
case db::consistency_level::LOCAL_SERIAL: return os << "LOCAL_SERIAL";
case db::consistency_level::LOCAL_ONE: return os << "LOCAL";
default: abort();
}
}
}
std::ostream&
operator<<(std::ostream& os, const exploded_clustering_prefix& ecp) {
// Can't pass to_hex() to transformed(), since it is overloaded, so wrap:
auto enhex = [] (auto&& x) { return to_hex(x); };
return fprint(os, "prefix{%s}", ::join(":", ecp._v | boost::adaptors::transformed(enhex)));
}
std::ostream&
operator<<(std::ostream& os, const atomic_cell_view& acv) {
if (acv.is_live()) {
return fprint(os, "atomic_cell{%s;ts=%d;expiry=%d,ttl=%d}",
to_hex(acv.value()),
acv.timestamp(),
acv.is_live_and_has_ttl() ? acv.expiry().time_since_epoch().count() : -1,
acv.is_live_and_has_ttl() ? acv.ttl().count() : 0);
} else {
return fprint(os, "atomic_cell{DEAD;ts=%d;deletion_time=%d}",
acv.timestamp(), acv.deletion_time().time_since_epoch().count());
}
}
std::ostream&
operator<<(std::ostream& os, const atomic_cell& ac) {
return os << atomic_cell_view(ac);
}
future<>
database::stop() {
return _compaction_manager.stop().then([this] {
// try to ensure that CL has done disk flushing
if (_commitlog != nullptr) {
return _commitlog->shutdown().then([this] {
return _commitlog->sync_all_segments();
});
}
return make_ready_future<>();
}).then([this] {
return parallel_for_each(_column_families, [this] (auto& val_pair) {
return val_pair.second->stop();
});
});
}
future<> database::flush_all_memtables() {
return parallel_for_each(_column_families, [this] (auto& cfp) {
return cfp.second->flush();
});
}
future<> database::truncate(db_clock::time_point truncated_at, sstring ksname, sstring cfname) {
auto& ks = find_keyspace(ksname);
auto& cf = find_column_family(ksname, cfname);
return truncate(truncated_at, ks, cf);
}
future<> database::truncate(db_clock::time_point truncated_at, const keyspace& ks, column_family& cf)
{
const auto durable = ks.metadata()->durable_writes();
const auto auto_snapshot = get_config().auto_snapshot();
future<> f = make_ready_future<>();
if (durable || auto_snapshot) {
// TODO:
// this is not really a guarantee at all that we've actually
// gotten all things to disk. Again, need queue-ish or something.
f = cf.flush();
} else {
cf.clear();
}
return cf.run_with_compaction_disabled([truncated_at, f = std::move(f), &cf, auto_snapshot, cfname = cf.schema()->cf_name()]() mutable {
return f.then([truncated_at, &cf, auto_snapshot, cfname = std::move(cfname)] {
dblog.debug("Discarding sstable data for truncated CF + indexes");
// TODO: notify truncation
future<> f = make_ready_future<>();
if (auto_snapshot) {
auto name = sprint("%d-%s", truncated_at.time_since_epoch().count(), cfname);
f = cf.snapshot(name);
}
return f.then([&cf, truncated_at] {
return cf.discard_sstables(truncated_at).then([&cf, truncated_at](db::replay_position rp) {
// TODO: indexes.
return db::system_keyspace::save_truncation_record(cf, truncated_at, rp);
});
});
});
});
}
const sstring& database::get_snitch_name() const {
return _cfg->endpoint_snitch();
}
future<> update_schema_version_and_announce(distributed<service::storage_proxy>& proxy)
{
return db::schema_tables::calculate_schema_digest(proxy).then([&proxy] (utils::UUID uuid) {
return proxy.local().get_db().invoke_on_all([uuid] (database& db) {
db.update_version(uuid);
return make_ready_future<>();
}).then([uuid] {
return db::system_keyspace::update_schema_version(uuid).then([uuid] {
return service::get_local_migration_manager().passive_announce(uuid);
});
});
});
}
// Snapshots: snapshotting the files themselves is easy: if more than one CF
// happens to link an SSTable twice, all but one will fail, and we will end up
// with one copy.
//
// The problem for us, is that the snapshot procedure is supposed to leave a
// manifest file inside its directory. So if we just call snapshot() from
// multiple shards, only the last one will succeed, writing its own SSTables to
// the manifest leaving all other shards' SSTables unaccounted for.
//
// Moreover, for things like drop table, the operation should only proceed when the
// snapshot is complete. That includes the manifest file being correctly written,
// and for this reason we need to wait for all shards to finish their snapshotting
// before we can move on.
//
// To know which files we must account for in the manifest, we will keep an
// SSTable set. Theoretically, we could just rescan the snapshot directory and
// see what's in there. But we would need to wait for all shards to finish
// before we can do that anyway. That is the hard part, and once that is done
// keeping the files set is not really a big deal.
//
// This code assumes that all shards will be snapshotting at the same time. So
// far this is a safe assumption, but if we ever want to take snapshots from a
// group of shards only, this code will have to be updated to account for that.
struct snapshot_manager {
std::unordered_set<sstring> files;
semaphore requests;
semaphore manifest_write;
snapshot_manager() : requests(0), manifest_write(0) {}
};
static thread_local std::unordered_map<sstring, lw_shared_ptr<snapshot_manager>> pending_snapshots;
static future<>
seal_snapshot(sstring jsondir) {
std::ostringstream ss;
int n = 0;
ss << "{" << std::endl << "\t\"files\" : [ ";
for (auto&& rf: pending_snapshots.at(jsondir)->files) {
if (n++ > 0) {
ss << ", ";
}
ss << "\"" << rf << "\"";
}
ss << " ]" << std::endl << "}" << std::endl;
auto json = ss.str();
auto jsonfile = jsondir + "/manifest.json";
dblog.debug("Storing manifest {}", jsonfile);
return recursive_touch_directory(jsondir).then([jsonfile, json = std::move(json)] {
return engine().open_file_dma(jsonfile, open_flags::wo | open_flags::create | open_flags::truncate).then([json](file f) {
return do_with(make_file_output_stream(std::move(f)), [json] (output_stream<char>& out) {
return out.write(json.c_str(), json.size()).then([&out] {
return out.flush();
}).then([&out] {
return out.close();
});
});
});
}).then([jsondir] {
return sync_directory(std::move(jsondir));
}).finally([jsondir] {
pending_snapshots.erase(jsondir);
return make_ready_future<>();
});
}
future<> column_family::snapshot(sstring name) {
return flush().then([this, name = std::move(name)]() {
auto tables = boost::copy_range<std::vector<sstables::shared_sstable>>(*_sstables | boost::adaptors::map_values);
return do_with(std::move(tables), [this, name](std::vector<sstables::shared_sstable> & tables) {
auto jsondir = _config.datadir + "/snapshots/" + name;
return parallel_for_each(tables, [name](sstables::shared_sstable sstable) {
auto dir = sstable->get_dir() + "/snapshots/" + name;
return recursive_touch_directory(dir).then([sstable, dir] {
return sstable->create_links(dir);
});
}).then([jsondir, &tables] {
// This is not just an optimization. If we have no files, jsondir may not have been created,
// and sync_directory would throw.
if (tables.size()) {
return sync_directory(std::move(jsondir));
} else {
return make_ready_future<>();
}
}).then([this, &tables, jsondir] {
auto shard = std::hash<sstring>()(jsondir) % smp::count;
std::unordered_set<sstring> table_names;
for (auto& sst : tables) {
auto f = sst->get_filename();
auto rf = f.substr(sst->get_dir().size() + 1);
table_names.insert(std::move(rf));
}
return smp::submit_to(shard, [requester = engine().cpu_id(), jsondir = std::move(jsondir),
tables = std::move(table_names), datadir = _config.datadir] {
if (pending_snapshots.count(jsondir) == 0) {
pending_snapshots.emplace(jsondir, make_lw_shared<snapshot_manager>());
}
auto snapshot = pending_snapshots.at(jsondir);
for (auto&& sst: tables) {
snapshot->files.insert(std::move(sst));
}
snapshot->requests.signal(1);
auto my_work = make_ready_future<>();
if (requester == engine().cpu_id()) {
my_work = snapshot->requests.wait(smp::count).then([jsondir = std::move(jsondir),
snapshot] () mutable {
return seal_snapshot(jsondir).then([snapshot] {
snapshot->manifest_write.signal(smp::count);
return make_ready_future<>();
});
});
}
return my_work.then([snapshot] {
return snapshot->manifest_write.wait(1);
}).then([snapshot] {});
});
});
});
});
}
future<> column_family::flush() {
// FIXME: this will synchronously wait for this write to finish, but doesn't guarantee
// anything about previous writes.
_stats.pending_flushes++;
return seal_active_memtable().finally([this]() mutable {
_stats.pending_flushes--;
// In origin memtable_switch_count is incremented inside
// ColumnFamilyMeetrics Flush.run
_stats.memtable_switch_count++;
return make_ready_future<>();
});
}
future<> column_family::flush(const db::replay_position& pos) {
// Technically possible if we've already issued the
// sstable write, but it is not done yet.
if (pos < _highest_flushed_rp) {
return make_ready_future<>();
}
// TODO: Origin looks at "secondary" memtables
// It also consideres "minReplayPosition", which is simply where
// the CL "started" (the first ever RP in this run).
// We ignore this for now and just say that if we're asked for
// a CF and it exists, we pretty much have to have data that needs
// flushing. Let's do it.
return seal_active_memtable();
}
void column_family::clear() {
_cache.clear();
_memtables->clear();
add_memtable();
}
// NOTE: does not need to be futurized, but might eventually, depending on
// if we implement notifications, whatnot.
future<db::replay_position> column_family::discard_sstables(db_clock::time_point truncated_at) {
assert(_stats.pending_compactions == 0);
db::replay_position rp;
auto gc_trunc = to_gc_clock(truncated_at);
auto pruned = make_lw_shared<sstable_list>();
for (auto&p : *_sstables) {
if (p.second->max_data_age() <= gc_trunc) {
rp = std::max(p.second->get_stats_metadata().position, rp);
p.second->mark_for_deletion();
continue;
}
pruned->emplace(p.first, p.second);
}
_sstables = std::move(pruned);
dblog.debug("cleaning out row cache");
_cache.clear();
return make_ready_future<db::replay_position>(rp);
}
std::ostream& operator<<(std::ostream& os, const user_types_metadata& m) {
os << "org.apache.cassandra.config.UTMetaData@" << &m;
return os;
}
std::ostream& operator<<(std::ostream& os, const keyspace_metadata& m) {
os << "KSMetaData{";
os << "name=" << m._name;
os << ", strategyClass=" << m._strategy_name;
os << ", strategyOptions={";
int n = 0;
for (auto& p : m._strategy_options) {
if (n++ != 0) {
os << ", ";
}
os << p.first << "=" << p.second;
}
os << "}";
os << ", cfMetaData={";
n = 0;
for (auto& p : m._cf_meta_data) {
if (n++ != 0) {
os << ", ";
}
os << p.first << "=" << p.second;
}
os << "}";
os << ", durable_writes=" << m._durable_writes;
os << ", userTypes=" << m._user_types;
os << "}";
return os;
}
Reference in New Issue View Git Blame Copy Permalink