handler0alter.cc

Go to the documentation of this file.

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

Copyright (C) 2005, 2010, Innobase Oy. All Rights Reserved.

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., 51 Franklin
St, Fifth Floor, Boston, MA 02110-1301 USA

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

/**************************************************/
#include <config.h>
#include <drizzled/error.h>
#include <drizzled/charset.h>
#include <drizzled/field.h>
#include <drizzled/table.h>
#include <drizzled/field/varstring.h>
#include <drizzled/internal/my_sys.h>

#include "log0log.h"
#include "row0merge.h"
#include "srv0srv.h"
#include "trx0trx.h"
#include "trx0roll.h"
#include "ha_prototypes.h"
#include "handler0alter.h"

#include "ha_innodb.h"
#include "handler0vars.h"

/*************************************************************/
static
void
innobase_col_to_mysql(
/*==================*/
  const dict_col_t* col,  
  const unsigned char*  data, 
  ulint     len,  
  Field*      field)  
{
  unsigned char*  ptr;
  unsigned char*  dest  = field->ptr;
  ulint flen  = field->pack_length();

  switch (col->mtype) {
  case DATA_INT:
    ut_ad(len == flen);

    /* Convert integer data from Innobase to little-endian
    format, sign bit restored to normal */

    for (ptr = dest + len; ptr != dest; ) {
      *--ptr = *data++;
    }

    if (!(field->flags & UNSIGNED_FLAG)) {
      ((byte*) dest)[len - 1] ^= 0x80;
    }

    break;

  case DATA_VARCHAR:
  case DATA_VARMYSQL:
  case DATA_BINARY:
    field->reset();

    if (field->type() == DRIZZLE_TYPE_VARCHAR) {
      /* This is a >= 5.0.3 type true VARCHAR. Store the
      length of the data to the first byte or the first
      two bytes of dest. */

      dest = row_mysql_store_true_var_len(
        dest, len, flen - field->key_length());
    }

    /* Copy the actual data */
    memcpy(dest, data, len);
    break;

  case DATA_BLOB:
    /* Store a pointer to the BLOB buffer to dest: the BLOB was
    already copied to the buffer in row_sel_store_mysql_rec */

    row_mysql_store_blob_ref(dest, flen, data, len);
    break;

#ifdef UNIV_DEBUG
  case DATA_MYSQL:
    ut_ad(flen >= len);
    ut_ad(DATA_MBMAXLEN(col->mbminmaxlen)
          >= DATA_MBMINLEN(col->mbminmaxlen));
    ut_ad(DATA_MBMAXLEN(col->mbminmaxlen)
          > DATA_MBMINLEN(col->mbminmaxlen) || flen == len);
    memcpy(dest, data, len);
    break;

  default:
  case DATA_SYS_CHILD:
  case DATA_SYS:
    /* These column types should never be shipped to MySQL. */
    ut_ad(0);

  case DATA_CHAR:
  case DATA_FIXBINARY:
  case DATA_FLOAT:
  case DATA_DOUBLE:
  case DATA_DECIMAL:
    /* Above are the valid column types for MySQL data. */
    ut_ad(flen == len);
#else /* UNIV_DEBUG */
  default:
#endif /* UNIV_DEBUG */
    memcpy(dest, data, len);
  }
}

/*************************************************************/
UNIV_INTERN
void
innobase_rec_to_mysql(
/*==================*/
  Table*      table,    
  const rec_t*    rec,    
  const dict_index_t* index,    
  const ulint*    offsets)  
{
  uint  n_fields  = table->getShare()->sizeFields();
  uint  i;

  ut_ad(n_fields == dict_table_get_n_user_cols(index->table));

  for (i = 0; i < n_fields; i++) {
    Field*    field = table->getField(i);
    ulint   ipos;
    ulint   ilen;
    const unsigned char*  ifield;

    field->reset();

    ipos = dict_index_get_nth_col_pos(index, i);

    if (UNIV_UNLIKELY(ipos == ULINT_UNDEFINED)) {
null_field:
      field->set_null();
      continue;
    }

    ifield = rec_get_nth_field(rec, offsets, ipos, &ilen);

    /* Assign the NULL flag */
    if (ilen == UNIV_SQL_NULL) {
      ut_ad(field->real_maybe_null());
      goto null_field;
    }

    field->set_notnull();

    innobase_col_to_mysql(
      dict_field_get_col(
        dict_index_get_nth_field(index, ipos)),
      ifield, ilen, field);
  }
}

