mirrors/scylladb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
next
scylladb/alternator/executor.cc
Botond Dénes 035aa90d4b Merge 'Alternator: add per-table batch latency metrics and test coverage' from Amnon Heiman 
This series fixes a metrics visibility gap in Alternator and adds regression coverage.

Until now, BatchGetItem and BatchWriteItem updated global latency histograms but did not consistently update per-table latency histograms. As a result, table-level latency dashboards could miss batch traffic.

It updates the batch read/write paths to compute request duration once and record it in both global and per-table latency metrics.

Add the missing tests, including a metric-agnostic helper and a dedicated per-table latency test that verifies latency counters increase for item and batch operations.

This change is metrics-only (no API/behavior change for requests) and improves observability consistency between global and per-table views.

Fixes #28721

**We assume the alternator per-table metrics exist, but the batch ones are not updated**

Closes scylladb/scylladb#28732

* github.com:scylladb/scylladb:
  test(alternator): add per-table latency coverage for item and batch ops
  alternator: track per-table latency for batch get/write operations
2026-03-16 17:18:00 +02:00

6215 lines
324 KiB
C++
Raw Permalink Blame History

/*
* Copyright 2019-present ScyllaDB
*/
/*
* SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
*/
#include <fmt/ranges.h>
#include <seastar/core/on_internal_error.hh>
#include "alternator/executor.hh"
#include "alternator/consumed_capacity.hh"
#include "auth/permission.hh"
#include "auth/resource.hh"
#include "cdc/log.hh"
#include "cdc/cdc_options.hh"
#include "auth/service.hh"
#include "db/config.hh"
#include "db/view/view_build_status.hh"
#include "locator/tablets.hh"
#include "mutation/tombstone.hh"
#include "locator/abstract_replication_strategy.hh"
#include "utils/log.hh"
#include "schema/schema_builder.hh"
#include "exceptions/exceptions.hh"
#include "service/client_state.hh"
#include "mutation/timestamp.hh"
#include "types/map.hh"
#include "schema/schema.hh"
#include "query/query-request.hh"
#include "query/query-result-reader.hh"
#include "cql3/selection/selection.hh"
#include "cql3/result_set.hh"
#include "bytes.hh"
#include "service/pager/query_pagers.hh"
#include <functional>
#include "error.hh"
#include "serialization.hh"
#include "expressions.hh"
#include "conditions.hh"
#include <optional>
#include "utils/assert.hh"
#include "utils/overloaded_functor.hh"
#include "mutation/collection_mutation.hh"
#include "schema/schema.hh"
#include "db/tags/extension.hh"
#include "db/tags/utils.hh"
#include "replica/database.hh"
#include "alternator/rmw_operation.hh"
#include <seastar/core/coroutine.hh>
#include <seastar/core/sleep.hh>
#include <seastar/core/loop.hh>
#include <seastar/coroutine/maybe_yield.hh>
#include <boost/range/algorithm/find_end.hpp>
#include <unordered_set>
#include "service/storage_proxy.hh"
#include "gms/feature_service.hh"
#include "gms/gossiper.hh"
#include "utils/error_injection.hh"
#include "db/schema_tables.hh"
#include "utils/rjson.hh"
#include "alternator/extract_from_attrs.hh"
#include "types/types.hh"
#include "db/system_keyspace.hh"
#include "cql3/statements/ks_prop_defs.hh"
#include "alternator/ttl_tag.hh"
using namespace std::chrono_literals;
logging::logger elogger("alternator-executor");
namespace std {
template <> struct hash<std::pair<sstring, sstring>> {
size_t operator () (const std::pair<sstring, sstring>& p) const {
return std::hash<sstring>()(p.first) * 1009 + std::hash<sstring>()(p.second) * 3;
}
};
}
namespace alternator {
// Alternator-specific table properties stored as hidden table tags:
//
// Alternator doesn't keep its own records of which Alternator tables exist
// or how each one was configured. Instead, an Alternator table is created
// as a CQL table - and that CQL table stores its own name, schema, views,
// and so on. However, there are some Alternator-specific properties of a
// table which are not part of a CQL schema but which we need to remember.
// We store those extra properties as hidden "tags" on the CQL table, using
// the schema's "tags extension". This extension provides a map<string,string>
// for the table that is stored persistently on disk, but also readable
// quickly from memory.
// Alternator also uses the tags extension to store user-defined tags on
// tables (the TagResource, UntagResource and ListTagsOfResource requests).
// So the internal tags are kept hidden from the user by using the prefix
// "system:" in their name (see tag_key_is_internal()).
// The following is the list of these Alternator-specific hidden tags that
// Alternator adds to tables:
//
// Tags storing the "ReadCapacityUnits" and "WriteCapacityUnits"
// configured for a table with BillingMode=PROVISIONED.
const sstring RCU_TAG_KEY("system:provisioned_rcu");
const sstring WCU_TAG_KEY("system:provisioned_wcu");
// Tag storing the table's original creation time, in milliseconds since the
// Unix epoch. All tables get this tag when they are created, but it may be
// missing in old tables created before this tag was introduced.
const sstring TABLE_CREATION_TIME_TAG_KEY("system:table_creation_time");
// If this tag is present, it stores the name of the attribute that was
// configured by UpdateTimeToLive to be the expiration-time attribute for
// this table.
extern const sstring TTL_TAG_KEY("system:ttl_attribute");
// This will be set to 1 in a case, where user DID NOT specify a range key.
// The way GSI / LSI is implemented by Alternator assumes user specified keys will come first
// in materialized view's key list. Then, if needed missing keys are added (current implementation
// of materialized views requires that all base hash / range keys were added to the view as well).
// Alternator allows only a single range key attribute to be specified by the user. So if
// the SPURIOUS_RANGE_KEY_ADDED_TO_GSI_AND_USER_DIDNT_SPECIFY_RANGE_KEY_TAG_KEY is set the user didn't specify any key and
// base table's keys were added as range keys. In all other cases either the first key is the user specified key,
// following ones are base table's keys added as needed or range key list will be empty.
static const sstring SPURIOUS_RANGE_KEY_ADDED_TO_GSI_AND_USER_DIDNT_SPECIFY_RANGE_KEY_TAG_KEY("system:spurious_range_key_added_to_gsi_and_user_didnt_specify_range_key");
// The following tags also have the "system:" prefix but are NOT used
// by Alternator to store table properties - only the user ever writes to
// them, as a way to configure the table. As such, these tags are writable
// (and readable) by the user, and not hidden by tag_key_is_internal().
// The reason why both hidden (internal) and user-configurable tags share the
// same "system:" prefix is historic.
// Setting the tag with a numeric value will enable a specific initial number
// of tablets (setting the value to 0 means enabling tablets with
// an automatic selection of the best number of tablets).
// Setting this tag to any non-numeric value (e.g., an empty string or the
// word "none") will ask to disable tablets.
static constexpr auto INITIAL_TABLETS_TAG_KEY = "system:initial_tablets";
enum class table_status {
active = 0,
creating,
updating,
deleting
};
static std::string_view table_status_to_sstring(table_status tbl_status) {
switch (tbl_status) {
case table_status::active:
return "ACTIVE";
case table_status::creating:
return "CREATING";
case table_status::updating:
return "UPDATING";
case table_status::deleting:
return "DELETING";
}
return "UNKNOWN";
}
static lw_shared_ptr<keyspace_metadata> create_keyspace_metadata(std::string_view keyspace_name, service::storage_proxy& sp, gms::gossiper& gossiper, api::timestamp_type,
const std::map<sstring, sstring>& tags_map, const gms::feature_service& feat, const db::tablets_mode_t::mode tablets_mode);
static map_type attrs_type() {
static thread_local auto t = map_type_impl::get_instance(utf8_type, bytes_type, true);
return t;
}
static const column_definition& attrs_column(const schema& schema) {
const column_definition* cdef = schema.get_column_definition(bytes(executor::ATTRS_COLUMN_NAME));
throwing_assert(cdef);
return *cdef;
}
static lw_shared_ptr<stats> get_stats_from_schema(service::storage_proxy& sp, const schema& schema) {
try {
replica::table& table = sp.local_db().find_column_family(schema.id());
if (!table.get_stats().alternator_stats) {
table.get_stats().alternator_stats = seastar::make_shared<table_stats>(schema.ks_name(), schema.cf_name());
}
return table.get_stats().alternator_stats->_stats;
} catch (std::runtime_error&) {
// If we're here it means that a table we are currently working on was deleted before the
// operation completed, returning a temporary object is fine, if the table get deleted so will its metrics
return make_lw_shared<stats>();
}
}
executor::body_writer make_streamed(rjson::value&& value) {
return [value = std::move(value)](output_stream<char>&& _out) mutable -> future<> {
auto out = std::move(_out);
std::exception_ptr ex;
try {
co_await rjson::print(value, out);
} catch (...) {
ex = std::current_exception();
}
co_await out.close();
co_await rjson::destroy_gently(std::move(value));
if (ex) {
co_await coroutine::return_exception_ptr(std::move(ex));
}
};
}
// make_streamed_with_extra_array() is variant of make_streamed() above, which
// builds a streaming response (a function writing to an output stream) from a
// JSON object (rjson::value) but adds to it at the end an additional array.
// The extra array is given a separate chunked_vector to avoid putting it
// inside the rjson::value - because RapidJSON does contiguous allocations for
// arrays which we want to avoid for potentially long arrays in Query/Scan
// responses (see #23535).
// If we ever fix RapidJSON to avoid contiguous allocations for arrays, or
// replace it entirely (#24458), we can remove this function and the function
// rjson::print_with_extra_array() which it calls.
executor::body_writer make_streamed_with_extra_array(rjson::value&& value,
std::string array_name, utils::chunked_vector<rjson::value>&& array) {
return [value = std::move(value), array_name = std::move(array_name), array = std::move(array)](output_stream<char>&& _out) mutable -> future<> {
auto out = std::move(_out);
std::exception_ptr ex;
try {
co_await rjson::print_with_extra_array(value, array_name, array, out);
} catch (...) {
ex = std::current_exception();
}
co_await out.close();
co_await rjson::destroy_gently(std::move(value));
// TODO: can/should we also destroy the array gently?
if (ex) {
co_await coroutine::return_exception_ptr(std::move(ex));
}
};
}
// This function throws api_error::validation if input value is not an object.
static void validate_is_object(const rjson::value& value, const char* caller) {
if (!value.IsObject()) {
throw api_error::validation(fmt::format("{} must be an object", caller));
}
}
// This function assumes the given value is an object and returns requested member value.
// If it is not possible, an api_error::validation is thrown.
static const rjson::value& get_member(const rjson::value& obj, const char* member_name, const char* caller) {
validate_is_object(obj, caller);
const rjson::value* ret = rjson::find(obj, member_name);
if (!ret) {
throw api_error::validation(fmt::format("{} is missing a mandatory member {}", caller, member_name));
}
return *ret;
}
// This function assumes the given value is an object with a single member, and returns this member.
// In case the requirements are not met, an api_error::validation is thrown.
static const rjson::value::Member& get_single_member(const rjson::value& v, const char* caller) {
if (!v.IsObject() || v.MemberCount() != 1) {
throw api_error::validation(format("{}: expected an object with a single member.", caller));
}
return *(v.MemberBegin());
}
class executor::describe_table_info_manager : public service::migration_listener::empty_listener {
executor &_executor;
struct table_info {
utils::simple_value_with_expiry<std::uint64_t> size_in_bytes;
};
std::unordered_map<std::pair<sstring, sstring>, table_info> info_for_tables;
bool active = false;
public:
describe_table_info_manager(executor& executor) : _executor(executor) {
_executor._proxy.data_dictionary().real_database_ptr()->get_notifier().register_listener(this);
active = true;
}
describe_table_info_manager(const describe_table_info_manager &) = delete;
describe_table_info_manager(describe_table_info_manager&&) = delete;
~describe_table_info_manager() {
if (active) {
on_fatal_internal_error(elogger, "describe_table_info_manager was not stopped before destruction");
}
}
describe_table_info_manager &operator = (const describe_table_info_manager &) = delete;
describe_table_info_manager &operator = (describe_table_info_manager&&) = delete;
static std::chrono::high_resolution_clock::time_point now() {
return std::chrono::high_resolution_clock::now();
}
std::optional<std::uint64_t> get_cached_size_in_bytes(const sstring &ks_name, const sstring &cf_name) const {
auto it = info_for_tables.find({ks_name, cf_name});
if (it != info_for_tables.end()) {
return it->second.size_in_bytes.get();
}
return std::nullopt;
}
void cache_size_in_bytes(sstring ks_name, sstring cf_name, std::uint64_t size_in_bytes, std::chrono::high_resolution_clock::time_point expiry) {
info_for_tables[{std::move(ks_name), std::move(cf_name)}].size_in_bytes.set_if_longer_expiry(size_in_bytes, expiry);
}
future<> stop() {
co_await _executor._proxy.data_dictionary().real_database_ptr()->get_notifier().unregister_listener(this);
active = false;
co_return;
}
void on_drop_column_family(const sstring& ks_name, const sstring& cf_name) override {
if (!ks_name.starts_with(executor::KEYSPACE_NAME_PREFIX)) return;
info_for_tables.erase({ks_name, cf_name});
}
};
executor::executor(gms::gossiper& gossiper,
service::storage_proxy& proxy,
service::storage_service& ss,
service::migration_manager& mm,
db::system_distributed_keyspace& sdks,
cdc::metadata& cdc_metadata,
smp_service_group ssg,
utils::updateable_value<uint32_t> default_timeout_in_ms)
: _gossiper(gossiper),
_ss(ss),
_proxy(proxy),
_mm(mm),
_sdks(sdks),
_cdc_metadata(cdc_metadata),
_enforce_authorization(_proxy.data_dictionary().get_config().alternator_enforce_authorization),
_warn_authorization(_proxy.data_dictionary().get_config().alternator_warn_authorization),
_ssg(ssg),
_parsed_expression_cache(std::make_unique<parsed::expression_cache>(
parsed::expression_cache::config{_proxy.data_dictionary().get_config().alternator_max_expression_cache_entries_per_shard},
_stats))
{
s_default_timeout_in_ms = std::move(default_timeout_in_ms);
_describe_table_info_manager = std::make_unique<describe_table_info_manager>(*this);
register_metrics(_metrics, _stats);
}
executor::~executor() = default;
static void set_table_creation_time(std::map<sstring, sstring>& tags_map, db_clock::time_point creation_time) {
auto tm = std::chrono::duration_cast<std::chrono::milliseconds>(creation_time.time_since_epoch()).count();
tags_map[TABLE_CREATION_TIME_TAG_KEY] = std::to_string(tm);
}
static double get_table_creation_time(const schema &schema) {
auto time = db::find_tag(schema, TABLE_CREATION_TIME_TAG_KEY);
if (time) {
try {
auto val = std::stoll(*time);
if (val > 0) {
return val / 1000.0;
}
}
catch(...) {}
}
return 0.0;
}
void executor::supplement_table_info(rjson::value& descr, const schema& schema, service::storage_proxy& sp) {
auto creation_time = get_table_creation_time(schema);
rjson::add(descr, "CreationDateTime", rjson::value(creation_time));
rjson::add(descr, "TableStatus", "ACTIVE");
rjson::add(descr, "TableId", rjson::from_string(schema.id().to_sstring()));
executor::supplement_table_stream_info(descr, schema, sp);
}
// We would have liked to support table names up to 255 bytes, like DynamoDB.
// But Scylla creates a directory whose name is the table's name plus 33
// bytes (dash and UUID), and since directory names are limited to 255 bytes,
// we need to limit table names to 222 bytes, instead of 255.
// See https://github.com/scylladb/scylla/issues/4480
// We actually have two limits here,
// * max_table_name_length is the limit that Alternator will impose on names
// of new Alternator tables.
// * max_auxiliary_table_name_length is the potentially higher absolute limit
// that Scylla imposes on the names of auxiliary tables that Alternator
// wants to create internally - i.e. materialized views or CDC log tables.
// The second limit might mean that it is not possible to add a GSI to an
// existing table, because the name of the new auxiliary table may go over
// the limit. The second limit is also one of the reasons why the first limit
// is set lower than 222 - to have room to enable streams which add the extra
// suffix "_scylla_cdc_log" to the table name.
static constexpr int max_table_name_length = 192;
static constexpr int max_auxiliary_table_name_length = 222;
static bool valid_table_name_chars(std::string_view name) {
for (auto c : name) {
if ((c < 'a' || c > 'z') &&
(c < 'A' || c > 'Z') &&
(c < '0' || c > '9') &&
c != '_' &&
c != '-' &&
c != '.') {
return false;
}
}
return true;
}
// validate_table_name() validates the TableName parameter in a request - it
// should only be called in CreateTable or when a request looking for an
// existing table failed to find it. validate_table_name() throws the
// appropriate api_error if this validation fails.
// The DynamoDB developer guide, https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.NamingRules
// specifies that table "names must be between 3 and 255 characters long and
// can contain only the following characters: a-z, A-Z, 0-9, _ (underscore),
// - (dash), . (dot)". However, Alternator only allows max_table_name_length
// characters (see above) - not 255.
static void validate_table_name(std::string_view name) {
if (name.length() < 3 || name.length() > max_table_name_length) {
throw api_error::validation(
format("TableName must be at least 3 characters long and at most {} characters long", max_table_name_length));
}
if (!valid_table_name_chars(name)) {
throw api_error::validation(
"TableName must satisfy regular expression pattern: [a-zA-Z0-9_.-]+");
}
}
// Validate that a CDC log table could be created for the base table with a
// given table_name, and if not, throw a user-visible api_error::validation.
// It is not possible to create a CDC log table if the table name is so long
// that adding the 15-character suffix "_scylla_cdc_log" (cdc_log_suffix)
// makes it go over max_auxiliary_table_name_length.
// Note that if max_table_name_length is set to less than 207 (which is
// max_auxiliary_table_name_length-15), then this function will never
// fail. However, it's still important to call it in UpdateTable, in case
// we have pre-existing tables with names longer than this to avoid #24598.
static void validate_cdc_log_name_length(std::string_view table_name) {
if (cdc::log_name(table_name).length() > max_auxiliary_table_name_length) {
// CDC will add cdc_log_suffix ("_scylla_cdc_log") to the table name
// to create its log table, and this will exceed the maximum allowed
// length. To provide a more helpful error message, we assume that
// cdc::log_name() always adds a suffix of the same length.
int suffix_len = cdc::log_name(table_name).length() - table_name.length();
throw api_error::validation(fmt::format("Streams cannot be enabled to a table whose name is longer than {} characters: {}",
max_auxiliary_table_name_length - suffix_len, table_name));
}
}
// In DynamoDB index names are local to a table, while in Scylla, materialized
// view names are global (in a keyspace). So we need to compose a unique name
// for the view taking into account both the table's name and the index name.
// We concatenate the table and index name separated by a delim character
// (a character not allowed by DynamoDB in ordinary table names, default: ":").
// The downside of this approach is that it limits the sum of the lengths,
// instead of each component individually as DynamoDB does.
// The view_name() function assumes the table_name has already been validated
// but validates the legality of index_name and the combination of both.
static std::string view_name(std::string_view table_name, std::string_view index_name, const std::string& delim = ":", bool validate_len = true) {
if (index_name.length() < 3) {
throw api_error::validation("IndexName must be at least 3 characters long");
}
if (!valid_table_name_chars(index_name)) {
throw api_error::validation(
fmt::format("IndexName '{}' must satisfy regular expression pattern: [a-zA-Z0-9_.-]+", index_name));
}
std::string ret = std::string(table_name) + delim + std::string(index_name);
if (ret.length() > max_auxiliary_table_name_length && validate_len) {
throw api_error::validation(
fmt::format("The total length of TableName ('{}') and IndexName ('{}') cannot exceed {} characters",
table_name, index_name, max_auxiliary_table_name_length - delim.size()));
}
return ret;
}
static std::string lsi_name(std::string_view table_name, std::string_view index_name, bool validate_len = true) {
return view_name(table_name, index_name, "!:", validate_len);
}
/** Extract table name from a request.
* Most requests expect the table's name to be listed in a "TableName" field.
* This convenience function returns the name or api_error in case the
* table name is missing or not a string.
*/
static std::optional<std::string> find_table_name(const rjson::value& request) {
const rjson::value* table_name_value = rjson::find(request, "TableName");
if (!table_name_value) {
return std::nullopt;
}
if (!table_name_value->IsString()) {
throw api_error::validation("Non-string TableName field in request");
}
std::string table_name = rjson::to_string(*table_name_value);
return table_name;
}
static std::string get_table_name(const rjson::value& request) {
auto name = find_table_name(request);
if (!name) {
throw api_error::validation("Missing TableName field in request");
}
return *name;
}
/** Extract table schema from a request.
* Many requests expect the table's name to be listed in a "TableName" field
* and need to look it up as an existing table. This convenience function
* does this, with the appropriate validation and api_error in case the table
* name is missing, invalid or the table doesn't exist. If everything is
* successful, it returns the table's schema.
*/
schema_ptr executor::find_table(service::storage_proxy& proxy, const rjson::value& request) {
auto table_name = find_table_name(request);
if (!table_name) {
return nullptr;
}
return find_table(proxy, *table_name);
}
schema_ptr executor::find_table(service::storage_proxy& proxy, std::string_view table_name) {
try {
return proxy.data_dictionary().find_schema(sstring(executor::KEYSPACE_NAME_PREFIX) + sstring(table_name), table_name);
} catch(data_dictionary::no_such_column_family&) {
// DynamoDB returns validation error even when table does not exist
// and the table name is invalid.
validate_table_name(table_name);
throw api_error::resource_not_found(
fmt::format("Requested resource not found: Table: {} not found", table_name));
}
}
schema_ptr get_table(service::storage_proxy& proxy, const rjson::value& request) {
auto schema = executor::find_table(proxy, request);
if (!schema) {
// if we get here then the name was missing, since syntax or missing actual CF
// checks throw. Slow path, but just call get_table_name to generate exception.
get_table_name(request);
}
return schema;
}
// try_get_internal_table() handles the special case that the given table_name
// begins with INTERNAL_TABLE_PREFIX (".scylla.alternator."). In that case,
// this function assumes that the rest of the name refers to an internal
// Scylla table (e.g., system table) and returns the schema of that table -
// or an exception if it doesn't exist. Otherwise, if table_name does not
// start with INTERNAL_TABLE_PREFIX, this function returns an empty schema_ptr
// and the caller should look for a normal Alternator table with that name.
static schema_ptr try_get_internal_table(data_dictionary::database db, std::string_view table_name) {
size_t it = table_name.find(executor::INTERNAL_TABLE_PREFIX);
if (it != 0) {
return schema_ptr{};
}
table_name.remove_prefix(executor::INTERNAL_TABLE_PREFIX.size());
size_t delim = table_name.find_first_of('.');
if (delim == std::string_view::npos) {
return schema_ptr{};
}
std::string_view ks_name = table_name.substr(0, delim);
table_name.remove_prefix(ks_name.size() + 1);
// Only internal keyspaces can be accessed to avoid leakage
auto ks = db.try_find_keyspace(ks_name);
if (!ks || !ks->is_internal()) {
return schema_ptr{};
}
try {
return db.find_schema(ks_name, table_name);
} catch (data_dictionary::no_such_column_family&) {
// DynamoDB returns validation error even when table does not exist
// and the table name is invalid.
validate_table_name(table_name);
throw api_error::resource_not_found(
fmt::format("Requested resource not found: Internal table: {}.{} not found", ks_name, table_name));
}
}
// get_table_or_view() is similar to to get_table(), except it returns either
// a table or a materialized view from which to read, based on the TableName
// and optional IndexName in the request. Only requests like Query and Scan
// which allow IndexName should use this function.
enum class table_or_view_type { base, lsi, gsi };
static std::pair<schema_ptr, table_or_view_type>
get_table_or_view(service::storage_proxy& proxy, const rjson::value& request) {
table_or_view_type type = table_or_view_type::base;
std::string table_name = get_table_name(request);
if (schema_ptr s = try_get_internal_table(proxy.data_dictionary(), table_name)) {
return {s, type};
}
std::string keyspace_name = executor::KEYSPACE_NAME_PREFIX + table_name;
const rjson::value* index_name = rjson::find(request, "IndexName");
std::string orig_table_name;
if (index_name) {
if (index_name->IsString()) {
orig_table_name = std::move(table_name);
table_name = view_name(orig_table_name, rjson::to_string_view(*index_name));
type = table_or_view_type::gsi;
} else {
throw api_error::validation(
fmt::format("Non-string IndexName '{}'", rjson::to_string_view(*index_name)));
}
// If no tables for global indexes were found, the index may be local
if (!proxy.data_dictionary().has_schema(keyspace_name, table_name)) {
type = table_or_view_type::lsi;
table_name = lsi_name(orig_table_name, rjson::to_string_view(*index_name));
}
}
try {
return { proxy.data_dictionary().find_schema(keyspace_name, table_name), type };
} catch(data_dictionary::no_such_column_family&) {
if (index_name) {
// DynamoDB returns a different error depending on whether the
// base table doesn't exist (ResourceNotFoundException) or it
// does exist but the index does not (ValidationException).
if (proxy.data_dictionary().has_schema(keyspace_name, orig_table_name)) {
throw api_error::validation(
fmt::format("Requested resource not found: Index '{}' for table '{}'", rjson::to_string_view(*index_name), orig_table_name));
} else {
throw api_error::resource_not_found(
fmt::format("Requested resource not found: Table: {} not found", orig_table_name));
}
} else {
throw api_error::resource_not_found(
fmt::format("Requested resource not found: Table: {} not found", table_name));
}
}
}
// get_table_for_write() is similar to get_table(), but additionally, if the
// configuration allows this, may also allow writing to system table with
// prefix INTERNAL_TABLE_PREFIX. This is analogous to the function
// get_table_or_view() above which allows *reading* internal tables.
static schema_ptr get_table_for_write(service::storage_proxy& proxy, const rjson::value& request) {
std::string table_name = get_table_name(request);
if (schema_ptr s = try_get_internal_table(proxy.data_dictionary(), table_name)) {
if (!proxy.data_dictionary().get_config().alternator_allow_system_table_write()) {
throw api_error::resource_not_found(fmt::format(
"Table {} is an internal table, and writing to it is forbidden"
" by the alternator_allow_system_table_write configuration",
table_name));
}
return s;
}
return executor::find_table(proxy, table_name);
}
// Convenience function for getting the value of a string attribute, or a
// default value if it is missing. If the attribute exists, but is not a
// string, a descriptive api_error is thrown.
static std::string get_string_attribute(const rjson::value& value, std::string_view attribute_name, const char* default_return) {
const rjson::value* attribute_value = rjson::find(value, attribute_name);
if (!attribute_value)
return default_return;
if (!attribute_value->IsString()) {
throw api_error::validation(fmt::format("Expected string value for attribute {}, got: {}",
attribute_name, value));
}
return rjson::to_string(*attribute_value);
}
// Convenience function for getting the value of a boolean attribute, or a
// default value if it is missing. If the attribute exists, but is not a
// bool, a descriptive api_error is thrown.
static bool get_bool_attribute(const rjson::value& value, std::string_view attribute_name, bool default_return) {
const rjson::value* attribute_value = rjson::find(value, attribute_name);
if (!attribute_value) {
return default_return;
}
if (!attribute_value->IsBool()) {
throw api_error::validation(fmt::format("Expected boolean value for attribute {}, got: {}",
attribute_name, value));
}
return attribute_value->GetBool();
}
// Convenience function for getting the value of an integer attribute, or
// an empty optional if it is missing. If the attribute exists, but is not
// an integer, a descriptive api_error is thrown.
static std::optional<int> get_int_attribute(const rjson::value& value, std::string_view attribute_name) {
const rjson::value* attribute_value = rjson::find(value, attribute_name);
if (!attribute_value)
return {};
if (!attribute_value->IsInt()) {
throw api_error::validation(fmt::format("Expected integer value for attribute {}, got: {}",
attribute_name, value));
}
return attribute_value->GetInt();
}
// Sets a KeySchema object inside the given JSON parent describing the key
// attributes of the given schema as being either HASH or RANGE keys.
// Additionally, adds to a given map mappings between the key attribute
// names and their type (as a DynamoDB type string).
void executor::describe_key_schema(rjson::value& parent, const schema& schema, std::unordered_map<std::string,std::string>* attribute_types, const std::map<sstring, sstring> *tags) {
rjson::value key_schema = rjson::empty_array();
const bool ignore_range_keys_as_spurious = tags != nullptr && tags->contains(SPURIOUS_RANGE_KEY_ADDED_TO_GSI_AND_USER_DIDNT_SPECIFY_RANGE_KEY_TAG_KEY);
for (const column_definition& cdef : schema.partition_key_columns()) {
rjson::value key = rjson::empty_object();
rjson::add(key, "AttributeName", rjson::from_string(cdef.name_as_text()));
rjson::add(key, "KeyType", "HASH");
rjson::push_back(key_schema, std::move(key));
if (attribute_types) {
(*attribute_types)[cdef.name_as_text()] = type_to_string(cdef.type);
}
}
if (!ignore_range_keys_as_spurious) {
// NOTE: user requested key (there can be at most one) will always come first.
// There might be more keys following it, which were added, but those were
// not requested by the user, so we ignore them.
for (const column_definition& cdef : schema.clustering_key_columns()) {
rjson::value key = rjson::empty_object();
rjson::add(key, "AttributeName", rjson::from_string(cdef.name_as_text()));
rjson::add(key, "KeyType", "RANGE");
rjson::push_back(key_schema, std::move(key));
if (attribute_types) {
(*attribute_types)[cdef.name_as_text()] = type_to_string(cdef.type);
}
break;
}
}
rjson::add(parent, "KeySchema", std::move(key_schema));
}
void executor::describe_key_schema(rjson::value& parent, const schema& schema, std::unordered_map<std::string,std::string>& attribute_types, const std::map<sstring, sstring> *tags) {
describe_key_schema(parent, schema, &attribute_types, tags);
}
static rjson::value generate_arn_for_table(const schema& schema) {
return rjson::from_string(format("arn:scylla:alternator:{}:scylla:table/{}", schema.ks_name(), schema.cf_name()));
}
static rjson::value generate_arn_for_index(const schema& schema, std::string_view index_name) {
return rjson::from_string(fmt::format(
"arn:scylla:alternator:{}:scylla:table/{}/index/{}",
schema.ks_name(), schema.cf_name(), index_name));
}
// The following function checks if a given view has finished building.
// We need this for describe_table() to know if a view is still backfilling,
// or active.
//
// Currently we don't have in view_ptr the knowledge whether a view finished
// building long ago - so checking this involves a somewhat inefficient, but
// still node-local, process:
// We need a table that can accurately tell that all nodes have finished
// building this view. system.built_views is not good enough because it only
// knows the view building status in the current node. In recent versions,
// after PR #19745, we have a local table system.view_build_status_v2 with
// global information, replacing the old system_distributed.view_build_status.
// In theory, there can be a period during upgrading an old cluster when this
// table is not yet available. However, since the IndexStatus is a new feature
// too, it is acceptable that it doesn't yet work in the middle of the update.
static future<bool> is_view_built(
view_ptr view,
service::storage_proxy& proxy,
service::client_state& client_state,
tracing::trace_state_ptr trace_state,
service_permit permit) {
auto schema = proxy.data_dictionary().find_table(
"system", db::system_keyspace::VIEW_BUILD_STATUS_V2).schema();
// The table system.view_build_status_v2 has "keyspace_name" and
// "view_name" as the partition key, and each clustering row has
// "host_id" as clustering key and a string "status". We need to
// read a single partition:
partition_key pk = partition_key::from_exploded(*schema,
{utf8_type->decompose(view->ks_name()),
utf8_type->decompose(view->cf_name())});
dht::partition_range_vector partition_ranges{
dht::partition_range(dht::decorate_key(*schema, pk))};
auto selection = cql3::selection::selection::wildcard(schema); // only for get_query_options()!
auto partition_slice = query::partition_slice(
{query::clustering_range::make_open_ended_both_sides()},
{}, // static columns
{schema->get_column_definition("status")->id}, // regular columns
selection->get_query_options());
auto command = ::make_lw_shared<query::read_command>(
schema->id(), schema->version(), partition_slice,
proxy.get_max_result_size(partition_slice),
query::tombstone_limit(proxy.get_tombstone_limit()));
service::storage_proxy::coordinator_query_result qr =
co_await proxy.query(
schema, std::move(command), std::move(partition_ranges),
db::consistency_level::LOCAL_ONE,
service::storage_proxy::coordinator_query_options(
executor::default_timeout(), std::move(permit), client_state, trace_state));
query::result_set rs = query::result_set::from_raw_result(
schema, partition_slice, *qr.query_result);
std::unordered_map<locator::host_id, sstring> statuses;
for (auto&& r : rs.rows()) {
auto host_id = r.get<utils::UUID>("host_id");
auto status = r.get<sstring>("status");
if (host_id && status) {
statuses.emplace(locator::host_id(*host_id), *status);
}
}
// A view is considered "built" if all nodes reported SUCCESS in having
// built this view. Note that we need this "SUCCESS" for all nodes in the
// cluster - even those that are temporarily down (their success is known
// by this node, even if they are down). Conversely, we don't care what is
// the recorded status for any node which is no longer in the cluster - it
// is possible we forgot to erase the status of nodes that left the
// cluster, but here we just ignore them and look at the nodes actually
// in the topology.
bool all_built = true;
auto token_metadata = proxy.get_token_metadata_ptr();
token_metadata->get_topology().for_each_node(
[&] (const locator::node& node) {
// Note: we could skip nodes in DCs which have no replication of
// this view. However, in practice even those nodes would run
// the view building (and just see empty content) so we don't
// need to bother with this skipping.
auto it = statuses.find(node.host_id());
if (it == statuses.end() || it->second != "SUCCESS") {
all_built = false;
}
});
co_return all_built;
}
future<> executor::cache_newly_calculated_size_on_all_shards(schema_ptr schema, std::uint64_t size_in_bytes, std::chrono::nanoseconds ttl) {
auto expiry = describe_table_info_manager::now() + ttl;
return container().invoke_on_all(
[schema, size_in_bytes, expiry] (executor& exec) {
exec._describe_table_info_manager->cache_size_in_bytes(schema->ks_name(), schema->cf_name(), size_in_bytes, expiry);
});
}
future<> executor::fill_table_size(rjson::value &table_description, schema_ptr schema, bool deleting) {
auto cached_size = _describe_table_info_manager->get_cached_size_in_bytes(schema->ks_name(), schema->cf_name());
std::uint64_t total_size = 0;
if (cached_size) {
total_size = *cached_size;
} else {
// there's no point in trying to estimate value of table that is being deleted, as other nodes more often than not might
// move forward with deletion faster than we calculate the size
if (!deleting) {
total_size = co_await _ss.estimate_total_sstable_volume(schema->id(), service::storage_service::ignore_errors::yes);
const auto expiry = std::chrono::seconds{ _proxy.data_dictionary().get_config().alternator_describe_table_info_cache_validity_in_seconds() };
// Note: we don't care when the notification of other shards will finish, as long as it will be done
// it's possible to get into race condition (next DescribeTable comes to other shard, that new shard doesn't have
// the size yet, so it will calculate it again) - this is not a problem, because it will call cache_newly_calculated_size_on_all_shards
// with expiry, which is extremely unlikely to be exactly the same as the previous one, all shards will keep the size coming with expiry that is further into the future.
// In case of the same expiry, some shards will have different size, which means DescribeTable will return different values depending on the shard
// which is also fine, as the specification doesn't give precision guarantees of any kind.
co_await cache_newly_calculated_size_on_all_shards(schema, total_size, expiry);
}
}
rjson::add(table_description, "TableSizeBytes", total_size);
}
future<rjson::value> executor::fill_table_description(schema_ptr schema, table_status tbl_status, service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit)
{
rjson::value table_description = rjson::empty_object();
auto tags_ptr = db::get_tags_of_table(schema);
rjson::add(table_description, "TableName", rjson::from_string(schema->cf_name()));
co_await fill_table_size(table_description, schema, tbl_status == table_status::deleting);
auto creation_timestamp = get_table_creation_time(*schema);
// FIXME: In DynamoDB the CreateTable implementation is asynchronous, and
// the table may be in "Creating" state until creating is finished.
// We don't currently do this in Alternator - instead CreateTable waits
// until the table is really available. So/ DescribeTable returns either
// ACTIVE or doesn't exist at all (and DescribeTable returns an error).
// The states CREATING and UPDATING are not currently returned.
rjson::add(table_description, "TableStatus", rjson::from_string(table_status_to_sstring(tbl_status)));
rjson::add(table_description, "TableArn", generate_arn_for_table(*schema));
rjson::add(table_description, "TableId", rjson::from_string(schema->id().to_sstring()));
rjson::add(table_description, "BillingModeSummary", rjson::empty_object());
rjson::add(table_description["BillingModeSummary"], "LastUpdateToPayPerRequestDateTime", rjson::value(creation_timestamp));
// In PAY_PER_REQUEST billing mode, provisioned capacity should return 0
int rcu = 0;
int wcu = 0;
bool is_pay_per_request = true;
if (tags_ptr) {
auto rcu_tag = tags_ptr->find(RCU_TAG_KEY);
auto wcu_tag = tags_ptr->find(WCU_TAG_KEY);
if (rcu_tag != tags_ptr->end() && wcu_tag != tags_ptr->end()) {
try {
rcu = std::stoi(rcu_tag->second);
wcu = std::stoi(wcu_tag->second);
is_pay_per_request = false;
} catch (...) {
rcu = 0;
wcu = 0;
}
}
}
if (is_pay_per_request) {
rjson::add(table_description["BillingModeSummary"], "BillingMode", "PAY_PER_REQUEST");
} else {
rjson::add(table_description["BillingModeSummary"], "BillingMode", "PROVISIONED");
}
rjson::add(table_description, "ProvisionedThroughput", rjson::empty_object());
rjson::add(table_description["ProvisionedThroughput"], "ReadCapacityUnits", rcu);
rjson::add(table_description["ProvisionedThroughput"], "WriteCapacityUnits", wcu);
rjson::add(table_description["ProvisionedThroughput"], "NumberOfDecreasesToday", 0);
data_dictionary::table t = _proxy.data_dictionary().find_column_family(schema);
if (tbl_status != table_status::deleting) {
rjson::add(table_description, "CreationDateTime", rjson::value(creation_timestamp));
std::unordered_map<std::string,std::string> key_attribute_types;
// Add base table's KeySchema and collect types for AttributeDefinitions:
executor::describe_key_schema(table_description, *schema, key_attribute_types, tags_ptr);
if (!t.views().empty()) {
rjson::value gsi_array = rjson::empty_array();
rjson::value lsi_array = rjson::empty_array();
for (const view_ptr& vptr : t.views()) {
rjson::value view_entry = rjson::empty_object();
const sstring& cf_name = vptr->cf_name();
size_t delim_it = cf_name.find(':');
if (delim_it == sstring::npos) {
elogger.error("Invalid internal index table name: {}", cf_name);
continue;
}
sstring index_name = cf_name.substr(delim_it + 1);
rjson::add(view_entry, "IndexName", rjson::from_string(index_name));
rjson::add(view_entry, "IndexArn", generate_arn_for_index(*schema, index_name));
// Add index's KeySchema and collect types for AttributeDefinitions:
executor::describe_key_schema(view_entry, *vptr, key_attribute_types, db::get_tags_of_table(vptr));
// Add projection type
rjson::value projection = rjson::empty_object();
rjson::add(projection, "ProjectionType", "ALL");
// FIXME: we have to get ProjectionType from the schema when it is added
rjson::add(view_entry, "Projection", std::move(projection));
// Local secondary indexes are marked by an extra '!' sign occurring before the ':' delimiter
bool is_lsi = (delim_it > 1 && cf_name[delim_it-1] == '!');
// Add IndexStatus and Backfilling flags, but only for GSIs -
// LSIs can only be created with the table itself and do not
// have a status. Alternator schema operations are synchronous
// so only two combinations of these flags are possible: ACTIVE
// (for a built view) or CREATING+Backfilling (if view building
// is in progress).
if (!is_lsi) {
if (co_await is_view_built(vptr, _proxy, client_state, trace_state, permit)) {
rjson::add(view_entry, "IndexStatus", "ACTIVE");
} else {
rjson::add(view_entry, "IndexStatus", "CREATING");
rjson::add(view_entry, "Backfilling", rjson::value(true));
}
}
rjson::value& index_array = is_lsi ? lsi_array : gsi_array;
rjson::push_back(index_array, std::move(view_entry));
}
if (!lsi_array.Empty()) {
rjson::add(table_description, "LocalSecondaryIndexes", std::move(lsi_array));
}
if (!gsi_array.Empty()) {
rjson::add(table_description, "GlobalSecondaryIndexes", std::move(gsi_array));
}
}
// Use map built by describe_key_schema() for base and indexes to produce
// AttributeDefinitions for all key columns:
rjson::value attribute_definitions = rjson::empty_array();
for (auto& type : key_attribute_types) {
rjson::value key = rjson::empty_object();
rjson::add(key, "AttributeName", rjson::from_string(type.first));
rjson::add(key, "AttributeType", rjson::from_string(type.second));
rjson::push_back(attribute_definitions, std::move(key));
}
rjson::add(table_description, "AttributeDefinitions", std::move(attribute_definitions));
}
executor::supplement_table_stream_info(table_description, *schema, _proxy);
co_return table_description;
}
bool is_alternator_keyspace(const sstring& ks_name) {
return ks_name.find(executor::KEYSPACE_NAME_PREFIX) == 0;
}
sstring executor::table_name(const schema& s) {
return s.cf_name();
}
future<executor::request_return_type> executor::describe_table(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.describe_table++;
elogger.trace("Describing table {}", request);
schema_ptr schema = get_table(_proxy, request);
get_stats_from_schema(_proxy, *schema)->api_operations.describe_table++;
tracing::add_alternator_table_name(trace_state, schema->cf_name());
rjson::value table_description = co_await fill_table_description(schema, table_status::active, client_state, trace_state, permit);
rjson::value response = rjson::empty_object();
rjson::add(response, "Table", std::move(table_description));
elogger.trace("returning {}", response);
co_return rjson::print(std::move(response));
}
// This function increments the authorization_failures counter, and may also
// log a warn-level message and/or throw an access_denied exception, depending
// on what enforce_authorization and warn_authorization are set to.
// Note that if enforce_authorization is false, this function will return
// without throwing. So a caller that doesn't want to continue after an
// authorization_error must explicitly return after calling this function.
static void authorization_error(alternator::stats& stats, bool enforce_authorization, bool warn_authorization, std::string msg) {
stats.authorization_failures++;
if (enforce_authorization) {
if (warn_authorization) {
elogger.warn("alternator_warn_authorization=true: {}", msg);
}
throw api_error::access_denied(std::move(msg));
} else {
if (warn_authorization) {
elogger.warn("If you set alternator_enforce_authorization=true the following will be enforced: {}", msg);
}
}
}
// Check CQL's Role-Based Access Control (RBAC) permission_to_check (MODIFY,
// SELECT, DROP, etc.) on the given table. When permission is denied an
// appropriate user-readable api_error::access_denied is thrown.
future<> verify_permission(
bool enforce_authorization,
bool warn_authorization,
const service::client_state& client_state,
const schema_ptr& schema,
auth::permission permission_to_check,
alternator::stats& stats) {
if (!enforce_authorization && !warn_authorization) {
co_return;
}
// Unfortunately, the fix for issue #23218 did not modify the function
// that we use here - check_has_permissions(). So if we want to allow
// writes to internal tables (from try_get_internal_table()) only to a
// superuser, we need to explicitly check it here.
if (permission_to_check == auth::permission::MODIFY && is_internal_keyspace(schema->ks_name())) {
if (!client_state.user() ||
!client_state.user()->name ||
!co_await client_state.get_auth_service()->underlying_role_manager().is_superuser(*client_state.user()->name)) {
sstring username = "<anonymous>";
if (client_state.user() && client_state.user()->name) {
username = client_state.user()->name.value();
}
authorization_error(stats, enforce_authorization, warn_authorization, fmt::format(
"Write access denied on internal table {}.{} to role {} because it is not a superuser",
schema->ks_name(), schema->cf_name(), username));
co_return;
}
}
auto resource = auth::make_data_resource(schema->ks_name(), schema->cf_name());
if (!client_state.user() || !client_state.user()->name ||
!co_await client_state.check_has_permission(auth::command_desc(permission_to_check, resource))) {
sstring username = "<anonymous>";
if (client_state.user() && client_state.user()->name) {
username = client_state.user()->name.value();
}
// Using exceptions for errors makes this function faster in the
// success path (when the operation is allowed).
authorization_error(stats, enforce_authorization, warn_authorization, fmt::format(
"{} access on table {}.{} is denied to role {}, client address {}",
auth::permissions::to_string(permission_to_check),
schema->ks_name(), schema->cf_name(), username, client_state.get_client_address()));
}
}
// Similar to verify_permission() above, but just for CREATE operations.
// Those do not operate on any specific table, so require permissions on
// ALL KEYSPACES instead of any specific table.
static future<> verify_create_permission(bool enforce_authorization, bool warn_authorization, const service::client_state& client_state, alternator::stats& stats) {
if (!enforce_authorization && !warn_authorization) {
co_return;
}
auto resource = auth::resource(auth::resource_kind::data);
if (!co_await client_state.check_has_permission(auth::command_desc(auth::permission::CREATE, resource))) {
sstring username = "<anonymous>";
if (client_state.user() && client_state.user()->name) {
username = client_state.user()->name.value();
}
authorization_error(stats, enforce_authorization, warn_authorization, fmt::format(
"CREATE access on ALL KEYSPACES is denied to role {}", username));
}
}
future<executor::request_return_type> executor::delete_table(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.delete_table++;
elogger.trace("Deleting table {}", request);
std::string table_name = get_table_name(request);
std::string keyspace_name = executor::KEYSPACE_NAME_PREFIX + table_name;
tracing::add_alternator_table_name(trace_state, table_name);
auto& p = _proxy.container();
schema_ptr schema = get_table(_proxy, request);
rjson::value table_description = co_await fill_table_description(schema, table_status::deleting, client_state, trace_state, permit);
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::DROP, _stats);
co_await _mm.container().invoke_on(0, [&, cs = client_state.move_to_other_shard()] (service::migration_manager& mm) -> future<> {
size_t retries = mm.get_concurrent_ddl_retries();
for (;;) {
auto group0_guard = co_await mm.start_group0_operation();
std::optional<data_dictionary::table> tbl = p.local().data_dictionary().try_find_table(keyspace_name, table_name);
if (!tbl) {
// DynamoDB returns validation error even when table does not exist
// and the table name is invalid.
validate_table_name(table_name);
throw api_error::resource_not_found(fmt::format("Requested resource not found: Table: {} not found", table_name));
}
auto m = co_await service::prepare_column_family_drop_announcement(p.local(), keyspace_name, table_name, group0_guard.write_timestamp(), service::drop_views::yes);
auto m2 = co_await service::prepare_keyspace_drop_announcement(p.local(), keyspace_name, group0_guard.write_timestamp());
std::move(m2.begin(), m2.end(), std::back_inserter(m));
// When deleting a table and its views, we need to remove this role's
// special permissions in those tables (undoing the "auto-grant" done
// by CreateTable). If we didn't do this, if a second role later
// recreates a table with the same name, the first role would still
// have permissions over the new table.
// To make things more robust we just remove *all* permissions for
// the deleted table (CQL's drop_table_statement also does this).
// Unfortunately, there is an API mismatch between this code (which
// uses separate group0_guard and vector<mutation>) and the function
// revoke_all() which uses a combined "group0_batch" structure - so
// we need to do some ugly back-and-forth conversions between the pair
// to the group0_batch and back to the pair :-(
service::group0_batch mc(std::move(group0_guard));
mc.add_mutations(std::move(m));
auto resource = auth::make_data_resource(schema->ks_name(), schema->cf_name());
co_await auth::revoke_all(*cs.get().get_auth_service(), resource, mc);
for (const view_ptr& v : tbl->views()) {
resource = auth::make_data_resource(v->ks_name(), v->cf_name());
co_await auth::revoke_all(*cs.get().get_auth_service(), resource, mc);
}
std::tie(m, group0_guard) = co_await std::move(mc).extract();
try {
co_await mm.announce(std::move(m), std::move(group0_guard), fmt::format("alternator-executor: delete {} table", table_name));
break;
} catch (const service::group0_concurrent_modification& ex) {
elogger.info("Failed to execute DeleteTable {} due to concurrent schema modifications. {}.",
table_name, retries ? "Retrying" : "Number of retries exceeded, giving up");
if (retries--) {
continue;
}
throw;
}
}
});
rjson::value response = rjson::empty_object();
rjson::add(response, "TableDescription", std::move(table_description));
elogger.trace("returning {}", response);
co_return rjson::print(std::move(response));
}
static data_type parse_key_type(std::string_view type) {
// Note that keys are only allowed to be string, blob or number (S/B/N).
// The other types: boolean and various lists or sets - are not allowed.
if (type.length() == 1) {
switch (type[0]) {
case 'S': return utf8_type;
case 'B': return bytes_type;
case 'N': return decimal_type; // FIXME: use a specialized Alternator type, not the general "decimal_type".
}
}
throw api_error::validation(
fmt::format("Invalid key type '{}', can only be S, B or N.", type));
}
static void add_column(schema_builder& builder, const std::string& name, const rjson::value& attribute_definitions, column_kind kind, bool computed_column=false) {
// FIXME: Currently, the column name ATTRS_COLUMN_NAME is not allowed
// because we use it for our untyped attribute map, and we can't have a
// second column with the same name. We should fix this, by renaming
// some column names which we want to reserve.
if (name == executor::ATTRS_COLUMN_NAME) {
throw api_error::validation(fmt::format("Column name '{}' is currently reserved. FIXME.", name));
}
for (auto it = attribute_definitions.Begin(); it != attribute_definitions.End(); ++it) {
const rjson::value& attribute_info = *it;
if (rjson::to_string_view(attribute_info["AttributeName"]) == name) {
std::string_view type = rjson::to_string_view(attribute_info["AttributeType"]);
data_type dt = parse_key_type(type);
if (computed_column) {
// Computed column for GSI (doesn't choose a real column as-is
// but rather extracts a single value from the ":attrs" map)
alternator_type at = type_info_from_string(type).atype;
builder.with_computed_column(to_bytes(name), dt, kind,
std::make_unique<extract_from_attrs_column_computation>(to_bytes(name), at));
} else {
builder.with_column(to_bytes(name), dt, kind);
}
return;
}
}
throw api_error::validation(
fmt::format("KeySchema key '{}' missing in AttributeDefinitions", name));
}
// Parse the KeySchema request attribute, which specifies the column names
// for a key. A KeySchema must include up to two elements, the first must be
// the HASH key name, and the second one, if exists, must be a RANGE key name.
// The function returns the two column names - the first is the hash key
// and always present, the second is the range key and may be an empty string.
static std::pair<std::string, std::string> parse_key_schema(const rjson::value& obj, std::string_view supplementary_context) {
const rjson::value *key_schema;
if (!obj.IsObject() || !(key_schema = rjson::find(obj, "KeySchema"))) {
throw api_error::validation("Missing KeySchema member");
}
if (!key_schema->IsArray() || key_schema->Size() < 1 || key_schema->Size() > 2) {
throw api_error::validation("KeySchema must list exactly one or two key columns");
}
if (!(*key_schema)[0].IsObject()) {
throw api_error::validation("First element of KeySchema must be an object");
}
const rjson::value *v = rjson::find((*key_schema)[0], "KeyType");
if (!v || !v->IsString() || rjson::to_string_view(*v) != "HASH") {
throw api_error::validation("First key in KeySchema must be a HASH key");
}
v = rjson::find((*key_schema)[0], "AttributeName");
if (!v || !v->IsString()) {
throw api_error::validation("First key in KeySchema must have string AttributeName");
}
validate_attr_name_length(supplementary_context, v->GetStringLength(), true, "HASH key in KeySchema - ");
std::string hash_key = rjson::to_string(*v);
std::string range_key;
if (key_schema->Size() == 2) {
if (!(*key_schema)[1].IsObject()) {
throw api_error::validation("Second element of KeySchema must be an object");
}
v = rjson::find((*key_schema)[1], "KeyType");
if (!v || !v->IsString() || rjson::to_string_view(*v) != "RANGE") {
throw api_error::validation("Second key in KeySchema must be a RANGE key");
}
v = rjson::find((*key_schema)[1], "AttributeName");
if (!v || !v->IsString()) {
throw api_error::validation("Second key in KeySchema must have string AttributeName");
}
validate_attr_name_length(supplementary_context, v->GetStringLength(), true, "RANGE key in KeySchema - ");
range_key = v->GetString();
}
return {hash_key, range_key};
}
static schema_ptr get_table_from_arn(service::storage_proxy& proxy, std::string_view arn) {
// Expected format: arn:scylla:alternator:${KEYSPACE_NAME}:scylla:table/${TABLE_NAME};
constexpr size_t prefix_size = sizeof("arn:scylla:alternator:") - 1;
// NOTE: This code returns AccessDeniedException if it's problematic to parse or recognize an arn.
// Technically, a properly formatted, but nonexistent arn *should* return AccessDeniedException,
// while an incorrectly formatted one should return ValidationException.
// Unfortunately, the rules are really uncertain, since DynamoDB
// states that arns are of the form arn:partition:service:region:account-id:resource-type/resource-id
// or similar - yet, for some arns that do not fit that pattern (e.g. "john"),
// it still returns AccessDeniedException rather than ValidationException.
// Consequently, this code simply falls back to AccessDeniedException,
// concluding that an error is an error and code which uses tagging
// must be ready for handling AccessDeniedException instances anyway.
try {
size_t keyspace_end = arn.find_first_of(':', prefix_size);
std::string_view keyspace_name = arn.substr(prefix_size, keyspace_end - prefix_size);
size_t table_start = arn.find_first_of('/');
std::string_view table_name = arn.substr(table_start + 1);
if (table_name.find('/') != std::string_view::npos) {
// A table name cannot contain a '/' - if it does, it's not a
// table ARN, it may be an index. DynamoDB returns a
// ValidationException in that case - see #10786.
throw api_error::validation(fmt::format("ResourceArn '{}' is not a valid table ARN", table_name));
}
// FIXME: remove sstring creation once find_schema gains a view-based interface
return proxy.data_dictionary().find_schema(sstring(keyspace_name), sstring(table_name));
} catch (const data_dictionary::no_such_column_family& e) {
throw api_error::resource_not_found(fmt::format("ResourceArn '{}' not found", arn));
} catch (const std::out_of_range& e) {
throw api_error::access_denied("Incorrect resource identifier");
}
}
static bool is_legal_tag_char(char c) {
// FIXME: According to docs, unicode strings should also be accepted.
// Alternator currently uses a simplified ASCII approach
return std::isalnum(c) || std::isspace(c)
|| c == '+' || c == '-' || c == '=' || c == '.' || c == '_' || c == ':' || c == '/' ;
}
static bool validate_legal_tag_chars(std::string_view tag) {
return std::all_of(tag.begin(), tag.end(), &is_legal_tag_char);
}
static const std::unordered_set<std::string_view> allowed_write_isolation_values = {
"f", "forbid", "forbid_rmw",
"a", "always", "always_use_lwt",
"o", "only_rmw_uses_lwt",
"u", "unsafe", "unsafe_rmw",
};
static void validate_tags(const std::map<sstring, sstring>& tags) {
auto it = tags.find(rmw_operation::WRITE_ISOLATION_TAG_KEY);
if (it != tags.end()) {
std::string_view value = it->second;
if (!allowed_write_isolation_values.contains(value)) {
throw api_error::validation(
fmt::format("Incorrect write isolation tag {}. Allowed values: {}", value, allowed_write_isolation_values));
}
}
}
static rmw_operation::write_isolation parse_write_isolation(std::string_view value) {
if (!value.empty()) {
switch (value[0]) {
case 'f':
return rmw_operation::write_isolation::FORBID_RMW;
case 'a':
return rmw_operation::write_isolation::LWT_ALWAYS;
case 'o':
return rmw_operation::write_isolation::LWT_RMW_ONLY;
case 'u':
return rmw_operation::write_isolation::UNSAFE_RMW;
}
}
// Shouldn't happen as validate_tags() / set_default_write_isolation()
// verify allow only a closed set of values.
return rmw_operation::default_write_isolation;
}
// This default_write_isolation is always overwritten in main.cc, which calls
// set_default_write_isolation().
rmw_operation::write_isolation rmw_operation::default_write_isolation =
rmw_operation::write_isolation::LWT_ALWAYS;
void rmw_operation::set_default_write_isolation(std::string_view value) {
if (value.empty()) {
throw std::runtime_error("When Alternator is enabled, write "
"isolation policy must be selected, using the "
"'--alternator-write-isolation' option. "
"See docs/alternator/alternator.md for instructions.");
}
if (!allowed_write_isolation_values.contains(value)) {
throw std::runtime_error(fmt::format("Invalid --alternator-write-isolation "
"setting '{}'. Allowed values: {}.",
value, allowed_write_isolation_values));
}
default_write_isolation = parse_write_isolation(value);
}
// Alternator uses tags whose keys start with the "system:" prefix for
// internal purposes. Those should not be readable by ListTagsOfResource,
// nor writable with TagResource or UntagResource (see #24098).
// Only a few specific system tags, currently only "system:write_isolation"
// and "system:initial_tablets", are deliberately intended to be set and read
// by the user, so are not considered "internal".
static bool tag_key_is_internal(std::string_view tag_key) {
return tag_key.starts_with("system:")
&& tag_key != rmw_operation::WRITE_ISOLATION_TAG_KEY
&& tag_key != INITIAL_TABLETS_TAG_KEY;
}
enum class update_tags_action { add_tags, delete_tags };
static void update_tags_map(const rjson::value& tags, std::map<sstring, sstring>& tags_map, update_tags_action action) {
if (action == update_tags_action::add_tags) {
for (auto it = tags.Begin(); it != tags.End(); ++it) {
if (!it->IsObject()) {
throw api_error::validation("invalid tag object");
}
const rjson::value* key = rjson::find(*it, "Key");
const rjson::value* value = rjson::find(*it, "Value");
if (!key || !key->IsString() || !value || !value->IsString()) {
throw api_error::validation("string Key and Value required");
}
auto tag_key = rjson::to_string_view(*key);
auto tag_value = rjson::to_string_view(*value);
if (tag_key.empty()) {
throw api_error::validation("A tag Key cannot be empty");
}
if (tag_key.size() > 128) {
throw api_error::validation("A tag Key is limited to 128 characters");
}
if (!validate_legal_tag_chars(tag_key)) {
throw api_error::validation("A tag Key can only contain letters, spaces, and [+-=._:/]");
}
if (tag_key_is_internal(tag_key)) {
throw api_error::validation(fmt::format("Tag key '{}' is reserved for internal use", tag_key));
}
// Note tag values are limited similarly to tag keys, but have a
// longer length limit, and *can* be empty.
if (tag_value.size() > 256) {
throw api_error::validation("A tag Value is limited to 256 characters");
}
if (!validate_legal_tag_chars(tag_value)) {
throw api_error::validation("A tag Value can only contain letters, spaces, and [+-=._:/]");
}
tags_map[sstring(tag_key)] = sstring(tag_value);
}
} else if (action == update_tags_action::delete_tags) {
for (auto it = tags.Begin(); it != tags.End(); ++it) {
auto tag_key = rjson::to_string_view(*it);
if (tag_key_is_internal(tag_key)) {
throw api_error::validation(fmt::format("Tag key '{}' is reserved for internal use", tag_key));
}
tags_map.erase(sstring(tag_key));
}
}
if (tags_map.size() > 50) {
throw api_error::validation("Number of Tags exceed the current limit for the provided ResourceArn");
}
validate_tags(tags_map);
}
const std::map<sstring, sstring>& get_tags_of_table_or_throw(schema_ptr schema) {
auto tags_ptr = db::get_tags_of_table(schema);
if (tags_ptr) {
return *tags_ptr;
} else {
throw api_error::validation(format("Table {} does not have valid tagging information", schema->ks_name()));
}
}
future<executor::request_return_type> executor::tag_resource(client_state& client_state, service_permit permit, rjson::value request) {
_stats.api_operations.tag_resource++;
const rjson::value* arn = rjson::find(request, "ResourceArn");
if (!arn || !arn->IsString()) {
co_return api_error::access_denied("Incorrect resource identifier");
}
schema_ptr schema = get_table_from_arn(_proxy, rjson::to_string_view(*arn));
get_stats_from_schema(_proxy, *schema)->api_operations.tag_resource++;
const rjson::value* tags = rjson::find(request, "Tags");
if (!tags || !tags->IsArray()) {
co_return api_error::validation("Cannot parse tags");
}
if (tags->Size() < 1) {
co_return api_error::validation("The number of tags must be at least 1") ;
}
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::ALTER, _stats);
co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [tags](std::map<sstring, sstring>& tags_map) {
update_tags_map(*tags, tags_map, update_tags_action::add_tags);
});
co_return ""; // empty response
}
future<executor::request_return_type> executor::untag_resource(client_state& client_state, service_permit permit, rjson::value request) {
_stats.api_operations.untag_resource++;
const rjson::value* arn = rjson::find(request, "ResourceArn");
if (!arn || !arn->IsString()) {
co_return api_error::access_denied("Incorrect resource identifier");
}
const rjson::value* tags = rjson::find(request, "TagKeys");
if (!tags || !tags->IsArray()) {
co_return api_error::validation(format("Cannot parse tag keys"));
}
schema_ptr schema = get_table_from_arn(_proxy, rjson::to_string_view(*arn));
get_stats_from_schema(_proxy, *schema)->api_operations.untag_resource++;
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::ALTER, _stats);
co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [tags](std::map<sstring, sstring>& tags_map) {
update_tags_map(*tags, tags_map, update_tags_action::delete_tags);
});
co_return ""; // empty response
}
future<executor::request_return_type> executor::list_tags_of_resource(client_state& client_state, service_permit permit, rjson::value request) {
_stats.api_operations.list_tags_of_resource++;
const rjson::value* arn = rjson::find(request, "ResourceArn");
if (!arn || !arn->IsString()) {
return make_ready_future<request_return_type>(api_error::access_denied("Incorrect resource identifier"));
}
schema_ptr schema = get_table_from_arn(_proxy, rjson::to_string_view(*arn));
get_stats_from_schema(_proxy, *schema)->api_operations.list_tags_of_resource++;
auto tags_map = get_tags_of_table_or_throw(schema);
rjson::value ret = rjson::empty_object();
rjson::add(ret, "Tags", rjson::empty_array());
rjson::value& tags = ret["Tags"];
for (auto& tag_entry : tags_map) {
if (tag_key_is_internal(tag_entry.first)) {
continue;
}
rjson::value new_entry = rjson::empty_object();
rjson::add(new_entry, "Key", rjson::from_string(tag_entry.first));
rjson::add(new_entry, "Value", rjson::from_string(tag_entry.second));
rjson::push_back(tags, std::move(new_entry));
}
return make_ready_future<executor::request_return_type>(rjson::print(std::move(ret)));
}
struct billing_mode_type {
bool provisioned = false;
int rcu;
int wcu;
};
static billing_mode_type verify_billing_mode(const rjson::value& request) {
// Alternator does not yet support billing or throughput limitations, but
// let's verify that BillingMode is at least legal.
std::string billing_mode = get_string_attribute(request, "BillingMode", "PROVISIONED");
if (billing_mode == "PAY_PER_REQUEST") {
if (rjson::find(request, "ProvisionedThroughput")) {
throw api_error::validation("When BillingMode=PAY_PER_REQUEST, ProvisionedThroughput cannot be specified.");
}
} else if (billing_mode == "PROVISIONED") {
const rjson::value *provisioned_throughput = rjson::find(request, "ProvisionedThroughput");
if (!provisioned_throughput) {
throw api_error::validation("When BillingMode=PROVISIONED, ProvisionedThroughput must be specified.");
}
const rjson::value& throughput = *provisioned_throughput;
auto rcu = get_int_attribute(throughput, "ReadCapacityUnits");
auto wcu = get_int_attribute(throughput, "WriteCapacityUnits");
if (!rcu.has_value()) {
throw api_error::validation("provisionedThroughput.readCapacityUnits is missing, when BillingMode=PROVISIONED, ProvisionedThroughput must be specified.");
}
if (!wcu.has_value()) {
throw api_error::validation("provisionedThroughput.writeCapacityUnits is missing, when BillingMode=PROVISIONED, ProvisionedThroughput must be specified.");
}
return billing_mode_type{true, *rcu, *wcu};
} else {
throw api_error::validation("Unknown BillingMode={}. Must be PAY_PER_REQUEST or PROVISIONED.");
}
return billing_mode_type();
}
// Validate that a AttributeDefinitions parameter in CreateTable is valid, and
// throws user-facing api_error::validation if it's not.
// In particular, verify that the same AttributeName doesn't appear more than
// once (Issue #13870).
// Return the set of attribute names defined in AttributeDefinitions - this
// set is useful for later verifying that all of them are used by some
// KeySchema (issue #19784)
static std::unordered_set<std::string> validate_attribute_definitions(std::string_view supplementary_context, const rjson::value& attribute_definitions) {
if (!attribute_definitions.IsArray()) {
throw api_error::validation("AttributeDefinitions must be an array");
}
std::unordered_set<std::string> seen_attribute_names;
for (auto it = attribute_definitions.Begin(); it != attribute_definitions.End(); ++it) {
const rjson::value* attribute_name = rjson::find(*it, "AttributeName");
if (!attribute_name) {
throw api_error::validation("AttributeName missing in AttributeDefinitions");
}
if (!attribute_name->IsString()) {
throw api_error::validation("AttributeName in AttributeDefinitions must be a string");
}
validate_attr_name_length(supplementary_context, attribute_name->GetStringLength(), true, "in AttributeDefinitions - ");
auto [it2, added] = seen_attribute_names.emplace(rjson::to_string_view(*attribute_name));
if (!added) {
throw api_error::validation(fmt::format("Duplicate AttributeName={} in AttributeDefinitions",
rjson::to_string_view(*attribute_name)));
}
const rjson::value* attribute_type = rjson::find(*it, "AttributeType");
if (!attribute_type) {
throw api_error::validation("AttributeType missing in AttributeDefinitions");
}
if (!attribute_type->IsString()) {
throw api_error::validation("AttributeType in AttributeDefinitions must be a string");
}
}
return seen_attribute_names;
}
// The following "extract_from_attrs_column_computation" implementation is
// what allows Alternator GSIs and LSIs to use in a materialized view's key a
// member from the ":attrs" map instead of a real column in the schema:
const bytes extract_from_attrs_column_computation::MAP_NAME = executor::ATTRS_COLUMN_NAME;
column_computation_ptr extract_from_attrs_column_computation::clone() const {
return std::make_unique<extract_from_attrs_column_computation>(*this);
}
// Serialize the *definition* of this column computation into a JSON
// string with a unique "type" string - TYPE_NAME - which then causes
// column_computation::deserialize() to create an object from this class.
bytes extract_from_attrs_column_computation::serialize() const {
rjson::value ret = rjson::empty_object();
rjson::add(ret, "type", TYPE_NAME);
rjson::add(ret, "attr_name", rjson::from_string(to_string_view(_attr_name)));
rjson::add(ret, "desired_type", represent_type(_desired_type).ident);
return to_bytes(rjson::print(ret));
}
// Construct an extract_from_attrs_column_computation object based on the
// saved output of serialize(). Calls on_internal_error() if the string
// doesn't match the expected output format of serialize(). "type" is not
// checked - we assume the caller (column_computation::deserialize()) won't
// call this constructor if "type" doesn't match.
extract_from_attrs_column_computation::extract_from_attrs_column_computation(const rjson::value &v) {
const rjson::value* attr_name = rjson::find(v, "attr_name");
if (attr_name->IsString()) {
_attr_name = bytes(to_bytes_view(rjson::to_string_view(*attr_name)));
const rjson::value* desired_type = rjson::find(v, "desired_type");
if (desired_type->IsString()) {
_desired_type = type_info_from_string(rjson::to_string_view(*desired_type)).atype;
switch (_desired_type) {
case alternator_type::S:
case alternator_type::B:
case alternator_type::N:
// We're done
return;
default:
// Fall through to on_internal_error below.
break;
}
}
}
on_internal_error(elogger, format("Improperly formatted alternator::extract_from_attrs_column_computation computed column definition: {}", v));
}
regular_column_transformation::result extract_from_attrs_column_computation::compute_value(
const schema& schema,
const partition_key& key,
const db::view::clustering_or_static_row& row) const
{
const column_definition* attrs_col = schema.get_column_definition(MAP_NAME);
if (!attrs_col || !attrs_col->is_regular() || !attrs_col->is_multi_cell()) {
on_internal_error(elogger, "extract_from_attrs_column_computation::compute_value() on a table without an attrs map");
}
// Look for the desired attribute _attr_name in the attrs_col map in row:
const atomic_cell_or_collection* attrs = row.cells().find_cell(attrs_col->id);
if (!attrs) {
return regular_column_transformation::result();
}
collection_mutation_view cmv = attrs->as_collection_mutation();
return cmv.with_deserialized(*attrs_col->type, [this] (const collection_mutation_view_description& cmvd) {
for (auto&& [key, cell] : cmvd.cells) {
if (key == _attr_name) {
return regular_column_transformation::result(cell,
std::bind(serialized_value_if_type, std::placeholders::_1, _desired_type));
}
}
return regular_column_transformation::result();
});
}
// extract_from_attrs_column_computation needs the whole row to compute
// value, it can't use just the partition key.
bytes extract_from_attrs_column_computation::compute_value(const schema&, const partition_key&) const {
on_internal_error(elogger, "extract_from_attrs_column_computation::compute_value called without row");
}
// Because `CreateTable` request creates GSI/LSI together with the base table (so the base table is empty),
// we can skip view building process and immediately mark the view as built on all nodes.
//
// However, we can do this only for tablet-based views because `view_building_worker` will automatically propagate
// this information to `system.built_views` table (see `view_building_worker::update_built_views()`).
// For vnode-based views, `view_builder` will process the view and mark it as built.
static future<> mark_view_schemas_as_built(utils::chunked_vector<mutation>& out, std::vector<schema_ptr> schemas, api::timestamp_type ts, service::storage_proxy& sp) {
auto token_metadata = sp.get_token_metadata_ptr();
for (auto& schema: schemas) {
if (schema->is_view()) {
for (auto& host_id: token_metadata->get_topology().get_all_host_ids()) {
auto view_status_mut = co_await sp.system_keyspace().make_view_build_status_mutation(ts, {schema->ks_name(), schema->cf_name()}, host_id, db::view::build_status::SUCCESS);
out.push_back(std::move(view_status_mut));
}
}
}
}
future<executor::request_return_type> executor::create_table_on_shard0(service::client_state&& client_state, tracing::trace_state_ptr trace_state, rjson::value request, bool enforce_authorization, bool warn_authorization, const db::tablets_mode_t::mode tablets_mode) {
throwing_assert(this_shard_id() == 0);
// We begin by parsing and validating the content of the CreateTable
// command. We can't inspect the current database schema at this point
// (e.g., verify that this table doesn't already exist) - we can only
// do this further down - after taking group0_guard.
std::string table_name = get_table_name(request);
validate_table_name(table_name);
if (table_name.find(executor::INTERNAL_TABLE_PREFIX) == 0) {
co_return api_error::validation(fmt::format("Prefix {} is reserved for accessing internal tables", executor::INTERNAL_TABLE_PREFIX));
}
std::string keyspace_name = executor::KEYSPACE_NAME_PREFIX + table_name;
const rjson::value* attribute_definitions = rjson::find(request, "AttributeDefinitions");
if (attribute_definitions == nullptr) {
co_return api_error::validation("Missing AttributeDefinitions in CreateTable request");
}
// Save the list of AttributeDefinitions in unused_attribute_definitions,
// and below remove each one as we see it in a KeySchema of the table or
// any of its GSIs or LSIs. If anything remains in this set at the end of
// this function, it's an error.
std::unordered_set<std::string> unused_attribute_definitions =
validate_attribute_definitions("", *attribute_definitions);
tracing::add_alternator_table_name(trace_state, table_name);
schema_builder builder(keyspace_name, table_name);
auto [hash_key, range_key] = parse_key_schema(request, "");
add_column(builder, hash_key, *attribute_definitions, column_kind::partition_key);
unused_attribute_definitions.erase(hash_key);
if (!range_key.empty()) {
add_column(builder, range_key, *attribute_definitions, column_kind::clustering_key);
unused_attribute_definitions.erase(range_key);
}
builder.with_column(bytes(executor::ATTRS_COLUMN_NAME), attrs_type(), column_kind::regular_column);
billing_mode_type bm = verify_billing_mode(request);
schema_ptr partial_schema = builder.build();
// Parse Local/GlobalSecondaryIndexes parameters before creating the
// base table, so if we have a parse errors we can fail without creating
// any table.
std::vector<schema_builder> view_builders;
std::unordered_set<std::string> index_names;
const rjson::value* lsi = rjson::find(request, "LocalSecondaryIndexes");
if (lsi) {
if (!lsi->IsArray()) {
throw api_error::validation("LocalSecondaryIndexes must be an array.");
}
for (const rjson::value& l : lsi->GetArray()) {
const rjson::value* index_name_v = rjson::find(l, "IndexName");
if (!index_name_v || !index_name_v->IsString()) {
throw api_error::validation("LocalSecondaryIndexes IndexName must be a string.");
}
std::string_view index_name = rjson::to_string_view(*index_name_v);
auto [it, added] = index_names.emplace(index_name);
if (!added) {
co_return api_error::validation(fmt::format("Duplicate IndexName '{}', ", index_name));
}
std::string vname(lsi_name(table_name, index_name));
elogger.trace("Adding LSI {}", index_name);
if (range_key.empty()) {
co_return api_error::validation("LocalSecondaryIndex requires that the base table have a range key");
}
// FIXME: read and handle "Projection" parameter. This will
// require the MV code to copy just parts of the attrs map.
schema_builder view_builder(keyspace_name, vname);
auto [view_hash_key, view_range_key] = parse_key_schema(l, "Local Secondary Index");
if (view_hash_key != hash_key) {
co_return api_error::validation("LocalSecondaryIndex hash key must match the base table hash key");
}
add_column(view_builder, view_hash_key, *attribute_definitions, column_kind::partition_key);
unused_attribute_definitions.erase(view_hash_key);
if (view_range_key.empty()) {
co_return api_error::validation("LocalSecondaryIndex must specify a sort key");
}
unused_attribute_definitions.erase(view_range_key);
if (view_range_key == hash_key) {
co_return api_error::validation("LocalSecondaryIndex sort key cannot be the same as hash key");
}
add_column(view_builder, view_range_key, *attribute_definitions, column_kind::clustering_key, view_range_key != range_key);
// Base key columns which aren't part of the index's key need to
// be added to the view nonetheless, as (additional) clustering
// key(s).
if (!range_key.empty() && view_range_key != range_key) {
add_column(view_builder, range_key, *attribute_definitions, column_kind::clustering_key);
}
view_builder.with_column(bytes(executor::ATTRS_COLUMN_NAME), attrs_type(), column_kind::regular_column);
// Note above we don't need to add virtual columns, as all
// base columns were copied to view. TODO: reconsider the need
// for virtual columns when we support Projection.
// LSIs have no tags, but Scylla's "synchronous_updates" feature
// (which an LSIs need), is actually implemented as a tag so we
// need to add it here:
std::map<sstring, sstring> tags_map = {{db::SYNCHRONOUS_VIEW_UPDATES_TAG_KEY, "true"}};
view_builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(tags_map));
view_builders.emplace_back(std::move(view_builder));
}
}
const rjson::value* gsi = rjson::find(request, "GlobalSecondaryIndexes");
if (gsi) {
if (!gsi->IsArray()) {
co_return api_error::validation("GlobalSecondaryIndexes must be an array.");
}
for (const rjson::value& g : gsi->GetArray()) {
const rjson::value* index_name_v = rjson::find(g, "IndexName");
if (!index_name_v || !index_name_v->IsString()) {
co_return api_error::validation("GlobalSecondaryIndexes IndexName must be a string.");
}
std::string_view index_name = rjson::to_string_view(*index_name_v);
auto [it, added] = index_names.emplace(index_name);
if (!added) {
co_return api_error::validation(fmt::format("Duplicate IndexName '{}', ", index_name));
}
std::string vname(view_name(table_name, index_name));
elogger.trace("Adding GSI {}", index_name);
// FIXME: read and handle "Projection" parameter. This will
// require the MV code to copy just parts of the attrs map.
schema_builder view_builder(keyspace_name, vname);
auto [view_hash_key, view_range_key] = parse_key_schema(g, "GlobalSecondaryIndexes");
// If an attribute is already a real column in the base table
// (i.e., a key attribute), we can use it directly as a view key.
// Otherwise, we need to add it as a "computed column", which
// extracts and deserializes the attribute from the ":attrs" map.
bool view_hash_key_real_column = partial_schema->get_column_definition(to_bytes(view_hash_key));
add_column(view_builder, view_hash_key, *attribute_definitions, column_kind::partition_key, !view_hash_key_real_column);
unused_attribute_definitions.erase(view_hash_key);
if (!view_range_key.empty()) {
bool view_range_key_real_column = partial_schema->get_column_definition(to_bytes(view_range_key));
add_column(view_builder, view_range_key, *attribute_definitions, column_kind::clustering_key, !view_range_key_real_column);
if (!partial_schema->get_column_definition(to_bytes(view_range_key)) &&
!partial_schema->get_column_definition(to_bytes(view_hash_key))) {
// FIXME: This warning should go away. See issue #6714
elogger.warn("Only 1 regular column from the base table should be used in the GSI key in order to ensure correct liveness management without assumptions");
}
unused_attribute_definitions.erase(view_range_key);
}
// Base key columns which aren't part of the index's key need to
// be added to the view nonetheless, as (additional) clustering
// key(s).
// NOTE: DescribeTable's implementation depends on those keys being added AFTER user specified keys.
bool spurious_base_key_added_as_range_key = false;
if (hash_key != view_hash_key && hash_key != view_range_key) {
add_column(view_builder, hash_key, *attribute_definitions, column_kind::clustering_key);
spurious_base_key_added_as_range_key = true;
}
if (!range_key.empty() && range_key != view_hash_key && range_key != view_range_key) {
add_column(view_builder, range_key, *attribute_definitions, column_kind::clustering_key);
spurious_base_key_added_as_range_key = true;
}
std::map<sstring, sstring> tags;
if (view_range_key.empty() && spurious_base_key_added_as_range_key) {
tags[SPURIOUS_RANGE_KEY_ADDED_TO_GSI_AND_USER_DIDNT_SPECIFY_RANGE_KEY_TAG_KEY] = "true";
}
view_builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(std::move(tags)));
view_builders.emplace_back(std::move(view_builder));
}
}
if (!unused_attribute_definitions.empty()) {
co_return api_error::validation(fmt::format(
"AttributeDefinitions defines spurious attributes not used by any KeySchema: {}",
unused_attribute_definitions));
}
// We don't yet support configuring server-side encryption (SSE) via the
// SSESpecifiction attribute, but an SSESpecification with Enabled=false
// is simply the default, and should be accepted:
rjson::value* sse_specification = rjson::find(request, "SSESpecification");
if (sse_specification && sse_specification->IsObject()) {
rjson::value* enabled = rjson::find(*sse_specification, "Enabled");
if (!enabled || !enabled->IsBool()) {
co_return api_error("ValidationException", "SSESpecification needs boolean Enabled");
}
if (enabled->GetBool()) {
// TODO: full support for SSESpecification
co_return api_error("ValidationException", "SSESpecification: configuring encryption-at-rest is not yet supported.");
}
}
rjson::value* stream_specification = rjson::find(request, "StreamSpecification");
if (stream_specification && stream_specification->IsObject()) {
if (executor::add_stream_options(*stream_specification, builder, _proxy)) {
validate_cdc_log_name_length(builder.cf_name());
}
}
// Parse the "Tags" parameter early, so we can avoid creating the table
// at all if this parsing failed.
const rjson::value* tags = rjson::find(request, "Tags");
std::map<sstring, sstring> tags_map;
if (tags && tags->IsArray()) {
update_tags_map(*tags, tags_map, update_tags_action::add_tags);
}
if (bm.provisioned) {
tags_map[RCU_TAG_KEY] = std::to_string(bm.rcu);
tags_map[WCU_TAG_KEY] = std::to_string(bm.wcu);
}
set_table_creation_time(tags_map, db_clock::now());
builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(tags_map));
co_await verify_create_permission(enforce_authorization, warn_authorization, client_state, _stats);
schema_ptr schema = builder.build();
for (auto& view_builder : view_builders) {
// Note below we don't need to add virtual columns, as all
// base columns were copied to view. TODO: reconsider the need
// for virtual columns when we support Projection.
for (const column_definition& regular_cdef : schema->regular_columns()) {
if (!view_builder.has_column(*cql3::to_identifier(regular_cdef))) {
view_builder.with_column(regular_cdef.name(), regular_cdef.type, column_kind::regular_column);
}
}
const bool include_all_columns = true;
view_builder.with_view_info(schema, include_all_columns, ""/*where clause*/);
}
size_t retries = _mm.get_concurrent_ddl_retries();
for (;;) {
auto group0_guard = co_await _mm.start_group0_operation();
auto ts = group0_guard.write_timestamp();
utils::chunked_vector<mutation> schema_mutations;
auto ksm = create_keyspace_metadata(keyspace_name, _proxy, _gossiper, ts, tags_map, _proxy.features(), tablets_mode);
locator::replication_strategy_params params(ksm->strategy_options(), ksm->initial_tablets(), ksm->consistency_option());
const auto& topo = _proxy.local_db().get_token_metadata().get_topology();
auto rs = locator::abstract_replication_strategy::create_replication_strategy(ksm->strategy_name(), params, topo);
// Alternator Streams doesn't yet work when the table uses tablets (#23838)
if (stream_specification && stream_specification->IsObject()) {
auto stream_enabled = rjson::find(*stream_specification, "StreamEnabled");
if (stream_enabled && stream_enabled->IsBool() && stream_enabled->GetBool()) {
if (rs->uses_tablets()) {
co_return api_error::validation("Streams not yet supported on a table using tablets (issue #23838). "
"If you want to use streams, create a table with vnodes by setting the tag 'system:initial_tablets' set to 'none'.");
}
}
}
// Creating an index in tablets mode requires the keyspace to be RF-rack-valid.
// GSI and LSI indexes are based on materialized views which require RF-rack-validity to avoid consistency issues.
if (!view_builders.empty() || _proxy.data_dictionary().get_config().rf_rack_valid_keyspaces()) {
try {
locator::assert_rf_rack_valid_keyspace(keyspace_name, _proxy.local_db().get_token_metadata_ptr(), *rs);
} catch (const std::invalid_argument& ex) {
if (!view_builders.empty()) {
co_return api_error::validation(fmt::format("GlobalSecondaryIndexes and LocalSecondaryIndexes on a table "
"using tablets require the number of racks in the cluster to be either 1 or 3"));
} else {
co_return api_error::validation(fmt::format("Cannot create table '{}' with tablets: the configuration "
"option 'rf_rack_valid_keyspaces' is enabled, which enforces that tables using tablets can only be created in clusters "
"that have either 1 or 3 racks", table_name));
}
}
}
try {
schema_mutations = service::prepare_new_keyspace_announcement(_proxy.local_db(), ksm, ts);
} catch (exceptions::already_exists_exception&) {
if (_proxy.data_dictionary().has_schema(keyspace_name, table_name)) {
co_return api_error::resource_in_use(fmt::format("Table {} already exists", table_name));
}
}
if (_proxy.data_dictionary().try_find_table(schema->id())) {
// This should never happen, the ID is supposed to be unique
co_return api_error::internal(format("Table with ID {} already exists", schema->id()));
}
std::vector<schema_ptr> schemas;
schemas.push_back(schema);
for (schema_builder& view_builder : view_builders) {
schemas.push_back(view_builder.build());
}
co_await service::prepare_new_column_families_announcement(schema_mutations, _proxy, *ksm, schemas, ts);
if (ksm->uses_tablets()) {
co_await mark_view_schemas_as_built(schema_mutations, schemas, ts, _proxy);
}
// If a role is allowed to create a table, we must give it permissions to
// use (and eventually delete) the specific table it just created (and
// also the view tables). This is known as "auto-grant".
// Unfortunately, there is an API mismatch between this code (which uses
// separate group0_guard and vector<mutation>) and the function
// grant_applicable_permissions() which uses a combined "group0_batch"
// structure - so we need to do some ugly back-and-forth conversions
// between the pair to the group0_batch and back to the pair :-(
service::group0_batch mc(std::move(group0_guard));
mc.add_mutations(std::move(schema_mutations));
if (client_state.user()) {
auto resource = auth::make_data_resource(schema->ks_name(), schema->cf_name());
co_await auth::grant_applicable_permissions(
*client_state.get_auth_service(), *client_state.user(), resource, mc);
for (const schema_builder& view_builder : view_builders) {
resource = auth::make_data_resource(view_builder.ks_name(), view_builder.cf_name());
co_await auth::grant_applicable_permissions(
*client_state.get_auth_service(), *client_state.user(), resource, mc);
}
}
std::tie(schema_mutations, group0_guard) = co_await std::move(mc).extract();
try {
co_await _mm.announce(std::move(schema_mutations), std::move(group0_guard), fmt::format("alternator-executor: create {} table", table_name));
break;
} catch (const service::group0_concurrent_modification& ex) {
elogger.info("Failed to execute CreateTable {} due to concurrent schema modifications. {}.",
table_name, retries ? "Retrying" : "Number of retries exceeded, giving up");
if (retries--) {
continue;
}
throw;
}
}
co_await _mm.wait_for_schema_agreement(_proxy.local_db(), db::timeout_clock::now() + 10s, nullptr);
rjson::value status = rjson::empty_object();
executor::supplement_table_info(request, *schema, _proxy);
rjson::add(status, "TableDescription", std::move(request));
co_return rjson::print(std::move(status));
}
future<executor::request_return_type> executor::create_table(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.create_table++;
elogger.trace("Creating table {}", request);
co_return co_await _mm.container().invoke_on(0, [&, tr = tracing::global_trace_state_ptr(trace_state), request = std::move(request), &e = this->container(), client_state_other_shard = client_state.move_to_other_shard(), enforce_authorization = bool(_enforce_authorization), warn_authorization = bool(_warn_authorization)]
(service::migration_manager& mm) mutable -> future<executor::request_return_type> {
const db::tablets_mode_t::mode tablets_mode = _proxy.data_dictionary().get_config().tablets_mode_for_new_keyspaces(); // type cast
// `invoke_on` hopped us to shard 0, but `this` points to `executor` is from 'old' shard, we need to hop it too.
co_return co_await e.local().create_table_on_shard0(client_state_other_shard.get(), tr, std::move(request), enforce_authorization, warn_authorization, std::move(tablets_mode));
});
}
// When UpdateTable adds a GSI, the type of its key columns must be specified
// in a AttributeDefinitions. If one of these key columns are *already* key
// columns of the base table or any of its prior GSIs or LSIs, the type
// given in AttributeDefinitions must match the type of the existing key -
// otherwise Alternator will not know which type to enforce in new writes.
// This function checks for such conflicts. It assumes that the structure of
// the given attribute_definitions was already validated (with
// validate_attribute_definitions()).
// This function should be called multiple times - once for the base schema
// and once for each of its views (existing GSIs and LSIs on this table).
static void check_attribute_definitions_conflicts(const rjson::value& attribute_definitions, const schema& schema) {
for (auto& def : schema.primary_key_columns()) {
std::string def_type = type_to_string(def.type);
for (auto it = attribute_definitions.Begin(); it != attribute_definitions.End(); ++it) {
const rjson::value& attribute_info = *it;
if (rjson::to_string_view(attribute_info["AttributeName"]) == def.name_as_text()) {
std::string_view type = rjson::to_string_view(attribute_info["AttributeType"]);
if (type != def_type) {
throw api_error::validation(fmt::format("AttributeDefinitions redefined {} to {} already a key attribute of type {} in this table", def.name_as_text(), type, def_type));
}
break;
}
}
}
}
future<executor::request_return_type> executor::update_table(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.update_table++;
elogger.trace("Updating table {}", request);
static const std::vector<sstring> unsupported = {
"ProvisionedThroughput",
"ReplicaUpdates",
"SSESpecification",
};
for (auto& s : unsupported) {
if (rjson::find(request, s)) {
co_await coroutine::return_exception(api_error::validation(s + " not supported"));
}
}
bool empty_request = true;
if (rjson::find(request, "BillingMode")) {
empty_request = false;
verify_billing_mode(request);
}
co_return co_await _mm.container().invoke_on(0, [&p = _proxy.container(), request = std::move(request), gt = tracing::global_trace_state_ptr(std::move(trace_state)), enforce_authorization = bool(_enforce_authorization), warn_authorization = bool(_warn_authorization), client_state_other_shard = client_state.move_to_other_shard(), empty_request, &e = this->container()]
(service::migration_manager& mm) mutable -> future<executor::request_return_type> {
schema_ptr schema;
size_t retries = mm.get_concurrent_ddl_retries();
for (;;) {
auto group0_guard = co_await mm.start_group0_operation();
schema_ptr tab = get_table(p.local(), request);
tracing::add_alternator_table_name(gt, tab->cf_name());
// the ugly but harmless conversion to string_view here is because
// Seastar's sstring is missing a find(std::string_view) :-()
if (std::string_view(tab->cf_name()).find(INTERNAL_TABLE_PREFIX) == 0) {
co_await coroutine::return_exception(api_error::validation(fmt::format("Prefix {} is reserved for accessing internal tables", INTERNAL_TABLE_PREFIX)));
}
schema_builder builder(tab);
rjson::value* stream_specification = rjson::find(request, "StreamSpecification");
if (stream_specification && stream_specification->IsObject()) {
empty_request = false;
if (add_stream_options(*stream_specification, builder, p.local())) {
validate_cdc_log_name_length(builder.cf_name());
}
// Alternator Streams doesn't yet work when the table uses tablets (#23838)
auto stream_enabled = rjson::find(*stream_specification, "StreamEnabled");
if (stream_enabled && stream_enabled->IsBool()) {
if (stream_enabled->GetBool()) {
if (p.local().local_db().find_keyspace(tab->ks_name()).get_replication_strategy().uses_tablets()) {
co_return api_error::validation("Streams not yet supported on a table using tablets (issue #23838). "
"If you want to enable streams, re-create this table with vnodes (with the tag 'system:initial_tablets' set to 'none').");
}
if (tab->cdc_options().enabled()) {
co_return api_error::validation("Table already has an enabled stream: TableName: " + tab->cf_name());
}
}
else if (!tab->cdc_options().enabled()) {
co_return api_error::validation("Table has no stream to disable: TableName: " + tab->cf_name());
}
}
}
schema = builder.build();
std::vector<view_ptr> new_views;
std::vector<std::string> dropped_views;
rjson::value* gsi_updates = rjson::find(request, "GlobalSecondaryIndexUpdates");
if (gsi_updates) {
if (!gsi_updates->IsArray()) {
co_return api_error::validation("GlobalSecondaryIndexUpdates must be an array");
}
if (gsi_updates->Size() > 1) {
// Although UpdateTable takes an array of operations and could
// support multiple Create and/or Delete operations in one
// command, DynamoDB doesn't actually allows this, and throws
// a LimitExceededException if this is attempted.
co_return api_error::limit_exceeded("GlobalSecondaryIndexUpdates only allows one index creation or deletion");
}
if (gsi_updates->Size() == 1) {
empty_request = false;
if (!(*gsi_updates)[0].IsObject() || (*gsi_updates)[0].MemberCount() != 1) {
co_return api_error::validation("GlobalSecondaryIndexUpdates array must contain one object with a Create, Delete or Update operation");
}
auto it = (*gsi_updates)[0].MemberBegin();
const std::string_view op = rjson::to_string_view(it->name);
if (!it->value.IsObject()) {
co_return api_error::validation("GlobalSecondaryIndexUpdates entries must be objects");
}
const rjson::value* index_name_v = rjson::find(it->value, "IndexName");
if (!index_name_v || !index_name_v->IsString()) {
co_return api_error::validation("GlobalSecondaryIndexUpdates operation must have IndexName");
}
std::string_view index_name = rjson::to_string_view(*index_name_v);
std::string_view table_name = schema->cf_name();
std::string_view keyspace_name = schema->ks_name();
std::string vname(view_name(table_name, index_name));
if (op == "Create") {
const rjson::value* attribute_definitions = rjson::find(request, "AttributeDefinitions");
if (!attribute_definitions) {
co_return api_error::validation("GlobalSecondaryIndexUpdates Create needs AttributeDefinitions");
}
std::unordered_set<std::string> unused_attribute_definitions =
validate_attribute_definitions("GlobalSecondaryIndexUpdates", *attribute_definitions);
check_attribute_definitions_conflicts(*attribute_definitions, *schema);
for (auto& view : p.local().data_dictionary().find_column_family(tab).views()) {
check_attribute_definitions_conflicts(*attribute_definitions, *view);
}
if (p.local().data_dictionary().has_schema(keyspace_name, vname)) {
// Surprisingly, DynamoDB uses validation error here, not resource_in_use
co_return api_error::validation(fmt::format(
"GSI {} already exists in table {}", index_name, table_name));
}
if (p.local().data_dictionary().has_schema(keyspace_name, lsi_name(table_name, index_name, false))) {
co_return api_error::validation(fmt::format(
"LSI {} already exists in table {}, can't use same name for GSI", index_name, table_name));
}
try {
locator::assert_rf_rack_valid_keyspace(keyspace_name, p.local().local_db().get_token_metadata_ptr(),
p.local().local_db().find_keyspace(keyspace_name).get_replication_strategy());
} catch (const std::invalid_argument& ex) {
co_return api_error::validation(fmt::format("GlobalSecondaryIndexes on a table "
"using tablets require the number of racks in the cluster to be either 1 or 3"));
}
elogger.trace("Adding GSI {}", index_name);
// FIXME: read and handle "Projection" parameter. This will
// require the MV code to copy just parts of the attrs map.
schema_builder view_builder(keyspace_name, vname);
auto [view_hash_key, view_range_key] = parse_key_schema(it->value, "GlobalSecondaryIndexUpdates");
// If an attribute is already a real column in the base
// table (i.e., a key attribute in the base table),
// we can use it directly as a view key. Otherwise, we
// need to add it as a "computed column", which extracts
// and deserializes the attribute from the ":attrs" map.
bool view_hash_key_real_column =
schema->get_column_definition(to_bytes(view_hash_key));
add_column(view_builder, view_hash_key, *attribute_definitions, column_kind::partition_key, !view_hash_key_real_column);
unused_attribute_definitions.erase(view_hash_key);
if (!view_range_key.empty()) {
bool view_range_key_real_column =
schema->get_column_definition(to_bytes(view_range_key));
add_column(view_builder, view_range_key, *attribute_definitions, column_kind::clustering_key, !view_range_key_real_column);
if (!schema->get_column_definition(to_bytes(view_range_key)) &&
!schema->get_column_definition(to_bytes(view_hash_key))) {
// FIXME: This warning should go away. See issue #6714
elogger.warn("Only 1 regular column from the base table should be used in the GSI key in order to ensure correct liveness management without assumptions");
}
unused_attribute_definitions.erase(view_range_key);
}
// Surprisingly, although DynamoDB checks for unused
// AttributeDefinitions in CreateTable, it does not
// check it in UpdateTable. We decided to check anyway.
if (!unused_attribute_definitions.empty()) {
co_return api_error::validation(fmt::format(
"AttributeDefinitions defines spurious attributes not used by any KeySchema: {}",
unused_attribute_definitions));
}
// Base key columns which aren't part of the index's key need to
// be added to the view nonetheless, as (additional) clustering
// key(s).
for (auto& def : schema->primary_key_columns()) {
if (def.name_as_text() != view_hash_key && def.name_as_text() != view_range_key) {
view_builder.with_column(def.name(), def.type, column_kind::clustering_key);
}
}
// GSIs have no tags:
view_builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>());
// Note below we don't need to add virtual columns, as all
// base columns were copied to view. TODO: reconsider the need
// for virtual columns when we support Projection.
for (const column_definition& regular_cdef : schema->regular_columns()) {
if (!view_builder.has_column(*cql3::to_identifier(regular_cdef))) {
view_builder.with_column(regular_cdef.name(), regular_cdef.type, column_kind::regular_column);
}
}
const bool include_all_columns = true;
view_builder.with_view_info(schema, include_all_columns, ""/*where clause*/);
new_views.emplace_back(view_builder.build());
} else if (op == "Delete") {
elogger.trace("Deleting GSI {}", index_name);
if (!p.local().data_dictionary().has_schema(keyspace_name, vname)) {
co_return api_error::resource_not_found(fmt::format("No GSI {} in table {}", index_name, table_name));
}
dropped_views.emplace_back(vname);
} else if (op == "Update") {
co_return api_error::validation("GlobalSecondaryIndexUpdates Update not yet supported");
} else {
co_return api_error::validation(fmt::format("GlobalSecondaryIndexUpdates supports a Create, Delete or Update operation, saw '{}'", op));
}
}
}
if (empty_request) {
co_return api_error::validation("UpdateTable requires one of GlobalSecondaryIndexUpdates, StreamSpecification or BillingMode to be specified");
}
co_await verify_permission(enforce_authorization, warn_authorization, client_state_other_shard.get(), schema, auth::permission::ALTER, e.local()._stats);
auto m = co_await service::prepare_column_family_update_announcement(p.local(), schema, std::vector<view_ptr>(), group0_guard.write_timestamp());
for (view_ptr view : new_views) {
auto m2 = co_await service::prepare_new_view_announcement(p.local(), view, group0_guard.write_timestamp());
std::move(m2.begin(), m2.end(), std::back_inserter(m));
}
for (const std::string& view_name : dropped_views) {
auto m2 = co_await service::prepare_view_drop_announcement(p.local(), schema->ks_name(), view_name, group0_guard.write_timestamp());
std::move(m2.begin(), m2.end(), std::back_inserter(m));
}
// If a role is allowed to create a GSI, we should give it permissions
// to read the GSI it just created. This is known as "auto-grant".
// Also, when we delete a GSI we should revoke any permissions set on
// it - so if it's ever created again the old permissions wouldn't be
// remembered for the new GSI. This is known as "auto-revoke"
if (client_state_other_shard.get().user() && (!new_views.empty() || !dropped_views.empty())) {
service::group0_batch mc(std::move(group0_guard));
mc.add_mutations(std::move(m));
for (view_ptr view : new_views) {
auto resource = auth::make_data_resource(view->ks_name(), view->cf_name());
co_await auth::grant_applicable_permissions(
*client_state_other_shard.get().get_auth_service(), *client_state_other_shard.get().user(), resource, mc);
}
for (const auto& view_name : dropped_views) {
auto resource = auth::make_data_resource(schema->ks_name(), view_name);
co_await auth::revoke_all(*client_state_other_shard.get().get_auth_service(), resource, mc);
}
std::tie(m, group0_guard) = co_await std::move(mc).extract();
}
try {
co_await mm.announce(std::move(m), std::move(group0_guard), format("alternator-executor: update {} table", tab->cf_name()));
break;
} catch (const service::group0_concurrent_modification& ex) {
elogger.info("Failed to execute UpdateTable {} due to concurrent schema modifications. {}.",
tab->cf_name(), retries ? "Retrying" : "Number of retries exceeded, giving up");
if (retries--) {
continue;
}
throw;
}
}
co_await mm.wait_for_schema_agreement(p.local().local_db(), db::timeout_clock::now() + 10s, nullptr);
rjson::value status = rjson::empty_object();
supplement_table_info(request, *schema, p.local());
rjson::add(status, "TableDescription", std::move(request));
co_return rjson::print(std::move(status));
});
}
// attribute_collector is a helper class used to accept several attribute
// puts or deletes, and collect them as single collection mutation.
// The implementation is somewhat complicated by the need of cells in a
// collection to be sorted by key order.
class attribute_collector {
std::map<bytes, atomic_cell, serialized_compare> collected;
void add(bytes&& name, atomic_cell&& cell) {
collected.emplace(std::move(name), std::move(cell));
}
void add(const bytes& name, atomic_cell&& cell) {
collected.emplace(name, std::move(cell));
}
public:
attribute_collector() : collected(attrs_type()->get_keys_type()->as_less_comparator()) { }
void put(bytes&& name, const bytes& val, api::timestamp_type ts) {
add(std::move(name), atomic_cell::make_live(*bytes_type, ts, val, atomic_cell::collection_member::yes));
}
void put(const bytes& name, const bytes& val, api::timestamp_type ts) {
add(name, atomic_cell::make_live(*bytes_type, ts, val, atomic_cell::collection_member::yes));
}
void del(bytes&& name, api::timestamp_type ts) {
add(std::move(name), atomic_cell::make_dead(ts, gc_clock::now()));
}
void del(const bytes& name, api::timestamp_type ts) {
add(name, atomic_cell::make_dead(ts, gc_clock::now()));
}
collection_mutation_description to_mut() {
collection_mutation_description ret;
for (auto&& e : collected) {
ret.cells.emplace_back(e.first, std::move(e.second));
}
return ret;
}
bool empty() const {
return collected.empty();
}
};
// After calling pk_from_json() and ck_from_json() to extract the pk and ck
// components of a key, and if that succeeded, call check_key() to further
// check that the key doesn't have any spurious components.
static void check_key(const rjson::value& key, const schema_ptr& schema) {
if (key.MemberCount() != (schema->clustering_key_size() == 0 ? 1 : 2)) {
throw api_error::validation("Given key attribute not in schema");
}
}
// Verify that a value parsed from the user input is legal. In particular,
// we check that the value is not an empty set, string or bytes - which is
// (somewhat artificially) forbidden by DynamoDB.
void validate_value(const rjson::value& v, const char* caller) {
if (!v.IsObject() || v.MemberCount() != 1) {
throw api_error::validation(format("{}: improperly formatted value '{}'", caller, v));
}
auto it = v.MemberBegin();
const std::string_view type = rjson::to_string_view(it->name);
if (type == "SS" || type == "BS" || type == "NS") {
if (!it->value.IsArray()) {
throw api_error::validation(format("{}: improperly formatted set '{}'", caller, v));
}
if (it->value.Size() == 0) {
throw api_error::validation(format("{}: empty set not allowed", caller));
}
} else if (type == "S" || type == "B") {
if (!it->value.IsString()) {
throw api_error::validation(format("{}: improperly formatted value '{}'", caller, v));
}
} else if (type == "N") {
if (!it->value.IsString()) {
// DynamoDB uses a SerializationException in this case, not ValidationException.
throw api_error::serialization(format("{}: number value must be encoded as string '{}'", caller, v));
}
} else if (type != "L" && type != "M" && type != "BOOL" && type != "NULL") {
// TODO: can do more sanity checks on the content of the above types.
throw api_error::validation(fmt::format("{}: unknown type {} for value {}", caller, type, v));
}
}
// The put_or_delete_item class builds the mutations needed by the PutItem and
// DeleteItem operations - either as stand-alone commands or part of a list
// of commands in BatchWriteItem.
// put_or_delete_item splits each operation into two stages: Constructing the
// object parses and validates the user input (throwing exceptions if there
// are input errors). Later, build() generates the actual mutation, with a
// specified timestamp. This split is needed because of the peculiar needs of
// BatchWriteItem and LWT. BatchWriteItem needs all parsing to happen before
// any writing happens (if one of the commands has an error, none of the
// writes should be done). LWT makes it impossible for the parse step to
// generate "mutation" objects, because the timestamp still isn't known.
class put_or_delete_item {
private:
partition_key _pk;
clustering_key _ck;
struct cell {
bytes column_name;
bytes value;
};
// PutItem: engaged _cells, write these cells to item (_pk, _ck).
// DeleteItem: disengaged _cells, delete the entire item (_pk, _ck).
std::optional<std::vector<cell>> _cells;
// WCU calculation takes into account some length in bytes,
// that length can have different meaning depends on the operation but the
// the calculation of length in bytes to WCU is the same.
uint64_t _length_in_bytes = 0;
public:
struct delete_item {};
struct put_item {};
put_or_delete_item(const rjson::value& key, schema_ptr schema, delete_item);
put_or_delete_item(const rjson::value& item, schema_ptr schema, put_item, std::unordered_map<bytes, std::string> key_attributes);
// put_or_delete_item doesn't keep a reference to schema (so it can be
// moved between shards for LWT) so it needs to be given again to build():
mutation build(schema_ptr schema, api::timestamp_type ts) const;
const partition_key& pk() const { return _pk; }
const clustering_key& ck() const { return _ck; }
uint64_t length_in_bytes() const noexcept {
return _length_in_bytes;
}
void set_length_in_bytes(uint64_t length) noexcept {
_length_in_bytes = length;
}
bool is_put_item() noexcept {
return _cells.has_value();
}
};
put_or_delete_item::put_or_delete_item(const rjson::value& key, schema_ptr schema, delete_item)
: _pk(pk_from_json(key, schema)), _ck(ck_from_json(key, schema)) {
check_key(key, schema);
}
// find_attribute() checks whether the named attribute is stored in the
// schema as a real column (we do this for key attribute, and for a GSI key)
// and if so, returns that column. If not, the function returns nullptr,
// telling the caller that the attribute is stored serialized in the
// ATTRS_COLUMN_NAME map - not in a stand-alone column in the schema.
static inline const column_definition* find_attribute(const schema& schema, const bytes& attribute_name) {
const column_definition* cdef = schema.get_column_definition(attribute_name);
// Although ATTRS_COLUMN_NAME exists as an actual column, when used as an
// attribute name it should refer to an attribute inside ATTRS_COLUMN_NAME
// not to ATTRS_COLUMN_NAME itself. This if() is needed for #5009.
if (cdef && cdef->name() == executor::ATTRS_COLUMN_NAME) {
return nullptr;
}
return cdef;
}
// Get a list of all attributes that serve as a key attributes for any of the
// GSIs or LSIs of this table, and the declared type for each (can be only
// "S", "B", or "N"). The implementation below will also list the base table's
// key columns (they are the views' clustering keys).
std::unordered_map<bytes, std::string> si_key_attributes(data_dictionary::table t) {
std::unordered_map<bytes, std::string> ret;
for (const view_ptr& v : t.views()) {
for (const column_definition& cdef : v->partition_key_columns()) {
ret[cdef.name()] = type_to_string(cdef.type);
}
for (const column_definition& cdef : v->clustering_key_columns()) {
ret[cdef.name()] = type_to_string(cdef.type);
}
}
return ret;
}
// When an attribute is a key (hash or sort) of one of the GSIs or LSIs on a
// table, DynamoDB refuses an update to that attribute with an unsuitable
// value. Unsuitable values are:
// 1. An empty string (those are normally allowed as values, but not allowed
// as keys, including GSI keys).
// 2. A value with a type different than that declared for the GSI key.
// Normally non-key attributes can take values of any type (DynamoDB is
// schema-less), but as soon as an attribute is used as a GSI key, it
// must be set only to the specific type declared for that key.
// (Note that a missing value for an GSI key attribute is fine - the update
// will happen on the base table, but won't reach the view table. In this
// case, this function simply won't be called for this attribute.)
//
// This function checks if the given attribute update is an update to some
// GSI's key, and if the value is unsuitable, an api_error::validation is
// thrown. The checking here is similar to the checking done in
// get_key_from_typed_value() for the base table's key columns.
//
// validate_value_if_index_key() should only be called after validate_value()
// already validated that the value itself has a valid form.
static inline void validate_value_if_index_key(
std::unordered_map<bytes, std::string> key_attributes,
const bytes& attribute,
const rjson::value& value) {
if (key_attributes.empty()) {
return;
}
auto it = key_attributes.find(attribute);
if (it == key_attributes.end()) {
// Given attribute is not a key column with a fixed type, so no
// more validation to do.
return;
}
const std::string& expected_type = it->second;
// We assume that validate_value() was previously called on this value,
// so value is known to be of the proper format (an object with one
// member, whose key and value are strings)
std::string_view value_type = rjson::to_string_view(value.MemberBegin()->name);
if (expected_type != value_type) {
throw api_error::validation(fmt::format(
"Type mismatch: expected type {} for GSI or LSI key attribute {}, got type {}",
expected_type, to_string_view(attribute), value_type));
}
std::string_view value_content = rjson::to_string_view(value.MemberBegin()->value);
if (value_content.empty()) {
throw api_error::validation(fmt::format(
"GSI or LSI key attribute {} cannot be set to an empty string", to_string_view(attribute)));
}
}
put_or_delete_item::put_or_delete_item(const rjson::value& item, schema_ptr schema, put_item, std::unordered_map<bytes, std::string> key_attributes)
: _pk(pk_from_json(item, schema)), _ck(ck_from_json(item, schema)) {
_cells = std::vector<cell>();
_cells->reserve(item.MemberCount());
for (auto it = item.MemberBegin(); it != item.MemberEnd(); ++it) {
bytes column_name = to_bytes(rjson::to_string_view(it->name));
validate_value(it->value, "PutItem");
const column_definition* cdef = find_attribute(*schema, column_name);
validate_attr_name_length("", column_name.size(), cdef && cdef->is_primary_key());
_length_in_bytes += column_name.size();
if (!cdef) {
// This attribute may be a key column of one of the GSI or LSI,
// in which case there are some limitations on the value.
validate_value_if_index_key(key_attributes, column_name, it->value);
bytes value = serialize_item(it->value);
if (value.size()) {
// ScyllaDB uses one extra byte compared to DynamoDB for the bytes length
_length_in_bytes += value.size() - 1;
}
_cells->push_back({std::move(column_name), serialize_item(it->value)});
} else if (!cdef->is_primary_key()) {
// Fixed-type regular columns were used for LSIs and also (in the
// slightly more distant past) GSIs, before they were moved to
// :attrs. We keep this branch for backward compatibility.
bytes value = get_key_from_typed_value(it->value, *cdef);
_cells->push_back({std::move(column_name), value});
if (value.size()) {
// ScyllaDB uses one extra byte compared to DynamoDB for the bytes length
_length_in_bytes += value.size() - 1;
}
}
}
if (_pk.representation().size() > 2) {
// ScyllaDB uses two extra bytes compared to DynamoDB for the key bytes length
_length_in_bytes += _pk.representation().size() - 2;
}
if (_ck.representation().size() > 2) {
// ScyllaDB uses two extra bytes compared to DynamoDB for the key bytes length
_length_in_bytes += _ck.representation().size() - 2;
}
}
mutation put_or_delete_item::build(schema_ptr schema, api::timestamp_type ts) const {
mutation m(schema, _pk);
// If there's no clustering key, a tombstone should be created directly
// on a partition, not on a clustering row - otherwise it will look like
// an open-ended range tombstone, which will crash on KA/LA sstable format.
// Ref: #6035
const bool use_partition_tombstone = schema->clustering_key_size() == 0;
if (!_cells) {
if (use_partition_tombstone) {
m.partition().apply(tombstone(ts, gc_clock::now()));
} else {
// a DeleteItem operation:
m.partition().clustered_row(*schema, _ck).apply(tombstone(ts, gc_clock::now()));
}
return m;
}
// else, a PutItem operation:
auto& row = m.partition().clustered_row(*schema, _ck);
attribute_collector attrs_collector;
for (auto& c : *_cells) {
const column_definition* cdef = find_attribute(*schema, c.column_name);
if (!cdef) {
attrs_collector.put(c.column_name, c.value, ts);
} else {
row.cells().apply(*cdef, atomic_cell::make_live(*cdef->type, ts, std::move(c.value)));
}
}
auto attrs = attrs_column(*schema);
if (!attrs_collector.empty()) {
auto serialized_map = attrs_collector.to_mut().serialize(*attrs_type());
row.cells().apply(attrs, std::move(serialized_map));
}
// To allow creation of an item with no attributes, we need a row marker.
row.apply(row_marker(ts));
// PutItem is supposed to completely replace the old item, so we need to
// also have a tombstone removing old cells. Important points:
// 1) Alternator's schema is dynamic, therefore we store data in a map
// in column :attrs. Since we're replacing a row, invalidating only
// :attrs is enough. Alternator base tables also had columns for LSI
// keys and GSI keys. New tables no longer have such columns, but old
// tables created in the past may still have them.
// 2) We use a collection tombstone for the :attrs column instead of a row
// tombstone. While a row tombstone would also replace the data, it has
// an undesirable side effect for CDC, which would report it as a
// separate deletion event. To model PutItem's "replace" semantic, we
// leverage a corner case: a collection tombstone at ts-1 paired with an
// upsert at ts is not reported by CDC as a separate REMOVE event. We
// can't use the timestamp ts, because when data and tombstone tie on
// timestamp, the tombstone wins. These tricks were introduced in
// Scylla to handle collection replacements in CQL (see #6084, PR #6491,
// e.g. cql3::maps::setter::execute()) and we utilize it to avoid
// emitting the REMOVE event (resolving #6930).
row.cells().apply(attrs, collection_mutation_description{tombstone{ts - 1, gc_clock::now()}}.serialize(*attrs.type));
// Note that for old tables created with regular LSI and GSI key columns,
// we must also delete the regular columns that are not part of the new
// schema consisting of pk, ck, and :attrs.
for (const auto& cdef : schema->regular_columns()) {
if (cdef.name_as_text() != executor::ATTRS_COLUMN_NAME) {
row.cells().apply(cdef, atomic_cell::make_dead(ts - 1, gc_clock::now()));
}
}
return m;
}
// The DynamoDB API doesn't let the client control the server's timeout, so
// we have a global default_timeout() for Alternator requests. The value of
// s_default_timeout_ms is overwritten in alternator::controller::start_server()
// based on the "alternator_timeout_in_ms" configuration parameter.
thread_local utils::updateable_value<uint32_t> executor::s_default_timeout_in_ms{10'000};
db::timeout_clock::time_point executor::default_timeout() {
return db::timeout_clock::now() + std::chrono::milliseconds(s_default_timeout_in_ms);
}
static lw_shared_ptr<query::read_command> previous_item_read_command(service::storage_proxy& proxy,
schema_ptr schema,
const clustering_key& ck,
shared_ptr<cql3::selection::selection> selection) {
std::vector<query::clustering_range> bounds;
if (schema->clustering_key_size() == 0) {
bounds.push_back(query::clustering_range::make_open_ended_both_sides());
} else {
bounds.push_back(query::clustering_range::make_singular(ck));
}
// FIXME: We pretend to take a selection (all callers currently give us a
// wildcard selection...) but here we read the entire item anyway. We
// should take the column list from selection instead of building it here.
auto regular_columns =
schema->regular_columns() | std::views::transform(&column_definition::id)
| std::ranges::to<query::column_id_vector>();
auto partition_slice = query::partition_slice(std::move(bounds), {}, std::move(regular_columns), selection->get_query_options());
return ::make_lw_shared<query::read_command>(schema->id(), schema->version(), partition_slice, proxy.get_max_result_size(partition_slice),
query::tombstone_limit(proxy.get_tombstone_limit()));
}
static dht::partition_range_vector to_partition_ranges(const schema& schema, const partition_key& pk) {
return dht::partition_range_vector{dht::partition_range(dht::decorate_key(schema, pk))};
}
static dht::partition_range_vector to_partition_ranges(const dht::decorated_key& pk) {
return dht::partition_range_vector{dht::partition_range(pk)};
}
// Parse the different options for the ReturnValues parameter. We parse all
// the known options, but only UpdateItem actually supports all of them. The
// other operations (DeleteItem and PutItem) will refuse some of them.
rmw_operation::returnvalues rmw_operation::parse_returnvalues(const rjson::value& request) {
const rjson::value* attribute_value = rjson::find(request, "ReturnValues");
if (!attribute_value) {
return rmw_operation::returnvalues::NONE;
}
if (!attribute_value->IsString()) {
throw api_error::validation(format("Expected string value for ReturnValues, got: {}", *attribute_value));
}
auto s = rjson::to_string_view(*attribute_value);
if (s == "NONE") {
return rmw_operation::returnvalues::NONE;
} else if (s == "ALL_OLD") {
return rmw_operation::returnvalues::ALL_OLD;
} else if (s == "UPDATED_OLD") {
return rmw_operation::returnvalues::UPDATED_OLD;
} else if (s == "ALL_NEW") {
return rmw_operation::returnvalues::ALL_NEW;
} else if (s == "UPDATED_NEW") {
return rmw_operation::returnvalues::UPDATED_NEW;
} else {
throw api_error::validation(fmt::format("Unrecognized value for ReturnValues: {}", s));
}
}
rmw_operation::returnvalues_on_condition_check_failure
rmw_operation::parse_returnvalues_on_condition_check_failure(const rjson::value& request) {
const rjson::value* attribute_value = rjson::find(request, "ReturnValuesOnConditionCheckFailure");
if (!attribute_value) {
return rmw_operation::returnvalues_on_condition_check_failure::NONE;
}
if (!attribute_value->IsString()) {
throw api_error::validation(format("Expected string value for ReturnValuesOnConditionCheckFailure, got: {}", *attribute_value));
}
auto s = rjson::to_string_view(*attribute_value);
if (s == "NONE") {
return rmw_operation::returnvalues_on_condition_check_failure::NONE;
} else if (s == "ALL_OLD") {
return rmw_operation::returnvalues_on_condition_check_failure::ALL_OLD;
} else {
throw api_error::validation(fmt::format("Unrecognized value for ReturnValuesOnConditionCheckFailure: {}", s));
}
}
rmw_operation::rmw_operation(service::storage_proxy& proxy, rjson::value&& request)
: _request(std::move(request))
, _schema(get_table_for_write(proxy, _request))
, _write_isolation(get_write_isolation_for_schema(_schema))
, _consumed_capacity(_request)
, _returnvalues(parse_returnvalues(_request))
, _returnvalues_on_condition_check_failure(parse_returnvalues_on_condition_check_failure(_request))
{
// _pk and _ck will be assigned later, by the subclass's constructor
// (each operation puts the key in a slightly different location in
// the request).
}
std::optional<mutation> rmw_operation::apply(foreign_ptr<lw_shared_ptr<query::result>> qr, const query::partition_slice& slice, api::timestamp_type ts, cdc::per_request_options& cdc_opts) {
if (qr->row_count()) {
auto selection = cql3::selection::selection::wildcard(_schema);
uint64_t item_length = 0;
auto previous_item = executor::describe_single_item(_schema, slice, *selection, *qr, {}, &item_length);
if (_consumed_capacity._total_bytes < item_length) {
_consumed_capacity._total_bytes = item_length;
}
if (previous_item) {
if (should_fill_preimage()) {
cdc_opts.preimage = make_lw_shared<cql3::untyped_result_set>(*_schema, std::move(qr), *selection, slice);
}
return apply(std::make_unique<rjson::value>(std::move(*previous_item)), ts, cdc_opts);
}
}
return apply(std::unique_ptr<rjson::value>(), ts, cdc_opts);
}
rmw_operation::write_isolation rmw_operation::get_write_isolation_for_schema(schema_ptr schema) {
const auto tags_ptr = db::get_tags_of_table(schema);
if (!tags_ptr) {
// Tags missing entirely from this table. This can't happen for a
// normal Alternator table, but can happen if get_table_for_write()
// allowed writing to a non-Alternator table (e.g., an internal table).
// If it is a system table, LWT will not work (and is also pointless
// for non-distributed tables), so use UNSAFE_RMW.
if(is_internal_keyspace(schema->ks_name())) {
return write_isolation::UNSAFE_RMW;
} else {
return default_write_isolation;
}
}
auto it = tags_ptr->find(WRITE_ISOLATION_TAG_KEY);
if (it == tags_ptr->end() || it->second.empty()) {
return default_write_isolation;
}
return parse_write_isolation(it->second);
}
// shard_for_execute() checks whether execute() must be called on a specific
// other shard. Running execute() on a specific shard is necessary only if it
// will use LWT (storage_proxy::cas()). This is because cas() can only be
// called on the specific shard owning (as per get_cas_shard()) _pk's token.
// Knowing if execute() will call cas() or not may depend on whether there is
// a read-before-write, but not just on it - depending on configuration,
// execute() may unconditionally use cas() for every write. Unfortunately,
// this requires duplicating here a bit of logic from execute().
// The returned cas_shard must be passed to execute() to ensure
// the tablet shard won't change. The caller must hold the returned object for
// the duration of execution, even if we were already on the right shard - so it doesn't move.
std::optional<service::cas_shard> rmw_operation::shard_for_execute(bool needs_read_before_write) {
if (_write_isolation == write_isolation::FORBID_RMW ||
(_write_isolation == write_isolation::LWT_RMW_ONLY && !needs_read_before_write) ||
_write_isolation == write_isolation::UNSAFE_RMW) {
return {};
}
// If we're still here, cas() *will* be called by execute(), so let's
// find the appropriate shard to run it on:
const auto token = dht::get_token(*_schema, _pk);
return service::cas_shard(*_schema, token);
}
// Build the return value from the different RMW operations (UpdateItem,
// PutItem, DeleteItem). All these return nothing by default, but can
// optionally return Attributes if requested via the ReturnValues option.
static executor::request_return_type rmw_operation_return(rjson::value&& attributes, const consumed_capacity_counter& consumed_capacity, uint64_t& metric) {
rjson::value ret = rjson::empty_object();
consumed_capacity.add_consumed_capacity_to_response_if_needed(ret);
metric += consumed_capacity.get_consumed_capacity_units();
if (!attributes.IsNull()) {
rjson::add(ret, "Attributes", std::move(attributes));
}
return rjson::print(std::move(ret));
}
static future<std::unique_ptr<rjson::value>> get_previous_item(
service::storage_proxy& proxy,
service::client_state& client_state,
schema_ptr schema,
const partition_key& pk,
const clustering_key& ck,
service_permit permit,
db::consistency_level cl,
uint64_t& item_length)
{
auto selection = cql3::selection::selection::wildcard(schema);
auto command = previous_item_read_command(proxy, schema, ck, selection);
command->allow_limit = db::allow_per_partition_rate_limit::yes;
return proxy.query(schema, command, to_partition_ranges(*schema, pk), cl, service::storage_proxy::coordinator_query_options(executor::default_timeout(), std::move(permit), client_state)).then(
[schema, command, selection = std::move(selection), &item_length] (service::storage_proxy::coordinator_query_result qr) {
auto previous_item = executor::describe_single_item(schema, command->slice, *selection, *qr.query_result, {}, &item_length);
if (previous_item) {
return make_ready_future<std::unique_ptr<rjson::value>>(std::make_unique<rjson::value>(std::move(*previous_item)));
} else {
return make_ready_future<std::unique_ptr<rjson::value>>();
}
});
}
static future<std::unique_ptr<rjson::value>> get_previous_item(
service::storage_proxy& proxy,
service::client_state& client_state,
schema_ptr schema,
const partition_key& pk,
const clustering_key& ck,
service_permit permit,
alternator::stats& global_stats,
alternator::stats& per_table_stats,
uint64_t& item_length)
{
global_stats.reads_before_write++;
per_table_stats.reads_before_write++;
return get_previous_item(proxy, client_state, schema, pk, ck, permit, db::consistency_level::LOCAL_QUORUM, item_length);
}
static future<uint64_t> get_previous_item_size(
service::storage_proxy& proxy,
service::client_state& client_state,
schema_ptr schema,
const partition_key& pk,
const clustering_key& ck,
service_permit permit) {
uint64_t item_length = 0;
// The use of get_previous_item here is for DynamoDB calculation compatibility mode,
// and the actual value is ignored. For performance reasons, we use CL_LOCAL_ONE.
co_await get_previous_item(proxy, client_state, schema, pk, ck, permit, db::consistency_level::LOCAL_ONE, item_length);
co_return item_length;
}
future<executor::request_return_type> rmw_operation::execute(service::storage_proxy& proxy,
std::optional<service::cas_shard> cas_shard,
service::client_state& client_state,
tracing::trace_state_ptr trace_state,
service_permit permit,
bool needs_read_before_write,
stats& global_stats,
stats& per_table_stats,
uint64_t& wcu_total) {
auto cdc_opts = cdc::per_request_options{
.alternator = true,
.alternator_streams_increased_compatibility = schema()->cdc_options().enabled() && proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
};
if (needs_read_before_write) {
if (_write_isolation == write_isolation::FORBID_RMW) {
throw api_error::validation("Read-modify-write operations are disabled by 'forbid_rmw' write isolation policy. Refer to https://github.com/scylladb/scylla/blob/master/docs/alternator/alternator.md#write-isolation-policies for more information.");
}
global_stats.reads_before_write++;
per_table_stats.reads_before_write++;
if (_write_isolation == write_isolation::UNSAFE_RMW) {
// This is the old, unsafe, read before write which does first
// a read, then a write. TODO: remove this mode entirely.
return get_previous_item(proxy, client_state, schema(), _pk, _ck, permit, global_stats, per_table_stats, _consumed_capacity._total_bytes).then(
[this, &proxy, &wcu_total, trace_state, permit = std::move(permit), cdc_opts = std::move(cdc_opts)] (std::unique_ptr<rjson::value> previous_item) mutable {
std::optional<mutation> m = apply(std::move(previous_item), api::new_timestamp(), cdc_opts);
if (!m) {
return make_ready_future<executor::request_return_type>(api_error::conditional_check_failed("The conditional request failed", std::move(_return_attributes)));
}
return proxy.mutate(utils::chunked_vector<mutation>{std::move(*m)}, db::consistency_level::LOCAL_QUORUM, executor::default_timeout(), trace_state, std::move(permit), db::allow_per_partition_rate_limit::yes, false, std::move(cdc_opts)).then([this,&wcu_total] () mutable {
return rmw_operation_return(std::move(_return_attributes), _consumed_capacity, wcu_total);
});
});
}
} else if (_write_isolation != write_isolation::LWT_ALWAYS) {
std::optional<mutation> m = apply(nullptr, api::new_timestamp(), cdc_opts);
throwing_assert(m); // !needs_read_before_write, so apply() did not check a condition
return proxy.mutate(utils::chunked_vector<mutation>{std::move(*m)}, db::consistency_level::LOCAL_QUORUM, executor::default_timeout(), trace_state, std::move(permit), db::allow_per_partition_rate_limit::yes, false, std::move(cdc_opts)).then([this, &wcu_total] () mutable {
return rmw_operation_return(std::move(_return_attributes), _consumed_capacity, wcu_total);
});
}
throwing_assert(cas_shard);
// If we're still here, we need to do this write using LWT:
global_stats.write_using_lwt++;
per_table_stats.write_using_lwt++;
auto timeout = executor::default_timeout();
auto selection = cql3::selection::selection::wildcard(schema());
auto read_command = needs_read_before_write ?
previous_item_read_command(proxy, schema(), _ck, selection) :
nullptr;
return proxy.cas(schema(), std::move(*cas_shard), *this, read_command, to_partition_ranges(*schema(), _pk),
{timeout, std::move(permit), client_state, trace_state},
db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM, timeout, timeout, true, std::move(cdc_opts)).then([this, read_command, &wcu_total] (bool is_applied) mutable {
if (!is_applied) {
return make_ready_future<executor::request_return_type>(api_error::conditional_check_failed("The conditional request failed", std::move(_return_attributes)));
}
return make_ready_future<executor::request_return_type>(rmw_operation_return(std::move(_return_attributes), _consumed_capacity, wcu_total));
});
}
static parsed::condition_expression get_parsed_condition_expression(parsed::expression_cache& parsed_expression_cache, rjson::value& request) {
rjson::value* condition_expression = rjson::find(request, "ConditionExpression");
if (!condition_expression) {
// Returning an empty() condition_expression means no condition.
return parsed::condition_expression{};
}
if (!condition_expression->IsString()) {
throw api_error::validation("ConditionExpression must be a string");
}
if (condition_expression->GetStringLength() == 0) {
throw api_error::validation("ConditionExpression must not be empty");
}
try {
return parsed_expression_cache.parse_condition_expression(rjson::to_string_view(*condition_expression), "ConditionExpression");
} catch(expressions_syntax_error& e) {
throw api_error::validation(e.what());
}
}
static bool check_needs_read_before_write(const parsed::condition_expression& condition_expression) {
// Theoretically, a condition expression may exist but not refer to the
// item at all. But this is not a useful case and there is no point in
// optimizing for it.
return !condition_expression.empty();
}
// Fail the expression if it has unused attribute names or values. This is
// how DynamoDB behaves, so we do too.
static void verify_all_are_used(const rjson::value* field,
const std::unordered_set<std::string>& used, const char* field_name, const char* operation) {
if (!field) {
return;
}
for (auto it = field->MemberBegin(); it != field->MemberEnd(); ++it) {
if (!used.contains(rjson::to_string(it->name))) {
throw api_error::validation(
format("{} has spurious '{}', not used in {}",
field_name, rjson::to_string_view(it->name), operation));
}
}
}
class put_item_operation : public rmw_operation {
private:
put_or_delete_item _mutation_builder;
public:
parsed::condition_expression _condition_expression;
put_item_operation(parsed::expression_cache& parsed_expression_cache, service::storage_proxy& proxy, rjson::value&& request)
: rmw_operation(proxy, std::move(request))
, _mutation_builder(rjson::get(_request, "Item"), schema(), put_or_delete_item::put_item{},
si_key_attributes(proxy.data_dictionary().find_table(schema()->ks_name(), schema()->cf_name()))) {
_pk = _mutation_builder.pk();
_ck = _mutation_builder.ck();
if (_returnvalues != returnvalues::NONE && _returnvalues != returnvalues::ALL_OLD) {
throw api_error::validation(format("PutItem supports only NONE or ALL_OLD for ReturnValues"));
}
_condition_expression = get_parsed_condition_expression(parsed_expression_cache, _request);
const rjson::value* expression_attribute_names = rjson::find(_request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(_request, "ExpressionAttributeValues");
if (!_condition_expression.empty()) {
std::unordered_set<std::string> used_attribute_names;
std::unordered_set<std::string> used_attribute_values;
resolve_condition_expression(_condition_expression,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
verify_all_are_used(expression_attribute_names, used_attribute_names,"ExpressionAttributeNames", "PutItem");
verify_all_are_used(expression_attribute_values, used_attribute_values,"ExpressionAttributeValues", "PutItem");
} else {
if (expression_attribute_names) {
throw api_error::validation("ExpressionAttributeNames cannot be used without ConditionExpression");
}
if (expression_attribute_values) {
throw api_error::validation("ExpressionAttributeValues cannot be used without ConditionExpression");
}
}
_consumed_capacity += _mutation_builder.length_in_bytes();
}
bool needs_read_before_write() const {
return _request.HasMember("Expected") ||
check_needs_read_before_write(_condition_expression) ||
_returnvalues == returnvalues::ALL_OLD;
}
virtual std::optional<mutation> apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts, cdc::per_request_options& cdc_opts) const override {
if (!verify_expected(_request, previous_item.get()) ||
!verify_condition_expression(_condition_expression, previous_item.get())) {
if (previous_item && _returnvalues_on_condition_check_failure ==
returnvalues_on_condition_check_failure::ALL_OLD) {
_return_attributes = std::move(*previous_item);
}
// If the update is to be cancelled because of an unfulfilled Expected
// condition, return an empty optional mutation, which is more
// efficient than throwing an exception.
return {};
}
if (_returnvalues == returnvalues::ALL_OLD && previous_item) {
_return_attributes = std::move(*previous_item);
} else {
_return_attributes = {};
}
return _mutation_builder.build(_schema, ts);
}
virtual ~put_item_operation() = default;
};
future<executor::request_return_type> executor::put_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.put_item++;
auto start_time = std::chrono::steady_clock::now();
elogger.trace("put_item {}", request);
auto op = make_shared<put_item_operation>(*_parsed_expression_cache, _proxy, std::move(request));
tracing::add_alternator_table_name(trace_state, op->schema()->cf_name());
const bool needs_read_before_write = op->needs_read_before_write();
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, op->schema(), auth::permission::MODIFY, _stats);
auto cas_shard = op->shard_for_execute(needs_read_before_write);
if (cas_shard && !cas_shard->this_shard()) {
_stats.api_operations.put_item--; // uncount on this shard, will be counted in other shard
_stats.shard_bounce_for_lwt++;
co_return co_await container().invoke_on(cas_shard->shard(), _ssg,
[request = std::move(*op).move_request(), cs = client_state.move_to_other_shard(), gt = tracing::global_trace_state_ptr(trace_state), permit = std::move(permit)]
(executor& e) mutable {
return do_with(cs.get(), [&e, request = std::move(request), trace_state = tracing::trace_state_ptr(gt)]
(service::client_state& client_state) mutable {
//FIXME: Instead of passing empty_service_permit() to the background operation,
// the current permit's lifetime should be prolonged, so that it's destructed
// only after all background operations are finished as well.
return e.put_item(client_state, std::move(trace_state), empty_service_permit(), std::move(request));
});
});
}
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *(op->schema()));
per_table_stats->api_operations.put_item++;
uint64_t wcu_total = 0;
auto res = co_await op->execute(_proxy, std::move(cas_shard), client_state, trace_state, std::move(permit), needs_read_before_write, _stats, *per_table_stats, wcu_total);
per_table_stats->operation_sizes.put_item_op_size_kb.add(bytes_to_kb_ceil(op->consumed_capacity()._total_bytes));
per_table_stats->wcu_total[stats::wcu_types::PUT_ITEM] += wcu_total;
_stats.wcu_total[stats::wcu_types::PUT_ITEM] += wcu_total;
per_table_stats->api_operations.put_item_latency.mark(std::chrono::steady_clock::now() - start_time);
_stats.api_operations.put_item_latency.mark(std::chrono::steady_clock::now() - start_time);
co_return res;
}
class delete_item_operation : public rmw_operation {
private:
put_or_delete_item _mutation_builder;
public:
parsed::condition_expression _condition_expression;
delete_item_operation(parsed::expression_cache& parsed_expression_cache, service::storage_proxy& proxy, rjson::value&& request)
: rmw_operation(proxy, std::move(request))
, _mutation_builder(rjson::get(_request, "Key"), schema(), put_or_delete_item::delete_item{}) {
_pk = _mutation_builder.pk();
_ck = _mutation_builder.ck();
if (_returnvalues != returnvalues::NONE && _returnvalues != returnvalues::ALL_OLD) {
throw api_error::validation(format("DeleteItem supports only NONE or ALL_OLD for ReturnValues"));
}
_condition_expression = get_parsed_condition_expression(parsed_expression_cache, _request);
const rjson::value* expression_attribute_names = rjson::find(_request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(_request, "ExpressionAttributeValues");
if (!_condition_expression.empty()) {
std::unordered_set<std::string> used_attribute_names;
std::unordered_set<std::string> used_attribute_values;
resolve_condition_expression(_condition_expression,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
verify_all_are_used(expression_attribute_names, used_attribute_names,"ExpressionAttributeNames", "DeleteItem");
verify_all_are_used(expression_attribute_values, used_attribute_values, "ExpressionAttributeValues", "DeleteItem");
} else {
if (expression_attribute_names) {
throw api_error::validation("ExpressionAttributeNames cannot be used without ConditionExpression");
}
if (expression_attribute_values) {
throw api_error::validation("ExpressionAttributeValues cannot be used without ConditionExpression");
}
}
}
bool needs_read_before_write() const {
return _request.HasMember("Expected") ||
check_needs_read_before_write(_condition_expression) ||
_returnvalues == returnvalues::ALL_OLD;
}
virtual std::optional<mutation> apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts, cdc::per_request_options& cdc_opts) const override {
if (!verify_expected(_request, previous_item.get()) ||
!verify_condition_expression(_condition_expression, previous_item.get())) {
if (previous_item && _returnvalues_on_condition_check_failure ==
returnvalues_on_condition_check_failure::ALL_OLD) {
_return_attributes = std::move(*previous_item);
}
// If the update is to be cancelled because of an unfulfilled Expected
// condition, return an empty optional mutation, which is more
// efficient than throwing an exception.
return {};
}
if (_returnvalues == returnvalues::ALL_OLD && previous_item) {
_return_attributes = std::move(*previous_item);
} else {
_return_attributes = {};
}
if (_consumed_capacity._total_bytes == 0) {
_consumed_capacity._total_bytes = 1;
}
return _mutation_builder.build(_schema, ts);
}
virtual ~delete_item_operation() = default;
};
future<executor::request_return_type> executor::delete_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.delete_item++;
auto start_time = std::chrono::steady_clock::now();
elogger.trace("delete_item {}", request);
auto op = make_shared<delete_item_operation>(*_parsed_expression_cache, _proxy, std::move(request));
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *(op->schema()));
tracing::add_alternator_table_name(trace_state, op->schema()->cf_name());
const bool needs_read_before_write = _proxy.data_dictionary().get_config().alternator_force_read_before_write() || op->needs_read_before_write();
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, op->schema(), auth::permission::MODIFY, _stats);
auto cas_shard = op->shard_for_execute(needs_read_before_write);
if (cas_shard && !cas_shard->this_shard()) {
_stats.api_operations.delete_item--; // uncount on this shard, will be counted in other shard
_stats.shard_bounce_for_lwt++;
per_table_stats->shard_bounce_for_lwt++;
co_return co_await container().invoke_on(cas_shard->shard(), _ssg,
[request = std::move(*op).move_request(), cs = client_state.move_to_other_shard(), gt = tracing::global_trace_state_ptr(trace_state), permit = std::move(permit)]
(executor& e) mutable {
return do_with(cs.get(), [&e, request = std::move(request), trace_state = tracing::trace_state_ptr(gt)]
(service::client_state& client_state) mutable {
//FIXME: Instead of passing empty_service_permit() to the background operation,
// the current permit's lifetime should be prolonged, so that it's destructed
// only after all background operations are finished as well.
return e.delete_item(client_state, std::move(trace_state), empty_service_permit(), std::move(request));
});
});
}
per_table_stats->api_operations.delete_item++;
uint64_t wcu_total = 0;
auto res = co_await op->execute(_proxy, std::move(cas_shard), client_state, trace_state, std::move(permit), needs_read_before_write, _stats, *per_table_stats, wcu_total);
if (op->consumed_capacity()._total_bytes > 1) {
per_table_stats->operation_sizes.delete_item_op_size_kb.add(bytes_to_kb_ceil(op->consumed_capacity()._total_bytes));
}
per_table_stats->wcu_total[stats::wcu_types::DELETE_ITEM] += wcu_total;
_stats.wcu_total[stats::wcu_types::DELETE_ITEM] += wcu_total;
per_table_stats->api_operations.delete_item_latency.mark(std::chrono::steady_clock::now() - start_time);
_stats.api_operations.delete_item_latency.mark(std::chrono::steady_clock::now() - start_time);
co_return res;
}
static schema_ptr get_table_from_batch_request(const service::storage_proxy& proxy, const rjson::value::ConstMemberIterator& batch_request) {
sstring table_name = rjson::to_sstring(batch_request->name); // JSON keys are always strings
try {
return proxy.data_dictionary().find_schema(sstring(executor::KEYSPACE_NAME_PREFIX) + table_name, table_name);
} catch(data_dictionary::no_such_column_family&) {
// DynamoDB returns validation error even when table does not exist
// and the table name is invalid.
validate_table_name(table_name);
throw api_error::resource_not_found(format("Requested resource not found: Table: {} not found", table_name));
}
}
using primary_key = std::pair<partition_key, clustering_key>;
struct primary_key_hash {
schema_ptr _s;
size_t operator()(const primary_key& key) const {
return utils::hash_combine(partition_key::hashing(*_s)(key.first), clustering_key::hashing(*_s)(key.second));
}
};
struct primary_key_equal {
schema_ptr _s;
bool operator()(const primary_key& k1, const primary_key& k2) const {
return partition_key::equality(*_s)(k1.first, k2.first) && clustering_key::equality(*_s)(k1.second, k2.second);
}
};
// This is a cas_request subclass for applying given put_or_delete_items to
// one partition using LWT as part as BatchWriteItem. This is a write-only
// operation, not needing the previous value of the item (the mutation to be
// done is known prior to starting the operation). Nevertheless, we want to
// do this mutation via LWT to ensure that it is serialized with other LWT
// mutations to the same partition.
//
// The std::vector<put_or_delete_item> must remain alive until the
// storage_proxy::cas() future is resolved.
class put_or_delete_item_cas_request : public service::cas_request {
schema_ptr schema;
const std::vector<put_or_delete_item>& _mutation_builders;
public:
put_or_delete_item_cas_request(schema_ptr s, const std::vector<put_or_delete_item>& b) :
schema(std::move(s)), _mutation_builders(b) { }
virtual ~put_or_delete_item_cas_request() = default;
virtual std::optional<mutation> apply(foreign_ptr<lw_shared_ptr<query::result>> qr, const query::partition_slice& slice, api::timestamp_type ts, cdc::per_request_options& cdc_opts) override {
std::optional<mutation> ret;
for (const put_or_delete_item& mutation_builder : _mutation_builders) {
// We assume all these builders have the same partition.
if (ret) {
ret->apply(mutation_builder.build(schema, ts));
} else {
ret = mutation_builder.build(schema, ts);
}
}
return ret;
}
};
future<> executor::cas_write(schema_ptr schema, service::cas_shard cas_shard, const dht::decorated_key& dk,
const std::vector<put_or_delete_item>& mutation_builders, service::client_state& client_state,
tracing::trace_state_ptr trace_state, service_permit permit)
{
if (!cas_shard.this_shard()) {
_stats.shard_bounce_for_lwt++;
return container().invoke_on(cas_shard.shard(), _ssg,
[cs = client_state.move_to_other_shard(),
&mb = mutation_builders,
&dk,
ks = schema->ks_name(),
cf = schema->cf_name(),
gt = tracing::global_trace_state_ptr(trace_state),
permit = std::move(permit)]
(executor& self) mutable {
return do_with(cs.get(), [&mb, &dk, ks = std::move(ks), cf = std::move(cf),
trace_state = tracing::trace_state_ptr(gt), &self]
(service::client_state& client_state) mutable {
auto schema = self._proxy.data_dictionary().find_schema(ks, cf);
service::cas_shard cas_shard(*schema, dk.token());
//FIXME: Instead of passing empty_service_permit() to the background operation,
// the current permit's lifetime should be prolonged, so that it's destructed
// only after all background operations are finished as well.
return self.cas_write(schema, std::move(cas_shard), dk, mb, client_state, std::move(trace_state), empty_service_permit());
});
});
}
auto timeout = executor::default_timeout();
auto op = std::make_unique<put_or_delete_item_cas_request>(schema, mutation_builders);
auto* op_ptr = op.get();
auto cdc_opts = cdc::per_request_options{
.alternator = true,
.alternator_streams_increased_compatibility =
schema->cdc_options().enabled() && _proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
};
return _proxy.cas(schema, std::move(cas_shard), *op_ptr, nullptr, to_partition_ranges(dk),
{timeout, std::move(permit), client_state, trace_state},
db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM,
timeout, timeout, true, std::move(cdc_opts)).finally([op = std::move(op)]{}).discard_result();
// We discarded cas()'s future value ("is_applied") because BatchWriteItem
// does not need to support conditional updates.
}
struct schema_decorated_key {
schema_ptr schema;
dht::decorated_key dk;
};
struct schema_decorated_key_hash {
size_t operator()(const schema_decorated_key& k) const {
return std::hash<dht::token>()(k.dk.token());
}
};
struct schema_decorated_key_equal {
bool operator()(const schema_decorated_key& k1, const schema_decorated_key& k2) const {
return k1.schema == k2.schema && k1.dk.equal(*k1.schema, k2.dk);
}
};
// FIXME: if we failed writing some of the mutations, need to return a list
// of these failed mutations rather than fail the whole write (issue #5650).
future<> executor::do_batch_write(
std::vector<std::pair<schema_ptr, put_or_delete_item>> mutation_builders,
service::client_state& client_state,
tracing::trace_state_ptr trace_state,
service_permit permit) {
if (mutation_builders.empty()) {
return make_ready_future<>();
}
// NOTE: technically, do_batch_write could be reworked to use LWT only for part
// of the batched requests and not use it for others, but it's not considered
// likely that a batch will contain both tables which always demand LWT and ones
// that don't - it's fragile to split a batch into multiple storage proxy requests though.
// Hence, the decision is conservative - if any table enforces LWT,the whole batch will use it.
const bool needs_lwt = std::ranges::any_of(mutation_builders | std::views::keys, [] (const schema_ptr& schema) {
return rmw_operation::get_write_isolation_for_schema(schema) == rmw_operation::write_isolation::LWT_ALWAYS;
});
if (!needs_lwt) {
// Do a normal write, without LWT:
utils::chunked_vector<mutation> mutations;
mutations.reserve(mutation_builders.size());
api::timestamp_type now = api::new_timestamp();
bool any_cdc_enabled = false;
for (auto& b : mutation_builders) {
mutations.push_back(b.second.build(b.first, now));
any_cdc_enabled |= b.first->cdc_options().enabled();
}
return _proxy.mutate(std::move(mutations),
db::consistency_level::LOCAL_QUORUM,
executor::default_timeout(),
trace_state,
std::move(permit),
db::allow_per_partition_rate_limit::yes,
false,
cdc::per_request_options{
.alternator = true,
.alternator_streams_increased_compatibility = any_cdc_enabled && _proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
});
} else {
// Do the write via LWT:
// Multiple mutations may be destined for the same partition, adding
// or deleting different items of one partition. Join them together
// because we can do them in one cas() call.
using map_type = std::unordered_map<schema_decorated_key,
std::vector<put_or_delete_item>,
schema_decorated_key_hash,
schema_decorated_key_equal>;
auto key_builders = std::make_unique<map_type>(1, schema_decorated_key_hash{}, schema_decorated_key_equal{});
for (auto&& b : std::move(mutation_builders)) {
auto [it, added] = key_builders->try_emplace(schema_decorated_key {
.schema = b.first,
.dk = dht::decorate_key(*b.first, b.second.pk())
});
it->second.push_back(std::move(b.second));
}
auto* key_builders_ptr = key_builders.get();
return parallel_for_each(*key_builders_ptr, [this, &client_state, trace_state, permit = std::move(permit)] (const auto& e) {
_stats.write_using_lwt++;
auto desired_shard = service::cas_shard(*e.first.schema, e.first.dk.token());
auto s = e.first.schema;
static const auto* injection_name = "alternator_executor_batch_write_wait";
return utils::get_local_injector().inject(injection_name, [s = std::move(s)] (auto& handler) -> future<> {
const auto ks = handler.get("keyspace");
const auto cf = handler.get("table");
const auto shard = std::atoll(handler.get("shard")->data());
if (ks == s->ks_name() && cf == s->cf_name() && shard == this_shard_id()) {
elogger.info("{}: hit", injection_name);
co_await handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::minutes{5});
elogger.info("{}: continue", injection_name);
}
}).then([&e, desired_shard = std::move(desired_shard),
&client_state, trace_state = std::move(trace_state), permit = std::move(permit), this]() mutable
{
return cas_write(e.first.schema, std::move(desired_shard), e.first.dk,
std::move(e.second), client_state, std::move(trace_state), std::move(permit));
});
}).finally([key_builders = std::move(key_builders)]{});
}
}
future<executor::request_return_type> executor::batch_write_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.batch_write_item++;
auto start_time = std::chrono::steady_clock::now();
const rjson::value& request_items = get_member(request, "RequestItems", "BatchWriteItem content");
validate_is_object(request_items, "RequestItems");
if (request_items.ObjectEmpty()) {
co_return api_error::validation("RequestItems can't be empty");
}
const auto maximum_batch_write_size = _proxy.data_dictionary().get_config().alternator_max_items_in_batch_write();
size_t total_items = 0;
for (auto it = request_items.MemberBegin(); it != request_items.MemberEnd(); ++it) {
if (!it->value.IsArray() || it->value.Empty()) {
co_return api_error::validation("Member of RequestItems must be a non-empty array of WriteRequest objects");
}
total_items += it->value.Size();
}
if (total_items > maximum_batch_write_size) {
co_return api_error::validation(fmt::format("Invalid length of BatchWriteItem command, got {} items, "
"maximum is {} (from configuration variable alternator_max_items_in_batch_write)", total_items, maximum_batch_write_size));
}
bool should_add_wcu = wcu_consumed_capacity_counter::should_add_capacity(request);
rjson::value consumed_capacity = rjson::empty_array();
std::vector<std::pair<schema_ptr, put_or_delete_item>> mutation_builders;
// WCU calculation is performed at the end of execution.
// We need to keep track of changes per table, both for internal metrics
// and to be able to return the values if should_add_wcu is true.
// For each table, we need its stats and schema.
std::vector<std::pair<lw_shared_ptr<stats>, schema_ptr>> per_table_wcu;
mutation_builders.reserve(request_items.MemberCount());
per_table_wcu.reserve(request_items.MemberCount());
for (auto it = request_items.MemberBegin(); it != request_items.MemberEnd(); ++it) {
schema_ptr schema = get_table_from_batch_request(_proxy, it);
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *(schema));
per_table_stats->api_operations.batch_write_item++;
per_table_stats->api_operations.batch_write_item_batch_total += it->value.Size();
per_table_stats->api_operations.batch_write_item_histogram.add(it->value.Size());
tracing::add_alternator_table_name(trace_state, schema->cf_name());
std::unordered_set<primary_key, primary_key_hash, primary_key_equal> used_keys(
1, primary_key_hash{schema}, primary_key_equal{schema});
for (auto& request : it->value.GetArray()) {
auto& r = get_single_member(request, "RequestItems element");
const auto r_name = rjson::to_string_view(r.name);
if (r_name == "PutRequest") {
const rjson::value& item = get_member(r.value, "Item", "PutRequest");
validate_is_object(item, "Item in PutRequest");
auto&& put_item = put_or_delete_item(
item, schema, put_or_delete_item::put_item{},
si_key_attributes(_proxy.data_dictionary().find_table(schema->ks_name(), schema->cf_name())));
mutation_builders.emplace_back(schema, std::move(put_item));
auto mut_key = std::make_pair(mutation_builders.back().second.pk(), mutation_builders.back().second.ck());
if (used_keys.contains(mut_key)) {
co_return api_error::validation("Provided list of item keys contains duplicates");
}
used_keys.insert(std::move(mut_key));
} else if (r_name == "DeleteRequest") {
const rjson::value& key = get_member(r.value, "Key", "DeleteRequest");
validate_is_object(key, "Key in DeleteRequest");
mutation_builders.emplace_back(schema, put_or_delete_item(
key, schema, put_or_delete_item::delete_item{}));
auto mut_key = std::make_pair(mutation_builders.back().second.pk(),
mutation_builders.back().second.ck());
if (used_keys.contains(mut_key)) {
co_return api_error::validation("Provided list of item keys contains duplicates");
}
used_keys.insert(std::move(mut_key));
} else {
co_return api_error::validation(fmt::format("Unknown BatchWriteItem request type: {}", r_name));
}
}
per_table_wcu.emplace_back(std::make_pair(per_table_stats, schema));
}
for (const auto& b : mutation_builders) {
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, b.first, auth::permission::MODIFY, _stats);
}
// If alternator_force_read_before_write is true we will first get the previous item size
// and only then do send the mutation.
if (_proxy.data_dictionary().get_config().alternator_force_read_before_write()) {
std::vector<future<uint64_t>> previous_items_sizes;
previous_items_sizes.reserve(mutation_builders.size());
// Parallel get all previous item sizes
for (const auto& b : mutation_builders) {
previous_items_sizes.emplace_back(get_previous_item_size(
_proxy,
client_state,
b.first,
b.second.pk(),
b.second.ck(),
permit));
}
size_t pos = 0;
// We are going to wait for all the requests
for (auto&& pi : previous_items_sizes) {
auto res = co_await std::move(pi);
if (mutation_builders[pos].second.length_in_bytes() < res) {
mutation_builders[pos].second.set_length_in_bytes(res);
}
pos++;
}
}
size_t wcu_put_units = 0;
size_t wcu_delete_units = 0;
size_t pos = 0;
size_t total_wcu;
// Here we calculate the per-table WCU.
// The size in the mutation is based either on the operation size,
// or, if we performed a read-before-write, on the larger of the operation size
// and the previous item's size.
for (const auto& w : per_table_wcu) {
total_wcu = 0;
// The following loop goes over all items from the same table
while(pos < mutation_builders.size() && w.second->id() == mutation_builders[pos].first->id()) {
uint64_t item_size = mutation_builders[pos].second.length_in_bytes();
size_t wcu = wcu_consumed_capacity_counter::get_units(item_size ? item_size : 1);
total_wcu += wcu;
if (mutation_builders[pos].second.is_put_item()) {
w.first->wcu_total[stats::PUT_ITEM] += wcu;
wcu_put_units += wcu;
} else {
w.first->wcu_total[stats::DELETE_ITEM] += wcu;
wcu_delete_units += wcu;
}
w.first->operation_sizes.batch_write_item_op_size_kb.add(bytes_to_kb_ceil(item_size));
pos++;
}
if (should_add_wcu) {
rjson::value entry = rjson::empty_object();
rjson::add(entry, "TableName", rjson::from_string(w.second->cf_name()));
rjson::add(entry, "CapacityUnits", total_wcu);
rjson::push_back(consumed_capacity, std::move(entry));
}
}
_stats.wcu_total[stats::PUT_ITEM] += wcu_put_units;
_stats.wcu_total[stats::DELETE_ITEM] += wcu_delete_units;
_stats.api_operations.batch_write_item_batch_total += total_items;
_stats.api_operations.batch_write_item_histogram.add(total_items);
co_await do_batch_write(std::move(mutation_builders), client_state, trace_state, std::move(permit));
// FIXME: Issue #5650: If we failed writing some of the updates,
// need to return a list of these failed updates in UnprocessedItems
// rather than fail the whole write (issue #5650).
rjson::value ret = rjson::empty_object();
rjson::add(ret, "UnprocessedItems", rjson::empty_object());
if (should_add_wcu) {
rjson::add(ret, "ConsumedCapacity", std::move(consumed_capacity));
}
auto duration = std::chrono::steady_clock::now() - start_time;
_stats.api_operations.batch_write_item_latency.mark(duration);
for (const auto& w : per_table_wcu) {
w.first->api_operations.batch_write_item_latency.mark(duration);
}
co_return rjson::print(std::move(ret));
}
static const std::string_view get_item_type_string(const rjson::value& v) {
const rjson::value::Member& mem = get_single_member(v, "Item");
return rjson::to_string_view(mem.name);
}
// attrs_to_get saves for each top-level attribute an attrs_to_get_node,
// a hierarchy of subparts that need to be kept. The following function
// takes a given JSON value and drops its parts which weren't asked to be
// kept. It modifies the given JSON value, or returns false to signify that
// the entire object should be dropped.
// Note that The JSON value is assumed to be encoded using the DynamoDB
// conventions - i.e., it is really a map whose key has a type string,
// and the value is the real object.
template<typename T>
static bool hierarchy_filter(rjson::value& val, const attribute_path_map_node<T>& h) {
if (!val.IsObject() || val.MemberCount() != 1) {
// This shouldn't happen. We shouldn't have stored malformed objects.
// But today Alternator does not validate the structure of nested
// documents before storing them, so this can happen on read.
throw api_error::internal(format("Malformed value object read: {}", val));
}
const char* type = val.MemberBegin()->name.GetString();
rjson::value& v = val.MemberBegin()->value;
if (h.has_members()) {
const auto& members = h.get_members();
if (type[0] != 'M' || !v.IsObject()) {
// If v is not an object (dictionary, map), none of the members
// can match.
return false;
}
rjson::value newv = rjson::empty_object();
for (auto it = v.MemberBegin(); it != v.MemberEnd(); ++it) {
std::string attr = rjson::to_string(it->name);
auto x = members.find(attr);
if (x != members.end()) {
if (x->second) {
// Only a part of this attribute is to be filtered, do it.
if (hierarchy_filter(it->value, *x->second)) {
// because newv started empty and attr are unique
// (keys of v), we can use add() here
rjson::add_with_string_name(newv, attr, std::move(it->value));
}
} else {
// The entire attribute is to be kept
rjson::add_with_string_name(newv, attr, std::move(it->value));
}
}
}
if (newv.MemberCount() == 0) {
return false;
}
v = newv;
} else if (h.has_indexes()) {
const auto& indexes = h.get_indexes();
if (type[0] != 'L' || !v.IsArray()) {
return false;
}
rjson::value newv = rjson::empty_array();
const auto& a = v.GetArray();
for (unsigned i = 0; i < v.Size(); i++) {
auto x = indexes.find(i);
if (x != indexes.end()) {
if (x->second) {
if (hierarchy_filter(a[i], *x->second)) {
rjson::push_back(newv, std::move(a[i]));
}
} else {
// The entire attribute is to be kept
rjson::push_back(newv, std::move(a[i]));
}
}
}
if (newv.Size() == 0) {
return false;
}
v = newv;
}
return true;
}
// Add a path to an attribute_path_map. Throws a validation error if the path
// "overlaps" with one already in the filter (one is a sub-path of the other)
// or "conflicts" with it (both a member and index is requested).
template<typename T>
void attribute_path_map_add(const char* source, attribute_path_map<T>& map, const parsed::path& p, T value = {}) {
using node = attribute_path_map_node<T>;
// The first step is to look for the top-level attribute (p.root()):
auto it = map.find(p.root());
if (it == map.end()) {
if (p.has_operators()) {
it = map.emplace(p.root(), node {std::nullopt}).first;
} else {
(void) map.emplace(p.root(), node {std::move(value)}).first;
// Value inserted for top-level node. We're done.
return;
}
} else if(!p.has_operators()) {
// If p is top-level and we already have it or a part of it
// in map, it's a forbidden overlapping path.
throw api_error::validation(fmt::format(
"Invalid {}: two document paths overlap at {}", source, p.root()));
} else if (it->second.has_value()) {
// If we're here, it != map.end() && p.has_operators && it->second.has_value().
// This means the top-level attribute already has a value, and we're
// trying to add a non-top-level value. It's an overlap.
throw api_error::validation(fmt::format("Invalid {}: two document paths overlap at {}", source, p.root()));
}
node* h = &it->second;
// The second step is to walk h from the top-level node to the inner node
// where we're supposed to insert the value:
for (const auto& op : p.operators()) {
std::visit(overloaded_functor {
[&] (const std::string& member) {
if (h->is_empty()) {
*h = node {typename node::members_t()};
} else if (h->has_indexes()) {
throw api_error::validation(format("Invalid {}: two document paths conflict at {}", source, p));
} else if (h->has_value()) {
throw api_error::validation(format("Invalid {}: two document paths overlap at {}", source, p));
}
typename node::members_t& members = h->get_members();
auto it = members.find(member);
if (it == members.end()) {
it = members.insert({member, std::make_unique<node>()}).first;
}
h = it->second.get();
},
[&] (unsigned index) {
if (h->is_empty()) {
*h = node {typename node::indexes_t()};
} else if (h->has_members()) {
throw api_error::validation(format("Invalid {}: two document paths conflict at {}", source, p));
} else if (h->has_value()) {
throw api_error::validation(format("Invalid {}: two document paths overlap at {}", source, p));
}
typename node::indexes_t& indexes = h->get_indexes();
auto it = indexes.find(index);
if (it == indexes.end()) {
it = indexes.insert({index, std::make_unique<node>()}).first;
}
h = it->second.get();
}
}, op);
}
// Finally, insert the value in the node h.
if (h->is_empty()) {
*h = node {std::move(value)};
} else {
throw api_error::validation(format("Invalid {}: two document paths overlap at {}", source, p));
}
}
// A very simplified version of the above function for the special case of
// adding only top-level attribute. It's not only simpler, we also use a
// different error message, referring to a "duplicate attribute"instead of
// "overlapping paths". DynamoDB also has this distinction (errors in
// AttributesToGet refer to duplicates, not overlaps, but errors in
// ProjectionExpression refer to overlap - even if it's an exact duplicate).
template<typename T>
void attribute_path_map_add(const char* source, attribute_path_map<T>& map, const std::string& attr, T value = {}) {
using node = attribute_path_map_node<T>;
auto it = map.find(attr);
if (it == map.end()) {
map.emplace(attr, node {std::move(value)});
} else {
throw api_error::validation(fmt::format(
"Invalid {}: Duplicate attribute: {}", source, attr));
}
}
// Parse the "Select" parameter of a Scan or Query operation, throwing a
// ValidationException in various forbidden combinations of options and
// finally returning one of three options:
// 1. regular - the default scan behavior of returning all or specific
// attributes ("ALL_ATTRIBUTES" or "SPECIFIC_ATTRIBUTES").
// 2. count - just count the items ("COUNT")
// 3. projection - return projected attributes ("ALL_PROJECTED_ATTRIBUTES")
// An ValidationException is thrown when recognizing an invalid combination
// of options - such as ALL_PROJECTED_ATTRIBUTES for a base table, or
// SPECIFIC_ATTRIBUTES without ProjectionExpression or AttributesToGet.
enum class select_type { regular, count, projection };
static select_type parse_select(const rjson::value& request, table_or_view_type table_type) {
const rjson::value* select_value = rjson::find(request, "Select");
if (!select_value) {
// If "Select" is not specified, it defaults to ALL_ATTRIBUTES
// on a base table, or ALL_PROJECTED_ATTRIBUTES on an index
return table_type == table_or_view_type::base ?
select_type::regular : select_type::projection;
}
if (!select_value->IsString()) {
throw api_error::validation("Select parameter must be a string");
}
std::string_view select = rjson::to_string_view(*select_value);
const bool has_attributes_to_get = request.HasMember("AttributesToGet");
const bool has_projection_expression = request.HasMember("ProjectionExpression");
if (select == "SPECIFIC_ATTRIBUTES") {
if (has_projection_expression || has_attributes_to_get) {
return select_type::regular;
}
throw api_error::validation("Select=SPECIFIC_ATTRIBUTES requires AttributesToGet or ProjectionExpression");
}
if (has_projection_expression || has_attributes_to_get) {
throw api_error::validation("AttributesToGet or ProjectionExpression require Select to be either SPECIFIC_ATTRIBUTES or missing");
}
if (select == "COUNT") {
return select_type::count;
}
if (select == "ALL_ATTRIBUTES") {
// FIXME: when we support projections (#5036), if this is a GSI and
// not all attributes are projected to it, we should throw.
return select_type::regular;
}
if (select == "ALL_PROJECTED_ATTRIBUTES") {
if (table_type == table_or_view_type::base) {
throw api_error::validation("ALL_PROJECTED_ATTRIBUTES only allowed for indexes");
}
return select_type::projection;
}
throw api_error::validation(fmt::format("Unknown Select value '{}'. Allowed choices: ALL_ATTRIBUTES, SPECIFIC_ATTRIBUTES, ALL_PROJECTED_ATTRIBUTES, COUNT",
select));
}
// calculate_attrs_to_get() takes either AttributesToGet or
// ProjectionExpression parameters (having both is *not* allowed),
// and returns the list of cells we need to read, or a disengaged optional
// when *all* attributes are to be returned.
// However, in our current implementation, only top-level attributes are
// stored as separate cells - a nested document is stored serialized together
// (as JSON) in the same cell. So this function return a map - each key is the
// top-level attribute we will need need to read, and the value for each
// top-level attribute is the partial hierarchy (struct hierarchy_filter)
// that we will need to extract from that serialized JSON.
// For example, if ProjectionExpression lists a.b and a.c[2], we
// return one top-level attribute name, "a", with the value "{b, c[2]}".
static std::optional<attrs_to_get> calculate_attrs_to_get(const rjson::value& req, parsed::expression_cache& parsed_expression_cache, std::unordered_set<std::string>& used_attribute_names, select_type select = select_type::regular) {
if (select == select_type::count) {
// An empty map asks to retrieve no attributes. Note that this is
// different from a disengaged optional which means retrieve all.
return attrs_to_get();
}
// FIXME: also need to handle select_type::projection
const bool has_attributes_to_get = req.HasMember("AttributesToGet");
const bool has_projection_expression = req.HasMember("ProjectionExpression");
if (has_attributes_to_get && has_projection_expression) {
throw api_error::validation(
format("GetItem does not allow both ProjectionExpression and AttributesToGet to be given together"));
}
if (has_attributes_to_get) {
const rjson::value& attributes_to_get = req["AttributesToGet"];
attrs_to_get ret;
for (auto it = attributes_to_get.Begin(); it != attributes_to_get.End(); ++it) {
attribute_path_map_add("AttributesToGet", ret, rjson::to_string(*it));
validate_attr_name_length("AttributesToGet", it->GetStringLength(), false);
}
if (ret.empty()) {
throw api_error::validation("Empty AttributesToGet is not allowed. Consider using Select=COUNT instead.");
}
return ret;
} else if (has_projection_expression) {
const rjson::value& projection_expression = req["ProjectionExpression"];
const rjson::value* expression_attribute_names = rjson::find(req, "ExpressionAttributeNames");
std::vector<parsed::path> paths_to_get;
try {
paths_to_get = parsed_expression_cache.parse_projection_expression(rjson::to_string_view(projection_expression));
} catch(expressions_syntax_error& e) {
throw api_error::validation(e.what());
}
resolve_projection_expression(paths_to_get, expression_attribute_names, used_attribute_names);
attrs_to_get ret;
for (const parsed::path& p : paths_to_get) {
attribute_path_map_add("ProjectionExpression", ret, p);
}
return ret;
}
// An disengaged optional asks to read everything
return std::nullopt;
}
/**
* Helper routine to extract data when we already have
* row, etc etc.
*
* Note: include_all_embedded_attributes means we should
* include all values in the `ATTRS_COLUMN_NAME` map column.
*
* We could change the behaviour to simply include all values
* from this column if the `ATTRS_COLUMN_NAME` is explicit in
* `attrs_to_get`, but I am scared to do that now in case
* there is some corner case in existing code.
*
* Explicit bool means we can be sure all previous calls are
* as before.
*/
void executor::describe_single_item(const cql3::selection::selection& selection,
const std::vector<managed_bytes_opt>& result_row,
const std::optional<attrs_to_get>& attrs_to_get,
rjson::value& item,
uint64_t* item_length_in_bytes,
bool include_all_embedded_attributes)
{
const auto& columns = selection.get_columns();
auto column_it = columns.begin();
for (const managed_bytes_opt& cell : result_row) {
if (!cell) {
++column_it;
continue;
}
std::string column_name = (*column_it)->name_as_text();
if (column_name != executor::ATTRS_COLUMN_NAME) {
if (item_length_in_bytes) {
(*item_length_in_bytes) += column_name.length() + cell->size();
}
if (!attrs_to_get || attrs_to_get->contains(column_name)) {
// item is expected to start empty, and column_name are unique
// so add() makes sense
rjson::add_with_string_name(item, column_name, rjson::empty_object());
rjson::value& field = item[column_name.c_str()];
cell->with_linearized([&] (bytes_view linearized_cell) {
rjson::add_with_string_name(field, type_to_string((*column_it)->type), json_key_column_value(linearized_cell, **column_it));
});
}
} else {
auto deserialized = attrs_type()->deserialize(*cell);
auto keys_and_values = value_cast<map_type_impl::native_type>(deserialized);
for (auto entry : keys_and_values) {
std::string attr_name = value_cast<sstring>(entry.first);
if (item_length_in_bytes) {
(*item_length_in_bytes) += attr_name.length();
}
if (include_all_embedded_attributes || !attrs_to_get || attrs_to_get->contains(attr_name)) {
bytes value = value_cast<bytes>(entry.second);
if (item_length_in_bytes && value.length()) {
// ScyllaDB uses one extra byte compared to DynamoDB for the bytes length
(*item_length_in_bytes) += value.length() - 1;
}
rjson::value v = deserialize_item(value);
if (attrs_to_get) {
auto it = attrs_to_get->find(attr_name);
if (it != attrs_to_get->end()) {
// attrs_to_get may have asked for only part of
// this attribute. hierarchy_filter() modifies v,
// and returns false when nothing is to be kept.
if (!hierarchy_filter(v, it->second)) {
continue;
}
}
}
// item is expected to start empty, and attribute
// names are unique so add() makes sense
rjson::add_with_string_name(item, attr_name, std::move(v));
} else if (item_length_in_bytes) {
(*item_length_in_bytes) += value_cast<bytes>(entry.second).length() - 1;
}
}
}
++column_it;
}
}
std::optional<rjson::value> executor::describe_single_item(schema_ptr schema,
const query::partition_slice& slice,
const cql3::selection::selection& selection,
const query::result& query_result,
const std::optional<attrs_to_get>& attrs_to_get,
uint64_t* item_length_in_bytes) {
rjson::value item = rjson::empty_object();
cql3::selection::result_set_builder builder(selection, gc_clock::now());
query::result_view::consume(query_result, slice, cql3::selection::result_set_builder::visitor(builder, *schema, selection));
auto result_set = builder.build();
if (result_set->empty()) {
if (item_length_in_bytes) {
// empty results is counted as having a minimal length (e.g. 1 byte).
(*item_length_in_bytes) += 1;
}
// If there is no matching item, we're supposed to return an empty
// object without an Item member - not one with an empty Item member
return {};
}
if (result_set->size() > 1) {
// If the result set contains multiple rows, the code should have
// called describe_multi_item(), not this function.
throw std::logic_error("describe_single_item() asked to describe multiple items");
}
describe_single_item(selection, *result_set->rows().begin(), attrs_to_get, item, item_length_in_bytes);
return item;
}
future<std::vector<rjson::value>> executor::describe_multi_item(schema_ptr schema,
const query::partition_slice&& slice,
shared_ptr<cql3::selection::selection> selection,
foreign_ptr<lw_shared_ptr<query::result>> query_result,
shared_ptr<const std::optional<attrs_to_get>> attrs_to_get,
noncopyable_function<void(uint64_t)> item_callback) {
cql3::selection::result_set_builder builder(*selection, gc_clock::now());
query::result_view::consume(*query_result, slice, cql3::selection::result_set_builder::visitor(builder, *schema, *selection));
auto result_set = builder.build();
std::vector<rjson::value> ret;
for (auto& result_row : result_set->rows()) {
rjson::value item = rjson::empty_object();
uint64_t item_length_in_bytes = 0;
describe_single_item(*selection, result_row, *attrs_to_get, item, &item_length_in_bytes);
item_callback(item_length_in_bytes);
ret.push_back(std::move(item));
co_await coroutine::maybe_yield();
}
co_return ret;
}
static bool check_needs_read_before_write(const parsed::value& v) {
return std::visit(overloaded_functor {
[&] (const parsed::constant& c) -> bool {
return false;
},
[&] (const parsed::value::function_call& f) -> bool {
return std::ranges::any_of(f._parameters, [&] (const parsed::value& param) {
return check_needs_read_before_write(param);
});
},
[&] (const parsed::path& p) -> bool {
return true;
}
}, v._value);
}
static bool check_needs_read_before_write(const attribute_path_map<parsed::update_expression::action>& update_expression) {
return std::ranges::any_of(update_expression, [](const auto& p) {
if (!p.second.has_value()) {
// If the action is not on the top-level attribute, we need to
// read the old item: we change only a part of the top-level
// attribute, and write the full top-level attribute back.
return true;
}
// Otherwise, the action p.second.get_value() is just on top-level
// attribute. Check if it needs read-before-write:
return std::visit(overloaded_functor {
[&] (const parsed::update_expression::action::set& a) -> bool {
return check_needs_read_before_write(a._rhs._v1) || (a._rhs._op != 'v' && check_needs_read_before_write(a._rhs._v2));
},
[&] (const parsed::update_expression::action::remove& a) -> bool {
return false;
},
[&] (const parsed::update_expression::action::add& a) -> bool {
return true;
},
[&] (const parsed::update_expression::action::del& a) -> bool {
return true;
}
}, p.second.get_value()._action);
});
}
/*!
* \brief estimate_value_size provides a rough size estimation
* for an rjson value object.
*
* When calculating RCU and WCU, we need to determine the length of the JSON representation
* (specifically, the length of each key and each value).
*
* When possible, this is calculated as a side effect of other operations.
* estimate_value_size is used when this calculation cannot be performed directly,
* but we still need an estimated value.
*
* It achieves this without streaming any values and uses a fixed size for numbers.
* The aim is not to provide a perfect 1-to-1 size calculation, as WCU is calculated
* in 1KB units. A ballpark estimate is sufficient.
*/
static size_t estimate_value_size(const rjson::value& value) {
size_t size = 0;
if (value.IsString()) {
size += value.GetStringLength();
}
else if (value.IsNumber()) {
size += 8;
}
else if (value.IsBool()) {
size += 5;
}
else if (value.IsArray()) {
for (auto& v : value.GetArray()) {
size += estimate_value_size(v); // Recursively calculate array element sizes
}
}
else if (value.IsObject()) {
for (auto it = value.MemberBegin(); it != value.MemberEnd(); ++it) {
size += it->name.GetStringLength(); // Size of the key
size += estimate_value_size(it->value); // Size of the value
}
}
return size;
}
class update_item_operation : public rmw_operation {
public:
// Some information parsed during the constructor to check for input
// errors, and cached to be used again during apply().
rjson::value* _attribute_updates;
// Instead of keeping a parsed::update_expression with an unsorted list
// list of actions, we keep them in an attribute_path_map which groups
// them by top-level attribute, and detects forbidden overlaps/conflicts.
attribute_path_map<parsed::update_expression::action> _update_expression;
// Saved list of GSI keys in the table being updated, used for
// validate_value_if_index_key()
std::unordered_map<bytes, std::string> _key_attributes;
parsed::condition_expression _condition_expression;
update_item_operation(parsed::expression_cache& parsed_expression_cache, service::storage_proxy& proxy, rjson::value&& request);
virtual ~update_item_operation() = default;
virtual std::optional<mutation> apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts, cdc::per_request_options& cdc_opts) const override;
bool needs_read_before_write() const;
private:
void delete_attribute(bytes&& column_name, const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts, deletable_row& row,
attribute_collector& modified_attrs) const;
void update_attribute(bytes&& column_name, const rjson::value& json_value, const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts,
deletable_row& row, attribute_collector& modified_attrs, const attribute_path_map_node<parsed::update_expression::action>* h = nullptr) const;
void apply_attribute_updates(const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts, deletable_row& row,
attribute_collector& modified_attrs, bool& any_updates, bool& any_deletes) const;
void apply_update_expression(const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts, deletable_row& row,
attribute_collector& modified_attrs, bool& any_updates, bool& any_deletes) const;
};
update_item_operation::update_item_operation(parsed::expression_cache& parsed_expression_cache, service::storage_proxy& proxy, rjson::value&& update_info)
: rmw_operation(proxy, std::move(update_info))
{
const rjson::value* key = rjson::find(_request, "Key");
if (!key) {
throw api_error::validation("UpdateItem requires a Key parameter");
}
_pk = pk_from_json(*key, _schema);
_ck = ck_from_json(*key, _schema);
check_key(*key, _schema);
const rjson::value* expression_attribute_names = rjson::find(_request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(_request, "ExpressionAttributeValues");
std::unordered_set<std::string> used_attribute_names;
std::unordered_set<std::string> used_attribute_values;
const rjson::value* update_expression = rjson::find(_request, "UpdateExpression");
if (update_expression) {
if (!update_expression->IsString()) {
throw api_error::validation("UpdateExpression must be a string");
}
try {
parsed::update_expression expr = parsed_expression_cache.parse_update_expression(rjson::to_string_view(*update_expression));
resolve_update_expression(expr,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
for (auto& action : expr.actions()) {
// Unfortunately we need to copy the action's path, because
// we std::move the action object.
auto p = action._path;
attribute_path_map_add("UpdateExpression", _update_expression, p, std::move(action));
}
} catch(expressions_syntax_error& e) {
throw api_error::validation(e.what());
}
}
_attribute_updates = rjson::find(_request, "AttributeUpdates");
if (_attribute_updates) {
if (!_attribute_updates->IsObject()) {
throw api_error::validation("AttributeUpdates must be an object");
}
for (auto it = std::as_const(*_attribute_updates).MemberBegin(); it != std::as_const(*_attribute_updates).MemberEnd(); ++it) {
validate_attr_name_length("AttributeUpdates", it->name.GetStringLength(), false);
}
}
_condition_expression = get_parsed_condition_expression(parsed_expression_cache, _request);
resolve_condition_expression(_condition_expression,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
verify_all_are_used(expression_attribute_names, used_attribute_names, "ExpressionAttributeNames", "UpdateItem");
verify_all_are_used(expression_attribute_values, used_attribute_values, "ExpressionAttributeValues", "UpdateItem");
// DynamoDB forbids having both old-style AttributeUpdates or Expected
// and new-style UpdateExpression or ConditionExpression in the same request
const rjson::value* expected = rjson::find(_request, "Expected");
if (update_expression && _attribute_updates) {
throw api_error::validation(
format("UpdateItem does not allow both AttributeUpdates and UpdateExpression to be given together"));
}
if (update_expression && expected) {
throw api_error::validation(
format("UpdateItem does not allow both old-style Expected and new-style UpdateExpression to be given together"));
}
if (_attribute_updates && !_condition_expression.empty()) {
throw api_error::validation(
format("UpdateItem does not allow both old-style AttributeUpdates and new-style ConditionExpression to be given together"));
}
if (_pk.representation().size() > 2) {
// ScyllaDB uses two extra bytes compared to DynamoDB for the key bytes length
_consumed_capacity._total_bytes += _pk.representation().size() - 2;
}
if (_ck.representation().size() > 2) {
// ScyllaDB uses two extra bytes compared to DynamoDB for the key bytes length
_consumed_capacity._total_bytes += _ck.representation().size() - 2;
}
if (expression_attribute_names) {
_consumed_capacity._total_bytes += estimate_value_size(*expression_attribute_names);
}
if (expression_attribute_values) {
_consumed_capacity._total_bytes += estimate_value_size(*expression_attribute_values);
}
_key_attributes = si_key_attributes(proxy.data_dictionary().find_table(
_schema->ks_name(), _schema->cf_name()));
}
// These are the cases where update_item_operation::apply() needs to use
// "previous_item" for certain AttributeUpdates operations (ADD or DELETE)
static bool check_needs_read_before_write_attribute_updates(rjson::value *attribute_updates) {
if (!attribute_updates) {
return false;
}
// We already confirmed in update_item_operation::update_item_operation()
// that _attribute_updates, when it exists, is a map
for (auto it = attribute_updates->MemberBegin(); it != attribute_updates->MemberEnd(); ++it) {
rjson::value* action = rjson::find(it->value, "Action");
if (action) {
std::string_view action_s = rjson::to_string_view(*action);
if (action_s == "ADD") {
return true;
}
// For DELETE operation, it only needs a read before write if the
// "Value" option is used. Without it, it's just a delete.
if (action_s == "DELETE" && it->value.HasMember("Value")) {
return true;
}
}
}
return false;
}
bool
update_item_operation::needs_read_before_write() const {
return check_needs_read_before_write(_update_expression) ||
check_needs_read_before_write(_condition_expression) ||
check_needs_read_before_write_attribute_updates(_attribute_updates) ||
_request.HasMember("Expected") ||
(_returnvalues != returnvalues::NONE && _returnvalues != returnvalues::UPDATED_NEW);
}
// action_result() returns the result of applying an UpdateItem action -
// this result is either a JSON object or an unset optional which indicates
// the action was a deletion. The caller (update_item_operation::apply()
// below) will either write this JSON as the content of a column, or
// use it as a piece in a bigger top-level attribute.
static std::optional<rjson::value> action_result(
const parsed::update_expression::action& action,
const rjson::value* previous_item) {
return std::visit(overloaded_functor {
[&] (const parsed::update_expression::action::set& a) -> std::optional<rjson::value> {
return calculate_value(a._rhs, previous_item);
},
[&] (const parsed::update_expression::action::remove& a) -> std::optional<rjson::value> {
return std::nullopt;
},
[&] (const parsed::update_expression::action::add& a) -> std::optional<rjson::value> {
parsed::value base;
parsed::value addition;
base.set_path(action._path);
addition.set_constant(a._valref);
rjson::value v1 = calculate_value(base, calculate_value_caller::UpdateExpression, previous_item);
rjson::value v2 = calculate_value(addition, calculate_value_caller::UpdateExpression, previous_item);
rjson::value result;
// An ADD can be used to create a new attribute (when
// v1.IsNull()) or to add to a pre-existing attribute:
if (v1.IsNull()) {
const auto v2_type = get_item_type_string(v2);
if (v2_type == "N" || v2_type == "SS" || v2_type == "NS" || v2_type == "BS") {
result = v2;
} else {
throw api_error::validation(format("An operand in the update expression has an incorrect data type: {}", v2));
}
} else {
const auto v1_type = get_item_type_string(v1);
if (v1_type == "N") {
if (get_item_type_string(v2) != "N") {
throw api_error::validation(fmt::format("Incorrect operand type for operator or function. Expected {}: {}", v1_type, rjson::print(v2)));
}
result = number_add(v1, v2);
} else if (v1_type == "SS" || v1_type == "NS" || v1_type == "BS") {
if (get_item_type_string(v2) != v1_type) {
throw api_error::validation(fmt::format("Incorrect operand type for operator or function. Expected {}: {}", v1_type, rjson::print(v2)));
}
result = set_sum(v1, v2);
} else {
throw api_error::validation(format("An operand in the update expression has an incorrect data type: {}", v1));
}
}
return result;
},
[&] (const parsed::update_expression::action::del& a) -> std::optional<rjson::value> {
parsed::value base;
parsed::value subset;
base.set_path(action._path);
subset.set_constant(a._valref);
rjson::value v1 = calculate_value(base, calculate_value_caller::UpdateExpression, previous_item);
rjson::value v2 = calculate_value(subset, calculate_value_caller::UpdateExpression, previous_item);
if (!v1.IsNull()) {
return set_diff(v1, v2);
}
// When we return nullopt here, we ask to *delete* this attribute,
// which is unnecessary because we know the attribute does not
// exist anyway. This is a waste, but a small one. Note that also
// for the "remove" action above we don't bother to check if the
// previous_item add anything to remove.
return std::nullopt;
}
}, action._action);
}
}
// Print an attribute_path_map_node<action> as the list of paths it contains:
template <> struct fmt::formatter<alternator::attribute_path_map_node<alternator::parsed::update_expression::action>> {
constexpr auto parse(format_parse_context& ctx) { return ctx.begin(); }
// this function recursively call into itself, so we have to forward declare it.
auto format(const alternator::attribute_path_map_node<alternator::parsed::update_expression::action>& h, fmt::format_context& ctx) const
-> decltype(ctx.out());
};
auto fmt::formatter<alternator::attribute_path_map_node<alternator::parsed::update_expression::action>>::format(const alternator::attribute_path_map_node<alternator::parsed::update_expression::action>& h, fmt::format_context& ctx) const
-> decltype(ctx.out()) {
auto out = ctx.out();
if (h.has_value()) {
out = fmt::format_to(out, " {}", h.get_value()._path);
} else if (h.has_members()) {
for (auto& member : h.get_members()) {
out = fmt::format_to(out, "{}", *member.second);
}
} else if (h.has_indexes()) {
for (auto& index : h.get_indexes()) {
out = fmt::format_to(out, "{}", *index.second);
}
}
return out;
}
namespace alternator {
// Apply the hierarchy of actions in an attribute_path_map_node<action> to a
// JSON object which uses DynamoDB's serialization conventions. The complete,
// unmodified, previous_item is also necessary for the right-hand sides of the
// actions. Modifies obj in-place or returns false if it is to be removed.
static bool hierarchy_actions(
rjson::value& obj,
const attribute_path_map_node<parsed::update_expression::action>& h,
const rjson::value* previous_item)
{
if (!obj.IsObject() || obj.MemberCount() != 1) {
// This shouldn't happen. We shouldn't have stored malformed objects.
// But today Alternator does not validate the structure of nested
// documents before storing them, so this can happen on read.
throw api_error::validation(format("Malformed value object read: {}", obj));
}
const char* type = obj.MemberBegin()->name.GetString();
rjson::value& v = obj.MemberBegin()->value;
if (h.has_value()) {
// Action replacing everything in this position in the hierarchy
std::optional<rjson::value> newv = action_result(h.get_value(), previous_item);
if (newv) {
obj = std::move(*newv);
} else {
return false;
}
} else if (h.has_members()) {
if (type[0] != 'M' || !v.IsObject()) {
// A .something on a non-map doesn't work.
throw api_error::validation(fmt::format("UpdateExpression: document paths not valid for this item:{}", h));
}
for (const auto& member : h.get_members()) {
std::string attr = member.first;
const attribute_path_map_node<parsed::update_expression::action>& subh = *member.second;
rjson::value *subobj = rjson::find(v, attr);
if (subobj) {
if (!hierarchy_actions(*subobj, subh, previous_item)) {
rjson::remove_member(v, attr);
}
} else {
// When a.b does not exist, setting a.b itself (i.e.
// subh.has_value()) is fine, but setting a.b.c is not.
if (subh.has_value()) {
std::optional<rjson::value> newv = action_result(subh.get_value(), previous_item);
if (newv) {
// This is the !subobj case, so v doesn't have an
// attr member so we can use add()
rjson::add_with_string_name(v, attr, std::move(*newv));
} else {
// Removing a.b when a is a map but a.b doesn't exist
// is silently ignored. It's not considered an error.
}
} else {
throw api_error::validation(format("UpdateExpression: document paths not valid for this item:{}", h));
}
}
}
} else if (h.has_indexes()) {
if (type[0] != 'L' || !v.IsArray()) {
// A [i] on a non-list doesn't work.
throw api_error::validation(format("UpdateExpression: document paths not valid for this item:{}", h));
}
unsigned nremoved = 0;
for (const auto& index : h.get_indexes()) {
unsigned i = index.first - nremoved;
const attribute_path_map_node<parsed::update_expression::action>& subh = *index.second;
if (i < v.Size()) {
if (!hierarchy_actions(v[i], subh, previous_item)) {
v.Erase(v.Begin() + i);
// If we have the actions "REMOVE a[1] SET a[3] = :val",
// the index 3 refers to the original indexes, before any
// items were removed. So we offset the next indexes
// (which are guaranteed to be higher than i - indexes is
// a sorted map) by an increased "nremoved".
nremoved++;
}
} else {
// If a[7] does not exist, setting a[7] itself (i.e.
// subh.has_value()) is fine - and appends an item, though
// not necessarily with index 7. But setting a[7].b will
// not work.
if (subh.has_value()) {
std::optional<rjson::value> newv = action_result(subh.get_value(), previous_item);
if (newv) {
rjson::push_back(v, std::move(*newv));
} else {
// Removing a[7] when the list has fewer elements is
// silently ignored. It's not considered an error.
}
} else {
throw api_error::validation(format("UpdateExpression: document paths not valid for this item:{}", h));
}
}
}
}
return true;
}
void update_item_operation::delete_attribute(bytes&& column_name, const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts,
deletable_row& row, attribute_collector& modified_attrs) const {
if (_returnvalues == returnvalues::ALL_NEW) {
rjson::remove_member(_return_attributes, to_string_view(column_name));
} else if (_returnvalues == returnvalues::UPDATED_OLD && previous_item) {
std::string_view cn = to_string_view(column_name);
const rjson::value* col = rjson::find(*previous_item, cn);
if (col) {
// In the UPDATED_OLD case the item starts empty and column
// names are unique, so we can use add()
rjson::add_with_string_name(_return_attributes, cn, rjson::copy(*col));
}
}
const column_definition* cdef = find_attribute(*_schema, column_name);
if (cdef) {
row.cells().apply(*cdef, atomic_cell::make_dead(ts, gc_clock::now()));
} else {
modified_attrs.del(std::move(column_name), ts);
}
}
void update_item_operation::update_attribute(bytes&& column_name, const rjson::value& json_value, const std::unique_ptr<rjson::value>& previous_item,
const api::timestamp_type ts, deletable_row& row, attribute_collector& modified_attrs,
const attribute_path_map_node<parsed::update_expression::action>* h) const {
if (_returnvalues == returnvalues::ALL_NEW) {
rjson::replace_with_string_name(_return_attributes, to_string_view(column_name), rjson::copy(json_value));
} else if (_returnvalues == returnvalues::UPDATED_NEW) {
rjson::value&& v = rjson::copy(json_value);
if (h) {
// If the operation was only on specific attribute paths,
// leave only them in _return_attributes.
if (hierarchy_filter(v, *h)) {
// In the UPDATED_NEW case, _return_attributes starts
// empty and the attribute names are unique, so we can
// use add().
rjson::add_with_string_name(_return_attributes, to_string_view(column_name), std::move(v));
}
} else {
rjson::add_with_string_name(_return_attributes, to_string_view(column_name), std::move(v));
}
} else if (_returnvalues == returnvalues::UPDATED_OLD && previous_item) {
std::string_view cn = to_string_view(column_name);
const rjson::value* col = rjson::find(*previous_item, cn);
if (col) {
rjson::value&& v = rjson::copy(*col);
if (h) {
if (hierarchy_filter(v, *h)) {
// In the UPDATED_OLD case, _return_attributes starts
// empty and the attribute names are unique, so we can
// use add().
rjson::add_with_string_name(_return_attributes, cn, std::move(v));
}
} else {
rjson::add_with_string_name(_return_attributes, cn, std::move(v));
}
}
}
const column_definition* cdef = find_attribute(*_schema, column_name);
if (cdef) {
bytes column_value = get_key_from_typed_value(json_value, *cdef);
row.cells().apply(*cdef, atomic_cell::make_live(*cdef->type, ts, column_value));
} else {
// This attribute may be a key column of one of the GSIs or LSIs,
// in which case there are some limitations on the value.
validate_value_if_index_key(_key_attributes, column_name, json_value);
modified_attrs.put(std::move(column_name), serialize_item(json_value), ts);
}
}
inline void update_item_operation::apply_attribute_updates(const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts, deletable_row& row,
attribute_collector& modified_attrs, bool& any_updates, bool& any_deletes) const {
for (auto it = _attribute_updates->MemberBegin(); it != _attribute_updates->MemberEnd(); ++it) {
// Note that it.key() is the name of the column, *it is the operation
bytes column_name = to_bytes(rjson::to_string_view(it->name));
const column_definition* cdef = _schema->get_column_definition(column_name);
if (cdef && cdef->is_primary_key()) {
throw api_error::validation(format("UpdateItem cannot update key column {}", rjson::to_string_view(it->name)));
}
std::string action = rjson::to_string((it->value)["Action"]);
if (action == "DELETE") {
// The DELETE operation can do two unrelated tasks. Without a
// "Value" option, it is used to delete an attribute. With a
// "Value" option, it is used to delete a set of elements from
// a set attribute of the same type.
if (it->value.HasMember("Value")) {
// Subtracting sets needs a read of previous_item, so
// check_needs_read_before_write_attribute_updates()
// returns true in this case, and previous_item is
// available to us when the item exists.
const rjson::value* v1 = previous_item ? rjson::find(*previous_item, to_string_view(column_name)) : nullptr;
const rjson::value& v2 = (it->value)["Value"];
validate_value(v2, "AttributeUpdates");
const auto v2_type = get_item_type_string(v2);
if (v2_type != "SS" && v2_type != "NS" && v2_type != "BS") {
throw api_error::validation(fmt::format("AttributeUpdates DELETE operation with Value only valid for sets, got type {}", v2_type));
}
if (v1) {
std::optional<rjson::value> result = set_diff(*v1, v2);
if (result) {
any_updates = true;
update_attribute(std::move(column_name), *result, previous_item, ts, row, modified_attrs);
} else {
// DynamoDB does not allow empty sets - if the
// result is empty, delete the attribute.
any_deletes = true;
delete_attribute(std::move(column_name), previous_item, ts, row, modified_attrs);
}
} else {
// if the attribute or item don't exist, the DELETE
// operation should silently do nothing - and not
// create an empty item. It's a waste to call
// do_delete() on an attribute we already know is
// deleted, so we can just mark any_deletes = true.
any_deletes = true;
}
} else {
any_deletes = true;
delete_attribute(std::move(column_name), previous_item, ts, row, modified_attrs);
}
} else if (action == "PUT") {
const rjson::value& value = (it->value)["Value"];
validate_value(value, "AttributeUpdates");
any_updates = true;
update_attribute(std::move(column_name), value, previous_item, ts, row, modified_attrs);
} else if (action == "ADD") {
// Note that check_needs_read_before_write_attribute_updates()
// made sure we retrieved previous_item (if exists) when there
// is an ADD action.
const rjson::value* v1 = previous_item ? rjson::find(*previous_item, to_string_view(column_name)) : nullptr;
const rjson::value& v2 = (it->value)["Value"];
validate_value(v2, "AttributeUpdates");
// An ADD can be used to create a new attribute (when
// !v1) or to add to a pre-existing attribute:
if (!v1) {
const auto v2_type = get_item_type_string(v2);
if (v2_type == "N" || v2_type == "SS" || v2_type == "NS" || v2_type == "BS" || v2_type == "L") {
any_updates = true;
update_attribute(std::move(column_name), v2, previous_item, ts, row, modified_attrs);
} else {
throw api_error::validation(format("An operand in the AttributeUpdates ADD has an incorrect data type: {}", v2));
}
} else {
const auto v1_type = get_item_type_string(*v1);
const auto v2_type = get_item_type_string(v2);
if (v2_type != v1_type) {
throw api_error::validation(fmt::format("Operand type mismatch in AttributeUpdates ADD. Expected {}, got {}", v1_type, v2_type));
}
if (v1_type == "N") {
any_updates = true;
update_attribute(std::move(column_name), number_add(*v1, v2), previous_item, ts, row, modified_attrs);
} else if (v1_type == "SS" || v1_type == "NS" || v1_type == "BS") {
any_updates = true;
update_attribute(std::move(column_name), set_sum(*v1, v2), previous_item, ts, row, modified_attrs);
} else if (v1_type == "L") {
// The DynamoDB documentation doesn't say it supports
// lists in ADD operations, but it turns out that it
// does. Interestingly, this is only true for
// AttributeUpdates (this code) - the similar ADD
// in UpdateExpression doesn't support lists.
any_updates = true;
update_attribute(std::move(column_name), list_concatenate(*v1, v2), previous_item, ts, row, modified_attrs);
} else {
throw api_error::validation(format("An operand in the AttributeUpdates ADD has an incorrect data type: {}", *v1));
}
}
} else {
throw api_error::validation(fmt::format("Unknown Action value '{}' in AttributeUpdates", action));
}
}
}
inline void update_item_operation::apply_update_expression(const std::unique_ptr<rjson::value>& previous_item, const api::timestamp_type ts, deletable_row& row,
attribute_collector& modified_attrs, bool& any_updates, bool& any_deletes) const {
for (auto& actions : _update_expression) {
// The actions of _update_expression are grouped by top-level
// attributes. Here, all actions in actions.second share the same
// top-level attribute actions.first.
std::string column_name = actions.first;
const column_definition* cdef = _schema->get_column_definition(to_bytes(column_name));
if (cdef && cdef->is_primary_key()) {
throw api_error::validation(fmt::format("UpdateItem cannot update key column {}", column_name));
}
if (actions.second.has_value()) {
// An action on a top-level attribute column_name. The single
// action is actions.second.get_value(). We can simply invoke
// the action and replace the attribute with its result:
std::optional<rjson::value> result = action_result(actions.second.get_value(), previous_item.get());
if (result) {
any_updates = true;
update_attribute(to_bytes(column_name), *result, previous_item, ts, row, modified_attrs, &actions.second);
} else {
any_deletes = true;
delete_attribute(to_bytes(column_name), previous_item, ts, row, modified_attrs);
}
} else {
// We have actions on a path or more than one path in the same
// top-level attribute column_name - but not on the top-level
// attribute as a whole. We already read the full top-level
// attribute (see check_needs_read_before_write()), and now we
// need to modify pieces of it and write back the entire
// top-level attribute.
if (!previous_item) {
throw api_error::validation(format("UpdateItem cannot update nested document path on non-existent item"));
}
const rjson::value* toplevel = rjson::find(*previous_item, column_name);
if (!toplevel) {
throw api_error::validation(fmt::format("UpdateItem cannot update document path: missing attribute {}", column_name));
}
rjson::value result = rjson::copy(*toplevel);
any_updates = true;
hierarchy_actions(result, actions.second, previous_item.get());
update_attribute(to_bytes(column_name), std::move(result), previous_item, ts, row, modified_attrs, &actions.second);
}
}
}
std::optional<mutation> update_item_operation::apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts, cdc::per_request_options& cdc_opts) const {
if (_consumed_capacity._total_bytes == 0) {
_consumed_capacity._total_bytes = 1;
}
if (!verify_expected(_request, previous_item.get()) || !verify_condition_expression(_condition_expression, previous_item.get())) {
if (previous_item && _returnvalues_on_condition_check_failure == returnvalues_on_condition_check_failure::ALL_OLD) {
_return_attributes = std::move(*previous_item);
}
// If the update is to be cancelled because of an unfulfilled
// condition, return an empty optional mutation, which is more
// efficient than throwing an exception.
return {};
}
// In the ReturnValues=ALL_NEW case, we make a copy of previous_item into
// _return_attributes and parts of it will be overwritten by the new
// updates (in do_update() and do_delete()). We need to make a copy and
// cannot overwrite previous_item directly because we still need its
// original content for update expressions. For example, the expression
// "REMOVE a SET b=a" is valid, and needs the original value of a to
// stick around.
// Note that for ReturnValues=ALL_OLD, we don't need to copy here, and
// can just move previous_item later, when we don't need it any more.
if (_returnvalues == returnvalues::ALL_NEW) {
if (previous_item) {
_return_attributes = rjson::copy(*previous_item);
} else {
// If there is no previous item, usually a new item is created
// and contains the given key. This may be cancelled at the end
// of this function if the update is just deletes.
_return_attributes = rjson::copy(rjson::get(_request, "Key"));
}
} else if (_returnvalues == returnvalues::UPDATED_OLD || _returnvalues == returnvalues::UPDATED_NEW) {
_return_attributes = rjson::empty_object();
}
mutation m(_schema, _pk);
bool any_updates = false;
bool any_deletes = false;
auto& row = m.partition().clustered_row(*_schema, _ck);
auto modified_attrs = attribute_collector();
if (!_update_expression.empty()) {
apply_update_expression(previous_item, ts, row, modified_attrs, any_updates, any_deletes);
}
if (_attribute_updates) {
apply_attribute_updates(previous_item, ts, row, modified_attrs, any_updates, any_deletes);
}
if (!modified_attrs.empty()) {
auto serialized_map = modified_attrs.to_mut().serialize(*attrs_type());
row.cells().apply(attrs_column(*_schema), std::move(serialized_map));
}
// To allow creation of an item with no attributes, we need a row marker.
// Note that unlike Scylla, even an "update" operation needs to add a row
// marker. An update with only DELETE operations must not add a row marker
// (this was issue #5862) but any other update, even an empty one, should.
if (any_updates || !any_deletes) {
row.apply(row_marker(ts));
} else if (_returnvalues == returnvalues::ALL_NEW && !previous_item) {
// There was no pre-existing item, and we're not creating one, so
// don't report the new item in the returned Attributes.
_return_attributes = rjson::null_value();
}
if (_returnvalues == returnvalues::ALL_OLD && previous_item) {
_return_attributes = std::move(*previous_item);
}
// ReturnValues=UPDATED_OLD/NEW never return an empty Attributes field,
// even if a new item was created. Instead it should be missing entirely.
if (_returnvalues == returnvalues::UPDATED_OLD || _returnvalues == returnvalues::UPDATED_NEW) {
if (_return_attributes.MemberCount() == 0) {
_return_attributes = rjson::null_value();
}
}
return m;
}
future<executor::request_return_type> executor::update_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.update_item++;
auto start_time = std::chrono::steady_clock::now();
elogger.trace("update_item {}", request);
auto op = make_shared<update_item_operation>(*_parsed_expression_cache, _proxy, std::move(request));
tracing::add_alternator_table_name(trace_state, op->schema()->cf_name());
const bool needs_read_before_write = _proxy.data_dictionary().get_config().alternator_force_read_before_write() || op->needs_read_before_write();
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, op->schema(), auth::permission::MODIFY, _stats);
auto cas_shard = op->shard_for_execute(needs_read_before_write);
if (cas_shard && !cas_shard->this_shard()) {
_stats.api_operations.update_item--; // uncount on this shard, will be counted in other shard
_stats.shard_bounce_for_lwt++;
co_return co_await container().invoke_on(cas_shard->shard(), _ssg,
[request = std::move(*op).move_request(), cs = client_state.move_to_other_shard(), gt = tracing::global_trace_state_ptr(trace_state), permit = std::move(permit)]
(executor& e) mutable {
return do_with(cs.get(), [&e, request = std::move(request), trace_state = tracing::trace_state_ptr(gt)]
(service::client_state& client_state) mutable {
//FIXME: Instead of passing empty_service_permit() to the background operation,
// the current permit's lifetime should be prolonged, so that it's destructed
// only after all background operations are finished as well.
return e.update_item(client_state, std::move(trace_state), empty_service_permit(), std::move(request));
});
});
}
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *(op->schema()));
per_table_stats->api_operations.update_item++;
uint64_t wcu_total = 0;
auto res = co_await op->execute(_proxy, std::move(cas_shard), client_state, trace_state, std::move(permit), needs_read_before_write, _stats, *per_table_stats, wcu_total);
per_table_stats->operation_sizes.update_item_op_size_kb.add(bytes_to_kb_ceil(op->consumed_capacity()._total_bytes));
per_table_stats->wcu_total[stats::wcu_types::UPDATE_ITEM] += wcu_total;
_stats.wcu_total[stats::wcu_types::UPDATE_ITEM] += wcu_total;
per_table_stats->api_operations.update_item_latency.mark(std::chrono::steady_clock::now() - start_time);
_stats.api_operations.update_item_latency.mark(std::chrono::steady_clock::now() - start_time);
co_return res;
}
// Check according to the request's "ConsistentRead" field, which consistency
// level we need to use for the read. The field can be True for strongly
// consistent reads, or False for eventually consistent reads, or if this
// field is absence, we default to eventually consistent reads.
// In Scylla, eventually-consistent reads are implemented as consistency
// level LOCAL_ONE, and strongly-consistent reads as LOCAL_QUORUM.
static db::consistency_level get_read_consistency(const rjson::value& request) {
const rjson::value* consistent_read_value = rjson::find(request, "ConsistentRead");
bool consistent_read = false;
if (consistent_read_value && !consistent_read_value->IsNull()) {
if (consistent_read_value->IsBool()) {
consistent_read = consistent_read_value->GetBool();
} else {
throw api_error::validation("ConsistentRead flag must be a boolean");
}
}
return consistent_read ? db::consistency_level::LOCAL_QUORUM : db::consistency_level::LOCAL_ONE;
}
// describe_item() wraps the result of describe_single_item() by a map
// as needed by the GetItem request. It should not be used for other purposes,
// use describe_single_item() instead.
static rjson::value describe_item(schema_ptr schema,
const query::partition_slice& slice,
const cql3::selection::selection& selection,
const query::result& query_result,
const std::optional<attrs_to_get>& attrs_to_get,
consumed_capacity_counter& consumed_capacity,
uint64_t& metric) {
std::optional<rjson::value> opt_item = executor::describe_single_item(std::move(schema), slice, selection, std::move(query_result), attrs_to_get, &consumed_capacity._total_bytes);
rjson::value item_descr = rjson::empty_object();
if (opt_item) {
rjson::add(item_descr, "Item", std::move(*opt_item));
}
consumed_capacity.add_consumed_capacity_to_response_if_needed(item_descr);
metric += consumed_capacity.get_half_units();
return item_descr;
}
future<executor::request_return_type> executor::get_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.get_item++;
auto start_time = std::chrono::steady_clock::now();
elogger.trace("Getting item {}", request);
schema_ptr schema = get_table(_proxy, request);
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *schema);
per_table_stats->api_operations.get_item++;
tracing::add_alternator_table_name(trace_state, schema->cf_name());
rjson::value& query_key = request["Key"];
db::consistency_level cl = get_read_consistency(request);
partition_key pk = pk_from_json(query_key, schema);
dht::partition_range_vector partition_ranges{dht::partition_range(dht::decorate_key(*schema, pk))};
std::vector<query::clustering_range> bounds;
if (schema->clustering_key_size() == 0) {
bounds.push_back(query::clustering_range::make_open_ended_both_sides());
} else {
clustering_key ck = ck_from_json(query_key, schema);
bounds.push_back(query::clustering_range::make_singular(std::move(ck)));
}
check_key(query_key, schema);
//TODO(sarna): It would be better to fetch only some attributes of the map, not all
auto regular_columns =
schema->regular_columns() | std::views::transform(&column_definition::id)
| std::ranges::to<query::column_id_vector>();
auto selection = cql3::selection::selection::wildcard(schema);
auto partition_slice = query::partition_slice(std::move(bounds), {}, std::move(regular_columns), selection->get_query_options());
auto command = ::make_lw_shared<query::read_command>(schema->id(), schema->version(), partition_slice, _proxy.get_max_result_size(partition_slice),
query::tombstone_limit(_proxy.get_tombstone_limit()));
std::unordered_set<std::string> used_attribute_names;
auto attrs_to_get = calculate_attrs_to_get(request, *_parsed_expression_cache, used_attribute_names);
const rjson::value* expression_attribute_names = rjson::find(request, "ExpressionAttributeNames");
verify_all_are_used(expression_attribute_names, used_attribute_names, "ExpressionAttributeNames", "GetItem");
rcu_consumed_capacity_counter add_capacity(request, cl == db::consistency_level::LOCAL_QUORUM);
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::SELECT, _stats);
service::storage_proxy::coordinator_query_result qr =
co_await _proxy.query(
schema, std::move(command), std::move(partition_ranges), cl,
service::storage_proxy::coordinator_query_options(executor::default_timeout(), std::move(permit), client_state, trace_state));
per_table_stats->api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
_stats.api_operations.get_item_latency.mark(std::chrono::steady_clock::now() - start_time);
uint64_t rcu_half_units = 0;
rjson::value res = describe_item(schema, partition_slice, *selection, *qr.query_result, std::move(attrs_to_get), add_capacity, rcu_half_units);
per_table_stats->rcu_half_units_total += rcu_half_units;
_stats.rcu_half_units_total += rcu_half_units;
// Update item size metrics only if we found an item.
if (qr.query_result->row_count().value_or(0) > 0) {
per_table_stats->operation_sizes.get_item_op_size_kb.add(bytes_to_kb_ceil(add_capacity._total_bytes));
}
co_return rjson::print(std::move(res));
}
static void check_big_object(const rjson::value& val, int& size_left);
static void check_big_array(const rjson::value& val, int& size_left);
bool is_big(const rjson::value& val, int big_size) {
if (val.IsString()) {
return ssize_t(val.GetStringLength()) > big_size;
} else if (val.IsObject()) {
check_big_object(val, big_size);
return big_size < 0;
} else if (val.IsArray()) {
check_big_array(val, big_size);
return big_size < 0;
}
return false;
}
static void check_big_array(const rjson::value& val, int& size_left) {
// Assume a fixed size of 10 bytes for each number, boolean, etc., or
// beginning of a sub-object. This doesn't have to be accurate.
size_left -= 10 * val.Size();
for (const auto& v : val.GetArray()) {
if (size_left < 0) {
return;
}
// Note that we avoid recursive calls for the leaves (anything except
// array or object) because usually those greatly outnumber the trunk.
if (v.IsString()) {
size_left -= v.GetStringLength();
} else if (v.IsObject()) {
check_big_object(v, size_left);
} else if (v.IsArray()) {
check_big_array(v, size_left);
}
}
}
static void check_big_object(const rjson::value& val, int& size_left) {
size_left -= 10 * val.MemberCount();
for (const auto& m : val.GetObject()) {
if (size_left < 0) {
return;
}
size_left -= m.name.GetStringLength();
if (m.value.IsString()) {
size_left -= m.value.GetStringLength();
} else if (m.value.IsObject()) {
check_big_object(m.value, size_left);
} else if (m.value.IsArray()) {
check_big_array(m.value, size_left);
}
}
}
future<executor::request_return_type> executor::batch_get_item(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
// FIXME: In this implementation, an unbounded batch size can cause
// unbounded response JSON object to be buffered in memory, unbounded
// parallelism of the requests, and unbounded amount of non-preemptable
// work in the following loops. So we should limit the batch size, and/or
// the response size, as DynamoDB does.
_stats.api_operations.batch_get_item++;
rjson::value& request_items = request["RequestItems"];
auto start_time = std::chrono::steady_clock::now();
// We need to validate all the parameters before starting any asynchronous
// query, and fail the entire request on any parse error. So we parse all
// the input into our own vector "requests", each element a table_requests
// listing all the request aimed at a single table. For efficiency, inside
// each table_requests we further group together all reads going to the
// same partition, so we can later send them together.
bool should_add_rcu = rcu_consumed_capacity_counter::should_add_capacity(request);
struct table_requests {
schema_ptr schema;
db::consistency_level cl;
::shared_ptr<const std::optional<alternator::attrs_to_get>> attrs_to_get;
// clustering_keys keeps a sorted set of clustering keys. It must
// be sorted for the read below (see #10827). Additionally each
// clustering key is mapped to the original rjson::value "Key".
using clustering_keys = std::map<clustering_key, rjson::value*, clustering_key::less_compare>;
std::unordered_map<partition_key, clustering_keys, partition_key::hashing, partition_key::equality> requests;
table_requests(schema_ptr s)
: schema(std::move(s))
, requests(8, partition_key::hashing(*schema), partition_key::equality(*schema))
{}
void add(rjson::value& key) {
auto pk = pk_from_json(key, schema);
auto it = requests.find(pk);
if (it == requests.end()) {
it = requests.emplace(pk, clustering_key::less_compare(*schema)).first;
}
auto ck = ck_from_json(key, schema);
if (auto [_, inserted] = it->second.emplace(ck, &key); !inserted) {
throw api_error::validation("Provided list of item keys contains duplicates");
}
}
};
std::vector<table_requests> requests;
uint batch_size = 0;
for (auto it = request_items.MemberBegin(); it != request_items.MemberEnd(); ++it) {
table_requests rs(get_table_from_batch_request(_proxy, it));
tracing::add_alternator_table_name(trace_state, rs.schema->cf_name());
rs.cl = get_read_consistency(it->value);
std::unordered_set<std::string> used_attribute_names;
rs.attrs_to_get = ::make_shared<const std::optional<attrs_to_get>>(calculate_attrs_to_get(it->value, *_parsed_expression_cache, used_attribute_names));
const rjson::value* expression_attribute_names = rjson::find(it->value, "ExpressionAttributeNames");
verify_all_are_used(expression_attribute_names, used_attribute_names,"ExpressionAttributeNames", "GetItem");
auto& keys = (it->value)["Keys"];
for (rjson::value& key : keys.GetArray()) {
rs.add(key);
check_key(key, rs.schema);
}
batch_size += rs.requests.size();
requests.emplace_back(std::move(rs));
}
for (const table_requests& tr : requests) {
co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, tr.schema, auth::permission::SELECT, _stats);
}
_stats.api_operations.batch_get_item_batch_total += batch_size;
_stats.api_operations.batch_get_item_histogram.add(batch_size);
// If we got here, all "requests" are valid, so let's start the
// requests for the different partitions all in parallel.
std::vector<future<std::vector<rjson::value>>> response_futures;
std::vector<uint64_t> consumed_rcu_half_units_per_table(requests.size());
for (size_t i = 0; i < requests.size(); i++) {
const table_requests& rs = requests[i];
bool is_quorum = rs.cl == db::consistency_level::LOCAL_QUORUM;
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *rs.schema);
per_table_stats->api_operations.batch_get_item_histogram.add(rs.requests.size());
for (const auto& [pk, cks] : rs.requests) {
dht::partition_range_vector partition_ranges{dht::partition_range(dht::decorate_key(*rs.schema, pk))};
std::vector<query::clustering_range> bounds;
if (rs.schema->clustering_key_size() == 0) {
bounds.push_back(query::clustering_range::make_open_ended_both_sides());
} else {
for (auto& ck : cks) {
bounds.push_back(query::clustering_range::make_singular(ck.first));
}
}
auto regular_columns =
rs.schema->regular_columns() | std::views::transform(&column_definition::id)
| std::ranges::to<query::column_id_vector>();
auto selection = cql3::selection::selection::wildcard(rs.schema);
auto partition_slice = query::partition_slice(std::move(bounds), {}, std::move(regular_columns), selection->get_query_options());
auto command = ::make_lw_shared<query::read_command>(rs.schema->id(), rs.schema->version(), partition_slice, _proxy.get_max_result_size(partition_slice),
query::tombstone_limit(_proxy.get_tombstone_limit()));
command->allow_limit = db::allow_per_partition_rate_limit::yes;
const auto item_callback = [is_quorum, per_table_stats, &rcus_per_table = consumed_rcu_half_units_per_table[i]](uint64_t size) {
rcus_per_table += rcu_consumed_capacity_counter::get_half_units(size, is_quorum);
// Update item size only if the item exists.
if (size > 0) {
per_table_stats->operation_sizes.batch_get_item_op_size_kb.add(bytes_to_kb_ceil(size));
}
};
future<std::vector<rjson::value>> f = _proxy.query(rs.schema, std::move(command), std::move(partition_ranges), rs.cl,
service::storage_proxy::coordinator_query_options(executor::default_timeout(), permit, client_state, trace_state)).then(
[schema = rs.schema, partition_slice = std::move(partition_slice), selection = std::move(selection), attrs_to_get = rs.attrs_to_get, item_callback = std::move(item_callback)] (service::storage_proxy::coordinator_query_result qr) mutable {
utils::get_local_injector().inject("alternator_batch_get_item", [] { throw std::runtime_error("batch_get_item injection"); });
return describe_multi_item(std::move(schema), std::move(partition_slice), std::move(selection), std::move(qr.query_result), std::move(attrs_to_get), std::move(item_callback));
});
response_futures.push_back(std::move(f));
}
}
// Wait for all requests to complete, and then return the response.
// In case of full failure (no reads succeeded), an arbitrary error
// from one of the operations will be returned.
bool some_succeeded = false;
std::exception_ptr eptr;
rjson::value response = rjson::empty_object();
rjson::add(response, "Responses", rjson::empty_object());
rjson::add(response, "UnprocessedKeys", rjson::empty_object());
auto fut_it = response_futures.begin();
rjson::value consumed_capacity = rjson::empty_array();
for (size_t i = 0; i < requests.size(); i++) {
const table_requests& rs = requests[i];
std::string table = table_name(*rs.schema);
for (const auto& [_, cks] : rs.requests) {
auto& fut = *fut_it;
++fut_it;
try {
std::vector<rjson::value> results = co_await std::move(fut);
some_succeeded = true;
if (!response["Responses"].HasMember(table)) {
rjson::add_with_string_name(response["Responses"], table, rjson::empty_array());
}
for (rjson::value& json : results) {
rjson::push_back(response["Responses"][table], std::move(json));
}
} catch(...) {
eptr = std::current_exception();
// This read of potentially several rows in one partition,
// failed. We need to add the row key(s) to UnprocessedKeys.
if (!response["UnprocessedKeys"].HasMember(table)) {
// Add the table's entry in UnprocessedKeys. Need to copy
// all the table's parameters from the request except the
// Keys field, which we start empty and then build below.
rjson::add_with_string_name(response["UnprocessedKeys"], table, rjson::empty_object());
rjson::value& unprocessed_item = response["UnprocessedKeys"][table];
rjson::value& request_item = request_items[table];
for (auto it = request_item.MemberBegin(); it != request_item.MemberEnd(); ++it) {
if (it->name != "Keys") {
rjson::add_with_string_name(unprocessed_item,
rjson::to_string_view(it->name), rjson::copy(it->value));
}
}
rjson::add_with_string_name(unprocessed_item, "Keys", rjson::empty_array());
}
for (auto& ck : cks) {
rjson::push_back(response["UnprocessedKeys"][table]["Keys"], std::move(*ck.second));
}
}
}
uint64_t rcu_half_units = consumed_rcu_half_units_per_table[i];
_stats.rcu_half_units_total += rcu_half_units;
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *rs.schema);
per_table_stats->rcu_half_units_total += rcu_half_units;
if (should_add_rcu) {
rjson::value entry = rjson::empty_object();
rjson::add(entry, "TableName", table);
rjson::add(entry, "CapacityUnits", rcu_half_units*0.5);
rjson::push_back(consumed_capacity, std::move(entry));
}
}
if (should_add_rcu) {
rjson::add(response, "ConsumedCapacity", std::move(consumed_capacity));
}
elogger.trace("Unprocessed keys: {}", response["UnprocessedKeys"]);
if (!some_succeeded && eptr) {
co_await coroutine::return_exception_ptr(std::move(eptr));
}
auto duration = std::chrono::steady_clock::now() - start_time;
_stats.api_operations.batch_get_item_latency.mark(duration);
for (const table_requests& rs : requests) {
lw_shared_ptr<stats> per_table_stats = get_stats_from_schema(_proxy, *rs.schema);
per_table_stats->api_operations.batch_get_item_latency.mark(duration);
}
if (is_big(response)) {
co_return make_streamed(std::move(response));
} else {
co_return rjson::print(std::move(response));
}
}
// "filter" represents a condition that can be applied to individual items
// read by a Query or Scan operation, to decide whether to keep the item.
// A filter is constructed from a Query or Scan request. This uses the
// relevant fields in the query (FilterExpression or QueryFilter/ScanFilter +
// ConditionalOperator). These fields are pre-checked and pre-parsed as much
// as possible, to ensure that later checking of many items is efficient.
class filter {
private:
// Holding QueryFilter/ScanFilter + ConditionalOperator:
struct conditions_filter {
bool require_all;
rjson::value conditions;
};
// Holding a parsed FilterExpression:
struct expression_filter {
parsed::condition_expression expression;
};
std::optional<std::variant<conditions_filter, expression_filter>> _imp;
public:
// Filtering for Scan and Query are very similar, but there are some
// small differences, especially the names of the request attributes.
enum class request_type { SCAN, QUERY };
// Note that a filter does not store pointers to the query used to
// construct it.
filter(parsed::expression_cache& parsed_expression_cache, const rjson::value& request, request_type rt,
std::unordered_set<std::string>& used_attribute_names,
std::unordered_set<std::string>& used_attribute_values);
bool check(const rjson::value& item) const;
bool filters_on(std::string_view attribute) const;
// for_filters_on() runs the given function on the attributes that the
// filter works on. It may run for the same attribute more than once if
// used more than once in the filter.
void for_filters_on(const noncopyable_function<void(std::string_view)>& func) const;
operator bool() const { return bool(_imp); }
};
filter::filter(parsed::expression_cache& parsed_expression_cache, const rjson::value& request, request_type rt,
std::unordered_set<std::string>& used_attribute_names,
std::unordered_set<std::string>& used_attribute_values) {
const rjson::value* expression = rjson::find(request, "FilterExpression");
const char* conditions_attribute = (rt == request_type::SCAN) ? "ScanFilter" : "QueryFilter";
const rjson::value* conditions = rjson::find(request, conditions_attribute);
auto conditional_operator = get_conditional_operator(request);
if (conditional_operator != conditional_operator_type::MISSING &&
(!conditions || (conditions->IsObject() && conditions->GetObject().ObjectEmpty()))) {
throw api_error::validation(
format("'ConditionalOperator' parameter cannot be specified for missing or empty {}",
conditions_attribute));
}
if (expression && conditions) {
throw api_error::validation(
format("FilterExpression and {} are not allowed together", conditions_attribute));
}
if (expression) {
if (!expression->IsString()) {
throw api_error::validation("FilterExpression must be a string");
}
if (expression->GetStringLength() == 0) {
throw api_error::validation("FilterExpression must not be empty");
}
if (rjson::find(request, "AttributesToGet")) {
throw api_error::validation("Cannot use both old-style and new-style parameters in same request: FilterExpression and AttributesToGet");
}
try {
auto parsed = parsed_expression_cache.parse_condition_expression(rjson::to_string_view(*expression), "FilterExpression");
const rjson::value* expression_attribute_names = rjson::find(request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(request, "ExpressionAttributeValues");
resolve_condition_expression(parsed,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
_imp = expression_filter { std::move(parsed) };
} catch(expressions_syntax_error& e) {
throw api_error::validation(e.what());
}
}
if (conditions) {
if (rjson::find(request, "ProjectionExpression")) {
throw api_error::validation(format("Cannot use both old-style and new-style parameters in same request: {} and ProjectionExpression", conditions_attribute));
}
bool require_all = conditional_operator != conditional_operator_type::OR;
_imp = conditions_filter { require_all, rjson::copy(*conditions) };
}
}
bool filter::check(const rjson::value& item) const {
if (!_imp) {
return true;
}
return std::visit(overloaded_functor {
[&] (const conditions_filter& f) -> bool {
return verify_condition(f.conditions, f.require_all, &item);
},
[&] (const expression_filter& f) -> bool {
return verify_condition_expression(f.expression, &item);
}
}, *_imp);
}
bool filter::filters_on(std::string_view attribute) const {
if (!_imp) {
return false;
}
return std::visit(overloaded_functor {
[&] (const conditions_filter& f) -> bool {
for (auto it = f.conditions.MemberBegin(); it != f.conditions.MemberEnd(); ++it) {
if (rjson::to_string_view(it->name) == attribute) {
return true;
}
}
return false;
},
[&] (const expression_filter& f) -> bool {
return condition_expression_on(f.expression, attribute);
}
}, *_imp);
}
void filter::for_filters_on(const noncopyable_function<void(std::string_view)>& func) const {
if (_imp) {
std::visit(overloaded_functor {
[&] (const conditions_filter& f) -> void {
for (auto it = f.conditions.MemberBegin(); it != f.conditions.MemberEnd(); ++it) {
func(rjson::to_string_view(it->name));
}
},
[&] (const expression_filter& f) -> void {
return for_condition_expression_on(f.expression, func);
}
}, *_imp);
}
}
class describe_items_visitor {
typedef std::vector<const column_definition*> columns_t;
const columns_t& _columns;
const std::optional<attrs_to_get>& _attrs_to_get;
std::unordered_set<std::string> _extra_filter_attrs;
const filter& _filter;
typename columns_t::const_iterator _column_it;
rjson::value _item;
// _items is a chunked_vector<rjson::value> instead of a RapidJson array
// (rjson::value) because unfortunately RapidJson arrays are stored
// contiguously in memory, and cause large allocations when a Query/Scan
// returns a long list of short items (issue #23535).
utils::chunked_vector<rjson::value> _items;
size_t _scanned_count;
public:
describe_items_visitor(const columns_t& columns, const std::optional<attrs_to_get>& attrs_to_get, filter& filter)
: _columns(columns)
, _attrs_to_get(attrs_to_get)
, _filter(filter)
, _column_it(columns.begin())
, _item(rjson::empty_object())
, _scanned_count(0)
{
// _filter.check() may need additional attributes not listed in
// _attrs_to_get (i.e., not requested as part of the output).
// We list those in _extra_filter_attrs. We will include them in
// the JSON but take them out before finally returning the JSON.
if (_attrs_to_get) {
_filter.for_filters_on([&] (std::string_view attr) {
std::string a(attr); // no heterogeneous maps searches :-(
if (!_attrs_to_get->contains(a)) {
_extra_filter_attrs.emplace(std::move(a));
}
});
}
}
void start_row() {
_column_it = _columns.begin();
}
void accept_value(managed_bytes_view_opt result_bytes_view) {
if (!result_bytes_view) {
++_column_it;
return;
}
result_bytes_view->with_linearized([this] (bytes_view bv) {
std::string column_name = (*_column_it)->name_as_text();
if (column_name != executor::ATTRS_COLUMN_NAME) {
if (!_attrs_to_get || _attrs_to_get->contains(column_name) || _extra_filter_attrs.contains(column_name)) {
if (!_item.HasMember(column_name.c_str())) {
rjson::add_with_string_name(_item, column_name, rjson::empty_object());
}
rjson::value& field = _item[column_name.c_str()];
rjson::add_with_string_name(field, type_to_string((*_column_it)->type), json_key_column_value(bv, **_column_it));
}
} else {
auto deserialized = attrs_type()->deserialize(bv);
auto keys_and_values = value_cast<map_type_impl::native_type>(deserialized);
for (auto entry : keys_and_values) {
std::string attr_name = value_cast<sstring>(entry.first);
if (!_attrs_to_get || _attrs_to_get->contains(attr_name) || _extra_filter_attrs.contains(attr_name)) {
bytes value = value_cast<bytes>(entry.second);
// Even if _attrs_to_get asked to keep only a part of a
// top-level attribute, we keep the entire attribute
// at this stage, because the item filter might still
// need the other parts (it was easier for us to keep
// extra_filter_attrs at top-level granularity). We'll
// filter the unneeded parts after item filtering.
rjson::add_with_string_name(_item, attr_name, deserialize_item(value));
}
}
}
});
++_column_it;
}
void end_row() {
if (_filter.check(_item)) {
// As noted above, we kept entire top-level attributes listed in
// _attrs_to_get. We may need to only keep parts of them.
if (_attrs_to_get) {
for (const auto& attr: *_attrs_to_get) {
// If !attr.has_value() it means we were asked not to keep
// attr entirely, but just parts of it.
if (!attr.second.has_value()) {
rjson::value* toplevel= rjson::find(_item, attr.first);
if (toplevel && !hierarchy_filter(*toplevel, attr.second)) {
rjson::remove_member(_item, attr.first);
}
}
}
}
// Remove the extra attributes _extra_filter_attrs which we had
// to add just for the filter, and not requested to be returned:
for (const auto& attr : _extra_filter_attrs) {
rjson::remove_member(_item, attr);
}
_items.push_back(std::move(_item));
}
_item = rjson::empty_object();
++_scanned_count;
}
utils::chunked_vector<rjson::value> get_items() && {
return std::move(_items);
}
size_t get_scanned_count() {
return _scanned_count;
}
};
// describe_items() returns a JSON object that includes members "Count"
// and "ScannedCount", but *not* "Items" - that is returned separately
// as a chunked_vector to avoid large contiguous allocations which
// RapidJSON does of its array. The caller should add "Items" to the
// returned JSON object if needed, or print it separately.
// The returned chunked_vector (the items) is std::optional<>, because
// the user may have requested only to count items, and not return any
// items - which is different from returning an empty list of items.
static future<std::tuple<rjson::value, std::optional<utils::chunked_vector<rjson::value>>, size_t>> describe_items(
const cql3::selection::selection& selection,
std::unique_ptr<cql3::result_set> result_set,
std::optional<attrs_to_get>&& attrs_to_get,
filter&& filter) {
describe_items_visitor visitor(selection.get_columns(), attrs_to_get, filter);
co_await result_set->visit_gently(visitor);
auto scanned_count = visitor.get_scanned_count();
utils::chunked_vector<rjson::value> items = std::move(visitor).get_items();
rjson::value items_descr = rjson::empty_object();
auto size = items.size();
rjson::add(items_descr, "Count", rjson::value(size));
rjson::add(items_descr, "ScannedCount", rjson::value(scanned_count));
// If attrs_to_get && attrs_to_get->empty(), this means the user asked not
// to get any attributes (i.e., a Scan or Query with Select=COUNT) and we
// shouldn't return "Items" at all.
// TODO: consider optimizing the case of Select=COUNT without a filter.
// In that case, we currently build a list of empty items and here drop
// it. We could just count the items and not bother with the empty items.
// (However, remember that when we do have a filter, we need the items).
std::optional<utils::chunked_vector<rjson::value>> opt_items;
if (!attrs_to_get || !attrs_to_get->empty()) {
opt_items = std::move(items);
}
co_return std::tuple(std::move(items_descr), std::move(opt_items), size);
}
static rjson::value encode_paging_state(const schema& schema, const service::pager::paging_state& paging_state) {
rjson::value last_evaluated_key = rjson::empty_object();
std::vector<bytes> exploded_pk = paging_state.get_partition_key().explode();
auto exploded_pk_it = exploded_pk.begin();
for (const column_definition& cdef : schema.partition_key_columns()) {
rjson::add_with_string_name(last_evaluated_key, std::string_view(cdef.name_as_text()), rjson::empty_object());
rjson::value& key_entry = last_evaluated_key[cdef.name_as_text()];
rjson::add_with_string_name(key_entry, type_to_string(cdef.type), json_key_column_value(*exploded_pk_it, cdef));
++exploded_pk_it;
}
auto pos = paging_state.get_position_in_partition();
if (pos.has_key()) {
// Alternator itself allows at most one column in clustering key, but
// user can use Alternator api to access system tables which might have
// multiple clustering key columns. So we need to handle that case here.
auto cdef_it = schema.clustering_key_columns().begin();
for(const auto &exploded_ck : pos.key().explode()) {
rjson::add_with_string_name(last_evaluated_key, std::string_view(cdef_it->name_as_text()), rjson::empty_object());
rjson::value& key_entry = last_evaluated_key[cdef_it->name_as_text()];
rjson::add_with_string_name(key_entry, type_to_string(cdef_it->type), json_key_column_value(exploded_ck, *cdef_it));
++cdef_it;
}
}
// To avoid possible conflicts (and thus having to reserve these names) we
// avoid adding the weight and region fields of the position to the paging
// state. Alternator will never need these as it doesn't have range
// tombstones (the only thing that can generate a position other than at(row)).
// We conditionally include these fields when reading CQL tables through alternator.
if (!is_alternator_keyspace(schema.ks_name()) && (!pos.has_key() || pos.get_bound_weight() != bound_weight::equal)) {
rjson::add_with_string_name(last_evaluated_key, scylla_paging_region, rjson::empty_object());
rjson::add(last_evaluated_key[scylla_paging_region.data()], "S", rjson::from_string(fmt::to_string(pos.region())));
rjson::add_with_string_name(last_evaluated_key, scylla_paging_weight, rjson::empty_object());
rjson::add(last_evaluated_key[scylla_paging_weight.data()], "N", static_cast<int>(pos.get_bound_weight()));
}
return last_evaluated_key;
}
// RapidJSON allocates arrays contiguously in memory, so we want to avoid
// returning a large number of items as a single rapidjson array, and use
// a chunked_vector instead. The following constant is an arbitrary cutoff
// point for when to switch from a rapidjson array to a chunked_vector.
static constexpr int max_items_for_rapidjson_array = 256;
static future<executor::request_return_type> do_query(service::storage_proxy& proxy,
schema_ptr table_schema,
const rjson::value* exclusive_start_key,
dht::partition_range_vector partition_ranges,
std::vector<query::clustering_range> ck_bounds,
std::optional<attrs_to_get> attrs_to_get,
uint32_t limit,
db::consistency_level cl,
filter filter,
query::partition_slice::option_set custom_opts,
service::client_state& client_state,
alternator::stats& stats,
tracing::trace_state_ptr trace_state,
service_permit permit,
bool enforce_authorization,
bool warn_authorization) {
lw_shared_ptr<service::pager::paging_state> old_paging_state = nullptr;
tracing::trace(trace_state, "Performing a database query");
// Reverse the schema and the clustering bounds as the underlying code expects
// reversed queries in the native reversed format.
auto query_schema = table_schema;
const bool reversed = custom_opts.contains<query::partition_slice::option::reversed>();
if (reversed) {
query_schema = table_schema->get_reversed();
std::reverse(ck_bounds.begin(), ck_bounds.end());
for (auto& bound : ck_bounds) {
bound = query::reverse(bound);
}
}
if (exclusive_start_key) {
partition_key pk = pk_from_json(*exclusive_start_key, table_schema);
auto pos = position_in_partition::for_partition_start();
if (table_schema->clustering_key_size() > 0) {
pos = pos_from_json(*exclusive_start_key, table_schema);
}
old_paging_state = make_lw_shared<service::pager::paging_state>(pk, pos, query::max_partitions, query_id::create_null_id(), service::pager::paging_state::replicas_per_token_range{}, std::nullopt, 0);
}
co_await verify_permission(enforce_authorization, warn_authorization, client_state, table_schema, auth::permission::SELECT, stats);
auto regular_columns =
table_schema->regular_columns() | std::views::transform(&column_definition::id)
| std::ranges::to<query::column_id_vector>();
auto static_columns =
table_schema->static_columns() | std::views::transform(&column_definition::id)
| std::ranges::to<query::column_id_vector>();
auto selection = cql3::selection::selection::wildcard(table_schema);
query::partition_slice::option_set opts = selection->get_query_options();
opts.add(custom_opts);
auto partition_slice = query::partition_slice(std::move(ck_bounds), std::move(static_columns), std::move(regular_columns), opts);
auto command = ::make_lw_shared<query::read_command>(query_schema->id(), query_schema->version(), partition_slice, proxy.get_max_result_size(partition_slice),
query::tombstone_limit(proxy.get_tombstone_limit()));
elogger.trace("Executing read query (reversed {}): table schema {}, query schema {}", partition_slice.is_reversed(), table_schema->version(), query_schema->version());
auto query_state_ptr = std::make_unique<service::query_state>(client_state, trace_state, std::move(permit));
// FIXME: should be moved above, set on opts, so get_max_result_size knows it?
command->slice.options.set<query::partition_slice::option::allow_short_read>();
auto query_options = std::make_unique<cql3::query_options>(cl, std::vector<cql3::raw_value>{});
query_options = std::make_unique<cql3::query_options>(std::move(query_options), std::move(old_paging_state));
auto p = service::pager::query_pagers::pager(proxy, query_schema, selection, *query_state_ptr, *query_options, command, std::move(partition_ranges), nullptr);
std::unique_ptr<cql3::result_set> rs = co_await p->fetch_page(limit, gc_clock::now(), executor::default_timeout());
if (!p->is_exhausted()) {
rs->get_metadata().set_paging_state(p->state());
}
auto paging_state = rs->get_metadata().paging_state();
bool has_filter = filter;
auto [items_descr, opt_items, size] = co_await describe_items(*selection, std::move(rs), std::move(attrs_to_get), std::move(filter));
if (paging_state) {
rjson::add(items_descr, "LastEvaluatedKey", encode_paging_state(*table_schema, *paging_state));
}
if (has_filter) {
stats.cql_stats.filtered_rows_read_total += p->stats().rows_read_total;
// update our "filtered_row_matched_total" for all the rows matched, despited the filter
stats.cql_stats.filtered_rows_matched_total += size;
}
if (opt_items) {
if (opt_items->size() >= max_items_for_rapidjson_array) {
// There are many items, better print the JSON and the array of
// items (opt_items) separately to avoid RapidJSON's contiguous
// allocation of arrays.
co_return make_streamed_with_extra_array(std::move(items_descr), "Items", std::move(*opt_items));
}
// There aren't many items in the chunked vector opt_items,
// let's just insert them into the JSON object and print the
// full JSON normally.
rjson::value items_json = rjson::empty_array();
for (auto& item : *opt_items) {
rjson::push_back(items_json, std::move(item));
}
rjson::add(items_descr, "Items", std::move(items_json));
}
if (is_big(items_descr)) {
co_return make_streamed(std::move(items_descr));
}
co_return rjson::print(std::move(items_descr));
}
static dht::token token_for_segment(int segment, int total_segments) {
throwing_assert(total_segments > 1 && segment >= 0 && segment < total_segments);
uint64_t delta = std::numeric_limits<uint64_t>::max() / total_segments;
return dht::token::from_int64(std::numeric_limits<int64_t>::min() + delta * segment);
}
static dht::partition_range get_range_for_segment(int segment, int total_segments) {
if (total_segments == 1) {
return dht::partition_range::make_open_ended_both_sides();
}
if (segment == 0) {
dht::token ending_token = token_for_segment(1, total_segments);
return dht::partition_range::make_ending_with(
dht::partition_range::bound(dht::ring_position::ending_at(ending_token), false));
} else if (segment == total_segments - 1) {
dht::token starting_token = token_for_segment(segment, total_segments);
return dht::partition_range::make_starting_with(
dht::partition_range::bound(dht::ring_position::starting_at(starting_token)));
} else {
dht::token starting_token = token_for_segment(segment, total_segments);
dht::token ending_token = token_for_segment(segment + 1, total_segments);
return dht::partition_range::make(
dht::partition_range::bound(dht::ring_position::starting_at(starting_token)),
dht::partition_range::bound(dht::ring_position::ending_at(ending_token), false)
);
}
}
future<executor::request_return_type> executor::scan(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.scan++;
elogger.trace("Scanning {}", request);
auto [schema, table_type] = get_table_or_view(_proxy, request);
tracing::add_alternator_table_name(trace_state, schema->cf_name());
get_stats_from_schema(_proxy, *schema)->api_operations.scan++;
auto segment = get_int_attribute(request, "Segment");
auto total_segments = get_int_attribute(request, "TotalSegments");
if (segment || total_segments) {
if (!segment || !total_segments) {
return make_ready_future<request_return_type>(api_error::validation(
"Both Segment and TotalSegments attributes need to be present for a parallel scan"));
}
if (*segment < 0 || *segment >= *total_segments) {
return make_ready_future<request_return_type>(api_error::validation(
"Segment must be non-negative and less than TotalSegments"));
}
if (*total_segments < 0 || *total_segments > 1000000) {
return make_ready_future<request_return_type>(api_error::validation(
"TotalSegments must be non-negative and less or equal to 1000000"));
}
}
rjson::value* exclusive_start_key = rjson::find(request, "ExclusiveStartKey");
db::consistency_level cl = get_read_consistency(request);
if (table_type == table_or_view_type::gsi && cl != db::consistency_level::LOCAL_ONE) {
return make_ready_future<request_return_type>(api_error::validation(
"Consistent reads are not allowed on global indexes (GSI)"));
}
rjson::value* limit_json = rjson::find(request, "Limit");
uint32_t limit = limit_json ? limit_json->GetUint64() : std::numeric_limits<uint32_t>::max();
if (limit <= 0) {
return make_ready_future<request_return_type>(api_error::validation("Limit must be greater than 0"));
}
select_type select = parse_select(request, table_type);
std::unordered_set<std::string> used_attribute_names;
std::unordered_set<std::string> used_attribute_values;
auto attrs_to_get = calculate_attrs_to_get(request, *_parsed_expression_cache, used_attribute_names, select);
dht::partition_range_vector partition_ranges;
if (segment) {
auto range = get_range_for_segment(*segment, *total_segments);
if (exclusive_start_key) {
auto ring_pos = dht::ring_position{dht::decorate_key(*schema, pk_from_json(*exclusive_start_key, schema))};
if (!range.contains(ring_pos, dht::ring_position_comparator(*schema))) {
return make_ready_future<request_return_type>(api_error::validation(
format("The provided starting key is invalid: Invalid ExclusiveStartKey. Please use ExclusiveStartKey "
"with correct Segment. TotalSegments: {} Segment: {}", *total_segments, *segment)));
}
}
partition_ranges.push_back(range);
} else {
partition_ranges.push_back(dht::partition_range::make_open_ended_both_sides());
}
std::vector<query::clustering_range> ck_bounds{query::clustering_range::make_open_ended_both_sides()};
filter filter(*_parsed_expression_cache, request, filter::request_type::SCAN, used_attribute_names, used_attribute_values);
// Note: Unlike Query, Scan does allow a filter on the key attributes.
// For some *specific* cases of key filtering, such an equality test on
// partition key or comparison operator for the sort key, we could have
// optimized the filtering by modifying partition_ranges and/or
// ck_bounds. We haven't done this optimization yet.
const rjson::value* expression_attribute_names = rjson::find(request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(request, "ExpressionAttributeValues");
verify_all_are_used(expression_attribute_names, used_attribute_names, "ExpressionAttributeNames", "Scan");
verify_all_are_used(expression_attribute_values, used_attribute_values, "ExpressionAttributeValues", "Scan");
return do_query(_proxy, schema, exclusive_start_key, std::move(partition_ranges), std::move(ck_bounds), std::move(attrs_to_get), limit, cl,
std::move(filter), query::partition_slice::option_set(), client_state, _stats, trace_state, std::move(permit), _enforce_authorization, _warn_authorization);
}
static dht::partition_range calculate_pk_bound(schema_ptr schema, const column_definition& pk_cdef, const rjson::value& comp_definition, const rjson::value& attrs) {
auto op = get_comparison_operator(comp_definition);
if (op != comparison_operator_type::EQ) {
throw api_error::validation(format("Hash key can only be restricted with equality operator (EQ). {} not supported.", comp_definition));
}
if (attrs.Size() != 1) {
throw api_error::validation(format("A single attribute is required for a hash key EQ restriction: {}", attrs));
}
bytes raw_value = get_key_from_typed_value(attrs[0], pk_cdef);
partition_key pk = partition_key::from_singular_bytes(*schema, std::move(raw_value));
auto decorated_key = dht::decorate_key(*schema, pk);
return dht::partition_range(decorated_key);
}
static query::clustering_range get_clustering_range_for_begins_with(bytes&& target, const clustering_key& ck, schema_ptr schema, data_type t) {
auto it = boost::range::find_end(target, bytes("\xFF"), std::not_equal_to<bytes::value_type>());
if (it != target.end()) {
++*it;
target.resize(std::distance(target.begin(), it) + 1);
clustering_key upper_limit = clustering_key::from_single_value(*schema, target);
return query::clustering_range::make(query::clustering_range::bound(ck), query::clustering_range::bound(upper_limit, false));
}
return query::clustering_range::make_starting_with(query::clustering_range::bound(ck));
}
static query::clustering_range calculate_ck_bound(schema_ptr schema, const column_definition& ck_cdef, const rjson::value& comp_definition, const rjson::value& attrs) {
auto op = get_comparison_operator(comp_definition);
const size_t expected_attrs_size = (op == comparison_operator_type::BETWEEN) ? 2 : 1;
if (attrs.Size() != expected_attrs_size) {
throw api_error::validation(format("{} arguments expected for a sort key restriction: {}", expected_attrs_size, attrs));
}
bytes raw_value = get_key_from_typed_value(attrs[0], ck_cdef);
clustering_key ck = clustering_key::from_single_value(*schema, raw_value);
switch (op) {
case comparison_operator_type::EQ:
return query::clustering_range(ck);
case comparison_operator_type::LE:
return query::clustering_range::make_ending_with(query::clustering_range::bound(ck));
case comparison_operator_type::LT:
return query::clustering_range::make_ending_with(query::clustering_range::bound(ck, false));
case comparison_operator_type::GE:
return query::clustering_range::make_starting_with(query::clustering_range::bound(ck));
case comparison_operator_type::GT:
return query::clustering_range::make_starting_with(query::clustering_range::bound(ck, false));
case comparison_operator_type::BETWEEN: {
bytes raw_upper_limit = get_key_from_typed_value(attrs[1], ck_cdef);
clustering_key upper_limit = clustering_key::from_single_value(*schema, raw_upper_limit);
return query::clustering_range::make(query::clustering_range::bound(ck), query::clustering_range::bound(upper_limit));
}
case comparison_operator_type::BEGINS_WITH: {
if (raw_value.empty()) {
return query::clustering_range::make_open_ended_both_sides();
}
// NOTICE(sarna): A range starting with given prefix and ending (non-inclusively) with a string "incremented" by a single
// character at the end. Throws for NUMBER instances.
if (!ck_cdef.type->is_compatible_with(*utf8_type)) {
throw api_error::validation(fmt::format("BEGINS_WITH operator cannot be applied to type {}", type_to_string(ck_cdef.type)));
}
return get_clustering_range_for_begins_with(std::move(raw_value), ck, schema, ck_cdef.type);
}
default:
throw api_error::validation(format("Operator {} not supported for sort key", comp_definition));
}
}
// Calculates primary key bounds from KeyConditions
static std::pair<dht::partition_range_vector, std::vector<query::clustering_range>>
calculate_bounds_conditions(schema_ptr schema, const rjson::value& conditions) {
dht::partition_range_vector partition_ranges;
std::vector<query::clustering_range> ck_bounds;
for (auto it = conditions.MemberBegin(); it != conditions.MemberEnd(); ++it) {
sstring key = rjson::to_sstring(it->name);
const rjson::value& condition = it->value;
const rjson::value& comp_definition = rjson::get(condition, "ComparisonOperator");
const rjson::value& attr_list = rjson::get(condition, "AttributeValueList");
const column_definition& pk_cdef = schema->partition_key_columns().front();
const column_definition* ck_cdef = schema->clustering_key_size() > 0 ? &schema->clustering_key_columns().front() : nullptr;
if (key == pk_cdef.name_as_text()) {
if (!partition_ranges.empty()) {
throw api_error::validation("Currently only a single restriction per key is allowed");
}
partition_ranges.push_back(calculate_pk_bound(schema, pk_cdef, comp_definition, attr_list));
}
if (ck_cdef && key == ck_cdef->name_as_text()) {
if (!ck_bounds.empty()) {
throw api_error::validation("Currently only a single restriction per key is allowed");
}
ck_bounds.push_back(calculate_ck_bound(schema, *ck_cdef, comp_definition, attr_list));
}
}
// Validate that a query's conditions must be on the hash key, and
// optionally also on the sort key if it exists.
if (partition_ranges.empty()) {
throw api_error::validation(format("Query missing condition on hash key '{}'", schema->partition_key_columns().front().name_as_text()));
}
if (schema->clustering_key_size() == 0) {
if (conditions.MemberCount() != 1) {
throw api_error::validation("Only one condition allowed in table with only hash key");
}
} else {
if (conditions.MemberCount() == 2 && ck_bounds.empty()) {
throw api_error::validation(format("Query missing condition on sort key '{}'", schema->clustering_key_columns().front().name_as_text()));
} else if (conditions.MemberCount() > 2) {
throw api_error::validation("Only one or two conditions allowed in table with hash key and sort key");
}
}
if (ck_bounds.empty()) {
ck_bounds.push_back(query::clustering_range::make_open_ended_both_sides());
}
return {std::move(partition_ranges), std::move(ck_bounds)};
}
// Extract the top-level column name specified in a KeyConditionExpression.
// If a nested attribute path is given, a ValidationException is generated.
// If the column name is a #reference to ExpressionAttributeNames, the
// reference is resolved.
// Note this function returns a string_view, which may refer to data in the
// given parsed::value or expression_attribute_names.
static std::string_view get_toplevel(const parsed::value& v,
const rjson::value* expression_attribute_names,
std::unordered_set<std::string>& used_attribute_names)
{
const parsed::path& path = std::get<parsed::path>(v._value);
if (path.has_operators()) {
throw api_error::validation("KeyConditionExpression does not support nested attributes");
}
std::string_view column_name = path.root();
if (column_name.size() > 0 && column_name[0] == '#') {
used_attribute_names.emplace(column_name);
if (!expression_attribute_names) {
throw api_error::validation(
fmt::format("ExpressionAttributeNames missing, entry '{}' required by KeyConditionExpression",
column_name));
}
const rjson::value* value = rjson::find(*expression_attribute_names, column_name);
if (!value || !value->IsString()) {
throw api_error::validation(
fmt::format("ExpressionAttributeNames missing entry '{}' required by KeyConditionExpression",
column_name));
}
column_name = rjson::to_string_view(*value);
}
return column_name;
}
// Extract a constant value specified in a KeyConditionExpression.
// This constant was originally parsed as a reference (:name) to a member of
// ExpressionAttributeValues, but at this point, after resolve_value(), it
// was already converted into a JSON value.
// This function decodes the value (using its given expected type) into bytes
// which Scylla uses as the actual key value. If the value has the wrong type,
// or the input had other problems, a ValidationException is thrown.
static bytes get_constant_value(const parsed::value& v,
const column_definition& column)
{
const parsed::constant& constant = std::get<parsed::constant>(v._value);
const parsed::constant::literal& lit = std::get<parsed::constant::literal>(constant._value);
return get_key_from_typed_value(*lit, column);
}
// condition_expression_and_list extracts a list of ANDed primitive conditions
// from a condition_expression. This is useful for KeyConditionExpression,
// which may not use OR or NOT. If the given condition_expression does use
// OR or NOT, this function throws a ValidationException.
static void condition_expression_and_list(
const parsed::condition_expression& condition_expression,
std::vector<const parsed::primitive_condition*>& conditions)
{
if (condition_expression._negated) {
throw api_error::validation("KeyConditionExpression cannot use NOT");
}
std::visit(overloaded_functor {
[&] (const parsed::primitive_condition& cond) {
conditions.push_back(&cond);
},
[&] (const parsed::condition_expression::condition_list& list) {
if (list.op == '|' && list.conditions.size() > 1) {
throw api_error::validation("KeyConditionExpression cannot use OR");
}
for (const parsed::condition_expression& cond : list.conditions) {
condition_expression_and_list(cond, conditions);
}
}
}, condition_expression._expression);
}
// Calculates primary key bounds from KeyConditionExpression
static std::pair<dht::partition_range_vector, std::vector<query::clustering_range>>
calculate_bounds_condition_expression(schema_ptr schema,
const rjson::value& expression,
const rjson::value* expression_attribute_values,
std::unordered_set<std::string>& used_attribute_values,
const rjson::value* expression_attribute_names,
std::unordered_set<std::string>& used_attribute_names,
parsed::expression_cache& parsed_expression_cache)
{
if (!expression.IsString()) {
throw api_error::validation("KeyConditionExpression must be a string");
}
if (expression.GetStringLength() == 0) {
throw api_error::validation("KeyConditionExpression must not be empty");
}
// We parse the KeyConditionExpression with the same parser we use for
// ConditionExpression. But KeyConditionExpression only supports a subset
// of the ConditionExpression features, so we have many additional
// verifications below that the key condition is legal. Briefly, a valid
// key condition must contain a single partition key and a single
// sort-key range.
parsed::condition_expression p;
try {
p = parsed_expression_cache.parse_condition_expression(rjson::to_string_view(expression), "KeyConditionExpression");
} catch(expressions_syntax_error& e) {
throw api_error::validation(e.what());
}
resolve_condition_expression(p,
expression_attribute_names, expression_attribute_values,
used_attribute_names, used_attribute_values);
std::vector<const parsed::primitive_condition*> conditions;
condition_expression_and_list(p, conditions);
if (conditions.size() < 1 || conditions.size() > 2) {
throw api_error::validation(
"KeyConditionExpression syntax error: must have 1 or 2 conditions");
}
// Scylla allows us to have an (equality) constraint on the partition key
// pk_cdef, and a range constraint on the *first* clustering key ck_cdef.
// Note that this is also good enough for our GSI implementation - the
// GSI's user-specified sort key will be the first clustering key.
// FIXME: In the case described in issue #5320 (base and GSI both have
// just hash key - but different ones), this may allow the user to Query
// using the base key which isn't officially part of the GSI.
const column_definition& pk_cdef = schema->partition_key_columns().front();
const column_definition* ck_cdef = schema->clustering_key_size() > 0 ?
&schema->clustering_key_columns().front() : nullptr;
dht::partition_range_vector partition_ranges;
std::vector<query::clustering_range> ck_bounds;
for (const parsed::primitive_condition* condp : conditions) {
const parsed::primitive_condition& cond = *condp;
// In all comparison operators, one operand must be a column name,
// the other is a constant (value reference). We remember which is
// which in toplevel_ind, and also the column name in key (not just
// for comparison operators).
std::string_view key;
int toplevel_ind;
switch (cond._values.size()) {
case 1: {
// The only legal single-value condition is a begin_with() function,
// and it must have two parameters - a top-level attribute and a
// value reference..
const parsed::value::function_call *f = std::get_if<parsed::value::function_call>(&cond._values[0]._value);
if (!f) {
throw api_error::validation("KeyConditionExpression cannot be just a value");
}
if (f->_function_name != "begins_with") {
throw api_error::validation(
fmt::format("KeyConditionExpression function '{}' not supported",f->_function_name));
}
if (f->_parameters.size() != 2 || !f->_parameters[0].is_path() ||
!f->_parameters[1].is_constant()) {
throw api_error::validation(
"KeyConditionExpression begins_with() takes attribute and value");
}
key = get_toplevel(f->_parameters[0], expression_attribute_names, used_attribute_names);
toplevel_ind = -1;
break;
}
case 2:
if (cond._values[0].is_path() && cond._values[1].is_constant()) {
toplevel_ind = 0;
} else if (cond._values[1].is_path() && cond._values[0].is_constant()) {
toplevel_ind = 1;
} else {
throw api_error::validation("KeyConditionExpression must compare attribute with constant");
}
key = get_toplevel(cond._values[toplevel_ind], expression_attribute_names, used_attribute_names);
break;
case 3:
// Only BETWEEN has three operands. First must be a column name,
// two other must be value references (constants):
if (cond._op != parsed::primitive_condition::type::BETWEEN) {
// Shouldn't happen unless we have a bug in the parser
throw std::logic_error(format("Wrong number of values {} in primitive_condition", cond._values.size()));
}
if (cond._values[0].is_path() && cond._values[1].is_constant() && cond._values[2].is_constant()) {
toplevel_ind = 0;
key = get_toplevel(cond._values[0], expression_attribute_names, used_attribute_names);
} else {
throw api_error::validation("KeyConditionExpression must compare attribute with constants");
}
break;
default:
// Shouldn't happen unless we have a bug in the parser
throw std::logic_error(format("Wrong number of values {} in primitive_condition", cond._values.size()));
}
if (cond._op == parsed::primitive_condition::type::IN) {
throw api_error::validation("KeyConditionExpression does not support IN operator");
} else if (cond._op == parsed::primitive_condition::type::NE) {
throw api_error::validation("KeyConditionExpression does not support NE operator");
} else if (cond._op == parsed::primitive_condition::type::EQ) {
// the EQ operator (=) is the only one which can be used for both
// the partition key and sort key:
if (sstring(key) == pk_cdef.name_as_text()) {
if (!partition_ranges.empty()) {
throw api_error::validation(
"KeyConditionExpression allows only one condition for each key");
}
bytes raw_value = get_constant_value(cond._values[!toplevel_ind], pk_cdef);
partition_key pk = partition_key::from_singular_bytes(*schema, std::move(raw_value));
auto decorated_key = dht::decorate_key(*schema, pk);
partition_ranges.push_back(dht::partition_range(decorated_key));
} else if (ck_cdef && sstring(key) == ck_cdef->name_as_text()) {
if (!ck_bounds.empty()) {
throw api_error::validation(
"KeyConditionExpression allows only one condition for each key");
}
bytes raw_value = get_constant_value(cond._values[!toplevel_ind], *ck_cdef);
clustering_key ck = clustering_key::from_single_value(*schema, raw_value);
ck_bounds.push_back(query::clustering_range(ck));
} else {
throw api_error::validation(
fmt::format("KeyConditionExpression condition on non-key attribute {}", key));
}
continue;
}
// If we're still here, it's any other operator besides EQ, and these
// are allowed *only* on the clustering key:
if (sstring(key) == pk_cdef.name_as_text()) {
throw api_error::validation(
fmt::format("KeyConditionExpression only '=' condition is supported on partition key {}", key));
} else if (!ck_cdef || sstring(key) != ck_cdef->name_as_text()) {
throw api_error::validation(
fmt::format("KeyConditionExpression condition on non-key attribute {}", key));
}
if (!ck_bounds.empty()) {
throw api_error::validation(
"KeyConditionExpression allows only one condition for each key");
}
if (cond._op == parsed::primitive_condition::type::BETWEEN) {
clustering_key ck1 = clustering_key::from_single_value(*schema,
get_constant_value(cond._values[1], *ck_cdef));
clustering_key ck2 = clustering_key::from_single_value(*schema,
get_constant_value(cond._values[2], *ck_cdef));
ck_bounds.push_back(query::clustering_range::make(
query::clustering_range::bound(ck1), query::clustering_range::bound(ck2)));
continue;
} else if (cond._values.size() == 1) {
// We already verified above, that this case this can only be a
// function call to begins_with(), with the first parameter the
// key, the second the value reference.
bytes raw_value = get_constant_value(
std::get<parsed::value::function_call>(cond._values[0]._value)._parameters[1], *ck_cdef);
if (!ck_cdef->type->is_compatible_with(*utf8_type)) {
// begins_with() supported on bytes and strings (both stored
// in the database as strings) but not on numbers.
throw api_error::validation(
fmt::format("KeyConditionExpression begins_with() not supported on type {}",
type_to_string(ck_cdef->type)));
} else if (raw_value.empty()) {
ck_bounds.push_back(query::clustering_range::make_open_ended_both_sides());
} else {
clustering_key ck = clustering_key::from_single_value(*schema, raw_value);
ck_bounds.push_back(get_clustering_range_for_begins_with(std::move(raw_value), ck, schema, ck_cdef->type));
}
continue;
}
// All remaining operator have one value reference parameter in index
// !toplevel_ind. Note how toplevel_ind==1 reverses the direction of
// an inequality.
bytes raw_value = get_constant_value(cond._values[!toplevel_ind], *ck_cdef);
clustering_key ck = clustering_key::from_single_value(*schema, raw_value);
if ((cond._op == parsed::primitive_condition::type::LT && toplevel_ind == 0) ||
(cond._op == parsed::primitive_condition::type::GT && toplevel_ind == 1)) {
ck_bounds.push_back(query::clustering_range::make_ending_with(query::clustering_range::bound(ck, false)));
} else if ((cond._op == parsed::primitive_condition::type::GT && toplevel_ind == 0) ||
(cond._op == parsed::primitive_condition::type::LT && toplevel_ind == 1)) {
ck_bounds.push_back(query::clustering_range::make_starting_with(query::clustering_range::bound(ck, false)));
} else if ((cond._op == parsed::primitive_condition::type::LE && toplevel_ind == 0) ||
(cond._op == parsed::primitive_condition::type::GE && toplevel_ind == 1)) {
ck_bounds.push_back(query::clustering_range::make_ending_with(query::clustering_range::bound(ck)));
} else if ((cond._op == parsed::primitive_condition::type::GE && toplevel_ind == 0) ||
(cond._op == parsed::primitive_condition::type::LE && toplevel_ind == 1)) {
ck_bounds.push_back(query::clustering_range::make_starting_with(query::clustering_range::bound(ck)));
}
}
if (partition_ranges.empty()) {
throw api_error::validation(
format("KeyConditionExpression requires a condition on partition key {}", pk_cdef.name_as_text()));
}
if (ck_bounds.empty()) {
ck_bounds.push_back(query::clustering_range::make_open_ended_both_sides());
}
return {std::move(partition_ranges), std::move(ck_bounds)};
}
future<executor::request_return_type> executor::query(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit, rjson::value request) {
_stats.api_operations.query++;
elogger.trace("Querying {}", request);
auto [schema, table_type] = get_table_or_view(_proxy, request);
get_stats_from_schema(_proxy, *schema)->api_operations.query++;
tracing::add_alternator_table_name(trace_state, schema->cf_name());
rjson::value* exclusive_start_key = rjson::find(request, "ExclusiveStartKey");
db::consistency_level cl = get_read_consistency(request);
if (table_type == table_or_view_type::gsi && cl != db::consistency_level::LOCAL_ONE) {
return make_ready_future<request_return_type>(api_error::validation(
"Consistent reads are not allowed on global indexes (GSI)"));
}
rjson::value* limit_json = rjson::find(request, "Limit");
uint32_t limit = limit_json ? limit_json->GetUint64() : std::numeric_limits<uint32_t>::max();
if (limit <= 0) {
return make_ready_future<request_return_type>(api_error::validation("Limit must be greater than 0"));
}
const bool forward = get_bool_attribute(request, "ScanIndexForward", true);
rjson::value* key_conditions = rjson::find(request, "KeyConditions");
rjson::value* key_condition_expression = rjson::find(request, "KeyConditionExpression");
std::unordered_set<std::string> used_attribute_values;
std::unordered_set<std::string> used_attribute_names;
if (key_conditions && key_condition_expression) {
throw api_error::validation("Query does not allow both "
"KeyConditions and KeyConditionExpression to be given together");
} else if (!key_conditions && !key_condition_expression) {
throw api_error::validation("Query must have one of "
"KeyConditions or KeyConditionExpression");
}
const rjson::value* expression_attribute_names = rjson::find(request, "ExpressionAttributeNames");
const rjson::value* expression_attribute_values = rjson::find(request, "ExpressionAttributeValues");
// exactly one of key_conditions or key_condition_expression
auto [partition_ranges, ck_bounds] = key_conditions
? calculate_bounds_conditions(schema, *key_conditions)
: calculate_bounds_condition_expression(schema, *key_condition_expression,
expression_attribute_values,
used_attribute_values,
expression_attribute_names,
used_attribute_names, *_parsed_expression_cache);
filter filter(*_parsed_expression_cache, request, filter::request_type::QUERY,
used_attribute_names, used_attribute_values);
// A query is not allowed to filter on the partition key or the sort key.
for (const column_definition& cdef : schema->partition_key_columns()) { // just one
if (filter.filters_on(cdef.name_as_text())) {
return make_ready_future<request_return_type>(api_error::validation(
format("QueryFilter can only contain non-primary key attributes: Partition key attribute: {}", cdef.name_as_text())));
}
}
for (const column_definition& cdef : schema->clustering_key_columns()) {
if (filter.filters_on(cdef.name_as_text())) {
return make_ready_future<request_return_type>(api_error::validation(
format("QueryFilter can only contain non-primary key attributes: Sort key attribute: {}", cdef.name_as_text())));
}
// FIXME: this "break" can avoid listing some clustering key columns
// we added for GSIs just because they existed in the base table -
// but not in all cases. We still have issue #5320.
break;
}
select_type select = parse_select(request, table_type);
auto attrs_to_get = calculate_attrs_to_get(request, *_parsed_expression_cache, used_attribute_names, select);
verify_all_are_used(expression_attribute_names, used_attribute_names, "ExpressionAttributeNames", "Query");
verify_all_are_used(expression_attribute_values, used_attribute_values, "ExpressionAttributeValues", "Query");
query::partition_slice::option_set opts;
opts.set_if<query::partition_slice::option::reversed>(!forward);
return do_query(_proxy, schema, exclusive_start_key, std::move(partition_ranges), std::move(ck_bounds), std::move(attrs_to_get), limit, cl,
std::move(filter), opts, client_state, _stats, std::move(trace_state), std::move(permit), _enforce_authorization, _warn_authorization);
}
future<executor::request_return_type> executor::list_tables(client_state& client_state, service_permit permit, rjson::value request) {
_stats.api_operations.list_tables++;
elogger.trace("Listing tables {}", request);
co_await utils::get_local_injector().inject("alternator_list_tables", [] (auto& handler) -> future<> {
handler.set("waiting", true);
co_await handler.wait_for_message(std::chrono::steady_clock::now() + std::chrono::minutes{5});
});
rjson::value* exclusive_start_json = rjson::find(request, "ExclusiveStartTableName");
rjson::value* limit_json = rjson::find(request, "Limit");
std::string exclusive_start = exclusive_start_json ? rjson::to_string(*exclusive_start_json) : "";
int limit = limit_json ? limit_json->GetInt() : 100;
if (limit < 1 || limit > 100) {
co_return api_error::validation("Limit must be greater than 0 and no greater than 100");
}
auto tables = _proxy.data_dictionary().get_tables(); // hold on to temporary, table_names isn't a container, it's a view
auto table_names = tables
| std::views::filter([this] (data_dictionary::table t) {
return t.schema()->ks_name().find(KEYSPACE_NAME_PREFIX) == 0 &&
!t.schema()->is_view() &&
!cdc::is_log_for_some_table(_proxy.local_db(), t.schema()->ks_name(), t.schema()->cf_name());
})
| std::views::transform([] (data_dictionary::table t) {
return t.schema()->cf_name();
});
rjson::value response = rjson::empty_object();
rjson::add(response, "TableNames", rjson::empty_array());
rjson::value& all_tables = response["TableNames"];
//TODO(sarna): Dynamo doesn't declare any ordering when listing tables,
// but our implementation is vulnerable to changes, because the tables
// are stored in an unordered map. We may consider (partially) sorting
// the results before returning them to the client, especially if there
// is an implicit order of elements that Dynamo imposes.
auto table_names_it = [&table_names, &exclusive_start] {
if (!exclusive_start.empty()) {
auto it = std::ranges::find_if(table_names, [&exclusive_start] (const sstring& table_name) { return table_name == exclusive_start; });
return std::next(it, it != table_names.end());
} else {
return table_names.begin();
}
}();
while (limit > 0 && table_names_it != table_names.end()) {
rjson::push_back(all_tables, rjson::from_string(*table_names_it));
--limit;
++table_names_it;
}
if (table_names_it != table_names.end()) {
auto& last_table_name = *std::prev(all_tables.End());
rjson::add(response, "LastEvaluatedTableName", rjson::copy(last_table_name));
}
co_return rjson::print(std::move(response));
}
future<executor::request_return_type> executor::describe_endpoints(client_state& client_state, service_permit permit, rjson::value request, std::string host_header) {
_stats.api_operations.describe_endpoints++;
// The alternator_describe_endpoints configuration can be used to disable
// the DescribeEndpoints operation, or set it to return a fixed string
std::string override = _proxy.data_dictionary().get_config().alternator_describe_endpoints();
if (!override.empty()) {
if (override == "disabled") {
_stats.unsupported_operations++;
co_return api_error::unknown_operation(
"DescribeEndpoints disabled by configuration (alternator_describe_endpoints=disabled)");
}
host_header = std::move(override);
}
rjson::value response = rjson::empty_object();
// Without having any configuration parameter to say otherwise, we tell
// the user to return to the same endpoint they used to reach us. The only
// way we can know this is through the "Host:" header in the request,
// which typically exists (and in fact is mandatory in HTTP 1.1).
// A "Host:" header includes both host name and port, exactly what we need
// to return.
if (host_header.empty()) {
co_return api_error::validation("DescribeEndpoints needs a 'Host:' header in request");
}
rjson::add(response, "Endpoints", rjson::empty_array());
rjson::push_back(response["Endpoints"], rjson::empty_object());
rjson::add(response["Endpoints"][0], "Address", rjson::from_string(host_header));
rjson::add(response["Endpoints"][0], "CachePeriodInMinutes", rjson::value(1440));
co_return rjson::print(std::move(response));
}
static locator::replication_strategy_config_options get_network_topology_options(service::storage_proxy& sp, gms::gossiper& gossiper, int rf) {
locator::replication_strategy_config_options options;
for (const auto& dc : sp.get_token_metadata_ptr()->get_datacenter_racks_token_owners() | std::views::keys) {
options.emplace(dc, std::to_string(rf));
}
return options;
}
future<executor::request_return_type> executor::describe_continuous_backups(client_state& client_state, service_permit permit, rjson::value request) {
_stats.api_operations.describe_continuous_backups++;
// Unlike most operations which return ResourceNotFound when the given
// table doesn't exists, this operation returns a TableNoteFoundException.
// So we can't use the usual get_table() wrapper and need a bit more code:
std::string table_name = get_table_name(request);
schema_ptr schema;
try {
schema = _proxy.data_dictionary().find_schema(sstring(executor::KEYSPACE_NAME_PREFIX) + table_name, table_name);
} catch(data_dictionary::no_such_column_family&) {
// DynamoDB returns validation error even when table does not exist
// and the table name is invalid.
validate_table_name(table_name);
throw api_error::table_not_found(
fmt::format("Table {} not found", table_name));
}
rjson::value desc = rjson::empty_object();
rjson::add(desc, "ContinuousBackupsStatus", "DISABLED");
rjson::value pitr = rjson::empty_object();
rjson::add(pitr, "PointInTimeRecoveryStatus", "DISABLED");
rjson::add(desc, "PointInTimeRecoveryDescription", std::move(pitr));
rjson::value response = rjson::empty_object();
rjson::add(response, "ContinuousBackupsDescription", std::move(desc));
co_return rjson::print(std::move(response));
}
// Create the metadata for the keyspace in which we put the alternator
// table if it doesn't already exist.
// Currently, we automatically configure the keyspace based on the number
// of nodes in the cluster: A cluster with 3 or more live nodes, gets RF=3.
// A smaller cluster (presumably, a test only), gets RF=1. The user may
// manually create the keyspace to override this predefined behavior.
static lw_shared_ptr<keyspace_metadata> create_keyspace_metadata(std::string_view keyspace_name, service::storage_proxy& sp, gms::gossiper& gossiper, api::timestamp_type ts,
const std::map<sstring, sstring>& tags_map, const gms::feature_service& feat, const db::tablets_mode_t::mode tablets_mode) {
// Whether to use tablets for the table (actually for the keyspace of the
// table) is determined by tablets_mode (taken from the configuration
// option "tablets_mode_for_new_keyspaces"), as well as the presence and
// the value of a per-table tag system:initial_tablets
// (INITIAL_TABLETS_TAG_KEY).
// Setting the tag with a numeric value will enable a specific initial number
// of tablets (setting the value to 0 means enabling tablets with
// an automatic selection of the best number of tablets).
// Setting this tag to any non-numeric value (e.g., an empty string or the
// word "none") will ask to disable tablets.
// When vnodes are asked for by the tag value, but tablets are enforced by config,
// throw an exception to the client.
std::optional<unsigned> initial_tablets;
if (feat.tablets) {
auto it = tags_map.find(INITIAL_TABLETS_TAG_KEY);
if (it != tags_map.end()) {
// Tag set. If it's a valid number, use it. If not - e.g., it's
// empty or a word like "none", disable tablets by setting
// initial_tablets to a disengaged optional.
try {
initial_tablets = std::stol(tags_map.at(INITIAL_TABLETS_TAG_KEY));
} catch (...) {
if (tablets_mode == db::tablets_mode_t::mode::enforced) {
throw api_error::validation(format("Tag {} containing non-numerical value requests vnodes, but vnodes are forbidden by configuration option `tablets_mode_for_new_keyspaces: enforced`", INITIAL_TABLETS_TAG_KEY));
}
initial_tablets = std::nullopt;
elogger.trace("Following {} tag containing non-numerical value, Alternator will attempt to create a keyspace {} with vnodes.", INITIAL_TABLETS_TAG_KEY, keyspace_name);
}
} else {
// No per-table tag present, use the value from config
if (tablets_mode == db::tablets_mode_t::mode::enabled || tablets_mode == db::tablets_mode_t::mode::enforced) {
initial_tablets = 0;
elogger.trace("Following the `tablets_mode_for_new_keyspaces` flag from the settings, Alternator will attempt to create a keyspace {} with tablets.", keyspace_name);
} else {
initial_tablets = std::nullopt;
elogger.trace("Following the `tablets_mode_for_new_keyspaces` flag from the settings, Alternator will attempt to create a keyspace {} with vnodes.", keyspace_name);
}
}
}
int endpoint_count = gossiper.num_endpoints();
int rf = 3;
if (endpoint_count < rf) {
rf = 1;
elogger.warn("Creating keyspace '{}' for Alternator with unsafe RF={} because cluster only has {} nodes.",
keyspace_name, rf, endpoint_count);
}
auto opts = get_network_topology_options(sp, gossiper, rf);
cql3::statements::ks_prop_defs props;
opts["class"] = sstring("NetworkTopologyStrategy");
props.add_property(cql3::statements::ks_prop_defs::KW_REPLICATION, opts);
std::map<sstring, sstring> tablet_opts;
if (initial_tablets) {
tablet_opts["initial"] = std::to_string(*initial_tablets);
}
tablet_opts["enabled"] = initial_tablets ? "true" : "false";
props.add_property(cql3::statements::ks_prop_defs::KW_TABLETS, std::move(tablet_opts));
props.validate();
return props.as_ks_metadata(sstring(keyspace_name), *sp.get_token_metadata_ptr(), feat, sp.local_db().get_config());
}
future<> executor::start() {
// Currently, nothing to do on initialization. We delay the keyspace
// creation (create_keyspace()) until a table is actually created.
return make_ready_future<>();
}
future<> executor::stop() {
co_await _describe_table_info_manager->stop();
// disconnect from the value source, but keep the value unchanged.
s_default_timeout_in_ms = utils::updateable_value<uint32_t>{s_default_timeout_in_ms()};
co_await _parsed_expression_cache->stop();
}
} // namespace alternator
Reference in New Issue View Git Blame Copy Permalink