mysql_execute_command(THD *thd)
{
  int res= 0;
  int  up_result= 0;
  LEX  *lex= thd->lex;
  /* first SELECT_LEX (have special meaning for many of non-SELECTcommands) */
  SELECT_LEX *select_lex= lex->first_select_lex();
  /* first table of first SELECT_LEX */
  TABLE_LIST *first_table= select_lex->table_list.first;
  /* list of all tables in query */
  TABLE_LIST *all_tables;
  /* most outer SELECT_LEX_UNIT of query */
  SELECT_LEX_UNIT *unit= &lex->unit;
#ifdef HAVE_REPLICATION
  /* have table map for update for multi-update statement (BUG#37051) */
  bool have_table_map_for_update= FALSE;
  /* */
  Rpl_filter *rpl_filter;
#endif
  DBUG_ENTER("mysql_execute_command");

  // check that we correctly marked first table for data insertion
  DBUG_ASSERT(!(sql_command_flags[lex->sql_command] & CF_INSERTS_DATA) ||
              first_table->for_insert_data);

  if (thd->security_ctx->password_expired &&
      lex->sql_command != SQLCOM_SET_OPTION &&
      lex->sql_command != SQLCOM_PREPARE &&
      lex->sql_command != SQLCOM_EXECUTE &&
      lex->sql_command != SQLCOM_DEALLOCATE_PREPARE)
  {
    my_error(ER_MUST_CHANGE_PASSWORD, MYF(0));
    DBUG_RETURN(1);
  }

  DBUG_ASSERT(thd->transaction.stmt.is_empty() || thd->in_sub_stmt);
  /*
    Each statement or replication event which might produce deadlock
    should handle transaction rollback on its own. So by the start of
    the next statement transaction rollback request should be fulfilled
    already.
  */
  DBUG_ASSERT(! thd->transaction_rollback_request || thd->in_sub_stmt);
  /*
    In many cases first table of main SELECT_LEX have special meaning =>
    check that it is first table in global list and relink it first in 
    queries_tables list if it is necessary (we need such relinking only
    for queries with subqueries in select list, in this case tables of
    subqueries will go to global list first)

    all_tables will differ from first_table only if most upper SELECT_LEX
    do not contain tables.

    Because of above in place where should be at least one table in most
    outer SELECT_LEX we have following check:
    DBUG_ASSERT(first_table == all_tables);
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
  */
  lex->first_lists_tables_same();
  lex->fix_first_select_number();
  /* should be assigned after making first tables same */
  all_tables= lex->query_tables;
  /* set context for commands which do not use setup_tables */
  select_lex->
    context.resolve_in_table_list_only(select_lex->
                                       table_list.first);

  /*
    Remember last commmand executed, so that we can use it in functions called by
    dispatch_command()
  */
  thd->last_sql_command= lex->sql_command;

  /*
    Reset warning count for each query that uses tables
    A better approach would be to reset this for any commands
    that is not a SHOW command or a select that only access local
    variables, but for now this is probably good enough.
  */
  if ((sql_command_flags[lex->sql_command] & CF_DIAGNOSTIC_STMT) != 0)
    thd->get_stmt_da()->set_warning_info_read_only(TRUE);
  else
  {
    thd->get_stmt_da()->set_warning_info_read_only(FALSE);
    if (all_tables)
      thd->get_stmt_da()->opt_clear_warning_info(thd->query_id);
  }

#ifdef HAVE_REPLICATION
  if (unlikely(thd->slave_thread))
  {
    if (lex->sql_command == SQLCOM_DROP_TRIGGER)
    {
      /*
        When dropping a trigger, we need to load its table name
        before checking slave filter rules.
      */
      add_table_for_trigger(thd, thd->lex->spname, 1, &all_tables);
      
      if (!all_tables)
      {
        /*
          If table name cannot be loaded,
          it means the trigger does not exists possibly because
          CREATE TRIGGER was previously skipped for this trigger
          according to slave filtering rules.
          Returning success without producing any errors in this case.
        */
        if (!thd->lex->create_info.if_exists())
          DBUG_RETURN(0);
        /*
          DROP TRIGGER IF NOT EXISTS will return without an error later
          after possibly writing the query to a binlog
        */
      }
      else // force searching in slave.cc:tables_ok()
        all_tables->updating= 1;
    }

    /*
      For fix of BUG#37051, the master stores the table map for update
      in the Query_log_event, and the value is assigned to
      thd->variables.table_map_for_update before executing the update
      query.

      If thd->variables.table_map_for_update is set, then we are
      replicating from a new master, we can use this value to apply
      filter rules without opening all the tables. However If
      thd->variables.table_map_for_update is not set, then we are
      replicating from an old master, so we just skip this and
      continue with the old method. And of course, the bug would still
      exist for old masters.
    */
    if (lex->sql_command == SQLCOM_UPDATE_MULTI &&
        thd->table_map_for_update)
    {
      have_table_map_for_update= TRUE;
      table_map table_map_for_update= thd->table_map_for_update;
      uint nr= 0;
      TABLE_LIST *table;
      for (table=all_tables; table; table=table->next_global, nr++)
      {
        if (table_map_for_update & ((table_map)1 << nr))
          table->updating= TRUE;
        else
          table->updating= FALSE;
      }

      if (all_tables_not_ok(thd, all_tables))
      {
        /* we warn the slave SQL thread */
        my_message(ER_SLAVE_IGNORED_TABLE, ER_THD(thd, ER_SLAVE_IGNORED_TABLE),
                   MYF(0));
      }
    }
    
    /*
      Check if statment should be skipped because of slave filtering
      rules

      Exceptions are:
      - UPDATE MULTI: For this statement, we want to check the filtering
        rules later in the code
      - SET: we always execute it (Not that many SET commands exists in
        the binary log anyway -- only 4.1 masters write SET statements,
	in 5.0 there are no SET statements in the binary log)
      - DROP TEMPORARY TABLE IF EXISTS: we always execute it (otherwise we
        have stale files on slave caused by exclusion of one tmp table).
    */
    if (!(lex->sql_command == SQLCOM_UPDATE_MULTI) &&
	!(lex->sql_command == SQLCOM_SET_OPTION) &&
	!((lex->sql_command == SQLCOM_DROP_TABLE ||
           lex->sql_command == SQLCOM_DROP_SEQUENCE) &&
          lex->tmp_table() && lex->if_exists()) &&
        all_tables_not_ok(thd, all_tables))
    {
      /* we warn the slave SQL thread */
      my_message(ER_SLAVE_IGNORED_TABLE, ER_THD(thd, ER_SLAVE_IGNORED_TABLE),
                 MYF(0));
      DBUG_RETURN(0);
    }
    /* 
       Execute deferred events first
    */
    if (slave_execute_deferred_events(thd))
      DBUG_RETURN(-1);
  }
  else
  {
#endif /* HAVE_REPLICATION */
    /*
      When option readonly is set deny operations which change non-temporary
      tables. Except for the replication thread and the 'super' users.
    */
    if (deny_updates_if_read_only_option(thd, all_tables))
    {
      my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
      DBUG_RETURN(-1);
    }
#ifdef HAVE_REPLICATION
  } /* endif unlikely slave */
#endif
  Opt_trace_start ots(thd, all_tables, lex->sql_command, &lex->var_list,
                      thd->query(), thd->query_length(),
                      thd->variables.character_set_client);

  Json_writer_object trace_command(thd);
  Json_writer_array trace_command_steps(thd, "steps");

  /* store old value of binlog format */
  enum_binlog_format orig_binlog_format,orig_current_stmt_binlog_format;

  thd->get_binlog_format(&orig_binlog_format,
                         &orig_current_stmt_binlog_format);
#ifdef WITH_WSREP
  if (WSREP(thd))
  {
    /*
      change LOCK TABLE WRITE to transaction
    */
    if (lex->sql_command== SQLCOM_LOCK_TABLES && wsrep_convert_LOCK_to_trx)
    {
      for (TABLE_LIST *table= all_tables; table; table= table->next_global)
      {
	if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
        {
	  lex->sql_command= SQLCOM_BEGIN;
	  thd->wsrep_converted_lock_session= true;
	  break;
	}
      }
    }
    if (lex->sql_command== SQLCOM_UNLOCK_TABLES &&
	thd->wsrep_converted_lock_session)
    {
      thd->wsrep_converted_lock_session= false;
      lex->sql_command= SQLCOM_COMMIT;
      lex->tx_release= TVL_NO;
    }

    /*
     * Bail out if DB snapshot has not been installed. We however,
     * allow SET and SHOW queries and reads from information schema
     * and dirty reads (if configured)
     */
    if (!(thd->wsrep_applier) &&
        !(wsrep_ready_get() && wsrep_reject_queries == WSREP_REJECT_NONE)  &&
        !(thd->variables.wsrep_dirty_reads &&
          (sql_command_flags[lex->sql_command] & CF_CHANGES_DATA) == 0)    &&
        !wsrep_tables_accessible_when_detached(all_tables)                 &&
        lex->sql_command != SQLCOM_SET_OPTION                              &&
        !wsrep_is_show_query(lex->sql_command))
    {
      my_message(ER_UNKNOWN_COM_ERROR,
                 "WSREP has not yet prepared node for application use", MYF(0));
      goto error;
    }
  }
#endif /* WITH_WSREP */
  status_var_increment(thd->status_var.com_stat[lex->sql_command]);
  thd->progress.report_to_client= MY_TEST(sql_command_flags[lex->sql_command] &
                                          CF_REPORT_PROGRESS);

  DBUG_ASSERT(thd->transaction.stmt.modified_non_trans_table == FALSE);

  /*
    Assign system variables with values specified by the clause
    SET STATEMENT var1=value1 [, var2=value2, ...] FOR <statement>
    if they are any.
  */
  if (run_set_statement_if_requested(thd, lex))
    goto error;

  if (thd->lex->mi.connection_name.str == NULL)
      thd->lex->mi.connection_name= thd->variables.default_master_connection;

  /*
    Force statement logging for DDL commands to allow us to update
    privilege, system or statistic tables directly without the updates
    getting logged.
  */
  if (!(sql_command_flags[lex->sql_command] &
        (CF_CAN_GENERATE_ROW_EVENTS | CF_FORCE_ORIGINAL_BINLOG_FORMAT |
         CF_STATUS_COMMAND)))
    thd->set_binlog_format_stmt();

  /*
    End a active transaction so that this command will have it's
    own transaction and will also sync the binary log. If a DDL is
    not run in it's own transaction it may simply never appear on
    the slave in case the outside transaction rolls back.
  */
  if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_BEGIN))
  {
    /*
      Note that this should never happen inside of stored functions
      or triggers as all such statements prohibited there.
    */
    DBUG_ASSERT(! thd->in_sub_stmt);
    /* Statement transaction still should not be started. */
    DBUG_ASSERT(thd->transaction.stmt.is_empty());
    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
    {
      /* Commit the normal transaction if one is active. */
      bool commit_failed= trans_commit_implicit(thd);
      /* Release metadata locks acquired in this transaction. */
      thd->release_transactional_locks();
      if (commit_failed)
      {
        WSREP_DEBUG("implicit commit failed, MDL released: %lld",
                    (longlong) thd->thread_id);
        goto error;
      }
    }
    thd->transaction.stmt.mark_trans_did_ddl();
#ifdef WITH_WSREP
    /* Clean up the previous transaction on implicit commit */
    if (WSREP_NNULL(thd) && wsrep_thd_is_local(thd) &&
        wsrep_after_statement(thd))
    {
      goto error;
    }
#endif /* WITH_WSREP */
  }

#ifndef DBUG_OFF
  if (lex->sql_command != SQLCOM_SET_OPTION)
    DEBUG_SYNC(thd,"before_execute_sql_command");
