MDEV-16799: Test wsrep.variables crash at sql_class.cc:639 thd_get_ha_data

Problem was that binlog_hton was not initialized fully when needed
i.e. when wsrep_on = true.

sql/events.cc

@@ -331,6 +331,7 @@ Events::create_event(THD *thd, Event_parse_data *parse_data,
 
   if (check_access(thd, EVENT_ACL, parse_data->dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   if (check_db_dir_existence(parse_data->dbname.str))
@@ -454,6 +455,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
 
   if (check_access(thd, EVENT_ACL, parse_data->dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   if (new_dbname)                               /* It's a rename */
@@ -569,6 +571,7 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists)
 
   if (check_access(thd, EVENT_ACL, dbname.str, NULL, NULL, 0, 0))
     DBUG_RETURN(TRUE);
+
   WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
 
   /*

sql/handler.cc

@@ -4455,6 +4455,7 @@ handler::ha_create_partitioning_metadata(const char *name, const char *old_name,
               (!old_name && strcmp(name, table_share->path.str)));
 
 
+  mark_trx_read_write();
   return create_partitioning_metadata(name, old_name, action_flag);
 }
 

sql/log.cc

@@ -1707,16 +1707,8 @@ binlog_trans_log_truncate(THD *thd, my_off_t pos)
 int binlog_init(void *p)
 {
   binlog_hton= (handlerton *)p;
-#ifdef WITH_WSREP
-  if (WSREP_ON)
-    binlog_hton->state= SHOW_OPTION_YES;
-  else
-  {
-#endif /* WITH_WSREP */
-  binlog_hton->state=opt_bin_log ? SHOW_OPTION_YES : SHOW_OPTION_NO;
-#ifdef WITH_WSREP
-  }
-#endif /* WITH_WSREP */
+  binlog_hton->state= (WSREP_ON || opt_bin_log) ? SHOW_OPTION_YES
+                                                : SHOW_OPTION_NO;
   binlog_hton->db_type=DB_TYPE_BINLOG;
   binlog_hton->savepoint_offset= sizeof(my_off_t);
   binlog_hton->close_connection= binlog_close_connection;

sql/log_event.cc

@@ -7753,11 +7753,6 @@ User_var_log_event(const char* buf, uint event_len,
       we keep the flags set to UNDEF_F.
     */
     uint bytes_read= ((val + val_len) - buf_start);
-    if (bytes_read > event_len)
-    {
-      error= true;
-      goto err;
-    }
     if ((data_written - bytes_read) > 0)
     {
       flags= (uint) *(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +

sql/sql_class.cc

@@ -6689,6 +6689,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
   DBUG_ENTER("THD::binlog_query");
   DBUG_PRINT("enter", ("qtype: %s  query: '%-.*s'",
                        show_query_type(qtype), (int) query_len, query_arg));
+
 #ifdef WITH_WSREP
   DBUG_ASSERT(query_arg && (WSREP_EMULATE_BINLOG(this)
 			    || mysql_bin_log.is_open()));

sql/sql_insert.cc

@@ -4333,6 +4333,8 @@ bool select_create::send_eof()
   if (!table->s->tmp_table)
   {
 #ifdef WITH_WSREP
+    if (WSREP_ON)
+    {
       /*
          append table level exclusive key for CTAS
       */
@@ -4359,22 +4361,26 @@ bool select_create::send_eof()
       }
       /* If commit fails, we should be able to reset the OK status. */
       thd->get_stmt_da()->set_overwrite_status(TRUE);
+    }
 #endif /* WITH_WSREP */
     trans_commit_stmt(thd);
     if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
       trans_commit_implicit(thd);
 #ifdef WITH_WSREP
-    thd->get_stmt_da()->set_overwrite_status(FALSE);
+    if (WSREP_ON)
+    {
+      thd->get_stmt_da()->set_overwrite_status(FALSE);
       mysql_mutex_lock(&thd->LOCK_wsrep_thd);
       if (thd->wsrep_conflict_state != NO_CONFLICT)
       {
-        WSREP_DEBUG("select_create commit failed, thd: %lu err: %d %s", 
+        WSREP_DEBUG("select_create commit failed, thd: %lu err: %d %s",
                     thd->thread_id, thd->wsrep_conflict_state, thd->query());
         mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
         abort_result_set();
         DBUG_RETURN(true);
       }
       mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
+    }
 #endif /* WITH_WSREP */
   }
   else if (!thd->is_current_stmt_binlog_format_row())

sql/sql_parse.cc

@@ -5622,6 +5622,7 @@ case SQLCOM_PREPARE:
 #ifdef WITH_WSREP
     else if (thd->spcont &&
              (thd->wsrep_conflict_state == MUST_ABORT ||
+              thd->wsrep_conflict_state == ABORTED    ||
               thd->wsrep_conflict_state == CERT_FAILURE))
     {
       /*
@@ -7017,7 +7018,9 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
 	      com_statement_info[thd->get_command()].m_key);
       MYSQL_SET_STATEMENT_TEXT(thd->m_statement_psi, thd->query(),
 	                       thd->query_length());
+      WSREP_DEBUG("Retry autocommit query: %s", thd->query());
     }
+
     mysql_parse(thd, rawbuf, length, parser_state);
 
     if (WSREP(thd)) {
@@ -7031,6 +7034,11 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
 
       if (thd->wsrep_conflict_state == MUST_REPLAY)
       {
+        mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
+	if (thd->lex->explain)
+          delete_explain_query(thd->lex);
+        mysql_mutex_lock(&thd->LOCK_wsrep_thd);
+
         wsrep_replay_transaction(thd);
       }
 
@@ -7044,7 +7052,7 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
             thd->lex->sql_command != SQLCOM_SELECT  &&
             (thd->wsrep_retry_counter < thd->variables.wsrep_retry_autocommit))
         {
-          WSREP_DEBUG("wsrep retrying AC query: %s", 
+          WSREP_DEBUG("wsrep retrying AC query: %s",
                       (thd->query()) ? thd->query() : "void");
 
 	  /* Performance Schema Interface instrumentation, end */
@@ -7061,27 +7069,31 @@ static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
         }
         else
         {
-          WSREP_DEBUG("%s, thd: %lu is_AC: %d, retry: %lu - %lu SQL: %s", 
-                      (thd->wsrep_conflict_state == ABORTED) ? 
+          WSREP_DEBUG("%s, thd: %lu is_AC: %d, retry: %lu - %lu SQL: %s",
+                      (thd->wsrep_conflict_state == ABORTED) ?
                       "BF Aborted" : "cert failure",
-                      thd->thread_id, is_autocommit, thd->wsrep_retry_counter, 
+                      thd->thread_id, is_autocommit, thd->wsrep_retry_counter,
                       thd->variables.wsrep_retry_autocommit, thd->query());
           my_error(ER_LOCK_DEADLOCK, MYF(0), "wsrep aborted transaction");
-          thd->killed= NOT_KILLED;
           thd->wsrep_conflict_state= NO_CONFLICT;
           if (thd->wsrep_conflict_state != REPLAYING)
             thd->wsrep_retry_counter= 0;             //  reset
         }
+
+	mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
+	thd->reset_killed();
       }
       else
       {
         set_if_smaller(thd->wsrep_retry_counter, 0); // reset; eventually ok
+        mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
       }
-      mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
     }
 
     /* If retry is requested clean up explain structure */
-    if (thd->wsrep_conflict_state == RETRY_AUTOCOMMIT && thd->lex->explain)
+    if ((thd->wsrep_conflict_state == RETRY_AUTOCOMMIT ||
+	 thd->wsrep_conflict_state == MUST_REPLAY )
+	&& thd->lex->explain)
         delete_explain_query(thd->lex);
 
   }  while (thd->wsrep_conflict_state== RETRY_AUTOCOMMIT);

sql/wsrep_hton.cc

@@ -24,6 +24,8 @@
 #include <cstdlib>
 #include "debug_sync.h"
 
+extern handlerton *binlog_hton;
+extern int binlog_close_connection(handlerton *hton, THD *thd);
 extern ulonglong thd_to_trx_id(THD *thd);
 
 extern "C" int thd_binlog_format(const MYSQL_THD thd);
@@ -173,7 +175,10 @@ wsrep_close_connection(handlerton*  hton, THD* thd)
   {
     DBUG_RETURN(0);
   }
-  DBUG_RETURN(wsrep_binlog_close_connection (thd));
+
+  if (wsrep_emulate_bin_log && thd_get_ha_data(thd, binlog_hton) != NULL)
+    binlog_hton->close_connection (binlog_hton, thd);
+  DBUG_RETURN(0);
 }
 
 /*
@@ -256,9 +261,9 @@ static int wsrep_rollback(handlerton *hton, THD *thd, bool all)
   }
 
   if ((all || !thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
-      thd->wsrep_conflict_state != MUST_REPLAY)
+      thd->variables.wsrep_on && thd->wsrep_conflict_state != MUST_REPLAY)
   {
-    if (WSREP(thd) && wsrep->post_rollback(wsrep, &thd->wsrep_ws_handle))
+    if (wsrep && wsrep->post_rollback(wsrep, &thd->wsrep_ws_handle))
     {
       DBUG_PRINT("wsrep", ("setting rollback fail"));
       WSREP_ERROR("settting rollback fail: thd: %llu, schema: %s, SQL: %s",
@@ -299,12 +304,14 @@ int wsrep_commit(handlerton *hton, THD *thd, bool all)
         Transaction didn't go through wsrep->pre_commit() so just roll back
         possible changes to clean state.
       */
-      if (WSREP(thd) && wsrep->post_rollback(wsrep, &thd->wsrep_ws_handle))
-      {
-        DBUG_PRINT("wsrep", ("setting rollback fail"));
-        WSREP_ERROR("settting rollback fail: thd: %llu, schema: %s, SQL: %s",
+      if (WSREP_PROVIDER_EXISTS) {
+        if (wsrep && wsrep->post_rollback(wsrep, &thd->wsrep_ws_handle))
+        {
+          DBUG_PRINT("wsrep", ("setting rollback fail"));
+          WSREP_ERROR("settting rollback fail: thd: %llu, schema: %s, SQL: %s",
                     (long long)thd->real_id, (thd->db ? thd->db : "(null)"),
                     thd->query());
+        }
       }
       wsrep_cleanup_transaction(thd);
     }

sql/wsrep_mysqld.h

@@ -204,6 +204,9 @@ extern void wsrep_prepend_PATH (const char* path);
 extern wsrep_seqno_t wsrep_locked_seqno;
 
 #define WSREP_ON                         \
+  (global_system_variables.wsrep_on)
+
+#define WSREP_ON_NEW                     \
   ((global_system_variables.wsrep_on) && \
    wsrep_provider                     && \
    strcmp(wsrep_provider, WSREP_NONE))