~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2003 MySQL AB

   This program is free software; you can redistribute it and/or modify

   it under the terms of the GNU General Public License as published by

   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,

   but WITHOUT ANY WARRANTY; without even the implied warranty of

   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License

   along with this program; if not, write to the Free Software

   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#define MYSQL_LEX 1

#include "mysql_priv.h"

#include "sql_repl.h"

#include "rpl_filter.h"

#include "repl_failsafe.h"

#include <m_ctype.h>

#include <myisam.h>

#include <my_dir.h>

/**

  @defgroup Runtime_Environment Runtime Environment

  @{

*/

static bool execute_sqlcom_select(THD *thd, TABLE_LIST *all_tables);

const char *any_db="*any*";	// Special symbol for check_access

const LEX_STRING command_name[]={

  { C_STRING_WITH_LEN("Sleep") },

  { C_STRING_WITH_LEN("Quit") },

  { C_STRING_WITH_LEN("Init DB") },

  { C_STRING_WITH_LEN("Query") },

  { C_STRING_WITH_LEN("Field List") },

  { C_STRING_WITH_LEN("Create DB") },

  { C_STRING_WITH_LEN("Drop DB") },

  { C_STRING_WITH_LEN("Refresh") },

  { C_STRING_WITH_LEN("Shutdown") },

  { C_STRING_WITH_LEN("Statistics") },

  { C_STRING_WITH_LEN("Processlist") },

  { C_STRING_WITH_LEN("Connect") },

  { C_STRING_WITH_LEN("Kill") },

  { C_STRING_WITH_LEN("Debug") },

  { C_STRING_WITH_LEN("Ping") },

  { C_STRING_WITH_LEN("Time") },

  { C_STRING_WITH_LEN("Delayed insert") },

  { C_STRING_WITH_LEN("Change user") },

  { C_STRING_WITH_LEN("Binlog Dump") },

  { C_STRING_WITH_LEN("Table Dump") },

  { C_STRING_WITH_LEN("Connect Out") },

  { C_STRING_WITH_LEN("Register Slave") },

  { C_STRING_WITH_LEN("Prepare") },

  { C_STRING_WITH_LEN("Execute") },

  { C_STRING_WITH_LEN("Long Data") },

  { C_STRING_WITH_LEN("Close stmt") },

  { C_STRING_WITH_LEN("Reset stmt") },

  { C_STRING_WITH_LEN("Set option") },

  { C_STRING_WITH_LEN("Fetch") },

  { C_STRING_WITH_LEN("Daemon") },

  { C_STRING_WITH_LEN("Error") }  // Last command number

};

const char *xa_state_names[]={

  "NON-EXISTING", "ACTIVE", "IDLE", "PREPARED"

};

static void unlock_locked_tables(THD *thd)

{

  if (thd->locked_tables)

  {

    thd->lock=thd->locked_tables;

    thd->locked_tables=0;			// Will be automatically closed

    close_thread_tables(thd);			// Free tables

  }

}

bool end_active_trans(THD *thd)

{

  int error=0;

  DBUG_ENTER("end_active_trans");

  if (unlikely(thd->in_sub_stmt))

  {

    my_error(ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG, MYF(0));

    DBUG_RETURN(1);

  }

  if (thd->transaction.xid_state.xa_state != XA_NOTR)

  {

    my_error(ER_XAER_RMFAIL, MYF(0),

             xa_state_names[thd->transaction.xid_state.xa_state]);

    DBUG_RETURN(1);

  }

  if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN |

		      OPTION_TABLE_LOCK))

  {

    DBUG_PRINT("info",("options: 0x%llx", thd->options));

    /* Safety if one did "drop table" on locked tables */

    if (!thd->locked_tables)

      thd->options&= ~OPTION_TABLE_LOCK;

    thd->server_status&= ~SERVER_STATUS_IN_TRANS;

    if (ha_commit(thd))

      error=1;

  }

  thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);

  thd->transaction.all.modified_non_trans_table= FALSE;

  DBUG_RETURN(error);

}

bool begin_trans(THD *thd)