#endif

  /*
    Check if we are in a read-only transaction and we're trying to
    execute a statement which should always be disallowed in such cases.

    Note that this check is done after any implicit commits.
  */
  if (thd->tx_read_only &&
      (sql_command_flags[lex->sql_command] & CF_DISALLOW_IN_RO_TRANS))
  {
    my_error(ER_CANT_EXECUTE_IN_READ_ONLY_TRANSACTION, MYF(0));
    goto error;
  }

  /*
    Close tables open by HANDLERs before executing DDL statement
    which is going to affect those tables.

    This should happen before temporary tables are pre-opened as
    otherwise we will get errors about attempt to re-open tables
    if table to be changed is open through HANDLER.

    Note that even although this is done before any privilege
    checks there is no security problem here as closing open
    HANDLER doesn't require any privileges anyway.
  */
  if (sql_command_flags[lex->sql_command] & CF_HA_CLOSE)
    mysql_ha_rm_tables(thd, all_tables);

  /*
    Pre-open temporary tables to simplify privilege checking
    for statements which need this.
  */
  if (sql_command_flags[lex->sql_command] & CF_PREOPEN_TMP_TABLES)
  {
    if (thd->open_temporary_tables(all_tables))
      goto error;
  }

  if (sql_command_flags[lex->sql_command] & CF_STATUS_COMMAND)
    thd->query_plan_flags|= QPLAN_STATUS;
  if (sql_command_flags[lex->sql_command] & CF_ADMIN_COMMAND)
    thd->query_plan_flags|= QPLAN_ADMIN;

  /* Start timeouts */
  thd->set_query_timer();

#ifdef WITH_WSREP
  /*
    Always start a new transaction for a wsrep THD unless the
    current command is DDL or explicit BEGIN. This will guarantee that
    the THD is BF abortable even if it does not generate any
    changes and takes only read locks. If the statement does not
    start a multi STMT transaction, the wsrep_transaction is
    committed as empty at the end of this function.

    Transaction is started for BEGIN in trans_begin(), for DDL the
    implicit commit took care of committing previous transaction
    above and a new transaction should not be started.

    Do not start transaction for stored procedures, it will be handled
    internally in SP processing.
  */
  if (WSREP_NNULL(thd)                    &&
      wsrep_thd_is_local(thd)             &&
      lex->sql_command != SQLCOM_BEGIN    &&
      lex->sql_command != SQLCOM_CALL     &&
      lex->sql_command != SQLCOM_EXECUTE  &&
      lex->sql_command != SQLCOM_EXECUTE_IMMEDIATE &&
      !(sql_command_flags[lex->sql_command] & CF_AUTO_COMMIT_TRANS))
  {
    wsrep_start_trx_if_not_started(thd);
  }
#endif /* WITH_WSREP */

  switch (lex->sql_command) {

  case SQLCOM_SHOW_EVENTS:
#ifndef HAVE_EVENT_SCHEDULER
    my_error(ER_NOT_SUPPORTED_YET, MYF(0), "embedded server");
    break;
#endif
  case SQLCOM_SHOW_STATUS:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
    execute_show_status(thd, all_tables);

    break;
  }
  case SQLCOM_SHOW_EXPLAIN:
  {
    if (!thd->security_ctx->priv_user[0] &&
        check_global_access(thd,PROCESS_ACL))
      break;

    /*
      The select should use only one table, it's the SHOW EXPLAIN pseudo-table
    */
    if (lex->sroutines.records || lex->query_tables->next_global)
    {
      my_message(ER_SET_CONSTANTS_ONLY, ER_THD(thd, ER_SET_CONSTANTS_ONLY),
		 MYF(0));
      goto error;
    }

    Item **it= lex->value_list.head_ref();
    if (!(*it)->basic_const_item() ||
        (*it)->fix_fields_if_needed_for_scalar(lex->thd, it))
    {
      my_message(ER_SET_CONSTANTS_ONLY, ER_THD(thd, ER_SET_CONSTANTS_ONLY),
		 MYF(0));
      goto error;
    }
  }
    /* fall through */
  case SQLCOM_SHOW_STATUS_PROC:
  case SQLCOM_SHOW_STATUS_FUNC:
  case SQLCOM_SHOW_STATUS_PACKAGE:
  case SQLCOM_SHOW_STATUS_PACKAGE_BODY:
  case SQLCOM_SHOW_DATABASES:
  case SQLCOM_SHOW_TABLES:
  case SQLCOM_SHOW_TRIGGERS:
  case SQLCOM_SHOW_TABLE_STATUS:
  case SQLCOM_SHOW_OPEN_TABLES:
  case SQLCOM_SHOW_GENERIC:
  case SQLCOM_SHOW_PLUGINS:
  case SQLCOM_SHOW_FIELDS:
  case SQLCOM_SHOW_KEYS:
  case SQLCOM_SHOW_VARIABLES:
  case SQLCOM_SHOW_CHARSETS:
  case SQLCOM_SHOW_COLLATIONS:
  case SQLCOM_SHOW_STORAGE_ENGINES:
  case SQLCOM_SHOW_PROFILE:
  case SQLCOM_SELECT:
  {
#ifdef WITH_WSREP
    if (lex->sql_command == SQLCOM_SELECT)
    {
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_READ);
    }
    else
    {
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
# ifdef ENABLED_PROFILING
      if (lex->sql_command == SQLCOM_SHOW_PROFILE)
        thd->profiling.discard_current_query();
# endif
    }
#endif /* WITH_WSREP */

    thd->status_var.last_query_cost= 0.0;

    /*
      lex->exchange != NULL implies SELECT .. INTO OUTFILE and this
      requires FILE_ACL access.
    */
    ulong privileges_requested= lex->exchange ? SELECT_ACL | FILE_ACL :
      SELECT_ACL;

    if (all_tables)
      res= check_table_access(thd,
                              privileges_requested,
                              all_tables, FALSE, UINT_MAX, FALSE);
    else
      res= check_access(thd, privileges_requested, any_db, NULL, NULL, 0, 0);

    if (!res)
      res= execute_sqlcom_select(thd, all_tables);

    break;
  }
  case SQLCOM_EXECUTE_IMMEDIATE:
  {
    mysql_sql_stmt_execute_immediate(thd);
    break;
  }
  case SQLCOM_PREPARE:
  {
    mysql_sql_stmt_prepare(thd);
    break;
  }
  case SQLCOM_EXECUTE:
  {
    mysql_sql_stmt_execute(thd);
    break;
  }
  case SQLCOM_DEALLOCATE_PREPARE:
  {
    mysql_sql_stmt_close(thd);
    break;
  }
  case SQLCOM_DO:
    if (check_table_access(thd, SELECT_ACL, all_tables, FALSE, UINT_MAX, FALSE)
        || open_and_lock_tables(thd, all_tables, TRUE, 0))
      goto error;

    res= mysql_do(thd, *lex->insert_list);
    break;

  case SQLCOM_EMPTY_QUERY:
    my_ok(thd);
    break;

  case SQLCOM_HELP:
    res= mysqld_help(thd,lex->help_arg);
    break;

#ifndef EMBEDDED_LIBRARY
  case SQLCOM_PURGE:
  {
    if (check_global_access(thd, SUPER_ACL))
      goto error;
    /* PURGE MASTER LOGS TO 'file' */
    res = purge_master_logs(thd, lex->to_log);
    break;
  }
  case SQLCOM_PURGE_BEFORE:
  {
    Item *it;

    if (check_global_access(thd, SUPER_ACL))
      goto error;
    /* PURGE MASTER LOGS BEFORE 'data' */
    it= (Item *)lex->value_list.head();
    if (it->fix_fields_if_needed_for_scalar(lex->thd, &it))
    {
      my_error(ER_WRONG_ARGUMENTS, MYF(0), "PURGE LOGS BEFORE");
      goto error;
    }
    it= new (thd->mem_root) Item_func_unix_timestamp(thd, it);
    it->fix_fields(thd, &it);
    res = purge_master_logs_before_date(thd, (ulong)it->val_int());
    break;
  }
#endif
  case SQLCOM_SHOW_WARNS:
  {
    res= mysqld_show_warnings(thd, (ulong)
			      ((1L << (uint) Sql_condition::WARN_LEVEL_NOTE) |
			       (1L << (uint) Sql_condition::WARN_LEVEL_WARN) |
			       (1L << (uint) Sql_condition::WARN_LEVEL_ERROR)
			       ));
    break;
  }
  case SQLCOM_SHOW_ERRORS:
  {
    res= mysqld_show_warnings(thd, (ulong)
			      (1L << (uint) Sql_condition::WARN_LEVEL_ERROR));
    break;
  }
  case SQLCOM_SHOW_PROFILES:
  {
#if defined(ENABLED_PROFILING)
    thd->profiling.discard_current_query();
    res= thd->profiling.show_profiles();
    if (res)
      goto error;
#else
    my_error(ER_FEATURE_DISABLED, MYF(0), "SHOW PROFILES", "enable-profiling");
    goto error;
#endif
    break;
  }

#ifdef HAVE_REPLICATION
  case SQLCOM_SHOW_SLAVE_HOSTS:
  {
    if (check_global_access(thd, REPL_SLAVE_ACL))
      goto error;
    res = show_slave_hosts(thd);
    break;
  }
  case SQLCOM_SHOW_RELAYLOG_EVENTS: /* fall through */
  case SQLCOM_SHOW_BINLOG_EVENTS:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
    if (check_global_access(thd, REPL_SLAVE_ACL))
      goto error;
    res = mysql_show_binlog_events(thd);
    break;
  }
#endif

  case SQLCOM_ASSIGN_TO_KEYCACHE:
  {
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    if (check_access(thd, INDEX_ACL, first_table->db.str,
                     &first_table->grant.privilege,
                     &first_table->grant.m_internal,
                     0, 0))
      goto error;
    res= mysql_assign_to_keycache(thd, first_table, &lex->ident);
    break;
  }
  case SQLCOM_PRELOAD_KEYS:
  {
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    if (check_access(thd, INDEX_ACL, first_table->db.str,
                     &first_table->grant.privilege,
                     &first_table->grant.m_internal,
                     0, 0))
      goto error;
    res = mysql_preload_keys(thd, first_table);
    break;
  }
#ifdef HAVE_REPLICATION
  case SQLCOM_CHANGE_MASTER:
  {
    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
    Master_info *mi;
    bool new_master= 0;
    bool master_info_added;

    if (check_global_access(thd, SUPER_ACL))
      goto error;
    /*
      In this code it's ok to use LOCK_active_mi as we are adding new things
      into master_info_index
    */
    mysql_mutex_lock(&LOCK_active_mi);
    if (!master_info_index)
    {
      mysql_mutex_unlock(&LOCK_active_mi);
      my_error(ER_SERVER_SHUTDOWN, MYF(0));
      goto error;
    }

    mi= master_info_index->get_master_info(&lex_mi->connection_name,
                                           Sql_condition::WARN_LEVEL_NOTE);

    if (mi == NULL)
    {
      /* New replication created */
      mi= new Master_info(&lex_mi->connection_name, relay_log_recovery); 
      if (unlikely(!mi || mi->error()))
      {
        delete mi;
        res= 1;
        mysql_mutex_unlock(&LOCK_active_mi);
        break;
      }
      new_master= 1;
    }

    res= change_master(thd, mi, &master_info_added);
    if (res && new_master)
    {
      /*
        If the new master was added by change_master(), remove it as it didn't
        work (this will free mi as well).

        If new master was not added, we still need to free mi.
      */
      if (master_info_added)
        master_info_index->remove_master_info(mi);
      else
        delete mi;
    }
    else
    {
      mi->rpl_filter= get_or_create_rpl_filter(lex_mi->connection_name.str,
                                               lex_mi->connection_name.length);
    }

    mysql_mutex_unlock(&LOCK_active_mi);
    break;
  }
  case SQLCOM_SHOW_SLAVE_STAT:
  {
    /* Accept one of two privileges */
    if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL))
      goto error;

    if (lex->verbose)
    {
      mysql_mutex_lock(&LOCK_active_mi);
      res= show_all_master_info(thd);
      mysql_mutex_unlock(&LOCK_active_mi);
    }
    else
    {
      LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
      Master_info *mi;
      if ((mi= get_master_info(&lex_mi->connection_name,
                               Sql_condition::WARN_LEVEL_ERROR)))
      {
        res= show_master_info(thd, mi, 0);
        mi->release();
      }
    }
    break;
  }
  case SQLCOM_SHOW_MASTER_STAT:
  {
    /* Accept one of two privileges */
    if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL))
      goto error;
    res = show_binlog_info(thd);
    break;
  }

