Skip navigation links

 Systems for MySQL

The world's most popular open source database

Contact a MySQL Representative

Login | Register

• MySQL.com
• Developer Zone
• Partners & Solutions
• Customer Login

• DevZone
• Downloads
• Documentation
• Articles
• Forums
• Bugs
• Forge
• Blogs

 

---

Learn about new MySQL releases, technical articles, events and more.

Subscribe Now »

mysql/src/5.1-dbg/sql/handler.cc

Go to the documentation of this file.

/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult 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; either version 2 of the License, or
   (at your option) any later version.

   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 */


/* Handler-calling-functions */

#ifdef USE_PRAGMA_IMPLEMENTATION
#pragma implementation                          // gcc: Class implementation
#endif

#include "mysql_priv.h"
#include "rpl_filter.h"
#include "ha_heap.h"
#include "ha_myisam.h"
#include "ha_myisammrg.h"


#include <myisampack.h>
#include <errno.h>

#ifdef WITH_NDBCLUSTER_STORAGE_ENGINE
#define NDB_MAX_ATTRIBUTES_IN_TABLE 128
#include "ha_ndbcluster.h"
#endif

#ifdef WITH_PARTITION_STORAGE_ENGINE
#include "ha_partition.h"
#endif

#ifdef WITH_INNOBASE_STORAGE_ENGINE
#include "ha_innodb.h"
#endif

/*
  While we have legacy_db_type, we have this array to
  check for dups and to find handlerton from legacy_db_type.
  Remove when legacy_db_type is finally gone
*/
st_plugin_int *hton2plugin[MAX_HA];

static handlerton *installed_htons[128];

#define BITMAP_STACKBUF_SIZE (128/8)

KEY_CREATE_INFO default_key_create_info= { HA_KEY_ALG_UNDEF, 0, {NullS,0} };

/* static functions defined in this file */

static handler *create_default(TABLE_SHARE *table, MEM_ROOT *mem_root);

static SHOW_COMP_OPTION have_yes= SHOW_OPTION_YES;

/* number of entries in handlertons[] */
ulong total_ha= 0;
/* number of storage engines (from handlertons[]) that support 2pc */
ulong total_ha_2pc= 0;
/* size of savepoint storage area (see ha_init) */
ulong savepoint_alloc_size= 0;

static const LEX_STRING sys_table_aliases[]=
{
  { C_STRING_WITH_LEN("INNOBASE") },  { C_STRING_WITH_LEN("INNODB") },
  { C_STRING_WITH_LEN("NDB") },       { C_STRING_WITH_LEN("NDBCLUSTER") },
  { C_STRING_WITH_LEN("HEAP") },      { C_STRING_WITH_LEN("MEMORY") },
  { C_STRING_WITH_LEN("MERGE") },     { C_STRING_WITH_LEN("MRG_MYISAM") },
  {NullS, 0}
};

const char *ha_row_type[] = {
  "", "FIXED", "DYNAMIC", "COMPRESSED", "REDUNDANT", "COMPACT", "?","?","?"
};

const char *tx_isolation_names[] =
{ "READ-UNCOMMITTED", "READ-COMMITTED", "REPEATABLE-READ", "SERIALIZABLE",
  NullS};
TYPELIB tx_isolation_typelib= {array_elements(tx_isolation_names)-1,"",
                               tx_isolation_names, NULL};

static TYPELIB known_extensions= {0,"known_exts", NULL, NULL};
uint known_extensions_id= 0;


/*
  Return the default storage engine handlerton for thread
  
  SYNOPSIS
    ha_default_handlerton(thd)
    thd         current thread
  
  RETURN
    pointer to handlerton
*/

handlerton *ha_default_handlerton(THD *thd)
{
  return (thd->variables.table_type != NULL) ?
          thd->variables.table_type :
          (global_system_variables.table_type != NULL ?
           global_system_variables.table_type : &myisam_hton);
}


/*
  Return the storage engine handlerton for the supplied name
  
  SYNOPSIS
    ha_resolve_by_name(thd, name)
    thd         current thread
    name        name of storage engine
  
  RETURN
    pointer to handlerton
*/