/*************************************************************/
UNIV_INTERN
void
innobase_rec_reset(
/*===============*/
  Table*      table)    
{
  uint  n_fields  = table->getShare()->sizeFields();
  uint  i;

  for (i = 0; i < n_fields; i++) {
    table->getField(i)->set_default();
  }
}

#if 0 // This is a part of the fast index code.
/******************************************************************/
static
void
innobase_convert_tablename(
/*=======================*/
  char* s)  
{

  char* slash = strchr(s, '/');

  if (slash) {
    char* t;
    /* Temporarily replace the '/' with NUL. */
    *slash = 0;
    strncpy(s, s, slash - s + 1);

    t = s + strlen(s);
    ut_ad(slash >= t);
    /* Append a  '.' after the database name. */
    *t++ = '.';
    slash++;
    /* Convert the table name. */
    strncpy(t, slash, slash - t + strlen(slash));
  }
}


/*******************************************************************/
static
int
innobase_check_index_keys(
/*======================*/
  const KeyInfo*  key_info, 
  ulint   num_of_keys,  
  const dict_table_t* table)  
{
  ulint   key_num;

  ut_ad(key_info);
  ut_ad(num_of_keys);

  for (key_num = 0; key_num < num_of_keys; key_num++) {
    const KeyInfo&  key = key_info[key_num];

    /* Check that the same index name does not appear
    twice in indexes to be created. */

    for (ulint i = 0; i < key_num; i++) {
      const KeyInfo&  key2 = key_info[i];

      if (0 == strcmp(key.name, key2.name)) {
        my_error(ER_WRONG_NAME_FOR_INDEX, MYF(0),
           key.name);

        return(ER_WRONG_NAME_FOR_INDEX);
      }
    }

    /* Check that the same index name does not already exist. */

    for (const dict_index_t* index
           = dict_table_get_first_index(table);
         index; index = dict_table_get_next_index(index)) {

      if (0 == strcmp(key.name, index->name)) {
        my_error(ER_WRONG_NAME_FOR_INDEX, MYF(0),
           key.name);

        return(ER_WRONG_NAME_FOR_INDEX);
      }
    }

    /* Check that MySQL does not try to create a column
    prefix index field on an inappropriate data type and
    that the same column does not appear twice in the index. */

    for (ulint i = 0; i < key.key_parts; i++) {
      const KeyPartInfo&  key_part1
        = key.key_part[i];
      const Field*    field
        = key_part1.field;
      ibool     is_unsigned;

      switch (get_innobase_type_from_mysql_type(
          &is_unsigned, field)) {
      default:
        break;
      case DATA_INT:
      case DATA_FLOAT:
      case DATA_DOUBLE:
      case DATA_DECIMAL:
        if (field->type() == DRIZZLE_TYPE_VARCHAR) {
          if (key_part1.length
              >= field->pack_length()
              - ((Field_varstring*) field)
              ->length_bytes) {
            break;
          }
        } else {
          if (key_part1.length
              >= field->pack_length()) {
            break;
          }
        }

        my_error(ER_WRONG_KEY_COLUMN, MYF(0),
           field->field_name);
        return(ER_WRONG_KEY_COLUMN);
      }

      for (ulint j = 0; j < i; j++) {
        const KeyPartInfo&  key_part2
          = key.key_part[j];

        if (strcmp(key_part1.field->field_name,
             key_part2.field->field_name)) {
          continue;
        }

        my_error(ER_WRONG_KEY_COLUMN, MYF(0),
           key_part1.field->field_name);
        return(ER_WRONG_KEY_COLUMN);
      }
    }
  }

  return(0);
}

