MDEV-16670 Improve execute direct (prepare and execute) performance

The patch detects if the processed PREPARE command is first part of the
pipelined DIRECT+EXECUTE commands with stmt id -1 for execution. For
that the vio class has been extended with peek method that lets us
verify that the server has already received next command, it is EXECUTE
with stmt id -1, and the packet is minimally complete.

If it is so, the execution is forwarded to the fused prepare+execute path
as if it was a single command. It lets a statement's execute leg run before
any real prepare/check_prepared_statement() has happened and the prepare
leg does basically query parsing only. Several places
downstream assumed that had already happened, so give Query_arena a single
predicate, is_direct_exec_unresolved() - true only for the one-shot fused
pass that never went through a real prepare - and fix each broken
assumption against it:

* Query cache: a cached response is replayed without ever running
  mysql_execute_command(), so the buffered prepare-response hook never
  fires; conversely, populating the cache captures that hook's bytes too
  (net_real_write() sees the whole flushed buffer), poisoning the entry
  for unrelated later clients. Direct-exec statements now opt out of the
  query cache entirely (query_cache_is_applicable /
  lex->safe_to_cache_query, saved and restored around just that one
  execution). Covered by test_mdev_16670_qc_hit_hang and
  test_mdev_16670_qc_poison.

* Reprepare: a mid-execute reprepare() (e.g. ER_NEED_REPREPARE) swaps in
  a lex that DID go through a real prepare, so the statement is no
  longer "unresolved" once that happens - tracked via a new
  IS_DIRECT_EXEC_REPREPARED flag so the direct-exec special-casing only
  applies to the original pass.

* System versioning / periods: skip_setup_conds() treated the fused pass
  as an ordinary re-execution rather than a first resolution, so
  vers_setup_conds()/period_setup_conds() never ran - a versioned table
  or FOR PORTION OF statement used the wrong predicate. Now counted as
  "first execute" via is_direct_exec_unresolved(). Covered by
  test_mdev_16670_versioned_rows and test_mdev_16670_trigger_versioned.

* Unsupported statements check used to be done as part of the
  check_prepared_statement, has been moved to the new
  check_stmt_command_supported function called from the
  Prepared_statement::prepare method. Covered by
  test_mdev_16670_unsupported_ps.

* CALL statements and queries referencing a stored function are excluded
  from the optimized path (skip_direct_exec_optimization()): their nested
  sub-statements run with thd->cur_stmt still pointing at the outer
  statement, which would otherwise misfire the buffered prepare-response
  hook. Also, the stored function use would cause guaranteed re-prepare.
  The flag is reset for them after the query parsing. Covered by
  test_mdev_16670_demoted_function_call.

Also adds regression tests for the reply-pair on parse error
(test_mdev_16670_parse_error_reply_pair), SQLSTATE/errno preservation
through the execute-leg fallback (test_mdev_16670_fallback_sqlstate),
general_log coverage (test_mdev_16670_general_log), a statement killed
mid-fallback (test_mdev_16670_kill_during_fallback), and prebind usage of
mariadb_stmt_execute_direct() with real bound parameters
(test_mdev_16670_prebind_direct).

It's possible to compile with global variable opt_exec_direct that controls
if the optimization should be used. Its existence can be switched on
using cmake option WITH_DIRECT_EXEC_OPTIMIZE_SWITCH. By default it does
not exist, as it is believed to be unneeded in the general case.

create_and_prepare_stmt() helper added to perform common statement creation
routine, and erase_stmt_safely() helper - for cleaning in
create_and_prepare_stmt()'s prepare-failure cleanup, direct_exec_prepare_ok_fallback()'s
OOM branch, and mysqld_stmt_direct_execute()'s packet_error branch.

Assisted-by: Claude Sonnet 5

CMakeLists.txt

