pthread_handler_t handle_slave_sql(void *arg)
{
  THD *thd;                     /* needs to be first for thread_stack */
  bool thd_added= false;
  char llbuff[22],llbuff1[22];
  char saved_log_name[FN_REFLEN];
  char saved_master_log_name[FN_REFLEN];
  my_off_t saved_log_pos= 0;
  my_off_t saved_master_log_pos= 0;
  my_off_t saved_skip= 0;

  Relay_log_info* rli = ((Master_info*)arg)->rli;
  const char *errmsg;
  bool mts_inited= false;
  Global_THD_manager *thd_manager= Global_THD_manager::get_instance();

  // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
  my_thread_init();
  DBUG_ENTER("handle_slave_sql");

  DBUG_ASSERT(rli->inited);
  mysql_mutex_lock(&rli->run_lock);
  DBUG_ASSERT(!rli->slave_running);
  errmsg= 0;
#ifndef DBUG_OFF
  rli->events_until_exit = abort_slave_event_count;
#endif

  thd = new THD; // note that contructor of THD uses DBUG_ !
  thd->thread_stack = (char*)&thd; // remember where our stack is
  mysql_mutex_lock(&rli->info_thd_lock);
  rli->info_thd= thd;

  #ifdef HAVE_PSI_INTERFACE
  // save the instrumentation for SQL thread in rli->info_thd->scheduler
  struct PSI_thread *psi= PSI_THREAD_CALL(get_thread)();
  thd_set_psi(rli->info_thd, psi);
  #endif

 /*
  Create Mts Submode.
  It is possible that we may not have deleted the last MTS submode in case
  terminate_slave_threads() returned with ER_STOP_SLAVE_SQL_THREAD_TIMEOUT
  while stopping the slave in the previous slave session.
 */
 if (rli->current_mts_submode)
   delete rli->current_mts_submode;

 if (mts_parallel_option != MTS_PARALLEL_TYPE_DB_NAME)
   rli->current_mts_submode= new Mts_submode_logical_clock();
 else
   rli->current_mts_submode= new Mts_submode_database();

  mysql_mutex_unlock(&rli->info_thd_lock);

  /* Inform waiting threads that slave has started */
  rli->slave_run_id++;
  rli->slave_running = 1;
  rli->reported_unsafe_warning= false;

  pthread_detach_this_thread();
  if (init_slave_thread(thd, SLAVE_THD_SQL))
  {
    /*
      TODO: this is currently broken - slave start and change master
      will be stuck if we fail here
    */
    mysql_cond_broadcast(&rli->start_cond);
    mysql_mutex_unlock(&rli->run_lock);
    rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR),
                "Failed during slave thread initialization");
    goto err;
  }
  thd->init_for_queries(rli);
  thd->temporary_tables = rli->save_temporary_tables; // restore temp tables
  set_thd_in_use_temporary_tables(rli);   // (re)set sql_thd in use for saved temp tables

  thd_manager->add_thd(thd);
  thd_added= true;

  /* MTS: starting the worker pool */
  if (slave_start_workers(rli, rli->opt_slave_parallel_workers, &mts_inited) != 0)
  {
    mysql_cond_broadcast(&rli->start_cond);
    mysql_mutex_unlock(&rli->run_lock);
    rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR),
                "Failed during slave workers initialization");
    goto err;
  }
  /*
    We are going to set slave_running to 1. Assuming slave I/O thread is
    alive and connected, this is going to make Seconds_Behind_Master be 0
    i.e. "caught up". Even if we're just at start of thread. Well it's ok, at
    the moment we start we can think we are caught up, and the next second we
    start receiving data so we realize we are not caught up and
    Seconds_Behind_Master grows. No big deal.
  */
  rli->abort_slave = 0;

  /*
    Reset errors for a clean start (otherwise, if the master is idle, the SQL
    thread may execute no Query_log_event, so the error will remain even
    though there's no problem anymore). Do not reset the master timestamp
    (imagine the slave has caught everything, the STOP SLAVE and START SLAVE:
    as we are not sure that we are going to receive a query, we want to
    remember the last master timestamp (to say how many seconds behind we are
    now.
    But the master timestamp is reset by RESET SLAVE & CHANGE MASTER.
  */
  rli->clear_error();
  if (rli->workers_array_initialized)
  {
    for(uint i= 0; i<rli->get_worker_count(); i++)
    {
      rli->get_worker(i)->clear_error();
    }
  }

  if (rli->update_is_transactional())
  {
    mysql_cond_broadcast(&rli->start_cond);
    mysql_mutex_unlock(&rli->run_lock);
    rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR),
                "Error checking if the relay log repository is transactional.");
    goto err;
  }

  if (!rli->is_transactional())
    rli->report(WARNING_LEVEL, 0,
    "If a crash happens this configuration does not guarantee that the relay "
    "log info will be consistent");

  mysql_mutex_unlock(&rli->run_lock);
  mysql_cond_broadcast(&rli->start_cond);

  DEBUG_SYNC(thd, "after_start_slave");

  //tell the I/O thread to take relay_log_space_limit into account from now on
  mysql_mutex_lock(&rli->log_space_lock);
  rli->ignore_log_space_limit= 0;
  mysql_mutex_unlock(&rli->log_space_lock);
  rli->trans_retries= 0; // start from "no error"
  DBUG_PRINT("info", ("rli->trans_retries: %lu", rli->trans_retries));

  if (rli->init_relay_log_pos(rli->get_group_relay_log_name(),
                              rli->get_group_relay_log_pos(),
                              true/*need_data_lock=true*/, &errmsg,
                              1 /*look for a description_event*/))
  { 
    rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, 
                "Error initializing relay log position: %s", errmsg);
    goto err;
  }
  THD_CHECK_SENTRY(thd);
