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


/* drop and alter of tables */

#include "mysql_priv.h"
#include <hash.h>
#include <myisam.h>

#ifdef __WIN__
#include <io.h>
#endif

extern HASH open_cache;
static 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,
				    enum enum_duplicates handle_duplicates,
                                    ORDER *order,
				    ha_rows *copied,ha_rows *deleted);

/*****************************************************************************
** Remove all possbile tables and give a compact errormessage for all
** wrong tables.
** This will wait for all users to free the table before dropping it
*****************************************************************************/

int mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists)
{
  int error;
  DBUG_ENTER("mysql_rm_table");

  /* mark for close and remove all cached entries */

  thd->mysys_var->current_mutex= &LOCK_open;
  thd->mysys_var->current_cond= &COND_refresh;
  VOID(pthread_mutex_lock(&LOCK_open));

  if (global_read_lock)
  {
    if (thd->global_read_lock)
    {
      my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE,MYF(0),
	       tables->real_name);
      error = 1;
      goto err;
    }
    while (global_read_lock && ! thd->killed)
    {
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
    }

  }
  error=mysql_rm_table_part2(thd,tables,if_exists,0);

 err:  
  VOID(pthread_cond_broadcast(&COND_refresh)); // Signal to refresh
  pthread_mutex_unlock(&LOCK_open);

  pthread_mutex_lock(&thd->mysys_var->mutex);
  thd->mysys_var->current_mutex= 0;
  thd->mysys_var->current_cond= 0;
  pthread_mutex_unlock(&thd->mysys_var->mutex);

  if (error)
    DBUG_RETURN(-1);
  send_ok(&thd->net);
  DBUG_RETURN(0);
}


int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
			 bool dont_log_query)
{
  TABLE_LIST *table;
  char	path[FN_REFLEN];
  String wrong_tables;
  db_type table_type;
  int error;
  bool some_tables_deleted=0;
  DBUG_ENTER("mysql_rm_table_part2");

  for (table=tables ; table ; table=table->next)
  {
    char *db=table->db ? table->db : thd->db;
    if (!close_temporary_table(thd, db, table->real_name))
    {
      some_tables_deleted=1;			// Log query
      continue;					// removed temporary table
    }

    abort_locked_tables(thd,db,table->real_name);
    while (remove_table_from_cache(thd,db,table->real_name) && !thd->killed)
    {
      dropping_tables++;
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
      dropping_tables--;
    }
    drop_locked_tables(thd,db,table->real_name);
    if (thd->killed)
    {
      VOID(pthread_cond_broadcast(&COND_refresh)); // Signal to refresh
      VOID(pthread_mutex_unlock(&LOCK_open));
      pthread_mutex_lock(&thd->mysys_var->mutex);
      thd->mysys_var->current_mutex= 0;
      thd->mysys_var->current_cond= 0;
      pthread_mutex_unlock(&thd->mysys_var->mutex);
      DBUG_RETURN(-1);
    }
    /* remove form file and isam files */
    (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table->real_name,
		   reg_ext);
    (void) unpack_filename(path,path);
    error=0;

    table_type=get_table_type(path);

    if (access(path,F_OK))
    {
      if (!if_exists)
	error=1;
    }
    else
    {
      char *end;
      *(end=fn_ext(path))=0;			// Remove extension
      error=ha_delete_table(table_type, path);
      if (error == ENOENT && if_exists)
	error = 0;
      if (!error || error == ENOENT)
      {
	/* Delete the table definition file */
	strmov(end,reg_ext);
	if (!(error=my_delete(path,MYF(MY_WME))))
	  some_tables_deleted=1;
      }
    }
    if (error)
    {
      if (wrong_tables.length())
	wrong_tables.append(',');
      wrong_tables.append(String(table->real_name));
    }
  }
  if (some_tables_deleted)
  {
    query_cache_invalidate3(thd, tables, 0);
    if (!dont_log_query)
    {
      mysql_update_log.write(thd, thd->query,thd->query_length);
      if (mysql_bin_log.is_open())
      {
	Query_log_event qinfo(thd, thd->query);
	mysql_bin_log.write(&qinfo);
      }
    }
  }

  error = 0;
  if (wrong_tables.length())
  {
    my_error(ER_BAD_TABLE_ERROR,MYF(0),wrong_tables.c_ptr());
    error=1;
  }
  DBUG_RETURN(error);
}


int quick_rm_table(enum db_type base,const char *db,
		   const char *table_name)
{
  char path[FN_REFLEN];
  int error=0;
  (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table_name,reg_ext);
  unpack_filename(path,path);
  if (my_delete(path,MYF(0)))
    error=1; /* purecov: inspected */
  sprintf(path,"%s/%s/%s",mysql_data_home,db,table_name);
  unpack_filename(path,path);
  return ha_delete_table(base,path) || error;
}

/*
  Sort keys in the following order:
  - PRIMARY KEY
  - UNIQUE keyws where all column are NOT NULL
  - 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)
{
  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 (a->name == primary_key_name)
      return -1;
    if (b->name == primary_key_name)
      return 1;
  }
  else if (b->flags & HA_NOSAME)
    return 1;					// Prefer b

  if ((a->flags ^ b->flags) & HA_FULLTEXT)
  {
    return (a->flags & HA_FULLTEXT) ? 1 : -1;
  }
  /*
    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);
}


/*****************************************************************************
 * Create a table.
 * If one creates a temporary table, this is automaticly opened
 ****************************************************************************/