{

  int error=0;

  if (unlikely(thd->in_sub_stmt))

  {

    my_error(ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG, MYF(0));

    return 1;

  }

  if (thd->locked_tables)

  {

    thd->lock=thd->locked_tables;

    thd->locked_tables=0;			// Will be automatically closed

    close_thread_tables(thd);			// Free tables

  }

  if (end_active_trans(thd))

    error= -1;

  else

  {

    LEX *lex= thd->lex;

    thd->options|= OPTION_BEGIN;

    thd->server_status|= SERVER_STATUS_IN_TRANS;

    if (lex->start_transaction_opt & MYSQL_START_TRANS_OPT_WITH_CONS_SNAPSHOT)

      error= ha_start_consistent_snapshot(thd);

  }

  return error;

}

#ifdef HAVE_REPLICATION

/**

  Returns true if all tables should be ignored.

*/

inline bool all_tables_not_ok(THD *thd, TABLE_LIST *tables)

{

  return rpl_filter->is_on() && tables &&

         !rpl_filter->tables_ok(thd->db, tables);

}

#endif

static bool some_non_temp_table_to_be_updated(THD *thd, TABLE_LIST *tables)

{

  for (TABLE_LIST *table= tables; table; table= table->next_global)

  {

    DBUG_ASSERT(table->db && table->table_name);

    if (table->updating &&

        !find_temporary_table(thd, table->db, table->table_name))

      return 1;

  }

  return 0;

}

/**

  Mark all commands that somehow changes a table.

  This is used to check number of updates / hour.

  sql_command is actually set to SQLCOM_END sometimes

  so we need the +1 to include it in the array.

  See COMMAND_FLAG_xxx for different type of commands

     2  - query that returns meaningful ROW_COUNT() -

          a number of modified rows

*/

uint sql_command_flags[SQLCOM_END+1];

void init_update_queries(void)

{

  bzero((uchar*) &sql_command_flags, sizeof(sql_command_flags));

  sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_DROP_INDEX]=     CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_DATABASES]=   CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_OPEN_TABLES]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_FIELDS]=      CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_KEYS]=        CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_VARIABLES]=   CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_CHARSETS]=    CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_COLLATIONS]=  CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_NEW_MASTER]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_BINLOGS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_SLAVE_HOSTS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_BINLOG_EVENTS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_WARNS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_ERRORS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_ENGINE_STATUS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_ENGINE_MUTEX]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_ENGINE_LOGS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_PROCESSLIST]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_CREATE_DB]=  CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_CREATE]=  CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_MASTER_STAT]=  CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_SLAVE_STAT]=  CF_STATUS_COMMAND;

   sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |

                                               CF_SHOW_TABLE_COMMAND);

  sql_command_flags[SQLCOM_SHOW_TABLE_STATUS]= (CF_STATUS_COMMAND |

                                                CF_SHOW_TABLE_COMMAND);

  /*

    The following is used to preserver CF_ROW_COUNT during the

    a CALL or EXECUTE statement, so the value generated by the

    last called (or executed) statement is preserved.

    See mysql_execute_command() for how CF_ROW_COUNT is used.

  */

  sql_command_flags[SQLCOM_EXECUTE]= 		CF_HAS_ROW_COUNT;

  /*

    The following admin table operations are allowed

    on log tables.

  */

  sql_command_flags[SQLCOM_REPAIR]=           CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_OPTIMIZE]=         CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_ANALYZE]=          CF_WRITE_LOGS_COMMAND;

}

bool is_update_query(enum enum_sql_command command)

{

  DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);

  return (sql_command_flags[command] & CF_CHANGES_DATA) != 0;

}

void execute_init_command(THD *thd, sys_var_str *init_command_var,

			  rw_lock_t *var_mutex)

{

  Vio* save_vio;

  ulong save_client_capabilities;

  thd_proc_info(thd, "Execution of init_command");

  /*

    We need to lock init_command_var because

    during execution of init_command_var query

    values of init_command_var can't be changed

  */

  rw_rdlock(var_mutex);

  save_client_capabilities= thd->client_capabilities;

  thd->client_capabilities|= CLIENT_MULTI_QUERIES;

  /*

    We don't need return result of execution to client side.

    To forbid this we should set thd->net.vio to 0.

  */

  save_vio= thd->net.vio;

  thd->net.vio= 0;

  dispatch_command(COM_QUERY, thd,

                   init_command_var->value,

                   init_command_var->value_length);

  rw_unlock(var_mutex);

  thd->client_capabilities= save_client_capabilities;

  thd->net.vio= save_vio;

}

