MDEV-662 Lag Free Alter On Slavebb-10.8-MENT-662-tmp

This commit implements two phase binloggable ALTER.
When a new

  @@session.binlog_alter_two_phase = YES

ALTER query gets logged in two parts, the START ALTER
and the COMMIT or ROLLBACK ALTER. START Alter is written
in binlog as soon as at its handling necessary locks are
acquired for the table. The timing is such that any
concurrent DML:s that update the same table are either
committed, thus logged into binary log having done work
on the old version of the table, or will be queued for
execution on its new version.

The "COMPLETE" COMMIT or ROLLBACK ALTER are written after
the most of ALTER work is done. When its result is positive
COMMIT ALTER is written, otherwise when there were errors
since START ALTER, ROLLBACK ALTER is written. Replication
of two-phase binloggable ALTER is cross-version safe.
Specifically the OLD slave merely does not recognized the
start alter part, still memorizing its gtid.

Two phase logged ALTER is read from binlog by mysqlbinlog
to produce BINLOG 'string', where 'string' contains base64
encoded Query_log_event containing either the start part of
ALTER, or a completion part. The Query details can be
displayed with `-v` flag, similarly to ROW format events.
Notice, mysqlbinlog output containing parts of two-phase
binloggable ALTER is processable correctly only by
binlog_alter_two_phase server.

Thanks to all people involved into early discussion of the
feature including Kristian Nielsen, those who helped to
design, implement and test: Sergei Golubchik, Andrei Elkin,
Sujatha Sivakumar, Brandon Nesterenko, Alice Sherepa.

1 files changed, 38 insertions, 2 deletions

diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index bab2afb957a..a39b5b22bb6 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -20,6 +20,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 +71,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:
@@ -213,13 +215,20 @@ void mysql_client_binlog_statement(THD* thd)
   rli= thd->rli_fake;
   if (!rli && (rli= thd->rli_fake= new Relay_log_info(FALSE, "BINLOG_BASE64_EVENT")))
     rli->sql_driver_thd= thd;
+  static LEX_CSTRING connection_name= { STRING_WITH_LEN("BINLOG_BASE64_EVENT") };
+  rli->mi= new Master_info(&connection_name, false);
   if (!(rgi= thd->rgi_fake))
     rgi= thd->rgi_fake= new rpl_group_info(rli);
   rgi->thd= thd;
-
+  thd->system_thread_info.rpl_sql_info=
+    new rpl_sql_thread_info(rli->mi->rpl_filter);
   const char *error= 0;
   Log_event *ev = 0;
   my_bool is_fragmented= FALSE;
+  sql_digest_state *m_digest;
+  PSI_statement_locker *m_statement_psi;
+  LEX_CSTRING save_db;
+  my_thread_id thread_id= 0;
 
   /*
     Out of memory check
@@ -373,7 +382,30 @@ void mysql_client_binlog_statement(THD* thd)
         LEX *backup_lex;
 
         thd->backup_and_reset_current_lex(&backup_lex);
+        if (ev->get_type_code() == QUERY_EVENT)
+        {
+          m_digest= thd->m_digest;
+          m_statement_psi= thd->m_statement_psi;
+          save_db.str= my_strndup(key_memory_THD_db, thd->db.str,
+              thd->db.length, MYF(MY_WME));
+          save_db.length= thd->db.length;
+          if (save_db.str == NULL)
+          {
+            my_error(ER_OUT_OF_RESOURCES, MYF(0));
+            goto end;
+          }
+          thd->m_digest= NULL;
+          thd->m_statement_psi= NULL;
+          thread_id= thd->variables.pseudo_thread_id;
+        }
         err= ev->apply_event(rgi);
+        if (ev->get_type_code() == QUERY_EVENT)
+        {
+          thd->m_digest= m_digest;
+          thd->m_statement_psi= m_statement_psi;
+          thd->reset_db(&save_db);
+          thd->variables.pseudo_thread_id= thread_id;
+        }
         thd->restore_current_lex(backup_lex);
       }
       thd->variables.option_bits=
@@ -413,5 +445,9 @@ end:
   thd->variables.option_bits= thd_options;
   rgi->slave_close_thread_tables(thd);
   my_free(buf);
+  delete rli->mi;
+  delete thd->system_thread_info.rpl_sql_info;
+  delete rgi;
+  rgi= thd->rgi_fake= NULL;
   DBUG_VOID_RETURN;
 }