MDEV-23033: All slaves crash once in ~24 hours and loop restart with signal 11

Problem:
=======
Upon deleting or updating a row in a parent table (with primary key), if
the child table has virtual column and an associated key with ON UPDATE
CASCADE/ON DELETE CASCADE, it will result in slave crash.

Analysis:
========
Tables which are related through foreign key require prelocking similar to
triggers. i.e If a table has triggers/foreign keys we should add all tables
and routines used by them to the prelocking set.  This prelocking happens
during 'open_and_lock_tables' call.  Each table being opened is checked for
foreign key references. If foreign key reference exists then the child
table is opened and it is linked to the table_list. Upon any modification
to  parent table its corresponding child tables are retried from table_list
and they are updated accordingly. This prelocking work fine on master.

On slave  prelocking works for following cases.
 - Statement/mixed based replication
 - In row based replication when trigger execution is enabled through
   'slave_run_triggers_for_rbr=YES/LOGGING/ENFORCE'

Otherwise it results in an assert/crash, as the parent table will not find
the corresponding child table and it will be NULL. Dereferencing NULL
pointer leads to slave server exit.

Fix:
===
Introduce a new 'slave_fk_event_map' flag similar to 'trg_event_map'. This
flag will ensure that when foreign key is enabled in row based replication
all the parent and child tables are prelocked, so that parent is able to
locate the child table.

Note: This issue is specific to slave, hence only slave needs to be
      upgraded.

3 files changed, 101 insertions, 65 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index c649e1f64fa..c32f31db1f6 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -10718,7 +10718,7 @@ int Rows_log_event::do_add_row_data(uchar *row_data, size_t length)
   There was the same problem with MERGE MYISAM tables and so here we try to
   go the same way.
 */
-static void restore_empty_query_table_list(LEX *lex)
+inline void restore_empty_query_table_list(LEX *lex)
 {
   if (lex->first_not_own_table())
       (*lex->first_not_own_table()->prev_global)= NULL;
@@ -10733,6 +10733,8 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
   TABLE* table;
   DBUG_ENTER("Rows_log_event::do_apply_event(Relay_log_info*)");
   int error= 0;
+  LEX *lex= thd->lex;
+  uint8 new_trg_event_map= get_trg_event_map();
   /*
     If m_table_id == ~0ULL, then we have a dummy event that does not
     contain any data.  In that case, we just remove all tables in the
@@ -10823,27 +10825,29 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
                       DBUG_ASSERT(!debug_sync_set_action(thd, STRING_WITH_LEN(action)));
                     };);
 
-    if (slave_run_triggers_for_rbr)
-    {
-      LEX *lex= thd->lex;
-      uint8 new_trg_event_map= get_trg_event_map();
-
-      /*
-        Trigger's procedures work with global table list. So we have to add
-        rgi->tables_to_lock content there to get trigger's in the list.
+    /*
+       Trigger's procedures work with global table list. So we have to add
+       rgi->tables_to_lock content there to get trigger's in the list.
 
-        Then restore_empty_query_table_list() restore the list as it was
-      */
-      DBUG_ASSERT(lex->query_tables == NULL);
-      if ((lex->query_tables= rgi->tables_to_lock))
-        rgi->tables_to_lock->prev_global= &lex->query_tables;
+       Then restore_empty_query_table_list() restore the list as it was
+    */
+    DBUG_ASSERT(lex->query_tables == NULL);
+    if ((lex->query_tables= rgi->tables_to_lock))
+      rgi->tables_to_lock->prev_global= &lex->query_tables;
 
-      for (TABLE_LIST *tables= rgi->tables_to_lock; tables;
-           tables= tables->next_global)
+    for (TABLE_LIST *tables= rgi->tables_to_lock; tables;
+         tables= tables->next_global)
+    {
+      if (slave_run_triggers_for_rbr)
       {
         tables->trg_event_map= new_trg_event_map;
         lex->query_tables_last= &tables->next_global;
       }
+      else if (!WSREP_ON)
+      {
+        tables->slave_fk_event_map= new_trg_event_map;
+        lex->query_tables_last= &tables->next_global;
+      }
     }
     if (open_and_lock_tables(thd, rgi->tables_to_lock, FALSE, 0))
     {
@@ -11193,8 +11197,7 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
   }
 
   /* remove trigger's tables */
-  if (slave_run_triggers_for_rbr)
-    restore_empty_query_table_list(thd->lex);
+  restore_empty_query_table_list(thd->lex);
 
 #if defined(WITH_WSREP) && defined(HAVE_QUERY_CACHE)
     if (WSREP(thd) && thd->wsrep_exec_mode == REPL_RECV)
@@ -11212,8 +11215,7 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
   DBUG_RETURN(error);
 
 err:
-  if (slave_run_triggers_for_rbr)
-    restore_empty_query_table_list(thd->lex);
+  restore_empty_query_table_list(thd->lex);
   rgi->slave_close_thread_tables(thd);
   DBUG_RETURN(error);
 }
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 8e57ea437b6..b8d18abb50c 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -4341,6 +4341,70 @@ bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_STRING *db,
   return false;
 }
 
