MDEV-22232: Fix CTAS replay & retry in case it gets BF-aborted

- Add selected tables as shared keys for CTAS certification
- Set proper security context on the replayer thread
- Disallow CTAS command retry

mysql-test/suite/galera/r/MDEV-22232.result

+connection node_2;
+connection node_1;
+connect con1,127.0.0.1,root,,test,$NODE_MYPORT_1;
+--- CTAS with empty result set ---
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+SET DEBUG_SYNC = 'create_table_select_before_create SIGNAL may_alter WAIT_FOR bf_abort';
+CREATE TABLE t2 SELECT * FROM t1;
+connection node_1;
+SET DEBUG_SYNC = 'now WAIT_FOR may_alter';
+ALTER TABLE t1 DROP FOREIGN KEY b, ALGORITHM=COPY;
+connection con1;
+ERROR 70100: Query execution was interrupted
+SET DEBUG_SYNC = 'RESET';
+--- CTAS with non-empty result set ---
+INSERT INTO t1 VALUES (10), (20), (30);
+SET DEBUG_SYNC = 'create_table_select_before_create SIGNAL may_alter WAIT_FOR bf_abort';
+CREATE TABLE t2 SELECT * FROM t1;
+connection node_1;
+SET DEBUG_SYNC = 'now WAIT_FOR may_alter';
+ALTER TABLE t1 DROP FOREIGN KEY b, ALGORITHM=COPY;
+connection con1;
+ERROR 70100: Query execution was interrupted
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t1;
+disconnect con1;
+disconnect node_2;
+disconnect node_1;

mysql-test/suite/galera/t/MDEV-22232.test

+#
+# MDEV-22232: CTAS execution crashes during replay.
+#
+# There were multiple problems and two failing scenarios with empty result set
+# and with non-empty result set:
+# - CTAS didn't add shared keys for selected tables
+# - Security context wasn't set on the replayer thread
+# - CTAS was retried after failure - now retry disabled
+
+--source include/galera_cluster.inc
+--source include/have_debug_sync.inc
+--source include/have_debug.inc
+
+--connect con1,127.0.0.1,root,,test,$NODE_MYPORT_1
+
+# Scenario 1
+--echo --- CTAS with empty result set ---
+CREATE TABLE t1 (a INT) ENGINE=InnoDB;
+
+# Run CTAS until the resulting table gets created,
+# then it gets BF aborted by ALTER.
+SET DEBUG_SYNC = 'create_table_select_before_create SIGNAL may_alter WAIT_FOR bf_abort';
+--send
+  CREATE TABLE t2 SELECT * FROM t1;
+
+# Wait for CTAS to reach the table create point,
+# start executing ALTER and BF abort CTAS.
+--connection node_1
+SET DEBUG_SYNC = 'now WAIT_FOR may_alter';
+--disable_result_log
+--error ER_ERROR_ON_RENAME
+ALTER TABLE t1 DROP FOREIGN KEY b, ALGORITHM=COPY;
+--enable_result_log
+
+--connection con1
+# CTAS gets BF aborted.
+--error ER_QUERY_INTERRUPTED
+--reap
+
+# Cleanup
+SET DEBUG_SYNC = 'RESET';
+
+
+# Scenario 2
+--echo --- CTAS with non-empty result set ---
+INSERT INTO t1 VALUES (10), (20), (30);
+
+# Run CTAS until the resulting table gets created,
+# then it gets BF aborted by ALTER.
+SET DEBUG_SYNC = 'create_table_select_before_create SIGNAL may_alter WAIT_FOR bf_abort';
+--send
+  CREATE TABLE t2 SELECT * FROM t1;
+
+# Wait for CTAS to reach the table create point,
+# start executing ALTER and BF abort CTAS.
+--connection node_1
+SET DEBUG_SYNC = 'now WAIT_FOR may_alter';
+--disable_result_log
+--error ER_ERROR_ON_RENAME
+ALTER TABLE t1 DROP FOREIGN KEY b, ALGORITHM=COPY;
+--enable_result_log
+
+--connection con1
+# CTAS gets BF aborted.
+--error ER_QUERY_INTERRUPTED
+--reap
+
+# Cleanup
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t1;
+--disconnect con1
+--source include/galera_end.inc

sql/sql_insert.cc

@@ -84,6 +84,7 @@
 #include "debug_sync.h"
 
 #ifdef WITH_WSREP
+#include "wsrep_mysqld.h" /* wsrep_append_table_keys() */
 #include "wsrep_trans_observer.h" /* wsrep_start_transction() */
 #endif /* WITH_WSREP */
 
@@ -4802,17 +4803,13 @@ bool select_create::send_eof()
                   thd->wsrep_trx_id(), thd->thread_id, thd->query_id);
 
       /*
-        append table level exclusive key for CTAS
+        For CTAS, append table level exclusive key for created table
+        and table level shared key for selected table.
       */