/*******************************************************************/
static
void
innobase_create_index_field_def(
/*============================*/
  KeyPartInfo*    key_part, 
  mem_heap_t*   heap,   
  merge_index_field_t*  index_field)  
{
  Field*    field;
  ibool   is_unsigned;
  ulint   col_type;

  ut_ad(key_part);
  ut_ad(index_field);

  field = key_part->field;
  ut_a(field);

  col_type = get_innobase_type_from_mysql_type(&is_unsigned, field);

  if (DATA_BLOB == col_type
      || (key_part->length < field->pack_length()
    && field->type() != DRIZZLE_TYPE_VARCHAR)
      || (field->type() == DRIZZLE_TYPE_VARCHAR
    && key_part->length < field->pack_length()
      - ((Field_varstring*)field)->length_bytes)) {

    index_field->prefix_len = key_part->length;
  } else {
    index_field->prefix_len = 0;
  }

  index_field->field_name = mem_heap_strdup(heap, field->field_name);

  return;
}

/*******************************************************************/
static
void
innobase_create_index_def(
/*======================*/
  KeyInfo*      key,    
  bool      new_primary,  
  bool      key_primary,  
  merge_index_def_t*  index,    
  mem_heap_t*   heap)   
{
  ulint i;
  ulint len;
  ulint n_fields = key->key_parts;
  char* index_name;

  index->fields = (merge_index_field_t*) mem_heap_alloc(
    heap, n_fields * sizeof *index->fields);

  index->ind_type = 0;
  index->n_fields = n_fields;
  len = strlen(key->name) + 1;
  index->name = index_name = (char*) mem_heap_alloc(heap,
                len + !new_primary);

  if (UNIV_LIKELY(!new_primary)) {
    *index_name++ = TEMP_INDEX_PREFIX;
  }

  memcpy(index_name, key->name, len);

  if (key->flags & HA_NOSAME) {
    index->ind_type |= DICT_UNIQUE;
  }

  if (key_primary) {
    index->ind_type |= DICT_CLUSTERED;
  }

  for (i = 0; i < n_fields; i++) {
    innobase_create_index_field_def(&key->key_part[i], heap,
            &index->fields[i]);
  }

  return;
}

/*******************************************************************/
static
void
innobase_copy_index_field_def(
/*==========================*/
  const dict_field_t* field,    
  merge_index_field_t*  index_field)  
{
  assert(field != NULL);
  assert(index_field != NULL);

  index_field->field_name = field->name;
  index_field->prefix_len = field->prefix_len;

  return;
}

/*******************************************************************/
static
void
innobase_copy_index_def(
/*====================*/
  const dict_index_t* index,  
  merge_index_def_t*  new_index,
  mem_heap_t*   heap) 
{
  ulint n_fields;
  ulint i;

  /* Note that we take only those fields that user defined to be
  in the index.  In the internal representation more colums were
  added and those colums are not copied .*/

  n_fields = index->n_user_defined_cols;

  new_index->fields = (merge_index_field_t*) mem_heap_alloc(
    heap, n_fields * sizeof *new_index->fields);

  /* When adding a PRIMARY KEY, we may convert a previous
  clustered index to a secondary index (UNIQUE NOT NULL). */
  new_index->ind_type = index->type & ~DICT_CLUSTERED;
  new_index->n_fields = n_fields;
  new_index->name = index->name;

  for (i = 0; i < n_fields; i++) {
    innobase_copy_index_field_def(&index->fields[i],
                &new_index->fields[i]);
  }

  return;
}