/**

  Execute commands from bootstrap_file.

  Used when creating the initial grant tables.

*/

pthread_handler_t handle_bootstrap(void *arg)

{

  THD *thd=(THD*) arg;

  FILE *file=bootstrap_file;

  char *buff;

  const char* found_semicolon= NULL;

  /* The following must be called before DBUG_ENTER */

  thd->thread_stack= (char*) &thd;

  if (my_thread_init() || thd->store_globals())

  {

    close_connection(thd, ER_OUT_OF_RESOURCES, 1);

    thd->fatal_error();

    goto end;

  }

  DBUG_ENTER("handle_bootstrap");

  pthread_detach_this_thread();

  thd->thread_stack= (char*) &thd;

  if (thd->variables.max_join_size == HA_POS_ERROR)

    thd->options |= OPTION_BIG_SELECTS;

  thd_proc_info(thd, 0);

  thd->version=refresh_version;

  thd->security_ctx->priv_user=

    thd->security_ctx->user= (char*) my_strdup("boot", MYF(MY_WME));

  thd->security_ctx->priv_host[0]=0;

  /*

    Make the "client" handle multiple results. This is necessary

    to enable stored procedures with SELECTs and Dynamic SQL

    in init-file.

  */

  thd->client_capabilities|= CLIENT_MULTI_RESULTS;

  buff= (char*) thd->net.buff;

  thd->init_for_queries();

  while (fgets(buff, thd->net.max_packet, file))

  {

    /* strlen() can't be deleted because fgets() doesn't return length */

    ulong length= (ulong) strlen(buff);

    while (buff[length-1] != '\n' && !feof(file))

    {

      /*

        We got only a part of the current string. Will try to increase

        net buffer then read the rest of the current string.

      */

      /* purecov: begin tested */

      if (net_realloc(&(thd->net), 2 * thd->net.max_packet))

      {

        net_end_statement(thd);

        bootstrap_error= 1;

        break;

      }

      buff= (char*) thd->net.buff;

      fgets(buff + length, thd->net.max_packet - length, file);

      length+= (ulong) strlen(buff + length);

      /* purecov: end */

    }

    if (bootstrap_error)

      break;                                    /* purecov: inspected */

    while (length && (my_isspace(thd->charset(), buff[length-1]) ||

                      buff[length-1] == ';'))

      length--;

    buff[length]=0;

    /* Skip lines starting with delimiter */

    if (strncmp(buff, STRING_WITH_LEN("delimiter")) == 0)

      continue;

    thd->query_length=length;

    thd->query= (char*) thd->memdup_w_gap(buff, length+1, 

                                          thd->db_length+1);

    thd->query[length] = '\0';

    DBUG_PRINT("query",("%-.4096s",thd->query));

    /*

      We don't need to obtain LOCK_thread_count here because in bootstrap

      mode we have only one thread.

    */

    thd->query_id=next_query_id();

    thd->set_time();

    mysql_parse(thd, thd->query, length, & found_semicolon);

    close_thread_tables(thd);			// Free tables

    bootstrap_error= thd->is_error();

    net_end_statement(thd);

    if (bootstrap_error)

      break;

    free_root(thd->mem_root,MYF(MY_KEEP_PREALLOC));

    free_root(&thd->transaction.mem_root,MYF(MY_KEEP_PREALLOC));

  }

end:

  net_end(&thd->net);

  thd->cleanup();

  delete thd;

  (void) pthread_mutex_lock(&LOCK_thread_count);

  thread_count--;

  (void) pthread_mutex_unlock(&LOCK_thread_count);

  (void) pthread_cond_broadcast(&COND_thread_count);

  my_thread_end();

  pthread_exit(0);

  DBUG_RETURN(0);

}