handlerton *ha_resolve_by_name(THD *thd, const LEX_STRING *name)
{
  const LEX_STRING *table_alias;
  st_plugin_int *plugin;

redo:
  /* my_strnncoll is a macro and gcc doesn't do early expansion of macro */
  if (thd && !my_charset_latin1.coll->strnncoll(&my_charset_latin1,
                           (const uchar *)name->str, name->length,
                           (const uchar *)STRING_WITH_LEN("DEFAULT"), 0))
    return ha_default_handlerton(thd);

  if ((plugin= plugin_lock(name, MYSQL_STORAGE_ENGINE_PLUGIN)))
  {
    handlerton *hton= (handlerton *)plugin->data;
    if (!(hton->flags & HTON_NOT_USER_SELECTABLE))
      return hton;
    plugin_unlock(plugin);
  }

  /*
    We check for the historical aliases.
  */
  for (table_alias= sys_table_aliases; table_alias->str; table_alias+= 2)
  {
    if (!my_strnncoll(&my_charset_latin1,
                      (const uchar *)name->str, name->length,
                      (const uchar *)table_alias->str, table_alias->length))
    {
      name= table_alias + 1;
      goto redo;
    }
  }

  return NULL;
}


const char *ha_get_storage_engine(enum legacy_db_type db_type)
{
  switch (db_type) {
  case DB_TYPE_DEFAULT:
    return "DEFAULT";
  default:
    if (db_type > DB_TYPE_UNKNOWN && db_type < DB_TYPE_DEFAULT &&
        installed_htons[db_type])
      return hton2plugin[installed_htons[db_type]->slot]->name.str;
    /* fall through */
  case DB_TYPE_UNKNOWN:
    return "UNKNOWN";
  }
}


static handler *create_default(TABLE_SHARE *table, MEM_ROOT *mem_root)
{
  handlerton *hton= ha_default_handlerton(current_thd);
  return (hton && hton->create) ? hton->create(table, mem_root) : NULL;
}


handlerton *ha_resolve_by_legacy_type(THD *thd, enum legacy_db_type db_type)
{
  switch (db_type) {
  case DB_TYPE_DEFAULT:
    return ha_default_handlerton(thd);
  case DB_TYPE_UNKNOWN:
    return NULL;
  default:
    if (db_type > DB_TYPE_UNKNOWN && db_type < DB_TYPE_DEFAULT)
      return installed_htons[db_type];
    return NULL;
  }
}


/* Use other database handler if databasehandler is not compiled in */

handlerton *ha_checktype(THD *thd, enum legacy_db_type database_type,
                          bool no_substitute, bool report_error)
{
  handlerton *hton= ha_resolve_by_legacy_type(thd, database_type);
  if (ha_storage_engine_is_enabled(hton))
    return hton;

  if (no_substitute)
  {
    if (report_error)
    {
      const char *engine_name= ha_get_storage_engine(database_type);
      my_error(ER_FEATURE_DISABLED,MYF(0),engine_name,engine_name);
    }
    return NULL;
  }

  switch (database_type) {
#ifndef NO_HASH
  case DB_TYPE_HASH:
    return ha_resolve_by_legacy_type(thd, DB_TYPE_HASH);
#endif
  case DB_TYPE_MRG_ISAM:
    return ha_resolve_by_legacy_type(thd, DB_TYPE_MRG_MYISAM);
  default:
    break;
  }

  return ha_default_handlerton(thd);
} /* ha_checktype */


handler *get_new_handler(TABLE_SHARE *share, MEM_ROOT *alloc,
                         handlerton *db_type)
{
  handler *file;
  DBUG_ENTER("get_new_handler");
  DBUG_PRINT("enter", ("alloc: 0x%lx", (long) alloc));

  if (db_type && db_type->state == SHOW_OPTION_YES && db_type->create)
  {
    if ((file= db_type->create(share, alloc)))
      file->init();
    DBUG_RETURN(file);
  }
  /*
    Try the default table type
    Here the call to current_thd() is ok as we call this function a lot of
    times but we enter this branch very seldom.
  */
  DBUG_RETURN(get_new_handler(share, alloc,
                              current_thd->variables.table_type));
}


#ifdef WITH_PARTITION_STORAGE_ENGINE
handler *get_ha_partition(partition_info *part_info)
{
  ha_partition *partition;
  DBUG_ENTER("get_ha_partition");
  if ((partition= new ha_partition(part_info)))
  {
    if (partition->initialise_partition(current_thd->mem_root))
    {
      delete partition;
      partition= 0;
    }
    else
      partition->init();
  }
  else
  {
    my_error(ER_OUTOFMEMORY, MYF(0), sizeof(ha_partition));
  }
  DBUG_RETURN(((handler*) partition));
}
#endif


