Bug#35115810 stmt-execute with new-params-bound = 0 [2/2]

The COM_STMT_EXECUTE message is handled by the StmtExecute Codec
which currently doesn't support messages with "new-params-bound = 0" which
is used when a prepared statement is executed twice, with the same
data-types, but possibly different values.

In the new-param-bound = 0 case, the types of the first execute of a
statement must be remembered.

Change
------

- handle new-param-bound = 0 properly

Change-Id: Ic71a9fd00b0f56d8cfc2d707873c164fed7495fd

Author: weigon

router/src/mysql_protocol/include/mysqlrouter/classic_protocol_codec_base.h

@@ -110,6 +110,7 @@ stdx::expected<size_t, std::error_code> encode(const T &v,
  *
  * @param buffer buffer to read from
  * @param caps protocol capabilities
+ * @tparam T the message class
  * @returns number of bytes read from 'buffers' and a T on success, or
  * std::error_code on error
  */
@@ -119,6 +120,28 @@ stdx::expected<std::pair<size_t, T>, std::error_code> decode(
   return Codec<T>::decode(buffer, caps);
 }
 
+/**
+ * decode a message from a buffer.
+ *
+ * @param buffer buffer to read from
+ * @param caps protocol capabilities
+ * @param args arguments that shall be forwarded to T's decode()
+ * @tparam T the message class
+ * @tparam Args Types of the extra arguments to be forwarded to T's decode()
+ * function.
+ * @returns number of bytes read from 'buffers' and a T on success, or
+ * std::error_code on error
+ */
+template <class T, class... Args>
+stdx::expected<std::pair<size_t, T>, std::error_code> decode(
+    const net::const_buffer &buffer, capabilities::value_type caps,
+    // clang-format off
+    Args &&... args
+    // clang-format on
+) {
+  return Codec<T>::decode(buffer, caps, std::forward<Args>(args)...);
+}
+
 namespace impl {
 
 /**

router/src/mysql_protocol/include/mysqlrouter/classic_protocol_codec_message.h

@@ -2111,17 +2111,13 @@ class Codec<borrowable::message::client::StmtExecute<Borrowed>>
     auto new_params_bound_res = accu.template step<bw::FixedInt<1>>();
     if (!accu.result()) return stdx::make_unexpected(accu.result().error());
 
-    auto new_params_bound = new_params_bound_res->value();
-    if (new_params_bound != 1) {
-      // new-params-bound is required as long as the decoder doesn't know the
-      // old param-defs
-      return stdx::make_unexpected(make_error_code(codec_errc::invalid_input));
-    }
-
     std::vector<typename value_type::ParamDef> types;
     std::vector<std::optional<typename value_type::string_type>> values;
 
-    if (new_params_bound == 1) {
+    auto new_params_bound = new_params_bound_res->value();
+    if (new_params_bound == 0) {
+      types = *metadata_res;
+    } else if (new_params_bound == 1) {
       types.reserve(param_count);
 
       for (size_t n{}; n < param_count; ++n) {
@@ -2130,6 +2126,8 @@ class Codec<borrowable::message::client::StmtExecute<Borrowed>>
 
         types.push_back(type_res->value());
       }
+    } else {
+      return stdx::make_unexpected(make_error_code(codec_errc::invalid_input));
     }
 
     const auto nullbits = nullbits_res->value();

router/src/routing/src/classic_frame.h

@@ -108,4 +108,56 @@ class ClassicFrame {
   }
 };
 
+/**
+ * receive a StmtExecute message from a channel.
+ *
+ * specialization of recv_msg<> as StmtExecute needs a the data from the
+ * StmtPrepareOk.
+ */
+template <>
+inline stdx::expected<classic_protocol::borrowed::message::client::StmtExecute,
+                      std::error_code>
+ClassicFrame::recv_msg<
+    classic_protocol::borrowed::message::client::StmtExecute>(
+    Channel *src_channel, ClassicProtocolState *src_protocol,
+    classic_protocol::capabilities::value_type caps) {
+  using msg_type = classic_protocol::borrowed::message::client::StmtExecute;
+
+  auto read_res = ClassicFrame::recv_frame_sequence(src_channel, src_protocol);
+  if (!read_res) return stdx::make_unexpected(read_res.error());
+
+  const auto &recv_buf = src_channel->recv_plain_view();
+
+  auto frame_decode_res = classic_protocol::decode<
+      classic_protocol::frame::Frame<classic_protocol::borrowed::wire::String>>(
+      net::buffer(recv_buf), caps);
+  if (!frame_decode_res) {
+    return stdx::make_unexpected(frame_decode_res.error());
+  }
+
+  src_protocol->seq_id(frame_decode_res->second.seq_id());
+
+  auto decode_res = classic_protocol::decode<msg_type>(
+      net::buffer(frame_decode_res->second.payload().value()), caps,
+      [src_protocol](auto stmt_id)
+          -> stdx::expected<std::vector<msg_type::ParamDef>, std::error_code> {
+        const auto it = src_protocol->prepared_statements().find(stmt_id);
+        if (it == src_protocol->prepared_statements().end()) {
+          return stdx::make_unexpected(make_error_code(
+              classic_protocol::codec_errc::statement_id_not_found));
+        }
+
+        std::vector<msg_type::ParamDef> params;
+        params.reserve(it->second.parameters.size());
+
+        for (const auto &param : it->second.parameters) {
+          params.emplace_back(param.type_and_flags);
+        }
+
+        return params;
+      });
+
+  return decode_res->second;
+}
+
 #endif

router/src/routing/src/classic_stmt_execute_forwarder.cc

@@ -26,6 +26,7 @@
 
 #include "classic_connection_base.h"
 #include "classic_frame.h"
+#include "hexify.h"
 #include "mysql/harness/stdx/expected.h"
 #include "mysql/harness/tls_error.h"
 #include "mysqld_error.h"  // mysql-server error-codes
@@ -61,7 +62,26 @@ StmtExecuteForwarder::process() {
 stdx::expected<Processor::Result, std::error_code>
 StmtExecuteForwarder::command() {
   if (auto &tr = tracer()) {
-    tr.trace(Tracer::Event().stage("stmt_execute::command"));
+    auto *socket_splicer = connection()->socket_splicer();
+    auto *src_channel = socket_splicer->client_channel();
+    auto *src_protocol = connection()->client_protocol();
+
+    auto msg_res = ClassicFrame::recv_msg<
+        classic_protocol::borrowed::message::client::StmtExecute>(src_channel,
+                                                                  src_protocol);
+    if (!msg_res) return recv_client_failed(msg_res.error());
+
+    const auto &recv_buf = src_channel->recv_plain_view();
+
+    tr.trace(Tracer::Event().stage(
+        "stmt_execute::command:\nstmt-id: " +              //
+        std::to_string(msg_res->statement_id()) + "\n" +   //
+        "flags: " + msg_res->flags().to_string() + "\n" +  //
+        "new-params-bound: " + std::to_string(msg_res->new_params_bound()) +
+        "\n" +  //
+        "types::size(): " + std::to_string(msg_res->types().size()) + "\n" +
+        "values::size(): " + std::to_string(msg_res->values().size()) + "\n" +
+        mysql_harness::hexify(recv_buf)));
   }
 
   auto &server_conn = connection()->socket_splicer()->server_conn();

router/tests/integration/test_routing_direct.cc

@@ -2161,6 +2161,17 @@ TEST_P(ConnectionTest, classic_protocol_prepare_execute) {
   };
   ASSERT_NO_ERROR(stmt.bind_params(params));
 
+  // execute again to trigger a StmtExecute with new-params-bound = 1.
+  {
+    auto exec_res = stmt.execute();
+    ASSERT_NO_ERROR(exec_res);
+
+    for ([[maybe_unused]] auto res : *exec_res) {
+      // drain the resultsets.
+    }
+  }
+
+  // execute again to trigger a StmtExecute with new-params-bound = 0.
   {
     auto exec_res = stmt.execute();
     ASSERT_NO_ERROR(exec_res);
@@ -2174,7 +2185,7 @@ TEST_P(ConnectionTest, classic_protocol_prepare_execute) {
     auto events_res = changed_event_counters(cli);
     ASSERT_NO_ERROR(events_res);
 
-    EXPECT_THAT(*events_res, ElementsAre(Pair("statement/com/Execute", 1),
+    EXPECT_THAT(*events_res, ElementsAre(Pair("statement/com/Execute", 2),
                                          Pair("statement/com/Prepare", 1)));
   }
 
@@ -2186,7 +2197,7 @@ TEST_P(ConnectionTest, classic_protocol_prepare_execute) {
     ASSERT_NO_ERROR(events_res);
 
     EXPECT_THAT(*events_res,
-                ElementsAre(Pair("statement/com/Execute", 1),
+                ElementsAre(Pair("statement/com/Execute", 2),
                             Pair("statement/com/Prepare", 1),
                             // explicit
                             Pair("statement/com/Reset Connection", 1),