MDEV-9947: COM_MULTI united response

libmysqld/lib_sql.cc

@@ -1170,7 +1170,8 @@ bool Protocol_binary::write()
 bool
 net_send_ok(THD *thd,
             uint server_status, uint statement_warn_count,
-            ulonglong affected_rows, ulonglong id, const char *message)
+            ulonglong affected_rows, ulonglong id, const char *message,
+            bool unused __attribute__((unused)))
 {
   DBUG_ENTER("emb_net_send_ok");
   MYSQL_DATA *data;

sql/log_event.cc

@@ -4439,7 +4439,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
 
         thd->enable_slow_log= thd->variables.sql_log_slow;
         mysql_parse(thd, thd->query(), thd->query_length(), &parser_state,
-                    FALSE);
+                    FALSE, FALSE);
         /* Finalize server status flags after executing a statement. */
         thd->update_server_status();
         log_slow_statement(thd);

sql/protocol.cc

@@ -35,7 +35,7 @@ static const unsigned int PACKET_BUFFER_EXTRA_ALLOC= 1024;
 /* Declared non-static only because of the embedded library. */
 bool net_send_error_packet(THD *, uint, const char *, const char *);
 /* Declared non-static only because of the embedded library. */
-bool net_send_ok(THD *, uint, uint, ulonglong, ulonglong, const char *);
+bool net_send_ok(THD *, uint, uint, ulonglong, ulonglong, const char *, bool);
 /* Declared non-static only because of the embedded library. */
 bool net_send_eof(THD *thd, uint server_status, uint statement_warn_count);
 #ifndef EMBEDDED_LIBRARY
@@ -208,7 +208,8 @@ bool net_send_error(THD *thd, uint sql_errno, const char *err,
 bool
 net_send_ok(THD *thd,
             uint server_status, uint statement_warn_count,
-            ulonglong affected_rows, ulonglong id, const char *message)
+            ulonglong affected_rows, ulonglong id, const char *message,
+            bool skip_flush)
 {
   NET *net= &thd->net;
   uchar buff[MYSQL_ERRMSG_SIZE+10],*pos;
@@ -250,7 +251,7 @@ net_send_ok(THD *thd,
   if (message && message[0])
     pos= net_store_data(pos, (uchar*) message, strlen(message));
   error= my_net_write(net, buff, (size_t) (pos-buff));
-  if (!error)
+  if (!error && !skip_flush)
     error= net_flush(net);
 
 
@@ -514,14 +515,16 @@ void Protocol::end_statement()
                    thd->get_stmt_da()->statement_warn_count(),
                    thd->get_stmt_da()->affected_rows(),
                    thd->get_stmt_da()->last_insert_id(),
-                   thd->get_stmt_da()->message());
+                   thd->get_stmt_da()->message(),
+                   thd->get_stmt_da()->skip_flush());
     break;
   case Diagnostics_area::DA_DISABLED:
     break;
   case Diagnostics_area::DA_EMPTY:
   default:
     DBUG_ASSERT(0);
-    error= send_ok(thd->server_status, 0, 0, 0, NULL);
+    error= send_ok(thd->server_status, 0, 0, 0, NULL,
+                   thd->get_stmt_da()->skip_flush());
     break;
   }
   if (!error)
@@ -540,12 +543,12 @@ void Protocol::end_statement()
 
 bool Protocol::send_ok(uint server_status, uint statement_warn_count,
                        ulonglong affected_rows, ulonglong last_insert_id,
-                       const char *message)
+                       const char *message, bool skip_flush)
 {
   DBUG_ENTER("Protocol::send_ok");
   const bool retval= 
     net_send_ok(thd, server_status, statement_warn_count,
-                affected_rows, last_insert_id, message);
+                affected_rows, last_insert_id, message, skip_flush);
   DBUG_RETURN(retval);
 }
 

sql/protocol.h

