mirrors/scylladb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge "Eliminate direct storage_proxy usage from cql3 statements" from Pavel E

Browse Source 
"

The token metadata and features should be kept on the query_processor itself,
so finally the "storage" API would look like this:

      6 .query()
      5 .get_max_result_size()
      2 .mutate_with_triggers()
      2 .cas()
      1 .truncate_blocking()

The get_max_result_size() is probably also worth moving away from storage,
it seem to have nothing to do with it.

tests: unit(dev)
"

* 'br-query-processor-in-cql-statements' of https://github.com/xemul/scylla:
  cql3: Generalize bounce-to-shard result creation
  cql3: Get data dictionary directly from query_processor
  create_keyspace_statement: Do not use proxy.shared_from_this()
  cas_request: Make read_command() accept query_processor
  select_statement: Replace all proxy-s with query_processor
  create_|alter_table_statement: Make check_restricted_table_properties() accept query_processor
  create_|alter_keyspace_statement: Make check_restricted_replication_strategy() accept query_processor
  role_management_statement: Make validate_cluster_support() accept query_processor
  drop_index_statement: Make lookup_indexed_table() accept query_processor
  batch_|modification_statement: Make get_mutations accept query_processor
  modification_statement: Replace most of proxy-s with query_processor
  batch_statement: Replace most of proxy-s with query_processor
  cql3: Make create_arg_types()/prepare_type() accept query_processor
  cql3: Make .validate_while_executing() accept query_processor
  cql3: Make execution stages carry query_processor over
  cql3: Make .validate() and .check_access() accept query_processor
This commit is contained in:
Avi Kivity
2021-12-23 15:59:15 +02:00
parent e4b2dfb54d d98dd0ff80
commit 9643f84d81
85 changed files with 363 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cql3/cql_statement.hh
View File

