~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 "config.h"

#include <plugin/myisam/myisam.h>

#include <drizzled/show.h>

#include <drizzled/error.h>

#include <drizzled/gettext.h>

#include <drizzled/data_home.h>

#include <drizzled/sql_parse.h>

#include <drizzled/my_hash.h>

#include <drizzled/sql_lex.h>

#include <drizzled/session.h>

#include <drizzled/sql_base.h>

#include "drizzled/strfunc.h"

#include <drizzled/db.h>

#include <drizzled/lock.h>

#include <drizzled/unireg.h>

#include <drizzled/item/int.h>

#include <drizzled/item/empty_string.h>

#include <drizzled/transaction_services.h>

#include "drizzled/transaction_services.h"

#include <drizzled/table_proto.h>

#include <drizzled/plugin/client.h>

#include <drizzled/table_identifier.h>

#include "drizzled/internal/m_string.h"

#include "drizzled/global_charset_info.h"

#include "drizzled/charset.h"

#include "drizzled/statement/alter_table.h"

#include "drizzled/sql_table.h"

#include "drizzled/pthread_globals.h"

#include <algorithm>

#include <sstream>

using namespace std;

namespace drizzled

{

extern plugin::StorageEngine *myisam_engine;

extern pid_t current_pid;

bool is_primary_key(KEY *key_info)

{

  static const char * primary_key_name="PRIMARY";

  return (strcmp(key_info->name, primary_key_name)==0);

}

const char* is_primary_key_name(const char* key_name)

{

  static const char * primary_key_name="PRIMARY";

  if (strcmp(key_name, primary_key_name)==0)

    return key_name;

  else

    return NULL;

}

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 bool prepare_blob_field(Session *session, CreateField *sql_field);

void set_table_default_charset(HA_CREATE_INFO *create_info, const char *db)

{

  /*

    If the table character set was not given explicitly,

    let's fetch the database default character set and

    apply it to the table.

  */

  if (create_info->default_table_charset == NULL)

    create_info->default_table_charset= plugin::StorageEngine::getSchemaCollation(db);

}

/*

  SYNOPSIS

    write_bin_log()

    session                           Thread object

    query                         Query to log

    query_length                  Length of query

  RETURN VALUES

    NONE

  DESCRIPTION

    Write the binlog if open, routine used in multiple places in this

    cursor

*/

void write_bin_log(Session *session,

                   char const *query)

{

  TransactionServices &transaction_services= TransactionServices::singleton();

  transaction_services.rawStatement(session, query);

}

/* Should should be refactored to go away */

void write_bin_log_drop_table(Session *session, bool if_exists, const char *db_name, const char *table_name)

{

  TransactionServices &transaction_services= TransactionServices::singleton();

  string built_query;

  if (if_exists)

    built_query.append("DROP TABLE IF EXISTS ");

  else

    built_query.append("DROP TABLE ");

  built_query.append("`");

  if (session->db.empty() || strcmp(db_name, session->db.c_str()) != 0)

  {

    built_query.append(db_name);

    built_query.append("`.`");

  }

  built_query.append(table_name);

  built_query.append("`");

  transaction_services.rawStatement(session, built_query);

}

/*

  Execute the drop of a normal or temporary table

  SYNOPSIS

    mysql_rm_table_part2()

    session			Thread Cursor

    tables		Tables to drop

    if_exists		If set, don't give an error if table doesn't exists.

			In this case we give an warning of level 'NOTE'

    drop_temporary	Only drop temporary tables

  TODO:

    When logging to the binary log, we should log

    tmp_tables and transactional tables as separate statements if we

    are in a transaction;  This is needed to get these tables into the

    cached binary log that is only written on COMMIT.

   The current code only writes DROP statements that only uses temporary

   tables to the cache binary log.  This should be ok on most cases, but

   not all.

 RETURN

   0	ok

   1	Error

   -1	Thread was killed

*/

int mysql_rm_table_part2(Session *session, TableList *tables, bool if_exists,

                         bool drop_temporary)

{

  TableList *table;

  String wrong_tables;

  int error= 0;

  bool foreign_key_error= false;

  pthread_mutex_lock(&LOCK_open); /* Part 2 of rm a table */

  /*

    If we have the table in the definition cache, we don't have to check the

    .frm cursor to find if the table is a normal table (not view) and what

    engine to use.

  */

  for (table= tables; table; table= table->next_local)

  {

    TableShare *share;

    table->db_type= NULL;

    if ((share= TableShare::getShare(table->db, table->table_name)))

      table->db_type= share->db_type();

  }

  if (not drop_temporary && lock_table_names_exclusively(session, tables))

  {

    pthread_mutex_unlock(&LOCK_open);

    return 1;

  }

  /* Don't give warnings for not found errors, as we already generate notes */

  session->no_warnings_for_error= 1;

  for (table= tables; table; table= table->next_local)

  {

    char *db=table->db;

    error= session->drop_temporary_table(table);

    switch (error) {

    case  0:

      // removed temporary table

      continue;

    case -1:

      error= 1;

      goto err_with_placeholders;

    default:

      // temporary table not found

      error= 0;

    }

    if (drop_temporary == false)

    {

      Table *locked_table;

      abort_locked_tables(session, db, table->table_name);

      remove_table_from_cache(session, db, table->table_name,

                              RTFC_WAIT_OTHER_THREAD_FLAG |

                              RTFC_CHECK_KILLED_FLAG);

      /*

        If the table was used in lock tables, remember it so that

        unlock_table_names can free it

      */

      if ((locked_table= drop_locked_tables(session, db, table->table_name)))

        table->table= locked_table;

      if (session->killed)

      {

        error= -1;

        goto err_with_placeholders;

      }

    }

    TableIdentifier identifier(db, table->table_name, table->internal_tmp_table ? INTERNAL_TMP_TABLE : STANDARD_TABLE);

    if (drop_temporary || not plugin::StorageEngine::doesTableExist(*session, identifier))

    {

      // Table was not found on disk and table can't be created from engine

      if (if_exists)

        push_warning_printf(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,

                            ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR),

                            table->table_name);

      else

        error= 1;

    }

    else

    {

      error= plugin::StorageEngine::dropTable(*session, identifier);

      if ((error == ENOENT || error == HA_ERR_NO_SUCH_TABLE) && if_exists)

      {

        error= 0;

        session->clear_error();

      }

      if (error == HA_ERR_ROW_IS_REFERENCED)

      {

        /* the table is referenced by a foreign key constraint */

        foreign_key_error= true;

      }

    }

    if (error == 0 || (if_exists && foreign_key_error == false))

    {

      TransactionServices &transaction_services= TransactionServices::singleton();

      transaction_services.dropTable(session, string(db), string(table->table_name), if_exists);

    }

    if (error)

    {

      if (wrong_tables.length())

        wrong_tables.append(',');

      wrong_tables.append(String(table->table_name,system_charset_info));

    }

  }