int mysql_create_table(THD *thd,const char *db, const char *table_name,
		       HA_CREATE_INFO *create_info,
		       List<create_field> &fields,
		       List<Key> &keys,bool tmp_table,bool no_log)
{
  char		path[FN_REFLEN];
  const char	*key_name;
  create_field	*sql_field,*dup_field;
  int		error= -1;
  uint		db_options,field,null_fields,blob_columns;
  ulong		pos;
  KEY	*key_info,*key_info_buffer;
  KEY_PART_INFO *key_part_info;
  int		auto_increment=0;
  handler	*file;
  DBUG_ENTER("mysql_create_table");

  /*
  ** Check for duplicate fields and check type of table to create
  */

  if (!fields.elements)
  {
    my_error(ER_TABLE_MUST_HAVE_COLUMNS,MYF(0));
    DBUG_RETURN(-1);
  }
  List_iterator<create_field> it(fields),it2(fields);
  null_fields=blob_columns=0;
  db_options=create_info->table_options;
  if (create_info->row_type == ROW_TYPE_DYNAMIC)
    db_options|=HA_OPTION_PACK_RECORD;
  file=get_new_handler((TABLE*) 0, create_info->db_type);

  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
      (file->table_flags() & HA_NO_TEMP_TABLES))
  {
    my_error(ER_ILLEGAL_HA,MYF(0),table_name);
    DBUG_RETURN(-1);
  }

  /* Don't pack keys in old tables if the user has requested this */

  while ((sql_field=it++))
  {
    if ((sql_field->flags & BLOB_FLAG) ||
	sql_field->sql_type == FIELD_TYPE_VAR_STRING &&
	create_info->row_type != ROW_TYPE_FIXED)
    {
      db_options|=HA_OPTION_PACK_RECORD;
    }
    if (!(sql_field->flags & NOT_NULL_FLAG))
      null_fields++;
    while ((dup_field=it2++) != sql_field)
    {
      if (my_strcasecmp(sql_field->field_name, dup_field->field_name) == 0)
      {
	my_error(ER_DUP_FIELDNAME,MYF(0),sql_field->field_name);
	DBUG_RETURN(-1);
      }
    }
    it2.rewind();
  }
  /* If fixed row records, we need one bit to check for deleted rows */
  if (!(db_options & HA_OPTION_PACK_RECORD))
    null_fields++;
  pos=(null_fields+7)/8;

  it.rewind();
  while ((sql_field=it++))
  {
    switch (sql_field->sql_type) {
    case FIELD_TYPE_BLOB:
    case FIELD_TYPE_MEDIUM_BLOB:
    case FIELD_TYPE_TINY_BLOB:
    case FIELD_TYPE_LONG_BLOB:
      sql_field->pack_flag=FIELDFLAG_BLOB |
	pack_length_to_packflag(sql_field->pack_length -
				portable_sizeof_char_ptr);
      if (sql_field->flags & BINARY_FLAG)
	sql_field->pack_flag|=FIELDFLAG_BINARY;
      sql_field->length=8;			// Unireg field length
      sql_field->unireg_check=Field::BLOB_FIELD;
      blob_columns++;
      break;
    case FIELD_TYPE_VAR_STRING:
    case FIELD_TYPE_STRING:
      sql_field->pack_flag=0;
      if (sql_field->flags & BINARY_FLAG)
	sql_field->pack_flag|=FIELDFLAG_BINARY;
      break;
    case FIELD_TYPE_ENUM:
      sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length) |
	FIELDFLAG_INTERVAL;
      sql_field->unireg_check=Field::INTERVAL_FIELD;
      break;
    case FIELD_TYPE_SET:
      sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length) |
	FIELDFLAG_BITFIELD;
      sql_field->unireg_check=Field::BIT_FIELD;
      break;
    case FIELD_TYPE_DATE:			// Rest of string types
    case FIELD_TYPE_NEWDATE:
    case FIELD_TYPE_TIME:
    case FIELD_TYPE_DATETIME:
    case FIELD_TYPE_NULL:
      sql_field->pack_flag=f_settype((uint) sql_field->sql_type);
      break;
    case FIELD_TYPE_TIMESTAMP:
      sql_field->unireg_check=Field::TIMESTAMP_FIELD;
      /* fall through */
    default:
      sql_field->pack_flag=(FIELDFLAG_NUMBER |
			    (sql_field->flags & UNSIGNED_FLAG ? 0 :
			     FIELDFLAG_DECIMAL) |
			    (sql_field->flags & ZEROFILL_FLAG ?
			     FIELDFLAG_ZEROFILL : 0) |
			    f_settype((uint) sql_field->sql_type) |
			    (sql_field->decimals << FIELDFLAG_DEC_SHIFT));
      break;
    }
    if (!(sql_field->flags & NOT_NULL_FLAG))
      sql_field->pack_flag|=FIELDFLAG_MAYBE_NULL;
    sql_field->offset= pos;
    if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
      auto_increment++;
    pos+=sql_field->pack_length;
  }
  if (auto_increment > 1)
  {
    my_error(ER_WRONG_AUTO_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  if (auto_increment &&
      (file->table_flags() & HA_NO_AUTO_INCREMENT))
  {
    my_error(ER_TABLE_CANT_HANDLE_AUTO_INCREMENT,MYF(0));
    DBUG_RETURN(-1);
  }

  if (blob_columns && (file->table_flags() & HA_NO_BLOBS))
  {
    my_error(ER_TABLE_CANT_HANDLE_BLOB,MYF(0));
    DBUG_RETURN(-1);
  }

  /* Create keys */

  List_iterator<Key> key_iterator(keys);
  uint key_parts=0,key_count=keys.elements;
  List<Key> keys_in_order;			// Add new keys here
  bool primary_key=0,unique_key=0;
  Key *key;
  uint tmp, key_number;
  tmp=min(file->max_keys(), MAX_KEY);
  if (key_count > tmp)
  {
    my_error(ER_TOO_MANY_KEYS,MYF(0),tmp);
    DBUG_RETURN(-1);
  }

  /* Calculate number of key segements */

  while ((key=key_iterator++))
  {
    tmp=max(file->max_key_parts(),MAX_REF_PARTS);
    if (key->columns.elements > tmp)
    {
      my_error(ER_TOO_MANY_KEY_PARTS,MYF(0),tmp);
      DBUG_RETURN(-1);
    }
    if (key->name() && strlen(key->name()) > NAME_LEN)
    {
      my_error(ER_TOO_LONG_IDENT, MYF(0), key->name());
      DBUG_RETURN(-1);
    }
    key_parts+=key->columns.elements;
  }

  key_info_buffer=key_info=(KEY*) sql_calloc(sizeof(KEY)*key_count);
  key_part_info=(KEY_PART_INFO*) sql_calloc(sizeof(KEY_PART_INFO)*key_parts);
  if (!key_info_buffer || ! key_part_info)
    DBUG_RETURN(-1);				// Out of memory

  key_iterator.rewind();
  key_number=0;
  for (; (key=key_iterator++) ; key_info++, key_number++)
  {
    uint key_length=0;
    key_part_spec *column;

    key_info->flags= (key->type == Key::MULTIPLE) ? 0 :
                     (key->type == Key::FULLTEXT) ? HA_FULLTEXT : HA_NOSAME;
    key_info->key_parts=(uint8) key->columns.elements;
    key_info->key_part=key_part_info;
    key_info->usable_key_parts= key_number;
    key_info->algorithm=key->algorithm;

    /* TODO: Add proper checks if handler supports key_type and algorithm */
    if (key->type == Key::FULLTEXT)
    {
      if (!(file->table_flags() & HA_CAN_FULLTEXT))
      {
        my_error(ER_TABLE_CANT_HANDLE_FULLTEXT, MYF(0));
        DBUG_RETURN(-1);
      }
    }

    List_iterator<key_part_spec> cols(key->columns);
    for (uint column_nr=0 ; (column=cols++) ; column_nr++)
    {
      it.rewind();
      field=0;
      while ((sql_field=it++) &&
	     my_strcasecmp(column->field_name,sql_field->field_name))
	field++;
      if (!sql_field)
      {
	my_printf_error(ER_KEY_COLUMN_DOES_NOT_EXITS,
			ER(ER_KEY_COLUMN_DOES_NOT_EXITS),MYF(0),
			column->field_name);
	DBUG_RETURN(-1);
      }
      if (f_is_blob(sql_field->pack_flag))
      {
	if (!(file->table_flags() & HA_BLOB_KEY))
	{
	  my_printf_error(ER_BLOB_USED_AS_KEY,ER(ER_BLOB_USED_AS_KEY),MYF(0),
			  column->field_name);
	  DBUG_RETURN(-1);
	}
	if (!column->length)
	{
          if (key->type == Key::FULLTEXT)
            column->length=1; /* ft-code ignores it anyway :-) */
          else
          {
            my_printf_error(ER_BLOB_KEY_WITHOUT_LENGTH,
                            ER(ER_BLOB_KEY_WITHOUT_LENGTH),MYF(0),
                            column->field_name);
            DBUG_RETURN(-1);
          }
	}
      }
      if (!(sql_field->flags & NOT_NULL_FLAG))
      {
	if (key->type == Key::PRIMARY)
	{
	  my_error(ER_PRIMARY_CANT_HAVE_NULL, MYF(0));
	  DBUG_RETURN(-1);
	}
	if (!(file->table_flags() & HA_NULL_KEY))
	{
	  my_printf_error(ER_NULL_COLUMN_IN_INDEX,ER(ER_NULL_COLUMN_IN_INDEX),
			  MYF(0),column->field_name);
	  DBUG_RETURN(-1);
	}
	key_info->flags|= HA_NULL_PART_KEY;
      }
      if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
      {
	if (column_nr == 0 || (file->table_flags() & HA_AUTO_PART_KEY))
	  auto_increment--;			// Field is used
      }
      key_part_info->fieldnr= field;
      key_part_info->offset=  (uint16) sql_field->offset;
      key_part_info->key_type=sql_field->pack_flag;
      uint length=sql_field->pack_length;
      if (column->length)
      {
	if (f_is_blob(sql_field->pack_flag))
	{
	  if ((length=column->length) > file->max_key_length() ||
	      length > file->max_key_part_length())
	  {
	    my_error(ER_WRONG_SUB_KEY,MYF(0));
	    DBUG_RETURN(-1);
	  }
	}
	else if (column->length > length ||
		 ((f_is_packed(sql_field->pack_flag) || 
		   ((file->table_flags() & HA_NO_PREFIX_CHAR_KEYS) &&
		    (key_info->flags & HA_NOSAME))) &&
		  column->length != length))
	{
	  my_error(ER_WRONG_SUB_KEY,MYF(0));
	  DBUG_RETURN(-1);
	}
	if (!(file->table_flags() & HA_NO_PREFIX_CHAR_KEYS))
	  length=column->length;
      }
      else if (length == 0)
      {
	my_printf_error(ER_WRONG_KEY_COLUMN, ER(ER_WRONG_KEY_COLUMN), MYF(0),
			column->field_name);
	  DBUG_RETURN(-1);
      }
      key_part_info->length=(uint8) length;
      /* Use packed keys for long strings on the first column */
      if (!(db_options & HA_OPTION_NO_PACK_KEYS) &&
	  (length >= KEY_DEFAULT_PACK_LENGTH &&
	   (sql_field->sql_type == FIELD_TYPE_STRING ||
	    sql_field->sql_type == FIELD_TYPE_VAR_STRING ||
	    sql_field->pack_flag & FIELDFLAG_BLOB)))
      {
	if (column_nr == 0 && (sql_field->pack_flag & FIELDFLAG_BLOB))
	  key_info->flags|= HA_BINARY_PACK_KEY;
	else
	  key_info->flags|= HA_PACK_KEY;
      }
      key_length+=length;
      key_part_info++;

      /* Create the key name based on the first column (if not given) */
      if (column_nr == 0)
      {
	if (key->type == Key::PRIMARY)
	{
	  if (primary_key)
	  {
	    my_error(ER_MULTIPLE_PRI_KEY,MYF(0));
	    DBUG_RETURN(-1);
	  }
	  key_name=primary_key_name;
	  primary_key=1;
	}
	else if (!(key_name = key->name()))
	  key_name=make_unique_key_name(sql_field->field_name,
					key_info_buffer,key_info);
	if (check_if_keyname_exists(key_name,key_info_buffer,key_info))
	{
	  my_error(ER_DUP_KEYNAME,MYF(0),key_name);
	  DBUG_RETURN(-1);
	}
	key_info->name=(char*) key_name;
      }
    }
    if (!(key_info->flags & HA_NULL_PART_KEY))
      unique_key=1;
    key_info->key_length=(uint16) key_length;
    uint max_key_length= max(file->max_key_length(), MAX_KEY_LENGTH);
    if (key_length > max_key_length && key->type != Key::FULLTEXT)
    {
      my_error(ER_TOO_LONG_KEY,MYF(0),max_key_length);
      DBUG_RETURN(-1);
    }
  }
  if (!unique_key && !primary_key &&
      (file->table_flags() & HA_REQUIRE_PRIMARY_KEY))
  {
    my_error(ER_REQUIRES_PRIMARY_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  if (auto_increment > 0)
  {
    my_error(ER_WRONG_AUTO_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  /* Sort keys in optimized order */
  qsort((gptr) key_info_buffer, key_count, sizeof(KEY), (qsort_cmp) sort_keys);

      /* Check if table exists */
  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
  {
    sprintf(path,"%s%s%lx_%lx_%x%s",mysql_tmpdir,tmp_file_prefix,
	    current_pid, thd->thread_id, thd->tmp_table++,reg_ext);
    create_info->table_options|=HA_CREATE_DELAY_KEY_WRITE;
  }
  else
    (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table_name,reg_ext);
  unpack_filename(path,path);
  /* Check if table already exists */
  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE)
      && find_temporary_table(thd,db,table_name))
  {
    if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
      DBUG_RETURN(0);
    my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
    DBUG_RETURN(-1);
  }
  VOID(pthread_mutex_lock(&LOCK_open));
  if (!tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
  {
    if (!access(path,F_OK))
    {
      VOID(pthread_mutex_unlock(&LOCK_open));
      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
	DBUG_RETURN(0);
      my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
      DBUG_RETURN(-1);
    }
  }

  thd->proc_info="creating table";

  create_info->create_statement = thd->query;
  create_info->table_options=db_options;
  if (rea_create_table(path, create_info, fields, key_count,
		       key_info_buffer))
  {
    /* my_error(ER_CANT_CREATE_TABLE,MYF(0),table_name,my_errno); */
    goto end;
  }
  if (!tmp_table && !no_log)
  {
    // Must be written before unlock
    mysql_update_log.write(thd,thd->query, thd->query_length);
    if (mysql_bin_log.is_open())
    {
      Query_log_event qinfo(thd, thd->query);
      mysql_bin_log.write(&qinfo);
    }
  }
  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
  {
    /* Open table and put in temporary table list */
    if (!(open_temporary_table(thd, path, db, table_name, 1)))
    {
      (void) rm_temporary_table(create_info->db_type, path);
      goto end;
    }
  }
  error=0;
end:
  VOID(pthread_mutex_unlock(&LOCK_open));
  thd->proc_info="After create";
  DBUG_RETURN(error);
}

/*
** Give the key name after the first field with an optional '_#' after
**/

static bool
check_if_keyname_exists(const char *name, KEY *start, KEY *end)
{
  for (KEY *key=start ; key != end ; key++)
    if (!my_strcasecmp(name,key->name))
      return 1;
  return 0;
}


static char *
make_unique_key_name(const char *field_name,KEY *start,KEY *end)
{
  char buff[MAX_FIELD_NAME],*buff_end;

  if (!check_if_keyname_exists(field_name,start,end))
    return (char*) field_name;			// Use fieldname
  buff_end=strmake(buff,field_name,MAX_FIELD_NAME-4);
  for (uint i=2 ; ; i++)
  {
    sprintf(buff_end,"_%d",i);
    if (!check_if_keyname_exists(buff,start,end))
      return sql_strdup(buff);
  }
}

/****************************************************************************
** Create table from a list of fields and items
****************************************************************************/

TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
			       const char *db, const char *name,
			       List<create_field> *extra_fields,
			       List<Key> *keys,
			       List<Item> *items,
			       MYSQL_LOCK **lock)
{
  TABLE tmp_table;		// Used during 'create_field()'
  TABLE *table;
  tmp_table.table_name=0;
  DBUG_ENTER("create_table_from_items");

  /* Add selected items to field list */
  List_iterator_fast<Item> it(*items);
  Item *item;
  Field *tmp_field;
  tmp_table.db_create_options=0;
  tmp_table.null_row=tmp_table.maybe_null=0;
  tmp_table.blob_ptr_size=portable_sizeof_char_ptr;
  tmp_table.db_low_byte_first= test(create_info->db_type == DB_TYPE_MYISAM ||
				    create_info->db_type == DB_TYPE_HEAP);

  while ((item=it++))
  {
    create_field *cr_field;
    if (strlen(item->name) > NAME_LEN ||
	check_column_name(item->name))
    {
      my_error(ER_WRONG_COLUMN_NAME,MYF(0),item->name);
      DBUG_RETURN(0);
    }
    Field *field;
    if (item->type() == Item::FUNC_ITEM)
      field=item->tmp_table_field(&tmp_table);
    else
      field=create_tmp_field(thd, &tmp_table, item, item->type(),
				  (Item_result_field***) 0, &tmp_field,0,0);
    if (!field ||
	!(cr_field=new create_field(field,(item->type() == Item::FIELD_ITEM ?
					   ((Item_field *)item)->field :
					   (Field*) 0))))
      DBUG_RETURN(0);
    extra_fields->push_back(cr_field);
  }
  /* create and lock table */
  /* QQ: This should be done atomic ! */
  if (mysql_create_table(thd,db,name,create_info,*extra_fields,
			 *keys,0,1)) // no logging
    DBUG_RETURN(0);
  if (!(table=open_table(thd,db,name,name,(bool*) 0)))
  {
    quick_rm_table(create_info->db_type,db,name);
    DBUG_RETURN(0);
  }
  table->reginfo.lock_type=TL_WRITE;
  if (!((*lock)=mysql_lock_tables(thd,&table,1)))
  {
    hash_delete(&open_cache,(byte*) table);
    quick_rm_table(create_info->db_type,db,name);
    DBUG_RETURN(0);
  }
  table->file->extra(HA_EXTRA_WRITE_CACHE);
  DBUG_RETURN(table);
}


/****************************************************************************
** Alter a table definition
****************************************************************************/

bool
mysql_rename_table(enum db_type base,
		   const char *old_db,
		   const char * old_name,
		   const char *new_db,
		   const char * new_name)
{
  char from[FN_REFLEN],to[FN_REFLEN];
  handler *file=get_new_handler((TABLE*) 0, base);
  int error=0;
  DBUG_ENTER("mysql_rename_table");
  (void) sprintf(from,"%s/%s/%s",mysql_data_home,old_db,old_name);
  (void) sprintf(to,"%s/%s/%s",mysql_data_home,new_db,new_name);
  fn_format(from,from,"","",4);
  fn_format(to,to,    "","",4);
  if (!(error=file->rename_table((const char*) from,(const char *) to)))
  {
    if (rename_file_ext(from,to,reg_ext))
    {
      error=my_errno;
      /* Restore old file name */
      file->rename_table((const char*) to,(const char *) from);
    }
  }
  delete file;
  if (error)
    my_error(ER_ERROR_ON_RENAME, MYF(0), from, to, error);
  DBUG_RETURN(error != 0);
}

/*
  close table in this thread and force close + reopen in other threads
  This assumes that the calling thread has lock on LOCK_open
  Win32 clients must also have a WRITE LOCK on the table !
*/

bool close_cached_table(THD *thd,TABLE *table)
{
  bool result=0;
  DBUG_ENTER("close_cached_table");
  if (table)
  {
    DBUG_PRINT("enter",("table: %s", table->table_name));
    VOID(table->file->extra(HA_EXTRA_FORCE_REOPEN)); // Close all data files
    /* Mark all tables that are in use as 'old' */
    mysql_lock_abort(thd,table);		 // end threads waiting on lock

#if defined(USING_TRANSACTIONS) || defined( __WIN__) || defined( __EMX__) || !defined(OS2)
    /* Wait until all there are no other threads that has this table open */
    while (remove_table_from_cache(thd,table->table_cache_key,
				   table->table_name))
    {
      dropping_tables++;
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
      dropping_tables--;
    }
#else
    (void) remove_table_from_cache(thd,table->table_cache_key,
				   table->table_name);
#endif
    /* When lock on LOCK_open is freed other threads can continue */
    pthread_cond_broadcast(&COND_refresh);

    /* Close lock if this is not got with LOCK TABLES */
    if (thd->lock)
    {
      mysql_unlock_tables(thd, thd->lock); thd->lock=0;	// Start locked threads
    }
    /* Close all copies of 'table'.  This also frees all LOCK TABLES lock */
    thd->open_tables=unlink_open_table(thd,thd->open_tables,table);
  }
  DBUG_RETURN(result);
}

static int send_check_errmsg(THD* thd, TABLE_LIST* table,
			     const char* operator_name, const char* errmsg)

{

  String* packet = &thd->packet;
  packet->length(0);
  net_store_data(packet, table->name);
  net_store_data(packet, (char*)operator_name);
  net_store_data(packet, "error");
  net_store_data(packet, errmsg);
  thd->net.last_error[0]=0;
  if (my_net_write(&thd->net, (char*) thd->packet.ptr(),
		   packet->length()))
    return -1;
  return 1;
}

static int prepare_for_restore(THD* thd, TABLE_LIST* table,
    HA_CHECK_OPT *check_opt)
{
  DBUG_ENTER("prepare_for_restore");

  if (table->table) // do not overwrite existing tables on restore
  {
    DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				  "table exists, will not overwrite on restore"
				  ));
  }
  else
  {
    char* backup_dir = thd->lex.backup_dir;
    char src_path[FN_REFLEN], dst_path[FN_REFLEN];
    char* table_name = table->name;
    char* db = thd->db ? thd->db : table->db;

    if (fn_format_relative_to_data_home(src_path, table_name, backup_dir,
					reg_ext))
      DBUG_RETURN(-1); // protect buffer overflow

    sprintf(dst_path, "%s/%s/%s", mysql_real_data_home, db, table_name);

    if (lock_and_wait_for_table_name(thd,table))
      DBUG_RETURN(-1);

    if (my_copy(src_path,
		fn_format(dst_path, dst_path,"", reg_ext, 4),
		MYF(MY_WME)))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				    "Failed copying .frm file"));
    }
    if (mysql_truncate(thd, table, 1))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				    "Failed generating table from .frm file"));
    }
  }

  // now we should be able to open the partially restored table
  // to finish the restore in the handler later on
  if (!(table->table = reopen_name_locked_table(thd, table)))
     unlock_table_name(thd, table);
  DBUG_RETURN(0);
}

