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/query_processor.hh

@@ -544,6 +544,7 @@ private:
                 [this, &client_state, &id_getter, d](const prepared_cache_key_type& key, const sstring& query_string) {
             return _prepared_cache.get(key, [this, &query_string, &client_state, d] {
                 auto prepared = get_statement(query_string, client_state, d);
+                prepared->calculate_metadata_id();
                 auto bound_terms = prepared->statement->get_bound_terms();
                 if (bound_terms > std::numeric_limits<uint16_t>::max()) {
                     throw exceptions::invalid_request_exception(

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>;
 

cql3/statements/prepared_statement.hh

@@ -50,7 +50,10 @@ public:
     const std::vector<seastar::lw_shared_ptr<column_specification>> bound_names;
     const std::vector<uint16_t> partition_key_bind_indices;
     std::vector<sstring> warnings;
+private:
+    cql_metadata_id_type _metadata_id;
 
+public:
     prepared_statement(audit::audit_info_ptr&& audit_info, seastar::shared_ptr<cql_statement> statement_, std::vector<seastar::lw_shared_ptr<column_specification>> bound_names_,
                        std::vector<uint16_t> partition_key_bind_indices, std::vector<sstring> warnings = {});
 
@@ -64,6 +67,10 @@ public:
     checked_weak_ptr checked_weak_from_this() const {
         return checked_weak_ptr(this->weak_from_this());
     }
+
+    void calculate_metadata_id();
+
+    cql_metadata_id_type get_metadata_id() const;
 };
 
 }

cql3/statements/raw/parsed_statement.cc

@@ -14,6 +14,7 @@
 #include "cql3/column_specification.hh"
 
 #include "cql3/cql_statement.hh"
+#include "cql3/result_set.hh"
 
 namespace cql3 {
 
@@ -47,6 +48,7 @@ prepared_statement::prepared_statement(
     , bound_names(std::move(bound_names_))
     , partition_key_bind_indices(std::move(partition_key_bind_indices))
     , warnings(std::move(warnings))
+    , _metadata_id(bytes{})
 {
     statement->set_audit_info(std::move(audit_info));
 }
@@ -66,6 +68,14 @@ prepared_statement::prepared_statement(audit::audit_info_ptr&& audit_info, ::sha
     : prepared_statement(std::move(audit_info), statement_, std::vector<lw_shared_ptr<column_specification>>(), std::vector<uint16_t>(), std::move(warnings))
 { }
 
+void prepared_statement::calculate_metadata_id() {
+    _metadata_id = statement->get_result_metadata()->calculate_metadata_id();
+}
+
+cql_metadata_id_type prepared_statement::get_metadata_id() const {
+    return _metadata_id;
+}
+
 }
 
 }

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/messages/result_message.hh

@@ -48,6 +48,10 @@ public:
         return _result_metadata;
     }
 
+    cql3::cql_metadata_id_type get_metadata_id() const {
+        return _prepared->get_metadata_id();
+    }
+
     class cql;
 private:
     static ::shared_ptr<const cql3::metadata> extract_result_metadata(::shared_ptr<cql3::cql_statement> statement);

transport/response.hh

@@ -76,7 +76,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& request_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

@@ -175,6 +175,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") {
@@ -1009,7 +1015,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>,
@@ -1104,7 +1110,8 @@ 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)));
+
+            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, q_state->query_state.get_trace_state(), version, cql_metadata_id_wrapper{}, skip_metadata)));
         }
     });
 }
@@ -1127,11 +1134,14 @@ 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) {
+        return _server._query_processor.local().prepare(std::move(query), client_state, dialect).then([this, stream, &client_state, trace_state] (auto msg) {
             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);
+            cql_metadata_id_wrapper metadata_id = is_metadata_id_supported(client_state)
+                ? cql_metadata_id_wrapper(msg->get_metadata_id())
+                : cql_metadata_id_wrapper();
+            return make_result(stream, *msg, trace_state, _version, std::move(metadata_id));
         });
     });
 }
@@ -1157,6 +1167,10 @@ process_execute_internal(service::client_state& client_state, distributed<cql3::
         throw exceptions::prepared_query_not_found_exception(id);
     }
 
+    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()), prepared->get_metadata_id())
+        : cql_metadata_id_wrapper();
+
     auto q_state = std::make_unique<cql_query_state>(client_state, trace_state, std::move(permit));
     auto& query_state = q_state->query_state;
     q_state->options = in.read_options(version, qp.local().get_cql_config());
@@ -1195,14 +1209,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)));
         }
     });
 }
@@ -1323,7 +1337,8 @@ 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)));
+
+            return cql_server::process_fn_return_type(make_foreign(make_result(stream, *msg, trace_state, version, cql_metadata_id_wrapper{})));
         }
     });
 }
@@ -1538,11 +1553,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 {
@@ -1557,8 +1574,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.has_response_metadata_id()) {
+            _response.write_short_bytes(_metadata_id.get_response_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 {
@@ -1578,7 +1598,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 {
@@ -1606,7 +1626,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);
@@ -1616,7 +1636,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;
 }
@@ -2024,7 +2044,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>();
@@ -2033,6 +2053,15 @@ 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.has_request_metadata_id() && metadata_id.has_response_metadata_id()) {
+        if (metadata_id.get_request_metadata_id() != metadata_id.get_response_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());
 
@@ -2044,6 +2073,10 @@ void cql_server::response::write(const cql3::metadata& m, bool no_metadata) {
         return;
     }
 
+    if (flags.contains<cql3::metadata::flag::METADATA_CHANGED>()) {
+        write_short_bytes(metadata_id.get_response_metadata_id()._metadata_id);
+    }
+
     auto names_i = m.get_names().begin();
 
     if (global_tables_spec) {
@@ -2096,6 +2129,28 @@ void cql_server::response::write(const cql3::prepared_metadata& m, uint8_t versi
     }
 }
 
+bool cql_metadata_id_wrapper::has_request_metadata_id() const {
+    return _request_metadata_id.has_value();
+}
+
+bool cql_metadata_id_wrapper::has_response_metadata_id() const {
+    return _response_metadata_id.has_value();
+}
+
+const cql3::cql_metadata_id_type& cql_metadata_id_wrapper::get_request_metadata_id() const {
+    if (!has_request_metadata_id()) {
+        on_internal_error(clogger, "request metadata_id is empty");
+    }
+    return _request_metadata_id.value();
+}
+
+const cql3::cql_metadata_id_type& cql_metadata_id_wrapper::get_response_metadata_id() const {
+    if (!has_response_metadata_id()) {
+        on_internal_error(clogger, "response metadata_id is empty");
+    }
+    return _response_metadata_id.value();
+}
+
 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) {