  /*

    It's safe to unlock LOCK_open: we have an exclusive lock

    on the table name.

  */

  pthread_mutex_unlock(&LOCK_open);

  error= 0;

  if (wrong_tables.length())

  {

    if (!foreign_key_error)

      my_printf_error(ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR), MYF(0),

                      wrong_tables.c_ptr());

    else

    {

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

    }

    error= 1;

  }

  pthread_mutex_lock(&LOCK_open); /* final bit in rm table lock */

err_with_placeholders:

  unlock_table_names(tables, NULL);

  pthread_mutex_unlock(&LOCK_open);

  session->no_warnings_for_error= 0;

  return(error);

}

/*

  Quickly remove a table.

  SYNOPSIS

    quick_rm_table()

      base                      The plugin::StorageEngine handle.

      db                        The database name.

      table_name                The table name.

      is_tmp                    If the table is temp.

  RETURN

    0           OK

    != 0        Error

*/

bool quick_rm_table(Session& session,

                    TableIdentifier &identifier)

{

  return (plugin::StorageEngine::dropTable(session, identifier));

}

/*

  Sort keys in the following order:

  - PRIMARY KEY

  - UNIQUE keys where all column are NOT NULL

  - UNIQUE keys that don't contain partial segments

  - Other UNIQUE keys

  - Normal keys

  - Fulltext keys

  This will make checking for duplicated keys faster and ensure that

  PRIMARY keys are prioritized.

*/

