~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2006 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 */

/*****************************************************************************

**

** This file implements classes defined in sql_class.h

** Especially the classes to handle a result from a select

**

*****************************************************************************/

#ifdef USE_PRAGMA_IMPLEMENTATION

#pragma implementation				// gcc: Class implementation

#endif

#include "mysql_priv.h"

#include "rpl_rli.h"

#include "rpl_record.h"

#include "slave.h"

#include <my_bitmap.h>

#include "log_event.h"

#include <m_ctype.h>

#include <sys/stat.h>

#include <thr_alarm.h>

#include <mysys_err.h>

/*

  The following is used to initialise Table_ident with a internal

  table name

*/

char internal_table_name[2]= "*";

char empty_c_string[1]= {0};    /* used for not defined db */

const char * const THD::DEFAULT_WHERE= "field list";

/*****************************************************************************

** Instansiate templates

*****************************************************************************/

#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION

/* Used templates */

template class List<Key>;

template class List_iterator<Key>;

template class List<Key_part_spec>;

template class List_iterator<Key_part_spec>;

template class List<Alter_drop>;

template class List_iterator<Alter_drop>;

template class List<Alter_column>;

template class List_iterator<Alter_column>;

#endif

/****************************************************************************

** User variables

****************************************************************************/

extern "C" uchar *get_var_key(user_var_entry *entry, size_t *length,

                              my_bool not_used __attribute__((unused)))

{

  *length= entry->name.length;

  return (uchar*) entry->name.str;

}

extern "C" void free_user_var(user_var_entry *entry)

{

  char *pos= (char*) entry+ALIGN_SIZE(sizeof(*entry));

  if (entry->value && entry->value != pos)

    my_free(entry->value, MYF(0));

  my_free((char*) entry,MYF(0));

}

bool Key_part_spec::operator==(const Key_part_spec& other) const

{

  return length == other.length &&

         field_name.length == other.field_name.length &&

         !strcmp(field_name.str, other.field_name.str);

}

/**

  Construct an (almost) deep copy of this key. Only those

  elements that are known to never change are not copied.

  If out of memory, a partial copy is returned and an error is set

  in THD.

*/

Key::Key(const Key &rhs, MEM_ROOT *mem_root)

  :type(rhs.type),

  key_create_info(rhs.key_create_info),

  columns(rhs.columns, mem_root),

  name(rhs.name),

  generated(rhs.generated)

{

  list_copy_and_replace_each_value(columns, mem_root);

}

/**

  Construct an (almost) deep copy of this foreign key. Only those

  elements that are known to never change are not copied.

  If out of memory, a partial copy is returned and an error is set

  in THD.

*/

Foreign_key::Foreign_key(const Foreign_key &rhs, MEM_ROOT *mem_root)

  :Key(rhs),

  ref_table(rhs.ref_table),

  ref_columns(rhs.ref_columns),

  delete_opt(rhs.delete_opt),

  update_opt(rhs.update_opt),

  match_opt(rhs.match_opt)

{

  list_copy_and_replace_each_value(ref_columns, mem_root);

}

/*

  Test if a foreign key (= generated key) is a prefix of the given key

  (ignoring key name, key type and order of columns)

  NOTES:

    This is only used to test if an index for a FOREIGN KEY exists

  IMPLEMENTATION

    We only compare field names

  RETURN

    0	Generated key is a prefix of other key

    1	Not equal

*/

bool foreign_key_prefix(Key *a, Key *b)