static int prepare_for_repair(THD* thd, TABLE_LIST* table,
    HA_CHECK_OPT *check_opt)
{
  DBUG_ENTER("prepare_for_repair");

  if (!(check_opt->sql_flags & TT_USEFRM))
  {
    DBUG_RETURN(0);
  }
  else
  {

    char from[FN_REFLEN],to[FN_REFLEN];
    char* db = thd->db ? thd->db : table->db;

    sprintf(from, "%s/%s/%s", mysql_real_data_home, db, table->name);
    fn_format(from, from, "", MI_NAME_DEXT, 4);
    sprintf(to,"%s-%lx_%lx", from, current_pid, thd->thread_id);


    my_rename(to, from, MYF(MY_WME));

    if (lock_and_wait_for_table_name(thd,table))
      DBUG_RETURN(-1);

    if (my_rename(from, to, MYF(MY_WME)))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "repair",
				    "Failed renaming .MYD file"));
    }
    if (mysql_truncate(thd, table, 1))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "repair",
				    "Failed generating table from .frm file"));
    }
    if (my_rename(to, from, MYF(MY_WME)))<commit_msg>"repair ... use_frm" crash fixed<commit_after>/* 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 */


/* drop and alter of tables */

#include "mysql_priv.h"
#include <hash.h>
#include <myisam.h>