/*******************************************************************/
static
merge_index_def_t*
innobase_create_key_def(
/*====================*/
  trx_t*    trx,    
  const dict_table_t*table,   
  mem_heap_t* heap,   
  KeyInfo*    key_info, 
  ulint&    n_keys)   
{
  ulint     i = 0;
  merge_index_def_t*  indexdef;
  merge_index_def_t*  indexdefs;
  bool      new_primary;

  indexdef = indexdefs = (merge_index_def_t*)
    mem_heap_alloc(heap, sizeof *indexdef
             * (n_keys + UT_LIST_GET_LEN(table->indexes)));

  /* If there is a primary key, it is always the first index
  defined for the table. */

  new_primary = !system_charset_info->strcasecmp(key_info->name, "PRIMARY");

  /* If there is a UNIQUE INDEX consisting entirely of NOT NULL
  columns and if the index does not contain column prefix(es)
  (only prefix/part of the column is indexed), MySQL will treat the
  index as a PRIMARY KEY unless the table already has one. */

  if (!new_primary && (key_info->flags & HA_NOSAME)
      && (!(key_info->flags & HA_KEY_HAS_PART_KEY_SEG))
      && row_table_got_default_clust_index(table)) {
    uint    key_part = key_info->key_parts;

    new_primary = TRUE;

    while (key_part--) {
      if (key_info->key_part[key_part].null_bit == 0) {
        new_primary = FALSE;
        break;
      }
    }
  }

  if (new_primary) {
    const dict_index_t* index;

    /* Create the PRIMARY key index definition */
    innobase_create_index_def(&key_info[i++], TRUE, TRUE,
            indexdef++, heap);

    row_mysql_lock_data_dictionary(trx);

    index = dict_table_get_first_index(table);

    /* Copy the index definitions of the old table.  Skip
    the old clustered index if it is a generated clustered
    index or a PRIMARY KEY.  If the clustered index is a
    UNIQUE INDEX, it must be converted to a secondary index. */

    if (dict_index_get_nth_col(index, 0)->mtype == DATA_SYS
        || !system_charset_info->strcasecmp(index->name, "PRIMARY")) 
    {
      index = dict_table_get_next_index(index);
    }

    while (index) {
      innobase_copy_index_def(index, indexdef++, heap);
      index = dict_table_get_next_index(index);
    }

    row_mysql_unlock_data_dictionary(trx);
  }

  /* Create definitions for added secondary indexes. */

  while (i < n_keys) {
    innobase_create_index_def(&key_info[i++], new_primary, FALSE,
            indexdef++, heap);
  }

  n_keys = indexdef - indexdefs;

  return(indexdefs);
}

/*******************************************************************/
static
char*
innobase_create_temporary_tablename(
/*================================*/
  mem_heap_t* heap,   
  char    id,   
  const char*     table_name) 
{
  char*     name;
  ulint     len;
  static const char suffix[] = "@0023 "; /* "# " */

  len = strlen(table_name);

  name = (char*) mem_heap_alloc(heap, len + sizeof suffix);
  memcpy(name, table_name, len);
  memcpy(name + len, suffix, sizeof suffix);
  name[len + (sizeof suffix - 2)] = id;

  return(name);
}