{

  /* Ensure that 'a' is the generated key */

  if (a->generated)

  {

    if (b->generated && a->columns.elements > b->columns.elements)

      swap_variables(Key*, a, b);               // Put shorter key in 'a'

  }

  else

  {

    if (!b->generated)

      return TRUE;                              // No foreign key

    swap_variables(Key*, a, b);                 // Put generated key in 'a'

  }

  /* Test if 'a' is a prefix of 'b' */

  if (a->columns.elements > b->columns.elements)

    return TRUE;                                // Can't be prefix

  List_iterator<Key_part_spec> col_it1(a->columns);

  List_iterator<Key_part_spec> col_it2(b->columns);

  const Key_part_spec *col1, *col2;

#ifdef ENABLE_WHEN_INNODB_CAN_HANDLE_SWAPED_FOREIGN_KEY_COLUMNS

  while ((col1= col_it1++))

  {

    bool found= 0;

    col_it2.rewind();

    while ((col2= col_it2++))

    {

      if (*col1 == *col2)

      {

        found= TRUE;

	break;

      }

    }

    if (!found)

      return TRUE;                              // Error

  }

  return FALSE;                                 // Is prefix

#else

  while ((col1= col_it1++))

  {

    col2= col_it2++;

    if (!(*col1 == *col2))

      return TRUE;

  }

  return FALSE;                                 // Is prefix

#endif

}

/****************************************************************************

** Thread specific functions

****************************************************************************/

Open_tables_state::Open_tables_state(ulong version_arg)

  :version(version_arg), state_flags(0U)

{

  reset_open_tables_state();

}

/*

  The following functions form part of the C plugin API

*/

extern "C" int mysql_tmpfile(const char *prefix)

{

  char filename[FN_REFLEN];

  File fd = create_temp_file(filename, mysql_tmpdir, prefix,

                             O_CREAT | O_EXCL | O_RDWR | O_TEMPORARY,

                             MYF(MY_WME));

  if (fd >= 0) {

    /*

      This can be removed once the following bug is fixed:

      Bug #28903  create_temp_file() doesn't honor O_TEMPORARY option

                  (file not removed) (Unix)

    */

    unlink(filename);

  }

  return fd;

}

extern "C"

int thd_in_lock_tables(const THD *thd)

{

  return test(thd->in_lock_tables);

}

extern "C"

int thd_tablespace_op(const THD *thd)

{

  return test(thd->tablespace_op);

}

extern "C"

const char *set_thd_proc_info(THD *thd, const char *info, 

                              const char *calling_function, 

                              const char *calling_file, 

                              const unsigned int calling_line)

{

  const char *old_info= thd->proc_info;

  DBUG_PRINT("proc_info", ("%s:%d  %s", calling_file, calling_line, 

                           (info != NULL) ? info : "(null)"));

  thd->proc_info= info;

  return old_info;

}

extern "C"

void **thd_ha_data(const THD *thd, const struct handlerton *hton)

{

  return (void **) &thd->ha_data[hton->slot].ha_ptr;

}

extern "C"

long long thd_test_options(const THD *thd, long long test_options)

{

  return thd->options & test_options;

}

extern "C"

int thd_sql_command(const THD *thd)

{

  return (int) thd->lex->sql_command;

}

extern "C"

int thd_tx_isolation(const THD *thd)

{

  return (int) thd->variables.tx_isolation;

}

extern "C"

void thd_inc_row_count(THD *thd)

{

  thd->row_count++;

}

/*

  Dumps a text description of a thread, its security context

  (user, host) and the current query.

  SYNOPSIS

    thd_security_context()

    thd                 current thread context

    buffer              pointer to preferred result buffer

    length              length of buffer

    max_query_len       how many chars of query to copy (0 for all)

  RETURN VALUES

    pointer to string

*/

extern "C"

char *thd_security_context(THD *thd, char *buffer, unsigned int length,

                           unsigned int max_query_len)