@@ -179,6 +179,14 @@ OPTION(ENABLED_JSON_WRITER_CONSISTENCY_CHECKS "Enable Json_writer_object / Json_
 IF(ENABLED_JSON_WRITER_CONSISTENCY_CHECKS)
   ADD_DEFINITIONS(-DENABLED_JSON_WRITER_CONSISTENCY_CHECKS)
 ENDIF()
+# ADD_DEFINITIONS, matching ENABLED_JSON_WRITER_CONSISTENCY_CHECKS above, rather
+# than #cmakedefine: this flag only compiles in the opt_exec_direct sysvar
+# itself (direct execution stays on either way), so no generated header needs
+# to carry it.
+OPTION(WITH_DIRECT_EXEC_OPTIMIZE_SWITCH "Compile in the opt_exec_direct sysvar for toggling pipelined direct-execution at runtime (direct execution itself is always on)" OFF)
+IF(WITH_DIRECT_EXEC_OPTIMIZE_SWITCH)
+  ADD_DEFINITIONS(-DWITH_DIRECT_EXEC_OPTIMIZE_SWITCH)
+ENDIF()
 OPTION(WITHOUT_SERVER "Build only the client library and clients" OFF)
 IF(UNIX)
   OPTION(WITH_VALGRIND "Valgrind instrumentation" OFF)

include/violite.h

@@ -281,5 +281,6 @@ struct st_vio
   int shutdown_flag;
   void *tp_ctx; /* threadpool context */
 #endif
+  size_t (*peek)(Vio *, uchar *, size_t);
 };
 #endif /* vio_violite_h_ */

sql/protocol.cc

@@ -1158,6 +1158,23 @@ static bool should_send_column_info(THD* thd, List<Item>* list, uint flags)
   if (flags & Protocol::SEND_FORCE_COLUMN_INFO)
     return true;
 
+  /*
+    This call is the direct-exec execute leg's own result-set metadata send.
+    It always comes after send_pipelined_prepare_response() has already run
+    for this same command (sql_prepare.cc), forcing its own, separate call
+    to this function with SEND_FORCE_COLUMN_INFO for the identical list -
+    so full column info already went out moments ago, in this same command.
+    Skip unconditionally rather than running metadata_columns_changed():
+    that call's own column_info_state was never touched by the forced send
+    above (it returned true via the SEND_FORCE_COLUMN_INFO check, never
+    reaching that state), so calling it now would see !initialized and
+    answer "changed" - the wrong answer, not just a redundant computation.
+  */
+  if (thd->cur_stmt->is_direct_exec_unresolved())
+  {
+    thd->status_var.skip_metadata_count++;
+    return false;
+  }
   auto &column_info_state= thd->cur_stmt->column_info_state;
 #ifndef DBUG_OFF
   auto cmd= thd->get_command();
@@ -1199,6 +1216,18 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
 {
   DBUG_ENTER("Protocol::send_result_set_metadata");
 
+  /*
+    Pipelined direct execution: emit the (buffered) prepare response before
+    this statement's own result-set metadata. This is the single choke point
+    through which every result-set-producing statement type sends its column
+    metadata (SELECT via select_send, but also SHOW/ANALYZE/CHECK/HELP/
+    HANDLER READ/XA RECOVER/... which call this directly), so hooking it here
+    covers all of them instead of one call site per statement type. No-op for
+    ordinary statements.
+  */
+  if (thd->cur_stmt && thd->cur_stmt->send_pipelined_prepare_response(list))
+    DBUG_RETURN(TRUE);
+
   bool send_column_info= should_send_column_info(thd, list, flags);
 
   if (flags & SEND_NUM_ROWS)

sql/sql_class.cc

@@ -3434,6 +3434,11 @@ bool select_send::send_result_set_metadata(List<Item> &list, uint flags)
     return FALSE;
   }
 #endif /* WITH_WSREP */
+  /*
+    Pipelined direct execution's prepare-response hook lives in
+    Protocol::send_result_set_metadata() itself (protocol.cc), not here - that
+    covers every statement type that sends a result set, not just SELECT.
+  */
   if (!(res= thd->protocol->send_result_set_metadata(&list, flags)))
     is_result_set_started= 1;
   return res;

sql/sql_class.h

@@ -1346,6 +1346,18 @@ public:
   virtual Type type() const;
   virtual ~Query_arena() = default;
 