/*******************************************************************/
UNIV_INTERN
int
ha_innobase::add_index(
/*===================*/
                       Session *session,
  Table*  i_table,  
  KeyInfo*  key_info, 
  uint  num_of_keys)  
{
  dict_index_t**  index;    
  dict_table_t* innodb_table; 
  dict_table_t* indexed_table;  
  merge_index_def_t* index_defs;  
  mem_heap_t*     heap;   
  trx_t*    trx;    
  ulint   num_of_idx;
  ulint   num_created = 0;
  ibool   dict_locked = FALSE;
  ulint   new_primary;
  int   error;

  ut_a(i_table);
  ut_a(key_info);
  ut_a(num_of_keys);

  if (srv_created_new_raw || srv_force_recovery) {
    return(HA_ERR_WRONG_COMMAND);
  }

  update_session(session);

  heap = mem_heap_create(1024);

  /* In case MySQL calls this in the middle of a SELECT query, release
  possible adaptive hash latch to avoid deadlocks of threads. */
  trx_search_latch_release_if_reserved(prebuilt->trx);
  trx_start_if_not_started(prebuilt->trx);

  /* Create a background transaction for the operations on
  the data dictionary tables. */
  trx = innobase_trx_allocate(user_session);
  trx_start_if_not_started(trx);

  innodb_table = indexed_table
    = dict_table_get(prebuilt->table->name, FALSE);

  if (UNIV_UNLIKELY(!innodb_table)) {
    error = HA_ERR_NO_SUCH_TABLE;
    goto err_exit;
  }

  /* Check if the index name is reserved. */
  if (innobase_index_name_is_reserved(trx, key_info, num_of_keys)) {
    error = -1;
  } else {
    /* Check that index keys are sensible */
    error = innobase_check_index_keys(key_info, num_of_keys,
              innodb_table);
  }

  if (UNIV_UNLIKELY(error)) {
err_exit:
    mem_heap_free(heap);
    trx_general_rollback_for_mysql(trx, NULL);
    trx_free_for_mysql(trx);
    trx_commit_for_mysql(prebuilt->trx);
    return(error);
  }

  /* Create table containing all indexes to be built in this
  alter table add index so that they are in the correct order
  in the table. */

  num_of_idx = num_of_keys;

  index_defs = innobase_create_key_def(
    trx, innodb_table, heap, key_info, num_of_idx);

  new_primary = DICT_CLUSTERED & index_defs[0].ind_type;

  /* Allocate memory for dictionary index definitions */

  index = (dict_index_t**) mem_heap_alloc(
    heap, num_of_idx * sizeof *index);

  /* Flag this transaction as a dictionary operation, so that
  the data dictionary will be locked in crash recovery. */
  trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);

  /* Acquire a lock on the table before creating any indexes. */
  error = row_merge_lock_table(prebuilt->trx, innodb_table,
             new_primary ? LOCK_X : LOCK_S);

  if (UNIV_UNLIKELY(error != DB_SUCCESS)) {

    goto error_handling;
  }

  /* Latch the InnoDB data dictionary exclusively so that no deadlocks
  or lock waits can happen in it during an index create operation. */

  row_mysql_lock_data_dictionary(trx);
  dict_locked = TRUE;

  ut_d(dict_table_check_for_dup_indexes(innodb_table, FALSE));

  /* If a new primary key is defined for the table we need
  to drop the original table and rebuild all indexes. */

  if (UNIV_UNLIKELY(new_primary)) {
    /* This transaction should be the only one
    operating on the table. */
    ut_a(innodb_table->n_mysql_handles_opened == 1);

    char* new_table_name = innobase_create_temporary_tablename(
      heap, '1', innodb_table->name);

    /* Clone the table. */
    trx_set_dict_operation(trx, TRX_DICT_OP_TABLE);
    indexed_table = row_merge_create_temporary_table(
      new_table_name, index_defs, innodb_table, trx);

    if (!indexed_table) {

      switch (trx->error_state) {
      case DB_TABLESPACE_ALREADY_EXISTS:
      case DB_DUPLICATE_KEY:
        innobase_convert_tablename(new_table_name);
        my_error(HA_ERR_TABLE_EXIST, MYF(0),
           new_table_name);
        error = HA_ERR_TABLE_EXIST;
        break;
      default:
        error = convert_error_code_to_mysql(
          trx->error_state, innodb_table->flags,
          user_session);
      }

      ut_d(dict_table_check_for_dup_indexes(innodb_table,
                    FALSE));
      row_mysql_unlock_data_dictionary(trx);
      goto err_exit;
    }

    trx->table_id = indexed_table->id;
  }

  /* Create the indexes in SYS_INDEXES and load into dictionary. */

  for (ulint i = 0; i < num_of_idx; i++) {

    index[i] = row_merge_create_index(trx, indexed_table,
              &index_defs[i]);

    if (!index[i]) {
      error = trx->error_state;
      goto error_handling;
    }

    num_created++;
  }

  ut_ad(error == DB_SUCCESS);

  /* We will need to rebuild index translation table. Set
  valid index entry count in the translation table to zero */
  share->idx_trans_tbl.index_count = 0;

  /* Commit the data dictionary transaction in order to release
  the table locks on the system tables.  This means that if
  MySQL crashes while creating a new primary key inside
  row_merge_build_indexes(), indexed_table will not be dropped
  by trx_rollback_active().  It will have to be recovered or
  dropped by the database administrator. */
  trx_commit_for_mysql(trx);

  row_mysql_unlock_data_dictionary(trx);
  dict_locked = FALSE;

  ut_a(trx->n_active_thrs == 0);
  ut_a(UT_LIST_GET_LEN(trx->signals) == 0);

  if (UNIV_UNLIKELY(new_primary)) {
    /* A primary key is to be built.  Acquire an exclusive
    table lock also on the table that is being created. */
    ut_ad(indexed_table != innodb_table);

    error = row_merge_lock_table(prebuilt->trx, indexed_table,
               LOCK_X);

    if (UNIV_UNLIKELY(error != DB_SUCCESS)) {

      goto error_handling;
    }
  }

  /* Read the clustered index of the table and build indexes
  based on this information using temporary files and merge sort. */
  error = row_merge_build_indexes(prebuilt->trx,
          innodb_table, indexed_table,
          index, num_of_idx, i_table);