/*
  Register handler error messages for use with my_error().

  SYNOPSIS
    ha_init_errors()

  RETURN
    0           OK
    != 0        Error
*/

static int ha_init_errors(void)
{
#define SETMSG(nr, msg) errmsgs[(nr) - HA_ERR_FIRST]= (msg)
  const char    **errmsgs;

  /* Allocate a pointer array for the error message strings. */
  /* Zerofill it to avoid uninitialized gaps. */
  if (! (errmsgs= (const char**) my_malloc(HA_ERR_ERRORS * sizeof(char*),
                                           MYF(MY_WME | MY_ZEROFILL))))
    return 1;

  /* Set the dedicated error messages. */
  SETMSG(HA_ERR_KEY_NOT_FOUND,          ER(ER_KEY_NOT_FOUND));
  SETMSG(HA_ERR_FOUND_DUPP_KEY,         ER(ER_DUP_KEY));
  SETMSG(HA_ERR_RECORD_CHANGED,         "Update wich is recoverable");
  SETMSG(HA_ERR_WRONG_INDEX,            "Wrong index given to function");
  SETMSG(HA_ERR_CRASHED,                ER(ER_NOT_KEYFILE));
  SETMSG(HA_ERR_WRONG_IN_RECORD,        ER(ER_CRASHED_ON_USAGE));
  SETMSG(HA_ERR_OUT_OF_MEM,             "Table handler out of memory");
  SETMSG(HA_ERR_NOT_A_TABLE,            "Incorrect file format '%.64s'");
  SETMSG(HA_ERR_WRONG_COMMAND,          "Command not supported");
  SETMSG(HA_ERR_OLD_FILE,               ER(ER_OLD_KEYFILE));
  SETMSG(HA_ERR_NO_ACTIVE_RECORD,       "No record read in update");
  SETMSG(HA_ERR_RECORD_DELETED,         "Intern record deleted");
  SETMSG(HA_ERR_RECORD_FILE_FULL,       ER(ER_RECORD_FILE_FULL));
  SETMSG(HA_ERR_INDEX_FILE_FULL,        "No more room in index file '%.64s'");
  SETMSG(HA_ERR_END_OF_FILE,            "End in next/prev/first/last");
  SETMSG(HA_ERR_UNSUPPORTED,            ER(ER_ILLEGAL_HA));
  SETMSG(HA_ERR_TO_BIG_ROW,             "Too big row");
  SETMSG(HA_WRONG_CREATE_OPTION,        "Wrong create option");
  SETMSG(HA_ERR_FOUND_DUPP_UNIQUE,      ER(ER_DUP_UNIQUE));
  SETMSG(HA_ERR_UNKNOWN_CHARSET,        "Can't open charset");
  SETMSG(HA_ERR_WRONG_MRG_TABLE_DEF,    ER(ER_WRONG_MRG_TABLE));
  SETMSG(HA_ERR_CRASHED_ON_REPAIR,      ER(ER_CRASHED_ON_REPAIR));
  SETMSG(HA_ERR_CRASHED_ON_USAGE,       ER(ER_CRASHED_ON_USAGE));
  SETMSG(HA_ERR_LOCK_WAIT_TIMEOUT,      ER(ER_LOCK_WAIT_TIMEOUT));
  SETMSG(HA_ERR_LOCK_TABLE_FULL,        ER(ER_LOCK_TABLE_FULL));
  SETMSG(HA_ERR_READ_ONLY_TRANSACTION,  ER(ER_READ_ONLY_TRANSACTION));
  SETMSG(HA_ERR_LOCK_DEADLOCK,          ER(ER_LOCK_DEADLOCK));
  SETMSG(HA_ERR_CANNOT_ADD_FOREIGN,     ER(ER_CANNOT_ADD_FOREIGN));
  SETMSG(HA_ERR_NO_REFERENCED_ROW,      ER(ER_NO_REFERENCED_ROW_2));
  SETMSG(HA_ERR_ROW_IS_REFERENCED,      ER(ER_ROW_IS_REFERENCED_2));
  SETMSG(HA_ERR_NO_SAVEPOINT,           "No savepoint with that name");
  SETMSG(HA_ERR_NON_UNIQUE_BLOCK_SIZE,  "Non unique key block size");
  SETMSG(HA_ERR_NO_SUCH_TABLE,          "No such table: '%.64s'");
  SETMSG(HA_ERR_TABLE_EXIST,            ER(ER_TABLE_EXISTS_ERROR));
  SETMSG(HA_ERR_NO_CONNECTION,          "Could not connect to storage engine");
  SETMSG(HA_ERR_TABLE_DEF_CHANGED,      ER(ER_TABLE_DEF_CHANGED));
  SETMSG(HA_ERR_FOREIGN_DUPLICATE_KEY,  "FK constraint would lead to duplicate key");
  SETMSG(HA_ERR_TABLE_NEEDS_UPGRADE,    ER(ER_TABLE_NEEDS_UPGRADE));
  SETMSG(HA_ERR_TABLE_READONLY,         ER(ER_OPEN_AS_READONLY));

  /* Register the error messages for use with my_error(). */
  return my_error_register(errmsgs, HA_ERR_FIRST, HA_ERR_LAST);
}