@@ -90,7 +90,7 @@ public:
*
* @param state the current client state
*/
virtual seastar::future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const = 0;
virtual seastar::future<> check_access(query_processor& qp, const service::client_state& state) const = 0;
/**
* Perform additional validation required by the statment.
@@ -98,7 +98,7 @@ public:
*
* @param state the current client state
*/
virtual void validate(service::storage_proxy& proxy, const service::client_state& state) const = 0;
virtual void validate(query_processor& qp, const service::client_state& state) const = 0;
/**
* Execute the statement and return the resulting result or null if there is no result.

20
cql3/query_processor.cc
View File

@@ -43,6 +43,7 @@
#include <seastar/core/metrics.hh>
#include "service/storage_proxy.hh"
#include "cql3/CqlParser.hpp"
#include "cql3/error_collector.hh"
#include "cql3/statements/batch_statement.hh"
@@ -503,7 +504,7 @@ query_processor::execute_direct(const sstring_view& query_string, service::query
metrics.regularStatementsExecuted.inc();
#endif
tracing::trace(query_state.get_trace_state(), "Processing a statement");
return cql_statement->check_access(_proxy, query_state.get_client_state()).then(
return cql_statement->check_access(*this, query_state.get_client_state()).then(
[this, cql_statement, &query_state, &options, warnings = move(warnings)] () mutable {
return process_authorized_statement(std::move(cql_statement), query_state, options).then(
[warnings = move(warnings)] (::shared_ptr<result_message> m) {
@@ -526,7 +527,7 @@ query_processor::execute_prepared(
::shared_ptr<cql_statement> statement = prepared->statement;
future<> fut = make_ready_future<>();
if (needs_authorization) {
fut = statement->check_access(_proxy, query_state.get_client_state()).then([this, &query_state, prepared = std::move(prepared), cache_key = std::move(cache_key)] () mutable {
fut = statement->check_access(*this, query_state.get_client_state()).then([this, &query_state, prepared = std::move(prepared), cache_key = std::move(cache_key)] () mutable {
return _authorized_prepared_cache.insert(*query_state.get_client_state().user(), std::move(cache_key), std::move(prepared)).handle_exception([this] (auto eptr) {
log.error("failed to cache the entry: {}", eptr);
});
@@ -545,7 +546,7 @@ query_processor::process_authorized_statement(const ::shared_ptr<cql_statement>
++_stats.queries_by_cl[size_t(options.get_consistency())];
statement->validate(_proxy, client_state);
statement->validate(*this, client_state);
auto fut = statement->execute(*this, query_state, options);
@@ -691,7 +692,7 @@ statements::prepared_statement::checked_weak_ptr query_processor::prepare_intern
if (p == nullptr) {
auto np = parse_statement(query_string)->prepare(_db, _cql_stats);
np->statement->raw_cql_statement = query_string;
np->statement->validate(_proxy, *_internal_state);
np->statement->validate(*this, *_internal_state);
p = std::move(np); // inserts it into map
}
return p->checked_weak_from_this();
@@ -849,7 +850,7 @@ query_processor::execute_internal(
} else {
auto p = parse_statement(query_string)->prepare(_db, _cql_stats);
p->statement->raw_cql_statement = query_string;
p->statement->validate(_proxy, *_internal_state);
p->statement->validate(*this, *_internal_state);
auto checked_weak_ptr = p->checked_weak_from_this();
return execute_with_params(std::move(checked_weak_ptr), cl, query_state, values).finally([p = std::move(p)] {});
}
@@ -875,14 +876,14 @@ query_processor::execute_batch(
service::query_state& query_state,
query_options& options,
std::unordered_map<prepared_cache_key_type, authorized_prepared_statements_cache::value_type> pending_authorization_entries) {
return batch->check_access(_proxy, query_state.get_client_state()).then([this, &query_state, &options, batch, pending_authorization_entries = std::move(pending_authorization_entries)] () mutable {
return batch->check_access(*this, query_state.get_client_state()).then([this, &query_state, &options, batch, pending_authorization_entries = std::move(pending_authorization_entries)] () mutable {
return parallel_for_each(pending_authorization_entries, [this, &query_state] (auto& e) {
return _authorized_prepared_cache.insert(*query_state.get_client_state().user(), e.first, std::move(e.second)).handle_exception([this] (auto eptr) {
log.error("failed to cache the entry: {}", eptr);
});
}).then([this, &query_state, &options, batch] {
batch->validate();
batch->validate(_proxy, query_state.get_client_state());
batch->validate(*this, query_state.get_client_state());
_stats.queries_by_cl[size_t(options.get_consistency())] += batch->get_statements().size();
if (log.is_enabled(logging::log_level::trace)) {
std::ostringstream oss;
@@ -998,4 +999,9 @@ future<> query_processor::query_internal(
return query_internal(query_string, db::consistency_level::ONE, {}, 1000, std::move(f));
}
shared_ptr<cql_transport::messages::result_message> query_processor::bounce_to_shard(unsigned shard, cql3::computed_function_values cached_fn_calls) {
_proxy.get_stats().replica_cross_shard_ops++;
return ::make_shared<cql_transport::messages::result_message::bounce_to_shard>(shard, std::move(cached_fn_calls));
}
}

2
cql3/query_processor.hh
View File

@@ -331,6 +331,8 @@ public:
friend class migration_subscriber;
shared_ptr<cql_transport::messages::result_message> bounce_to_shard(unsigned shard, cql3::computed_function_values cached_fn_calls);
private:
query_options make_internal_options(
const statements::prepared_statement::checked_weak_ptr& p,

13
cql3/statements/alter_keyspace_statement.cc
View File

@@ -62,11 +62,11 @@ const sstring& cql3::statements::alter_keyspace_statement::keyspace() const {
return _name;
}
future<> cql3::statements::alter_keyspace_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
return state.has_keyspace_access(proxy.local_db(), _name, auth::permission::ALTER);
future<> cql3::statements::alter_keyspace_statement::check_access(query_processor& qp, const service::client_state& state) const {
return state.has_keyspace_access(qp.proxy().local_db(), _name, auth::permission::ALTER);
}
void cql3::statements::alter_keyspace_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void cql3::statements::alter_keyspace_statement::validate(query_processor& qp, const service::client_state& state) const {
auto tmp = _name;
std::transform(tmp.begin(), tmp.end(), tmp.begin(), ::tolower);
if (is_system_keyspace(tmp)) {
@@ -93,9 +93,8 @@ void cql3::statements::alter_keyspace_statement::validate(service::storage_proxy
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_processor& qp) const {
try {
service::storage_proxy& proxy = qp.proxy();
auto old_ksm = proxy.data_dictionary().find_keyspace(_name).metadata();
const auto& tm = *proxy.get_token_metadata_ptr();
auto old_ksm = qp.db().find_keyspace(_name).metadata();
const auto& tm = *qp.proxy().get_token_metadata_ptr();
auto m = qp.get_migration_manager().prepare_keyspace_update_announcement(_attrs->as_ks_metadata_update(old_ksm, tm));
@@ -120,7 +119,7 @@ static logging::logger mylogger("alter_keyspace");
future<::shared_ptr<cql_transport::messages::result_message>>
cql3::statements::alter_keyspace_statement::execute(query_processor& qp, service::query_state& state, const query_options& options) const {
std::optional<sstring> warning = check_restricted_replication_strategy(qp.proxy(), keyspace(), *_attrs);
std::optional<sstring> warning = check_restricted_replication_strategy(qp, keyspace(), *_attrs);
return schema_altering_statement::execute(qp, state, options).then([this, warning = std::move(warning)] (::shared_ptr<messages::result_message> msg) {
if (warning) {
msg->add_warning(*warning);

4
cql3/statements/alter_keyspace_statement.hh
View File

@@ -62,8 +62,8 @@ public:
const sstring& keyspace() const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
void validate(service::storage_proxy& proxy, const service::client_state& state) const override;
future<> check_access(query_processor& qp, const service::client_state& state) const override;
void validate(query_processor& qp, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual future<::shared_ptr<messages::result_message>> execute(query_processor& qp, service::query_state& state, const query_options& options) const override;

4
cql3/statements/alter_role_statement.hh
View File

@@ -66,9 +66,9 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

4
cql3/statements/alter_service_level_statement.cc
View File

@@ -42,10 +42,10 @@ cql3::statements::alter_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<alter_service_level_statement>(*this));
}
void alter_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void alter_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> alter_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> alter_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::ALTER, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/alter_service_level_statement.hh
View File

@@ -37,8 +37,8 @@ class alter_service_level_statement final : public service_level_statement {
public:
alter_service_level_statement(sstring service_level, shared_ptr<sl_prop_defs> attrs);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

8
cql3/statements/alter_table_statement.cc
View File

@@ -75,13 +75,13 @@ alter_table_statement::alter_table_statement(cf_name name,
{
}
future<> alter_table_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> alter_table_statement::check_access(query_processor& qp, const service::client_state& state) const {
using cdt = auth::command_desc::type;
return state.has_column_family_access(proxy.local_db(), keyspace(), column_family(), auth::permission::ALTER,
return state.has_column_family_access(qp.proxy().local_db(), keyspace(), column_family(), auth::permission::ALTER,
_type == type::opts ? cdt::ALTER_WITH_OPTS : cdt::OTHER);
}
void alter_table_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
void alter_table_statement::validate(query_processor& qp, const service::client_state& state) const
{
// validated in prepare_schema_mutations()
}
@@ -436,7 +436,7 @@ cql3::statements::alter_table_statement::prepare(data_dictionary::database db, c
future<::shared_ptr<messages::result_message>>
alter_table_statement::execute(query_processor& qp, service::query_state& state, const query_options& options) const {
std::optional<sstring> warning = check_restricted_table_properties(qp.proxy(), keyspace(), column_family(), *_properties);
std::optional<sstring> warning = check_restricted_table_properties(qp, keyspace(), column_family(), *_properties);
return schema_altering_statement::execute(qp, state, options).then([this, warning = std::move(warning)] (::shared_ptr<messages::result_message> msg) {
if (warning) {
msg->add_warning(*warning);

4
cql3/statements/alter_table_statement.hh
View File

@@ -81,8 +81,8 @@ public:
std::optional<cf_prop_defs> properties,
renames_type renames);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual void validate(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(query_processor& qp, const service::client_state& state) const override;
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual future<::shared_ptr<messages::result_message>> execute(query_processor& qp, service::query_state& state, const query_options& options) const override;

6
cql3/statements/alter_type_statement.cc
View File

@@ -68,12 +68,12 @@ void alter_type_statement::prepare_keyspace(const service::client_state& state)
}
}
future<> alter_type_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> alter_type_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_keyspace_access(proxy.local_db(), keyspace(), auth::permission::ALTER);
return state.has_keyspace_access(qp.proxy().local_db(), keyspace(), auth::permission::ALTER);
}
void alter_type_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
void alter_type_statement::validate(query_processor& qp, const service::client_state& state) const
{
// Validation is left to announceMigration as it's easier to do it while constructing the updated type.
// It doesn't really change anything anyway.

4
cql3/statements/alter_type_statement.hh
View File

@@ -65,9 +65,9 @@ public:
virtual void prepare_keyspace(const service::client_state& state) override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual void validate(query_processor& qp, const service::client_state& state) const override;
virtual const sstring& keyspace() const override;

6
cql3/statements/alter_view_statement.cc
View File

@@ -60,10 +60,10 @@ alter_view_statement::alter_view_statement(cf_name view_name, std::optional<cf_p
{
}
future<> alter_view_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> alter_view_statement::check_access(query_processor& qp, const service::client_state& state) const
{
try {
const data_dictionary::database db = proxy.data_dictionary();
const data_dictionary::database db = qp.db();
auto&& s = db.find_schema(keyspace(), column_family());
if (s->is_view()) {
return state.has_column_family_access(db.real_database(), keyspace(), s->view_info()->base_name(), auth::permission::ALTER);
@@ -74,7 +74,7 @@ future<> alter_view_statement::check_access(service::storage_proxy& proxy, const
return make_ready_future<>();
}
void alter_view_statement::validate(service::storage_proxy&, const service::client_state& state) const
void alter_view_statement::validate(query_processor&, const service::client_state& state) const
{
// validated in prepare_schema_mutations()
}

4
cql3/statements/alter_view_statement.hh
View File

@@ -62,9 +62,9 @@ private:
public:
alter_view_statement(cf_name view_name, std::optional<cf_prop_defs> properties);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;

4
cql3/statements/attach_service_level_statement.cc
View File

@@ -41,10 +41,10 @@ cql3::statements::attach_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<attach_service_level_statement>(*this));
}
void attach_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void attach_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> attach_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> attach_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::AUTHORIZE, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/attach_service_level_statement.hh
View File

@@ -36,8 +36,8 @@ class attach_service_level_statement final : public service_level_statement {
public:
attach_service_level_statement(sstring service_level, sstring role_name);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

4
cql3/statements/authentication_statement.cc
View File

@@ -57,10 +57,10 @@ bool cql3::statements::authentication_statement::depends_on_column_family(
}
void cql3::statements::authentication_statement::validate(
service::storage_proxy&,
query_processor&,
const service::client_state& state) const {
}
future<> cql3::statements::authentication_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> cql3::statements::authentication_statement::check_access(query_processor& qp, const service::client_state& state) const {
return make_ready_future<>();
}

4
cql3/statements/authentication_statement.hh
View File

@@ -59,9 +59,9 @@ public:
bool depends_on_column_family(const sstring& cf_name) const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
future<> check_access(query_processor& qp, const service::client_state& state) const override;
void validate(service::storage_proxy&, const service::client_state& state) const override;
void validate(query_processor&, const service::client_state& state) const override;
};
}

4
cql3/statements/authorization_statement.cc
View File

@@ -59,11 +59,11 @@ bool cql3::statements::authorization_statement::depends_on_column_family(
}
void cql3::statements::authorization_statement::validate(
service::storage_proxy&,
query_processor&,
const service::client_state& state) const {
}
future<> cql3::statements::authorization_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> cql3::statements::authorization_statement::check_access(query_processor& qp, const service::client_state& state) const {
return make_ready_future<>();
}

4
cql3/statements/authorization_statement.hh
View File

@@ -63,9 +63,9 @@ public:
bool depends_on_column_family(const sstring& cf_name) const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
future<> check_access(query_processor& qp, const service::client_state& state) const override;
void validate(service::storage_proxy&, const service::client_state& state) const override;
void validate(query_processor&, const service::client_state& state) const override;
protected:
static void maybe_correct_resource(auth::resource&, const service::client_state&);

51
cql3/statements/batch_statement.cc
View File

@@ -113,11 +113,11 @@ uint32_t batch_statement::get_bound_terms() const
return _bound_terms;
}
future<> batch_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> batch_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return parallel_for_each(_statements.begin(), _statements.end(), [&proxy, &state](auto&& s) {
return parallel_for_each(_statements.begin(), _statements.end(), [&qp, &state](auto&& s) {
if (s.needs_authorization) {
return s.statement->check_access(proxy, state);
return s.statement->check_access(qp, state);
} else {
return make_ready_future<>();
}
@@ -177,10 +177,10 @@ void batch_statement::validate()
}
}
void batch_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
void batch_statement::validate(query_processor& qp, const service::client_state& state) const
{
for (auto&& s : _statements) {
s.statement->validate(proxy, state);
s.statement->validate(qp, state);
}
}
@@ -189,20 +189,20 @@ const std::vector<batch_statement::single_statement>& batch_statement::get_state
return _statements;
}
future<std::vector<mutation>> batch_statement::get_mutations(service::storage_proxy& storage, const query_options& options,
future<std::vector<mutation>> batch_statement::get_mutations(query_processor& qp, const query_options& options,
db::timeout_clock::time_point timeout, bool local, api::timestamp_type now, service::query_state& query_state) const {
// Do not process in parallel because operations like list append/prepend depend on execution order.
using mutation_set_type = std::unordered_set<mutation, mutation_hash_by_key, mutation_equals_by_key>;
return do_with(mutation_set_type(), [this, &storage, &options, timeout, now, local, &query_state] (auto& result) mutable {
return do_with(mutation_set_type(), [this, &qp, &options, timeout, now, local, &query_state] (auto& result) mutable {
result.reserve(_statements.size());
return do_for_each(boost::make_counting_iterator<size_t>(0),
boost::make_counting_iterator<size_t>(_statements.size()),
[this, &storage, &options, now, local, &result, timeout, &query_state] (size_t i) {
[this, &qp, &options, now, local, &result, timeout, &query_state] (size_t i) {
auto&& statement = _statements[i].statement;
statement->inc_cql_stats(query_state.get_client_state().is_internal());
auto&& statement_options = options.for_statement(i);
auto timestamp = _attrs->get_timestamp(now, statement_options);
return statement->get_mutations(storage, statement_options, timeout, local, timestamp, query_state).then([&result] (auto&& more) {
return statement->get_mutations(qp, statement_options, timeout, local, timestamp, query_state).then([&result] (auto&& more) {
for (auto&& m : more) {
// We want unordered_set::try_emplace(), but we don't have it
auto pos = result.find(m);
@@ -225,13 +225,13 @@ future<std::vector<mutation>> batch_statement::get_mutations(service::storage_pr
});
}
void batch_statement::verify_batch_size(service::storage_proxy& proxy, const std::vector<mutation>& mutations) {
void batch_statement::verify_batch_size(query_processor& qp, const std::vector<mutation>& mutations) {
if (mutations.size() <= 1) {
return; // We only warn for batch spanning multiple mutations
}
size_t warn_threshold = proxy.data_dictionary().get_config().batch_size_warn_threshold_in_kb() * 1024;
size_t fail_threshold = proxy.data_dictionary().get_config().batch_size_fail_threshold_in_kb() * 1024;
size_t warn_threshold = qp.db().get_config().batch_size_warn_threshold_in_kb() * 1024;
size_t fail_threshold = qp.db().get_config().batch_size_fail_threshold_in_kb() * 1024;
size_t size = 0;
for (auto&m : mutations) {
@@ -262,7 +262,7 @@ struct batch_statement_executor {
static thread_local inheriting_concrete_execution_stage<
future<shared_ptr<cql_transport::messages::result_message>>,
const batch_statement*,
service::storage_proxy&,
query_processor&,
service::query_state&,
const query_options&,
bool,
@@ -270,14 +270,13 @@ static thread_local inheriting_concrete_execution_stage<
future<shared_ptr<cql_transport::messages::result_message>> batch_statement::execute(
query_processor& qp, service::query_state& state, const query_options& options) const {
service::storage_proxy& storage = qp.proxy();
cql3::util::validate_timestamp(options, _attrs);
return batch_stage(this, seastar::ref(storage), seastar::ref(state),
return batch_stage(this, seastar::ref(qp), seastar::ref(state),
seastar::cref(options), false, options.get_timestamp(state));
}
future<shared_ptr<cql_transport::messages::result_message>> batch_statement::do_execute(
service::storage_proxy& storage,
query_processor& qp,
service::query_state& query_state, const query_options& options,
bool local, api::timestamp_type now) const
{
@@ -291,16 +290,16 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::do_
if (_has_conditions) {
++_stats.cas_batches;
_stats.statements_in_cas_batches += _statements.size();
return execute_with_conditions(storage, options, query_state);
return execute_with_conditions(qp, options, query_state);
}
++_stats.batches;
_stats.statements_in_batches += _statements.size();
auto timeout = db::timeout_clock::now() + get_timeout(query_state.get_client_state(), options);
return get_mutations(storage, options, timeout, local, now, query_state).then([this, &storage, &options, timeout, tr_state = query_state.get_trace_state(),
return get_mutations(qp, options, timeout, local, now, query_state).then([this, &qp, &options, timeout, tr_state = query_state.get_trace_state(),
permit = query_state.get_permit()] (std::vector<mutation> ms) mutable {
return execute_without_conditions(storage, std::move(ms), options.get_consistency(), timeout, std::move(tr_state), std::move(permit));
return execute_without_conditions(qp, std::move(ms), options.get_consistency(), timeout, std::move(tr_state), std::move(permit));
}).then([] {
return make_ready_future<shared_ptr<cql_transport::messages::result_message>>(
make_shared<cql_transport::messages::result_message::void_message>());
@@ -308,7 +307,7 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::do_
}
future<> batch_statement::execute_without_conditions(
service::storage_proxy& storage,
query_processor& qp,
std::vector<mutation> mutations,
db::consistency_level cl,
db::timeout_clock::time_point timeout,
@@ -326,7 +325,7 @@ future<> batch_statement::execute_without_conditions(
}
}));
#endif
verify_batch_size(storage, mutations);
verify_batch_size(qp, mutations);
bool mutate_atomic = true;
if (_type != type::LOGGED) {
@@ -340,11 +339,11 @@ future<> batch_statement::execute_without_conditions(
mutate_atomic = false;
}
}
return storage.mutate_with_triggers(std::move(mutations), cl, timeout, mutate_atomic, std::move(tr_state), std::move(permit));
return qp.proxy().mutate_with_triggers(std::move(mutations), cl, timeout, mutate_atomic, std::move(tr_state), std::move(permit));
}
future<shared_ptr<cql_transport::messages::result_message>> batch_statement::execute_with_conditions(
service::storage_proxy& proxy,
query_processor& qp,
const query_options& options,
service::query_state& qs) const {
@@ -391,12 +390,12 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
auto shard = service::storage_proxy::cas_shard(*_statements[0].statement->s, request->key()[0].start()->value().as_decorated_key().token());
if (shard != this_shard_id()) {
proxy.get_stats().replica_cross_shard_ops++;
return make_ready_future<shared_ptr<cql_transport::messages::result_message>>(
::make_shared<cql_transport::messages::result_message::bounce_to_shard>(shard, std::move(cached_fn_calls)));
qp.bounce_to_shard(shard, std::move(cached_fn_calls))
);
}
return proxy.cas(schema, request, request->read_command(proxy), request->key(),
return qp.proxy().cas(schema, request, request->read_command(qp), request->key(),
{read_timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()},
cl_for_paxos, cl_for_learn, batch_timeout, cas_timeout).then([this, request] (bool is_applied) {
return request->build_cas_result_set(_metadata, _columns_of_cas_result_set, is_applied);

14
cql3/statements/batch_statement.hh
View File

@@ -122,7 +122,7 @@ public:
virtual uint32_t get_bound_terms() const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
// Validates a prepared batch statement without validating its nested statements.
void validate();
@@ -133,11 +133,11 @@ public:
// The batch itself will be validated in either Parsed#prepare() - for regular CQL3 batches,
// or in QueryProcessor.processBatch() - for native protocol batches.
virtual void validate(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual void validate(query_processor& qp, const service::client_state& state) const override;
const std::vector<single_statement>& get_statements();
private:
future<std::vector<mutation>> get_mutations(service::storage_proxy& storage, const query_options& options, db::timeout_clock::time_point timeout,
future<std::vector<mutation>> get_mutations(query_processor& qp, const query_options& options, db::timeout_clock::time_point timeout,
bool local, api::timestamp_type now, service::query_state& query_state) const;
public:
@@ -145,19 +145,19 @@ public:
* Checks batch size to ensure threshold is met. If not, a warning is logged.
* @param cfs ColumnFamilies that will store the batch's mutations.
*/
static void verify_batch_size(service::storage_proxy& proxy, const std::vector<mutation>& mutations);
static void verify_batch_size(query_processor& qp, const std::vector<mutation>& mutations);
virtual future<shared_ptr<cql_transport::messages::result_message>> execute(
query_processor& qp, service::query_state& state, const query_options& options) const override;
private:
friend class batch_statement_executor;
future<shared_ptr<cql_transport::messages::result_message>> do_execute(
service::storage_proxy& storage,
query_processor& qp,
service::query_state& query_state, const query_options& options,
bool local, api::timestamp_type now) const;
future<> execute_without_conditions(
service::storage_proxy& storage,
query_processor& qp,
std::vector<mutation> mutations,
db::consistency_level cl,
db::timeout_clock::time_point timeout,
@@ -165,7 +165,7 @@ private:
service_permit permit) const;
future<shared_ptr<cql_transport::messages::result_message>> execute_with_conditions(
service::storage_proxy& storage,
query_processor& qp,
const query_options& options,
service::query_state& state) const;

5
cql3/statements/cas_request.cc
View File

