MW-322 - CTAS fix merged to 5.5-v25 branch

Signed-off-by: Jan Lindström <jan.lindstrom@mariadb.com>

sql/handler.cc

@@ -5426,6 +5426,11 @@ void ha_wsrep_fake_trx_id(THD *thd)
     DBUG_VOID_RETURN;
   }
 
+  if (thd->wsrep_ws_handle.trx_id != WSREP_UNDEFINED_TRX_ID)
+  {
+    WSREP_DEBUG("fake trx id skipped: %lu", thd->wsrep_ws_handle.trx_id);
+    DBUG_VOID_RETURN;
+  }
   handlerton *hton= installed_htons[DB_TYPE_INNODB];
   if (hton && hton->wsrep_fake_trx_id)
   {

sql/sql_base.cc

@@ -9506,11 +9506,22 @@ void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
       I_P_List_iterator<TABLE, TABLE_share> it(share->free_tables);
 #ifndef DBUG_OFF
       if (remove_type == TDC_RT_REMOVE_ALL)
+#ifdef WITH_WSREP
       {
-        DBUG_ASSERT(share->used_tables.is_empty());
+        /* following assert may cause false posivive fire for CTAS */
+        if (thd->lex->sql_command != SQLCOM_CREATE_TABLE)
+        {
+#endif /* WITH_WSREP */
+          {
+            DBUG_ASSERT(share->used_tables.is_empty());
+          }
+#ifdef WITH_WSREP
+        }
       }
       else if (remove_type == TDC_RT_REMOVE_NOT_OWN ||
                remove_type == TDC_RT_REMOVE_NOT_OWN_AND_MARK_NOT_USABLE)
+#endif /* WITH_WSREP */
+
       {
         I_P_List_iterator<TABLE, TABLE_share> it2(share->used_tables);
         while ((table= it2++))

sql/sql_insert.cc

@@ -4313,9 +4313,38 @@ bool select_create::send_eof()
     */
     if (!table->s->tmp_table)
     {
+ #ifdef WITH_WSREP
+      /*
+         append table level exclusive key for CTAS
+      */
+      wsrep_key_arr_t key_arr= {0, 0};
+      wsrep_prepare_keys_for_isolation(thd,
+                                       create_table->db,
+                                       create_table->table_name,
+                                       table_list,
+                                       &key_arr);
+      int rcode = wsrep->append_key(
+                                wsrep,
+                                &thd->wsrep_ws_handle,
+                                key_arr.keys, //&wkey,
+                                key_arr.keys_len,
+                                WSREP_KEY_EXCLUSIVE,
+                                false);
+      wsrep_keys_free(&key_arr);
+      if (rcode) {
+        DBUG_PRINT("wsrep", ("row key failed: %d", rcode));
+        WSREP_ERROR("Appending table key for CTAS failed: %s, %d",
+                    (wsrep_thd_query(thd)) ?
+                    wsrep_thd_query(thd) : "void", rcode);
+        return true;
+      }
+      /* If commit fails, we should be able to reset the OK status. */
+      thd->stmt_da->can_overwrite_status= TRUE;
+#endif /* WITH_WSREP */
       trans_commit_stmt(thd);
       trans_commit_implicit(thd);
 #ifdef WITH_WSREP
+      thd->stmt_da->can_overwrite_status= FALSE;
       mysql_mutex_lock(&thd->LOCK_wsrep_thd);
       if (thd->wsrep_conflict_state != NO_CONFLICT)
       {

sql/wsrep_mysqld.cc

@@ -944,17 +944,7 @@ bool wsrep_sync_wait (THD* thd, uint mask)
   return false;
 }
 
-/*
+void wsrep_keys_free(wsrep_key_arr_t* key_arr)
- * Helpers to deal with TOI key arrays
- */
-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)
     {
@@ -1019,11 +1009,11 @@ static bool wsrep_prepare_key_for_isolation(const char* db,
 }
 
 /* Prepare key list from db/table and table_list */
-static bool wsrep_prepare_keys_for_isolation(THD*              thd,
+bool wsrep_prepare_keys_for_isolation(THD*              thd,
-                                             const char*       db,
+                                      const char*       db,
-                                             const char*       table,
+                                      const char*       table,
-                                             const TABLE_LIST* table_list,
+                                      const TABLE_LIST* table_list,
-                                             wsrep_key_arr_t*  ka)
+                                      wsrep_key_arr_t*  ka)
 {
     ka->keys= 0;
     ka->keys_len= 0;
@@ -1554,7 +1544,7 @@ wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
     }
     else if (request_thd->lex->sql_command == SQLCOM_DROP_TABLE)
     {
-      WSREP_DEBUG("DROP caused BF abort");
+      WSREP_DEBUG("DROP caused BF abort, conf %d", granted_thd->wsrep_conflict_state);
       ticket->wsrep_report(wsrep_debug);
       mysql_mutex_unlock(&granted_thd->LOCK_wsrep_thd);
       wsrep_abort_thd((void*)request_thd, (void*)granted_thd, 1);

sql/wsrep_mysqld.h

@@ -322,4 +322,15 @@ const wsrep_uuid_t* wsrep_xid_uuid(const xid_t*);
 wsrep_seqno_t wsrep_xid_seqno(const xid_t*);
 extern "C" int wsrep_is_wsrep_xid(const void* xid);
 
+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);
 #endif /* WSREP_MYSQLD_H */

storage/innobase/handler/ha_innodb.cc

@@ -6000,10 +6000,14 @@ report_error:
 						   prebuilt->table->flags,
 						   user_thd);
 #ifdef WITH_WSREP
-	if (!error_result                                &&
+	if (!error_result
-	    wsrep_thd_exec_mode(user_thd) == LOCAL_STATE &&
+	    && wsrep_on(user_thd)
-	    wsrep_on(user_thd)                           &&
+	    && wsrep_thd_exec_mode(user_thd) == LOCAL_STATE
-	    !wsrep_consistency_check(user_thd))
+	    && !wsrep_consistency_check(user_thd))
+	    && (sql_command != SQLCOM_CREATE_TABLE)
+	    && (sql_command != SQLCOM_LOAD ||
+		thd_binlog_format(user_thd) == BINLOG_FORMAT_ROW)) {
+
 	{
 		if (wsrep_append_keys(user_thd, false, record, NULL))
 		{
@@ -12788,8 +12792,10 @@ wsrep_innobase_kill_one_trx(
  		    wsrep_thd_thread_id(thd),
 		    victim_trx->id);
 
-	WSREP_DEBUG("Aborting query: %s", 
+	WSREP_DEBUG("Aborting query: %s conf %d trx: %lu",
-		  (thd && wsrep_thd_query(thd)) ? wsrep_thd_query(thd) : "void");
+		    (thd && wsrep_thd_query(thd)) ? wsrep_thd_query(thd) : "void",
+		    wsrep_thd_conflict_state(thd),
+		    wsrep_thd_ws_handle(thd)->trx_id);
 
 	wsrep_thd_LOCK(thd);
 
@@ -12841,9 +12847,8 @@ wsrep_innobase_kill_one_trx(
 		} else {
 			rcode = wsrep->abort_pre_commit(
 				wsrep, bf_seqno,
-				(wsrep_trx_id_t)victim_trx->id
+				(wsrep_trx_id_t)wsrep_thd_ws_handle(thd)->trx_id
 			);
-			
 			switch (rcode) {
 			case WSREP_WARNING:
 				WSREP_DEBUG("cancel commit warning: %llu",
@@ -12973,15 +12978,17 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 			my_bool signal)
 {
 	DBUG_ENTER("wsrep_innobase_abort_thd");
-	trx_t* victim_trx = thd_to_trx(victim_thd);
+	
-	trx_t* bf_trx     = (bf_thd) ? thd_to_trx(bf_thd) : NULL;
+	trx_t* victim_trx	= thd_to_trx(victim_thd);
+	trx_t* bf_trx		= (bf_thd) ? thd_to_trx(bf_thd) : NULL;
+
+	WSREP_DEBUG("abort transaction: BF: %s victim: %s victim conf: %d",
+			wsrep_thd_query(bf_thd),
+			wsrep_thd_query(victim_thd),
+			wsrep_thd_conflict_state(victim_thd));
 
 	ut_ad(!mutex_own(&kernel_mutex));
 
-	WSREP_DEBUG("abort transaction: BF: %s victim: %s", 
-		    wsrep_thd_query(bf_thd),
-		    wsrep_thd_query(victim_thd));
-
 	if (victim_trx)
 	{
 		int rcode = wsrep_innobase_kill_one_trx(
@@ -13001,24 +13008,24 @@ wsrep_abort_transaction(handlerton* hton, THD *bf_thd, THD *victim_thd,
 static int innobase_wsrep_set_checkpoint(handlerton* hton, const XID* xid)
 {
 	DBUG_ASSERT(hton == innodb_hton_ptr);
-        if (wsrep_is_wsrep_xid(xid)) {
+	if (wsrep_is_wsrep_xid(xid)) {
-                mtr_t mtr;
+		mtr_t mtr;
-                mtr_start(&mtr);
+		mtr_start(&mtr);
-                trx_sysf_t* sys_header = trx_sysf_get(&mtr);
+		trx_sysf_t* sys_header = trx_sysf_get(&mtr);
-                trx_sys_update_wsrep_checkpoint(xid, sys_header, &mtr);
+		trx_sys_update_wsrep_checkpoint(xid, sys_header, &mtr);
-                mtr_commit(&mtr);
+		mtr_commit(&mtr);
-                innobase_flush_logs(hton);
+		innobase_flush_logs(hton);
-                return 0;
+		return 0;
-        } else {
+	} else {
-                return 1;
+		return 1;
-        }
+	}
 }
 
 static int innobase_wsrep_get_checkpoint(handlerton* hton, XID* xid)
 {
 	DBUG_ASSERT(hton == innodb_hton_ptr);
-        trx_sys_read_wsrep_checkpoint(xid);
+	trx_sys_read_wsrep_checkpoint(xid);
-        return 0;
+	return 0;
 }
 
 static void
@@ -13030,6 +13037,7 @@ wsrep_fake_trx_id(
 	mutex_enter(&kernel_mutex);
 	trx_id_t trx_id = trx_sys_get_new_trx_id();
 	mutex_exit(&kernel_mutex);
+	WSREP_DEBUG("innodb fake trx id: %lu thd: %s", trx_id, wsrep_thd_query(thd));
 
 	(void *)wsrep_ws_handle_for_trx(wsrep_thd_ws_handle(thd), trx_id);
 }