/*
  Unregister handler error messages.

  SYNOPSIS
    ha_finish_errors()

  RETURN
    0           OK
    != 0        Error
*/

static int ha_finish_errors(void)
{
  const char    **errmsgs;

  /* Allocate a pointer array for the error message strings. */
  if (! (errmsgs= my_error_unregister(HA_ERR_FIRST, HA_ERR_LAST)))
    return 1;
  my_free((gptr) errmsgs, MYF(0));
  return 0;
}


int ha_finalize_handlerton(st_plugin_int *plugin)
{
  handlerton *hton= (handlerton *)plugin->data;
  DBUG_ENTER("ha_finalize_handlerton");

  switch (hton->state)
  {
  case SHOW_OPTION_NO:
  case SHOW_OPTION_DISABLED:
    break;
  case SHOW_OPTION_YES:
    if (installed_htons[hton->db_type] == hton)
      installed_htons[hton->db_type]= NULL;
    if (hton->panic && hton->panic(HA_PANIC_CLOSE))
      DBUG_RETURN(1);
    break;
  };
  DBUG_RETURN(0);
}


int ha_initialize_handlerton(st_plugin_int *plugin)
{
  handlerton *hton= ((st_mysql_storage_engine *)plugin->plugin->info)->handlerton;
  DBUG_ENTER("ha_initialize_handlerton");

  plugin->data= hton; // shortcut for the future

  /*
    the switch below and hton->state should be removed when
    command-line options for plugins will be implemented
  */
  switch (hton->state) {
  case SHOW_OPTION_NO:
    break;
  case SHOW_OPTION_YES:
    {
      uint tmp;
      /* now check the db_type for conflict */
      if (hton->db_type <= DB_TYPE_UNKNOWN ||
          hton->db_type >= DB_TYPE_DEFAULT ||
          installed_htons[hton->db_type])
      {
        int idx= (int) DB_TYPE_FIRST_DYNAMIC;

        while (idx < (int) DB_TYPE_DEFAULT && installed_htons[idx])
          idx++;

        if (idx == (int) DB_TYPE_DEFAULT)
        {
          sql_print_warning("Too many storage engines!");
          DBUG_RETURN(1);
        }
        if (hton->db_type != DB_TYPE_UNKNOWN)
          sql_print_warning("Storage engine '%s' has conflicting typecode. "
                            "Assigning value %d.", plugin->plugin->name, idx);
        hton->db_type= (enum legacy_db_type) idx;
      }
      installed_htons[hton->db_type]= hton;
      tmp= hton->savepoint_offset;
      hton->savepoint_offset= savepoint_alloc_size;
      savepoint_alloc_size+= tmp;
      hton->slot= total_ha++;
      hton2plugin[hton->slot]=plugin;
      if (hton->prepare)
        total_ha_2pc++;
      break;
    }
    /* fall through */
  default:
    hton->state= SHOW_OPTION_DISABLED;
    break;
  }
  DBUG_RETURN(0);
}

int ha_init()
{
  int error= 0;
  DBUG_ENTER("ha_init");

  if (ha_init_errors())
    DBUG_RETURN(1);

  DBUG_ASSERT(total_ha < MAX_HA);
  /*
    Check if there is a transaction-capable storage engine besides the
    binary log (which is considered a transaction-capable storage engine in
    counting total_ha)
  */
  opt_using_transactions= total_ha>(ulong)opt_bin_log;
  savepoint_alloc_size+= sizeof(SAVEPOINT);
  DBUG_RETURN(error);
}