static int sort_keys(KEY *a, KEY *b)

{

  ulong a_flags= a->flags, b_flags= b->flags;

  if (a_flags & HA_NOSAME)

  {

    if (!(b_flags & HA_NOSAME))

      return -1;

    if ((a_flags ^ b_flags) & (HA_NULL_PART_KEY))

    {

      /* Sort NOT NULL keys before other keys */

      return (a_flags & (HA_NULL_PART_KEY)) ? 1 : -1;

    }

    if (is_primary_key(a))

      return -1;

    if (is_primary_key(b))

      return 1;

    /* Sort keys don't containing partial segments before others */

    if ((a_flags ^ b_flags) & HA_KEY_HAS_PART_KEY_SEG)

      return (a_flags & HA_KEY_HAS_PART_KEY_SEG) ? 1 : -1;

  }

  else if (b_flags & HA_NOSAME)

    return 1;					// Prefer b

  /*

    Prefer original key order.	usable_key_parts contains here

    the original key position.

  */

  return ((a->usable_key_parts < b->usable_key_parts) ? -1 :

          (a->usable_key_parts > b->usable_key_parts) ? 1 :

          0);

}

/*

  Check TYPELIB (set or enum) for duplicates

  SYNOPSIS

    check_duplicates_in_interval()

    set_or_name   "SET" or "ENUM" string for warning message

    name	  name of the checked column

    typelib	  list of values for the column

    dup_val_count  returns count of duplicate elements

  DESCRIPTION

    This function prints an warning for each value in list

    which has some duplicates on its right

  RETURN VALUES

    0             ok

    1             Error

*/

static bool check_duplicates_in_interval(const char *set_or_name,

                                         const char *name, TYPELIB *typelib,

                                         const CHARSET_INFO * const cs,

                                         unsigned int *dup_val_count)

{

  TYPELIB tmp= *typelib;

  const char **cur_value= typelib->type_names;

  unsigned int *cur_length= typelib->type_lengths;

  *dup_val_count= 0;

  for ( ; tmp.count > 1; cur_value++, cur_length++)

  {

    tmp.type_names++;

    tmp.type_lengths++;

    tmp.count--;

    if (find_type2(&tmp, (const char*)*cur_value, *cur_length, cs))

    {

      my_error(ER_DUPLICATED_VALUE_IN_TYPE, MYF(0),

               name,*cur_value,set_or_name);

      return 1;

    }

  }

  return 0;

}

/*

  Check TYPELIB (set or enum) max and total lengths

  SYNOPSIS

    calculate_interval_lengths()

    cs            charset+collation pair of the interval

    typelib       list of values for the column

    max_length    length of the longest item

    tot_length    sum of the item lengths

  DESCRIPTION

    After this function call:

    - ENUM uses max_length

    - SET uses tot_length.

  RETURN VALUES

    void

*/

static void calculate_interval_lengths(const CHARSET_INFO * const cs,

                                       TYPELIB *interval,

                                       uint32_t *max_length,

                                       uint32_t *tot_length)

{

  const char **pos;

  uint32_t *len;

  *max_length= *tot_length= 0;

  for (pos= interval->type_names, len= interval->type_lengths;

       *pos ; pos++, len++)

  {

    uint32_t length= cs->cset->numchars(cs, *pos, *pos + *len);

    *tot_length+= length;

    set_if_bigger(*max_length, (uint32_t)length);

  }

}

/*

  Prepare a create_table instance for packing

  SYNOPSIS

    prepare_create_field()

    sql_field     field to prepare for packing

    blob_columns  count for BLOBs

    timestamps    count for timestamps

  DESCRIPTION

    This function prepares a CreateField instance.

    Fields such as pack_flag are valid after this call.

  RETURN VALUES

   0	ok

   1	Error

*/

