transport: implement SCYLLA_USE_METADATA_ID support

Metadata id was introduced in CQLv5 to make metadata of prepared
statement consistent between driver and database. This commit introduces
a protocol extension that allows to use the same mechanism in CQLv4.

This change:
 - Introduce SCYLLA_USE_METADATA_ID protocol extension for CQLv4
 - Introduce METADATA_CHANGED flag in RESULT. The flag cames directly
   from CQLv5 binary protocol. In CQLv4, the bit was never used, so we
   assume it is safe to reuse it.
 - Implement handling of metadata_id and METADATA_CHANGED in RESULT rows
 - Implement returning metadata_id in RESULT prepared
 - Implement reading metadata_id from EXECUTE
 - Added description of SCYLLA_USE_METADATA_ID in documentation

Metadata_id is wrapped in cql_metadata_id_wrapper because we need to
distinguish the following situations:
 - Metadata_id is not supported by the protocol (e.g. CQLv4 without the
   extension is used)
 - Metadata_id is supported by the protocol but not set - e.g. PREPARE
   query is being handled: it doesn't contain metadata_id in the
   request but the reply (RESULT prepared) must contain metadata_id
 - Metadata_id is supported by the protocol and set, any number of
   bytes >= 0 is allowed, according to the CQLv5 protocol specification

Fixes scylladb#20860

Author: andrzej-jackowski-scylladb

cql3/result_set.hh

@@ -27,12 +27,14 @@ public:
         GLOBAL_TABLES_SPEC = 0,
         HAS_MORE_PAGES = 1,
         NO_METADATA = 2,
+        METADATA_CHANGED = 3,
     };
 
     using flag_enum = super_enum<flag,
         flag::GLOBAL_TABLES_SPEC,
         flag::HAS_MORE_PAGES,
-        flag::NO_METADATA>;
+        flag::NO_METADATA,
+        flag::METADATA_CHANGED>;
 
     using flag_enum_set = enum_set<flag_enum>;
 

docs/dev/protocol-extensions.md

@@ -224,3 +224,15 @@ Having a designated flag gives the ability to skip tablet metadata generation
 
 The feature is identified by the `TABLETS_ROUTING_V1` key, which is meant to be sent
 in the SUPPORTED message.
+
+## Negotiate sending metadata id
+
+This extension allows the driver to inform the database that it is aware of
+`metadata id` and is able to interpret the metadata id information.
+
+Metadata id was originally introduced in CQLv5 to make metadata of prepared statement
+consistent between driver and database. This extension allows to use
+the same mechanism for other protocol versions, such as CQLv4.
+
+The feature is identified by the `SCYLLA_USE_METADATA_ID` key, which is meant to be sent
+in the SUPPORTED message.

transport/cql_protocol_extension.cc