error_handling:

  /* After an error, remove all those index definitions from the
  dictionary which were defined. */

  switch (error) {
    const char* old_name;
    char*   tmp_name;
  case DB_SUCCESS:
    ut_a(!dict_locked);
    row_mysql_lock_data_dictionary(trx);
    dict_locked = TRUE;

    ut_d(dict_table_check_for_dup_indexes(prebuilt->table, TRUE));

    if (!new_primary) {
      error = row_merge_rename_indexes(trx, indexed_table);

      if (error != DB_SUCCESS) {
        row_merge_drop_indexes(trx, indexed_table,
                   index, num_created);
      }

      goto convert_error;
    }

    /* If a new primary key was defined for the table and
    there was no error at this point, we can now rename
    the old table as a temporary table, rename the new
    temporary table as the old table and drop the old table. */
    old_name = innodb_table->name;
    tmp_name = innobase_create_temporary_tablename(heap, '2',
                     old_name);

    error = row_merge_rename_tables(innodb_table, indexed_table,
            tmp_name, trx);

    if (error != DB_SUCCESS) {

      row_merge_drop_table(trx, indexed_table);

      switch (error) {
      case DB_TABLESPACE_ALREADY_EXISTS:
      case DB_DUPLICATE_KEY:
        innobase_convert_tablename(tmp_name);
        my_error(HA_ERR_TABLE_EXIST, MYF(0), tmp_name);
        error = HA_ERR_TABLE_EXIST;
        break;
      default:
        goto convert_error;
      }
      break;
    }

    trx_commit_for_mysql(prebuilt->trx);
    row_prebuilt_free(prebuilt, TRUE);
    prebuilt = row_create_prebuilt(indexed_table);

    indexed_table->n_mysql_handles_opened++;

    error = row_merge_drop_table(trx, innodb_table);
    innodb_table = indexed_table;
    goto convert_error;

  case DB_TOO_BIG_RECORD:
    my_error(HA_ERR_TO_BIG_ROW, MYF(0));
    goto error;
  case DB_PRIMARY_KEY_IS_NULL:
    my_error(ER_PRIMARY_CANT_HAVE_NULL, MYF(0));
    /* fall through */
  case DB_DUPLICATE_KEY:
error:
    prebuilt->trx->error_info = NULL;
    /* fall through */
  default:
    trx->error_state = DB_SUCCESS;

    if (new_primary) {
      if (indexed_table != innodb_table) {
        row_merge_drop_table(trx, indexed_table);
      }
    } else {
      if (!dict_locked) {
        row_mysql_lock_data_dictionary(trx);
        dict_locked = TRUE;
      }

      row_merge_drop_indexes(trx, indexed_table,
                 index, num_created);
    }

convert_error:
    error = convert_error_code_to_mysql(error,
                innodb_table->flags,
                user_session);
  }

  mem_heap_free(heap);
  trx_commit_for_mysql(trx);
  if (prebuilt->trx) {
    trx_commit_for_mysql(prebuilt->trx);
  }

  if (dict_locked) {
    ut_d(dict_table_check_for_dup_indexes(innodb_table, FALSE));
    row_mysql_unlock_data_dictionary(trx);
  }

  trx_free_for_mysql(trx);

  /* There might be work for utility threads.*/
  srv_active_wake_master_thread();

  return(error);
}

