~drizzle-trunk/drizzle/development

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

/* drop and alter of tables */

#include "mysql_priv.h"

#include <hash.h>

#include <myisam.h>

#include <my_dir.h>

#include "sql_show.h"

int creating_table= 0;        // How many mysql_create_table are running

const char *primary_key_name="PRIMARY";

static bool check_if_keyname_exists(const char *name,KEY *start, KEY *end);

static char *make_unique_key_name(const char *field_name,KEY *start,KEY *end);

static int copy_data_between_tables(TABLE *from,TABLE *to,

                                    List<Create_field> &create, bool ignore,

                                    uint order_num, ORDER *order,

                                    ha_rows *copied,ha_rows *deleted,

                                    enum enum_enable_or_disable keys_onoff,

                                    bool error_if_not_empty);

static bool prepare_blob_field(THD *thd, Create_field *sql_field);

static bool check_engine(THD *, const char *, HA_CREATE_INFO *);

static int

mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,

                           Alter_info *alter_info,

                           bool tmp_table,

                               uint *db_options,

                               handler *file, KEY **key_info_buffer,

                               uint *key_count, int select_field_count);

static bool

mysql_prepare_alter_table(THD *thd, TABLE *table,

                          HA_CREATE_INFO *create_info,

                          Alter_info *alter_info);

#ifndef DBUG_OFF

/* Wait until we get a 'mysql_kill' signal */

static void wait_for_kill_signal(THD *thd)

{

  while (thd->killed == 0)

    sleep(1);

  // Reset signal and continue as if nothing happend

  thd->killed= THD::NOT_KILLED;

}

#endif

/*

  Translate a file name to a table name (WL #1324).

  SYNOPSIS

    filename_to_tablename()

      from                      The file name in my_charset_filename.

      to                OUT     The table name in system_charset_info.

      to_length                 The size of the table name buffer.

  RETURN

    Table name length.

*/

uint filename_to_tablename(const char *from, char *to, uint to_length)

{

  uint errors;

  uint res;

  if (!memcmp(from, tmp_file_prefix, tmp_file_prefix_length))

  {

    /* Temporary table name. */

    res= (strnmov(to, from, to_length) - to);

  }

  else

  {

    res= strconvert(&my_charset_filename, from,

                    system_charset_info,  to, to_length, &errors);

    if (errors) // Old 5.0 name

    {

      res= (strxnmov(to, to_length, MYSQL50_TABLE_NAME_PREFIX,  from, NullS) -

            to);

      sql_print_error("Invalid (old?) table or database name '%s'", from);

      /*

        TODO: add a stored procedure for fix table and database names,

        and mention its name in error log.

      */

    }

  }

  return(res);

}

/*

  Translate a table name to a file name (WL #1324).

  SYNOPSIS

    tablename_to_filename()

      from                      The table name in system_charset_info.

      to                OUT     The file name in my_charset_filename.

      to_length                 The size of the file name buffer.

  RETURN

    File name length.

*/

uint tablename_to_filename(const char *from, char *to, uint to_length)

{

  uint errors, length;

  if (from[0] == '#' && !strncmp(from, MYSQL50_TABLE_NAME_PREFIX,

                                 MYSQL50_TABLE_NAME_PREFIX_LENGTH))

    return((uint) (strmake(to, from+MYSQL50_TABLE_NAME_PREFIX_LENGTH,

                                to_length-1) -

                        (from + MYSQL50_TABLE_NAME_PREFIX_LENGTH)));

  length= strconvert(system_charset_info, from,

                     &my_charset_filename, to, to_length, &errors);

  if (check_if_legal_tablename(to) &&

      length + 4 < to_length)

  {

    memcpy(to + length, "@@@", 4);

    length+= 3;

  }

  return(length);

}