+  /*
+    True only before a direct-exec statement's fused execute has gone
+    through any real prepare. See
+    Prepared_statement::is_direct_exec_unresolved().
+
+    Declared here, not on Statement, so sql_select.cc/table.cc can read it
+    through thd->stmt_arena without seeing Prepared_statement (defined only
+    in sql_prepare.cc). A nested SP/trigger instruction repoints stmt_arena
+    to its own arena, which correctly inherits this default (false).
+  */
+  virtual bool is_direct_exec_unresolved() const { return false; }
+
   inline bool is_stmt_prepare() const { return state == STMT_INITIALIZED; }
   inline bool is_stmt_prepare_or_first_sp_execute() const
   { return (int)state < (int)STMT_PREPARED; }
@@ -1972,6 +1984,17 @@ public:
   trg_event_type current_trg_event();
   bool push_current_trg_event(trg_event_type trg_event);
   void pop_current_trg_event();
+  /*
+    For pipelined direct execution: emit the (buffered, un-flushed) prepare
+    response just before the execute leg sends its first reply - its own
+    result-set metadata, a LOAD DATA LOCAL file request, or (for a statement
+    without a result set) its OK. Called with the resolved result columns
+    (NULL/empty for a 0-column statement). Returns true on network error.
+    No-op (returns false) for statements that are not a pipelined direct
+    execution, and on every call after the first one.
+  */
+  virtual bool send_pipelined_prepare_response(List<Item> *fields)
+  { return false; }
 };
 
 

sql/sql_load.cc