#ifndef DBUG_OFF
  {
    char llbuf1[22], llbuf2[22];
    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
                        llstr(my_b_tell(rli->cur_log),llbuf1),
                        llstr(rli->get_event_relay_log_pos(),llbuf2)));
    DBUG_ASSERT(rli->get_event_relay_log_pos() >= BIN_LOG_HEADER_SIZE);
    /*
      Wonder if this is correct. I (Guilhem) wonder if my_b_tell() returns the
      correct position when it's called just after my_b_seek() (the questionable
      stuff is those "seek is done on next read" comments in the my_b_seek()
      source code).
      The crude reality is that this assertion randomly fails whereas
      replication seems to work fine. And there is no easy explanation why it
      fails (as we my_b_seek(rli->event_relay_log_pos) at the very end of
      init_relay_log_pos() called above). Maybe the assertion would be
      meaningful if we held rli->data_lock between the my_b_seek() and the
      DBUG_ASSERT().

      DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->get_event_relay_log_pos());
    */
  }
#endif
  DBUG_ASSERT(rli->info_thd == thd);

#ifdef WITH_NDBCLUSTER_STORAGE_ENGINE
  /* engine specific hook, to be made generic */
  if (ndb_wait_setup_func && ndb_wait_setup_func(opt_ndb_wait_setup))
  {
    sql_print_warning("Slave SQL thread : NDB : Tables not available after %lu"
                      " seconds.  Consider increasing --ndb-wait-setup value",
                      opt_ndb_wait_setup);
  }