-      wsrep_key_arr_t key_arr= {0, 0};
-      wsrep_prepare_keys_for_isolation(thd,
-                                       create_table->db.str,
-                                       create_table->table_name.str,
-                                       table_list,
-                                       &key_arr);
-      int rcode= wsrep_thd_append_key(thd, key_arr.keys, key_arr.keys_len,
-                                      WSREP_SERVICE_KEY_EXCLUSIVE);
-      wsrep_keys_free(&key_arr);
+      int rcode= wsrep_append_table_keys(thd, create_table, table_list,
+              WSREP_SERVICE_KEY_EXCLUSIVE);
+      rcode= rcode || wsrep_append_table_keys(thd, nullptr, select_tables,
+              WSREP_SERVICE_KEY_SHARED);
       if (rcode)
       {
         DBUG_PRINT("wsrep", ("row key failed: %d", rcode));

sql/sql_table.cc

@@ -11634,8 +11634,18 @@ bool Sql_cmd_create_table_like::execute(THD *thd)
   Alter_info alter_info(lex->alter_info, thd->mem_root);
 
 #ifdef WITH_WSREP
+  bool wsrep_ctas= false;
   // If CREATE TABLE AS SELECT and wsrep_on
-  const bool wsrep_ctas= (select_lex->item_list.elements && WSREP(thd));
+  if (WSREP(thd) && (select_lex->item_list.elements ||
+     // Only CTAS may be applied not using TOI.
+     (wsrep_thd_is_applying(thd) && !wsrep_thd_is_toi(thd))))
+  {
+    wsrep_ctas= true;
+
+    // MDEV-22232: Disable CTAS retry by setting the retry counter to the
+    // threshold value.
+    thd->wsrep_retry_counter= thd->variables.wsrep_retry_autocommit;
+  }
 
   // This will be used in THD::decide_logging_format if CTAS
   Enable_wsrep_ctas_guard wsrep_ctas_guard(thd, wsrep_ctas);

sql/wsrep_client_service.cc

@@ -283,6 +283,13 @@ enum wsrep::provider::status Wsrep_client_service::replay()
     original THD state during replication event applying.
    */
   THD *replayer_thd= new THD(true, true);
+  // Replace the security context of the replayer with the security context
+  // of the original THD. Since security context class doesn't have proper
+  // copy constructors, we need to store the original one and set it back
+  // before destruction so that THD desctruction doesn't cause double-free
+  // on the replaced security context.
+  Security_context old_ctx = replayer_thd->main_security_ctx;
+  replayer_thd->main_security_ctx = m_thd->main_security_ctx;
   replayer_thd->thread_stack= m_thd->thread_stack;
   replayer_thd->real_id= pthread_self();
   replayer_thd->prior_thr_create_utime=
@@ -299,6 +306,7 @@ enum wsrep::provider::status Wsrep_client_service::replay()
     replayer_service.replay_status(ret);
   }
 
+  replayer_thd->main_security_ctx = old_ctx;
   delete replayer_thd;
   DBUG_RETURN(ret);
 }

sql/wsrep_mysqld.cc

@@ -1266,7 +1266,13 @@ bool wsrep_sync_wait(THD* thd, enum enum_sql_command command)
   return res;
 }
 
-void wsrep_keys_free(wsrep_key_arr_t* key_arr)
+typedef struct wsrep_key_arr
+{
+    wsrep_key_t* keys;
+    size_t       keys_len;
+} wsrep_key_arr_t;
+
+static void wsrep_keys_free(wsrep_key_arr_t* key_arr)
 {
     for (size_t i= 0; i < key_arr->keys_len; ++i)
     {
@@ -1516,18 +1522,30 @@ static bool wsrep_prepare_keys_for_isolation(THD*              thd,
 }
 
 /*
- * Prepare key list from db/table and table_list
+ * Prepare key list from db/table and table_list and append it to Wsrep
+ * with the given key type.
  *
  * Return zero in case of success, 1 in case of failure.
  */
-
-bool wsrep_prepare_keys_for_isolation(THD*              thd,
-                                      const char*       db,
-                                      const char*       table,
-                                      const TABLE_LIST* table_list,
-                                      wsrep_key_arr_t*  ka)
-{
-  return wsrep_prepare_keys_for_isolation(thd, db, table, table_list, NULL, ka);
+int wsrep_append_table_keys(THD* thd,
+                            TABLE_LIST* first_table,
+                            TABLE_LIST* table_list,
+                            Wsrep_service_key_type key_type)
+{
+   wsrep_key_arr_t key_arr= {0, 0};
+   const char* db_name= first_table ? first_table->db.str : NULL;
+   const char* table_name= first_table ? first_table->table_name.str : NULL;
+   int rcode= wsrep_prepare_keys_for_isolation(thd, db_name, table_name,
+                                               table_list, NULL, &key_arr);
+
+   if (!rcode && key_arr.keys_len)
+   {
+     rcode= wsrep_thd_append_key(thd, key_arr.keys,
+                                 key_arr.keys_len, key_type);
+   }
+
+   wsrep_keys_free(&key_arr);
+   return rcode;
 }
 
 bool wsrep_prepare_key(const uchar* cache_key, size_t cache_key_len,

sql/wsrep_mysqld.h

@@ -347,17 +347,10 @@ int wsrep_must_ignore_error(THD* thd);
 
 bool wsrep_replicate_GTID(THD* thd);
 
-typedef struct wsrep_key_arr
-{
-    wsrep_key_t* keys;
-    size_t       keys_len;
-} wsrep_key_arr_t;
-bool wsrep_prepare_keys_for_isolation(THD*              thd,
-                                      const char*       db,
-                                      const char*       table,
-                                      const TABLE_LIST* table_list,
-                                      wsrep_key_arr_t*  ka);
-void wsrep_keys_free(wsrep_key_arr_t* key_arr);
+int wsrep_append_table_keys(THD* thd,
+                            TABLE_LIST* first_table,
+                            TABLE_LIST* table_list,
+                            Wsrep_service_key_type key_type);
 
 extern void
 wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,