@@ -546,6 +546,15 @@ int mysql_load(THD *thd, const sql_exchange *ex, TABLE_LIST *table_list,
 #ifndef EMBEDDED_LIBRARY
   if (read_file_from_client)
   {
+    /*
+      Pipelined direct execution: the file-request packet is the first thing
+      the client sees from this statement, so the (buffered) prepare response
+      must go out before it. No-op for ordinary statements. Tables are already
+      open+locked and fields validated here, i.e. the prepare would have
+      succeeded.
+    */
+    if (thd->cur_stmt && thd->cur_stmt->send_pipelined_prepare_response(NULL))
+      DBUG_RETURN(TRUE);
     (void)net_request_file(&thd->net,ex->file_name);
     file = -1;
   }

sql/sql_parse.cc

@@ -117,6 +117,10 @@ static bool wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
   @{
 */
 
+#ifdef WITH_DIRECT_EXEC_OPTIMIZE_SWITCH
+my_bool opt_exec_direct= 1;
+#endif
+
 static bool execute_sqlcom_select(THD *thd, TABLE_LIST *all_tables);
 static void sql_kill(THD *thd, my_thread_id id, killed_state state, killed_type type);
 static void sql_kill_user(THD *thd, LEX_USER *user, killed_state state);
@@ -1846,7 +1850,41 @@ dispatch_command_return dispatch_command(enum enum_server_command command,
   }
   case COM_STMT_PREPARE:
   {
-    mysqld_stmt_prepare(thd, packet, packet_length);
+    /*
+      If a COM_STMT_EXECUTE for stmt_id -1 is already pipelined right behind
+      this prepare, fuse the two into a single command (parse, then execute
+      the just-parsed statement) so the exchange is flushed only once.
+    */
+#ifdef WITH_DIRECT_EXEC_OPTIMIZE_SWITCH
+    if (opt_exec_direct && direct_execution(thd->net.vio))
+#else
+    if (direct_execution(thd->net.vio))
+#endif
+    {
+      /*
+        log_slow_done/wsrep_after_statement() only if the execute leg ran:
+        Prepared_statement::execute() already logged/audited then, same as
+        the ordinary COM_STMT_EXECUTE case above. On an early failure (parse
+        error, read error, malformed packet) nothing executed, so neither
+        applies - same as an ordinary COM_STMT_PREPARE failure.
+      */
+      bool force_close= false;
+      bool executed= mysqld_stmt_direct_execute(thd, packet, packet_length,
+                                                &force_close);
+      log_slow_done= executed;
+      if (force_close)
+        error= true;
+#ifdef WITH_WSREP
+      if (executed && WSREP(thd))
+      {
+          (void)wsrep_after_statement(thd);
+      }
+#endif /* WITH_WSREP */
+    }
+    else
+    {
+      mysqld_stmt_prepare(thd, packet, packet_length);
+    }
     break;
   }
   case COM_STMT_CLOSE:

sql/sql_prepare.h

@@ -81,6 +81,16 @@ private:
 };
 
 
+struct st_vio;
+bool direct_execution(struct st_vio *vio);
+/*
+  Returns true if the execute leg ran (and so already audited/logged itself).
+  Sets *force_close to true if the connection must be closed once the
+  current reply has been sent (a desynced pipelined read) - the caller
+  should fold that into dispatch_command()'s own close-connection decision.
+*/
+bool mysqld_stmt_direct_execute(THD *thd, const char *packet,
+                                uint packet_length, bool *force_close);
 void mysqld_stmt_prepare(THD *thd, const char *packet, uint packet_length);
 void mysqld_stmt_execute(THD *thd, char *packet, uint packet_length);
 void mysqld_stmt_execute_bulk(THD *thd, char *packet, uint packet_length);

sql/sql_select.cc

@@ -1167,8 +1167,25 @@ Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
 static
 bool skip_setup_conds(THD *thd)
 {
-  return (!thd->stmt_arena->is_conventional()
-          && !thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
+  /*
+    A pipelined direct execution's fused execute is the only resolution
+    pass its statement will ever get without going through a real prepare
+    (see Query_arena::is_direct_exec_unresolved()) - unlike an ordinary
+    prepared statement, which resolves once during a separate PREPARE
+    (state < STMT_PREPARED) and replays the result on every execute. Count
+    that one pass as "first execute" too, so that e.g. period/versioning
+    conditions get built on it instead of being silently skipped forever.
+
+    Read through thd->stmt_arena, not thd->cur_stmt: a nested SP/trigger
+    instruction executing inside the fused statement's own execute leg
+    switches thd->stmt_arena to its own arena (thd->cur_stmt stays pointed
+    at the outer statement throughout), so its is_direct_exec_unresolved()
+    correctly reads as false via the Query_arena default - it is never the
+    fused statement's own one-shot resolution pass, however it was reached.
+  */
+  bool is_first_resolve= thd->stmt_arena->is_stmt_prepare_or_first_sp_execute() ||
+                         thd->stmt_arena->is_direct_exec_unresolved();
+  return (!thd->stmt_arena->is_conventional() && !is_first_resolve)
          || thd->lex->is_view_context_analysis();
 }
 

sql/sys_vars.cc

@@ -7801,3 +7801,12 @@ static Sys_var_path Sys_path(
         "path", "Comma-separated list of schema names that defines the search "
         "order for stored routines",
         SESSION_VAR(path), NO_CMD_LINE, NOT_IN_BINLOG);
+
+#ifdef WITH_DIRECT_EXEC_OPTIMIZE_SWITCH
+extern my_bool opt_exec_direct;
+
+static Sys_var_mybool Sys_optimize_direct_exec(
+    "opt_exec_direct", "Optimize direct prepared statement execution",
+    GLOBAL_VAR(opt_exec_direct), CMD_LINE(OPT_ARG), DEFAULT(TRUE),
+    NO_MUTEX_GUARD, NOT_IN_BINLOG);
+#endif

sql/table.cc

@@ -6349,7 +6349,8 @@ bool TABLE_LIST::create_field_translation(THD *thd)
                        get_unit()));
 
   if (thd->stmt_arena->is_conventional() ||
-      thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
+      thd->stmt_arena->is_stmt_prepare_or_first_sp_execute() ||
+      thd->stmt_arena->is_direct_exec_unresolved())
   {
     /* initialize lists */
     used_items.empty();

sql/threadpool_winsockets.cc

@@ -181,6 +181,35 @@ static size_t my_vio_read(st_vio *vio, uchar *dest, size_t sz)
   return sock->m_orig_vio_read(vio, dest, sz);
 }
 