int prepare_create_field(CreateField *sql_field,

                         uint32_t *blob_columns,

                         int *timestamps,

                         int *timestamps_with_niladic)

{

  unsigned int dup_val_count;

  /*

    This code came from mysql_prepare_create_table.

    Indent preserved to make patching easier

  */

  assert(sql_field->charset);

  switch (sql_field->sql_type) {

  case DRIZZLE_TYPE_BLOB:

    sql_field->pack_flag= pack_length_to_packflag(sql_field->pack_length - portable_sizeof_char_ptr);

    sql_field->length= 8; // Unireg field length

    (*blob_columns)++;

    break;

  case DRIZZLE_TYPE_VARCHAR:

    sql_field->pack_flag=0;

    break;

  case DRIZZLE_TYPE_ENUM:

    sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length);

    if (check_duplicates_in_interval("ENUM",

                                     sql_field->field_name,

                                     sql_field->interval,

                                     sql_field->charset,

                                     &dup_val_count))

      return 1;

    break;

  case DRIZZLE_TYPE_DATE:  // Rest of string types

  case DRIZZLE_TYPE_DATETIME:

  case DRIZZLE_TYPE_NULL:

    sql_field->pack_flag=f_settype((uint32_t) sql_field->sql_type);

    break;

  case DRIZZLE_TYPE_DECIMAL:

    sql_field->pack_flag= 0;

    break;

  case DRIZZLE_TYPE_TIMESTAMP:

    /* We should replace old TIMESTAMP fields with their newer analogs */

    if (sql_field->unireg_check == Field::TIMESTAMP_OLD_FIELD)

    {

      if (!*timestamps)

      {

        sql_field->unireg_check= Field::TIMESTAMP_DNUN_FIELD;

        (*timestamps_with_niladic)++;

      }

      else

        sql_field->unireg_check= Field::NONE;

    }

    else if (sql_field->unireg_check != Field::NONE)

      (*timestamps_with_niladic)++;

    (*timestamps)++;

    /* fall-through */

  default:

    sql_field->pack_flag=(0 |

                          f_settype((uint32_t) sql_field->sql_type));

    break;

  }

  return 0;

}

static int mysql_prepare_create_table(Session *session,

                                      HA_CREATE_INFO *create_info,

                                      message::Table &create_proto,

                                      AlterInfo *alter_info,

                                      bool tmp_table,

                                      uint32_t *db_options,

                                      KEY **key_info_buffer,

                                      uint32_t *key_count,

                                      int select_field_count)