{

  String str(buffer, length, &my_charset_latin1);

  const Security_context *sctx= &thd->main_security_ctx;

  char header[64];

  int len;

  len= my_snprintf(header, sizeof(header),

                   "MySQL thread id %lu, query id %lu",

                   thd->thread_id, (ulong) thd->query_id);

  str.length(0);

  str.append(header, len);

  if (sctx->host)

  {

    str.append(' ');

    str.append(sctx->host);

  }

  if (sctx->ip)

  {

    str.append(' ');

    str.append(sctx->ip);

  }

  if (sctx->user)

  {

    str.append(' ');

    str.append(sctx->user);

  }

  if (thd->proc_info)

  {

    str.append(' ');

    str.append(thd->proc_info);

  }

  if (thd->query)

  {

    if (max_query_len < 1)

      len= thd->query_length;

    else

      len= min(thd->query_length, max_query_len);

    str.append('\n');

    str.append(thd->query, len);

  }

  if (str.c_ptr_safe() == buffer)

    return buffer;

  return thd->strmake(str.ptr(), str.length());

}

/**

  Clear this diagnostics area. 

  Normally called at the end of a statement.

*/

void

Diagnostics_area::reset_diagnostics_area()

{

#ifdef DBUG_OFF

  can_overwrite_status= FALSE;

  /** Don't take chances in production */

  m_message[0]= '\0';

  m_sql_errno= 0;

  m_server_status= 0;

  m_affected_rows= 0;

  m_last_insert_id= 0;

  m_total_warn_count= 0;

#endif

  is_sent= FALSE;

  /** Tiny reset in debug mode to see garbage right away */

  m_status= DA_EMPTY;

}

/**

  Set OK status -- ends commands that do not return a

  result set, e.g. INSERT/UPDATE/DELETE.

*/

void

Diagnostics_area::set_ok_status(THD *thd, ha_rows affected_rows_arg,

                                ulonglong last_insert_id_arg,

                                const char *message_arg)

{

  DBUG_ASSERT(! is_set());

#ifdef DBUG_OFF

  /*

    In production, refuse to overwrite an error or a custom response

    with an OK packet.

  */

  if (is_error() || is_disabled())

    return;

#endif

  /** Only allowed to report success if has not yet reported an error */

  m_server_status= thd->server_status;

  m_total_warn_count= thd->total_warn_count;

  m_affected_rows= affected_rows_arg;

  m_last_insert_id= last_insert_id_arg;

  if (message_arg)

    strmake(m_message, message_arg, sizeof(m_message) - 1);

  else

    m_message[0]= '\0';

  m_status= DA_OK;

}

/**

  Set EOF status.

*/

void

Diagnostics_area::set_eof_status(THD *thd)

{

  /** Only allowed to report eof if has not yet reported an error */

  DBUG_ASSERT(! is_set());

#ifdef DBUG_OFF

  /*

    In production, refuse to overwrite an error or a custom response

    with an EOF packet.

  */

  if (is_error() || is_disabled())

    return;

#endif

  m_server_status= thd->server_status;

  /*

    If inside a stored procedure, do not return the total

    number of warnings, since they are not available to the client

    anyway.

  */

  m_total_warn_count= thd->total_warn_count;

  m_status= DA_EOF;

}

/**

  Set ERROR status.

*/

void

Diagnostics_area::set_error_status(THD *thd, uint sql_errno_arg,

                                   const char *message_arg)

{

  /*

    Only allowed to report error if has not yet reported a success

    The only exception is when we flush the message to the client,

    an error can happen during the flush.

  */

  DBUG_ASSERT(! is_set() || can_overwrite_status);

#ifdef DBUG_OFF

  /*

    In production, refuse to overwrite a custom response with an

    ERROR packet.

  */

  if (is_disabled())

    return;

#endif

  m_sql_errno= sql_errno_arg;

  strmake(m_message, message_arg, sizeof(m_message) - 1);

  m_status= DA_ERROR;

}

/**

  Mark the diagnostics area as 'DISABLED'.

  This is used in rare cases when the COM_ command at hand sends a response

  in a custom format. One example is the query cache, another is

  COM_STMT_PREPARE.

*/

void

Diagnostics_area::disable_status()

{

  DBUG_ASSERT(! is_set());

  m_status= DA_DISABLED;

}