@@ -62,7 +62,7 @@ class Protocol
 
   virtual bool send_ok(uint server_status, uint statement_warn_count,
                        ulonglong affected_rows, ulonglong last_insert_id,
-                       const char *message);
+                       const char *message, bool skip_flush);
 
   virtual bool send_eof(uint server_status, uint statement_warn_count);
 

sql/sql_error.cc

@@ -347,6 +347,7 @@ void
 Diagnostics_area::reset_diagnostics_area()
 {
   DBUG_ENTER("reset_diagnostics_area");
+  m_skip_flush= FALSE;
 #ifdef DBUG_OFF
   m_can_overwrite_status= FALSE;
   /** Don't take chances in production */

sql/sql_error.h

@@ -704,6 +704,12 @@ class Diagnostics_area
   const char *message() const
   { DBUG_ASSERT(m_status == DA_ERROR || m_status == DA_OK); return m_message; }
 
+  bool skip_flush() const
+  { DBUG_ASSERT(m_status == DA_OK); return m_skip_flush; }
+
+  void set_skip_flush()
+  { m_skip_flush= TRUE; }
+
   uint sql_errno() const
   { DBUG_ASSERT(m_status == DA_ERROR); return m_sql_errno; }
 
@@ -857,6 +863,9 @@ class Diagnostics_area
   /** Set to make set_error_status after set_{ok,eof}_status possible. */
   bool m_can_overwrite_status;
 
+  /** Skip flushing network buffer after writing OK (for COM_MULTI) */
+  bool m_skip_flush;
+
   /** Message buffer. Can be used by OK or ERROR status. */
   char m_message[MYSQL_ERRMSG_SIZE];
 

sql/sql_parse.cc

@@ -111,7 +111,9 @@
 #include "wsrep_thd.h"
 
 static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
-                              Parser_state *parser_state, bool is_next_command);
+                              Parser_state *parser_state,
+                              bool is_com_multi,
+                              bool is_next_command);
 
 /**
   @defgroup Runtime_Environment Runtime Environment
@@ -1020,7 +1022,7 @@ static void handle_bootstrap_impl(THD *thd)
       break;
     }
 
-    mysql_parse(thd, thd->query(), length, &parser_state, FALSE);
+    mysql_parse(thd, thd->query(), length, &parser_state, FALSE, FALSE);
 
     bootstrap_error= thd->is_error();
     thd->protocol->end_statement();
@@ -1633,6 +1635,8 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     drop_more_results= !MY_TEST(thd->server_status &
                                 SERVER_MORE_RESULTS_EXISTS);
     thd->server_status|= SERVER_MORE_RESULTS_EXISTS;
+    if (is_com_multi)
+      thd->get_stmt_da()->set_skip_flush();
   }
 
   switch (command) {
@@ -1784,10 +1788,10 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 
     if (WSREP_ON)
       wsrep_mysql_parse(thd, thd->query(), thd->query_length(), &parser_state,
-                        is_next_command);
+                        is_com_multi, is_next_command);
     else
       mysql_parse(thd, thd->query(), thd->query_length(), &parser_state,
-                  is_next_command);
+                  is_com_multi, is_next_command);
 
     while (!thd->killed && (parser_state.m_lip.found_semicolon != NULL) &&
            ! thd->is_error())
@@ -1873,10 +1877,10 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 
       if (WSREP_ON)
         wsrep_mysql_parse(thd, beginning_of_next_stmt, length, &parser_state,
-                          is_next_command);
+                          is_com_multi, is_next_command);
       else
         mysql_parse(thd, beginning_of_next_stmt, length, &parser_state,
-                    is_next_command);
+                    is_com_multi, is_next_command);
 
     }
 
@@ -1930,7 +1934,12 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     thd->reset_for_next_command();
     // thd->reset_for_next_command reset state => restore it
     if (is_next_command)
+    {
       thd->server_status|= SERVER_MORE_RESULTS_EXISTS;
+      if (is_com_multi)
+        thd->get_stmt_da()->set_skip_flush();
+    }
+
     lex_start(thd);
     /* Must be before we init the table list. */
     if (lower_case_table_names)
