Fix FK constraint violation in applier, after ALTER TABLE ADD FK

Adding a FK constraint to an existing table (ALTER TABLE ADD FOREIGN
KEY) causes the applier to fail, if a concurrent DML statement that
violate the new constraint (i.e. a DELETE or UPDATE of record in the
parent table).

For exmaple, the following scenario causes a crash in the applier:

1. ALTER successfully adds FK constraint in node_1
2. On node_2 is UPDATE is in pre_commit() and has certified successfully
3. ALTER is delivered in node_2 and BF aborts DML
4. Applying UPDATE event causes FK violation in node_1

To avoid this situation it is necessary for UPDATE to fail during
certification. And for the UPDATE to fail certfication it is necessary
that ALTER appends certification keys for both the child and the parent
table. Before this patch, ALTER TABLE ADD FK only appended keys for
child table which is ALTERed.

4 files changed, 155 insertions, 118 deletions

diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index 0b4636c1f0f..1d69b06d798 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -106,14 +106,10 @@ bool Alter_table_statement::execute(THD *thd)
       (!thd->is_current_stmt_binlog_format_row() ||
        !find_temporary_table(thd, first_table)))
   {
-    if (wsrep_to_isolation_begin(thd,
-                                 lex->name.str ? select_lex->db : NULL,
-                                 lex->name.str ? lex->name.str : NULL,
-                                 first_table))
-    {
-      WSREP_WARN("ALTER TABLE isolation failure");
-      DBUG_RETURN(TRUE);
-    }
+    WSREP_TO_ISOLATION_BEGIN_ALTER(((lex->name.str) ? select_lex->db : NULL),
+                                    ((lex->name.str) ? lex->name.str : NULL),
+                                    first_table,
+                                    &alter_info);
 
     thd->variables.auto_increment_offset = 1;
     thd->variables.auto_increment_increment = 1;
@@ -128,6 +124,11 @@ bool Alter_table_statement::execute(THD *thd)
                             lex->ignore, lex->online);
 
 #ifdef WITH_WSREP
+error:
+  {
+    WSREP_WARN("ALTER TABLE isolation failure");
+    DBUG_RETURN(TRUE);
+  }
 #endif /* WITH_WSREP */
   DBUG_RETURN(result);
 }
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index 2093ecc85ee..a0c1c89eecc 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -214,6 +214,11 @@ inline bool is_supported_parser_charset(CHARSET_INFO *cs)
 #define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)                   \
   if (WSREP(thd) && wsrep_to_isolation_begin(thd, db_, table_, table_list_)) goto error;
 