THD::THD()

   :Statement(&main_lex, &main_mem_root, CONVENTIONAL_EXECUTION,

              /* statement id */ 0),

   Open_tables_state(refresh_version), rli_fake(0),

   lock_id(&main_lock_id),

   user_time(0), in_sub_stmt(0),

   binlog_table_maps(0), binlog_flags(0UL),

   arg_of_last_insert_id_function(FALSE),

   first_successful_insert_id_in_prev_stmt(0),

   first_successful_insert_id_in_prev_stmt_for_binlog(0),

   first_successful_insert_id_in_cur_stmt(0),

   stmt_depends_on_first_successful_insert_id_in_prev_stmt(FALSE),

   global_read_lock(0),

   is_fatal_error(0),

   transaction_rollback_request(0),

   is_fatal_sub_stmt_error(0),

   rand_used(0),

   time_zone_used(0),

   in_lock_tables(0),

   bootstrap(0),

   derived_tables_processing(FALSE),

   m_lip(NULL),

  /*

    @todo The following is a work around for online backup and the DDL blocker.

          It should be removed when the generalized solution is in place.

          This is needed to ensure the restore (which uses DDL) is not blocked

          when the DDL blocker is engaged.

  */

   DDL_exception(FALSE)

{

  ulong tmp;

  /*

    Pass nominal parameters to init_alloc_root only to ensure that

    the destructor works OK in case of an error. The main_mem_root

    will be re-initialized in init_for_queries().

  */

  init_sql_alloc(&main_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);

  stmt_arena= this;

  thread_stack= 0;

  catalog= (char*)"std"; // the only catalog we have for now

  main_security_ctx.init();

  security_ctx= &main_security_ctx;

  some_tables_deleted=no_errors=password= 0;

  query_start_used= 0;

  count_cuted_fields= CHECK_FIELD_IGNORE;

  killed= NOT_KILLED;

  col_access=0;

  is_slave_error= thread_specific_used= FALSE;

  hash_clear(&handler_tables_hash);

  tmp_table=0;

  used_tables=0;

  cuted_fields= sent_row_count= row_count= 0L;

  limit_found_rows= 0;

  row_count_func= -1;

  statement_id_counter= 0UL;

  // Must be reset to handle error with THD's created for init of mysqld

  lex->current_select= 0;

  start_time=(time_t) 0;

  start_utime= 0L;

  utime_after_lock= 0L;

  current_linfo =  0;

  slave_thread = 0;

  bzero(&variables, sizeof(variables));

  thread_id= 0;

  one_shot_set= 0;

  file_id = 0;

  query_id= 0;

  warn_id= 0;

  db_charset= global_system_variables.collation_database;

  bzero(ha_data, sizeof(ha_data));

  mysys_var=0;

  binlog_evt_union.do_union= FALSE;

  enable_slow_log= 0;

#ifndef DBUG_OFF

  dbug_sentry=THD_SENTRY_MAGIC;

#endif

  net.vio=0;

  client_capabilities= 0;                       // minimalistic client

  system_thread= NON_SYSTEM_THREAD;

  cleanup_done= abort_on_warning= no_warnings_for_error= 0;

  peer_port= 0;					// For SHOW PROCESSLIST

  transaction.m_pending_rows_event= 0;

  transaction.on= 1;

#ifdef SIGNAL_WITH_VIO_CLOSE

  active_vio = 0;

#endif

  pthread_mutex_init(&LOCK_delete, MY_MUTEX_INIT_FAST);

  /* Variables with default values */

  proc_info="login";

  where= THD::DEFAULT_WHERE;

  server_id = ::server_id;

  slave_net = 0;

  command=COM_CONNECT;

  *scramble= '\0';

  init();

  /* Initialize sub structures */

  init_sql_alloc(&warn_root, WARN_ALLOC_BLOCK_SIZE, WARN_ALLOC_PREALLOC_SIZE);

  user_connect=(USER_CONN *)0;

  hash_init(&user_vars, system_charset_info, USER_VARS_HASH_SIZE, 0, 0,

	    (hash_get_key) get_var_key,

	    (hash_free_key) free_user_var, 0);

  /* For user vars replication*/

  if (opt_bin_log)

    my_init_dynamic_array(&user_var_events,

			  sizeof(BINLOG_USER_VAR_EVENT *), 16, 16);

  else

    bzero((char*) &user_var_events, sizeof(user_var_events));

  /* Protocol */

  protocol= &protocol_text;			// Default protocol

  protocol_text.init(this);

  tablespace_op=FALSE;

  tmp= sql_rnd_with_mutex();

  randominit(&rand, tmp + (ulong) &rand, tmp + (ulong) ::global_query_id);

  substitute_null_with_insert_id = FALSE;

  thr_lock_info_init(&lock_info); /* safety: will be reset after start */

  thr_lock_owner_init(&main_lock_id, &lock_info);

  m_internal_handler= NULL;

}