+/*
+  Peek at incoming data without consuming it, thread-pool-buffer aware.
+
+  Without this override, a caller peeking the wire (e.g. direct_execution()
+  detecting a pipelined COM_STMT_EXECUTE) would only ever see the raw
+  socket, never bytes the one-shot AIO read already pulled into m_buf_ptr -
+  a deterministic miss for any payload small enough to fit in one such read.
+
+  Mirrors vio_peek()'s own two-source shape: serve from the AIO buffer
+  first, then fall through to m_orig_vio_peek() for anything beyond what's
+  already buffered. Never advances m_buf_off - a peek must not consume what
+  a later my_vio_read() still needs to see.
+*/
+static size_t my_vio_peek(st_vio *vio, uchar *dest, size_t sz)
+{
+  auto sock= (win_aiosocket *) vio->tp_ctx;
+  DBUG_ASSERT(sock);
+
+  auto buffered= std::min(sock->buffer_remaining(), sz);
+
+  if (buffered > 0)
+    memcpy(dest, sock->m_buf_ptr + sock->m_buf_off, buffered);
+
+  if (buffered == sz)
+    return buffered;
+
+  return buffered + sock->m_orig_vio_peek(vio, dest + buffered, sz - buffered);
+}
+
 DWORD win_aiosocket::begin_read()
 {
   DWORD err = ERROR_SUCCESS;
@@ -255,6 +284,18 @@ void win_aiosocket::init(Vio *vio)
 
   m_orig_vio_read= vio->read;
   vio->read= my_vio_read;
+
+  /*
+    Only override peek if this transport has one to fall back to (e.g. not
+    a named pipe) - otherwise my_vio_peek() would call a NULL
+    m_orig_vio_peek() once a caller asked for more than m_buf_ptr's
+    leftovers, turning "no peek support" into a crash.
+  */
+  if (vio->peek)
+  {
+    m_orig_vio_peek= vio->peek;
+    vio->peek= my_vio_peek;
+  }
 }
 
 void init_win_aio_buffers(unsigned int n_buffers)

sql/threadpool_winsockets.h

@@ -39,9 +39,10 @@ struct win_aiosocket
   size_t m_buf_datalen{};
 
   /*  Vio handling */
-  /** Pointer to original vio->vio_read/vio->has_data function */
+  /** Pointer to original vio->vio_read/vio->has_data/vio->peek function */
   size_t (*m_orig_vio_read)(st_vio *, unsigned char *, size_t){};
   char (*m_orig_vio_has_data)(st_vio *){};
+  size_t (*m_orig_vio_peek)(st_vio *, unsigned char *, size_t){};
 
 
 

vio/vio.c

@@ -137,6 +137,7 @@ static void vio_init(Vio *vio, enum enum_vio_type type,
     vio->viodelete	=vio_ssl_delete;
     vio->vioerrno	=vio_errno;
     vio->read		=vio_ssl_read;
+    vio->peek     =vio_ssl_peek;
     vio->write		=vio_ssl_write;
     vio->fastsend	=vio_fastsend;
     vio->viokeepalive	=vio_keepalive;
@@ -172,6 +173,7 @@ static void vio_init(Vio *vio, enum enum_vio_type type,
   vio->timeout          =vio_socket_timeout;
   vio->has_data         = ((flags & VIO_BUFFERED_READ) ?
                            vio_buff_has_data : has_no_data);
+  vio->peek             =vio_peek;
   DBUG_VOID_RETURN;
 }
 

vio/vio_priv.h

@@ -43,6 +43,7 @@ uint vio_pending_pipe(Vio* vio);
 
 int	vio_socket_shutdown(Vio *vio, int how);
 my_bool	vio_buff_has_data(Vio *vio);
+size_t	vio_peek(Vio *vio, uchar *buf, size_t size);
 int	vio_socket_io_wait(Vio *vio, enum enum_vio_io_event event);
 int	vio_socket_timeout(Vio *vio, uint which, my_bool old_mode);
 
@@ -50,6 +51,7 @@ int	vio_socket_timeout(Vio *vio, uint which, my_bool old_mode);
 #include "my_net.h"			/* needed because of struct in_addr */
 
 size_t	vio_ssl_read(Vio *vio,uchar* buf,	size_t size);
+size_t  vio_ssl_peek(Vio *vio, uchar *buf, size_t size);
 size_t	vio_ssl_write(Vio *vio,const uchar* buf, size_t size);
 
 /* When the workday is over... */

vio/viosocket.c

@@ -251,6 +251,56 @@ my_bool vio_buff_has_data(Vio *vio)
 }
 
 