#endif /* HAVE_REPLICATION */
  case SQLCOM_SHOW_ENGINE_STATUS:
    {
      if (check_global_access(thd, PROCESS_ACL))
        goto error;
      res = ha_show_status(thd, lex->create_info.db_type, HA_ENGINE_STATUS);
      break;
    }
  case SQLCOM_SHOW_ENGINE_MUTEX:
    {
      if (check_global_access(thd, PROCESS_ACL))
        goto error;
      res = ha_show_status(thd, lex->create_info.db_type, HA_ENGINE_MUTEX);
      break;
    }
  case SQLCOM_CREATE_INDEX:
  case SQLCOM_DROP_INDEX:
  /*
    CREATE INDEX and DROP INDEX are implemented by calling ALTER
    TABLE with proper arguments.

    In the future ALTER TABLE will notice that the request is to
    only add indexes and create these one by one for the existing
    table without having to do a full rebuild.
  */
  {
    /* Prepare stack copies to be re-execution safe */
    HA_CREATE_INFO create_info;
    Alter_info alter_info(lex->alter_info, thd->mem_root);

    if (unlikely(thd->is_fatal_error)) /* out of memory creating alter_info */
      goto error;

    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    if (check_one_table_access(thd, INDEX_ACL, all_tables))
      goto error; /* purecov: inspected */
    WSREP_TO_ISOLATION_BEGIN(first_table->db.str, first_table->table_name.str, NULL);

    bzero((char*) &create_info, sizeof(create_info));
    create_info.db_type= 0;
    create_info.row_type= ROW_TYPE_NOT_USED;
    create_info.default_table_charset= thd->variables.collation_database;

    res= mysql_alter_table(thd, &first_table->db, &first_table->table_name,
                           &create_info, first_table, &alter_info,
                           0, (ORDER*) 0, 0);
    break;
  }
#ifdef HAVE_REPLICATION
  case SQLCOM_SLAVE_START:
  {
    LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
    Master_info *mi;
    int load_error;

    load_error= rpl_load_gtid_slave_state(thd);

    /*
      We don't need to ensure that only one user is using master_info
      as start_slave is protected against simultaneous usage
    */
    if (unlikely((mi= get_master_info(&lex_mi->connection_name,
                                      Sql_condition::WARN_LEVEL_ERROR))))
    {
      if (load_error)
      {
        /*
          We cannot start a slave using GTID if we cannot load the
          GTID position from the mysql.gtid_slave_pos table. But we
          can allow non-GTID replication (useful eg. during upgrade).
        */
        if (mi->using_gtid != Master_info::USE_GTID_NO)
        {
          mi->release();
          break;
        }
        else
          thd->clear_error();
      }
      if (!start_slave(thd, mi, 1 /* net report*/))
        my_ok(thd);
      mi->release();
    }
    break;
  }
  case SQLCOM_SLAVE_STOP:
  {
    LEX_MASTER_INFO *lex_mi;
    Master_info *mi;
    /*
      If the client thread has locked tables, a deadlock is possible.
      Assume that
      - the client thread does LOCK TABLE t READ.
      - then the master updates t.
      - then the SQL slave thread wants to update t,
        so it waits for the client thread because t is locked by it.
    - then the client thread does SLAVE STOP.
      SLAVE STOP waits for the SQL slave thread to terminate its
      update t, which waits for the client thread because t is locked by it.
      To prevent that, refuse SLAVE STOP if the
      client thread has locked tables
    */
    if (thd->locked_tables_mode ||
        thd->in_active_multi_stmt_transaction() ||
        thd->global_read_lock.is_acquired())
    {
      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                 ER_THD(thd, ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
      goto error;
    }

    lex_mi= &thd->lex->mi;
    if ((mi= get_master_info(&lex_mi->connection_name,
                             Sql_condition::WARN_LEVEL_ERROR)))
    {
      if (stop_slave(thd, mi, 1/* net report*/))
        res= 1;
      mi->release();
      if (rpl_parallel_resize_pool_if_no_slaves())
        res= 1;
      if (!res)
        my_ok(thd);
    }
    break;
  }
  case SQLCOM_SLAVE_ALL_START:
  {
    mysql_mutex_lock(&LOCK_active_mi);
    if (master_info_index && !master_info_index->start_all_slaves(thd))
      my_ok(thd);
    mysql_mutex_unlock(&LOCK_active_mi);
    break;
  }
  case SQLCOM_SLAVE_ALL_STOP:
  {
    if (thd->locked_tables_mode ||
        thd->in_active_multi_stmt_transaction() ||
        thd->global_read_lock.is_acquired())
    {
      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                 ER_THD(thd, ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
      goto error;
    }
    mysql_mutex_lock(&LOCK_active_mi);
    if (master_info_index && !master_info_index->stop_all_slaves(thd))
      my_ok(thd);      
    mysql_mutex_unlock(&LOCK_active_mi);
    break;
  }
#endif /* HAVE_REPLICATION */
  case SQLCOM_RENAME_TABLE:
  {
    if (check_rename_table(thd, first_table, all_tables))
      goto error;

    WSREP_TO_ISOLATION_BEGIN(0, 0, first_table);

    if (mysql_rename_tables(thd, first_table, 0))
      goto error;
    break;
  }
#ifndef EMBEDDED_LIBRARY
  case SQLCOM_SHOW_BINLOGS:
#ifdef DONT_ALLOW_SHOW_COMMANDS
    my_message(ER_NOT_ALLOWED_COMMAND, ER_THD(thd, ER_NOT_ALLOWED_COMMAND),
               MYF(0)); /* purecov: inspected */
    goto error;
#else
    {
      if (check_global_access(thd, SUPER_ACL | REPL_CLIENT_ACL))
	goto error;
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
      res = show_binlogs(thd);
      break;
    }
#endif
#endif /* EMBEDDED_LIBRARY */
  case SQLCOM_SHOW_CREATE:
  {
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
#ifdef DONT_ALLOW_SHOW_COMMANDS
    my_message(ER_NOT_ALLOWED_COMMAND, ER_THD(thd, ER_NOT_ALLOWED_COMMAND),
               MYF(0)); /* purecov: inspected */
    goto error;
#else
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);

     /*
        Access check:
        SHOW CREATE TABLE require any privileges on the table level (ie
        effecting all columns in the table).
        SHOW CREATE VIEW require the SHOW_VIEW and SELECT ACLs on the table
        level.
        NOTE: SHOW_VIEW ACL is checked when the view is created.
      */

      DBUG_PRINT("debug", ("lex->only_view: %d, table: %s.%s",
                           lex->table_type == TABLE_TYPE_VIEW,
                           first_table->db.str, first_table->table_name.str));
      res= mysqld_show_create(thd, first_table);
      break;
#endif
  }
  case SQLCOM_CHECKSUM:
  {
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_READ);

    if (check_table_access(thd, SELECT_ACL, all_tables,
                           FALSE, UINT_MAX, FALSE))
      goto error; /* purecov: inspected */

    res = mysql_checksum_table(thd, first_table, &lex->check_opt);
    break;
  }
  case SQLCOM_UPDATE:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);
    ha_rows found= 0, updated= 0;
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);

    if (update_precheck(thd, all_tables))
      break;

    /*
      UPDATE IGNORE can be unsafe. We therefore use row based
      logging if mixed or row based logging is available.
      TODO: Check if the order of the output of the select statement is
      deterministic. Waiting for BUG#42415
    */
    if (lex->ignore)
      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_UPDATE_IGNORE);

    DBUG_ASSERT(select_lex->offset_limit == 0);
    unit->set_limit(select_lex);
    MYSQL_UPDATE_START(thd->query());
    res= up_result= mysql_update(thd, all_tables,
                                  select_lex->item_list,
                                  lex->value_list,
                                  select_lex->where,
                                  select_lex->order_list.elements,
                                  select_lex->order_list.first,
                                  unit->select_limit_cnt,
                                  lex->ignore, &found, &updated);
    MYSQL_UPDATE_DONE(res, found, updated);
    /* mysql_update return 2 if we need to switch to multi-update */
    if (up_result != 2)
      break;
    if (thd->lex->period_conditions.is_set())
    {
      DBUG_ASSERT(0); // Should never happen
      goto error;
    }
  }
  /* fall through */
  case SQLCOM_UPDATE_MULTI:
  {
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    /* if we switched from normal update, rights are checked */
    if (up_result != 2)
    {
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);
      if ((res= multi_update_precheck(thd, all_tables)))
        break;
    }
    else
      res= 0;

    unit->set_limit(select_lex);
    /*
      We can not use mysql_explain_union() because of parameters of
      mysql_select in mysql_multi_update so just set the option if needed
    */
    if (thd->lex->describe)
    {
      select_lex->set_explain_type(FALSE);
      select_lex->options|= SELECT_DESCRIBE;
    }

    res= mysql_multi_update_prepare(thd);

#ifdef HAVE_REPLICATION
    /* Check slave filtering rules */
    if (unlikely(thd->slave_thread && !have_table_map_for_update))
    {
      if (all_tables_not_ok(thd, all_tables))
      {
        if (res!= 0)
        {
          res= 0;             /* don't care of prev failure  */
          thd->clear_error(); /* filters are of highest prior */
        }
        /* we warn the slave SQL thread */
        my_error(ER_SLAVE_IGNORED_TABLE, MYF(0));
        break;
      }
      if (res)
        break;
    }
    else
    {
#endif /* HAVE_REPLICATION */
      if (res)
        break;
      if (opt_readonly &&
	  !(thd->security_ctx->master_access & SUPER_ACL) &&
	  some_non_temp_table_to_be_updated(thd, all_tables))
      {
	my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
	break;
      }
#ifdef HAVE_REPLICATION
    }  /* unlikely */
#endif
    {
      multi_update *result_obj;
      MYSQL_MULTI_UPDATE_START(thd->query());
      res= mysql_multi_update(thd, all_tables,
                              &select_lex->item_list,
                              &lex->value_list,
                              select_lex->where,
                              select_lex->options,
                              lex->duplicates,
                              lex->ignore,
                              unit,
                              select_lex,
                              &result_obj);
      if (result_obj)
      {
        MYSQL_MULTI_UPDATE_DONE(res, result_obj->num_found(),
                                result_obj->num_updated());
        res= FALSE; /* Ignore errors here */
        delete result_obj;
      }
      else
      {
        MYSQL_MULTI_UPDATE_DONE(1, 0, 0);
      }
    }
    break;
  }
  case SQLCOM_REPLACE:
#ifndef DBUG_OFF
    if (mysql_bin_log.is_open())
    {
      /*
        Generate an incident log event before writing the real event
        to the binary log.  We put this event is before the statement
        since that makes it simpler to check that the statement was
        not executed on the slave (since incidents usually stop the
        slave).

        Observe that any row events that are generated will be
        generated before.

        This is only for testing purposes and will not be present in a
        release build.
      */

      Incident incident= INCIDENT_NONE;
      DBUG_PRINT("debug", ("Just before generate_incident()"));
      DBUG_EXECUTE_IF("incident_database_resync_on_replace",
                      incident= INCIDENT_LOST_EVENTS;);
      if (incident)
      {
        Incident_log_event ev(thd, incident);
        (void) mysql_bin_log.write(&ev);        /* error is ignored */
        if (mysql_bin_log.rotate_and_purge(true))
        {
          res= 1;
          break;
        }
      }
      DBUG_PRINT("debug", ("Just after generate_incident()"));
    }