#ifdef __WIN__
#include <io.h>
#endif

extern HASH open_cache;
static 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,
				    enum enum_duplicates handle_duplicates,
                                    ORDER *order,
				    ha_rows *copied,ha_rows *deleted);

/*****************************************************************************
** Remove all possbile tables and give a compact errormessage for all
** wrong tables.
** This will wait for all users to free the table before dropping it
*****************************************************************************/

int mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists)
{
  int error;
  DBUG_ENTER("mysql_rm_table");

  /* mark for close and remove all cached entries */

  thd->mysys_var->current_mutex= &LOCK_open;
  thd->mysys_var->current_cond= &COND_refresh;
  VOID(pthread_mutex_lock(&LOCK_open));

  if (global_read_lock)
  {
    if (thd->global_read_lock)
    {
      my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE,MYF(0),
	       tables->real_name);
      error = 1;
      goto err;
    }
    while (global_read_lock && ! thd->killed)
    {
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
    }

  }
  error=mysql_rm_table_part2(thd,tables,if_exists,0);

 err:  
  VOID(pthread_cond_broadcast(&COND_refresh)); // Signal to refresh
  pthread_mutex_unlock(&LOCK_open);

  pthread_mutex_lock(&thd->mysys_var->mutex);
  thd->mysys_var->current_mutex= 0;
  thd->mysys_var->current_cond= 0;
  pthread_mutex_unlock(&thd->mysys_var->mutex);

  if (error)
    DBUG_RETURN(-1);
  send_ok(&thd->net);
  DBUG_RETURN(0);
}