/*

  Creates path to a file: mysql_data_dir/db/table.ext

  SYNOPSIS

   build_table_filename()

     buff                       Where to write result in my_charset_filename.

                                This may be the same as table_name.

     bufflen                    buff size

     db                         Database name in system_charset_info.

     table_name                 Table name in system_charset_info.

     ext                        File extension.

     flags                      FN_FROM_IS_TMP or FN_TO_IS_TMP or FN_IS_TMP

                                table_name is temporary, do not change.

  NOTES

    Uses database and table name, and extension to create

    a file name in mysql_data_dir. Database and table

    names are converted from system_charset_info into "fscs".

    Unless flags indicate a temporary table name.

    'db' is always converted.

    'ext' is not converted.

    The conversion suppression is required for ALTER TABLE. This

    statement creates intermediate tables. These are regular

    (non-temporary) tables with a temporary name. Their path names must

    be derivable from the table name. So we cannot use

    build_tmptable_filename() for them.

  RETURN

    path length

*/

uint build_table_filename(char *buff, size_t bufflen, const char *db,

                          const char *table_name, const char *ext, uint flags)

{

  char dbbuff[FN_REFLEN];

  char tbbuff[FN_REFLEN];

  if (flags & FN_IS_TMP) // FN_FROM_IS_TMP | FN_TO_IS_TMP

    strnmov(tbbuff, table_name, sizeof(tbbuff));

  else

    VOID(tablename_to_filename(table_name, tbbuff, sizeof(tbbuff)));

  VOID(tablename_to_filename(db, dbbuff, sizeof(dbbuff)));

  char *end = buff + bufflen;

  /* Don't add FN_ROOTDIR if mysql_data_home already includes it */

  char *pos = strnmov(buff, mysql_data_home, bufflen);

  int rootdir_len= strlen(FN_ROOTDIR);

  if (pos - rootdir_len >= buff &&

      memcmp(pos - rootdir_len, FN_ROOTDIR, rootdir_len) != 0)

    pos= strnmov(pos, FN_ROOTDIR, end - pos);

  pos= strxnmov(pos, end - pos, dbbuff, FN_ROOTDIR, NullS);

#ifdef USE_SYMDIR

  unpack_dirname(buff, buff);

  pos= strend(buff);

#endif

  pos= strxnmov(pos, end - pos, tbbuff, ext, NullS);

  return(pos - buff);

}

/*

  Creates path to a file: mysql_tmpdir/#sql1234_12_1.ext

  SYNOPSIS

   build_tmptable_filename()

     thd                        The thread handle.

     buff                       Where to write result in my_charset_filename.

     bufflen                    buff size

  NOTES

    Uses current_pid, thread_id, and tmp_table counter to create

    a file name in mysql_tmpdir.

  RETURN

    path length

*/

uint build_tmptable_filename(THD* thd, char *buff, size_t bufflen)

{

  char *p= strnmov(buff, mysql_tmpdir, bufflen);

  snprintf(p, bufflen - (p - buff), "/%s%lx_%lx_%x%s",

	      tmp_file_prefix, current_pid,

              thd->thread_id, thd->tmp_table++, reg_ext);

  if (lower_case_table_names)

  {

    /* Convert all except tmpdir to lower case */

    my_casedn_str(files_charset_info, p);

  }

  uint length= unpack_filename(buff, buff);

  return(length);

}

/*

--------------------------------------------------------------------------

   MODULE: DDL log

   -----------------

   This module is used to ensure that we can recover from crashes that occur

   in the middle of a meta-data operation in MySQL. E.g. DROP TABLE t1, t2;

   We need to ensure that both t1 and t2 are dropped and not only t1 and

   also that each table drop is entirely done and not "half-baked".

   To support this we create log entries for each meta-data statement in the

   ddl log while we are executing. These entries are dropped when the

   operation is completed.

   At recovery those entries that were not completed will be executed.

   There is only one ddl log in the system and it is protected by a mutex

   and there is a global struct that contains information about its current

   state.

   History:

   First version written in 2006 by Mikael Ronstrom

--------------------------------------------------------------------------

*/

struct st_global_ddl_log

{