+static my_bool socket_peek_read(Vio *vio, uint *bytes);
+
+/*
+  Peek at incoming data without consuming it.
+
+  Copies up to 'size' bytes that a subsequent read would return into 'buf',
+  taking data from the buffered-read buffer first and then from the socket
+  (via MSG_PEEK). This never blocks: if fewer than 'size' bytes are available
+  right now, only the available bytes are returned. The socket is left
+  untouched, so the following my_net_read() sees exactly the same stream.
+
+  @return number of bytes copied into 'buf' (may be less than 'size').
+*/
+
+size_t vio_peek(Vio *vio, uchar *buf, size_t size)
+{
+  size_t buffered= 0;
+  uint avail= 0;
+  ssize_t ret;
+  DBUG_ENTER("vio_peek");
+
+  /*
+    First, serve bytes still sitting in the buffered-read buffer. These are
+    the not-yet-consumed leftovers of an earlier vio_read(); the socket's
+    unread bytes logically follow them.
+  */
+  if (vio->read_pos < vio->read_end)
+  {
+    buffered= MY_MIN((size_t) (vio->read_end - vio->read_pos), size);
+    memcpy(buf, vio->read_pos, buffered);
+    if (buffered == size)
+      DBUG_RETURN(buffered);
+  }
+
+  /*
+    Peek the remainder from the socket without consuming or blocking. Only
+    request what is already available so recv(MSG_PEEK) cannot block.
+  */
+  if (socket_peek_read(vio, &avail) || avail == 0)
+    DBUG_RETURN(buffered);
+
+  ret= mysql_socket_recv(vio->mysql_socket, (SOCKBUF_T *) (buf + buffered),
+                         MY_MIN(size - buffered, (size_t) avail), MSG_PEEK);
+  if (ret <= 0)
+    DBUG_RETURN(buffered);
+
+  DBUG_RETURN(buffered + (size_t) ret);
+}
+
+
 size_t vio_write(Vio *vio, const uchar* buf, size_t size)
 {
   ssize_t ret;

vio/viossl.c

@@ -185,6 +185,34 @@ size_t vio_ssl_read(Vio *vio, uchar *buf, size_t size)
 
 }
 
+size_t vio_ssl_peek(Vio *vio, uchar *buf, size_t size)
+{
+  int ret;
+  SSL *ssl= vio->ssl_arg;
+  DBUG_ENTER("vio_ssl_peek");
+  DBUG_PRINT("enter", ("sd: %d  buf: %p  size: %zu  ssl: %p",
+                       (int) mysql_socket_getfd(vio->mysql_socket), buf, size,
+                       vio->ssl_arg));
+
+  /*
+    "Quick peek": only look at application data that is already decrypted and
+    waiting in the SSL layer's plaintext buffer (clearOutputBuffer). If fewer
+    than 'size' bytes are buffered, return 0 immediately instead of calling
+    SSL_peek(), which would read from - and block on, up to the read timeout -
+    the socket. This keeps pipelined-execute detection free for the common
+    case where nothing is waiting behind the prepare.
+
+    When at least 'size' decrypted bytes are pending, SSL_peek() returns them
+    without touching the socket, so it neither blocks nor needs a retry loop.
+  */
+  if ((size_t) SSL_pending(ssl) < size)
+    DBUG_RETURN(0);
+
+  ret= SSL_peek(ssl, buf, (int) size);
+
+  DBUG_PRINT("exit", ("%d", ret));
+  DBUG_RETURN(ret < 0 ? 0 : (size_t) ret);
+}
 
 size_t vio_ssl_write(Vio *vio, const uchar *buf, size_t size)
 {