+/**
+  Extend the table_list to include foreign tables for prelocking.
+
+  @param[in]  thd              Thread context.
+  @param[in]  prelocking_ctx   Prelocking context of the statement.
+  @param[in]  table_list       Table list element for table.
+  @param[in]  sp               Routine body.
+  @param[out] need_prelocking  Set to TRUE if method detects that prelocking
+                               required, not changed otherwise.
+
+  @retval FALSE  Success.
+  @retval TRUE   Failure (OOM).
+*/
+inline bool
+prepare_fk_prelocking_list(THD *thd, Query_tables_list *prelocking_ctx,
+                           TABLE_LIST *table_list, bool *need_prelocking,
+                           uint8 op)
+{
+  List <FOREIGN_KEY_INFO> fk_list;
+  List_iterator<FOREIGN_KEY_INFO> fk_list_it(fk_list);
+  FOREIGN_KEY_INFO *fk;
+  Query_arena *arena, backup;
+
+  arena= thd->activate_stmt_arena_if_needed(&backup);
+
+  table_list->table->file->get_parent_foreign_key_list(thd, &fk_list);
+  if (thd->is_error())
+  {
+    if (arena)
+      thd->restore_active_arena(arena, &backup);
+    return TRUE;
+  }
+
+  *need_prelocking= TRUE;
+
+  while ((fk= fk_list_it++))
+  {
+    // FK_OPTION_RESTRICT and FK_OPTION_NO_ACTION only need read access
+    static bool can_write[]= { true, false, true, true, false, true };
+    thr_lock_type lock_type;
+
+    if ((op & (1 << TRG_EVENT_DELETE) && can_write[fk->delete_method])
+        || (op & (1 << TRG_EVENT_UPDATE) && can_write[fk->update_method]))
+      lock_type= TL_WRITE_ALLOW_WRITE;
+    else
+      lock_type= TL_READ;
+
+    if (table_already_fk_prelocked(prelocking_ctx->query_tables,
+          fk->foreign_db, fk->foreign_table,
+          lock_type))
+      continue;
+
+    TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
+    tl->init_one_table_for_prelocking(fk->foreign_db->str, fk->foreign_db->length,
+        fk->foreign_table->str, fk->foreign_table->length,
+        NULL, lock_type, false, table_list->belong_to_view,
+        op, &prelocking_ctx->query_tables_last);
+  }
+  if (arena)
+    thd->restore_active_arena(arena, &backup);
+
+  return FALSE;
+}
+
 
 /**
   Defines how prelocking algorithm for DML statements should handle table list
@@ -4381,55 +4445,21 @@ handle_table(THD *thd, Query_tables_list *prelocking_ctx,
           add_tables_and_routines_for_triggers(thd, prelocking_ctx, table_list))
         return TRUE;
     }
-
     if (table_list->table->file->referenced_by_foreign_key())
     {
-      List <FOREIGN_KEY_INFO> fk_list;
-      List_iterator<FOREIGN_KEY_INFO> fk_list_it(fk_list);
-      FOREIGN_KEY_INFO *fk;
-      Query_arena *arena, backup;
-
-      arena= thd->activate_stmt_arena_if_needed(&backup);
-
-      table_list->table->file->get_parent_foreign_key_list(thd, &fk_list);
-      if (thd->is_error())
-      {
-        if (arena)
-          thd->restore_active_arena(arena, &backup);
-        return TRUE;
-      }
-
-      *need_prelocking= TRUE;
-
-      while ((fk= fk_list_it++))
-      {
-        // FK_OPTION_RESTRICT and FK_OPTION_NO_ACTION only need read access
-        static bool can_write[]= { true, false, true, true, false, true };
-        uint8 op= table_list->trg_event_map;
-        thr_lock_type lock_type;
-
-        if ((op & (1 << TRG_EVENT_DELETE) && can_write[fk->delete_method])
-         || (op & (1 << TRG_EVENT_UPDATE) && can_write[fk->update_method]))
-          lock_type= TL_WRITE_ALLOW_WRITE;
-        else
-          lock_type= TL_READ;
-
-        if (table_already_fk_prelocked(prelocking_ctx->query_tables,
-                                       fk->foreign_db, fk->foreign_table,
-                                       lock_type))
-          continue;
-
-        TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
-        tl->init_one_table_for_prelocking(fk->foreign_db->str, fk->foreign_db->length,
-                           fk->foreign_table->str, fk->foreign_table->length,
-                           NULL, lock_type, false, table_list->belong_to_view,
-                           op, &prelocking_ctx->query_tables_last);
-      }
-      if (arena)
-        thd->restore_active_arena(arena, &backup);
+      return (prepare_fk_prelocking_list(thd, prelocking_ctx, table_list,
+                                         need_prelocking,
+                                         table_list->trg_event_map));
     }
   }
+  else if (table_list->slave_fk_event_map &&
+           table_list->table->file->referenced_by_foreign_key())
+  {
+      return (prepare_fk_prelocking_list(thd, prelocking_ctx,
+                                         table_list, need_prelocking,
+                                         table_list->slave_fk_event_map));
 
+  }
   return FALSE;
 }
 
diff --git a/sql/table.h b/sql/table.h
index 9a864f7ce9f..57706655d9b 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -2277,8 +2277,12 @@ struct TABLE_LIST
     Indicates what triggers we need to pre-load for this TABLE_LIST
     when opening an associated TABLE. This is filled after
     the parsed tree is created.
+
+    slave_fk_event_map is filled on the slave side with bitmaps value
+    representing row-based event operation to help find and prelock
+    possible FK constrain-related child tables.
   */
-  uint8 trg_event_map;
+  uint8 trg_event_map, slave_fk_event_map;
   /* TRUE <=> this table is a const one and was optimized away. */
   bool optimized_away;