  /*

    We need to adjust buffer size to be able to handle downgrades/upgrades

    where IO_SIZE has changed. We'll set the buffer size such that we can

    handle that the buffer size was upto 4 times bigger in the version

    that wrote the DDL log.

  */

  char file_entry_buf[4*IO_SIZE];

  char file_name_str[FN_REFLEN];

  char *file_name;

  DDL_LOG_MEMORY_ENTRY *first_free;

  DDL_LOG_MEMORY_ENTRY *first_used;

  uint num_entries;

  File file_id;

  uint name_len;

  uint io_size;

  bool inited;

  bool do_release;

  bool recovery_phase;

  st_global_ddl_log() : inited(false), do_release(false) {}

};

st_global_ddl_log global_ddl_log;

pthread_mutex_t LOCK_gdl;

#define DDL_LOG_ENTRY_TYPE_POS 0

#define DDL_LOG_ACTION_TYPE_POS 1

#define DDL_LOG_PHASE_POS 2

#define DDL_LOG_NEXT_ENTRY_POS 4

#define DDL_LOG_NAME_POS 8

#define DDL_LOG_NUM_ENTRY_POS 0

#define DDL_LOG_NAME_LEN_POS 4

#define DDL_LOG_IO_SIZE_POS 8

/*

  Read one entry from ddl log file

  SYNOPSIS

    read_ddl_log_file_entry()

    entry_no                     Entry number to read

  RETURN VALUES

    true                         Error

    false                        Success

*/

static bool read_ddl_log_file_entry(uint entry_no)

{

  bool error= false;

  File file_id= global_ddl_log.file_id;

  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;

  ssize_t io_size= (ssize_t)global_ddl_log.io_size;

  if (pread(file_id, file_entry_buf, io_size, io_size * entry_no) != io_size)

    error= true;

  return(error);

}

/*

  Write one entry from ddl log file

  SYNOPSIS

    write_ddl_log_file_entry()

    entry_no                     Entry number to read

  RETURN VALUES

    true                         Error

    false                        Success

*/

static bool write_ddl_log_file_entry(uint entry_no)

{

  bool error= false;

  File file_id= global_ddl_log.file_id;

  char *file_entry_buf= (char*)global_ddl_log.file_entry_buf;

  if (pwrite(file_id, (uchar*)file_entry_buf,

                IO_SIZE, IO_SIZE * entry_no) != IO_SIZE)

    error= true;

  return(error);

}

/*

  Write ddl log header

  SYNOPSIS

    write_ddl_log_header()

  RETURN VALUES

    true                      Error

    false                     Success

*/

static bool write_ddl_log_header()

{

  uint16 const_var;

  bool error= false;

  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NUM_ENTRY_POS],

            global_ddl_log.num_entries);

  const_var= FN_LEN;

  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_LEN_POS],

            (ulong) const_var);

  const_var= IO_SIZE;

  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_IO_SIZE_POS],

            (ulong) const_var);

  if (write_ddl_log_file_entry(0UL))

  {

    sql_print_error("Error writing ddl log header");

    return(true);

  }

  VOID(sync_ddl_log());

  return(error);

}

/*

  Create ddl log file name

  SYNOPSIS

    create_ddl_log_file_name()

    file_name                   Filename setup

  RETURN VALUES

    NONE

*/

static inline void create_ddl_log_file_name(char *file_name)

{

  strxmov(file_name, mysql_data_home, "/", "ddl_log.log", NullS);

}

/*

  Read header of ddl log file

  SYNOPSIS

    read_ddl_log_header()

  RETURN VALUES

    > 0                  Last entry in ddl log

    0                    No entries in ddl log

  DESCRIPTION

    When we read the ddl log header we get information about maximum sizes

    of names in the ddl log and we also get information about the number

    of entries in the ddl log.

*/

static uint read_ddl_log_header()