#endif
    /* fall through */
  case SQLCOM_INSERT:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);
    DBUG_ASSERT(first_table == all_tables && first_table != 0);

    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);

    /*
      Since INSERT DELAYED doesn't support temporary tables, we could
      not pre-open temporary tables for SQLCOM_INSERT / SQLCOM_REPLACE.
      Open them here instead.
    */
    if (first_table->lock_type != TL_WRITE_DELAYED)
    {
      res= (thd->open_temporary_tables(all_tables)) ? TRUE : FALSE;
      if (res)
        break;
    }

    if ((res= insert_precheck(thd, all_tables)))
      break;

    MYSQL_INSERT_START(thd->query());
    res= mysql_insert(thd, all_tables, lex->field_list, lex->many_values,
		      lex->update_list, lex->value_list,
                      lex->duplicates, lex->ignore);
    MYSQL_INSERT_DONE(res, (ulong) thd->get_row_count_func());
    /*
      If we have inserted into a VIEW, and the base table has
      AUTO_INCREMENT column, but this column is not accessible through
      a view, then we should restore LAST_INSERT_ID to the value it
      had before the statement.
    */
    if (first_table->view && !first_table->contain_auto_increment)
      thd->first_successful_insert_id_in_cur_stmt=
        thd->first_successful_insert_id_in_prev_stmt;

#ifdef ENABLED_DEBUG_SYNC
    DBUG_EXECUTE_IF("after_mysql_insert",
                    {
                      const char act1[]=
                        "now "
                        "wait_for signal.continue";
                      const char act2[]=
                        "now "
                        "signal signal.continued";
                      DBUG_ASSERT(debug_sync_service);
                      DBUG_ASSERT(!debug_sync_set_action(thd,
                                                         STRING_WITH_LEN(act1)));
                      DBUG_ASSERT(!debug_sync_set_action(thd,
                                                         STRING_WITH_LEN(act2)));
                    };);
    DEBUG_SYNC(thd, "after_mysql_insert");
#endif
    break;
  }
  case SQLCOM_REPLACE_SELECT:
  case SQLCOM_INSERT_SELECT:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_INSERT_REPLACE);
    select_insert *sel_result;
    bool explain= MY_TEST(lex->describe);
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);

    if ((res= insert_precheck(thd, all_tables)))
      break;
#ifdef WITH_WSREP
    if (WSREP(thd) && thd->wsrep_consistency_check == CONSISTENCY_CHECK_DECLARED)
    {
      thd->wsrep_consistency_check = CONSISTENCY_CHECK_RUNNING;
      WSREP_TO_ISOLATION_BEGIN(first_table->db.str, first_table->table_name.str, NULL);
    }