void THD::push_internal_handler(Internal_error_handler *handler)

{

  /*

    TODO: The current implementation is limited to 1 handler at a time only.

    THD and sp_rcontext need to be modified to use a common handler stack.

  */

  DBUG_ASSERT(m_internal_handler == NULL);

  m_internal_handler= handler;

}

bool THD::handle_error(uint sql_errno, const char *message,

                       MYSQL_ERROR::enum_warning_level level)

{

  if (m_internal_handler)

  {

    return m_internal_handler->handle_error(sql_errno, message, level, this);

  }

  return FALSE;                                 // 'FALSE', as per coding style

}

void THD::pop_internal_handler()

{

  DBUG_ASSERT(m_internal_handler != NULL);

  m_internal_handler= NULL;

}

extern "C"

void *thd_alloc(MYSQL_THD thd, unsigned int size)

{

  return thd->alloc(size);

}

extern "C"

void *thd_calloc(MYSQL_THD thd, unsigned int size)

{

  return thd->calloc(size);

}

extern "C"

char *thd_strdup(MYSQL_THD thd, const char *str)

{

  return thd->strdup(str);

}

extern "C"

char *thd_strmake(MYSQL_THD thd, const char *str, unsigned int size)

{

  return thd->strmake(str, size);

}

extern "C"

LEX_STRING *thd_make_lex_string(THD *thd, LEX_STRING *lex_str,

                                const char *str, unsigned int size,

                                int allocate_lex_string)

{

  return thd->make_lex_string(lex_str, str, size,

                              (bool) allocate_lex_string);

}

extern "C"

void *thd_memdup(MYSQL_THD thd, const void* str, unsigned int size)

{

  return thd->memdup(str, size);

}

extern "C"

void thd_get_xid(const MYSQL_THD thd, MYSQL_XID *xid)

{

  *xid = *(MYSQL_XID *) &thd->transaction.xid_state.xid;

}

/*

  Init common variables that has to be reset on start and on change_user

*/

void THD::init(void)

{

  pthread_mutex_lock(&LOCK_global_system_variables);

  plugin_thdvar_init(this);

  variables.time_format= date_time_format_copy((THD*) 0,

					       variables.time_format);

  variables.date_format= date_time_format_copy((THD*) 0,

					       variables.date_format);

  variables.datetime_format= date_time_format_copy((THD*) 0,

						   variables.datetime_format);

  /*

    variables= global_system_variables above has reset

    variables.pseudo_thread_id to 0. We need to correct it here to

    avoid temporary tables replication failure.

  */

  variables.pseudo_thread_id= thread_id;

  pthread_mutex_unlock(&LOCK_global_system_variables);

  server_status= SERVER_STATUS_AUTOCOMMIT;

  options= thd_startup_options;

  if (variables.max_join_size == HA_POS_ERROR)

    options |= OPTION_BIG_SELECTS;

  else

    options &= ~OPTION_BIG_SELECTS;

  transaction.all.modified_non_trans_table= transaction.stmt.modified_non_trans_table= FALSE;

  open_options=ha_open_options;

  update_lock_default= (variables.low_priority_updates ?

			TL_WRITE_LOW_PRIORITY :

			TL_WRITE);

  session_tx_isolation= (enum_tx_isolation) variables.tx_isolation;

  warn_list.empty();

  bzero((char*) warn_count, sizeof(warn_count));

  total_warn_count= 0;

  update_charset();

  reset_current_stmt_binlog_row_based();

  bzero((char *) &status_var, sizeof(status_var));

}