/*
  close, flush or restart databases
  Ignore this for other databases than ours
*/

static my_bool panic_handlerton(THD *unused1, st_plugin_int *plugin, void *arg)
{
  handlerton *hton= (handlerton *)plugin->data;
  if (hton->state == SHOW_OPTION_YES && hton->panic)
    ((int*)arg)[0]|= hton->panic((enum ha_panic_function)((int*)arg)[1]);
  return FALSE;
}


int ha_panic(enum ha_panic_function flag)
{
  int error[2];

  error[0]= 0; error[1]= (int)flag;
  plugin_foreach(NULL, panic_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, error);

  if (flag == HA_PANIC_CLOSE && ha_finish_errors())
    error[0]= 1;
  return error[0];
} /* ha_panic */

static my_bool dropdb_handlerton(THD *unused1, st_plugin_int *plugin,
                                 void *path)
{
  handlerton *hton= (handlerton *)plugin->data;
  if (hton->state == SHOW_OPTION_YES && hton->drop_database)
    hton->drop_database((char *)path);
  return FALSE;
}


void ha_drop_database(char* path)
{
  plugin_foreach(NULL, dropdb_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, path);
}


static my_bool closecon_handlerton(THD *thd, st_plugin_int *plugin,
                                   void *unused)
{
  handlerton *hton= (handlerton *)plugin->data;
  /*
    there's no need to rollback here as all transactions must
    be rolled back already
  */
  if (hton->state == SHOW_OPTION_YES && hton->close_connection &&
      thd->ha_data[hton->slot])
    hton->close_connection(thd);
  return FALSE;
}


/* don't bother to rollback here, it's done already */
void ha_close_connection(THD* thd)
{
  plugin_foreach(thd, closecon_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, 0);
}

/* ========================================================================
 ======================= TRANSACTIONS ===================================*/

/*
  Register a storage engine for a transaction

  DESCRIPTION
    Every storage engine MUST call this function when it starts
    a transaction or a statement (that is it must be called both for the
    "beginning of transaction" and "beginning of statement").
    Only storage engines registered for the transaction/statement
    will know when to commit/rollback it.

  NOTE
    trans_register_ha is idempotent - storage engine may register many
    times per transaction.

*/
void trans_register_ha(THD *thd, bool all, handlerton *ht_arg)
{
  THD_TRANS *trans;
  handlerton **ht;
  DBUG_ENTER("trans_register_ha");
  DBUG_PRINT("enter",("%s", all ? "all" : "stmt"));

  if (all)
  {
    trans= &thd->transaction.all;
    thd->server_status|= SERVER_STATUS_IN_TRANS;
  }
  else
    trans= &thd->transaction.stmt;

  for (ht=trans->ht; *ht; ht++)
    if (*ht == ht_arg)
      DBUG_VOID_RETURN;  /* already registered, return */

  trans->ht[trans->nht++]=ht_arg;
  DBUG_ASSERT(*ht == ht_arg);
  trans->no_2pc|=(ht_arg->prepare==0);
  if (thd->transaction.xid_state.xid.is_null())
    thd->transaction.xid_state.xid.set(thd->query_id);
  DBUG_VOID_RETURN;
}

/*
  RETURN
      0  - ok
      1  - error, transaction was rolled back
*/
int ha_prepare(THD *thd)
{
  int error=0, all=1;
  THD_TRANS *trans=all ? &thd->transaction.all : &thd->transaction.stmt;
  handlerton **ht=trans->ht;
  DBUG_ENTER("ha_prepare");
#ifdef USING_TRANSACTIONS
  if (trans->nht)
  {
    for (; *ht; ht++)
    {
      int err;
      statistic_increment(thd->status_var.ha_prepare_count,&LOCK_status);
      if ((*ht)->prepare)
      {
        if ((err= (*(*ht)->prepare)(thd, all)))
        {
          my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
          ha_rollback_trans(thd, all);
          error=1;
          break;
        }
      }
      else
      {
        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                            ER_ILLEGAL_HA, ER(ER_ILLEGAL_HA),
                            hton2plugin[(*ht)->slot]->name.str);
      }
    }
  }
#endif /* USING_TRANSACTIONS */
  DBUG_RETURN(error);
}