{

  char *file_entry_buf= (char*)global_ddl_log.file_entry_buf;

  char file_name[FN_REFLEN];

  uint entry_no;

  bool successful_open= false;

  create_ddl_log_file_name(file_name);

  if ((global_ddl_log.file_id= my_open(file_name,

                                        O_RDWR | O_BINARY, MYF(0))) >= 0)

  {

    if (read_ddl_log_file_entry(0UL))

    {

      /* Write message into error log */

      sql_print_error("Failed to read ddl log file in recovery");

    }

    else

      successful_open= true;

  }

  entry_no= uint4korr(&file_entry_buf[DDL_LOG_NUM_ENTRY_POS]);

  global_ddl_log.name_len= uint4korr(&file_entry_buf[DDL_LOG_NAME_LEN_POS]);

  if (successful_open)

  {

    global_ddl_log.io_size= uint4korr(&file_entry_buf[DDL_LOG_IO_SIZE_POS]);

    assert(global_ddl_log.io_size <=

                sizeof(global_ddl_log.file_entry_buf));

  }

  else

  {

    entry_no= 0;

  }

  global_ddl_log.first_free= NULL;

  global_ddl_log.first_used= NULL;

  global_ddl_log.num_entries= 0;

  VOID(pthread_mutex_init(&LOCK_gdl, MY_MUTEX_INIT_FAST));

  global_ddl_log.do_release= true;

  return(entry_no);

}

/*

  Read a ddl log entry

  SYNOPSIS

    read_ddl_log_entry()

    read_entry               Number of entry to read

    out:entry_info           Information from entry

  RETURN VALUES

    true                     Error

    false                    Success

  DESCRIPTION

    Read a specified entry in the ddl log

*/

bool read_ddl_log_entry(uint read_entry, DDL_LOG_ENTRY *ddl_log_entry)

{

  char *file_entry_buf= (char*)&global_ddl_log.file_entry_buf;

  uint inx;

  uchar single_char;

  if (read_ddl_log_file_entry(read_entry))

  {

    return(true);

  }

  ddl_log_entry->entry_pos= read_entry;

  single_char= file_entry_buf[DDL_LOG_ENTRY_TYPE_POS];

  ddl_log_entry->entry_type= (enum ddl_log_entry_code)single_char;

  single_char= file_entry_buf[DDL_LOG_ACTION_TYPE_POS];

  ddl_log_entry->action_type= (enum ddl_log_action_code)single_char;

  ddl_log_entry->phase= file_entry_buf[DDL_LOG_PHASE_POS];

  ddl_log_entry->next_entry= uint4korr(&file_entry_buf[DDL_LOG_NEXT_ENTRY_POS]);

  ddl_log_entry->name= &file_entry_buf[DDL_LOG_NAME_POS];

  inx= DDL_LOG_NAME_POS + global_ddl_log.name_len;

  ddl_log_entry->from_name= &file_entry_buf[inx];

  inx+= global_ddl_log.name_len;

  ddl_log_entry->handler_name= &file_entry_buf[inx];

  return(false);

}

/*

  Initialise ddl log

  SYNOPSIS

    init_ddl_log()

  DESCRIPTION

    Write the header of the ddl log file and length of names. Also set

    number of entries to zero.

  RETURN VALUES

    true                     Error

    false                    Success

*/

static bool init_ddl_log()

{

  char file_name[FN_REFLEN];

  if (global_ddl_log.inited)

    goto end;

  global_ddl_log.io_size= IO_SIZE;

  create_ddl_log_file_name(file_name);

  if ((global_ddl_log.file_id= my_create(file_name,

                                         CREATE_MODE,

                                         O_RDWR | O_TRUNC | O_BINARY,

                                         MYF(MY_WME))) < 0)

  {

    /* Couldn't create ddl log file, this is serious error */

    sql_print_error("Failed to open ddl log file");

    return(true);

  }

  global_ddl_log.inited= true;

  if (write_ddl_log_header())

  {

    VOID(my_close(global_ddl_log.file_id, MYF(MY_WME)));

    global_ddl_log.inited= false;

    return(true);

  }

end:

  return(false);

}