/*

  Init THD for query processing.

  This has to be called once before we call mysql_parse.

  See also comments in sql_class.h.

*/

void THD::init_for_queries()

{

  set_time(); 

  ha_enable_transaction(this,TRUE);

  reset_root_defaults(mem_root, variables.query_alloc_block_size,

                      variables.query_prealloc_size);

  reset_root_defaults(&transaction.mem_root,

                      variables.trans_alloc_block_size,

                      variables.trans_prealloc_size);

  transaction.xid_state.xid.null();

  transaction.xid_state.in_thd=1;

}

/* Do operations that may take a long time */

void THD::cleanup(void)

{

  DBUG_ENTER("THD::cleanup");

  DBUG_ASSERT(cleanup_done == 0);

  killed= KILL_CONNECTION;

#ifdef ENABLE_WHEN_BINLOG_WILL_BE_ABLE_TO_PREPARE

  if (transaction.xid_state.xa_state == XA_PREPARED)

  {

#error xid_state in the cache should be replaced by the allocated value

  }

#endif

  {

    ha_rollback(this);

    xid_cache_delete(&transaction.xid_state);

  }

  if (locked_tables)

  {

    lock=locked_tables; locked_tables=0;

    close_thread_tables(this);

  }

  mysql_ha_cleanup(this);

  delete_dynamic(&user_var_events);

  hash_free(&user_vars);

  close_temporary_tables(this);

  my_free((char*) variables.time_format, MYF(MY_ALLOW_ZERO_PTR));

  my_free((char*) variables.date_format, MYF(MY_ALLOW_ZERO_PTR));

  my_free((char*) variables.datetime_format, MYF(MY_ALLOW_ZERO_PTR));

  if (global_read_lock)

    unlock_global_read_lock(this);

  cleanup_done=1;

  DBUG_VOID_RETURN;

}

/*

   Set the proc_info field.

   Do not use this method directly, use THD_SET_PROC_INFO instead.

 */

#ifndef DBUG_OFF

void THD::set_proc_info(const char* file, int line, const char* info)

{

  /*

     Implementation note:

     file and line correspond to the __FILE__ and __LINE__ where

     THD_SET_PROC_INFO was called.

     These two parameters are provided to help instrumenting the code.

   */

  DBUG_PRINT("info", ("THD::set_proc_info(%s, %d, %s)",

                      file, line, (info ? info : NullS)));

  proc_info= info;

}

#endif

THD::~THD()

{

  THD_CHECK_SENTRY(this);

  DBUG_ENTER("~THD()");

  /* Ensure that no one is using THD */

  pthread_mutex_lock(&LOCK_delete);

  pthread_mutex_unlock(&LOCK_delete);

  add_to_status(&global_status_var, &status_var);

  /* Close connection */

  if (net.vio)

  {

    vio_delete(net.vio);

    net_end(&net);

  }

  if (!cleanup_done)

    cleanup();

  ha_close_connection(this);

  plugin_thdvar_cleanup(this);

  DBUG_PRINT("info", ("freeing security context"));

  main_security_ctx.destroy();

  safeFree(db);

  free_root(&warn_root,MYF(0));

  free_root(&transaction.mem_root,MYF(0));

  mysys_var=0;					// Safety (shouldn't be needed)

  pthread_mutex_destroy(&LOCK_delete);

#ifndef DBUG_OFF

  dbug_sentry= THD_SENTRY_GONE;

#endif  

  if (rli_fake)

  {

    delete rli_fake;