#endif

  DBUG_PRINT("master_info",("log_file_name: %s  position: %s",
                            rli->get_group_master_log_name(),
                            llstr(rli->get_group_master_log_pos(),llbuff)));
  sql_print_information("Slave SQL thread initialized, starting replication in \
log '%s' at position %s, relay log '%s' position: %s", rli->get_rpl_log_name(),
                        llstr(rli->get_group_master_log_pos(),llbuff),rli->get_group_relay_log_name(),
                        llstr(rli->get_group_relay_log_pos(),llbuff1));

  if (check_temp_dir(rli->slave_patternload_file))
  {
    rli->report(ERROR_LEVEL, thd->get_stmt_da()->mysql_errno(),
                "Unable to use slave's temporary directory %s - %s", 
                slave_load_tmpdir, thd->get_stmt_da()->message_text());
    goto err;
  }

  /* execute init_slave variable */
  if (opt_init_slave.length)
  {
    execute_init_command(thd, &opt_init_slave, &LOCK_sys_init_slave);
    if (thd->is_slave_error)
    {
      rli->report(ERROR_LEVEL, thd->get_stmt_da()->mysql_errno(),
                  "Slave SQL thread aborted. Can't execute init_slave query,"
                  "'%s'", thd->get_stmt_da()->message_text());
      goto err;
    }
  }

  /*
    First check until condition - probably there is nothing to execute. We
    do not want to wait for next event in this case.
  */
  mysql_mutex_lock(&rli->data_lock);
  if (rli->slave_skip_counter)
  {
    strmake(saved_log_name, rli->get_group_relay_log_name(), FN_REFLEN - 1);
    strmake(saved_master_log_name, rli->get_group_master_log_name(), FN_REFLEN - 1);
    saved_log_pos= rli->get_group_relay_log_pos();
    saved_master_log_pos= rli->get_group_master_log_pos();
    saved_skip= rli->slave_skip_counter;
  }
  if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
      rli->is_until_satisfied(thd, NULL))
  {
    mysql_mutex_unlock(&rli->data_lock);
    goto err;
  }
  mysql_mutex_unlock(&rli->data_lock);

  /* Read queries from the IO/THREAD until this thread is killed */

  while (!sql_slave_killed(thd,rli))
  {
    THD_STAGE_INFO(thd, stage_reading_event_from_the_relay_log);
    DBUG_ASSERT(rli->info_thd == thd);
    THD_CHECK_SENTRY(thd);

    if (saved_skip && rli->slave_skip_counter == 0)
    {
      sql_print_information("'SQL_SLAVE_SKIP_COUNTER=%ld' executed at "
        "relay_log_file='%s', relay_log_pos='%ld', master_log_name='%s', "
        "master_log_pos='%ld' and new position at "
        "relay_log_file='%s', relay_log_pos='%ld', master_log_name='%s', "
        "master_log_pos='%ld' ",
        (ulong) saved_skip, saved_log_name, (ulong) saved_log_pos,
        saved_master_log_name, (ulong) saved_master_log_pos,
        rli->get_group_relay_log_name(), (ulong) rli->get_group_relay_log_pos(),
        rli->get_group_master_log_name(), (ulong) rli->get_group_master_log_pos());
      saved_skip= 0;
    }
    
    if (exec_relay_log_event(thd,rli))
    {
      DBUG_PRINT("info", ("exec_relay_log_event() failed"));
      // do not scare the user if SQL thread was simply killed or stopped
      if (!sql_slave_killed(thd,rli))
      {
        /*
          retrieve as much info as possible from the thd and, error
          codes and warnings and print this to the error log as to
          allow the user to locate the error
        */
        uint32 const last_errno= rli->last_error().number;

        if (thd->is_error())
        {
          char const *const errmsg= thd->get_stmt_da()->message_text();

          DBUG_PRINT("info",
                     ("thd->get_stmt_da()->get_mysql_errno()=%d; "
                      "rli->last_error.number=%d",
                      thd->get_stmt_da()->mysql_errno(), last_errno));
          if (last_errno == 0)
          {
            /*
 	      This function is reporting an error which was not reported
 	      while executing exec_relay_log_event().
 	    */ 
            rli->report(ERROR_LEVEL, thd->get_stmt_da()->mysql_errno(),
                        "%s", errmsg);
          }
          else if (last_errno != thd->get_stmt_da()->mysql_errno())
          {
            /*
             * An error was reported while executing exec_relay_log_event()
             * however the error code differs from what is in the thread.
             * This function prints out more information to help finding
             * what caused the problem.
             */  
            sql_print_error("Slave (additional info): %s Error_code: %d",
                            errmsg, thd->get_stmt_da()->mysql_errno());
          }
        }

        /* Print any warnings issued */
        Diagnostics_area::Sql_condition_iterator it=
          thd->get_stmt_da()->sql_conditions();
        const Sql_condition *err;
        /*
          Added controlled slave thread cancel for replication
          of user-defined variables.
        */
        bool udf_error = false;
        while ((err= it++))
        {
          if (err->mysql_errno() == ER_CANT_OPEN_LIBRARY)
            udf_error = true;
          sql_print_warning("Slave: %s Error_code: %d",
                            err->message_text(), err->mysql_errno());
        }
        if (udf_error)
          sql_print_error("Error loading user-defined library, slave SQL "
            "thread aborted. Install the missing library, and restart the "
            "slave SQL thread with \"SLAVE START\". We stopped at log '%s' "
            "position %s", rli->get_rpl_log_name(),
            llstr(rli->get_group_master_log_pos(), llbuff));
        else
          sql_print_error("\
Error running query, slave SQL thread aborted. Fix the problem, and restart \
the slave SQL thread with \"SLAVE START\". We stopped at log \
'%s' position %s", rli->get_rpl_log_name(),
llstr(rli->get_group_master_log_pos(), llbuff));
      }
      goto err;
    }
  }

  /* Thread stopped. Print the current replication position to the log */
  sql_print_information("Slave SQL thread exiting, replication stopped in log "
                        "'%s' at position %s",
                        rli->get_rpl_log_name(),
                        llstr(rli->get_group_master_log_pos(), llbuff));

 err:

  slave_stop_workers(rli, &mts_inited); // stopping worker pool
  if (rli->recovery_groups_inited)
  {
    bitmap_free(&rli->recovery_groups);
    rli->mts_recovery_group_cnt= 0;
    rli->recovery_groups_inited= false;
  }

  /*
    Some events set some playgrounds, which won't be cleared because thread
    stops. Stopping of this thread may not be known to these events ("stop"
    request is detected only by the present function, not by events), so we
    must "proactively" clear playgrounds:
  */
  thd->clear_error();
  rli->cleanup_context(thd, 1);
  /*
    Some extra safety, which should not been needed (normally, event deletion
    should already have done these assignments (each event which sets these
    variables is supposed to set them to 0 before terminating)).
  */
  thd->catalog= 0;
  thd->reset_query();
  thd->reset_db(NULL, 0);

  THD_STAGE_INFO(thd, stage_waiting_for_slave_mutex_on_exit);
  mysql_mutex_lock(&rli->run_lock);
  /* We need data_lock, at least to wake up any waiting master_pos_wait() */
  mysql_mutex_lock(&rli->data_lock);
  DBUG_ASSERT(rli->slave_running == 1); // tracking buffer overrun
  /* When master_pos_wait() wakes up it will check this and terminate */
  rli->slave_running= 0;
  /* Forget the relay log's format */
  rli->set_rli_description_event(NULL);
  /* Wake up master_pos_wait() */
  mysql_mutex_unlock(&rli->data_lock);
  DBUG_PRINT("info",("Signaling possibly waiting master_pos_wait() functions"));
  mysql_cond_broadcast(&rli->data_cond);
  rli->ignore_log_space_limit= 0; /* don't need any lock */
  /* we die so won't remember charset - re-update them on next thread start */
  rli->cached_charset_invalidate();
  rli->save_temporary_tables = thd->temporary_tables;

  /*
    TODO: see if we can do this conditionally in next_event() instead
    to avoid unneeded position re-init
  */
  thd->temporary_tables = 0; // remove tempation from destructor to close them
  DBUG_ASSERT(thd->net.buff != 0);
  net_end(&thd->net); // destructor will not free it, because we are weird
  DBUG_ASSERT(rli->info_thd == thd);
  THD_CHECK_SENTRY(thd);
  mysql_mutex_lock(&rli->info_thd_lock);
  rli->info_thd= NULL;
  mysql_mutex_unlock(&rli->info_thd_lock);
  set_thd_in_use_temporary_tables(rli);  // (re)set info_thd in use for saved temp tables

  thd->release_resources();
  THD_CHECK_SENTRY(thd);
  if (thd_added)
    thd_manager->remove_thd(thd);

  /*
    The thd can only be destructed after indirect references
    through mi->rli->info_thd are cleared: mi->rli->info_thd= NULL.

    For instance, user thread might be issuing show_slave_status
    and attempting to read mi->rli->info_thd->get_proc_info().
    Therefore thd must only be deleted after info_thd is set
    to NULL.
  */
  delete thd;

 /*
  Note: the order of the broadcast and unlock calls below (first broadcast, then unlock)
  is important. Otherwise a killer_thread can execute between the calls and
  delete the mi structure leading to a crash! (see BUG#25306 for details)
 */ 
  mysql_cond_broadcast(&rli->stop_cond);
  DBUG_EXECUTE_IF("simulate_slave_delay_at_terminate_bug38694", sleep(5););
  mysql_mutex_unlock(&rli->run_lock);  // tell the world we are done

  DBUG_LEAVE;                            // Must match DBUG_ENTER()
  my_thread_end();
  ERR_remove_state(0);
  pthread_exit(0);
  return 0;                             // Avoid compiler warnings
}