/*
  RETURN
      0  - ok
      1  - transaction was rolled back
      2  - error during commit, data may be inconsistent
*/
int ha_commit_trans(THD *thd, bool all)
{
  int error= 0, cookie= 0;
  THD_TRANS *trans= all ? &thd->transaction.all : &thd->transaction.stmt;
  bool is_real_trans= all || thd->transaction.all.nht == 0;
  handlerton **ht= trans->ht;
  my_xid xid= thd->transaction.xid_state.xid.get_my_xid();
  DBUG_ENTER("ha_commit_trans");

  if (thd->in_sub_stmt)
  {
    /*
      Since we don't support nested statement transactions in 5.0,
      we can't commit or rollback stmt transactions while we are inside
      stored functions or triggers. So we simply do nothing now.
      TODO: This should be fixed in later ( >= 5.1) releases.
    */
    if (!all)
      DBUG_RETURN(0);
    /*
      We assume that all statements which commit or rollback main transaction
      are prohibited inside of stored functions or triggers. So they should
      bail out with error even before ha_commit_trans() call. To be 100% safe
      let us throw error in non-debug builds.
    */
    DBUG_ASSERT(0);
    my_error(ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG, MYF(0));
    DBUG_RETURN(2);
  }
#ifdef USING_TRANSACTIONS
  if (trans->nht)
  {
    if (is_real_trans && wait_if_global_read_lock(thd, 0, 0))
    {
      ha_rollback_trans(thd, all);
      DBUG_RETURN(1);
    }
    DBUG_EXECUTE_IF("crash_commit_before", abort(););

    /* Close all cursors that can not survive COMMIT */
    if (is_real_trans)                          /* not a statement commit */
      thd->stmt_map.close_transient_cursors();

    if (!trans->no_2pc && trans->nht > 1)
    {
      for (; *ht && !error; ht++)
      {
        int err;
        if ((err= (*(*ht)->prepare)(thd, all)))
        {
          my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
          error= 1;
        }
        statistic_increment(thd->status_var.ha_prepare_count,&LOCK_status);
      }
      DBUG_EXECUTE_IF("crash_commit_after_prepare", abort(););
      if (error || (is_real_trans && xid &&
                    (error= !(cookie= tc_log->log(thd, xid)))))
      {
        ha_rollback_trans(thd, all);
        error= 1;
        goto end;
      }
      DBUG_EXECUTE_IF("crash_commit_after_log", abort(););
    }
    error=ha_commit_one_phase(thd, all) ? cookie ? 2 : 1 : 0;
    DBUG_EXECUTE_IF("crash_commit_before_unlog", abort(););
    if (cookie)
      tc_log->unlog(cookie, xid);
    DBUG_EXECUTE_IF("crash_commit_after", abort(););
end:
    if (is_real_trans)
      start_waiting_global_read_lock(thd);
  }
#endif /* USING_TRANSACTIONS */
  DBUG_RETURN(error);
}

/*
  NOTE - this function does not care about global read lock.
  A caller should.
*/
int ha_commit_one_phase(THD *thd, bool all)
{
  int error=0;
  THD_TRANS *trans=all ? &thd->transaction.all : &thd->transaction.stmt;
  bool is_real_trans=all || thd->transaction.all.nht == 0;
  handlerton **ht=trans->ht;
  DBUG_ENTER("ha_commit_one_phase");
#ifdef USING_TRANSACTIONS
  if (trans->nht)
  {
    for (ht=trans->ht; *ht; ht++)
    {
      int err;
      if ((err= (*(*ht)->commit)(thd, all)))
      {
        my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
        error=1;
      }
      statistic_increment(thd->status_var.ha_commit_count,&LOCK_status);
      *ht= 0;
    }
    trans->nht=0;
    trans->no_2pc=0;
    if (is_real_trans)
      thd->transaction.xid_state.xid.null();
    if (all)
    {
#ifdef HAVE_QUERY_CACHE
      if (thd->transaction.changed_tables)
        query_cache.invalidate(thd->transaction.changed_tables);
#endif
      thd->variables.tx_isolation=thd->session_tx_isolation;
      thd->transaction.cleanup();
    }
  }
#endif /* USING_TRANSACTIONS */
  DBUG_RETURN(error);
}