/* This works because items are allocated with sql_alloc() */

void free_items(Item *item)

{

  Item *next;

  DBUG_ENTER("free_items");

  for (; item ; item=next)

  {

    next=item->next;

    item->delete_self();

  }

  DBUG_VOID_RETURN;

}

/* This works because items are allocated with sql_alloc() */

void cleanup_items(Item *item)

{

  DBUG_ENTER("cleanup_items");

  for (; item ; item=item->next)

    item->cleanup();

  DBUG_VOID_RETURN;

}

/**

  Ends the current transaction and (maybe) begin the next.

  @param thd            Current thread

  @param completion     Completion type

  @retval

    0   OK

*/

int end_trans(THD *thd, enum enum_mysql_completiontype completion)

{

  bool do_release= 0;

  int res= 0;

  DBUG_ENTER("end_trans");

  if (unlikely(thd->in_sub_stmt))

  {

    my_error(ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG, MYF(0));

    DBUG_RETURN(1);

  }

  if (thd->transaction.xid_state.xa_state != XA_NOTR)

  {

    my_error(ER_XAER_RMFAIL, MYF(0),

             xa_state_names[thd->transaction.xid_state.xa_state]);

    DBUG_RETURN(1);

  }

  switch (completion) {

  case COMMIT:

    /*

     We don't use end_active_trans() here to ensure that this works

     even if there is a problem with the OPTION_AUTO_COMMIT flag

     (Which of course should never happen...)

    */

    thd->server_status&= ~SERVER_STATUS_IN_TRANS;

    res= ha_commit(thd);

    thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);

    thd->transaction.all.modified_non_trans_table= FALSE;

    break;

  case COMMIT_RELEASE:

    do_release= 1; /* fall through */

  case COMMIT_AND_CHAIN:

    res= end_active_trans(thd);

    if (!res && completion == COMMIT_AND_CHAIN)

      res= begin_trans(thd);

    break;

  case ROLLBACK_RELEASE:

    do_release= 1; /* fall through */

  case ROLLBACK:

  case ROLLBACK_AND_CHAIN:

  {

    thd->server_status&= ~SERVER_STATUS_IN_TRANS;

    if (ha_rollback(thd))

      res= -1;

    thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);

    thd->transaction.all.modified_non_trans_table= FALSE;

    if (!res && (completion == ROLLBACK_AND_CHAIN))

      res= begin_trans(thd);

    break;

  }

  default:

    res= -1;

    my_error(ER_UNKNOWN_COM_ERROR, MYF(0));

    DBUG_RETURN(-1);

  }

  if (res < 0)

    my_error(thd->killed_errno(), MYF(0));

  else if ((res == 0) && do_release)

    thd->killed= THD::KILL_CONNECTION;

  DBUG_RETURN(res);

}

/**

  Read one command from connection and execute it (query or simple command).

  This function is called in loop from thread function.

  For profiling to work, it must never be called recursively.

  @retval

    0  success

  @retval

    1  request of thread shutdown (see dispatch_command() description)

*/

bool do_command(THD *thd)