/*******************************************************************/
UNIV_INTERN
int
ha_innobase::prepare_drop_index(
/*============================*/
                                Session *session,
  Table*  i_table,  
  uint* key_num,  
  uint  num_of_keys)  
{
  trx_t*    trx;
  int   err = 0;
  uint    n_key;

  ut_ad(i_table);
  ut_ad(key_num);
  ut_ad(num_of_keys);
  if (srv_created_new_raw || srv_force_recovery) {
    return(HA_ERR_WRONG_COMMAND);
  }

  update_session(session);

  trx_search_latch_release_if_reserved(prebuilt->trx);
  trx = prebuilt->trx;

  /* Test and mark all the indexes to be dropped */

  row_mysql_lock_data_dictionary(trx);
  ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));

  /* Check that none of the indexes have previously been flagged
  for deletion. */
  {
    const dict_index_t* index
      = dict_table_get_first_index(prebuilt->table);
    do {
      ut_a(!index->to_be_dropped);
      index = dict_table_get_next_index(index);
    } while (index);
  }

  for (n_key = 0; n_key < num_of_keys; n_key++) {
    const KeyInfo*  key;
    dict_index_t* index;

    key = i_table->key_info + key_num[n_key];
    index = dict_table_get_index_on_name_and_min_id(
      prebuilt->table, key->name);

    if (!index) {
      errmsg_printf(ERRMSG_LVL_ERROR, "InnoDB could not find key n:o %u "
          "with name %s for table %s",
          key_num[n_key],
          key ? key->name : "NULL",
          prebuilt->table->name);

      err = HA_ERR_KEY_NOT_FOUND;
      goto func_exit;
    }

    /* Refuse to drop the clustered index.  It would be
    better to automatically generate a clustered index,
    but drizzled::alter_table() will call this method only
    after ha_innobase::add_index(). */

    if (dict_index_is_clust(index)) {
      my_error(ER_REQUIRES_PRIMARY_KEY, MYF(0));
      err = -1;
      goto func_exit;
    }

    index->to_be_dropped = TRUE;
  }

  /* If FOREIGN_KEY_CHECKS = 1 you may not drop an index defined
  for a foreign key constraint because InnoDB requires that both
  tables contain indexes for the constraint. Such index can
  be dropped only if FOREIGN_KEY_CHECKS is set to 0.
  Note that CREATE INDEX id ON table does a CREATE INDEX and
  DROP INDEX, and we can ignore here foreign keys because a
  new index for the foreign key has already been created.

  We check for the foreign key constraints after marking the
  candidate indexes for deletion, because when we check for an
  equivalent foreign index we don't want to select an index that
  is later deleted. */

  if (trx->check_foreigns
      && session_sql_command(user_session) != SQLCOM_CREATE_INDEX) {
    dict_index_t* index;

    for (index = dict_table_get_first_index(prebuilt->table);
         index;
         index = dict_table_get_next_index(index)) {
      dict_foreign_t* foreign;

      if (!index->to_be_dropped) {

        continue;
      }

      /* Check if the index is referenced. */
      foreign = dict_table_get_referenced_constraint(
        prebuilt->table, index);

      if (foreign) {
index_needed:
        trx_set_detailed_error(
          trx,
          "Index needed in foreign key "
          "constraint");

        trx->error_info = index;

        err = HA_ERR_DROP_INDEX_FK;
        break;
      } else {
        /* Check if this index references some
        other table */
        foreign = dict_table_get_foreign_constraint(
          prebuilt->table, index);

        if (foreign) {
          ut_a(foreign->foreign_index == index);

          /* Search for an equivalent index that
          the foreign key constraint could use
          if this index were to be deleted. */
          if (!dict_foreign_find_equiv_index(
            foreign)) {

            goto index_needed;
          }
        }
      }
    }
  } else if (session_sql_command(user_session) == SQLCOM_CREATE_INDEX) {
    /* This is a drop of a foreign key constraint index that
    was created by MySQL when the constraint was added.  MySQL
    does this when the user creates an index explicitly which
    can be used in place of the automatically generated index. */

    dict_index_t* index;

    for (index = dict_table_get_first_index(prebuilt->table);
         index;
         index = dict_table_get_next_index(index)) {
      dict_foreign_t* foreign;

      if (!index->to_be_dropped) {

        continue;
      }

      /* Check if this index references some other table */
      foreign = dict_table_get_foreign_constraint(
        prebuilt->table, index);

      if (foreign == NULL) {

        continue;
      }

      ut_a(foreign->foreign_index == index);

      /* Search for an equivalent index that the
      foreign key constraint could use if this index
      were to be deleted. */

      if (!dict_foreign_find_equiv_index(foreign)) {
        trx_set_detailed_error(
          trx,
          "Index needed in foreign key "
          "constraint");

        trx->error_info = foreign->foreign_index;

        err = HA_ERR_DROP_INDEX_FK;
        break;
      }
    }
  }

