1 files changed, 61 insertions, 6 deletions

diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index bab2afb957a..c229655e915 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -1,5 +1,6 @@
 /*
    Copyright (c) 2005, 2013, Oracle and/or its affiliates.
+   Copyright (c) 2022, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -20,6 +21,7 @@
 #include "sql_parse.h"
 #include "sql_acl.h"
 #include "rpl_rli.h"
+#include "rpl_mi.h"
 #include "slave.h"
 #include "log_event.h"
 
@@ -70,7 +72,8 @@ static int check_event_type(int type, Relay_log_info *rli)
 
     /* It is always allowed to execute FD events. */
     return 0;
-    
+
+  case QUERY_EVENT:
   case TABLE_MAP_EVENT:
   case WRITE_ROWS_EVENT_V1:
   case UPDATE_ROWS_EVENT_V1:
@@ -167,6 +170,57 @@ int binlog_defragment(THD *thd)
   return 0;
 }
 
+/**
+  Wraps Log_event::apply_event to save and restore
+  session context in case of Query_log_event.
+
+  @param ev   replication event
+  @param rgi  execution context for the event
+
+  @return
+    0         on success,
+    non-zero  otherwise.
+*/
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+int save_restore_context_apply_event(Log_event *ev, rpl_group_info *rgi)
+{
+  if (ev->get_type_code() != QUERY_EVENT)
+    return ev->apply_event(rgi);
+
+  THD *thd= rgi->thd;
+  Relay_log_info *rli= thd->rli_fake;
+  DBUG_ASSERT(!rli->mi);
+  LEX_CSTRING connection_name= { STRING_WITH_LEN("BINLOG_BASE64_EVENT") };
+
+  if (!(rli->mi= new Master_info(&connection_name, false)))
+  {
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+    return -1;
+  }
+
+  sql_digest_state *m_digest= thd->m_digest;
+  PSI_statement_locker *m_statement_psi= thd->m_statement_psi;;
+  LEX_CSTRING save_db= thd->db;
+  my_thread_id m_thread_id= thd->variables.pseudo_thread_id;
+
+  thd->system_thread_info.rpl_sql_info= NULL;
+  thd->reset_db(&null_clex_str);
+
+  thd->m_digest= NULL;
+  thd->m_statement_psi= NULL;
+
+  int err= ev->apply_event(rgi);
+
+  thd->m_digest= m_digest;
+  thd->m_statement_psi= m_statement_psi;
+  thd->variables.pseudo_thread_id= m_thread_id;
+  thd->reset_db(&save_db);
+  delete rli->mi;
+  rli->mi= NULL;
+
+  return err;
+}
+#endif
 
 /**
   Execute a BINLOG statement.
@@ -216,11 +270,9 @@ void mysql_client_binlog_statement(THD* thd)
   if (!(rgi= thd->rgi_fake))
     rgi= thd->rgi_fake= new rpl_group_info(rli);
   rgi->thd= thd;
-
   const char *error= 0;
   Log_event *ev = 0;
   my_bool is_fragmented= FALSE;
-
   /*
     Out of memory check
   */
@@ -373,7 +425,7 @@ void mysql_client_binlog_statement(THD* thd)
         LEX *backup_lex;
 
         thd->backup_and_reset_current_lex(&backup_lex);
-        err= ev->apply_event(rgi);
+        err= save_restore_context_apply_event(ev, rgi);
         thd->restore_current_lex(backup_lex);
       }
       thd->variables.option_bits=
@@ -389,7 +441,7 @@ void mysql_client_binlog_statement(THD* thd)
         i.e. when this thread terminates.
       */
       if (ev->get_type_code() != FORMAT_DESCRIPTION_EVENT)
-        delete ev; 
+        delete ev;
       ev= 0;
       if (err)
       {
@@ -397,7 +449,8 @@ void mysql_client_binlog_statement(THD* thd)
           TODO: Maybe a better error message since the BINLOG statement
           now contains several events.
         */
-        my_error(ER_UNKNOWN_ERROR, MYF(0));
+        if (!thd->is_error())
+          my_error(ER_UNKNOWN_ERROR, MYF(0));
         goto end;
       }
     }
@@ -413,5 +466,7 @@ end:
   thd->variables.option_bits= thd_options;
   rgi->slave_close_thread_tables(thd);
   my_free(buf);
+  delete rgi;
+  rgi= thd->rgi_fake= NULL;
   DBUG_VOID_RETURN;
 }