#endif /* WITH_WSREP */

    /*
      INSERT...SELECT...ON DUPLICATE KEY UPDATE/REPLACE SELECT/
      INSERT...IGNORE...SELECT can be unsafe, unless ORDER BY PRIMARY KEY
      clause is used in SELECT statement. We therefore use row based
      logging if mixed or row based logging is available.
      TODO: Check if the order of the output of the select statement is
      deterministic. Waiting for BUG#42415
    */
    if (lex->sql_command == SQLCOM_INSERT_SELECT &&
        lex->duplicates == DUP_UPDATE)
      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_SELECT_UPDATE);

    if (lex->sql_command == SQLCOM_INSERT_SELECT && lex->ignore)
      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_IGNORE_SELECT);

    if (lex->sql_command == SQLCOM_REPLACE_SELECT)
      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_REPLACE_SELECT);

    /* Fix lock for first table */
    if (first_table->lock_type == TL_WRITE_DELAYED)
      first_table->lock_type= TL_WRITE;

    /* Don't unlock tables until command is written to binary log */
    select_lex->options|= SELECT_NO_UNLOCK;

    unit->set_limit(select_lex);

    if (!(res=open_and_lock_tables(thd, all_tables, TRUE, 0)))
    {
      MYSQL_INSERT_SELECT_START(thd->query());
      /*
        Only the INSERT table should be merged. Other will be handled by
        select.
      */
      /* Skip first table, which is the table we are inserting in */
      TABLE_LIST *second_table= first_table->next_local;
      /*
        This is a hack: this leaves select_lex->table_list in an inconsistent
        state as 'elements' does not contain number of elements in the list.
        Moreover, if second_table == NULL then 'next' becomes invalid.
        TODO: fix it by removing the front element (restoring of it should
        be done properly as well)
      */
      select_lex->table_list.first= second_table;
      select_lex->context.table_list= 
        select_lex->context.first_name_resolution_table= second_table;
      res= mysql_insert_select_prepare(thd);
      if (!res && (sel_result= new (thd->mem_root) select_insert(thd,
                                                             first_table,
                                                             first_table->table,
                                                             &lex->field_list,
                                                             &lex->update_list,
                                                             &lex->value_list,
                                                             lex->duplicates,
                                                             lex->ignore)))
      {
        if (lex->analyze_stmt)
          ((select_result_interceptor*)sel_result)->disable_my_ok_calls();

        if (explain)
          res= mysql_explain_union(thd, &thd->lex->unit, sel_result);
        else
          res= handle_select(thd, lex, sel_result, OPTION_SETUP_TABLES_DONE);
        /*
          Invalidate the table in the query cache if something changed
          after unlocking when changes become visible.
          TODO: this is workaround. right way will be move invalidating in
          the unlock procedure.
        */
        if (!res && first_table->lock_type ==  TL_WRITE_CONCURRENT_INSERT &&
            thd->lock)
        {
          /* INSERT ... SELECT should invalidate only the very first table */
          TABLE_LIST *save_table= first_table->next_local;
          first_table->next_local= 0;
          query_cache_invalidate3(thd, first_table, 1);
          first_table->next_local= save_table;
        }
        if (explain)
        {
          /*
            sel_result needs to be cleaned up properly.
            INSERT... SELECT statement will call either send_eof() or
            abort_result_set(). EXPLAIN doesn't call either, so we need
            to cleanup manually.
          */
          sel_result->abort_result_set();
        }
        delete sel_result;
      }

      if (!res && (explain || lex->analyze_stmt))
        res= thd->lex->explain->send_explain(thd);

      /* revert changes for SP */
      MYSQL_INSERT_SELECT_DONE(res, (ulong) thd->get_row_count_func());
      select_lex->table_list.first= first_table;
    }
    /*
      If we have inserted into a VIEW, and the base table has
      AUTO_INCREMENT column, but this column is not accessible through
      a view, then we should restore LAST_INSERT_ID to the value it
      had before the statement.
    */
    if (first_table->view && !first_table->contain_auto_increment)
      thd->first_successful_insert_id_in_cur_stmt=
        thd->first_successful_insert_id_in_prev_stmt;

    break;
  }
  case SQLCOM_DELETE:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);
    select_result *sel_result= NULL;
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);

    if ((res= delete_precheck(thd, all_tables)))
      break;
    DBUG_ASSERT(select_lex->offset_limit == 0);
    unit->set_limit(select_lex);

    MYSQL_DELETE_START(thd->query());
    Protocol * UNINIT_VAR(save_protocol);
    bool replaced_protocol= false;

    if (!select_lex->item_list.is_empty())
    {
      /* This is DELETE ... RETURNING.  It will return output to the client */
      if (thd->lex->analyze_stmt)
      {
        /* 
          Actually, it is ANALYZE .. DELETE .. RETURNING. We need to produce
          output and then discard it.
        */
        sel_result= new (thd->mem_root) select_send_analyze(thd);
        replaced_protocol= true;
        save_protocol= thd->protocol;
        thd->protocol= new Protocol_discard(thd);
      }
      else
      {
        if (!lex->result && !(sel_result= new (thd->mem_root) select_send(thd)))
          goto error;
      }
    }

    res = mysql_delete(thd, all_tables, 
                       select_lex->where, &select_lex->order_list,
                       unit->select_limit_cnt, select_lex->options,
                       lex->result ? lex->result : sel_result);

    if (replaced_protocol)
    {
      delete thd->protocol;
      thd->protocol= save_protocol;
    }

    if (thd->lex->analyze_stmt || thd->lex->describe)
    {
      if (!res)
        res= thd->lex->explain->send_explain(thd);
    }

    delete sel_result;
    MYSQL_DELETE_DONE(res, (ulong) thd->get_row_count_func());
    break;
  }
  case SQLCOM_DELETE_MULTI:
  {
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    TABLE_LIST *aux_tables= thd->lex->auxiliary_table_list.first;
    multi_delete *result;
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_UPDATE_DELETE);

    if ((res= multi_delete_precheck(thd, all_tables)))
      break;

    /* condition will be TRUE on SP re-excuting */
    if (select_lex->item_list.elements != 0)
      select_lex->item_list.empty();
    if (add_item_to_list(thd, new (thd->mem_root) Item_null(thd)))
      goto error;

    THD_STAGE_INFO(thd, stage_init);
    if ((res= open_and_lock_tables(thd, all_tables, TRUE, 0)))
      break;

    MYSQL_MULTI_DELETE_START(thd->query());
    if (unlikely(res= mysql_multi_delete_prepare(thd)))
    {
      MYSQL_MULTI_DELETE_DONE(1, 0);
      goto error;
    }

    if (likely(!thd->is_fatal_error))
    {
      result= new (thd->mem_root) multi_delete(thd, aux_tables,
                                               lex->table_count);
      if (likely(result))
      {
        if (unlikely(select_lex->vers_setup_conds(thd, aux_tables)))
          goto multi_delete_error;
        res= mysql_select(thd,
                          select_lex->get_table_list(),
                          select_lex->with_wild,
                          select_lex->item_list,
                          select_lex->where,
                          0, (ORDER *)NULL, (ORDER *)NULL, (Item *)NULL,
                          (ORDER *)NULL,
                          (select_lex->options | thd->variables.option_bits |
                          SELECT_NO_JOIN_CACHE | SELECT_NO_UNLOCK |
                          OPTION_SETUP_TABLES_DONE) & ~OPTION_BUFFER_RESULT,
                          result, unit, select_lex);
        res|= (int)(thd->is_error());

        MYSQL_MULTI_DELETE_DONE(res, result->num_deleted());
        if (res)
          result->abort_result_set(); /* for both DELETE and EXPLAIN DELETE */
        else
        {
          if (lex->describe || lex->analyze_stmt)
            res= thd->lex->explain->send_explain(thd);
        }
      multi_delete_error:
        delete result;
      }
    }
    else
    {
      res= TRUE;                                // Error
      MYSQL_MULTI_DELETE_DONE(1, 0);
    }
    break;
  }
  case SQLCOM_DROP_SEQUENCE:
  case SQLCOM_DROP_TABLE:
  {
    int result;
    DBUG_ASSERT(first_table == all_tables && first_table != 0);

    thd->open_options|= HA_OPEN_FOR_REPAIR;
    result= thd->open_temporary_tables(all_tables);
    thd->open_options&= ~HA_OPEN_FOR_REPAIR;
    if (result)
      goto error;
    if (!lex->tmp_table())
    {
      if (check_table_access(thd, DROP_ACL, all_tables, FALSE, UINT_MAX, FALSE))
	goto error;				/* purecov: inspected */
    }
    else
    {
      status_var_decrement(thd->status_var.com_stat[lex->sql_command]);
      status_var_increment(thd->status_var.com_drop_tmp_table);

      /* So that DROP TEMPORARY TABLE gets to binlog at commit/rollback */
      thd->variables.option_bits|= OPTION_KEEP_LOG;
    }
    /*
      If we are a slave, we should add IF EXISTS if the query executed
      on the master without an error. This will help a slave to
      recover from multi-table DROP TABLE that was aborted in the
      middle.
    */
    if (thd->slave_thread && !thd->slave_expected_error &&
        slave_ddl_exec_mode_options == SLAVE_EXEC_MODE_IDEMPOTENT)
      lex->create_info.set(DDL_options_st::OPT_IF_EXISTS);

    if (WSREP(thd))
    {
      for (TABLE_LIST *table= all_tables; table; table= table->next_global)
      {
        if (!lex->tmp_table() &&
           (!thd->is_current_stmt_binlog_format_row() ||
	    !thd->find_temporary_table(table)))
        {
          WSREP_TO_ISOLATION_BEGIN(NULL, NULL, all_tables);
          break;
        }
      }
    }
    
    /* DDL and binlog write order are protected by metadata locks. */
    res= mysql_rm_table(thd, first_table, lex->if_exists(), lex->tmp_table(),
                        lex->table_type == TABLE_TYPE_SEQUENCE);

    /*
      When dropping temporary tables if @@session_track_state_change is ON
      then send the boolean tracker in the OK packet
    */
    if(!res && (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
    {
      SESSION_TRACKER_CHANGED(thd, SESSION_STATE_CHANGE_TRACKER, NULL);
    }
    break;
  }
  case SQLCOM_SHOW_PROCESSLIST:
    if (!thd->security_ctx->priv_user[0] &&
        check_global_access(thd,PROCESS_ACL))
      break;
    mysqld_list_processes(thd,
			  (thd->security_ctx->master_access & PROCESS_ACL ?
                           NullS :
                           thd->security_ctx->priv_user),
                          lex->verbose);
    break;
  case SQLCOM_SHOW_AUTHORS:
    res= mysqld_show_authors(thd);
    break;
  case SQLCOM_SHOW_CONTRIBUTORS:
    res= mysqld_show_contributors(thd);
    break;
  case SQLCOM_SHOW_PRIVILEGES:
    res= mysqld_show_privileges(thd);
    break;
  case SQLCOM_SHOW_ENGINE_LOGS:
#ifdef DONT_ALLOW_SHOW_COMMANDS
    my_message(ER_NOT_ALLOWED_COMMAND, ER_THD(thd, ER_NOT_ALLOWED_COMMAND),
               MYF(0));	/* purecov: inspected */
    goto error;
#else
    {
      if (check_access(thd, FILE_ACL, any_db, NULL, NULL, 0, 0))
	goto error;
      res= ha_show_status(thd, lex->create_info.db_type, HA_ENGINE_LOGS);
      break;
    }
#endif
  case SQLCOM_CHANGE_DB:
  {
    if (!mysql_change_db(thd, &select_lex->db, FALSE))
      my_ok(thd);

    break;
  }

  case SQLCOM_LOAD:
  {
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    uint privilege= (lex->duplicates == DUP_REPLACE ?
		     INSERT_ACL | DELETE_ACL : INSERT_ACL) |
                    (lex->local_file ? 0 : FILE_ACL);

    if (lex->local_file)
    {
      if (!(thd->client_capabilities & CLIENT_LOCAL_FILES) ||
          !opt_local_infile)
      {
	my_message(ER_NOT_ALLOWED_COMMAND, ER_THD(thd, ER_NOT_ALLOWED_COMMAND), MYF(0));
	goto error;
      }
    }

    if (check_one_table_access(thd, privilege, all_tables))
      goto error;

    res= mysql_load(thd, lex->exchange, first_table, lex->field_list,
                    lex->update_list, lex->value_list, lex->duplicates,
                    lex->ignore, (bool) lex->local_file);
    break;
  }

  case SQLCOM_SET_OPTION:
  {
    List<set_var_base> *lex_var_list= &lex->var_list;

    if ((check_table_access(thd, SELECT_ACL, all_tables, FALSE, UINT_MAX, FALSE)
         || open_and_lock_tables(thd, all_tables, TRUE, 0)))
      goto error;
    if (likely(!(res= sql_set_variables(thd, lex_var_list, true))))
    {
      if (likely(!thd->is_error()))
        my_ok(thd);
    }
    else
    {
      /*
        We encountered some sort of error, but no message was sent.
        Send something semi-generic here since we don't know which
        assignment in the list caused the error.
      */
      if (!thd->is_error())
        my_error(ER_WRONG_ARGUMENTS,MYF(0),"SET");
      goto error;
    }

    break;
  }

  case SQLCOM_UNLOCK_TABLES:
    /*
      It is critical for mysqldump --single-transaction --master-data that
      UNLOCK TABLES does not implicitely commit a connection which has only
      done FLUSH TABLES WITH READ LOCK + BEGIN. If this assumption becomes
      false, mysqldump will not work.
    */
    if (thd->variables.option_bits & OPTION_TABLE_LOCK)
    {
      res= trans_commit_implicit(thd);
      thd->locked_tables_list.unlock_locked_tables(thd);
      thd->release_transactional_locks();
      thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
    }
    if (thd->global_read_lock.is_acquired() &&
        thd->current_backup_stage == BACKUP_FINISHED)
      thd->global_read_lock.unlock_global_read_lock(thd);
    if (res)
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_LOCK_TABLES:
    /* We must end the transaction first, regardless of anything */
    res= trans_commit_implicit(thd);
    thd->locked_tables_list.unlock_locked_tables(thd);
    /* Release transactional metadata locks. */
    thd->release_transactional_locks();
    if (res)
      goto error;

#ifdef WITH_WSREP
    /* Clean up the previous transaction on implicit commit. */
    if (wsrep_on(thd) && !wsrep_not_committed(thd) && wsrep_after_statement(thd))
      goto error;
#endif

    /* We can't have any kind of table locks while backup is active */
    if (thd->current_backup_stage != BACKUP_FINISHED)
    {
      my_error(ER_BACKUP_LOCK_IS_ACTIVE, MYF(0));
      goto error;
    }

    /* Should not lock tables while BACKUP LOCK is active */
    if (thd->mdl_backup_lock)
    {
      my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
      goto error;
    }

    /*
      Here we have to pre-open temporary tables for LOCK TABLES.

      CF_PREOPEN_TMP_TABLES is not set for this SQL statement simply
      because LOCK TABLES calls close_thread_tables() as a first thing
      (it's called from unlock_locked_tables() above). So even if
      CF_PREOPEN_TMP_TABLES was set and the tables would be pre-opened
      in a usual way, they would have been closed.
    */
    if (thd->open_temporary_tables(all_tables))
      goto error;

    if (lock_tables_precheck(thd, all_tables))
      goto error;

    thd->variables.option_bits|= OPTION_TABLE_LOCK;

    res= lock_tables_open_and_lock_tables(thd, all_tables);

    if (res)
    {
      thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
    }
    else
    {
#ifdef HAVE_QUERY_CACHE
      if (thd->variables.query_cache_wlock_invalidate)
	query_cache.invalidate_locked_for_write(thd, first_table);
#endif /*HAVE_QUERY_CACHE*/
      my_ok(thd);
    }
    break;
  case SQLCOM_BACKUP:
    if (check_global_access(thd, RELOAD_ACL))
      goto error;
    if (!(res= run_backup_stage(thd, lex->backup_stage)))
      my_ok(thd);
    break;
  case SQLCOM_BACKUP_LOCK:
    if (check_global_access(thd, RELOAD_ACL))
      goto error;
    /* first table is set for lock. For unlock the list is empty */
    if (first_table)
      res= backup_lock(thd, first_table);
    else
      backup_unlock(thd);
    if (!res)
      my_ok(thd);
    break;
  case SQLCOM_CREATE_DB:
  {
    if (prepare_db_action(thd, lex->create_info.or_replace() ?
                          (CREATE_ACL | DROP_ACL) : CREATE_ACL,
                          &lex->name))
      break;
    WSREP_TO_ISOLATION_BEGIN(lex->name.str, NULL, NULL);
    res= mysql_create_db(thd, &lex->name,
                         lex->create_info, &lex->create_info);
    break;
  }
  case SQLCOM_DROP_DB:
  {
    if (prepare_db_action(thd, DROP_ACL, &lex->name))
      break;
    WSREP_TO_ISOLATION_BEGIN(lex->name.str, NULL, NULL);
    res= mysql_rm_db(thd, &lex->name, lex->if_exists());
    break;
  }
  case SQLCOM_ALTER_DB_UPGRADE:
  {
    LEX_CSTRING *db= &lex->name;
#ifdef HAVE_REPLICATION
    if (thd->slave_thread)
    {
      rpl_filter= thd->system_thread_info.rpl_sql_info->rpl_filter;
      if (!rpl_filter->db_ok(db->str) ||
          !rpl_filter->db_ok_with_wild_table(db->str))
      {
        res= 1;
        my_message(ER_SLAVE_IGNORED_TABLE, ER_THD(thd, ER_SLAVE_IGNORED_TABLE), MYF(0));
        break;
      }
    }
#endif
    if (check_db_name((LEX_STRING*) db))
    {
      my_error(ER_WRONG_DB_NAME, MYF(0), db->str);
      break;
    }
    if (check_access(thd, ALTER_ACL, db->str, NULL, NULL, 1, 0) ||
        check_access(thd, DROP_ACL, db->str, NULL, NULL, 1, 0) ||
        check_access(thd, CREATE_ACL, db->str, NULL, NULL, 1, 0))
    {
      res= 1;
      break;
    }
    WSREP_TO_ISOLATION_BEGIN(db->str, NULL, NULL);
    res= mysql_upgrade_db(thd, db);
    if (!res)
      my_ok(thd);
    break;
  }
  case SQLCOM_ALTER_DB:
  {
    LEX_CSTRING *db= &lex->name;
    if (prepare_db_action(thd, ALTER_ACL, db))
      break;
    WSREP_TO_ISOLATION_BEGIN(db->str, NULL, NULL);
    res= mysql_alter_db(thd, db, &lex->create_info);
    break;
  }
  case SQLCOM_SHOW_CREATE_DB:
  {
    char db_name_buff[NAME_LEN+1];
    LEX_CSTRING db_name;
    DBUG_EXECUTE_IF("4x_server_emul",
                    my_error(ER_UNKNOWN_ERROR, MYF(0)); goto error;);

    db_name.str= db_name_buff;
    db_name.length= lex->name.length;
    strmov(db_name_buff, lex->name.str);

    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);

    if (check_db_name((LEX_STRING*) &db_name))
    {
      my_error(ER_WRONG_DB_NAME, MYF(0), db_name.str);
      break;
    }
    res= mysqld_show_create_db(thd, &db_name, &lex->name, lex->create_info);
    break;
  }
  case SQLCOM_CREATE_EVENT:
  case SQLCOM_ALTER_EVENT:
  #ifdef HAVE_EVENT_SCHEDULER
  do
  {
    DBUG_ASSERT(lex->event_parse_data);
    if (lex->table_or_sp_used())
    {
      my_error(ER_SUBQUERIES_NOT_SUPPORTED, MYF(0), "CREATE/ALTER EVENT");
      break;
    }

    res= sp_process_definer(thd);
    if (res)
      break;

    switch (lex->sql_command) {
    case SQLCOM_CREATE_EVENT:
    {
      res= Events::create_event(thd, lex->event_parse_data);
      break;
    }
    case SQLCOM_ALTER_EVENT:
      res= Events::update_event(thd, lex->event_parse_data,
                                lex->spname ? &lex->spname->m_db : NULL,
                                lex->spname ? &lex->spname->m_name : NULL);
      break;
    default:
      DBUG_ASSERT(0);
    }
    DBUG_PRINT("info",("DDL error code=%d", res));
    if (!res)
      my_ok(thd);

  } while (0);
  /* Don't do it, if we are inside a SP */
  if (!thd->spcont)
  {
    sp_head::destroy(lex->sphead);
    lex->sphead= NULL;
  }
  /* lex->unit.cleanup() is called outside, no need to call it here */
  break;
  case SQLCOM_SHOW_CREATE_EVENT:
    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
    res= Events::show_create_event(thd, &lex->spname->m_db,
                                   &lex->spname->m_name);
    break;
  case SQLCOM_DROP_EVENT:
    if (!(res= Events::drop_event(thd,
                                  &lex->spname->m_db, &lex->spname->m_name,
                                  lex->if_exists())))
      my_ok(thd);
    break;
#else
    my_error(ER_NOT_SUPPORTED_YET,MYF(0),"embedded server");
    break;
#endif
  case SQLCOM_CREATE_FUNCTION:                  // UDF function
  {
    if (check_access(thd, lex->create_info.or_replace() ?
                          (INSERT_ACL | DELETE_ACL) : INSERT_ACL,
                     "mysql", NULL, NULL, 1, 0))
      break;
#ifdef HAVE_DLOPEN
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    if (!(res = mysql_create_function(thd, &lex->udf)))
      my_ok(thd);
#else
    my_error(ER_CANT_OPEN_LIBRARY, MYF(0), lex->udf.dl, 0, "feature disabled");
    res= TRUE;
#endif
    break;
  }
#ifndef NO_EMBEDDED_ACCESS_CHECKS
  case SQLCOM_CREATE_USER:
  case SQLCOM_CREATE_ROLE:
  {
    if (check_access(thd, lex->create_info.or_replace() ?
                          INSERT_ACL | DELETE_ACL : INSERT_ACL,
                     "mysql", NULL, NULL, 1, 1) &&
        check_global_access(thd,CREATE_USER_ACL))
      break;
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    /* Conditionally writes to binlog */
    if (!(res= mysql_create_user(thd, lex->users_list,
                                 lex->sql_command == SQLCOM_CREATE_ROLE)))
      my_ok(thd);
    break;
  }
  case SQLCOM_DROP_USER:
  case SQLCOM_DROP_ROLE:
  {
    if (check_access(thd, DELETE_ACL, "mysql", NULL, NULL, 1, 1) &&
        check_global_access(thd,CREATE_USER_ACL))
      break;
    /* Conditionally writes to binlog */
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    if (!(res= mysql_drop_user(thd, lex->users_list,
                               lex->sql_command == SQLCOM_DROP_ROLE)))
      my_ok(thd);
    break;
  }
  case SQLCOM_ALTER_USER:
  case SQLCOM_RENAME_USER:
  {
    if (check_access(thd, UPDATE_ACL, "mysql", NULL, NULL, 1, 1) &&
        check_global_access(thd,CREATE_USER_ACL))
      break;
    /* Conditionally writes to binlog */
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    if (lex->sql_command == SQLCOM_ALTER_USER)
      res= mysql_alter_user(thd, lex->users_list);
    else
      res= mysql_rename_user(thd, lex->users_list);
    if (!res)
      my_ok(thd);
    break;
  }
  case SQLCOM_REVOKE_ALL:
  {
    if (check_access(thd, UPDATE_ACL, "mysql", NULL, NULL, 1, 1) &&
        check_global_access(thd,CREATE_USER_ACL))
      break;

    /* Conditionally writes to binlog */
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    if (!(res = mysql_revoke_all(thd, lex->users_list)))
      my_ok(thd);
    break;
  }
  case SQLCOM_REVOKE:
  case SQLCOM_GRANT:
  {
    if (lex->type != TYPE_ENUM_PROXY &&
        check_access(thd, lex->grant | lex->grant_tot_col | GRANT_ACL,
                     first_table ?  first_table->db.str : select_lex->db.str,
                     first_table ? &first_table->grant.privilege : NULL,
                     first_table ? &first_table->grant.m_internal : NULL,
                     first_table ? 0 : 1, 0))
      goto error;

    /* Replicate current user as grantor */
    thd->binlog_invoker(false);

    if (thd->security_ctx->user)              // If not replication
    {
      LEX_USER *user;
      bool first_user= TRUE;

      List_iterator <LEX_USER> user_list(lex->users_list);
      while ((user= user_list++))
      {
        if (specialflag & SPECIAL_NO_RESOLVE &&
            hostname_requires_resolving(user->host.str))
          push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
                              ER_WARN_HOSTNAME_WONT_WORK,
                              ER_THD(thd, ER_WARN_HOSTNAME_WONT_WORK));

        /*
          GRANT/REVOKE PROXY has the target user as a first entry in the list. 
         */
        if (lex->type == TYPE_ENUM_PROXY && first_user)
        {
          if (!(user= get_current_user(thd, user)) || !user->host.str)
            goto error;

          first_user= FALSE;
          if (acl_check_proxy_grant_access (thd, user->host.str, user->user.str,
                                        lex->grant & GRANT_ACL))
            goto error;
        } 
      }
    }
    if (first_table)
    {
      const Sp_handler *sph= Sp_handler::handler((stored_procedure_type)
                                                 lex->type);
      if (sph)
      {
        uint grants= lex->all_privileges 
		   ? (PROC_ACLS & ~GRANT_ACL) | (lex->grant & GRANT_ACL)
		   : lex->grant;
        if (check_grant_routine(thd, grants | GRANT_ACL, all_tables, sph, 0))
	  goto error;
        /* Conditionally writes to binlog */
        WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
        res= mysql_routine_grant(thd, all_tables, sph,
                                 lex->users_list, grants,
                                 lex->sql_command == SQLCOM_REVOKE, TRUE);
        if (!res)
          my_ok(thd);
      }
      else
      {
	if (check_grant(thd,(lex->grant | lex->grant_tot_col | GRANT_ACL),
                        all_tables, FALSE, UINT_MAX, FALSE))
	  goto error;
        /* Conditionally writes to binlog */
        WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
        res= mysql_table_grant(thd, all_tables, lex->users_list,
			       lex->columns, lex->grant,
			       lex->sql_command == SQLCOM_REVOKE);
      }
    }
    else
    {
      if (lex->columns.elements || (lex->type && lex->type != TYPE_ENUM_PROXY))
      {
	my_message(ER_ILLEGAL_GRANT_FOR_TABLE, ER_THD(thd, ER_ILLEGAL_GRANT_FOR_TABLE),
                   MYF(0));
        goto error;
      }
      else
      {
        WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
        /* Conditionally writes to binlog */
        res= mysql_grant(thd, select_lex->db.str, lex->users_list, lex->grant,
                         lex->sql_command == SQLCOM_REVOKE,
                         lex->type == TYPE_ENUM_PROXY);
      }
      if (!res)
      {
	if (lex->sql_command == SQLCOM_GRANT)
	{
	  List_iterator <LEX_USER> str_list(lex->users_list);
	  LEX_USER *user, *tmp_user;
	  while ((tmp_user=str_list++))
          {
            if (!(user= get_current_user(thd, tmp_user)))
              goto error;
	    reset_mqh(user, 0);
          }
	}
      }
    }
    break;
  }
  case SQLCOM_REVOKE_ROLE:
  case SQLCOM_GRANT_ROLE:
  {
    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
    if (!(res= mysql_grant_role(thd, lex->users_list,
                                lex->sql_command != SQLCOM_GRANT_ROLE)))
      my_ok(thd);
    break;
  }