+#define WSREP_TO_ISOLATION_BEGIN_ALTER(db_, table_, table_list_, alter_info_) \
+  if (WSREP(thd) && wsrep_to_isolation_begin(thd, db_, table_,                \
+                                             table_list_, alter_info_))       \
+    goto error;
+
 #define WSREP_TO_ISOLATION_END                                              \
   if (WSREP(thd) || (thd && thd->wsrep_exec_mode==TOTAL_ORDER))             \
     wsrep_to_isolation_end(thd);
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 54fdf430f86..a7950754666 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -981,43 +981,137 @@ static bool wsrep_prepare_key_for_isolation(const char* db,
                                             wsrep_buf_t* key,
                                             size_t* key_len)
 {
-    if (*key_len < 2) return false;
+  if (*key_len < 2) return false;
 
-    switch (wsrep_protocol_version)
-    {
-    case 0:
-        *key_len= 0;
-        break;
-    case 1:
-    case 2:
-    case 3:
+  switch (wsrep_protocol_version)
+  {
+  case 0:
+    *key_len= 0;
+    break;
+  case 1:
+  case 2:
+  case 3:
+  {
+    *key_len= 0;
+    if (db)
     {
-        *key_len= 0;
-        if (db)
-        {
-            // sql_print_information("%s.%s", db, table);
-            if (db)
-            {
-                key[*key_len].ptr= db;
-                key[*key_len].len= strlen(db);
-                ++(*key_len);
-                if (table)
-                {
-                    key[*key_len].ptr= table;
-                    key[*key_len].len= strlen(table);
-                    ++(*key_len);
-                }
-            }
-        }
-        break;
+      // sql_print_information("%s.%s", db, table);
+      key[*key_len].ptr= db;
+      key[*key_len].len= strlen(db);
+      ++(*key_len);
+      if (table)
+      {
+        key[*key_len].ptr= table;
+        key[*key_len].len= strlen(table);
+        ++(*key_len);
+      }
     }
-    default:
+    break;
+  }
+  default:
+    return false;
+  }
+  return true;
+}
+
+
+static bool wsrep_prepare_key_for_isolation(const char* db,
+                                            const char* table,
+                                            wsrep_key_arr_t* ka)
+{
+  wsrep_key_t* tmp;
+  tmp= (wsrep_key_t*)my_realloc(ka->keys,
+                                (ka->keys_len + 1) * sizeof(wsrep_key_t),
+                                MYF(0));
+  if (!tmp)
+  {
+    WSREP_ERROR("Can't allocate memory for key_array");
+    return false;
+  }
+  ka->keys= tmp;
+  if (!(ka->keys[ka->keys_len].key_parts= (wsrep_buf_t*)
+        my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
+  {
+    WSREP_ERROR("Can't allocate memory for key_parts");
+    return false;
+  }
+  ka->keys[ka->keys_len].key_parts_num= 2;
+  ++ka->keys_len;
+  if (!wsrep_prepare_key_for_isolation(db, table,
+                                       (wsrep_buf_t*)ka->keys[ka->keys_len - 1].key_parts,
+                                       &ka->keys[ka->keys_len - 1].key_parts_num))
+  {
+    WSREP_ERROR("Preparing keys for isolation failed");
+    return false;
+  }
+
+  return true;
+}
+
+
+static bool wsrep_prepare_keys_for_alter_add_fk(char* child_table_db,
+                                                Alter_info* alter_info,
+                                                wsrep_key_arr_t* ka)
+{
+  Key *key;
+  List_iterator<Key> key_iterator(alter_info->key_list);
+  while ((key= key_iterator++))
+  {
+    if (key->type == Key::FOREIGN_KEY)
+    {
+      Foreign_key *fk_key= (Foreign_key *)key;
+      const char *db_name= fk_key->ref_table->db.str;
+      const char *table_name= fk_key->ref_table->table.str;
+      if (!db_name)
+      {
+        db_name= child_table_db;
+      }
+      if (!wsrep_prepare_key_for_isolation(db_name, table_name, ka))
+      {
         return false;
+      }
     }
+  }
+  return true;
+}
 
-    return true;
+
+static bool wsrep_prepare_keys_for_isolation(THD*              thd,
+                                             const char*       db,
+                                             const char*       table,
+                                             const TABLE_LIST* table_list,
+                                             Alter_info*       alter_info,
+                                             wsrep_key_arr_t*  ka)
+{
+  ka->keys= 0;
+  ka->keys_len= 0;
+
+  if (db || table)
+  {
+    if (!wsrep_prepare_key_for_isolation(db, table, ka))
+      goto err;
+  }
+
+  for (const TABLE_LIST* table= table_list; table; table= table->next_global)
+  {
+    if (!wsrep_prepare_key_for_isolation(table->db, table->table_name, ka))
+      goto err;
+  }
+
+  if (alter_info && (alter_info->flags & ALTER_FOREIGN_KEY))
+  {
+    if (!wsrep_prepare_keys_for_alter_add_fk(table_list->db, alter_info, ka))
+      goto err;
+  }
+
+  return false;
+
+err:
+  wsrep_keys_free(ka);
+  return true;
 }
 
+
 /* Prepare key list from db/table and table_list */
 bool wsrep_prepare_keys_for_isolation(THD*              thd,
                                       const char*       db,
@@ -1025,78 +1119,7 @@ bool wsrep_prepare_keys_for_isolation(THD*              thd,
                                       const TABLE_LIST* table_list,
                                       wsrep_key_arr_t*  ka)
 {
-    ka->keys= 0;
-    ka->keys_len= 0;
-
-    extern TABLE* find_temporary_table(THD*, const TABLE_LIST*);
-
-    if (db || table)
-    {
-        TABLE_LIST tmp_table;
-        bzero((char*) &tmp_table,sizeof(tmp_table));
-        tmp_table.table_name= (char*)db;
-        tmp_table.db= (char*)table;
-        if (!table || !find_temporary_table(thd, &tmp_table))
-        {
-            if (!(ka->keys= (wsrep_key_t*)my_malloc(sizeof(wsrep_key_t), MYF(0))))
-            {
-                WSREP_ERROR("Can't allocate memory for key_array");
-                goto err;
-            }
-            ka->keys_len= 1;
-            if (!(ka->keys[0].key_parts= (wsrep_buf_t*)
-                  my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
-            {
-                WSREP_ERROR("Can't allocate memory for key_parts");
-                goto err;
-            }
-            ka->keys[0].key_parts_num= 2;
-            if (!wsrep_prepare_key_for_isolation(
-                    db, table,
-                    (wsrep_buf_t*)ka->keys[0].key_parts,
-                    &ka->keys[0].key_parts_num))
-            {
-                WSREP_ERROR("Preparing keys for isolation failed");
-                goto err;
-            }
-        }
-    }
-
-    for (const TABLE_LIST* table= table_list; table; table= table->next_global)
-    {
-        if (!find_temporary_table(thd, table))
-        {
-            wsrep_key_t* tmp;
-            tmp= (wsrep_key_t*)my_realloc(
-                ka->keys, (ka->keys_len + 1) * sizeof(wsrep_key_t), MYF(0));
-            if (!tmp)
-            {
-                WSREP_ERROR("Can't allocate memory for key_array");
-                goto err;
-            }
-            ka->keys= tmp;
-            if (!(ka->keys[ka->keys_len].key_parts= (wsrep_buf_t*)
-                  my_malloc(sizeof(wsrep_buf_t)*2, MYF(0))))
-            {
-                WSREP_ERROR("Can't allocate memory for key_parts");
-                goto err;
-            }
-            ka->keys[ka->keys_len].key_parts_num= 2;
-            ++ka->keys_len;
-            if (!wsrep_prepare_key_for_isolation(
-                    table->db, table->table_name,
-                    (wsrep_buf_t*)ka->keys[ka->keys_len - 1].key_parts,
-                    &ka->keys[ka->keys_len - 1].key_parts_num))
-            {
-                WSREP_ERROR("Preparing keys for isolation failed");
-                goto err;
-            }
-        }
-    }
-    return true;
-err:
-    wsrep_keys_free(ka);
-    return false;
+  return wsrep_prepare_keys_for_isolation(thd, db, table, table_list, NULL, ka);
 }
 
 
@@ -1260,7 +1283,8 @@ create_view_query(THD *thd, uchar** buf, size_t* buf_len)
 }
 
 static int wsrep_TOI_begin(THD *thd, char *db_, char *table_,
-                           const TABLE_LIST* table_list)
+                           const TABLE_LIST* table_list,
+                           Alter_info* alter_info)
 {
   wsrep_status_t ret(WSREP_WARNING);
   uchar* buf(0);
@@ -1295,8 +1319,10 @@ static int wsrep_TOI_begin(THD *thd, char *db_, char *table_,
 
   wsrep_key_arr_t key_arr= {0, 0};
   struct wsrep_buf buff = { buf, buf_len };
-  if (!buf_err                                                    &&
-      wsrep_prepare_keys_for_isolation(thd, db_, table_, table_list, &key_arr)&&
+  if (!buf_err                                                                  &&
+      !wsrep_prepare_keys_for_isolation(thd, db_, table_,
+                                        table_list, alter_info, &key_arr)       &&
+      key_arr.keys_len > 0                                                      &&
       WSREP_OK == (ret = wsrep->to_execute_start(wsrep, thd->thread_id,
                                                  key_arr.keys, key_arr.keys_len,
                                                  &buff, 1,
@@ -1417,9 +1443,9 @@ static void wsrep_RSU_end(THD *thd)
 }
 
 int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
-                             const TABLE_LIST* table_list)
+                             const TABLE_LIST* table_list,
+                             Alter_info* alter_info)
 {
-
   /*
     No isolation for applier or replaying threads.
    */
@@ -1469,9 +1495,12 @@ int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
   if (thd->variables.wsrep_on && thd->wsrep_exec_mode==LOCAL_STATE)
   {
     switch (wsrep_OSU_method_options) {
-    case WSREP_OSU_TOI: ret =  wsrep_TOI_begin(thd, db_, table_,
-                                               table_list); break;
-    case WSREP_OSU_RSU: ret =  wsrep_RSU_begin(thd, db_, table_); break;
+    case WSREP_OSU_TOI:
+      ret= wsrep_TOI_begin(thd, db_, table_, table_list, alter_info);
+      break;
+    case WSREP_OSU_RSU:
+      ret= wsrep_RSU_begin(thd, db_, table_);
+      break;
     }
     if (!ret)
     {
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index 56e3baae7cc..b1dc5ba452b 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -305,8 +305,10 @@ extern PSI_mutex_key key_LOCK_wsrep_slave_threads;
 extern PSI_mutex_key key_LOCK_wsrep_desync;
 #endif /* HAVE_PSI_INTERFACE */
 struct TABLE_LIST;
+class Alter_info;
 int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
-                             const TABLE_LIST* table_list);
+                             const TABLE_LIST* table_list,
+                             Alter_info* alter_info = NULL);
 void wsrep_to_isolation_end(THD *thd);
 void wsrep_cleanup_transaction(THD *thd);
 int wsrep_to_buf_helper(