@@ -2272,6 +2281,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       thd->m_digest= save_digest;
 
       /* release old buffer */
+      net_flush(net);
       DBUG_ASSERT(net->buff == net->write_pos); // nothing to send
       my_free(readbuff);
     }
@@ -7508,7 +7518,9 @@ void mysql_init_multi_delete(LEX *lex)
 }
 
 static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
-                              Parser_state *parser_state, bool is_next_command)
+                              Parser_state *parser_state,
+                              bool is_com_multi,
+                              bool is_next_command)
 {
 #ifdef WITH_WSREP
   bool is_autocommit=
@@ -7527,7 +7539,8 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
       MYSQL_SET_STATEMENT_TEXT(thd->m_statement_psi, thd->query(),
 	                       thd->query_length());
     }
-    mysql_parse(thd, rawbuf, length, parser_state, is_next_command);
+    mysql_parse(thd, rawbuf, length, parser_state, is_com_multi,
+                is_next_command);
 
     if (WSREP(thd)) {
       /* wsrep BF abort in query exec phase */
@@ -7629,7 +7642,9 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
 */
 
 void mysql_parse(THD *thd, char *rawbuf, uint length,
-                 Parser_state *parser_state, bool is_next_command)
+                 Parser_state *parser_state,
+                 bool is_com_multi,
+                 bool is_next_command)
 {
   int error __attribute__((unused));
   DBUG_ENTER("mysql_parse");
@@ -7654,7 +7669,11 @@ void mysql_parse(THD *thd, char *rawbuf, uint length,
   lex_start(thd);
   thd->reset_for_next_command();
   if (is_next_command)
+  {
     thd->server_status|= SERVER_MORE_RESULTS_EXISTS;
+    if (is_com_multi)
+      thd->get_stmt_da()->set_skip_flush();
+  }
 
   if (query_cache_send_result_to_client(thd, rawbuf, length) <= 0)
   {

sql/sql_parse.h

@@ -88,7 +88,8 @@ bool is_log_table_write_query(enum enum_sql_command command);
 bool alloc_query(THD *thd, const char *packet, uint packet_length);
 void mysql_init_select(LEX *lex);
 void mysql_parse(THD *thd, char *rawbuf, uint length,
-                 Parser_state *parser_state, bool is_com_multi);
+                 Parser_state *parser_state, bool is_com_multi,
+                 bool is_next_command);
 bool mysql_new_select(LEX *lex, bool move_down);
 void create_select_for_variable(const char *var_name);
 void create_table_set_open_action_and_adjust_tables(LEX *lex);

sql/sql_prepare.cc

@@ -277,7 +277,7 @@ class Protocol_local :public Protocol
 
   virtual bool send_ok(uint server_status, uint statement_warn_count,
                        ulonglong affected_rows, ulonglong last_insert_id,
-                       const char *message);
+                       const char *message, bool skip_flush);
 
   virtual bool send_eof(uint server_status, uint statement_warn_count);
   virtual bool send_error(uint sql_errno, const char *err_msg, const char* sqlstate);
@@ -4877,7 +4877,7 @@ bool Protocol_local::send_out_parameters(List<Item_param> *sp_params)
 bool
 Protocol_local::send_ok(uint server_status, uint statement_warn_count,
                         ulonglong affected_rows, ulonglong last_insert_id,
-                        const char *message)
+                        const char *message, bool skip_flush)
 {
   /*
     Just make sure nothing is sent to the client, we have grabbed

sql/wsrep_sst.cc

@@ -957,7 +957,7 @@ static int run_sql_command(THD *thd, const char *query)
     return -1;
   }
 
-  mysql_parse(thd, thd->query(), thd->query_length(), &ps, FALSE);
+  mysql_parse(thd, thd->query(), thd->query_length(), &ps, FALSE, FALSE);
   if (thd->is_error())
   {
     int const err= thd->get_stmt_da()->sql_errno();