#endif /*!NO_EMBEDDED_ACCESS_CHECKS*/
  case SQLCOM_RESET:
    /*
      RESET commands are never written to the binary log, so we have to
      initialize this variable because RESET shares the same code as FLUSH
    */
    lex->no_write_to_binlog= 1;
    /* fall through */
  case SQLCOM_FLUSH:
  {
    int write_to_binlog;
    if (check_global_access(thd,RELOAD_ACL))
      goto error;

    if (first_table && lex->type & (REFRESH_READ_LOCK|REFRESH_FOR_EXPORT))
    {
      /* Check table-level privileges. */
      if (check_table_access(thd, PRIV_LOCK_TABLES, all_tables,
                             FALSE, UINT_MAX, FALSE))
        goto error;

      if (flush_tables_with_read_lock(thd, all_tables))
        goto error;

      my_ok(thd);
      break;
    }

#ifdef WITH_WSREP
    if (lex->type & (
    REFRESH_GRANT                           |
    REFRESH_HOSTS                           |
#ifdef HAVE_OPENSSL
    REFRESH_DES_KEY_FILE                    |
#endif
    /*
      Write all flush log statements except
      FLUSH LOGS
      FLUSH BINARY LOGS
      Check reload_acl_and_cache for why.
    */
    REFRESH_RELAY_LOG                       |
    REFRESH_SLOW_LOG                        |
    REFRESH_GENERAL_LOG                     |
    REFRESH_ENGINE_LOG                      |
    REFRESH_ERROR_LOG                       |
#ifdef HAVE_QUERY_CACHE
    REFRESH_QUERY_CACHE_FREE                |
#endif /* HAVE_QUERY_CACHE */
    REFRESH_STATUS                          |
    REFRESH_USER_RESOURCES))
    {
      WSREP_TO_ISOLATION_BEGIN_WRTCHK(WSREP_MYSQL_DB, NULL, NULL);
    }
#endif /* WITH_WSREP*/

#ifdef HAVE_REPLICATION
    if (lex->type & REFRESH_READ_LOCK)
    {
      /*
        We need to pause any parallel replication slave workers during FLUSH
        TABLES WITH READ LOCK. Otherwise we might cause a deadlock, as
        worker threads eun run in arbitrary order but need to commit in a
        specific given order.
      */
      if (rpl_pause_for_ftwrl(thd))
        goto error;
    }
#endif
    /*
      reload_acl_and_cache() will tell us if we are allowed to write to the
      binlog or not.
    */
    if (!reload_acl_and_cache(thd, lex->type, first_table, &write_to_binlog))
    {
#ifdef WITH_WSREP
      if ((lex->type & REFRESH_TABLES) && !(lex->type & (REFRESH_FOR_EXPORT|REFRESH_READ_LOCK)))
      {
        /*
          This is done after reload_acl_and_cache is because
          LOCK TABLES is not replicated in galera, the upgrade of which
          is checked in reload_acl_and_cache.
          Hence, done after/if we are able to upgrade locks.
        */
        if (first_table)
        {
          WSREP_TO_ISOLATION_BEGIN_WRTCHK(NULL, NULL, first_table);
        }
        else
        {
          WSREP_TO_ISOLATION_BEGIN_WRTCHK(WSREP_MYSQL_DB, NULL, NULL);
        }
      }
#endif /* WITH_WSREP */
      /*
        We WANT to write and we CAN write.
        ! we write after unlocking the table.
      */
      /*
        Presumably, RESET and binlog writing doesn't require synchronization
      */

      if (write_to_binlog > 0)  // we should write
      { 
        if (!lex->no_write_to_binlog)
          res= write_bin_log(thd, FALSE, thd->query(), thd->query_length());
      } else if (write_to_binlog < 0) 
      {
        /* 
           We should not write, but rather report error because 
           reload_acl_and_cache binlog interactions failed 
         */
        res= 1;
      }

      if (!res)
        my_ok(thd);
    } 
    else
      res= 1;                                   // reload_acl_and_cache failed
#ifdef HAVE_REPLICATION
    if (lex->type & REFRESH_READ_LOCK)
      rpl_unpause_after_ftwrl(thd);
#endif
    
    break;
  }
  case SQLCOM_KILL:
  {
    if (lex->table_or_sp_used())
    {
      my_error(ER_SUBQUERIES_NOT_SUPPORTED, MYF(0), "KILL");
      break;
    }

    if (lex->kill_type == KILL_TYPE_ID || lex->kill_type == KILL_TYPE_QUERY)
    {
      Item *it= (Item *)lex->value_list.head();
      if (it->fix_fields_if_needed_for_scalar(lex->thd, &it))
      {
        my_message(ER_SET_CONSTANTS_ONLY, ER_THD(thd, ER_SET_CONSTANTS_ONLY),
                   MYF(0));
        goto error;
      }
      sql_kill(thd, it->val_int(), lex->kill_signal, lex->kill_type);
    }
    else
      sql_kill_user(thd, get_current_user(thd, lex->users_list.head()),
                    lex->kill_signal);
    break;
  }
  case SQLCOM_SHUTDOWN:
#ifndef EMBEDDED_LIBRARY
    DBUG_EXECUTE_IF("crash_shutdown", DBUG_SUICIDE(););
    if (check_global_access(thd,SHUTDOWN_ACL))
      goto error;
    kill_mysql(thd);
    my_ok(thd);
#else
    my_error(ER_NOT_SUPPORTED_YET, MYF(0), "embedded server");
#endif
    break;