@@ -18,7 +18,8 @@ namespace cql_transport {
 static const std::map<cql_protocol_extension, seastar::sstring> EXTENSION_NAMES = {
     {cql_protocol_extension::LWT_ADD_METADATA_MARK, "SCYLLA_LWT_ADD_METADATA_MARK"},
     {cql_protocol_extension::RATE_LIMIT_ERROR, "SCYLLA_RATE_LIMIT_ERROR"},
-    {cql_protocol_extension::TABLETS_ROUTING_V1, "TABLETS_ROUTING_V1"}
+    {cql_protocol_extension::TABLETS_ROUTING_V1, "TABLETS_ROUTING_V1"},
+    {cql_protocol_extension::USE_METADATA_ID, "SCYLLA_USE_METADATA_ID"}
 };
 
 cql_protocol_extension_enum_set supported_cql_protocol_extensions() {

transport/cql_protocol_extension.hh

@@ -28,13 +28,15 @@ namespace cql_transport {
 enum class cql_protocol_extension {
     LWT_ADD_METADATA_MARK,
     RATE_LIMIT_ERROR,
-    TABLETS_ROUTING_V1
+    TABLETS_ROUTING_V1,
+    USE_METADATA_ID
 };
 
 using cql_protocol_extension_enum = super_enum<cql_protocol_extension,
     cql_protocol_extension::LWT_ADD_METADATA_MARK,
     cql_protocol_extension::RATE_LIMIT_ERROR,
-    cql_protocol_extension::TABLETS_ROUTING_V1>;
+    cql_protocol_extension::TABLETS_ROUTING_V1,
+    cql_protocol_extension::USE_METADATA_ID>;
 
 using cql_protocol_extension_enum_set = enum_set<cql_protocol_extension_enum>;
 

transport/response.hh

@@ -8,6 +8,7 @@
 
 #pragma once
 
+#include "cql3/prepared_statements_cache.hh"
 #include "server.hh"
 
 namespace cql_transport {
@@ -76,7 +77,7 @@ public:
     void write_string_bytes_map(const std::unordered_map<sstring, bytes>& map);
     void write_value(bytes_opt value);
     void write_value(std::optional<managed_bytes_view> value);
-    void write(const cql3::metadata& m, bool skip = false);
+    void write(const cql3::metadata& m, const cql_metadata_id_wrapper& metadata_id, bool no_metadata = false);
     void write(const cql3::prepared_metadata& m, uint8_t version);
 
     // Make a non-owning scattered_message of the response. Remains valid as long

transport/server.cc

@@ -8,6 +8,7 @@
 
 #include "server.hh"
 
+#include "cql3/prepared_statements_cache.hh"
 #include "cql3/statements/batch_statement.hh"
 #include "cql3/statements/modification_statement.hh"
 #include <seastar/core/scheduling.hh>
@@ -27,6 +28,7 @@
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/future-util.hh>
 #include <seastar/core/seastar.hh>
+#include <seastar/core/shared_ptr.hh>
 #include <seastar/net/byteorder.hh>
 #include <seastar/core/metrics.hh>
 #include <seastar/net/byteorder.hh>
@@ -174,6 +176,12 @@ sstring to_string(const event::schema_change::target_type t) {
     SCYLLA_ASSERT(false && "unreachable");
 }
 
+bool is_metadata_id_supported(const service::client_state& client_state) {
+    // TODO: metadata_id is mandatory in CQLv5, so extend the check below
+    // when CQLv5 support is implemented
+    return client_state.is_protocol_extension_set(cql_transport::cql_protocol_extension::USE_METADATA_ID);
+}
+
 event::event_type parse_event_type(const sstring& value)
 {
     if (value == "TOPOLOGY_CHANGE") {
@@ -982,7 +990,7 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_op
 
 std::unique_ptr<cql_server::response>
 make_result(int16_t stream, messages::result_message& msg, const tracing::trace_state_ptr& tr_state,
-        cql_protocol_version_type version, bool skip_metadata = false);
+        cql_protocol_version_type version, cql_metadata_id_wrapper&& metadata_id, bool skip_metadata = false);
 
 template <typename Process>
     requires std::is_invocable_r_v<future<cql_server::process_fn_return_type>,
@@ -1078,7 +1086,9 @@ process_query_internal(service::client_state& client_state, distributed<cql3::qu
             return cql_server::process_fn_return_type(convert_error_message_to_coordinator_result(msg.get()));
         } else {
             tracing::trace(q_state->query_state.get_trace_state(), "Done processing - preparing a result");
-            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, q_state->query_state.get_trace_state(), version, skip_metadata)));
+            cql_metadata_id_wrapper metadata_id{is_metadata_id_supported(q_state->query_state.get_client_state())};
+
+            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, q_state->query_state.get_trace_state(), version, std::move(metadata_id), skip_metadata)));
         }
     });
 }