/*

  Execute one action in a ddl log entry

  SYNOPSIS

    execute_ddl_log_action()

    ddl_log_entry              Information in action entry to execute

  RETURN VALUES

    true                       Error

    false                      Success

*/

static int execute_ddl_log_action(THD *thd, DDL_LOG_ENTRY *ddl_log_entry)

{

  bool frm_action= false;

  LEX_STRING handler_name;

  handler *file= NULL;

  MEM_ROOT mem_root;

  int error= true;

  char to_path[FN_REFLEN];

  char from_path[FN_REFLEN];

  handlerton *hton;

  if (ddl_log_entry->entry_type == DDL_IGNORE_LOG_ENTRY_CODE)

  {

    return(false);

  }

  handler_name.str= (char*)ddl_log_entry->handler_name;

  handler_name.length= strlen(ddl_log_entry->handler_name);

  init_sql_alloc(&mem_root, TABLE_ALLOC_BLOCK_SIZE, 0); 

  if (!strcmp(ddl_log_entry->handler_name, reg_ext))

    frm_action= true;

  else

  {

    plugin_ref plugin= ha_resolve_by_name(thd, &handler_name);

    if (!plugin)

    {

      my_error(ER_ILLEGAL_HA, MYF(0), ddl_log_entry->handler_name);

      goto error;

    }

    hton= plugin_data(plugin, handlerton*);

    file= get_new_handler((TABLE_SHARE*)0, &mem_root, hton);

    if (!file)

    {

      my_error(ER_OUTOFMEMORY, MYF(0), sizeof(handler));

      goto error;

    }

  }

  switch (ddl_log_entry->action_type)

  {

    case DDL_LOG_REPLACE_ACTION:

    case DDL_LOG_DELETE_ACTION:

    {

      if (ddl_log_entry->phase == 0)

      {

        if (frm_action)

        {

          strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);

          if ((error= my_delete(to_path, MYF(MY_WME))))

          {

            if (my_errno != ENOENT)

              break;

          }

        }

        else

        {

          if ((error= file->ha_delete_table(ddl_log_entry->name)))

          {

            if (error != ENOENT && error != HA_ERR_NO_SUCH_TABLE)

              break;

          }

        }

        if ((deactivate_ddl_log_entry(ddl_log_entry->entry_pos)))

          break;

        VOID(sync_ddl_log());

        error= false;

        if (ddl_log_entry->action_type == DDL_LOG_DELETE_ACTION)

          break;

      }

      assert(ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION);

      /*

        Fall through and perform the rename action of the replace

        action. We have already indicated the success of the delete

        action in the log entry by stepping up the phase.

      */

    }

    case DDL_LOG_RENAME_ACTION:

    {

      error= true;

      if (frm_action)

      {

        strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);

        strxmov(from_path, ddl_log_entry->from_name, reg_ext, NullS);

        if (my_rename(from_path, to_path, MYF(MY_WME)))

          break;

      }

      else

      {

        if (file->ha_rename_table(ddl_log_entry->from_name,

                                  ddl_log_entry->name))

          break;

      }

      if ((deactivate_ddl_log_entry(ddl_log_entry->entry_pos)))

        break;

      VOID(sync_ddl_log());

      error= false;

      break;

    }

    default:

      assert(0);

      break;

  }

  delete file;

error:

  free_root(&mem_root, MYF(0)); 

  return(error);

}

/*

  Get a free entry in the ddl log

  SYNOPSIS

    get_free_ddl_log_entry()

    out:active_entry                A ddl log memory entry returned

  RETURN VALUES

    true                       Error

    false                      Success

*/

static bool get_free_ddl_log_entry(DDL_LOG_MEMORY_ENTRY **active_entry,

                                   bool *write_header)