int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
			 bool dont_log_query)
{
  TABLE_LIST *table;
  char	path[FN_REFLEN];
  String wrong_tables;
  db_type table_type;
  int error;
  bool some_tables_deleted=0;
  DBUG_ENTER("mysql_rm_table_part2");

  for (table=tables ; table ; table=table->next)
  {
    char *db=table->db ? table->db : thd->db;
    if (!close_temporary_table(thd, db, table->real_name))
    {
      some_tables_deleted=1;			// Log query
      continue;					// removed temporary table
    }

    abort_locked_tables(thd,db,table->real_name);
    while (remove_table_from_cache(thd,db,table->real_name) && !thd->killed)
    {
      dropping_tables++;
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
      dropping_tables--;
    }
    drop_locked_tables(thd,db,table->real_name);
    if (thd->killed)
    {
      VOID(pthread_cond_broadcast(&COND_refresh)); // Signal to refresh
      VOID(pthread_mutex_unlock(&LOCK_open));
      pthread_mutex_lock(&thd->mysys_var->mutex);
      thd->mysys_var->current_mutex= 0;
      thd->mysys_var->current_cond= 0;
      pthread_mutex_unlock(&thd->mysys_var->mutex);
      DBUG_RETURN(-1);
    }
    /* remove form file and isam files */
    (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table->real_name,
		   reg_ext);
    (void) unpack_filename(path,path);
    error=0;

    table_type=get_table_type(path);

    if (access(path,F_OK))
    {
      if (!if_exists)
	error=1;
    }
    else
    {
      char *end;
      *(end=fn_ext(path))=0;			// Remove extension
      error=ha_delete_table(table_type, path);
      if (error == ENOENT && if_exists)
	error = 0;
      if (!error || error == ENOENT)
      {
	/* Delete the table definition file */
	strmov(end,reg_ext);
	if (!(error=my_delete(path,MYF(MY_WME))))
	  some_tables_deleted=1;
      }
    }
    if (error)
    {
      if (wrong_tables.length())
	wrong_tables.append(',');
      wrong_tables.append(String(table->real_name));
    }
  }
  if (some_tables_deleted)
  {
    query_cache_invalidate3(thd, tables, 0);
    if (!dont_log_query)
    {
      mysql_update_log.write(thd, thd->query,thd->query_length);
      if (mysql_bin_log.is_open())
      {
	Query_log_event qinfo(thd, thd->query);
	mysql_bin_log.write(&qinfo);
      }
    }
  }

  error = 0;
  if (wrong_tables.length())
  {
    my_error(ER_BAD_TABLE_ERROR,MYF(0),wrong_tables.c_ptr());
    error=1;
  }
  DBUG_RETURN(error);
}


int quick_rm_table(enum db_type base,const char *db,
		   const char *table_name)
{
  char path[FN_REFLEN];
  int error=0;
  (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table_name,reg_ext);
  unpack_filename(path,path);
  if (my_delete(path,MYF(0)))
    error=1; /* purecov: inspected */
  sprintf(path,"%s/%s/%s",mysql_data_home,db,table_name);
  unpack_filename(path,path);
  return ha_delete_table(base,path) || error;
}

/*
  Sort keys in the following order:
  - PRIMARY KEY
  - UNIQUE keyws where all column are NOT NULL
  - 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)
{
  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 (a->name == primary_key_name)
      return -1;
    if (b->name == primary_key_name)
      return 1;
  }
  else if (b->flags & HA_NOSAME)
    return 1;					// Prefer b

  if ((a->flags ^ b->flags) & HA_FULLTEXT)
  {
    return (a->flags & HA_FULLTEXT) ? 1 : -1;
  }
  /*
    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);
}


/*****************************************************************************
 * Create a table.
 * If one creates a temporary table, this is automaticly opened
 ****************************************************************************/