@@ -1101,11 +1111,13 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_pr
             return qp.prepare(std::move(query), client_state, dialect).discard_result();
     }).then([this, query, stream, &client_state, trace_state, dialect] () mutable {
         tracing::trace(trace_state, "Done preparing on remote shards");
-        return _server._query_processor.local().prepare(std::move(query), client_state, dialect).then([this, stream, trace_state] (auto msg) {
+        cql_metadata_id_wrapper metadata_id{is_metadata_id_supported(client_state)};
+
+        return _server._query_processor.local().prepare(std::move(query), client_state, dialect).then([this, stream, trace_state, metadata_id = std::move(metadata_id)] (auto msg) mutable {
             tracing::trace(trace_state, "Done preparing on a local shard - preparing a result. ID is [{}]", seastar::value_of([&msg] {
                 return messages::result_message::prepared::cql::get_id(msg);
             }));
-            return make_result(stream, *msg, trace_state, _version);
+            return make_result(stream, *msg, trace_state, _version, std::move(metadata_id));
         });
     });
 }
@@ -1116,6 +1128,11 @@ process_execute_internal(service::client_state& client_state, distributed<cql3::
         service_permit permit, tracing::trace_state_ptr trace_state, bool init_trace, cql3::computed_function_values cached_pk_fn_calls,
         cql3::dialect dialect) {
     cql3::prepared_cache_key_type cache_key(in.read_short_bytes(), dialect);
+
+    cql_metadata_id_wrapper metadata_id = is_metadata_id_supported(client_state)
+        ? cql_metadata_id_wrapper(cql3::cql_metadata_id_type(in.read_short_bytes()))
+        : cql_metadata_id_wrapper(false);
+
     auto& id = cql3::prepared_cache_key_type::cql_id(cache_key);
     bool needs_authorization = false;
 
@@ -1169,14 +1186,14 @@ process_execute_internal(service::client_state& client_state, distributed<cql3::
 
     tracing::trace(trace_state, "Processing a statement");
     return qp.local().execute_prepared_without_checking_exception_message(query_state, std::move(stmt), options, std::move(prepared), std::move(cache_key), needs_authorization)
-            .then([trace_state = query_state.get_trace_state(), skip_metadata, q_state = std::move(q_state), stream, version] (auto msg) {
+            .then([trace_state = query_state.get_trace_state(), skip_metadata, q_state = std::move(q_state), stream, version, metadata_id = std::move(metadata_id)] (auto msg) mutable {
         if (msg->move_to_shard()) {
             return cql_server::process_fn_return_type(make_foreign(dynamic_pointer_cast<messages::result_message::bounce_to_shard>(msg)));
         } else if (msg->is_exception()) {
             return cql_server::process_fn_return_type(convert_error_message_to_coordinator_result(msg.get()));
         } else {
             tracing::trace(q_state->query_state.get_trace_state(), "Done processing - preparing a result");
-            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, q_state->query_state.get_trace_state(), version, skip_metadata)));
+            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, q_state->query_state.get_trace_state(), version, std::move(metadata_id), skip_metadata)));
         }
     });
 }
@@ -1296,7 +1313,9 @@ process_batch_internal(service::client_state& client_state, distributed<cql3::qu
             return cql_server::process_fn_return_type(convert_error_message_to_coordinator_result(msg.get()));
         } else {
             tracing::trace(q_state->query_state.get_trace_state(), "Done processing - preparing a result");
-            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, trace_state, version)));
+            cql_metadata_id_wrapper metadata_id{is_metadata_id_supported(q_state->query_state.get_client_state())};
+
+            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, trace_state, version, std::move(metadata_id))));
         }
     });
 }
