Fixed MDEV-8428: Mangled DML statements on 2nd level slave when enabling binlog checksums

Fix was to add a test in Query_log_event::Query_log_event() if we are using
CREATE ... SELECT and in this case use trans cache, like we do on the master.
This avoid using (with doesn't have checksum)

Other things:
- Removed dummy call my_checksum(0L, NULL, 0)
- More DBUG_PRINT
- Cleaned up Log_event::need_checksum() to make it more readable (similar as in MySQL 5.6)
- Renamed variable that was hiding another one in create_table_imp()

1 files changed, 35 insertions, 19 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 69b981f3f47..a75e2137f69 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -1083,12 +1083,22 @@ my_bool Log_event::need_checksum()
      and Stop event) 
      provides their checksum alg preference through Log_event::checksum_alg.
   */
-  ret= ((checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF) ?
-        (checksum_alg != BINLOG_CHECKSUM_ALG_OFF) :
-        ((binlog_checksum_options != BINLOG_CHECKSUM_ALG_OFF) &&
-         (cache_type == Log_event::EVENT_NO_CACHE)) ?
-        MY_TEST(binlog_checksum_options) : FALSE);
-
+  if (checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF)
+    ret= (checksum_alg != BINLOG_CHECKSUM_ALG_OFF);
+  else
+  {
+    if (binlog_checksum_options != BINLOG_CHECKSUM_ALG_OFF &&
+           cache_type == Log_event::EVENT_NO_CACHE)
+    {
+      checksum_alg= binlog_checksum_options;
+      ret= MY_TEST(binlog_checksum_options);
+    }
+    else
+    {
+      ret= FALSE;
+      checksum_alg= (uint8) BINLOG_CHECKSUM_ALG_OFF;
+    }
+  }
   /*
     FD calls the methods before data_written has been calculated.
     The following invariant claims if the current is not the first
@@ -1099,10 +1109,6 @@ my_bool Log_event::need_checksum()
   DBUG_ASSERT(get_type_code() != FORMAT_DESCRIPTION_EVENT || ret ||
               data_written == 0);
 
-  if (checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF)
-    checksum_alg= ret ? // calculated value stored
-      (uint8) binlog_checksum_options : (uint8) BINLOG_CHECKSUM_ALG_OFF;
-
   DBUG_ASSERT(!ret || 
               ((checksum_alg == binlog_checksum_options ||
                /* 
@@ -1142,17 +1148,19 @@ bool Log_event::wrapper_my_b_safe_write(IO_CACHE* file, const uchar* buf, ulong
 
 bool Log_event::write_footer(IO_CACHE* file) 
 {
+  DBUG_ENTER("write_footer");
   /*
      footer contains the checksum-algorithm descriptor 
      followed by the checksum value
   */
   if (need_checksum())
   {
+    DBUG_PRINT("info", ("Writing checksum"));
     uchar buf[BINLOG_CHECKSUM_LEN];
     int4store(buf, crc);
-    return (my_b_safe_write(file, (uchar*) buf, sizeof(buf)));
+    DBUG_RETURN(my_b_safe_write(file, (uchar*) buf, sizeof(buf)));
   }
-  return 0;
+  DBUG_RETURN(0);
 }
 
 /*
@@ -1174,7 +1182,7 @@ bool Log_event::write_header(IO_CACHE* file, ulong event_data_length)
 
   if (need_checksum())
   {
-    crc= my_checksum(0L, NULL, 0);
+    crc= 0;
     data_written += BINLOG_CHECKSUM_LEN;
   }
 
@@ -3066,6 +3074,7 @@ Query_log_event::Query_log_event()
       query_arg         - array of char representing the query
       query_length      - size of the  `query_arg' array
       using_trans       - there is a modified transactional table
+      direct            - Don't cache statement
       suppress_use      - suppress the generation of 'USE' statements
       errcode           - the error code of the query
       
@@ -3184,10 +3193,17 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
     break;
 
     case SQLCOM_CREATE_TABLE:
-      trx_cache= (lex->select_lex.item_list.elements &&
-                  thd->is_current_stmt_binlog_format_row());
-      use_cache= (lex->create_info.tmp_table() &&
-                   thd->in_multi_stmt_transaction_mode()) || trx_cache;
+      /*
+        If we are using CREATE ... SELECT or if we are a slave
+        executing BEGIN...COMMIT (generated by CREATE...SELECT) we
+        have to use the transactional cache to ensure we don't
+        calculate any checksum for the CREATE part.
+      */
+      trx_cache= ((lex->select_lex.item_list.elements &&
+                  thd->is_current_stmt_binlog_format_row()) ||
+                  (thd->variables.option_bits & OPTION_GTID_BEGIN));
+      use_cache= ((lex->create_info.tmp_table() &&
+                   thd->in_multi_stmt_transaction_mode()) || trx_cache);
       break;
     case SQLCOM_SET_OPTION:
       if (lex->autocommit)
@@ -3218,8 +3234,8 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   else
     cache_type= Log_event::EVENT_STMT_CACHE;
   DBUG_ASSERT(cache_type != Log_event::EVENT_INVALID_CACHE);
-  DBUG_PRINT("info",("Query_log_event has flags2: %lu  sql_mode: %llu",
-                     (ulong) flags2, sql_mode));
+  DBUG_PRINT("info",("Query_log_event has flags2: %lu  sql_mode: %llu  cache_tye: %d",
+                     (ulong) flags2, sql_mode, cache_type));
 }
 #endif /* MYSQL_CLIENT */