int mysql_create_table(THD *thd,const char *db, const char *table_name,
		       HA_CREATE_INFO *create_info,
		       List<create_field> &fields,
		       List<Key> &keys,bool tmp_table,bool no_log)
{
  char		path[FN_REFLEN];
  const char	*key_name;
  create_field	*sql_field,*dup_field;
  int		error= -1;
  uint		db_options,field,null_fields,blob_columns;
  ulong		pos;
  KEY	*key_info,*key_info_buffer;
  KEY_PART_INFO *key_part_info;
  int		auto_increment=0;
  handler	*file;
  DBUG_ENTER("mysql_create_table");

  /*
  ** Check for duplicate fields and check type of table to create
  */

  if (!fields.elements)
  {
    my_error(ER_TABLE_MUST_HAVE_COLUMNS,MYF(0));
    DBUG_RETURN(-1);
  }
  List_iterator<create_field> it(fields),it2(fields);
  null_fields=blob_columns=0;
  db_options=create_info->table_options;
  if (create_info->row_type == ROW_TYPE_DYNAMIC)
    db_options|=HA_OPTION_PACK_RECORD;
  file=get_new_handler((TABLE*) 0, create_info->db_type);

  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
      (file->table_flags() & HA_NO_TEMP_TABLES))
  {
    my_error(ER_ILLEGAL_HA,MYF(0),table_name);
    DBUG_RETURN(-1);
  }

  /* Don't pack keys in old tables if the user has requested this */

  while ((sql_field=it++))
  {
    if ((sql_field->flags & BLOB_FLAG) ||
	sql_field->sql_type == FIELD_TYPE_VAR_STRING &&
	create_info->row_type != ROW_TYPE_FIXED)
    {
      db_options|=HA_OPTION_PACK_RECORD;
    }
    if (!(sql_field->flags & NOT_NULL_FLAG))
      null_fields++;
    while ((dup_field=it2++) != sql_field)
    {
      if (my_strcasecmp(sql_field->field_name, dup_field->field_name) == 0)
      {
	my_error(ER_DUP_FIELDNAME,MYF(0),sql_field->field_name);
	DBUG_RETURN(-1);
      }
    }
    it2.rewind();
  }
  /* If fixed row records, we need one bit to check for deleted rows */
  if (!(db_options & HA_OPTION_PACK_RECORD))
    null_fields++;
  pos=(null_fields+7)/8;

  it.rewind();
  while ((sql_field=it++))
  {
    switch (sql_field->sql_type) {
    case FIELD_TYPE_BLOB:
    case FIELD_TYPE_MEDIUM_BLOB:
    case FIELD_TYPE_TINY_BLOB:
    case FIELD_TYPE_LONG_BLOB:
      sql_field->pack_flag=FIELDFLAG_BLOB |
	pack_length_to_packflag(sql_field->pack_length -
				portable_sizeof_char_ptr);
      if (sql_field->flags & BINARY_FLAG)
	sql_field->pack_flag|=FIELDFLAG_BINARY;
      sql_field->length=8;			// Unireg field length
      sql_field->unireg_check=Field::BLOB_FIELD;
      blob_columns++;
      break;
    case FIELD_TYPE_VAR_STRING:
    case FIELD_TYPE_STRING:
      sql_field->pack_flag=0;
      if (sql_field->flags & BINARY_FLAG)
	sql_field->pack_flag|=FIELDFLAG_BINARY;
      break;
    case FIELD_TYPE_ENUM:
      sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length) |
	FIELDFLAG_INTERVAL;
      sql_field->unireg_check=Field::INTERVAL_FIELD;
      break;
    case FIELD_TYPE_SET:
      sql_field->pack_flag=pack_length_to_packflag(sql_field->pack_length) |
	FIELDFLAG_BITFIELD;
      sql_field->unireg_check=Field::BIT_FIELD;
      break;
    case FIELD_TYPE_DATE:			// Rest of string types
    case FIELD_TYPE_NEWDATE:
    case FIELD_TYPE_TIME:
    case FIELD_TYPE_DATETIME:
    case FIELD_TYPE_NULL:
      sql_field->pack_flag=f_settype((uint) sql_field->sql_type);
      break;
    case FIELD_TYPE_TIMESTAMP:
      sql_field->unireg_check=Field::TIMESTAMP_FIELD;
      /* fall through */
    default:
      sql_field->pack_flag=(FIELDFLAG_NUMBER |
			    (sql_field->flags & UNSIGNED_FLAG ? 0 :
			     FIELDFLAG_DECIMAL) |
			    (sql_field->flags & ZEROFILL_FLAG ?
			     FIELDFLAG_ZEROFILL : 0) |
			    f_settype((uint) sql_field->sql_type) |
			    (sql_field->decimals << FIELDFLAG_DEC_SHIFT));
      break;
    }
    if (!(sql_field->flags & NOT_NULL_FLAG))
      sql_field->pack_flag|=FIELDFLAG_MAYBE_NULL;
    sql_field->offset= pos;
    if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
      auto_increment++;
    pos+=sql_field->pack_length;
  }
  if (auto_increment > 1)
  {
    my_error(ER_WRONG_AUTO_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  if (auto_increment &&
      (file->table_flags() & HA_NO_AUTO_INCREMENT))
  {
    my_error(ER_TABLE_CANT_HANDLE_AUTO_INCREMENT,MYF(0));
    DBUG_RETURN(-1);
  }

  if (blob_columns && (file->table_flags() & HA_NO_BLOBS))
  {
    my_error(ER_TABLE_CANT_HANDLE_BLOB,MYF(0));
    DBUG_RETURN(-1);
  }

  /* Create keys */

  List_iterator<Key> key_iterator(keys);
  uint key_parts=0,key_count=keys.elements;
  List<Key> keys_in_order;			// Add new keys here
  bool primary_key=0,unique_key=0;
  Key *key;
  uint tmp, key_number;
  tmp=min(file->max_keys(), MAX_KEY);
  if (key_count > tmp)
  {
    my_error(ER_TOO_MANY_KEYS,MYF(0),tmp);
    DBUG_RETURN(-1);
  }

  /* Calculate number of key segements */

  while ((key=key_iterator++))
  {
    tmp=max(file->max_key_parts(),MAX_REF_PARTS);
    if (key->columns.elements > tmp)
    {
      my_error(ER_TOO_MANY_KEY_PARTS,MYF(0),tmp);
      DBUG_RETURN(-1);
    }
    if (key->name() && strlen(key->name()) > NAME_LEN)
    {
      my_error(ER_TOO_LONG_IDENT, MYF(0), key->name());
      DBUG_RETURN(-1);
    }
    key_parts+=key->columns.elements;
  }

  key_info_buffer=key_info=(KEY*) sql_calloc(sizeof(KEY)*key_count);
  key_part_info=(KEY_PART_INFO*) sql_calloc(sizeof(KEY_PART_INFO)*key_parts);
  if (!key_info_buffer || ! key_part_info)
    DBUG_RETURN(-1);				// Out of memory

  key_iterator.rewind();
  key_number=0;
  for (; (key=key_iterator++) ; key_info++, key_number++)
  {
    uint key_length=0;
    key_part_spec *column;

    key_info->flags= (key->type == Key::MULTIPLE) ? 0 :
                     (key->type == Key::FULLTEXT) ? HA_FULLTEXT : HA_NOSAME;
    key_info->key_parts=(uint8) key->columns.elements;
    key_info->key_part=key_part_info;
    key_info->usable_key_parts= key_number;
    key_info->algorithm=key->algorithm;

    /* TODO: Add proper checks if handler supports key_type and algorithm */
    if (key->type == Key::FULLTEXT)
    {
      if (!(file->table_flags() & HA_CAN_FULLTEXT))
      {
        my_error(ER_TABLE_CANT_HANDLE_FULLTEXT, MYF(0));
        DBUG_RETURN(-1);
      }
    }

    List_iterator<key_part_spec> cols(key->columns);
    for (uint column_nr=0 ; (column=cols++) ; column_nr++)
    {
      it.rewind();
      field=0;
      while ((sql_field=it++) &&
	     my_strcasecmp(column->field_name,sql_field->field_name))
	field++;
      if (!sql_field)
      {
	my_printf_error(ER_KEY_COLUMN_DOES_NOT_EXITS,
			ER(ER_KEY_COLUMN_DOES_NOT_EXITS),MYF(0),
			column->field_name);
	DBUG_RETURN(-1);
      }
      if (f_is_blob(sql_field->pack_flag))
      {
	if (!(file->table_flags() & HA_BLOB_KEY))
	{
	  my_printf_error(ER_BLOB_USED_AS_KEY,ER(ER_BLOB_USED_AS_KEY),MYF(0),
			  column->field_name);
	  DBUG_RETURN(-1);
	}
	if (!column->length)
	{
          if (key->type == Key::FULLTEXT)
            column->length=1; /* ft-code ignores it anyway :-) */
          else
          {
            my_printf_error(ER_BLOB_KEY_WITHOUT_LENGTH,
                            ER(ER_BLOB_KEY_WITHOUT_LENGTH),MYF(0),
                            column->field_name);
            DBUG_RETURN(-1);
          }
	}
      }
      if (!(sql_field->flags & NOT_NULL_FLAG))
      {
	if (key->type == Key::PRIMARY)
	{
	  my_error(ER_PRIMARY_CANT_HAVE_NULL, MYF(0));
	  DBUG_RETURN(-1);
	}
	if (!(file->table_flags() & HA_NULL_KEY))
	{
	  my_printf_error(ER_NULL_COLUMN_IN_INDEX,ER(ER_NULL_COLUMN_IN_INDEX),
			  MYF(0),column->field_name);
	  DBUG_RETURN(-1);
	}
	key_info->flags|= HA_NULL_PART_KEY;
      }
      if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
      {
	if (column_nr == 0 || (file->table_flags() & HA_AUTO_PART_KEY))
	  auto_increment--;			// Field is used
      }
      key_part_info->fieldnr= field;
      key_part_info->offset=  (uint16) sql_field->offset;
      key_part_info->key_type=sql_field->pack_flag;
      uint length=sql_field->pack_length;
      if (column->length)
      {
	if (f_is_blob(sql_field->pack_flag))
	{
	  if ((length=column->length) > file->max_key_length() ||
	      length > file->max_key_part_length())
	  {
	    my_error(ER_WRONG_SUB_KEY,MYF(0));
	    DBUG_RETURN(-1);
	  }
	}
	else if (column->length > length ||
		 ((f_is_packed(sql_field->pack_flag) || 
		   ((file->table_flags() & HA_NO_PREFIX_CHAR_KEYS) &&
		    (key_info->flags & HA_NOSAME))) &&
		  column->length != length))
	{
	  my_error(ER_WRONG_SUB_KEY,MYF(0));
	  DBUG_RETURN(-1);
	}
	if (!(file->table_flags() & HA_NO_PREFIX_CHAR_KEYS))
	  length=column->length;
      }
      else if (length == 0)
      {
	my_printf_error(ER_WRONG_KEY_COLUMN, ER(ER_WRONG_KEY_COLUMN), MYF(0),
			column->field_name);
	  DBUG_RETURN(-1);
      }
      key_part_info->length=(uint8) length;
      /* Use packed keys for long strings on the first column */
      if (!(db_options & HA_OPTION_NO_PACK_KEYS) &&
	  (length >= KEY_DEFAULT_PACK_LENGTH &&
	   (sql_field->sql_type == FIELD_TYPE_STRING ||
	    sql_field->sql_type == FIELD_TYPE_VAR_STRING ||
	    sql_field->pack_flag & FIELDFLAG_BLOB)))
      {
	if (column_nr == 0 && (sql_field->pack_flag & FIELDFLAG_BLOB))
	  key_info->flags|= HA_BINARY_PACK_KEY;
	else
	  key_info->flags|= HA_PACK_KEY;
      }
      key_length+=length;
      key_part_info++;

      /* Create the key name based on the first column (if not given) */
      if (column_nr == 0)
      {
	if (key->type == Key::PRIMARY)
	{
	  if (primary_key)
	  {
	    my_error(ER_MULTIPLE_PRI_KEY,MYF(0));
	    DBUG_RETURN(-1);
	  }
	  key_name=primary_key_name;
	  primary_key=1;
	}
	else if (!(key_name = key->name()))
	  key_name=make_unique_key_name(sql_field->field_name,
					key_info_buffer,key_info);
	if (check_if_keyname_exists(key_name,key_info_buffer,key_info))
	{
	  my_error(ER_DUP_KEYNAME,MYF(0),key_name);
	  DBUG_RETURN(-1);
	}
	key_info->name=(char*) key_name;
      }
    }
    if (!(key_info->flags & HA_NULL_PART_KEY))
      unique_key=1;
    key_info->key_length=(uint16) key_length;
    uint max_key_length= max(file->max_key_length(), MAX_KEY_LENGTH);
    if (key_length > max_key_length && key->type != Key::FULLTEXT)
    {
      my_error(ER_TOO_LONG_KEY,MYF(0),max_key_length);
      DBUG_RETURN(-1);
    }
  }
  if (!unique_key && !primary_key &&
      (file->table_flags() & HA_REQUIRE_PRIMARY_KEY))
  {
    my_error(ER_REQUIRES_PRIMARY_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  if (auto_increment > 0)
  {
    my_error(ER_WRONG_AUTO_KEY,MYF(0));
    DBUG_RETURN(-1);
  }
  /* Sort keys in optimized order */
  qsort((gptr) key_info_buffer, key_count, sizeof(KEY), (qsort_cmp) sort_keys);

      /* Check if table exists */
  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
  {
    sprintf(path,"%s%s%lx_%lx_%x%s",mysql_tmpdir,tmp_file_prefix,
	    current_pid, thd->thread_id, thd->tmp_table++,reg_ext);
    create_info->table_options|=HA_CREATE_DELAY_KEY_WRITE;
  }
  else
    (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,db,table_name,reg_ext);
  unpack_filename(path,path);
  /* Check if table already exists */
  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE)
      && find_temporary_table(thd,db,table_name))
  {
    if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
      DBUG_RETURN(0);
    my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
    DBUG_RETURN(-1);
  }
  VOID(pthread_mutex_lock(&LOCK_open));
  if (!tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
  {
    if (!access(path,F_OK))
    {
      VOID(pthread_mutex_unlock(&LOCK_open));
      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
	DBUG_RETURN(0);
      my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
      DBUG_RETURN(-1);
    }
  }

  thd->proc_info="creating table";

  create_info->create_statement = thd->query;
  create_info->table_options=db_options;
  if (rea_create_table(path, create_info, fields, key_count,
		       key_info_buffer))
  {
    /* my_error(ER_CANT_CREATE_TABLE,MYF(0),table_name,my_errno); */
    goto end;
  }
  if (!tmp_table && !no_log)
  {
    // Must be written before unlock
    mysql_update_log.write(thd,thd->query, thd->query_length);
    if (mysql_bin_log.is_open())
    {
      Query_log_event qinfo(thd, thd->query);
      mysql_bin_log.write(&qinfo);
    }
  }
  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
  {
    /* Open table and put in temporary table list */
    if (!(open_temporary_table(thd, path, db, table_name, 1)))
    {
      (void) rm_temporary_table(create_info->db_type, path);
      goto end;
    }
  }
  error=0;
end:
  VOID(pthread_mutex_unlock(&LOCK_open));
  thd->proc_info="After create";
  DBUG_RETURN(error);
}

/*
** Give the key name after the first field with an optional '_#' after
**/

static bool
check_if_keyname_exists(const char *name, KEY *start, KEY *end)
{
  for (KEY *key=start ; key != end ; key++)
    if (!my_strcasecmp(name,key->name))
      return 1;
  return 0;
}


static char *
make_unique_key_name(const char *field_name,KEY *start,KEY *end)
{
  char buff[MAX_FIELD_NAME],*buff_end;

  if (!check_if_keyname_exists(field_name,start,end))
    return (char*) field_name;			// Use fieldname
  buff_end=strmake(buff,field_name,MAX_FIELD_NAME-4);
  for (uint i=2 ; ; i++)
  {
    sprintf(buff_end,"_%d",i);
    if (!check_if_keyname_exists(buff,start,end))
      return sql_strdup(buff);
  }
}

/****************************************************************************
** Create table from a list of fields and items
****************************************************************************/

TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
			       const char *db, const char *name,
			       List<create_field> *extra_fields,
			       List<Key> *keys,
			       List<Item> *items,
			       MYSQL_LOCK **lock)
{
  TABLE tmp_table;		// Used during 'create_field()'
  TABLE *table;
  tmp_table.table_name=0;
  DBUG_ENTER("create_table_from_items");

  /* Add selected items to field list */
  List_iterator_fast<Item> it(*items);
  Item *item;
  Field *tmp_field;
  tmp_table.db_create_options=0;
  tmp_table.null_row=tmp_table.maybe_null=0;
  tmp_table.blob_ptr_size=portable_sizeof_char_ptr;
  tmp_table.db_low_byte_first= test(create_info->db_type == DB_TYPE_MYISAM ||
				    create_info->db_type == DB_TYPE_HEAP);

  while ((item=it++))
  {
    create_field *cr_field;
    if (strlen(item->name) > NAME_LEN ||
	check_column_name(item->name))
    {
      my_error(ER_WRONG_COLUMN_NAME,MYF(0),item->name);
      DBUG_RETURN(0);
    }
    Field *field;
    if (item->type() == Item::FUNC_ITEM)
      field=item->tmp_table_field(&tmp_table);
    else
      field=create_tmp_field(thd, &tmp_table, item, item->type(),
				  (Item_result_field***) 0, &tmp_field,0,0);
    if (!field ||
	!(cr_field=new create_field(field,(item->type() == Item::FIELD_ITEM ?
					   ((Item_field *)item)->field :
					   (Field*) 0))))
      DBUG_RETURN(0);
    extra_fields->push_back(cr_field);
  }
  /* create and lock table */
  /* QQ: This should be done atomic ! */
  if (mysql_create_table(thd,db,name,create_info,*extra_fields,
			 *keys,0,1)) // no logging
    DBUG_RETURN(0);
  if (!(table=open_table(thd,db,name,name,(bool*) 0)))
  {
    quick_rm_table(create_info->db_type,db,name);
    DBUG_RETURN(0);
  }
  table->reginfo.lock_type=TL_WRITE;
  if (!((*lock)=mysql_lock_tables(thd,&table,1)))
  {
    hash_delete(&open_cache,(byte*) table);
    quick_rm_table(create_info->db_type,db,name);
    DBUG_RETURN(0);
  }
  table->file->extra(HA_EXTRA_WRITE_CACHE);
  DBUG_RETURN(table);
}


/****************************************************************************
** Alter a table definition
****************************************************************************/

bool
mysql_rename_table(enum db_type base,
		   const char *old_db,
		   const char * old_name,
		   const char *new_db,
		   const char * new_name)
{
  char from[FN_REFLEN],to[FN_REFLEN];
  handler *file=get_new_handler((TABLE*) 0, base);
  int error=0;
  DBUG_ENTER("mysql_rename_table");
  (void) sprintf(from,"%s/%s/%s",mysql_data_home,old_db,old_name);
  (void) sprintf(to,"%s/%s/%s",mysql_data_home,new_db,new_name);
  fn_format(from,from,"","",4);
  fn_format(to,to,    "","",4);
  if (!(error=file->rename_table((const char*) from,(const char *) to)))
  {
    if (rename_file_ext(from,to,reg_ext))
    {
      error=my_errno;
      /* Restore old file name */
      file->rename_table((const char*) to,(const char *) from);
    }
  }
  delete file;
  if (error)
    my_error(ER_ERROR_ON_RENAME, MYF(0), from, to, error);
  DBUG_RETURN(error != 0);
}

/*
  close table in this thread and force close + reopen in other threads
  This assumes that the calling thread has lock on LOCK_open
  Win32 clients must also have a WRITE LOCK on the table !
*/

bool close_cached_table(THD *thd,TABLE *table)
{
  bool result=0;
  DBUG_ENTER("close_cached_table");
  if (table)
  {
    DBUG_PRINT("enter",("table: %s", table->table_name));
    VOID(table->file->extra(HA_EXTRA_FORCE_REOPEN)); // Close all data files
    /* Mark all tables that are in use as 'old' */
    mysql_lock_abort(thd,table);		 // end threads waiting on lock

#if defined(USING_TRANSACTIONS) || defined( __WIN__) || defined( __EMX__) || !defined(OS2)
    /* Wait until all there are no other threads that has this table open */
    while (remove_table_from_cache(thd,table->table_cache_key,
				   table->table_name))
    {
      dropping_tables++;
      (void) pthread_cond_wait(&COND_refresh,&LOCK_open);
      dropping_tables--;
    }
#else
    (void) remove_table_from_cache(thd,table->table_cache_key,
				   table->table_name);
#endif
    /* When lock on LOCK_open is freed other threads can continue */
    pthread_cond_broadcast(&COND_refresh);

    /* Close lock if this is not got with LOCK TABLES */
    if (thd->lock)
    {
      mysql_unlock_tables(thd, thd->lock); thd->lock=0;	// Start locked threads
    }
    /* Close all copies of 'table'.  This also frees all LOCK TABLES lock */
    thd->open_tables=unlink_open_table(thd,thd->open_tables,table);
  }
  DBUG_RETURN(result);
}

static int send_check_errmsg(THD* thd, TABLE_LIST* table,
			     const char* operator_name, const char* errmsg)

{

  String* packet = &thd->packet;
  packet->length(0);
  net_store_data(packet, table->name);
  net_store_data(packet, (char*)operator_name);
  net_store_data(packet, "error");
  net_store_data(packet, errmsg);
  thd->net.last_error[0]=0;
  if (my_net_write(&thd->net, (char*) thd->packet.ptr(),
		   packet->length()))
    return -1;
  return 1;
}

static int prepare_for_restore(THD* thd, TABLE_LIST* table,
    HA_CHECK_OPT *check_opt)
{
  DBUG_ENTER("prepare_for_restore");

  if (table->table) // do not overwrite existing tables on restore
  {
    DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				  "table exists, will not overwrite on restore"
				  ));
  }
  else
  {
    char* backup_dir = thd->lex.backup_dir;
    char src_path[FN_REFLEN], dst_path[FN_REFLEN];
    char* table_name = table->name;
    char* db = thd->db ? thd->db : table->db;

    if (fn_format_relative_to_data_home(src_path, table_name, backup_dir,
					reg_ext))
      DBUG_RETURN(-1); // protect buffer overflow

    sprintf(dst_path, "%s/%s/%s", mysql_real_data_home, db, table_name);

    if (lock_and_wait_for_table_name(thd,table))
      DBUG_RETURN(-1);

    if (my_copy(src_path,
		fn_format(dst_path, dst_path,"", reg_ext, 4),
		MYF(MY_WME)))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				    "Failed copying .frm file"));
    }
    if (mysql_truncate(thd, table, 1))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "restore",
				    "Failed generating table from .frm file"));
    }
  }

  // now we should be able to open the partially restored table
  // to finish the restore in the handler later on
  if (!(table->table = reopen_name_locked_table(thd, table)))
     unlock_table_name(thd, table);
  DBUG_RETURN(0);
}