@@ -1510,11 +1529,13 @@ class cql_server::fmt_visitor : public messages::result_message::visitor_base {
     uint8_t _version;
     cql_server::response& _response;
     bool _skip_metadata;
+    cql_metadata_id_wrapper _metadata_id;
 public:
-    fmt_visitor(uint8_t version, cql_server::response& response, bool skip_metadata)
+    fmt_visitor(uint8_t version, cql_server::response& response, bool skip_metadata, cql_metadata_id_wrapper&& metadata_id)
         : _version{version}
         , _response{response}
         , _skip_metadata{skip_metadata}
+        , _metadata_id(std::move(metadata_id))
     { }
 
     virtual void visit(const messages::result_message::void_message&) override {
@@ -1529,8 +1550,11 @@ class cql_server::fmt_visitor : public messages::result_message::visitor_base {
     virtual void visit(const messages::result_message::prepared::cql& m) override {
         _response.write_int(0x0004);
         _response.write_short_bytes(m.get_id());
+        if (_metadata_id.is_supported_by_protocol()) {
+            _response.write_short_bytes(m.result_metadata()->calculate_metadata_id()._metadata_id);
+        }
         _response.write(m.metadata(), _version);
-        _response.write(*m.result_metadata());
+        _response.write(*m.result_metadata(), _metadata_id);
     }
 
     virtual void visit(const messages::result_message::schema_change& m) override {
@@ -1550,7 +1574,7 @@ class cql_server::fmt_visitor : public messages::result_message::visitor_base {
     virtual void visit(const messages::result_message::rows& m) override {
         _response.write_int(0x0002);
         auto& rs = m.rs();
-        _response.write(rs.get_metadata(), _skip_metadata);
+        _response.write(rs.get_metadata(), _metadata_id, _skip_metadata);
         auto row_count_plhldr = _response.write_int_placeholder();
 
         class visitor {
@@ -1578,7 +1602,7 @@ class cql_server::fmt_visitor : public messages::result_message::visitor_base {
 
 std::unique_ptr<cql_server::response>
 make_result(int16_t stream, messages::result_message& msg, const tracing::trace_state_ptr& tr_state,
-        cql_protocol_version_type version, bool skip_metadata) {
+        cql_protocol_version_type version, cql_metadata_id_wrapper&& metadata_id, bool skip_metadata) {
     auto response = std::make_unique<cql_server::response>(stream, cql_binary_opcode::RESULT, tr_state);
     if (__builtin_expect(!msg.warnings().empty() && version > 3, false)) {
         response->set_frame_flag(cql_frame_flags::warning);
@@ -1588,7 +1612,7 @@ make_result(int16_t stream, messages::result_message& msg, const tracing::trace_
         response->set_frame_flag(cql_frame_flags::custom_payload);
         response->write_string_bytes_map(msg.custom_payload().value());
     }
-    cql_server::fmt_visitor fmt{version, *response, skip_metadata};
+    cql_server::fmt_visitor fmt{version, *response, skip_metadata, std::move(metadata_id)};
     msg.accept(fmt);
     return response;
 }
@@ -1996,7 +2020,7 @@ thread_local const type_codec::type_id_to_type_type type_codec::type_id_to_type
     { inet_addr_type, type_id::INET },
 };
 
-void cql_server::response::write(const cql3::metadata& m, bool no_metadata) {
+void cql_server::response::write(const cql3::metadata& m, const cql_metadata_id_wrapper& metadata_id, bool no_metadata) {
     auto flags = m.flags();
     bool global_tables_spec = m.flags().contains<cql3::metadata::flag::GLOBAL_TABLES_SPEC>();
     bool has_more_pages = m.flags().contains<cql3::metadata::flag::HAS_MORE_PAGES>();
@@ -2005,6 +2029,18 @@ void cql_server::response::write(const cql3::metadata& m, bool no_metadata) {
         flags.set<cql3::metadata::flag::NO_METADATA>();
     }
 
+    cql3::cql_metadata_id_type calculated_metadata_id{bytes{}};
+    if (metadata_id.is_non_empty())
+    {
+        calculated_metadata_id = m.calculate_metadata_id();
+        if (calculated_metadata_id != metadata_id.get_metadata_id())
+        {
+            flags.remove<cql3::metadata::flag::NO_METADATA>();
+            flags.set<cql3::metadata::flag::METADATA_CHANGED>();
+            no_metadata = false;
+        }
+    }
+
     write_int(flags.mask());
     write_int(m.column_count());
 
@@ -2016,6 +2052,10 @@ void cql_server::response::write(const cql3::metadata& m, bool no_metadata) {
         return;
     }
 
+    if (flags.contains<cql3::metadata::flag::METADATA_CHANGED>()) {
+        write_bytes_as_string(calculated_metadata_id._metadata_id);
+    }
+
     auto names_i = m.get_names().begin();
 
     if (global_tables_spec) {
@@ -2068,6 +2108,13 @@ void cql_server::response::write(const cql3::prepared_metadata& m, uint8_t versi
     }
 }
 
+const cql3::cql_metadata_id_type& cql_metadata_id_wrapper::get_metadata_id() const {
+    if (_state != state::SUPPORTED_BY_PROTOCOL_NON_EMPTY) {
+        on_internal_error(clogger, "Metadata id is unsupported or empty");
+    }
+    return _metadata_id;
+}
+
 future<utils::chunked_vector<client_data>> cql_server::get_client_data() {
     utils::chunked_vector<client_data> ret;
     co_await for_each_gently([&ret] (const generic_server::connection& c) {

transport/server.hh

@@ -11,6 +11,7 @@
 #include "auth/service.hh"
 #include <seastar/core/seastar.hh>
 #include <seastar/core/scheduling.hh>
+#include "cql3/prepared_statements_cache.hh"
 #include "service/endpoint_lifecycle_subscriber.hh"
 #include "service/migration_listener.hh"
 #include "auth/authenticator.hh"
@@ -148,6 +149,39 @@ struct connection_service_level_params {
     sstring scheduling_group_name;
 };
 
+class cql_metadata_id_wrapper {
+public:
+    enum class state {
+        UNSUPPORTED_BY_PROTOCOL,
+        SUPPORTED_BY_PROTOCOL_EMPTY, // metadata_id supported by protocol but no value provided (e.g. handling QUERY or BATCH)
+        SUPPORTED_BY_PROTOCOL_NON_EMPTY // metadata_id supported by protocol and provided (i.e. handling EXECUTE)
+    };
+private:
+    cql3::cql_metadata_id_type _metadata_id; 
+    state _state;
+
+public:
+    explicit cql_metadata_id_wrapper(bool supported)
+        : _metadata_id{bytes{}}
+        , _state(supported ? state::SUPPORTED_BY_PROTOCOL_EMPTY : state::UNSUPPORTED_BY_PROTOCOL)
+    { }
+
+    explicit cql_metadata_id_wrapper(cql3::cql_metadata_id_type&& metadata_id)
+        : _metadata_id(std::move(metadata_id))
+        , _state(state::SUPPORTED_BY_PROTOCOL_NON_EMPTY)
+    { }
+
+    bool is_supported_by_protocol() const {
+        return _state != state::UNSUPPORTED_BY_PROTOCOL;
+    }
+
+    bool is_non_empty() const {
+        return _state == state::SUPPORTED_BY_PROTOCOL_NON_EMPTY;
+    }
+
+    const cql3::cql_metadata_id_type& get_metadata_id() const;
+};
+
 class cql_server : public seastar::peering_sharded_service<cql_server>, public generic_server::server {
 private:
     struct transport_stats {
@@ -212,7 +246,7 @@ private:
     class fmt_visitor;
     friend class connection;
     friend std::unique_ptr<cql_server::response> make_result(int16_t stream, messages::result_message& msg,
-            const tracing::trace_state_ptr& tr_state, cql_protocol_version_type version, bool skip_metadata);
+            const tracing::trace_state_ptr& tr_state, cql_protocol_version_type version, cql_metadata_id_wrapper&& metadata_id, bool skip_metadata);
 
     class connection : public generic_server::connection {
         cql_server& _server;