{

  bool return_value;

  char *packet= 0;

  ulong packet_length;

  NET *net= &thd->net;

  enum enum_server_command command;

  DBUG_ENTER("do_command");

  /*

    indicator of uninitialized lex => normal flow of errors handling

    (see my_message_sql)

  */

  thd->lex->current_select= 0;

  /*

    This thread will do a blocking read from the client which

    will be interrupted when the next command is received from

    the client, the connection is closed or "net_wait_timeout"

    number of seconds has passed

  */

  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);

  /*

    XXX: this code is here only to clear possible errors of init_connect. 

    Consider moving to init_connect() instead.

  */

  thd->clear_error();				// Clear error message

  thd->main_da.reset_diagnostics_area();

  net_new_transaction(net);

  packet_length= my_net_read(net);

  if (packet_length == packet_error)

  {

    DBUG_PRINT("info",("Got error %d reading command from socket %s",

		       net->error,

		       vio_description(net->vio)));

    /* Check if we can continue without closing the connection */

    /* The error must be set. */

    DBUG_ASSERT(thd->is_error());

    net_end_statement(thd);

    if (net->error != 3)

    {

      return_value= TRUE;                       // We have to close it.

      goto out;

    }

    net->error= 0;

    return_value= FALSE;

    goto out;

  }

  packet= (char*) net->read_pos;

  /*

    'packet_length' contains length of data, as it was stored in packet

    header. In case of malformed header, my_net_read returns zero.

    If packet_length is not zero, my_net_read ensures that the returned

    number of bytes was actually read from network.

    There is also an extra safety measure in my_net_read:

    it sets packet[packet_length]= 0, but only for non-zero packets.

  */

  if (packet_length == 0)                       /* safety */

  {

    /* Initialize with COM_SLEEP packet */

    packet[0]= (uchar) COM_SLEEP;

    packet_length= 1;

  }

  /* Do not rely on my_net_read, extra safety against programming errors. */

  packet[packet_length]= '\0';                  /* safety */

  command= (enum enum_server_command) (uchar) packet[0];

  if (command >= COM_END)

    command= COM_END;                           // Wrong command

  DBUG_PRINT("info",("Command on %s = %d (%s)",

                     vio_description(net->vio), command,

                     command_name[command].str));

  /* Restore read timeout value */

  my_net_set_read_timeout(net, thd->variables.net_read_timeout);

  DBUG_ASSERT(packet_length);

  return_value= dispatch_command(command, thd, packet+1, (uint) (packet_length-1));

out:

  DBUG_RETURN(return_value);

}

/**

  Determine if an attempt to update a non-temporary table while the

  read-only option was enabled has been made.

  This is a helper function to mysql_execute_command.

  @note SQLCOM_MULTI_UPDATE is an exception and dealt with elsewhere.

  @see mysql_execute_command

  @returns Status code

    @retval TRUE The statement should be denied.

    @retval FALSE The statement isn't updating any relevant tables.

*/

static my_bool deny_updates_if_read_only_option(THD *thd,

                                                TABLE_LIST *all_tables)

{

  DBUG_ENTER("deny_updates_if_read_only_option");

  if (!opt_readonly)

    DBUG_RETURN(FALSE);

  LEX *lex= thd->lex;

  if (!(sql_command_flags[lex->sql_command] & CF_CHANGES_DATA))

    DBUG_RETURN(FALSE);

  /* Multi update is an exception and is dealt with later. */

  if (lex->sql_command == SQLCOM_UPDATE_MULTI)

    DBUG_RETURN(FALSE);

  const my_bool create_temp_tables= 

    (lex->sql_command == SQLCOM_CREATE_TABLE) &&

    (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE);

  const my_bool drop_temp_tables= 

    (lex->sql_command == SQLCOM_DROP_TABLE) &&

    lex->drop_temporary;

  const my_bool update_real_tables=

    some_non_temp_table_to_be_updated(thd, all_tables) &&

    !(create_temp_tables || drop_temp_tables);

  const my_bool create_or_drop_databases=

    (lex->sql_command == SQLCOM_CREATE_DB) ||

    (lex->sql_command == SQLCOM_DROP_DB);

  if (update_real_tables || create_or_drop_databases)

  {

      /*

        An attempt was made to modify one or more non-temporary tables.

      */

      DBUG_RETURN(TRUE);

  }

  /* Assuming that only temporary tables are modified. */

  DBUG_RETURN(FALSE);

}

/**

  Perform one connection-level (COM_XXXX) command.

  @param command         type of command to perform

  @param thd             connection handle

  @param packet          data for the command, packet is always null-terminated

  @param packet_length   length of packet + 1 (to show that data is

                         null-terminated) except for COM_SLEEP, where it

                         can be zero.

  @todo

    set thd->lex->sql_command to SQLCOM_END here.

  @todo

    The following has to be changed to an 8 byte integer

  @retval

    0   ok

  @retval

    1   request of thread shutdown, i. e. if command is

        COM_QUIT/COM_SHUTDOWN

*/

bool dispatch_command(enum enum_server_command command, THD *thd,

		      char* packet, uint packet_length)