static int prepare_for_repair(THD* thd, TABLE_LIST* table,
    HA_CHECK_OPT *check_opt)
{
  DBUG_ENTER("prepare_for_repair");

  if (!(check_opt->sql_flags & TT_USEFRM))
  {
    DBUG_RETURN(0);
  }
  else
  {

    char from[FN_REFLEN],tmp[FN_REFLEN];
    char* db = thd->db ? thd->db : table->db;

    sprintf(from, "%s/%s/%s", mysql_real_data_home, db, table->name);
    fn_format(from, from, "", MI_NAME_DEXT, 4);
    sprintf(tmp,"%s-%lx_%lx", from, current_pid, thd->thread_id);

    close_cached_table(thd,table->table);

    if (lock_and_wait_for_table_name(thd,table))
      DBUG_RETURN(-1);

    if (my_rename(from, tmp, MYF(MY_WME)))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "repair",
				    "Failed renaming .MYD file"));
    }
    if (mysql_truncate(thd, table, 1))
    {
      unlock_table_name(thd, table);
      DBUG_RETURN(send_check_errmsg(thd, table, "repair",
				    "Failed generating table from .frm file"));
    }
    if (my_rename(tmp, from, MYF(MY_WME)))
{'fraction_non_alphanumeric': 0.1541599570585078, 'fraction_numerical': 0.00704956163893362, 'mean_word_length': 3.650607422200033, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 530, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4354878', 'n_tokens_mistral': 21408, 'n_tokens_neox': 20637, 'n_words': 3662}