{

  const char	*key_name;

  CreateField	*sql_field,*dup_field;

  uint		field,null_fields,blob_columns,max_key_length;

  ulong		record_offset= 0;

  KEY		*key_info;

  KEY_PART_INFO *key_part_info;

  int		timestamps= 0, timestamps_with_niladic= 0;

  int		field_no,dup_no;

  int		select_field_pos,auto_increment=0;

  List_iterator<CreateField> it(alter_info->create_list);

  List_iterator<CreateField> it2(alter_info->create_list);

  uint32_t total_uneven_bit_length= 0;

  plugin::StorageEngine *engine= plugin::StorageEngine::findByName(create_proto.engine().name());

  select_field_pos= alter_info->create_list.elements - select_field_count;

  null_fields=blob_columns=0;

  max_key_length= engine->max_key_length();

  for (field_no=0; (sql_field=it++) ; field_no++)

  {

    const CHARSET_INFO *save_cs;

    /*

      Initialize length from its original value (number of characters),

      which was set in the parser. This is necessary if we're

      executing a prepared statement for the second time.

    */

    sql_field->length= sql_field->char_length;

    if (!sql_field->charset)

      sql_field->charset= create_info->default_table_charset;

    /*

      table_charset is set in ALTER Table if we want change character set

      for all varchar/char columns.

      But the table charset must not affect the BLOB fields, so don't

      allow to change my_charset_bin to somethig else.

    */

    if (create_info->table_charset && sql_field->charset != &my_charset_bin)

      sql_field->charset= create_info->table_charset;

    save_cs= sql_field->charset;

    if ((sql_field->flags & BINCMP_FLAG) &&

        !(sql_field->charset= get_charset_by_csname(sql_field->charset->csname, MY_CS_BINSORT)))

    {

      char tmp[64];

      char *tmp_pos= tmp;

      strncpy(tmp_pos, save_cs->csname, sizeof(tmp)-4);

      tmp_pos+= strlen(tmp);

      strncpy(tmp_pos, STRING_WITH_LEN("_bin"));

      my_error(ER_UNKNOWN_COLLATION, MYF(0), tmp);

      return(true);

    }

    /*

      Convert the default value from client character

      set into the column character set if necessary.

    */

    if (sql_field->def &&

        save_cs != sql_field->def->collation.collation &&

        (sql_field->sql_type == DRIZZLE_TYPE_ENUM))

    {

      /*

        Starting from 5.1 we work here with a copy of CreateField

        created by the caller, not with the instance that was

        originally created during parsing. It's OK to create

        a temporary item and initialize with it a member of the

        copy -- this item will be thrown away along with the copy

        at the end of execution, and thus not introduce a dangling

        pointer in the parsed tree of a prepared statement or a

        stored procedure statement.

      */

      sql_field->def= sql_field->def->safe_charset_converter(save_cs);

      if (sql_field->def == NULL)

      {

        /* Could not convert */

        my_error(ER_INVALID_DEFAULT, MYF(0), sql_field->field_name);

        return(true);

      }

    }

    if (sql_field->sql_type == DRIZZLE_TYPE_ENUM)

    {

      uint32_t dummy;

      const CHARSET_INFO * const cs= sql_field->charset;

      TYPELIB *interval= sql_field->interval;

      /*

        Create typelib from interval_list, and if necessary

        convert strings from client character set to the

        column character set.

      */

      if (!interval)

      {

        /*

          Create the typelib in runtime memory - we will free the

          occupied memory at the same time when we free this

          sql_field -- at the end of execution.

        */

        interval= sql_field->interval= typelib(session->mem_root,

                                               sql_field->interval_list);

        List_iterator<String> int_it(sql_field->interval_list);

        String conv, *tmp;

        char comma_buf[4];

        int comma_length= cs->cset->wc_mb(cs, ',', (unsigned char*) comma_buf,

                                          (unsigned char*) comma_buf +

                                          sizeof(comma_buf));

        assert(comma_length > 0);

        for (uint32_t i= 0; (tmp= int_it++); i++)

        {

          uint32_t lengthsp;

          if (String::needs_conversion(tmp->length(), tmp->charset(),

                                       cs, &dummy))

          {

            uint32_t cnv_errs;

            conv.copy(tmp->ptr(), tmp->length(), tmp->charset(), cs, &cnv_errs);

            interval->type_names[i]= strmake_root(session->mem_root, conv.ptr(),

                                                  conv.length());

            interval->type_lengths[i]= conv.length();

          }

          // Strip trailing spaces.

          lengthsp= cs->cset->lengthsp(cs, interval->type_names[i],

                                       interval->type_lengths[i]);

          interval->type_lengths[i]= lengthsp;

          ((unsigned char *)interval->type_names[i])[lengthsp]= '\0';

        }

        sql_field->interval_list.empty(); // Don't need interval_list anymore

      }

      /* DRIZZLE_TYPE_ENUM */

      {

        uint32_t field_length;

        assert(sql_field->sql_type == DRIZZLE_TYPE_ENUM);

        if (sql_field->def != NULL)

        {

          String str, *def= sql_field->def->val_str(&str);

          if (def == NULL) /* SQL "NULL" maps to NULL */

          {

            if ((sql_field->flags & NOT_NULL_FLAG) != 0)

            {

              my_error(ER_INVALID_DEFAULT, MYF(0), sql_field->field_name);

              return(true);

            }

            /* else, the defaults yield the correct length for NULLs. */

          }

          else /* not NULL */

          {

            def->length(cs->cset->lengthsp(cs, def->ptr(), def->length()));