MDEV-28055: Galera ps-protocol fixes

* Fix test galera.MW-44 to make it work with --ps-protocol
* Skip test galera.MW-328C under --ps-protocol This test
  relies on wsrep_retry_autocommit, which has no effect
  under ps-protocol.
* Return WSREP related errors on COM_STMT_PREPARE commands
  Change wsrep_command_no_result() to allow sending back errors
  when a statement is prepared. For example, to handle deadlock
  error due to BF aborted transaction during prepare.
* Add sync waiting before statement prepare
  When a statement is prepared, tables used in the statement may be
  opened and checked for existence. Because of that, some tests (for
  example galera_create_table_as_select) that CREATE a table in one node
  and then SELECT from the same table in another node may result in errors
  due to non existing table.
  To make tests behave similarly under normal and PS protocol, we add a
  call to sync wait before preparing statements that would sync wait
  during normal execution.
Reviewed-by: Jan Lindström <jan.lindstrom@mariadb.com>

mysql-test/suite/galera/t/MW-328C.test

@@ -7,6 +7,7 @@
 # masks all deadlock errors
 #
 
+--source include/no_protocol.inc
 --source include/galera_cluster.inc
 --source suite/galera/t/MW-328-header.inc
 

mysql-test/suite/galera/t/MW-44.test

@@ -19,7 +19,11 @@ SET SESSION wsrep_osu_method=RSU;
 ALTER TABLE t1 ADD COLUMN f2 INTEGER;
 SET SESSION wsrep_osu_method=TOI;
 
---let $wait_condition = SELECT COUNT(*) = 2 FROM mysql.general_log WHERE argument LIKE "CREATE%" OR argument LIKE "ALTER%"
+--let $wait_condition = SELECT COUNT(*) = 1 FROM mysql.general_log WHERE argument LIKE "CREATE%" AND command_type != 'Prepare'
+--let $wait_condition_on_error_output = SELECT * FROM mysql.general_log
+--source include/wait_condition_with_debug.inc
+
+--let $wait_condition = SELECT COUNT(*) = 1 FROM mysql.general_log WHERE argument LIKE "ALTER%" AND command_type != 'Prepare'
 --let $wait_condition_on_error_output = SELECT * FROM mysql.general_log
 --source include/wait_condition_with_debug.inc
 

sql/sql_parse.cc

@@ -1167,8 +1167,7 @@ static bool wsrep_tables_accessible_when_detached(const TABLE_LIST *tables)
 
 static bool wsrep_command_no_result(char command)
 {
-  return (command == COM_STMT_PREPARE          ||
-          command == COM_STMT_FETCH            ||
+  return (command == COM_STMT_FETCH            ||
           command == COM_STMT_SEND_LONG_DATA   ||
           command == COM_STMT_CLOSE);
 }

sql/sql_prepare.cc

@@ -2394,6 +2394,10 @@ static bool check_prepared_statement(Prepared_statement *stmt)
       goto error;
   }
 
+#ifdef WITH_WSREP
+    if (wsrep_sync_wait(thd, sql_command))
+      goto error;
+#endif
   switch (sql_command) {
   case SQLCOM_REPLACE:
   case SQLCOM_INSERT:

sql/wsrep_mysqld.cc

@@ -1176,6 +1176,73 @@ wsrep_sync_wait_upto (THD*          thd,
   return ret;
 }
 
+bool wsrep_is_show_query(enum enum_sql_command command)
+{
+  DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
+  return (sql_command_flags[command] & CF_STATUS_COMMAND) != 0;
+}
+
+static bool wsrep_is_diagnostic_query(enum enum_sql_command command)
+{
+  assert(command >= 0 && command <= SQLCOM_END);
+  return (sql_command_flags[command] & CF_DIAGNOSTIC_STMT) != 0;
+}
+
+static enum enum_wsrep_sync_wait
+wsrep_sync_wait_mask_for_command(enum enum_sql_command command)
+{
+  switch (command)
+  {
+  case SQLCOM_SELECT:
+  case SQLCOM_CHECKSUM:
+    return WSREP_SYNC_WAIT_BEFORE_READ;
+  case SQLCOM_DELETE:
+  case SQLCOM_DELETE_MULTI:
+  case SQLCOM_UPDATE:
+  case SQLCOM_UPDATE_MULTI:
+    return WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE;
+  case SQLCOM_REPLACE:
+  case SQLCOM_INSERT:
+  case SQLCOM_REPLACE_SELECT:
+  case SQLCOM_INSERT_SELECT:
+    return WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE;
+  default:
+    if (wsrep_is_diagnostic_query(command))
+    {
+      return WSREP_SYNC_WAIT_NONE;
+    }
+    if (wsrep_is_show_query(command))
+    {
+      switch (command)
+      {
+      case SQLCOM_SHOW_PROFILE:
+      case SQLCOM_SHOW_PROFILES:
+      case SQLCOM_SHOW_SLAVE_HOSTS:
+      case SQLCOM_SHOW_RELAYLOG_EVENTS:
+      case SQLCOM_SHOW_SLAVE_STAT:
+      case SQLCOM_SHOW_MASTER_STAT:
+      case SQLCOM_SHOW_ENGINE_STATUS:
+      case SQLCOM_SHOW_ENGINE_MUTEX:
+      case SQLCOM_SHOW_ENGINE_LOGS:
+      case SQLCOM_SHOW_PROCESSLIST:
+      case SQLCOM_SHOW_PRIVILEGES:
+        return WSREP_SYNC_WAIT_NONE;
+      default:
+        return WSREP_SYNC_WAIT_BEFORE_SHOW;
+      }
+    }
+  }
+  return WSREP_SYNC_WAIT_NONE;
+}
+
+bool wsrep_sync_wait(THD* thd, enum enum_sql_command command)
+{
+  bool res = false;
+  if (WSREP_CLIENT(thd) && thd->variables.wsrep_sync_wait)
+    res = wsrep_sync_wait(thd, wsrep_sync_wait_mask_for_command(command));
+  return res;
+}
+
 void wsrep_keys_free(wsrep_key_arr_t* key_arr)
 {
     for (size_t i= 0; i < key_arr->keys_len; ++i)
@@ -2706,11 +2773,6 @@ extern  bool wsrep_thd_ignore_table(THD *thd)
   return thd->wsrep_ignore_table;
 }
 
-bool wsrep_is_show_query(enum enum_sql_command command)
-{
-  DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
-  return (sql_command_flags[command] & CF_STATUS_COMMAND) != 0;
-}
 bool wsrep_create_like_table(THD* thd, TABLE_LIST* table,
                              TABLE_LIST* src_table,
                              HA_CREATE_INFO *create_info)

sql/wsrep_mysqld.h

@@ -208,6 +208,7 @@ extern bool wsrep_start_replication(const char *wsrep_cluster_address);
 extern void wsrep_shutdown_replication();
 extern bool wsrep_must_sync_wait (THD* thd, uint mask= WSREP_SYNC_WAIT_BEFORE_READ);
 extern bool wsrep_sync_wait (THD* thd, uint mask= WSREP_SYNC_WAIT_BEFORE_READ);
+extern bool wsrep_sync_wait (THD* thd, enum enum_sql_command command);
 extern enum wsrep::provider::status
 wsrep_sync_wait_upto (THD* thd, wsrep_gtid_t* upto, int timeout);
 extern int  wsrep_check_opts();