@@ -46,6 +46,7 @@
#include "transport/messages/result_message.hh"
#include "types/map.hh"
#include "service/storage_proxy.hh"
#include "cql3/query_processor.hh"
namespace cql3::statements {
@@ -85,7 +86,7 @@ std::optional<mutation> cas_request::apply_updates(api::timestamp_type ts) const
return mutation_set;
}
lw_shared_ptr<query::read_command> cas_request::read_command(service::storage_proxy& proxy) const {
lw_shared_ptr<query::read_command> cas_request::read_command(query_processor& qp) const {
column_set columns_to_read(_schema->all_columns_count());
std::vector<query::clustering_range> ranges;
@@ -120,7 +121,7 @@ lw_shared_ptr<query::read_command> cas_request::read_command(service::storage_pr
options.set(query::partition_slice::option::always_return_static_content);
query::partition_slice ps(std::move(ranges), *_schema, columns_to_read, options);
ps.set_partition_row_limit(max_rows);
return make_lw_shared<query::read_command>(_schema->id(), _schema->version(), std::move(ps), proxy.get_max_result_size(ps));
return make_lw_shared<query::read_command>(_schema->id(), _schema->version(), std::move(ps), qp.proxy().get_max_result_size(ps));
}
bool cas_request::applies_to() const {

8
cql3/statements/cas_request.hh
View File

@@ -41,12 +41,6 @@
#include "service/paxos/cas_request.hh"
#include "cql3/statements/modification_statement.hh"
namespace service {
class storage_proxy;
} // namespace service
namespace cql3::statements {
using namespace std::chrono;
@@ -96,7 +90,7 @@ public:
return _rows;
}
lw_shared_ptr<query::read_command> read_command(service::storage_proxy& proxy) const;
lw_shared_ptr<query::read_command> read_command(query_processor& qp) const;
void add_row_update(const modification_statement& stmt_arg, std::vector<query::clustering_range> ranges_arg,
modification_statement::json_cache_opt json_cache_arg, const query_options& options_arg);

10
cql3/statements/create_aggregate_statement.cc
View File

@@ -35,16 +35,16 @@ namespace cql3 {
namespace statements {
shared_ptr<functions::function> create_aggregate_statement::create(service::storage_proxy& proxy, functions::function* old) const {
if (!proxy.features().cluster_supports_user_defined_aggregates()) {
shared_ptr<functions::function> create_aggregate_statement::create(query_processor& qp, functions::function* old) const {
if (!qp.proxy().features().cluster_supports_user_defined_aggregates()) {
throw exceptions::invalid_request_exception("Cluster does not support user-defined aggregates, upgrade the whole cluster in order to use UDA");
}
if (old && !dynamic_cast<functions::user_aggregate*>(old)) {
throw exceptions::invalid_request_exception(format("Cannot replace '{}' which is not a user defined aggregate", *old));
}
data_type state_type = prepare_type(proxy, *_stype);
data_type state_type = prepare_type(qp, *_stype);
auto&& db = proxy.data_dictionary();
auto&& db = qp.db();
std::vector<data_type> acc_types{state_type};
acc_types.insert(acc_types.end(), _arg_types.begin(), _arg_types.end());
auto state_func = dynamic_pointer_cast<functions::scalar_function>(functions::functions::find(functions::function_name{_name.keyspace, _sfunc}, acc_types));
@@ -74,7 +74,7 @@ create_aggregate_statement::prepare_schema_mutations(query_processor& qp) const
::shared_ptr<cql_transport::event::schema_change> ret;
std::vector<mutation> m;
auto aggregate = dynamic_pointer_cast<functions::user_aggregate>(validate_while_executing(qp.proxy()));
auto aggregate = dynamic_pointer_cast<functions::user_aggregate>(validate_while_executing(qp));
if (aggregate) {
m = co_await qp.get_migration_manager().prepare_new_aggregate_announcement(aggregate);
ret = create_schema_change(*aggregate, true);

2
cql3/statements/create_aggregate_statement.hh
View File

@@ -39,7 +39,7 @@ class create_aggregate_statement final : public create_function_statement_base {
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
virtual shared_ptr<functions::function> create(service::storage_proxy& proxy, functions::function* old) const override;
virtual shared_ptr<functions::function> create(query_processor& qp, functions::function* old) const override;
sstring _sfunc;
shared_ptr<cql3_type::raw> _stype;

8
cql3/statements/create_function_statement.cc
View File

@@ -35,20 +35,20 @@ namespace cql3 {
namespace statements {
shared_ptr<functions::function> create_function_statement::create(service::storage_proxy& proxy, functions::function* old) const {
shared_ptr<functions::function> create_function_statement::create(query_processor& qp, functions::function* old) const {
if (old && !dynamic_cast<functions::user_function*>(old)) {
throw exceptions::invalid_request_exception(format("Cannot replace '{}' which is not a user defined function", *old));
}
if (_language != "lua" && _language != "xwasm") {
throw exceptions::invalid_request_exception(format("Language '{}' is not supported", _language));
}
data_type return_type = prepare_type(proxy, *_return_type);
data_type return_type = prepare_type(qp, *_return_type);
std::vector<sstring> arg_names;
for (const auto& arg_name : _arg_names) {
arg_names.push_back(arg_name->to_string());
}
auto&& db = proxy.data_dictionary();
auto&& db = qp.db();
if (_language == "lua") {
auto cfg = lua::make_runtime_config(db.get_config());
functions::user_function::context ctx = functions::user_function::lua_context {
@@ -81,7 +81,7 @@ create_function_statement::prepare_schema_mutations(query_processor& qp) const {
::shared_ptr<cql_transport::event::schema_change> ret;
std::vector<mutation> m;
auto func = dynamic_pointer_cast<functions::user_function>(validate_while_executing(qp.proxy()));
auto func = dynamic_pointer_cast<functions::user_function>(validate_while_executing(qp));
if (func) {
m = co_await qp.get_migration_manager().prepare_new_function_announcement(func);

2
cql3/statements/create_function_statement.hh
View File

@@ -38,7 +38,7 @@ class create_function_statement final : public create_function_statement_base {
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
virtual shared_ptr<functions::function> create(service::storage_proxy& proxy, functions::function* old) const override;
virtual shared_ptr<functions::function> create(query_processor& qp, functions::function* old) const override;
sstring _language;
sstring _body;
std::vector<shared_ptr<column_identifier>> _arg_names;

12
cql3/statements/create_index_statement.cc
View File

@@ -78,12 +78,12 @@ create_index_statement::create_index_statement(cf_name name,
}
future<>
create_index_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
return state.has_column_family_access(proxy.local_db(), keyspace(), column_family(), auth::permission::ALTER);
create_index_statement::check_access(query_processor& qp, const service::client_state& state) const {
return state.has_column_family_access(qp.proxy().local_db(), keyspace(), column_family(), auth::permission::ALTER);
}
void
create_index_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
create_index_statement::validate(query_processor& qp, const service::client_state& state) const
{
if (_raw_targets.empty() && !_properties->is_custom) {
throw exceptions::invalid_request_exception("Only CUSTOM indexes can be created without specifying a target column");
@@ -92,8 +92,8 @@ create_index_statement::validate(service::storage_proxy& proxy, const service::c
_properties->validate();
}
std::vector<::shared_ptr<index_target>> create_index_statement::validate_while_executing(service::storage_proxy& proxy) const {
auto db = proxy.data_dictionary();
std::vector<::shared_ptr<index_target>> create_index_statement::validate_while_executing(query_processor& qp) const {
auto db = qp.db();
auto schema = validation::validate_column_family(db.real_database(), keyspace(), column_family());
if (schema->is_counter()) {
@@ -279,7 +279,7 @@ void create_index_statement::validate_targets_for_multi_column_index(std::vector
}
schema_ptr create_index_statement::build_index_schema(query_processor& qp) const {
auto targets = validate_while_executing(qp.proxy());
auto targets = validate_while_executing(qp);
data_dictionary::database db = qp.db();
auto schema = db.find_schema(keyspace(), column_family());

6
cql3/statements/create_index_statement.hh
View File

@@ -76,8 +76,8 @@ public:
std::vector<::shared_ptr<index_target::raw>> raw_targets,
::shared_ptr<index_prop_defs> properties, bool if_not_exists);
future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
void validate(service::storage_proxy&, const service::client_state& state) const override;
future<> check_access(query_processor& qp, const service::client_state& state) const override;
void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
@@ -94,7 +94,7 @@ private:
const sstring& name,
index_metadata_kind kind,
const index_options_map& options);
std::vector<::shared_ptr<index_target>> validate_while_executing(service::storage_proxy& proxy) const;
std::vector<::shared_ptr<index_target>> validate_while_executing(query_processor& qp) const;
schema_ptr build_index_schema(query_processor& qp) const;
};

15
cql3/statements/create_keyspace_statement.cc
View File

@@ -74,12 +74,12 @@ const sstring& create_keyspace_statement::keyspace() const
return _name;
}
future<> create_keyspace_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> create_keyspace_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_all_keyspaces_access(auth::permission::CREATE);
}
void create_keyspace_statement::validate(service::storage_proxy&, const service::client_state& state) const
void create_keyspace_statement::validate(query_processor&, const service::client_state& state) const
{
std::string name;
name.resize(_name.length());
@@ -115,8 +115,7 @@ void create_keyspace_statement::validate(service::storage_proxy&, const service:
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> create_keyspace_statement::prepare_schema_mutations(query_processor& qp) const {
using namespace cql_transport;
auto p = qp.proxy().shared_from_this();
const auto& tm = *p->get_token_metadata_ptr();
const auto& tm = *qp.proxy().get_token_metadata_ptr();
::shared_ptr<event::schema_change> ret;
std::vector<mutation> m;
@@ -161,7 +160,7 @@ future<> cql3::statements::create_keyspace_statement::grant_permissions_to_creat
// errors (such as unknown replication strategy name or unknown options
// to a known replication strategy) are done elsewhere.
std::optional<sstring> check_restricted_replication_strategy(
service::storage_proxy& proxy,
query_processor& qp,
const sstring& keyspace,
const ks_prop_defs& attrs)
{
@@ -174,7 +173,7 @@ std::optional<sstring> check_restricted_replication_strategy(
// may have in the future - multiple racks or DCs. So depending on how
// protective we are configured, let's prevent it or allow with a warning:
if (replication_strategy == "org.apache.cassandra.locator.SimpleStrategy") {
switch(proxy.data_dictionary().get_config().restrict_replication_simplestrategy()) {
switch(qp.db().get_config().restrict_replication_simplestrategy()) {
case db::tri_mode_restriction_t::mode::TRUE:
throw exceptions::configuration_exception(
"SimpleStrategy replication class is not recommended, and "
@@ -191,7 +190,7 @@ std::optional<sstring> check_restricted_replication_strategy(
case db::tri_mode_restriction_t::mode::FALSE:
// Scylla was configured to allow SimpleStrategy, but let's warn
// if it's used on a cluster which *already* has multiple DCs:
if (proxy.get_token_metadata_ptr()->get_topology().get_datacenter_endpoints().size() > 1) {
if (qp.proxy().get_token_metadata_ptr()->get_topology().get_datacenter_endpoints().size() > 1) {
return "Using SimpleStrategy in a multi-datacenter environment is not recommended.";
}
break;
@@ -202,7 +201,7 @@ std::optional<sstring> check_restricted_replication_strategy(
future<::shared_ptr<messages::result_message>>
create_keyspace_statement::execute(query_processor& qp, service::query_state& state, const query_options& options) const {
std::optional<sstring> warning = check_restricted_replication_strategy(qp.proxy(), keyspace(), *_attrs);
std::optional<sstring> warning = check_restricted_replication_strategy(qp, keyspace(), *_attrs);
return schema_altering_statement::execute(qp, state, options).then([this, warning = std::move(warning)] (::shared_ptr<messages::result_message> msg) {
if (warning) {
msg->add_warning(*warning);

6
cql3/statements/create_keyspace_statement.hh
View File

@@ -83,7 +83,7 @@ public:
virtual const sstring& keyspace() const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
/**
* The <code>CqlParser</code> only goes as far as extracting the keyword arguments
@@ -92,7 +92,7 @@ public:
*
* @throws InvalidRequestException if arguments are missing or unacceptable
*/
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
@@ -108,7 +108,7 @@ public:
};
std::optional<sstring> check_restricted_replication_strategy(
service::storage_proxy& proxy,
query_processor& qp,
const sstring& keyspace,
const ks_prop_defs& attrs);

4
cql3/statements/create_role_statement.hh
View File

@@ -72,9 +72,9 @@ public:
future<> grant_permissions_to_creator(const service::client_state&) const;
void validate(service::storage_proxy&, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

4
cql3/statements/create_service_level_statement.cc
View File

@@ -42,10 +42,10 @@ cql3::statements::create_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<create_service_level_statement>(*this));
}
void create_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void create_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> create_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> create_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::CREATE, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/create_service_level_statement.hh
View File

@@ -38,8 +38,8 @@ class create_service_level_statement final : public service_level_statement {
public:
create_service_level_statement(sstring service_level, shared_ptr<sl_prop_defs> attrs, bool if_not_exists);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

12
cql3/statements/create_table_statement.cc
View File

@@ -82,11 +82,11 @@ create_table_statement::create_table_statement(cf_name name,
{
}
future<> create_table_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
return state.has_keyspace_access(proxy.local_db(), keyspace(), auth::permission::CREATE);
future<> create_table_statement::check_access(query_processor& qp, const service::client_state& state) const {
return state.has_keyspace_access(qp.proxy().local_db(), keyspace(), auth::permission::CREATE);
}
void create_table_statement::validate(service::storage_proxy&, const service::client_state& state) const {
void create_table_statement::validate(query_processor&, const service::client_state& state) const {
// validated in announceMigration()
}
@@ -453,7 +453,7 @@ void create_table_statement::raw_statement::add_column_alias(::shared_ptr<column
// legal but restricted by the configuration. Checks for other of errors
// in the table's options are done elsewhere.
std::optional<sstring> check_restricted_table_properties(
service::storage_proxy& proxy,
query_processor& qp,
const sstring& keyspace, const sstring& table,
const cf_prop_defs& cfprops)
{
@@ -463,7 +463,7 @@ std::optional<sstring> check_restricted_table_properties(
// in prepare_schema_mutations(), in the middle of execute).
auto strategy = cfprops.get_compaction_strategy_class();
if (strategy && *strategy == sstables::compaction_strategy_type::date_tiered) {
switch(proxy.data_dictionary().get_config().restrict_dtcs()) {
switch(qp.db().get_config().restrict_dtcs()) {
case db::tri_mode_restriction_t::mode::TRUE:
throw exceptions::configuration_exception(
"DateTieredCompactionStrategy is deprecated, and "
@@ -485,7 +485,7 @@ std::optional<sstring> check_restricted_table_properties(
future<::shared_ptr<messages::result_message>>
create_table_statement::execute(query_processor& qp, service::query_state& state, const query_options& options) const {
std::optional<sstring> warning = check_restricted_table_properties(qp.proxy(), keyspace(), column_family(), *_properties);
std::optional<sstring> warning = check_restricted_table_properties(qp, keyspace(), column_family(), *_properties);
return schema_altering_statement::execute(qp, state, options).then([this, warning = std::move(warning)] (::shared_ptr<messages::result_message> msg) {
if (warning) {
msg->add_warning(*warning);

6
cql3/statements/create_table_statement.hh
View File

@@ -99,9 +99,9 @@ public:
column_set_type static_columns,
const std::optional<utils::UUID>& id);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
@@ -160,7 +160,7 @@ public:
};
std::optional<sstring> check_restricted_table_properties(
service::storage_proxy& proxy,
query_processor& qp,
const sstring& keyspace, const sstring& table,
const cf_prop_defs& cfprops);

6
cql3/statements/create_type_statement.cc
View File

@@ -72,9 +72,9 @@ void create_type_statement::add_definition(::shared_ptr<column_identifier> name,
_column_types.emplace_back(type);
}
future<> create_type_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> create_type_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_keyspace_access(proxy.local_db(), keyspace(), auth::permission::CREATE);
return state.has_keyspace_access(qp.proxy().local_db(), keyspace(), auth::permission::CREATE);
}
inline bool create_type_statement::type_exists_in(data_dictionary::keyspace ks) const
@@ -83,7 +83,7 @@ inline bool create_type_statement::type_exists_in(data_dictionary::keyspace ks)
return keyspace_types.contains(_name.get_user_type_name());
}
void create_type_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
void create_type_statement::validate(query_processor& qp, const service::client_state& state) const
{
if (_column_types.size() > max_udt_fields) {
throw exceptions::invalid_request_exception(format("A user type cannot have more than {} fields", max_udt_fields));

4
cql3/statements/create_type_statement.hh
View File

@@ -62,9 +62,9 @@ public:
void add_definition(::shared_ptr<column_identifier> name, ::shared_ptr<cql3_type::raw> type);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual const sstring& keyspace() const override;

6
cql3/statements/create_view_statement.cc
View File

@@ -91,11 +91,11 @@ create_view_statement::create_view_statement(
{
}
future<> create_view_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
return state.has_column_family_access(proxy.local_db(), keyspace(), _base_name.get_column_family(), auth::permission::ALTER);
future<> create_view_statement::check_access(query_processor& qp, const service::client_state& state) const {
return state.has_column_family_access(qp.proxy().local_db(), keyspace(), _base_name.get_column_family(), auth::permission::ALTER);
}
void create_view_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void create_view_statement::validate(query_processor& qp, const service::client_state& state) const {
}
static const column_definition* get_column_definition(const schema& schema, column_identifier::raw& identifier) {

4
cql3/statements/create_view_statement.hh
View File

@@ -68,8 +68,8 @@ public:
}
// Functions we need to override to subclass schema_altering_statement
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;

4
cql3/statements/detach_service_level_statement.cc
View File

@@ -40,10 +40,10 @@ cql3::statements::detach_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<detach_service_level_statement>(*this));
}
void detach_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void detach_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> detach_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> detach_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::AUTHORIZE, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/detach_service_level_statement.hh
View File

@@ -34,8 +34,8 @@ class detach_service_level_statement final : public service_level_statement {
public:
detach_service_level_statement(sstring role_name);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

2
cql3/statements/drop_aggregate_statement.cc
View File

@@ -41,7 +41,7 @@ drop_aggregate_statement::prepare_schema_mutations(query_processor& qp) const {
::shared_ptr<cql_transport::event::schema_change> ret;
std::vector<mutation> m;
auto func = validate_while_executing(qp.proxy());
auto func = validate_while_executing(qp);
if (func) {
auto user_aggr = dynamic_pointer_cast<functions::user_aggregate>(func);
if (!user_aggr) {

2
cql3/statements/drop_function_statement.cc
View File

@@ -41,7 +41,7 @@ drop_function_statement::prepare_schema_mutations(query_processor& qp) const {
::shared_ptr<cql_transport::event::schema_change> ret;
std::vector<mutation> m;
auto func = validate_while_executing(qp.proxy());
auto func = validate_while_executing(qp);
if (func) {
auto user_func = dynamic_pointer_cast<functions::user_function>(func);

16
cql3/statements/drop_index_statement.cc
View File

@@ -69,20 +69,20 @@ const sstring& drop_index_statement::column_family() const
cf_statement::column_family();
}
future<> drop_index_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> drop_index_statement::check_access(query_processor& qp, const service::client_state& state) const
{
auto cfm = lookup_indexed_table(proxy);
auto cfm = lookup_indexed_table(qp);
if (!cfm) {
return make_ready_future<>();
}
return state.has_column_family_access(proxy.local_db(), cfm->ks_name(), cfm->cf_name(), auth::permission::ALTER);
return state.has_column_family_access(qp.proxy().local_db(), cfm->ks_name(), cfm->cf_name(), auth::permission::ALTER);
}
void drop_index_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const
void drop_index_statement::validate(query_processor& qp, const service::client_state& state) const
{
// validated in lookup_indexed_table()
auto db = proxy.data_dictionary();
auto db = qp.db();
if (db.has_keyspace(keyspace())) {
auto schema = db.find_indexed_table(keyspace(), _index_name);
if (schema) {
@@ -92,7 +92,7 @@ void drop_index_statement::validate(service::storage_proxy& proxy, const service
}
schema_ptr drop_index_statement::make_drop_idex_schema(query_processor& qp) const {
auto cfm = lookup_indexed_table(qp.proxy());
auto cfm = lookup_indexed_table(qp);
if (!cfm) {
return nullptr;
}
@@ -128,9 +128,9 @@ drop_index_statement::prepare(data_dictionary::database db, cql_stats& stats) {
return std::make_unique<prepared_statement>(make_shared<drop_index_statement>(*this));
}
schema_ptr drop_index_statement::lookup_indexed_table(service::storage_proxy& proxy) const
schema_ptr drop_index_statement::lookup_indexed_table(query_processor& qp) const
{
auto& db = proxy.data_dictionary();
auto db = qp.db();
if (!db.has_keyspace(keyspace())) {
throw exceptions::keyspace_not_defined_exception(format("Keyspace {} does not exist", keyspace()));
}

6
cql3/statements/drop_index_statement.hh
View File

@@ -71,15 +71,15 @@ public:
virtual const sstring& column_family() const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
private:
schema_ptr lookup_indexed_table(service::storage_proxy& proxy) const;
schema_ptr lookup_indexed_table(query_processor& qp) const;
schema_ptr make_drop_idex_schema(query_processor& qp) const;
};

6
cql3/statements/drop_keyspace_statement.cc
View File

@@ -59,12 +59,12 @@ drop_keyspace_statement::drop_keyspace_statement(const sstring& keyspace, bool i
{
}
future<> drop_keyspace_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> drop_keyspace_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_keyspace_access(proxy.local_db(), keyspace(), auth::permission::DROP);
return state.has_keyspace_access(qp.proxy().local_db(), keyspace(), auth::permission::DROP);
}
void drop_keyspace_statement::validate(service::storage_proxy&, const service::client_state& state) const
void drop_keyspace_statement::validate(query_processor&, const service::client_state& state) const
{
warn(unimplemented::cause::VALIDATION);
#if 0

4
cql3/statements/drop_keyspace_statement.hh
View File

@@ -55,9 +55,9 @@ class drop_keyspace_statement : public schema_altering_statement {
public:
drop_keyspace_statement(const sstring& keyspace, bool if_exists);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual const sstring& keyspace() const override;

4
cql3/statements/drop_role_statement.hh
View File

@@ -64,9 +64,9 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual void validate(service::storage_proxy&, const service::client_state&) const override;
virtual void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

4
cql3/statements/drop_service_level_statement.cc
View File

@@ -40,10 +40,10 @@ cql3::statements::drop_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<drop_service_level_statement>(*this));
}
void drop_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void drop_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> drop_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> drop_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::DROP, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/drop_service_level_statement.hh
View File

@@ -35,8 +35,8 @@ class drop_service_level_statement final : public service_level_statement {
public:
drop_service_level_statement(sstring service_level, bool if_exists);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

6
cql3/statements/drop_table_statement.cc
View File

@@ -57,11 +57,11 @@ drop_table_statement::drop_table_statement(cf_name cf_name, bool if_exists)
{
}
future<> drop_table_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> drop_table_statement::check_access(query_processor& qp, const service::client_state& state) const
{
// invalid_request_exception is only thrown synchronously.
try {
return state.has_column_family_access(proxy.local_db(), keyspace(), column_family(), auth::permission::DROP);
return state.has_column_family_access(qp.proxy().local_db(), keyspace(), column_family(), auth::permission::DROP);
} catch (exceptions::invalid_request_exception&) {
if (!_if_exists) {
throw;
@@ -70,7 +70,7 @@ future<> drop_table_statement::check_access(service::storage_proxy& proxy, const
}
}
void drop_table_statement::validate(service::storage_proxy&, const service::client_state& state) const
void drop_table_statement::validate(query_processor&, const service::client_state& state) const
{
// validated in prepare_schema_mutations()
}

4
cql3/statements/drop_table_statement.hh
View File

@@ -55,9 +55,9 @@ class drop_table_statement : public schema_altering_statement {
public:
drop_table_statement(cf_name cf_name, bool if_exists);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;

12
cql3/statements/drop_type_statement.cc
View File

@@ -68,18 +68,18 @@ void drop_type_statement::prepare_keyspace(const service::client_state& state)
}
}
future<> drop_type_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> drop_type_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_keyspace_access(proxy.local_db(), keyspace(), auth::permission::DROP);
return state.has_keyspace_access(qp.proxy().local_db(), keyspace(), auth::permission::DROP);
}
void drop_type_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void drop_type_statement::validate(query_processor& qp, const service::client_state& state) const {
// validation is done at execution time
}
void drop_type_statement::validate_while_executing(service::storage_proxy& proxy) const {
void drop_type_statement::validate_while_executing(query_processor& qp) const {
try {
auto&& ks = proxy.data_dictionary().find_keyspace(keyspace());
auto&& ks = qp.db().find_keyspace(keyspace());
auto&& all_types = ks.metadata()->user_types().get_all_types();
auto old = all_types.find(_name.get_user_type_name());
if (old == all_types.end()) {
@@ -153,7 +153,7 @@ const sstring& drop_type_statement::keyspace() const
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>>
drop_type_statement::prepare_schema_mutations(query_processor& qp) const {
validate_while_executing(qp.proxy());
validate_while_executing(qp);
data_dictionary::database db = qp.db();

6
cql3/statements/drop_type_statement.hh
View File

@@ -56,9 +56,9 @@ public:
virtual void prepare_keyspace(const service::client_state& state) override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual const sstring& keyspace() const override;
@@ -67,7 +67,7 @@ public:
virtual std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
private:
void validate_while_executing(service::storage_proxy&) const;
void validate_while_executing(query_processor&) const;
};
}

6
cql3/statements/drop_view_statement.cc
View File

@@ -58,10 +58,10 @@ drop_view_statement::drop_view_statement(cf_name view_name, bool if_exists)
{
}
future<> drop_view_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> drop_view_statement::check_access(query_processor& qp, const service::client_state& state) const
{
try {
const data_dictionary::database db = proxy.data_dictionary();
const data_dictionary::database db = qp.db();
auto&& s = db.find_schema(keyspace(), column_family());
if (s->is_view()) {
return state.has_column_family_access(db.real_database(), keyspace(), s->view_info()->base_name(), auth::permission::ALTER);
@@ -72,7 +72,7 @@ future<> drop_view_statement::check_access(service::storage_proxy& proxy, const
return make_ready_future<>();
}
void drop_view_statement::validate(service::storage_proxy&, const service::client_state& state) const
void drop_view_statement::validate(query_processor&, const service::client_state& state) const
{
// validated in migration_manager::announce_view_drop()
}

4
cql3/statements/drop_view_statement.hh
View File

@@ -61,9 +61,9 @@ private:
public:
drop_view_statement(cf_name view_name, bool if_exists);
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
future<std::pair<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>>> prepare_schema_mutations(query_processor& qp) const override;

27
cql3/statements/function_statement.cc
View File

@@ -26,11 +26,12 @@
#include "data_dictionary/data_dictionary.hh"
#include "gms/feature_service.hh"
#include "service/storage_proxy.hh"
#include "cql3/query_processor.hh"
namespace cql3 {
namespace statements {
future<> function_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const { return make_ready_future<>(); }
future<> function_statement::check_access(query_processor& qp, const service::client_state& state) const { return make_ready_future<>(); }
function_statement::function_statement(
functions::function_name name, std::vector<shared_ptr<cql3_type::raw>> raw_arg_types)
@@ -49,11 +50,11 @@ shared_ptr<cql_transport::event::schema_change> function_statement::create_schem
return ::make_shared<event::schema_change>(change, type, func.name().keyspace, std::move(options));
}
data_type function_statement::prepare_type(service::storage_proxy& proxy, cql3_type::raw& t) const {
data_type function_statement::prepare_type(query_processor& qp, cql3_type::raw& t) const {
if (t.is_user_type() && t.is_frozen()) {
throw exceptions::invalid_request_exception("User defined argument and return types should not be frozen");
}
auto&& db = proxy.data_dictionary();
auto&& db = qp.db();
// At the CQL level the argument and return types should not have
// the frozen keyword.
// We and cassandra 3 support only frozen UDT arguments and
@@ -68,14 +69,14 @@ data_type function_statement::prepare_type(service::storage_proxy& proxy, cql3_t
return prepared;
}
void function_statement::create_arg_types(service::storage_proxy& proxy) const {
if (!proxy.features().cluster_supports_user_defined_functions()) {
void function_statement::create_arg_types(query_processor& qp) const {
if (!qp.proxy().features().cluster_supports_user_defined_functions()) {
throw exceptions::invalid_request_exception("User defined functions are disabled. Set enable_user_defined_functions and experimental_features:udf to enable them");
}
if (_arg_types.empty()) {
for (const auto& arg_type : _raw_arg_types) {
_arg_types.push_back(prepare_type(proxy, *arg_type));
_arg_types.push_back(prepare_type(qp, *arg_type));
}
}
}
@@ -90,15 +91,15 @@ create_function_statement_base::create_function_statement_base(functions::functi
std::vector<shared_ptr<cql3_type::raw>> raw_arg_types, bool or_replace, bool if_not_exists)
: function_statement(std::move(name), std::move(raw_arg_types)), _or_replace(or_replace), _if_not_exists(if_not_exists) {}
void create_function_statement_base::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void create_function_statement_base::validate(query_processor& qp, const service::client_state& state) const {
// validation happens during execution
}
shared_ptr<functions::function> create_function_statement_base::validate_while_executing(service::storage_proxy& proxy) const {
create_arg_types(proxy);
shared_ptr<functions::function> create_function_statement_base::validate_while_executing(query_processor& qp) const {
create_arg_types(qp);
auto old = functions::functions::find(_name, _arg_types);
if (!old || _or_replace) {
return create(proxy, old.get());
return create(qp, old.get());
}
if (!_if_not_exists) {
throw exceptions::invalid_request_exception(format("The function '{}' already exists", old));
@@ -110,12 +111,12 @@ drop_function_statement_base::drop_function_statement_base(functions::function_n
std::vector<shared_ptr<cql3_type::raw>> arg_types, bool args_present, bool if_exists)
: function_statement(std::move(name), std::move(arg_types)), _args_present(args_present), _if_exists(if_exists) {}
void drop_function_statement_base::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void drop_function_statement_base::validate(query_processor& qp, const service::client_state& state) const {
// validation happens during execution
}
shared_ptr<functions::function> drop_function_statement_base::validate_while_executing(service::storage_proxy& proxy) const {
create_arg_types(proxy);
shared_ptr<functions::function> drop_function_statement_base::validate_while_executing(query_processor& qp) const {
create_arg_types(qp);
shared_ptr<functions::function> func;
if (_args_present) {
func = functions::functions::find(_name, _arg_types);

18
cql3/statements/function_statement.hh
View File

@@ -35,7 +35,7 @@ namespace statements {
class function_statement : public schema_altering_statement {
protected:
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void prepare_keyspace(const service::client_state& state) override;
functions::function_name _name;
std::vector<shared_ptr<cql3_type::raw>> _raw_arg_types;
@@ -43,17 +43,17 @@ protected:
static shared_ptr<cql_transport::event::schema_change> create_schema_change(
const functions::function& func, bool created);
function_statement(functions::function_name name, std::vector<shared_ptr<cql3_type::raw>> raw_arg_types);
void create_arg_types(service::storage_proxy& proxy) const;
data_type prepare_type(service::storage_proxy& proxy, cql3_type::raw &t) const;
virtual shared_ptr<functions::function> validate_while_executing(service::storage_proxy&) const = 0;
void create_arg_types(query_processor& qp) const;
data_type prepare_type(query_processor& qp, cql3_type::raw &t) const;
virtual shared_ptr<functions::function> validate_while_executing(query_processor&) const = 0;
};
// common logic for creating UDF and UDA
class create_function_statement_base : public function_statement {
protected:
virtual void validate(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual shared_ptr<functions::function> create(service::storage_proxy& proxy, functions::function* old) const = 0;
virtual shared_ptr<functions::function> validate_while_executing(service::storage_proxy&) const override;
virtual void validate(query_processor& qp, const service::client_state& state) const override;
virtual shared_ptr<functions::function> create(query_processor& qp, functions::function* old) const = 0;
virtual shared_ptr<functions::function> validate_while_executing(query_processor&) const override;
bool _or_replace;
bool _if_not_exists;
@@ -65,8 +65,8 @@ protected:
// common logic for dropping UDF and UDA
class drop_function_statement_base : public function_statement {
protected:
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual shared_ptr<functions::function> validate_while_executing(service::storage_proxy&) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual shared_ptr<functions::function> validate_while_executing(query_processor&) const override;
bool _args_present;
bool _if_exists;

2
cql3/statements/grant_role_statement.hh
View File

@@ -65,7 +65,7 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

4
cql3/statements/list_permissions_statement.cc
View File

@@ -65,13 +65,13 @@ std::unique_ptr<cql3::statements::prepared_statement> cql3::statements::list_per
}
void cql3::statements::list_permissions_statement::validate(
service::storage_proxy& proxy,
query_processor& qp,
const service::client_state& state) const {
// a check to ensure the existence of the user isn't being leaked by user existence check.
state.ensure_not_anonymous();
}
future<> cql3::statements::list_permissions_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> cql3::statements::list_permissions_statement::check_access(query_processor& qp, const service::client_state& state) const {
if (_resource) {
maybe_correct_resource(*_resource, state);
return state.ensure_exists(*_resource);

4
cql3/statements/list_permissions_statement.hh
View File

@@ -65,9 +65,9 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
future<> check_access(query_processor& qp, const service::client_state&) const override;
future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state& , const query_options&) const override;

2
cql3/statements/list_roles_statement.hh
View File

@@ -66,7 +66,7 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

4
cql3/statements/list_service_level_attachments_statement.cc
View File

@@ -45,10 +45,10 @@ cql3::statements::list_service_level_attachments_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<list_service_level_attachments_statement>(*this));
}
void list_service_level_attachments_statement::validate(service::storage_proxy &, const service::client_state &) const {
void list_service_level_attachments_statement::validate(query_processor &, const service::client_state &) const {
}
future<> list_service_level_attachments_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> list_service_level_attachments_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::DESCRIBE, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/list_service_level_attachments_statement.hh
View File

@@ -36,8 +36,8 @@ public:
list_service_level_attachments_statement(sstring role_name);
list_service_level_attachments_statement();
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

4
cql3/statements/list_service_level_statement.cc
View File

@@ -42,10 +42,10 @@ cql3::statements::list_service_level_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<list_service_level_statement>(*this));
}
void list_service_level_statement::validate(service::storage_proxy &, const service::client_state &) const {
void list_service_level_statement::validate(query_processor &, const service::client_state &) const {
}
future<> list_service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> list_service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return state.ensure_has_permission(auth::command_desc{.permission = auth::permission::DESCRIBE, .resource = auth::root_service_level_resource()});
}

4
cql3/statements/list_service_level_statement.hh
View File

@@ -35,8 +35,8 @@ class list_service_level_statement final : public service_level_statement {
public:
list_service_level_statement(sstring service_level, bool describe_all);
std::unique_ptr<cql3::statements::prepared_statement> prepare(data_dictionary::database db, cql_stats &stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
virtual future<> check_access(service::storage_proxy& sp, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;
};

4
cql3/statements/list_users_statement.cc
View File

@@ -51,10 +51,10 @@ std::unique_ptr<cql3::statements::prepared_statement> cql3::statements::list_use
return std::make_unique<prepared_statement>(::make_shared<list_users_statement>(*this));
}
void cql3::statements::list_users_statement::validate(service::storage_proxy& proxy, const service::client_state& state) const {
void cql3::statements::list_users_statement::validate(query_processor& qp, const service::client_state& state) const {
}
future<> cql3::statements::list_users_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> cql3::statements::list_users_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return make_ready_future();
}

4
cql3/statements/list_users_statement.hh
View File

@@ -54,8 +54,8 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
void validate(service::storage_proxy&, const service::client_state&) const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
future<> check_access(query_processor& qp, const service::client_state&) const override;
future<::shared_ptr<cql_transport::messages::result_message>> execute(query_processor&
, service::query_state&
, const query_options&) const override;

42
cql3/statements/modification_statement.cc
View File

@@ -125,8 +125,8 @@ gc_clock::duration modification_statement::get_time_to_live(const query_options&
return gc_clock::duration(attrs->get_time_to_live(options));
}
future<> modification_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
const data_dictionary::database db = proxy.data_dictionary();
future<> modification_statement::check_access(query_processor& qp, const service::client_state& state) const {
const data_dictionary::database db = qp.db();
auto f = state.has_column_family_access(db.real_database(), keyspace(), column_family(), auth::permission::MODIFY);
if (has_conditions()) {
f = f.then([this, &state, db] {
@@ -137,7 +137,7 @@ future<> modification_statement::check_access(service::storage_proxy& proxy, con
}
future<std::vector<mutation>>
modification_statement::get_mutations(service::storage_proxy& proxy, const query_options& options, db::timeout_clock::time_point timeout, bool local, int64_t now, service::query_state& qs) const {
modification_statement::get_mutations(query_processor& qp, const query_options& options, db::timeout_clock::time_point timeout, bool local, int64_t now, service::query_state& qs) const {
if (_restrictions->range_or_slice_eq_null(options)) { // See #7852 and #9290.
throw exceptions::invalid_request_exception("Invalid null value in condition for a key column");
}
@@ -154,9 +154,9 @@ modification_statement::get_mutations(service::storage_proxy& proxy, const query
}
if (requires_read()) {
lw_shared_ptr<query::read_command> cmd = read_command(proxy, ranges, cl);
lw_shared_ptr<query::read_command> cmd = read_command(qp, ranges, cl);
// FIXME: ignoring "local"
f = proxy.query(s, cmd, dht::partition_range_vector(keys), cl,
f = qp.proxy().query(s, cmd, dht::partition_range_vector(keys), cl,
{timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()}).then(
[this, cmd] (auto cqr) {
@@ -232,14 +232,14 @@ std::vector<mutation> modification_statement::apply_updates(
}
lw_shared_ptr<query::read_command>
modification_statement::read_command(service::storage_proxy& proxy, query::clustering_row_ranges ranges, db::consistency_level cl) const {
modification_statement::read_command(query_processor& qp, query::clustering_row_ranges ranges, db::consistency_level cl) const {
try {
validate_for_read(cl);
} catch (exceptions::invalid_request_exception& e) {
throw exceptions::invalid_request_exception(format("Write operation require a read but consistency {} is not supported on reads", cl));
}
query::partition_slice ps(std::move(ranges), *s, columns_to_read(), update_parameters::options);
const auto max_result_size = proxy.get_max_result_size(ps);
const auto max_result_size = qp.proxy().get_max_result_size(ps);
return make_lw_shared<query::read_command>(s->id(), s->version(), std::move(ps), query::max_result_size(max_result_size));
}
@@ -263,19 +263,18 @@ struct modification_statement_executor {
static thread_local inheriting_concrete_execution_stage<
future<::shared_ptr<cql_transport::messages::result_message>>,
const modification_statement*,
service::storage_proxy&,
query_processor&,
service::query_state&,
const query_options&> modify_stage{"cql3_modification", modification_statement_executor::get()};
future<::shared_ptr<cql_transport::messages::result_message>>
modification_statement::execute(query_processor& qp, service::query_state& qs, const query_options& options) const {
cql3::util::validate_timestamp(options, attrs);
service::storage_proxy& proxy = qp.proxy();
return modify_stage(this, seastar::ref(proxy), seastar::ref(qs), seastar::cref(options));
return modify_stage(this, seastar::ref(qp), seastar::ref(qs), seastar::cref(options));
}
future<::shared_ptr<cql_transport::messages::result_message>>
modification_statement::do_execute(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const {
modification_statement::do_execute(query_processor& qp, service::query_state& qs, const query_options& options) const {
if (has_conditions() && options.get_protocol_version() == 1) {
throw exceptions::invalid_request_exception("Conditional updates are not supported by the protocol version in use. You need to upgrade to a driver using the native protocol v2.");
}
@@ -285,30 +284,30 @@ modification_statement::do_execute(service::storage_proxy& proxy, service::query
inc_cql_stats(qs.get_client_state().is_internal());
if (has_conditions()) {
return execute_with_condition(proxy, qs, options);
return execute_with_condition(qp, qs, options);
}
return execute_without_condition(proxy, qs, options).then([] {
return execute_without_condition(qp, qs, options).then([] {
return make_ready_future<::shared_ptr<cql_transport::messages::result_message>>(
::shared_ptr<cql_transport::messages::result_message>{});
});
}
future<>
modification_statement::execute_without_condition(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const {
modification_statement::execute_without_condition(query_processor& qp, service::query_state& qs, const query_options& options) const {
auto cl = options.get_consistency();
auto timeout = db::timeout_clock::now() + get_timeout(qs.get_client_state(), options);
return get_mutations(proxy, options, timeout, false, options.get_timestamp(qs), qs).then([this, cl, timeout, &proxy, &qs] (auto mutations) {
return get_mutations(qp, options, timeout, false, options.get_timestamp(qs), qs).then([this, cl, timeout, &qp, &qs] (auto mutations) {
if (mutations.empty()) {
return now();
}
return proxy.mutate_with_triggers(std::move(mutations), cl, timeout, false, qs.get_trace_state(), qs.get_permit(), this->is_raw_counter_shard_write());
return qp.proxy().mutate_with_triggers(std::move(mutations), cl, timeout, false, qs.get_trace_state(), qs.get_permit(), this->is_raw_counter_shard_write());
});
}
future<::shared_ptr<cql_transport::messages::result_message>>
modification_statement::execute_with_condition(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const {
modification_statement::execute_with_condition(query_processor& qp, service::query_state& qs, const query_options& options) const {
auto cl_for_learn = options.get_consistency();
auto cl_for_paxos = options.check_serial_consistency();
@@ -335,13 +334,12 @@ modification_statement::execute_with_condition(service::storage_proxy& proxy, se
auto shard = service::storage_proxy::cas_shard(*s, request->key()[0].start()->value().as_decorated_key().token());
if (shard != this_shard_id()) {
proxy.get_stats().replica_cross_shard_ops++;
return make_ready_future<shared_ptr<cql_transport::messages::result_message>>(
::make_shared<cql_transport::messages::result_message::bounce_to_shard>(shard,
std::move(const_cast<cql3::query_options&>(options).take_cached_pk_function_calls())));
qp.bounce_to_shard(shard, std::move(const_cast<cql3::query_options&>(options).take_cached_pk_function_calls()))
);
}
return proxy.cas(s, request, request->read_command(proxy), request->key(),
return qp.proxy().cas(s, request, request->read_command(qp), request->key(),
{read_timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()},
cl_for_paxos, cl_for_learn, statement_timeout, cas_timeout).then([this, request] (bool is_applied) {
return request->build_cas_result_set(_metadata, _columns_of_cas_result_set, is_applied);
@@ -551,7 +549,7 @@ modification_statement::prepare_conditions(data_dictionary::database db, const s
} // namespace raw
void
modification_statement::validate(service::storage_proxy&, const service::client_state& state) const {
modification_statement::validate(query_processor&, const service::client_state& state) const {
if (has_conditions() && attrs->is_timestamp_set()) {
throw exceptions::invalid_request_exception("Cannot provide custom timestamp for conditional updates");
}

14
cql3/statements/modification_statement.hh
View File

@@ -160,10 +160,10 @@ public:
gc_clock::duration get_time_to_live(const query_options& options) const;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
// Validate before execute, using client state and current schema
void validate(service::storage_proxy&, const service::client_state& state) const override;
void validate(query_processor&, const service::client_state& state) const override;
virtual bool depends_on_keyspace(const sstring& ks_name) const override;
@@ -228,7 +228,7 @@ public:
// Build a read_command instance to fetch the previous mutation from storage. The mutation is
// fetched if we need to check LWT conditions or apply updates to non-frozen list elements.
lw_shared_ptr<query::read_command> read_command(service::storage_proxy& proxy, query::clustering_row_ranges ranges, db::consistency_level cl) const;
lw_shared_ptr<query::read_command> read_command(query_processor& qp, query::clustering_row_ranges ranges, db::consistency_level cl) const;
// Create a mutation object for the update operation represented by this modification statement.
// A single mutation object for lightweight transactions, which can only span one partition, or a vector
// of mutations, one per partition key, for statements which affect multiple partition keys,
@@ -251,7 +251,7 @@ public:
private:
future<::shared_ptr<cql_transport::messages::result_message>>
do_execute(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const;
do_execute(query_processor& qp, service::query_state& qs, const query_options& options) const;
friend class modification_statement_executor;
public:
// True if the statement has IF conditions. Pre-computed during prepare.
@@ -266,10 +266,10 @@ public:
private:
future<>
execute_without_condition(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const;
execute_without_condition(query_processor& qp, service::query_state& qs, const query_options& options) const;
future<::shared_ptr<cql_transport::messages::result_message>>
execute_with_condition(service::storage_proxy& proxy, service::query_state& qs, const query_options& options) const;
execute_with_condition(query_processor& qp, service::query_state& qs, const query_options& options) const;
public:
/**
@@ -282,7 +282,7 @@ public:
* @return vector of the mutations
* @throws invalid_request_exception on invalid requests
*/
future<std::vector<mutation>> get_mutations(service::storage_proxy& proxy, const query_options& options, db::timeout_clock::time_point timeout, bool local, int64_t now, service::query_state& qs) const;
future<std::vector<mutation>> get_mutations(query_processor& qp, const query_options& options, db::timeout_clock::time_point timeout, bool local, int64_t now, service::query_state& qs) const;
virtual json_cache_opt maybe_prepare_json_cache(const query_options& options) const;
protected:

4
cql3/statements/permission_altering_statement.cc
View File

@@ -68,10 +68,10 @@ cql3::statements::permission_altering_statement::permission_altering_statement(
, _role_name(rn.to_string()) {
}
void cql3::statements::permission_altering_statement::validate(service::storage_proxy&, const service::client_state&) const {
void cql3::statements::permission_altering_statement::validate(query_processor&, const service::client_state&) const {
}
future<> cql3::statements::permission_altering_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> cql3::statements::permission_altering_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
maybe_correct_resource(_resource, state);

4
cql3/statements/permission_altering_statement.hh
View File

@@ -60,8 +60,8 @@ protected:
public:
permission_altering_statement(auth::permission_set, auth::resource, const cql3::role_name&);
void validate(service::storage_proxy&, const service::client_state&) const override;
future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
void validate(query_processor&, const service::client_state&) const override;
future<> check_access(query_processor& qp, const service::client_state&) const override;
};
}

2
cql3/statements/revoke_role_statement.hh
View File

@@ -65,7 +65,7 @@ public:
std::unique_ptr<prepared_statement> prepare(data_dictionary::database db, cql_stats& stats) override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state&) const override;
virtual future<> check_access(query_processor& qp, const service::client_state&) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor&, service::query_state&, const query_options&) const override;

26
cql3/statements/role-management-statements.cc
View File

@@ -83,7 +83,7 @@ static future<result_message_ptr> void_result_message() {
return make_ready_future<result_message_ptr>(nullptr);
}
void validate_cluster_support(service::storage_proxy&) {
void validate_cluster_support(query_processor& qp) {
}
//
@@ -106,11 +106,11 @@ future<> create_role_statement::grant_permissions_to_creator(const service::clie
});
}
void create_role_statement::validate(service::storage_proxy& p, const service::client_state&) const {
validate_cluster_support(p);
void create_role_statement::validate(query_processor& qp, const service::client_state&) const {
validate_cluster_support(qp);
}
future<> create_role_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> create_role_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return async([this, &state] {
@@ -164,11 +164,11 @@ std::unique_ptr<prepared_statement> alter_role_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<alter_role_statement>(*this));
}
void alter_role_statement::validate(service::storage_proxy& p, const service::client_state&) const {
validate_cluster_support(p);
void alter_role_statement::validate(query_processor& qp, const service::client_state&) const {
validate_cluster_support(qp);
}
future<> alter_role_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> alter_role_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return async([this, &state] {
@@ -245,15 +245,15 @@ std::unique_ptr<prepared_statement> drop_role_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<drop_role_statement>(*this));
}
void drop_role_statement::validate(service::storage_proxy& p, const service::client_state& state) const {
validate_cluster_support(p);
void drop_role_statement::validate(query_processor& qp, const service::client_state& state) const {
validate_cluster_support(qp);
if (*state.user() == auth::authenticated_user(_role)) {
throw request_validations::invalid_request("Cannot DROP primary role for current login.");
}
}
future<> drop_role_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> drop_role_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return async([this, &state] {
@@ -302,7 +302,7 @@ std::unique_ptr<prepared_statement> list_roles_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<list_roles_statement>(*this));
}
future<> list_roles_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> list_roles_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return async([this, &state] {
@@ -437,7 +437,7 @@ std::unique_ptr<prepared_statement> grant_role_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<grant_role_statement>(*this));
}
future<> grant_role_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> grant_role_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return do_with(auth::make_role_resource(_role), [this, &state](const auto& r) {
@@ -465,7 +465,7 @@ std::unique_ptr<prepared_statement> revoke_role_statement::prepare(
return std::make_unique<prepared_statement>(::make_shared<revoke_role_statement>(*this));
}
future<> revoke_role_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> revoke_role_statement::check_access(query_processor& qp, const service::client_state& state) const {
state.ensure_not_anonymous();
return do_with(auth::make_role_resource(_role), [this, &state](const auto& r) {

108
cql3/statements/select_statement.cc
View File

@@ -179,9 +179,9 @@ uint32_t select_statement::get_bound_terms() const {
return _bound_terms;
}
future<> select_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const {
future<> select_statement::check_access(query_processor& qp, const service::client_state& state) const {
try {
const data_dictionary::database db = proxy.data_dictionary();
const data_dictionary::database db = qp.db();
auto&& s = db.find_schema(keyspace(), column_family());
auto& cf_name = s->is_view() ? s->view_info()->base_name() : column_family();
return state.has_column_family_access(db.real_database(), keyspace(), cf_name, auth::permission::SELECT);
@@ -191,7 +191,7 @@ future<> select_statement::check_access(service::storage_proxy& proxy, const ser
}
}
void select_statement::validate(service::storage_proxy&, const service::client_state& state) const {
void select_statement::validate(query_processor&, const service::client_state& state) const {
// Nothing to do, all validation has been done by raw_statemet::prepare()
}
@@ -289,7 +289,7 @@ struct select_statement_executor {
static thread_local inheriting_concrete_execution_stage<
future<shared_ptr<cql_transport::messages::result_message>>,
const select_statement*,
service::storage_proxy&,
query_processor&,
service::query_state&,
const query_options&> select_stage{"cql3_select", select_statement_executor::get()};
@@ -298,12 +298,11 @@ select_statement::execute(query_processor& qp,
service::query_state& state,
const query_options& options) const
{
service::storage_proxy& proxy = qp.proxy();
return select_stage(this, seastar::ref(proxy), seastar::ref(state), seastar::cref(options));
return select_stage(this, seastar::ref(qp), seastar::ref(state), seastar::cref(options));
}
future<shared_ptr<cql_transport::messages::result_message>>
select_statement::do_execute(service::storage_proxy& proxy,
select_statement::do_execute(query_processor& qp,
service::query_state& state,
const query_options& options) const
{
@@ -328,7 +327,7 @@ select_statement::do_execute(service::storage_proxy& proxy,
_stats.select_partition_range_scan_no_bypass_cache += _range_scan_no_bypass_cache;
auto slice = make_partition_slice(options);
auto max_result_size = proxy.get_max_result_size(slice);
auto max_result_size = qp.proxy().get_max_result_size(slice);
auto command = ::make_lw_shared<query::read_command>(
_schema->id(),
_schema->version(),
@@ -365,17 +364,16 @@ select_statement::do_execute(service::storage_proxy& proxy,
}
unsigned shard = dht::shard_of(*_schema, key_ranges[0].start()->value().as_decorated_key().token());
if (this_shard_id() != shard) {
proxy.get_stats().replica_cross_shard_ops++;
return make_ready_future<shared_ptr<cql_transport::messages::result_message>>(
::make_shared<cql_transport::messages::result_message::bounce_to_shard>(shard,
std::move(const_cast<cql3::query_options&>(options).take_cached_pk_function_calls())));
qp.bounce_to_shard(shard, std::move(const_cast<cql3::query_options&>(options).take_cached_pk_function_calls()))
);
}
}
if (!aggregate && !_restrictions_need_filtering && (page_size <= 0
|| !service::pager::query_pagers::may_need_paging(*_schema, page_size,
*command, key_ranges))) {
return execute(proxy, command, std::move(key_ranges), state, options, now);
return execute(qp, command, std::move(key_ranges), state, options, now);
}
command->slice.options.set<query::partition_slice::option::allow_short_read>();
@@ -491,7 +489,7 @@ generate_base_key_from_index_pk(const partition_key& index_pk, const std::option
}
lw_shared_ptr<query::read_command>
indexed_table_select_statement::prepare_command_for_base_query(service::storage_proxy& proxy, const query_options& options,
indexed_table_select_statement::prepare_command_for_base_query(query_processor& qp, const query_options& options,
service::query_state& state, gc_clock::time_point now, bool use_paging) const {
auto slice = make_partition_slice(options);
if (use_paging) {
@@ -505,7 +503,7 @@ indexed_table_select_statement::prepare_command_for_base_query(service::storage_
_schema->id(),
_schema->version(),
std::move(slice),
proxy.get_max_result_size(slice),
qp.proxy().get_max_result_size(slice),
query::row_limit(get_limit(options)),
query::partition_limit(query::max_partitions),
now,
@@ -518,17 +516,17 @@ indexed_table_select_statement::prepare_command_for_base_query(service::storage_
future<std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>>
indexed_table_select_statement::do_execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
dht::partition_range_vector&& partition_ranges,
service::query_state& state,
const query_options& options,
gc_clock::time_point now,
lw_shared_ptr<const service::pager::paging_state> paging_state) const {
using value_type = std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>;
auto cmd = prepare_command_for_base_query(proxy, options, state, now, bool(paging_state));
auto cmd = prepare_command_for_base_query(qp, options, state, now, bool(paging_state));
auto timeout = db::timeout_clock::now() + get_timeout(state.get_client_state(), options);
uint32_t queried_ranges_count = partition_ranges.size();
service::query_ranges_to_vnodes_generator ranges_to_vnodes(proxy.get_token_metadata_ptr(), _schema, std::move(partition_ranges));
service::query_ranges_to_vnodes_generator ranges_to_vnodes(qp.proxy().get_token_metadata_ptr(), _schema, std::move(partition_ranges));
struct base_query_state {
query::result_merger merger;
@@ -545,12 +543,12 @@ indexed_table_select_statement::do_execute_base_query(
const bool is_paged = bool(paging_state);
base_query_state query_state{cmd->get_row_limit() * queried_ranges_count, std::move(ranges_to_vnodes)};
return do_with(std::move(query_state), [this, is_paged, &proxy, &state, &options, cmd, timeout] (auto&& query_state) {
return do_with(std::move(query_state), [this, is_paged, &qp, &state, &options, cmd, timeout] (auto&& query_state) {
auto& merger = query_state.merger;
auto& ranges_to_vnodes = query_state.ranges_to_vnodes;
auto& concurrency = query_state.concurrency;
auto& previous_result_size = query_state.previous_result_size;
return repeat([this, is_paged, &previous_result_size, &ranges_to_vnodes, &merger, &proxy, &state, &options, &concurrency, cmd, timeout]() {
return repeat([this, is_paged, &previous_result_size, &ranges_to_vnodes, &merger, &qp, &state, &options, &concurrency, cmd, timeout]() {
// Starting with 1 range, we check if the result was a short read, and if not,
// we continue exponentially, asking for 2x more ranges than before
dht::partition_range_vector prange = ranges_to_vnodes(concurrency);
@@ -587,7 +585,7 @@ indexed_table_select_statement::do_execute_base_query(
if (previous_result_size < query::result_memory_limiter::maximum_result_size && concurrency < max_base_table_query_concurrency) {
concurrency *= 2;
}
return proxy.query(_schema, command, std::move(prange), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
return qp.proxy().query(_schema, command, std::move(prange), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
.then([is_paged, &previous_result_size, &ranges_to_vnodes, &merger] (service::storage_proxy::coordinator_query_result qr) {
auto is_short_read = qr.query_result->is_short_read();
// Results larger than 1MB should be shipped to the client immediately
@@ -606,13 +604,13 @@ indexed_table_select_statement::do_execute_base_query(
future<shared_ptr<cql_transport::messages::result_message>>
indexed_table_select_statement::execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
dht::partition_range_vector&& partition_ranges,
service::query_state& state,
const query_options& options,
gc_clock::time_point now,
lw_shared_ptr<const service::pager::paging_state> paging_state) const {
return do_execute_base_query(proxy, std::move(partition_ranges), state, options, now, paging_state).then_unpack(
return do_execute_base_query(qp, std::move(partition_ranges), state, options, now, paging_state).then_unpack(
[this, &state, &options, now, paging_state = std::move(paging_state)] (foreign_ptr<lw_shared_ptr<query::result>> result, lw_shared_ptr<query::read_command> cmd) {
return process_base_query_results(std::move(result), std::move(cmd), state, options, now, std::move(paging_state));
});
@@ -620,14 +618,14 @@ indexed_table_select_statement::execute_base_query(
future<std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>>
indexed_table_select_statement::do_execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
std::vector<primary_key>&& primary_keys,
service::query_state& state,
const query_options& options,
gc_clock::time_point now,
lw_shared_ptr<const service::pager::paging_state> paging_state) const {
using value_type = std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>;
auto cmd = prepare_command_for_base_query(proxy, options, state, now, bool(paging_state));
auto cmd = prepare_command_for_base_query(qp, options, state, now, bool(paging_state));
auto timeout = db::timeout_clock::now() + get_timeout(state.get_client_state(), options);
struct base_query_state {
@@ -647,13 +645,13 @@ indexed_table_select_statement::do_execute_base_query(
base_query_state query_state{cmd->get_row_limit(), std::move(primary_keys)};
const bool is_paged = bool(paging_state);
return do_with(std::move(query_state), [this, is_paged, &proxy, &state, &options, cmd, timeout] (auto&& query_state) {
return do_with(std::move(query_state), [this, is_paged, &qp, &state, &options, cmd, timeout] (auto&& query_state) {
auto &merger = query_state.merger;
auto &keys = query_state.primary_keys;
auto &key_it = query_state.current_primary_key;
auto &previous_result_size = query_state.previous_result_size;
auto &next_iteration_size = query_state.next_iteration_size;
return repeat([this, is_paged, &previous_result_size, &next_iteration_size, &keys, &key_it, &merger, &proxy, &state, &options, cmd, timeout]() {
return repeat([this, is_paged, &previous_result_size, &next_iteration_size, &keys, &key_it, &merger, &qp, &state, &options, cmd, timeout]() {
// Starting with 1 key, we check if the result was a short read, and if not,
// we continue exponentially, asking for 2x more key than before
auto already_done = std::distance(keys.begin(), key_it);
@@ -667,7 +665,7 @@ indexed_table_select_statement::do_execute_base_query(
auto command = ::make_lw_shared<query::read_command>(*cmd);
query::result_merger oneshot_merger(cmd->get_row_limit(), query::max_partitions);
return map_reduce(key_it, key_it_end, [this, &proxy, &state, &options, cmd, timeout] (auto& key) {
return map_reduce(key_it, key_it_end, [this, &qp, &state, &options, cmd, timeout] (auto& key) {
auto command = ::make_lw_shared<query::read_command>(*cmd);
// for each partition, read just one clustering row (TODO: can
// get all needed rows of one partition at once.)
@@ -675,7 +673,7 @@ indexed_table_select_statement::do_execute_base_query(
if (key.clustering) {
command->slice._row_ranges.push_back(query::clustering_range::make_singular(key.clustering));
}
return proxy.query(_schema, command, {dht::partition_range::make_singular(key.partition)}, options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
return qp.proxy().query(_schema, command, {dht::partition_range::make_singular(key.partition)}, options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
.then([] (service::storage_proxy::coordinator_query_result qr) {
return std::move(qr.query_result);
});
@@ -698,20 +696,20 @@ indexed_table_select_statement::do_execute_base_query(
future<shared_ptr<cql_transport::messages::result_message>>
indexed_table_select_statement::execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
std::vector<primary_key>&& primary_keys,
service::query_state& state,
const query_options& options,
gc_clock::time_point now,
lw_shared_ptr<const service::pager::paging_state> paging_state) const {
return do_execute_base_query(proxy, std::move(primary_keys), state, options, now, paging_state).then_unpack(
return do_execute_base_query(qp, std::move(primary_keys), state, options, now, paging_state).then_unpack(
[this, &state, &options, now, paging_state = std::move(paging_state)] (foreign_ptr<lw_shared_ptr<query::result>> result, lw_shared_ptr<query::read_command> cmd) {
return process_base_query_results(std::move(result), std::move(cmd), state, options, now, std::move(paging_state));
});
}
future<shared_ptr<cql_transport::messages::result_message>>
select_statement::execute(service::storage_proxy& proxy,
select_statement::execute(query_processor& qp,
lw_shared_ptr<query::read_command> cmd,
dht::partition_range_vector&& partition_ranges,
service::query_state& state,
@@ -724,13 +722,13 @@ select_statement::execute(service::storage_proxy& proxy,
// doing post-query ordering.
auto timeout = db::timeout_clock::now() + get_timeout(state.get_client_state(), options);
if (needs_post_query_ordering() && _limit) {
return do_with(std::forward<dht::partition_range_vector>(partition_ranges), [this, &proxy, &state, &options, cmd, timeout](auto& prs) {
return do_with(std::forward<dht::partition_range_vector>(partition_ranges), [this, &qp, &state, &options, cmd, timeout](auto& prs) {
assert(cmd->partition_limit == query::max_partitions);
query::result_merger merger(cmd->get_row_limit() * prs.size(), query::max_partitions);
return map_reduce(prs.begin(), prs.end(), [this, &proxy, &state, &options, cmd, timeout] (auto& pr) {
return map_reduce(prs.begin(), prs.end(), [this, &qp, &state, &options, cmd, timeout] (auto& pr) {
dht::partition_range_vector prange { pr };
auto command = ::make_lw_shared<query::read_command>(*cmd);
return proxy.query(_schema,
return qp.proxy().query(_schema,
command,
std::move(prange),
options.get_consistency(),
@@ -742,7 +740,7 @@ select_statement::execute(service::storage_proxy& proxy,
return this->process_results(std::move(result), cmd, options, now);
});
} else {
return proxy.query(_schema, cmd, std::move(partition_ranges), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
return qp.proxy().query(_schema, cmd, std::move(partition_ranges), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
.then([this, &options, now, cmd] (service::storage_proxy::coordinator_query_result qr) {
return this->process_results(std::move(qr.query_result), cmd, options, now);
});
@@ -1003,7 +1001,7 @@ lw_shared_ptr<const service::pager::paging_state> indexed_table_select_statement
}
future<shared_ptr<cql_transport::messages::result_message>>
indexed_table_select_statement::do_execute(service::storage_proxy& proxy,
indexed_table_select_statement::do_execute(query_processor& qp,
service::query_state& state,
const query_options& options) const
{
@@ -1069,10 +1067,10 @@ indexed_table_select_statement::do_execute(service::storage_proxy& proxy,
const bool aggregate = _selection->is_aggregate() || has_group_by();
if (aggregate) {
return do_with(cql3::selection::result_set_builder(*_selection, now, options.get_cql_serialization_format(), *_group_by_cell_indices), std::make_unique<cql3::query_options>(cql3::query_options(options)),
[this, &options, &proxy, &state, now, whole_partitions, partition_slices] (cql3::selection::result_set_builder& builder, std::unique_ptr<cql3::query_options>& internal_options) {
[this, &options, &qp, &state, now, whole_partitions, partition_slices] (cql3::selection::result_set_builder& builder, std::unique_ptr<cql3::query_options>& internal_options) {
// page size is set to the internal count page size, regardless of the user-provided value
internal_options.reset(new cql3::query_options(std::move(internal_options), options.get_paging_state(), internal_paging_size));
return repeat([this, &builder, &options, &internal_options, &proxy, &state, now, whole_partitions, partition_slices] () {
return repeat([this, &builder, &options, &internal_options, &qp, &state, now, whole_partitions, partition_slices] () {
auto consume_results = [this, &builder, &options, &internal_options, &state] (foreign_ptr<lw_shared_ptr<query::result>> results, lw_shared_ptr<query::read_command> cmd, lw_shared_ptr<const service::pager::paging_state> paging_state) {
if (paging_state) {
paging_state = generate_view_paging_state_from_base_query_results(paging_state, results, state, options);
@@ -1090,17 +1088,17 @@ indexed_table_select_statement::do_execute(service::storage_proxy& proxy,
};
if (whole_partitions || partition_slices) {
return find_index_partition_ranges(proxy, state, *internal_options).then_unpack(
[this, now, &state, &internal_options, &proxy, consume_results = std::move(consume_results)] (dht::partition_range_vector partition_ranges, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return do_execute_base_query(proxy, std::move(partition_ranges), state, *internal_options, now, paging_state)
return find_index_partition_ranges(qp, state, *internal_options).then_unpack(
[this, now, &state, &internal_options, &qp, consume_results = std::move(consume_results)] (dht::partition_range_vector partition_ranges, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return do_execute_base_query(qp, std::move(partition_ranges), state, *internal_options, now, paging_state)
.then_unpack([paging_state, consume_results = std::move(consume_results)](foreign_ptr<lw_shared_ptr<query::result>> results, lw_shared_ptr<query::read_command> cmd) {
return consume_results(std::move(results), std::move(cmd), std::move(paging_state));
});
});
} else {
return find_index_clustering_rows(proxy, state, *internal_options).then_unpack(
[this, now, &state, &internal_options, &proxy, consume_results = std::move(consume_results)] (std::vector<primary_key> primary_keys, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->do_execute_base_query(proxy, std::move(primary_keys), state, *internal_options, now, paging_state)
return find_index_clustering_rows(qp, state, *internal_options).then_unpack(
[this, now, &state, &internal_options, &qp, consume_results = std::move(consume_results)] (std::vector<primary_key> primary_keys, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->do_execute_base_query(qp, std::move(primary_keys), state, *internal_options, now, paging_state)
.then_unpack([paging_state, consume_results = std::move(consume_results)](foreign_ptr<lw_shared_ptr<query::result>> results, lw_shared_ptr<query::read_command> cmd) {
return consume_results(std::move(results), std::move(cmd), std::move(paging_state));
});
@@ -1120,15 +1118,15 @@ indexed_table_select_statement::do_execute(service::storage_proxy& proxy,
tracing::trace(state.get_trace_state(), "Consulting index {} for a single slice of keys", _index.metadata().name());
// In this case, can use our normal query machinery, which retrieves
// entire partitions or the same slice for many partitions.
return find_index_partition_ranges(proxy, state, options).then_unpack([now, &state, &options, &proxy, this] (dht::partition_range_vector partition_ranges, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->execute_base_query(proxy, std::move(partition_ranges), state, options, now, std::move(paging_state));
return find_index_partition_ranges(qp, state, options).then_unpack([now, &state, &options, &qp, this] (dht::partition_range_vector partition_ranges, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->execute_base_query(qp, std::move(partition_ranges), state, options, now, std::move(paging_state));
});
} else {
tracing::trace(state.get_trace_state(), "Consulting index {} for a list of rows containing keys", _index.metadata().name());
// In this case, we need to retrieve a list of rows (not entire
// partitions) and then retrieve those specific rows.
return find_index_clustering_rows(proxy, state, options).then_unpack([now, &state, &options, &proxy, this] (std::vector<primary_key> primary_keys, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->execute_base_query(proxy, std::move(primary_keys), state, options, now, std::move(paging_state));
return find_index_clustering_rows(qp, state, options).then_unpack([now, &state, &options, &qp, this] (std::vector<primary_key> primary_keys, lw_shared_ptr<const service::pager::paging_state> paging_state) {
return this->execute_base_query(qp, std::move(primary_keys), state, options, now, std::move(paging_state));
});
}
}
@@ -1188,7 +1186,7 @@ query::partition_slice indexed_table_select_statement::get_partition_slice_for_l
// the posting-list for a particular value of the indexed column.
// Remember a secondary index can only be created on a single column.
future<::shared_ptr<cql_transport::messages::result_message::rows>>
indexed_table_select_statement::read_posting_list(service::storage_proxy& proxy,
indexed_table_select_statement::read_posting_list(query_processor& qp,
const query_options& options,
uint64_t limit,
service::query_state& state,
@@ -1203,7 +1201,7 @@ indexed_table_select_statement::read_posting_list(service::storage_proxy& proxy,
_view_schema->id(),
_view_schema->version(),
partition_slice,
proxy.get_max_result_size(partition_slice),
qp.proxy().get_max_result_size(partition_slice),
query::row_limit(limit),
query::partition_limit(query::max_partitions),
now,
@@ -1225,7 +1223,7 @@ indexed_table_select_statement::read_posting_list(service::storage_proxy& proxy,
int32_t page_size = options.get_page_size();
if (page_size <= 0 || !service::pager::query_pagers::may_need_paging(*_view_schema, page_size, *cmd, partition_ranges)) {
return proxy.query(_view_schema, cmd, std::move(partition_ranges), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
return qp.proxy().query(_view_schema, cmd, std::move(partition_ranges), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()})
.then([this, now, &options, selection = std::move(selection), partition_slice = std::move(partition_slice)] (service::storage_proxy::coordinator_query_result qr) {
cql3::selection::result_set_builder builder(*selection, now, options.get_cql_serialization_format());
query::result_view::consume(*qr.query_result,
@@ -1246,14 +1244,14 @@ indexed_table_select_statement::read_posting_list(service::storage_proxy& proxy,
// Note: the partitions keys returned by this function are sorted
// in token order. See issue #3423.
future<std::tuple<dht::partition_range_vector, lw_shared_ptr<const service::pager::paging_state>>>
indexed_table_select_statement::find_index_partition_ranges(service::storage_proxy& proxy,
indexed_table_select_statement::find_index_partition_ranges(query_processor& qp,
service::query_state& state,
const query_options& options) const
{
using value_type = std::tuple<dht::partition_range_vector, lw_shared_ptr<const service::pager::paging_state>>;
auto now = gc_clock::now();
auto timeout = db::timeout_clock::now() + get_timeout(state.get_client_state(), options);
return read_posting_list(proxy, options, get_limit(options), state, now, timeout, false).then(
return read_posting_list(qp, options, get_limit(options), state, now, timeout, false).then(
[this, now, &options] (::shared_ptr<cql_transport::messages::result_message::rows> rows) {
auto rs = cql3::untyped_result_set(rows);
dht::partition_range_vector partition_ranges;
@@ -1289,12 +1287,12 @@ indexed_table_select_statement::find_index_partition_ranges(service::storage_pro
// Note: the partitions keys returned by this function are sorted
// in token order. See issue #3423.
future<std::tuple<std::vector<indexed_table_select_statement::primary_key>, lw_shared_ptr<const service::pager::paging_state>>>
indexed_table_select_statement::find_index_clustering_rows(service::storage_proxy& proxy, service::query_state& state, const query_options& options) const
indexed_table_select_statement::find_index_clustering_rows(query_processor& qp, service::query_state& state, const query_options& options) const
{
using value_type = std::tuple<std::vector<indexed_table_select_statement::primary_key>, lw_shared_ptr<const service::pager::paging_state>>;
auto now = gc_clock::now();
auto timeout = db::timeout_clock::now() + get_timeout(state.get_client_state(), options);
return read_posting_list(proxy, options, get_limit(options), state, now, timeout, true).then(
return read_posting_list(qp, options, get_limit(options), state, now, timeout, true).then(
[this, now, &options] (::shared_ptr<cql_transport::messages::result_message::rows> rows) {
auto rs = cql3::untyped_result_set(rows);

26
cql3/statements/select_statement.hh
View File

@@ -107,7 +107,7 @@ protected:
bool _range_scan_no_bypass_cache = false;
std::unique_ptr<cql3::attributes> _attrs;
protected :
virtual future<::shared_ptr<cql_transport::messages::result_message>> do_execute(service::storage_proxy& proxy,
virtual future<::shared_ptr<cql_transport::messages::result_message>> do_execute(query_processor& qp,
service::query_state& state, const query_options& options) const;
friend class select_statement_executor;
public:
@@ -126,15 +126,15 @@ public:
virtual ::shared_ptr<const cql3::metadata> get_result_metadata() const override;
virtual uint32_t get_bound_terms() const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual bool depends_on_keyspace(const sstring& ks_name) const override;
virtual bool depends_on_column_family(const sstring& cf_name) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>> execute(query_processor& qp,
service::query_state& state, const query_options& options) const override;
future<::shared_ptr<cql_transport::messages::result_message>> execute(service::storage_proxy& proxy,
future<::shared_ptr<cql_transport::messages::result_message>> execute(query_processor& qp,
lw_shared_ptr<query::read_command> cmd, dht::partition_range_vector&& partition_ranges, service::query_state& state,
const query_options& options, gc_clock::time_point now) const;
@@ -228,17 +228,17 @@ public:
std::unique_ptr<cql3::attributes> attrs);
private:
virtual future<::shared_ptr<cql_transport::messages::result_message>> do_execute(service::storage_proxy& proxy,
virtual future<::shared_ptr<cql_transport::messages::result_message>> do_execute(query_processor& qp,
service::query_state& state, const query_options& options) const override;
lw_shared_ptr<const service::pager::paging_state> generate_view_paging_state_from_base_query_results(lw_shared_ptr<const service::pager::paging_state> paging_state,
const foreign_ptr<lw_shared_ptr<query::result>>& results, service::query_state& state, const query_options& options) const;
future<std::tuple<dht::partition_range_vector, lw_shared_ptr<const service::pager::paging_state>>> find_index_partition_ranges(service::storage_proxy& proxy,
future<std::tuple<dht::partition_range_vector, lw_shared_ptr<const service::pager::paging_state>>> find_index_partition_ranges(query_processor& qp,
service::query_state& state,
const query_options& options) const;
future<std::tuple<std::vector<primary_key>, lw_shared_ptr<const service::pager::paging_state>>> find_index_clustering_rows(service::storage_proxy& proxy,
future<std::tuple<std::vector<primary_key>, lw_shared_ptr<const service::pager::paging_state>>> find_index_clustering_rows(query_processor& qp,
service::query_state& state,
const query_options& options) const;
@@ -252,12 +252,12 @@ private:
lw_shared_ptr<const service::pager::paging_state> paging_state) const;
lw_shared_ptr<query::read_command>
prepare_command_for_base_query(service::storage_proxy& proxy, const query_options& options, service::query_state& state, gc_clock::time_point now,
prepare_command_for_base_query(query_processor& qp, const query_options& options, service::query_state& state, gc_clock::time_point now,
bool use_paging) const;
future<std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>>
do_execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
dht::partition_range_vector&& partition_ranges,
service::query_state& state,
const query_options& options,
@@ -265,7 +265,7 @@ private:
lw_shared_ptr<const service::pager::paging_state> paging_state) const;
future<shared_ptr<cql_transport::messages::result_message>>
execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
dht::partition_range_vector&& partition_ranges,
service::query_state& state,
const query_options& options,
@@ -283,7 +283,7 @@ private:
// Keys are ordered in token order (see #3423)
future<std::tuple<foreign_ptr<lw_shared_ptr<query::result>>, lw_shared_ptr<query::read_command>>>
do_execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
std::vector<primary_key>&& primary_keys,
service::query_state& state,
const query_options& options,
@@ -291,7 +291,7 @@ private:
lw_shared_ptr<const service::pager::paging_state> paging_state) const;
future<shared_ptr<cql_transport::messages::result_message>>
execute_base_query(
service::storage_proxy& proxy,
query_processor& qp,
std::vector<primary_key>&& primary_keys,
service::query_state& state,
const query_options& options,
@@ -304,7 +304,7 @@ private:
}
future<::shared_ptr<cql_transport::messages::result_message::rows>>read_posting_list(
service::storage_proxy& proxy,
query_processor& qp,
const query_options& options,
uint64_t limit,
service::query_state& state,

4
cql3/statements/service_level_statement.cc
View File

@@ -41,11 +41,11 @@ bool service_level_statement::depends_on_column_family(
}
void service_level_statement::validate(
service::storage_proxy &,
query_processor&,
const service::client_state &state) const {
}
future<> service_level_statement::check_access(service::storage_proxy& sp, const service::client_state &state) const {
future<> service_level_statement::check_access(query_processor& qp, const service::client_state &state) const {
return make_ready_future<>();
}

4
cql3/statements/service_level_statement.hh
View File

@@ -60,9 +60,9 @@ public:
bool depends_on_column_family(const sstring& cf_name) const override;
future<> check_access(service::storage_proxy& sp, const service::client_state& state) const override;
future<> check_access(query_processor& qp, const service::client_state& state) const override;
void validate(service::storage_proxy&, const service::client_state& state) const override;
void validate(query_processor&, const service::client_state& state) const override;
};
}

6
cql3/statements/truncate_statement.cc
View File

@@ -77,12 +77,12 @@ bool truncate_statement::depends_on_column_family(const sstring& cf_name) const
return false;
}
future<> truncate_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> truncate_statement::check_access(query_processor& qp, const service::client_state& state) const
{
return state.has_column_family_access(proxy.local_db(), keyspace(), column_family(), auth::permission::MODIFY);
return state.has_column_family_access(qp.proxy().local_db(), keyspace(), column_family(), auth::permission::MODIFY);
}
void truncate_statement::validate(service::storage_proxy&, const service::client_state& state) const
void truncate_statement::validate(query_processor&, const service::client_state& state) const
{
warn(unimplemented::cause::VALIDATION);
#if 0

4
cql3/statements/truncate_statement.hh
View File

@@ -62,9 +62,9 @@ public:
virtual bool depends_on_column_family(const sstring& cf_name) const override;
virtual future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual future<::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor& qp, service::query_state& state, const query_options& options) const override;

4
cql3/statements/use_statement.cc
View File

@@ -84,13 +84,13 @@ bool use_statement::depends_on_column_family(const sstring& cf_name) const
return false;
}
future<> use_statement::check_access(service::storage_proxy& proxy, const service::client_state& state) const
future<> use_statement::check_access(query_processor& qp, const service::client_state& state) const
{
state.validate_login();
return make_ready_future<>();
}
void use_statement::validate(service::storage_proxy&, const service::client_state& state) const
void use_statement::validate(query_processor&, const service::client_state& state) const
{
}

4
cql3/statements/use_statement.hh
View File

@@ -63,9 +63,9 @@ public:
virtual bool depends_on_column_family(const seastar::sstring& cf_name) const override;
virtual seastar::future<> check_access(service::storage_proxy& proxy, const service::client_state& state) const override;
virtual seastar::future<> check_access(query_processor& qp, const service::client_state& state) const override;
virtual void validate(service::storage_proxy&, const service::client_state& state) const override;
virtual void validate(query_processor&, const service::client_state& state) const override;
virtual seastar::future<seastar::shared_ptr<cql_transport::messages::result_message>>
execute(query_processor& qp, service::query_state& state, const query_options& options) const override;

2
test/lib/cql_test_env.cc
View File

@@ -275,7 +275,7 @@ public:
auto& qo = cql3::query_options::DEFAULT;
auto timeout = db::timeout_clock::now() + qs->get_client_state().get_timeout_config().write_timeout;
return modif_stmt->get_mutations(local_qp().proxy(), qo, timeout, false, qo.get_timestamp(*qs), *qs)
return modif_stmt->get_mutations(local_qp(), qo, timeout, false, qo.get_timestamp(*qs), *qs)
.finally([qs, modif_stmt = std::move(modif_stmt)] {});
}