#ifndef NO_EMBEDDED_ACCESS_CHECKS
  case SQLCOM_SHOW_CREATE_USER:
  {
    LEX_USER *grant_user= lex->grant_user;
    if (!grant_user)
      goto error;

    res = mysql_show_create_user(thd, grant_user);
    break;
  }
  case SQLCOM_SHOW_GRANTS:
  {
    LEX_USER *grant_user= lex->grant_user;
    if (!grant_user)
      goto error;

    WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
    res = mysql_show_grants(thd, grant_user);
    break;
  }
#endif
  case SQLCOM_HA_OPEN:
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    if (check_table_access(thd, SELECT_ACL, all_tables, FALSE, UINT_MAX, FALSE))
      goto error;
    /* Close temporary tables which were pre-opened for privilege checking. */
    close_thread_tables(thd);
    all_tables->table= NULL;
    res= mysql_ha_open(thd, first_table, 0);
    break;
  case SQLCOM_HA_CLOSE:
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    res= mysql_ha_close(thd, first_table);
    break;
  case SQLCOM_HA_READ:
    DBUG_ASSERT(first_table == all_tables && first_table != 0);
    /*
      There is no need to check for table permissions here, because
      if a user has no permissions to read a table, he won't be
      able to open it (with SQLCOM_HA_OPEN) in the first place.
    */
    unit->set_limit(select_lex);

    res= mysql_ha_read(thd, first_table, lex->ha_read_mode, lex->ident.str,
                       lex->insert_list, lex->ha_rkey_mode, select_lex->where,
                       unit->select_limit_cnt, unit->offset_limit_cnt);
    break;

  case SQLCOM_BEGIN:
    DBUG_PRINT("info", ("Executing SQLCOM_BEGIN  thd: %p", thd));
    if (trans_begin(thd, lex->start_transaction_opt))
    {
      thd->release_transactional_locks();
      WSREP_DEBUG("BEGIN failed, MDL released: %lld",
                  (longlong) thd->thread_id);
      WSREP_DEBUG("stmt_da, sql_errno: %d", (thd->get_stmt_da()->is_error()) ? thd->get_stmt_da()->sql_errno() : 0);
      goto error;
    }
    my_ok(thd);
    break;
  case SQLCOM_COMMIT:
  {
    DBUG_ASSERT(thd->lock == NULL ||
                thd->locked_tables_mode == LTM_LOCK_TABLES);
    bool tx_chain= (lex->tx_chain == TVL_YES ||
                    (thd->variables.completion_type == 1 &&
                     lex->tx_chain != TVL_NO));
    bool tx_release= (lex->tx_release == TVL_YES ||
                      (thd->variables.completion_type == 2 &&
                       lex->tx_release != TVL_NO));
    bool commit_failed= trans_commit(thd);
    thd->release_transactional_locks();
    if (commit_failed)
    {
      WSREP_DEBUG("COMMIT failed, MDL released: %lld",
                  (longlong) thd->thread_id);
      goto error;
    }
    /* Begin transaction with the same isolation level. */
    if (tx_chain)
    {
      if (trans_begin(thd))
        goto error;
    }
    else
    {
      /* Reset the isolation level and access mode if no chaining transaction.*/
      trans_reset_one_shot_chistics(thd);
    }
    /* Disconnect the current client connection. */
    if (tx_release)
    {
      thd->set_killed(KILL_CONNECTION);
      thd->print_aborted_warning(3, "RELEASE");
    }
    my_ok(thd);
    break;
  }
  case SQLCOM_ROLLBACK:
  {
    DBUG_ASSERT(thd->lock == NULL ||
                thd->locked_tables_mode == LTM_LOCK_TABLES);
    bool tx_chain= (lex->tx_chain == TVL_YES ||
                    (thd->variables.completion_type == 1 &&
                     lex->tx_chain != TVL_NO));
    bool tx_release= (lex->tx_release == TVL_YES ||
                      (thd->variables.completion_type == 2 &&
                       lex->tx_release != TVL_NO));
    bool rollback_failed= trans_rollback(thd);
    thd->release_transactional_locks();

    if (rollback_failed)
    {
      WSREP_DEBUG("rollback failed, MDL released: %lld",
                  (longlong) thd->thread_id);
      goto error;
    }
    /* Begin transaction with the same isolation level. */
    if (tx_chain)
    {
#ifdef WITH_WSREP
      /* If there are pending changes after rollback we should clear them */
      if (wsrep_on(thd) && wsrep_has_changes(thd))
        wsrep_after_statement(thd);
#endif
      if (trans_begin(thd))
        goto error;
    }
    else
    {
      /* Reset the isolation level and access mode if no chaining transaction.*/
      trans_reset_one_shot_chistics(thd);
    }
    /* Disconnect the current client connection. */
    if (tx_release)
      thd->set_killed(KILL_CONNECTION);
    my_ok(thd);
   break;
  }
  case SQLCOM_RELEASE_SAVEPOINT:
    if (trans_release_savepoint(thd, lex->ident))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_ROLLBACK_TO_SAVEPOINT:
    if (trans_rollback_to_savepoint(thd, lex->ident))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_SAVEPOINT:
    if (trans_savepoint(thd, lex->ident))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_CREATE_PROCEDURE:
  case SQLCOM_CREATE_SPFUNCTION:
  case SQLCOM_CREATE_PACKAGE:
  case SQLCOM_CREATE_PACKAGE_BODY:
  {
    if (mysql_create_routine(thd, lex))
      goto error;
    my_ok(thd);
    break; /* break super switch */
  } /* end case group bracket */
  case SQLCOM_COMPOUND:
    DBUG_ASSERT(all_tables == 0);
    DBUG_ASSERT(thd->in_sub_stmt == 0);
    lex->sphead->m_sql_mode= thd->variables.sql_mode;
    if (do_execute_sp(thd, lex->sphead))
      goto error;
    break;

  case SQLCOM_ALTER_PROCEDURE:
  case SQLCOM_ALTER_FUNCTION:
    {
      int sp_result;
      const Sp_handler *sph= Sp_handler::handler(lex->sql_command);
      if (check_routine_access(thd, ALTER_PROC_ACL, &lex->spname->m_db,
                               &lex->spname->m_name, sph, 0))
        goto error;

      /*
        Note that if you implement the capability of ALTER FUNCTION to
        alter the body of the function, this command should be made to
        follow the restrictions that log-bin-trust-function-creators=0
        already puts on CREATE FUNCTION.
      */
      /* Conditionally writes to binlog */
      sp_result= sph->sp_update_routine(thd, lex->spname, &lex->sp_chistics);
      switch (sp_result)
      {
      case SP_OK:
	my_ok(thd);
	break;
      case SP_KEY_NOT_FOUND:
	my_error(ER_SP_DOES_NOT_EXIST, MYF(0),
                 sph->type_str(), ErrConvDQName(lex->spname).ptr());
	goto error;
      default:
	my_error(ER_SP_CANT_ALTER, MYF(0),
                 sph->type_str(), ErrConvDQName(lex->spname).ptr());
	goto error;
      }
      break;
    }
  case SQLCOM_DROP_PROCEDURE:
  case SQLCOM_DROP_FUNCTION:
  case SQLCOM_DROP_PACKAGE:
  case SQLCOM_DROP_PACKAGE_BODY:
    {
#ifdef HAVE_DLOPEN
      if (lex->sql_command == SQLCOM_DROP_FUNCTION &&
          ! lex->spname->m_explicit_name)
      {
        /* DROP FUNCTION <non qualified name> */
        enum drop_udf_result rc= mysql_drop_function(thd,
                                                     &lex->spname->m_name);
        if (rc == UDF_DEL_RESULT_DELETED)
        {
          my_ok(thd);
          break;
        }

        if (rc == UDF_DEL_RESULT_ERROR)
          goto error;

        DBUG_ASSERT(rc == UDF_DEL_RESULT_ABSENT);

        // If there was no current database, so it can not be SP
        if (lex->spname->m_db.str == NULL)
        {
          if (lex->if_exists())
          {
            push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                                ER_SP_DOES_NOT_EXIST, ER_THD(thd, ER_SP_DOES_NOT_EXIST),
                                "FUNCTION (UDF)", lex->spname->m_name.str);
            res= FALSE;
            my_ok(thd);
            break;
          }
          my_error(ER_SP_DOES_NOT_EXIST, MYF(0),
                   "FUNCTION (UDF)", lex->spname->m_name.str);
          goto error;
        }
        /* Fall thought to test for a stored function */
      }
#endif

      int sp_result;
      const Sp_handler *sph= Sp_handler::handler(lex->sql_command);

      if (check_routine_access(thd, ALTER_PROC_ACL, &lex->spname->m_db, &lex->spname->m_name,
                               Sp_handler::handler(lex->sql_command), 0))
        goto error;
      WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);

      /* Conditionally writes to binlog */
      sp_result= sph->sp_drop_routine(thd, lex->spname);

#ifndef NO_EMBEDDED_ACCESS_CHECKS
      /*
        We're going to issue an implicit REVOKE statement so we close all
        open tables. We have to keep metadata locks as this ensures that
        this statement is atomic against concurent FLUSH TABLES WITH READ
        LOCK. Deadlocks which can arise due to fact that this implicit
        statement takes metadata locks should be detected by a deadlock
        detector in MDL subsystem and reported as errors.

        No need to commit/rollback statement transaction, it's not started.

        TODO: Long-term we should either ensure that implicit REVOKE statement
              is written into binary log as a separate statement or make both
              dropping of routine and implicit REVOKE parts of one fully atomic
              statement.
      */
      DBUG_ASSERT(thd->transaction.stmt.is_empty());
      close_thread_tables(thd);

      if (sp_result != SP_KEY_NOT_FOUND &&
          sp_automatic_privileges && !opt_noacl &&
          sp_revoke_privileges(thd, lex->spname->m_db.str, lex->spname->m_name.str,
                               Sp_handler::handler(lex->sql_command)))
      {
        push_warning(thd, Sql_condition::WARN_LEVEL_WARN,
                     ER_PROC_AUTO_REVOKE_FAIL,
                     ER_THD(thd, ER_PROC_AUTO_REVOKE_FAIL));
        /* If this happens, an error should have been reported. */
        goto error;
      }
#endif

      res= sp_result;
      switch (sp_result) {
      case SP_OK:
	my_ok(thd);
	break;
      case SP_KEY_NOT_FOUND:
	if (lex->if_exists())
	{
          res= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
	  push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
			      ER_SP_DOES_NOT_EXIST, ER_THD(thd, ER_SP_DOES_NOT_EXIST),
                              sph->type_str(),
                              ErrConvDQName(lex->spname).ptr());
          if (!res)
            my_ok(thd);
	  break;
	}
	my_error(ER_SP_DOES_NOT_EXIST, MYF(0),
                 sph->type_str(), ErrConvDQName(lex->spname).ptr());
	goto error;
      default:
	my_error(ER_SP_DROP_FAILED, MYF(0),
                 sph->type_str(), ErrConvDQName(lex->spname).ptr());
	goto error;
      }
      break;
    }
  case SQLCOM_SHOW_CREATE_PROC:
  case SQLCOM_SHOW_CREATE_FUNC:
  case SQLCOM_SHOW_CREATE_PACKAGE:
  case SQLCOM_SHOW_CREATE_PACKAGE_BODY:
    {
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
      const Sp_handler *sph= Sp_handler::handler(lex->sql_command);
      if (sph->sp_show_create_routine(thd, lex->spname))
        goto error;
      break;
    }
  case SQLCOM_SHOW_PROC_CODE:
  case SQLCOM_SHOW_FUNC_CODE:
  case SQLCOM_SHOW_PACKAGE_BODY_CODE:
    {
#ifndef DBUG_OFF
      Database_qualified_name pkgname(&null_clex_str, &null_clex_str);
      sp_head *sp;
      const Sp_handler *sph= Sp_handler::handler(lex->sql_command);
      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
      if (sph->sp_resolve_package_routine(thd, thd->lex->sphead,
                                          lex->spname, &sph, &pkgname))
        return true;
      if (sph->sp_cache_routine(thd, lex->spname, false, &sp))
        goto error;
      if (!sp || sp->show_routine_code(thd))
      {
        /* We don't distinguish between errors for now */
        my_error(ER_SP_DOES_NOT_EXIST, MYF(0),
                 sph->type_str(), lex->spname->m_name.str);
        goto error;
      }
      break;
#else
      my_error(ER_FEATURE_DISABLED, MYF(0),
               "SHOW PROCEDURE|FUNCTION CODE", "--with-debug");
      goto error;
#endif // ifndef DBUG_OFF
    }
  case SQLCOM_SHOW_CREATE_TRIGGER:
    {
      if (check_ident_length(&lex->spname->m_name))
        goto error;

      WSREP_SYNC_WAIT(thd, WSREP_SYNC_WAIT_BEFORE_SHOW);
      if (show_create_trigger(thd, lex->spname))
        goto error; /* Error has been already logged. */

      break;
    }
  case SQLCOM_CREATE_VIEW:
    {
      /*
        Note: SQLCOM_CREATE_VIEW also handles 'ALTER VIEW' commands
        as specified through the thd->lex->create_view->mode flag.
      */
      res= mysql_create_view(thd, first_table, thd->lex->create_view->mode);
      break;
    }
  case SQLCOM_DROP_VIEW:
    {
      if (check_table_access(thd, DROP_ACL, all_tables, FALSE, UINT_MAX, FALSE))
        goto error;
      /* Conditionally writes to binlog. */
      WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);
      res= mysql_drop_view(thd, first_table, thd->lex->drop_mode);
      break;
    }
  case SQLCOM_CREATE_TRIGGER:
  {
    /* Conditionally writes to binlog. */
    res= mysql_create_or_drop_trigger(thd, all_tables, 1);

    break;
  }
  case SQLCOM_DROP_TRIGGER:
  {
    /* Conditionally writes to binlog. */
    res= mysql_create_or_drop_trigger(thd, all_tables, 0);
    break;
  }
  case SQLCOM_XA_START:
#ifdef WITH_WSREP
    if (WSREP_ON)
    {
      my_error(ER_NOT_SUPPORTED_YET, MYF(0),
               "XA transactions with Galera replication");
      break;
    }
#endif /* WITH_WSREP */
    if (trans_xa_start(thd))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_XA_END:
    if (trans_xa_end(thd))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_XA_PREPARE:
    if (trans_xa_prepare(thd))
      goto error;
    my_ok(thd);
    break;
  case SQLCOM_XA_COMMIT:
  {
    bool commit_failed= trans_xa_commit(thd);
    if (commit_failed)
    {
      WSREP_DEBUG("XA commit failed, MDL released: %lld",
                  (longlong) thd->thread_id);
      goto error;
    }
    /*
      We've just done a commit, reset transaction
      isolation level and access mode to the session default.
    */
    trans_reset_one_shot_chistics(thd);
    my_ok(thd);
    break;
  }
  case SQLCOM_XA_ROLLBACK:
  {
    bool rollback_failed= trans_xa_rollback(thd);
    if (rollback_failed)
    {
      WSREP_DEBUG("XA rollback failed, MDL released: %lld",
                  (longlong) thd->thread_id);
      goto error;
    }
    /*
      We've just done a rollback, reset transaction
      isolation level and access mode to the session default.
    */
    trans_reset_one_shot_chistics(thd);
    my_ok(thd);
    break;
  }
  case SQLCOM_XA_RECOVER:
    res= mysql_xa_recover(thd);
    break;
  case SQLCOM_ALTER_TABLESPACE:
    if (check_global_access(thd, CREATE_TABLESPACE_ACL))
      break;
    if (!(res= mysql_alter_tablespace(thd, lex->alter_tablespace_info)))
      my_ok(thd);
    break;
  case SQLCOM_INSTALL_PLUGIN:
    if (! (res= mysql_install_plugin(thd, &thd->lex->comment,
                                     &thd->lex->ident)))
      my_ok(thd);
    break;
  case SQLCOM_UNINSTALL_PLUGIN:
    if (! (res= mysql_uninstall_plugin(thd, &thd->lex->comment,
                                       &thd->lex->ident)))
      my_ok(thd);
    break;
  case SQLCOM_BINLOG_BASE64_EVENT:
  {
#ifndef EMBEDDED_LIBRARY
    mysql_client_binlog_statement(thd);
#else /* EMBEDDED_LIBRARY */
    my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "embedded");
#endif /* EMBEDDED_LIBRARY */
    break;
  }
  case SQLCOM_CREATE_SERVER:
  {
    DBUG_PRINT("info", ("case SQLCOM_CREATE_SERVER"));

    if (check_global_access(thd, SUPER_ACL))
      break;

    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);

    res= create_server(thd, &lex->server_options);
    break;
  }
  case SQLCOM_ALTER_SERVER:
  {
    int error;
    DBUG_PRINT("info", ("case SQLCOM_ALTER_SERVER"));

    if (check_global_access(thd, SUPER_ACL))
      break;

    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);

    if (unlikely((error= alter_server(thd, &lex->server_options))))
    {
      DBUG_PRINT("info", ("problem altering server <%s>",
                          lex->server_options.server_name.str));
      my_error(error, MYF(0), lex->server_options.server_name.str);
      break;
    }
    my_ok(thd, 1);
    break;
  }
  case SQLCOM_DROP_SERVER:
  {
    int err_code;
    DBUG_PRINT("info", ("case SQLCOM_DROP_SERVER"));

    if (check_global_access(thd, SUPER_ACL))
      break;

    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL);

    if ((err_code= drop_server(thd, &lex->server_options)))
    {
      if (! lex->if_exists() && err_code == ER_FOREIGN_SERVER_DOESNT_EXIST)
      {
        DBUG_PRINT("info", ("problem dropping server %s",
                            lex->server_options.server_name.str));
        my_error(err_code, MYF(0), lex->server_options.server_name.str);
      }
      else
      {
        my_ok(thd, 0);
      }
      break;
    }
    my_ok(thd, 1);
    break;
  }
  case SQLCOM_ANALYZE:
  case SQLCOM_CHECK:
  case SQLCOM_OPTIMIZE:
  case SQLCOM_REPAIR:
  case SQLCOM_TRUNCATE:
  case SQLCOM_CREATE_TABLE:
  case SQLCOM_CREATE_SEQUENCE:
  case SQLCOM_ALTER_TABLE:
      DBUG_ASSERT(first_table == all_tables && first_table != 0);
    /* fall through */
  case SQLCOM_ALTER_SEQUENCE:
  case SQLCOM_SIGNAL:
  case SQLCOM_RESIGNAL:
  case SQLCOM_GET_DIAGNOSTICS:
  case SQLCOM_CALL:
    DBUG_ASSERT(lex->m_sql_cmd != NULL);
    res= lex->m_sql_cmd->execute(thd);
    break;
  default:

#ifndef EMBEDDED_LIBRARY
    DBUG_ASSERT(0);                             /* Impossible */
#endif
    my_ok(thd);
    break;
  }
  THD_STAGE_INFO(thd, stage_query_end);
  thd->update_stats();

  goto finish;

error:
#ifdef WITH_WSREP
wsrep_error_label:
#endif
  res= true;

finish:

  thd->reset_query_timer();
  DBUG_ASSERT(!thd->in_active_multi_stmt_transaction() ||
               thd->in_multi_stmt_transaction_mode());

  lex->unit.cleanup();

  /* close/reopen tables that were marked to need reopen under LOCK TABLES */
  if (unlikely(thd->locked_tables_list.some_table_marked_for_reopen) &&
      !thd->lex->requires_prelocking())
    thd->locked_tables_list.reopen_tables(thd, true);

  if (! thd->in_sub_stmt)
  {
    if (thd->killed != NOT_KILLED)
    {
      /* report error issued during command execution */
      if (thd->killed_errno())
      {
        /* If we already sent 'ok', we can ignore any kill query statements */
        if (! thd->get_stmt_da()->is_set())
          thd->send_kill_message();
      }
      thd->reset_kill_query();
    }
    if (unlikely(thd->is_error()) ||
        (thd->variables.option_bits & OPTION_MASTER_SQL_ERROR))
    {
      THD_STAGE_INFO(thd, stage_rollback);
      trans_rollback_stmt(thd);
    }
    else
    {
      /* If commit fails, we should be able to reset the OK status. */
      THD_STAGE_INFO(thd, stage_commit);
      thd->get_stmt_da()->set_overwrite_status(true);
      trans_commit_stmt(thd);
      thd->get_stmt_da()->set_overwrite_status(false);
    }
    ha_maria_implicit_commit(thd, FALSE);
  }

  /* Free tables. Set stage 'closing tables' */
  close_thread_tables(thd);


#ifndef DBUG_OFF
  if (lex->sql_command != SQLCOM_SET_OPTION && ! thd->in_sub_stmt)
    DEBUG_SYNC(thd, "execute_command_after_close_tables");
#endif
  if (!(sql_command_flags[lex->sql_command] &
        (CF_CAN_GENERATE_ROW_EVENTS | CF_FORCE_ORIGINAL_BINLOG_FORMAT |
         CF_STATUS_COMMAND)))
    thd->set_binlog_format(orig_binlog_format,
                           orig_current_stmt_binlog_format);

  if (! thd->in_sub_stmt && thd->transaction_rollback_request)
  {
    /*
      We are not in sub-statement and transaction rollback was requested by
      one of storage engines (e.g. due to deadlock). Rollback transaction in
      all storage engines including binary log.
    */
    THD_STAGE_INFO(thd, stage_rollback_implicit);
    trans_rollback_implicit(thd);
    thd->release_transactional_locks();
  }
  else if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
  {
    /* No transaction control allowed in sub-statements. */
    DBUG_ASSERT(! thd->in_sub_stmt);
    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
    {
      THD_STAGE_INFO(thd, stage_commit_implicit);
      /* If commit fails, we should be able to reset the OK status. */
      thd->get_stmt_da()->set_overwrite_status(true);
      /* Commit the normal transaction if one is active. */
      trans_commit_implicit(thd);
      thd->get_stmt_da()->set_overwrite_status(false);
      thd->release_transactional_locks();
    }
  }
  else if (! thd->in_sub_stmt && ! thd->in_multi_stmt_transaction_mode())
  {
    /*
      - If inside a multi-statement transaction,
      defer the release of metadata locks until the current
      transaction is either committed or rolled back. This prevents
      other statements from modifying the table for the entire
      duration of this transaction.  This provides commit ordering
      and guarantees serializability across multiple transactions.
      - If in autocommit mode, or outside a transactional context,
      automatically release metadata locks of the current statement.
    */
    thd->release_transactional_locks();
  }
  else if (! thd->in_sub_stmt)
  {
    thd->mdl_context.release_statement_locks();
  }

  THD_STAGE_INFO(thd, stage_starting_cleanup);

  TRANSACT_TRACKER(add_trx_state_from_thd(thd));

#ifdef WITH_WSREP
  thd->wsrep_consistency_check= NO_CONSISTENCY_CHECK;
  
  WSREP_TO_ISOLATION_END;
  /*
    Force release of transactional locks if not in active MST and wsrep is on.
  */
  if (WSREP(thd) &&
      ! thd->in_sub_stmt &&
      ! thd->in_active_multi_stmt_transaction() &&
      thd->mdl_context.has_transactional_locks())
  {
    WSREP_DEBUG("Forcing release of transactional locks for thd: %lld",
                (longlong) thd->thread_id);
    thd->release_transactional_locks();
  }

  /*
    Current command did not start multi STMT transaction and the command
    did not cause commit to happen (e.g. read only). Commit the wsrep
    transaction as empty.
  */
  if (!thd->in_active_multi_stmt_transaction() &&
      !thd->in_sub_stmt &&
      thd->wsrep_trx().active() &&
      thd->wsrep_trx().state() == wsrep::transaction::s_executing)
  {
    wsrep_commit_empty(thd, true);
  }

  /* assume PA safety for next transaction */
  thd->wsrep_PA_safe= true;
#endif /* WITH_WSREP */

  DBUG_RETURN(res || thd->is_error());
 }