func_exit:
  if (err) {
    /* Undo our changes since there was some sort of error. */
    dict_index_t* index
      = dict_table_get_first_index(prebuilt->table);

    do {
      index->to_be_dropped = FALSE;
      index = dict_table_get_next_index(index);
    } while (index);
  }

  ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
  row_mysql_unlock_data_dictionary(trx);

  return(err);
}

/*******************************************************************/
UNIV_INTERN
int
ha_innobase::final_drop_index(
/*==========================*/
                              Session *session,
  Table*  )   
{
  dict_index_t* index;    
  trx_t*    trx;    
  int   err;

  if (srv_created_new_raw || srv_force_recovery) {
    return(HA_ERR_WRONG_COMMAND);
  }

  update_session(session);

  trx_search_latch_release_if_reserved(prebuilt->trx);
  trx_start_if_not_started(prebuilt->trx);

  /* Create a background transaction for the operations on
  the data dictionary tables. */
  trx = innobase_trx_allocate(user_session);
  trx_start_if_not_started(trx);

  /* Flag this transaction as a dictionary operation, so that
  the data dictionary will be locked in crash recovery. */
  trx_set_dict_operation(trx, TRX_DICT_OP_INDEX);

  /* Lock the table exclusively, to ensure that no active
  transaction depends on an index that is being dropped. */
  err = convert_error_code_to_mysql(
    row_merge_lock_table(prebuilt->trx, prebuilt->table, LOCK_X),
    prebuilt->table->flags, user_session);

  row_mysql_lock_data_dictionary(trx);
  ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));

  if (UNIV_UNLIKELY(err)) {

    /* Unmark the indexes to be dropped. */
    for (index = dict_table_get_first_index(prebuilt->table);
         index; index = dict_table_get_next_index(index)) {

      index->to_be_dropped = FALSE;
    }

    goto func_exit;
  }

  /* Drop indexes marked to be dropped */

  index = dict_table_get_first_index(prebuilt->table);

  while (index) {
    dict_index_t* next_index;

    next_index = dict_table_get_next_index(index);

    if (index->to_be_dropped) {

      row_merge_drop_index(index, prebuilt->table, trx);
    }

    index = next_index;
  }

  /* Check that all flagged indexes were dropped. */
  for (index = dict_table_get_first_index(prebuilt->table);
       index; index = dict_table_get_next_index(index)) {
    ut_a(!index->to_be_dropped);
  }

  /* We will need to rebuild index translation table. Set
  valid index entry count in the translation table to zero */
  share->idx_trans_tbl.index_count = 0;

func_exit:
  ut_d(dict_table_check_for_dup_indexes(prebuilt->table, FALSE));
  trx_commit_for_mysql(trx);
  trx_commit_for_mysql(prebuilt->trx);
  row_mysql_unlock_data_dictionary(trx);

  /* Flush the log to reduce probability that the .frm files and
  the InnoDB data dictionary get out-of-sync if the user runs
  with innodb_flush_log_at_trx_commit = 0 */

  log_buffer_flush_to_disk();

  trx_free_for_mysql(trx);

  /* Tell the InnoDB server that there might be work for
  utility threads: */

  srv_active_wake_master_thread();

  return(err);
}
#endif