{

  NET *net= &thd->net;

  bool error= 0;

  DBUG_ENTER("dispatch_command");

  DBUG_PRINT("info",("packet: '%*.s'; command: %d", packet_length, packet, command));

  thd->command=command;

  /*

    Commands which always take a long time are logged into

    the slow log only if opt_log_slow_admin_statements is set.

  */

  thd->enable_slow_log= TRUE;

  thd->lex->sql_command= SQLCOM_END; /* to avoid confusing VIEW detectors */

  thd->set_time();

  VOID(pthread_mutex_lock(&LOCK_thread_count));

  thd->query_id= global_query_id;

  switch( command ) {

  /* Ignore these statements. */

  case COM_STATISTICS:

  case COM_PING:

    break;

  /* Only increase id on these statements but don't count them. */

  case COM_STMT_PREPARE: 

  case COM_STMT_CLOSE:

  case COM_STMT_RESET:

    next_query_id();

    break;

  /* Increase id and count all other statements. */

  default:

    statistic_increment(thd->status_var.questions, &LOCK_status);

    next_query_id();

  }

  thread_running++;

  /* TODO: set thd->lex->sql_command to SQLCOM_END here */

  VOID(pthread_mutex_unlock(&LOCK_thread_count));

  thd->server_status&=

           ~(SERVER_QUERY_NO_INDEX_USED | SERVER_QUERY_NO_GOOD_INDEX_USED);

  switch (command) {

  case COM_INIT_DB:

  {

    LEX_STRING tmp;

    status_var_increment(thd->status_var.com_stat[SQLCOM_CHANGE_DB]);

    thd->convert_string(&tmp, system_charset_info,

                        packet, packet_length, thd->charset());

    if (!mysql_change_db(thd, &tmp, FALSE))

    {

      general_log_write(thd, command, thd->db, thd->db_length);

      my_ok(thd);

    }

    break;

  }

  case COM_REGISTER_SLAVE:

  {

    if (!register_slave(thd, (uchar*)packet, packet_length))

      my_ok(thd);

    break;

  }

  case COM_CHANGE_USER:

  {

    status_var_increment(thd->status_var.com_other);

    char *user= (char*) packet, *packet_end= packet + packet_length;

    /* Safe because there is always a trailing \0 at the end of the packet */

    char *passwd= strend(user)+1;

    thd->clear_error();                         // if errors from rollback

    /*

      Old clients send null-terminated string ('\0' for empty string) for

      password.  New clients send the size (1 byte) + string (not null

      terminated, so also '\0' for empty string).

      Cast *passwd to an unsigned char, so that it doesn't extend the sign

      for *passwd > 127 and become 2**32-127 after casting to uint.

    */

    char db_buff[NAME_LEN+1];                 // buffer to store db in utf8

    char *db= passwd;

    char *save_db;

    /*

      If there is no password supplied, the packet must contain '\0',

      in any type of handshake (4.1 or pre-4.1).

     */

    if (passwd >= packet_end)

    {

      my_message(ER_UNKNOWN_COM_ERROR, ER(ER_UNKNOWN_COM_ERROR), MYF(0));

      break;

    }

    uint passwd_len= (thd->client_capabilities & CLIENT_SECURE_CONNECTION ?

                      (uchar)(*passwd++) : strlen(passwd));

    uint dummy_errors, save_db_length, db_length;

    int res;

    Security_context save_security_ctx= *thd->security_ctx;

    USER_CONN *save_user_connect;

    db+= passwd_len + 1;

    /*

      Database name is always NUL-terminated, so in case of empty database

      the packet must contain at least the trailing '\0'.

    */

    if (db >= packet_end)

    {

      my_message(ER_UNKNOWN_COM_ERROR, ER(ER_UNKNOWN_COM_ERROR), MYF(0));

      break;

    }

    db_length= strlen(db);

    char *ptr= db + db_length + 1;

    uint cs_number= 0;

    if (ptr < packet_end)

    {

      if (ptr + 2 > packet_end)

      {

        my_message(ER_UNKNOWN_COM_ERROR, ER(ER_UNKNOWN_COM_ERROR), MYF(0));

        break;

      }

      cs_number= uint2korr(ptr);