int ha_rollback_trans(THD *thd, bool all)
{
  int error=0;
  THD_TRANS *trans=all ? &thd->transaction.all : &thd->transaction.stmt;
  bool is_real_trans=all || thd->transaction.all.nht == 0;
  DBUG_ENTER("ha_rollback_trans");
  if (thd->in_sub_stmt)
  {
    /*
      If we are inside stored function or trigger we should not commit or
      rollback current statement transaction. See comment in ha_commit_trans()
      call for more information.
    */
    if (!all)
      DBUG_RETURN(0);
    DBUG_ASSERT(0);
    my_error(ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG, MYF(0));
    DBUG_RETURN(1);
  }
#ifdef USING_TRANSACTIONS
  if (trans->nht)
  {
    /* Close all cursors that can not survive ROLLBACK */
    if (is_real_trans)                          /* not a statement commit */
      thd->stmt_map.close_transient_cursors();

    for (handlerton **ht=trans->ht; *ht; ht++)
    {
      int err;
      if ((err= (*(*ht)->rollback)(thd, all)))
      { // cannot happen
        my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
        error=1;
      }
      statistic_increment(thd->status_var.ha_rollback_count,&LOCK_status);
      *ht= 0;
    }
    trans->nht=0;
    trans->no_2pc=0;
    if (is_real_trans)
      thd->transaction.xid_state.xid.null();
    if (all)
    {
      thd->variables.tx_isolation=thd->session_tx_isolation;
      thd->transaction.cleanup();
    }
  }
#endif /* USING_TRANSACTIONS */
  /*
    If a non-transactional table was updated, warn; don't warn if this is a
    slave thread (because when a slave thread executes a ROLLBACK, it has
    been read from the binary log, so it's 100% sure and normal to produce
    error ER_WARNING_NOT_COMPLETE_ROLLBACK. If we sent the warning to the
    slave SQL thread, it would not stop the thread but just be printed in
    the error log; but we don't want users to wonder why they have this
    message in the error log, so we don't send it.
  */
  if (is_real_trans && (thd->options & OPTION_STATUS_NO_TRANS_UPDATE) &&
      !thd->slave_thread)
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                 ER_WARNING_NOT_COMPLETE_ROLLBACK,
                 ER(ER_WARNING_NOT_COMPLETE_ROLLBACK));
  DBUG_RETURN(error);
}

/*
  This is used to commit or rollback a single statement depending on the value
  of error. Note that if the autocommit is on, then the following call inside
  InnoDB will commit or rollback the whole transaction (= the statement). The
  autocommit mechanism built into InnoDB is based on counting locks, but if
  the user has used LOCK TABLES then that mechanism does not know to do the
  commit.
*/

int ha_autocommit_or_rollback(THD *thd, int error)
{
  DBUG_ENTER("ha_autocommit_or_rollback");
#ifdef USING_TRANSACTIONS
  if (thd->transaction.stmt.nht)
  {
    if (!error)
    {
      if (ha_commit_stmt(thd))
        error=1;
    }
    else
      (void) ha_rollback_stmt(thd);

    thd->variables.tx_isolation=thd->session_tx_isolation;
  }
#endif
  DBUG_RETURN(error);
}


struct xahton_st {
  XID *xid;
  int result;
};

static my_bool xacommit_handlerton(THD *unused1, st_plugin_int *plugin,
                                   void *arg)
{
  handlerton *hton= (handlerton *)plugin->data;
  if (hton->state == SHOW_OPTION_YES && hton->recover)
  {
    hton->commit_by_xid(((struct xahton_st *)arg)->xid);
    ((struct xahton_st *)arg)->result= 0;
  }
  return FALSE;
}

static my_bool xarollback_handlerton(THD *unused1, st_plugin_int *plugin,
                                     void *arg)
{
  handlerton *hton= (handlerton *)plugin->data;
  if (hton->state == SHOW_OPTION_YES && hton->recover)
  {
    hton->rollback_by_xid(((struct xahton_st *)arg)->xid);
    ((struct xahton_st *)arg)->result= 0;
  }
  return FALSE;
}


int ha_commit_or_rollback_by_xid(XID *xid, bool commit)
{
  struct xahton_st xaop;
  xaop.xid= xid;
  xaop.result= 1;

  plugin_foreach(NULL, commit ? xacommit_handlerton : xarollback_handlerton,
                 MYSQL_STORAGE_ENGINE_PLUGIN, &xaop);

  return xaop.result;
}


#ifndef DBUG_OFF
/* this does not need to be multi-byte safe or anything */
static char* xid_to_str(char *buf, XID *xid)
{
  int i;
  char *s=buf;
  *s++='\'';
  for (i=0; i < xid->gtrid_length+xid->bqual_length; i++)
  {
    uchar c=(uchar)xid->