{

  DDL_LOG_MEMORY_ENTRY *used_entry;

  DDL_LOG_MEMORY_ENTRY *first_used= global_ddl_log.first_used;

  if (global_ddl_log.first_free == NULL)

  {

    if (!(used_entry= (DDL_LOG_MEMORY_ENTRY*)my_malloc(

                              sizeof(DDL_LOG_MEMORY_ENTRY), MYF(MY_WME))))

    {

      sql_print_error("Failed to allocate memory for ddl log free list");

      return(true);

    }

    global_ddl_log.num_entries++;

    used_entry->entry_pos= global_ddl_log.num_entries;

    *write_header= true;

  }

  else

  {

    used_entry= global_ddl_log.first_free;

    global_ddl_log.first_free= used_entry->next_log_entry;

    *write_header= false;

  }

  /*

    Move from free list to used list

  */

  used_entry->next_log_entry= first_used;

  used_entry->prev_log_entry= NULL;

  global_ddl_log.first_used= used_entry;

  if (first_used)

    first_used->prev_log_entry= used_entry;

  *active_entry= used_entry;

  return(false);

}

/*

  External interface methods for the DDL log Module

  ---------------------------------------------------

*/

/*

  SYNOPSIS

    write_ddl_log_entry()

    ddl_log_entry         Information about log entry

    out:entry_written     Entry information written into   

  RETURN VALUES

    true                      Error

    false                     Success

  DESCRIPTION

    A careful write of the ddl log is performed to ensure that we can

    handle crashes occurring during CREATE and ALTER TABLE processing.

*/

bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,

                         DDL_LOG_MEMORY_ENTRY **active_entry)

{

  bool error, write_header;

  if (init_ddl_log())

  {

    return(true);

  }

  global_ddl_log.file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]=

                                    (char)DDL_LOG_ENTRY_CODE;

  global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS]=

                                    (char)ddl_log_entry->action_type;

  global_ddl_log.file_entry_buf[DDL_LOG_PHASE_POS]= 0;

  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NEXT_ENTRY_POS],

            ddl_log_entry->next_entry);

  assert(strlen(ddl_log_entry->name) < FN_LEN);

  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],

          ddl_log_entry->name, FN_LEN - 1);

  if (ddl_log_entry->action_type == DDL_LOG_RENAME_ACTION ||

      ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION)

  {

    assert(strlen(ddl_log_entry->from_name) < FN_LEN);

    strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_LEN],

          ddl_log_entry->from_name, FN_LEN - 1);

  }

  else

    global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_LEN]= 0;

  assert(strlen(ddl_log_entry->handler_name) < FN_LEN);

  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (2*FN_LEN)],

          ddl_log_entry->handler_name, FN_LEN - 1);

  if (get_free_ddl_log_entry(active_entry, &write_header))

  {

    return(true);

  }

  error= false;

  if (write_ddl_log_file_entry((*active_entry)->entry_pos))

  {

    error= true;

    sql_print_error("Failed to write entry_no = %u",

                    (*active_entry)->entry_pos);

  }

  if (write_header && !error)

  {

    VOID(sync_ddl_log());

    if (write_ddl_log_header())

      error= true;

  }

  if (error)

    release_ddl_log_memory_entry(*active_entry);

  return(error);

}

/*

  Write final entry in the ddl log

  SYNOPSIS

    write_execute_ddl_log_entry()

    first_entry                    First entry in linked list of entries

                                   to execute, if 0 = NULL it means that

                                   the entry is removed and the entries

                                   are put into the free list.

    complete                       Flag indicating we are simply writing

                                   info about that entry has been completed

    in:out:active_entry            Entry to execute, 0 = NULL if the entry

                                   is written first time and needs to be

                                   returned. In this case the entry written

                                   is returned in this parameter

  RETURN VALUES

    true                           Error

    false                          Success

  DESCRIPTION

    This is the last write in the ddl log. The previous log entries have

    already been written but not yet synched to disk.

    We write a couple of log entries that describes action to perform.

    This entries are set-up in a linked list, however only when a first

    execute entry is put as the first entry these will be executed.

    This routine writes this first 

*/ 

bool write_execute_ddl_log_entry(uint first_entry,

                                 bool complete,