~drizzle-trunk/drizzle/development

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

#include <drizzled/server_includes.h>

#include <mysys/my_bit.h>

#include "myisampack.h"

#include "ha_myisam.h"

#include "myisamdef.h"

#include <drizzled/util/test.h>

#include <drizzled/error.h>

#include <drizzled/errmsg_print.h>

#include <drizzled/gettext.h>

#include <drizzled/session.h>

#include <drizzled/plugin/protocol.h>

#include <drizzled/table.h>

#include <drizzled/field/timestamp.h>

#include <string>

#include <algorithm>

using namespace std;

static const string engine_name("MyISAM");

ulong myisam_recover_options= HA_RECOVER_NONE;

pthread_mutex_t THR_LOCK_myisam= PTHREAD_MUTEX_INITIALIZER;

static uint32_t repair_threads;

static uint32_t block_size;

static uint64_t max_sort_file_size;

static uint64_t sort_buffer_size;

/* bits in myisam_recover_options */

const char *myisam_recover_names[] =

{ "DEFAULT", "BACKUP", "FORCE", "QUICK", NULL};

TYPELIB myisam_recover_typelib= {array_elements(myisam_recover_names)-1,"",

                                 myisam_recover_names, NULL};

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

** MyISAM tables

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

static const char *ha_myisam_exts[] = {

  ".MYI",

  ".MYD",

  NULL

};

class MyisamEngine : public StorageEngine

{

public:

  MyisamEngine(string name_arg)

   : StorageEngine(name_arg, HTON_CAN_RECREATE | HTON_TEMPORARY_ONLY | HTON_FILE_BASED) {}

  virtual handler *create(TableShare *table,

                          MEM_ROOT *mem_root)

  {

    return new (mem_root) ha_myisam(this, table);

  }

  const char **bas_ext() const {

    return ha_myisam_exts;

  }

  int createTableImplementation(Session *, const char *table_name,

                                Table *table_arg,

                                HA_CREATE_INFO *ha_create_info,

                                drizzled::message::Table*);

  int renameTableImplementation(Session*, const char *from, const char *to);

  int deleteTableImplementation(Session*, const string table_name);

};

// collect errors printed by mi_check routines

static void mi_check_print_msg(MI_CHECK *param,	const char* msg_type,

                               const char *fmt, va_list args)

{

  Session* session = (Session*)param->session;

  drizzled::plugin::Protocol *protocol= session->protocol;

  uint32_t length, msg_length;

  char msgbuf[MI_MAX_MSG_BUF];

  char name[NAME_LEN*2+2];

  msg_length= vsnprintf(msgbuf, sizeof(msgbuf), fmt, args);

  msgbuf[sizeof(msgbuf) - 1] = 0; // healthy paranoia

  if (!session->protocol->isConnected())

  {

    errmsg_printf(ERRMSG_LVL_ERROR, "%s",msgbuf);

    return;

  }

  if (param->testflag & (T_CREATE_MISSING_KEYS | T_SAFE_REPAIR |

			 T_AUTO_REPAIR))

  {

    my_message(ER_NOT_KEYFILE,msgbuf,MYF(MY_WME));

    return;

  }

  length= sprintf(name,"%s.%s",param->db_name,param->table_name);

  /*

    TODO: switch from protocol to push_warning here. The main reason we didn't

    it yet is parallel repair. Due to following trace:

    mi_check_print_msg/push_warning/sql_alloc/my_pthread_getspecific_ptr.

    Also we likely need to lock mutex here (in both cases with protocol and

    push_warning).

  */

  protocol->prepareForResend();

  protocol->store(name, length);

  protocol->store(param->op_name);

  protocol->store(msg_type);

  protocol->store(msgbuf, msg_length);

  if (protocol->write())

    errmsg_printf(ERRMSG_LVL_ERROR, "Failed on drizzleclient_net_write, writing to stderr instead: %s\n",

		    msgbuf);

  return;

}

/*

  Convert Table object to MyISAM key and column definition

  SYNOPSIS

    table2myisam()

      table_arg   in     Table object.

      keydef_out  out    MyISAM key definition.

      recinfo_out out    MyISAM column definition.

      records_out out    Number of fields.

  DESCRIPTION

    This function will allocate and initialize MyISAM key and column

    definition for further use in mi_create or for a check for underlying

    table conformance in merge engine.

    The caller needs to free *recinfo_out after use. Since *recinfo_out

    and *keydef_out are allocated with a my_multi_malloc, *keydef_out

    is freed automatically when *recinfo_out is freed.

  RETURN VALUE

    0  OK

    !0 error code

*/

static int table2myisam(Table *table_arg, MI_KEYDEF **keydef_out,

                        MI_COLUMNDEF **recinfo_out, uint32_t *records_out)

{

  uint32_t i, j, recpos, minpos, fieldpos, temp_length, length;

  enum ha_base_keytype type= HA_KEYTYPE_BINARY;

  unsigned char *record;

  KEY *pos;

  MI_KEYDEF *keydef;

  MI_COLUMNDEF *recinfo, *recinfo_pos;

  HA_KEYSEG *keyseg;

  TableShare *share= table_arg->s;

  uint32_t options= share->db_options_in_use;

  if (!(my_multi_malloc(MYF(MY_WME),

          recinfo_out, (share->fields * 2 + 2) * sizeof(MI_COLUMNDEF),

          keydef_out, share->keys * sizeof(MI_KEYDEF),

          &keyseg,

          (share->key_parts + share->keys) * sizeof(HA_KEYSEG),

          NULL)))

    return(HA_ERR_OUT_OF_MEM); /* purecov: inspected */

  keydef= *keydef_out;

  recinfo= *recinfo_out;

  pos= table_arg->key_info;

  for (i= 0; i < share->keys; i++, pos++)

  {

    keydef[i].flag= ((uint16_t) pos->flags & (HA_NOSAME));

    keydef[i].key_alg= HA_KEY_ALG_BTREE;

    keydef[i].block_length= pos->block_size;

    keydef[i].seg= keyseg;

    keydef[i].keysegs= pos->key_parts;

    for (j= 0; j < pos->key_parts; j++)

    {

      Field *field= pos->key_part[j].field;

      type= field->key_type();

      keydef[i].seg[j].flag= pos->key_part[j].key_part_flag;

      if (options & HA_OPTION_PACK_KEYS ||

          (pos->flags & (HA_PACK_KEY | HA_BINARY_PACK_KEY |

                         HA_SPACE_PACK_USED)))

      {

        if (pos->key_part[j].length > 8 &&

            (type == HA_KEYTYPE_TEXT ||

             type == HA_KEYTYPE_NUM ||

             (type == HA_KEYTYPE_BINARY && !field->zero_pack())))

        {

          /* No blobs here */

          if (j == 0)

            keydef[i].flag|= HA_PACK_KEY;

          if ((((int) (pos->key_part[j].length - field->decimals())) >= 4))

            keydef[i].seg[j].flag|= HA_SPACE_PACK;

        }

        else if (j == 0 && (!(pos->flags & HA_NOSAME) || pos->key_length > 16))

          keydef[i].flag|= HA_BINARY_PACK_KEY;

      }

      keydef[i].seg[j].type= (int) type;

      keydef[i].seg[j].start= pos->key_part[j].offset;

      keydef[i].seg[j].length= pos->key_part[j].length;

      keydef[i].seg[j].bit_start= keydef[i].seg[j].bit_end=

        keydef[i].seg[j].bit_length= 0;

      keydef[i].seg[j].bit_pos= 0;

      keydef[i].seg[j].language= field->charset()->number;

      if (field->null_ptr)

      {

        keydef[i].seg[j].null_bit= field->null_bit;

        keydef[i].seg[j].null_pos= (uint) (field->null_ptr-

                                           (unsigned char*) table_arg->record[0]);

      }

      else

      {

        keydef[i].seg[j].null_bit= 0;

        keydef[i].seg[j].null_pos= 0;

      }

      if (field->type() == DRIZZLE_TYPE_BLOB)

      {

        keydef[i].seg[j].flag|= HA_BLOB_PART;

        /* save number of bytes used to pack length */

        keydef[i].seg[j].bit_start= (uint) (field->pack_length() -

                                            share->blob_ptr_size);

      }

    }

    keyseg+= pos->key_parts;

  }

  if (table_arg->found_next_number_field)

    keydef[share->next_number_index].flag|= HA_AUTO_KEY;

  record= table_arg->record[0];

  recpos= 0;

  recinfo_pos= recinfo;

  while (recpos < (uint) share->stored_rec_length)

  {

    Field **field, *found= 0;

    minpos= share->reclength;

    length= 0;

    for (field= table_arg->field; *field; field++)

    {

      if ((fieldpos= (*field)->offset(record)) >= recpos &&

          fieldpos <= minpos)

      {

        /* skip null fields */

        if (!(temp_length= (*field)->pack_length_in_rec()))

          continue; /* Skip null-fields */

        if (! found || fieldpos < minpos ||

            (fieldpos == minpos && temp_length < length))

        {

          minpos= fieldpos;

          found= *field;

          length= temp_length;

        }

      }

    }

    if (recpos != minpos)

    { // Reserved space (Null bits?)

      memset(recinfo_pos, 0, sizeof(*recinfo_pos));

      recinfo_pos->type= (int) FIELD_NORMAL;

      recinfo_pos++->length= (uint16_t) (minpos - recpos);

    }

    if (!found)

      break;

    if (found->flags & BLOB_FLAG)

      recinfo_pos->type= (int) FIELD_BLOB;

    else if (found->type() == DRIZZLE_TYPE_VARCHAR)

      recinfo_pos->type= FIELD_VARCHAR;

    else if (!(options & HA_OPTION_PACK_RECORD))

      recinfo_pos->type= (int) FIELD_NORMAL;

    else if (found->zero_pack())

      recinfo_pos->type= (int) FIELD_SKIP_ZERO;

    else

      recinfo_pos->type= (int) ((length <= 3) ?  FIELD_NORMAL : FIELD_SKIP_PRESPACE);

    if (found->null_ptr)

    {

      recinfo_pos->null_bit= found->null_bit;

      recinfo_pos->null_pos= (uint) (found->null_ptr -

                                     (unsigned char*) table_arg->record[0]);

    }

    else

    {

      recinfo_pos->null_bit= 0;

      recinfo_pos->null_pos= 0;

    }

    (recinfo_pos++)->length= (uint16_t) length;

    recpos= minpos + length;

  }

  *records_out= (uint) (recinfo_pos - recinfo);

  return(0);

}

/*

  Check for underlying table conformance

  SYNOPSIS

    check_definition()

      t1_keyinfo       in    First table key definition

      t1_recinfo       in    First table record definition

      t1_keys          in    Number of keys in first table

      t1_recs          in    Number of records in first table

      t2_keyinfo       in    Second table key definition

      t2_recinfo       in    Second table record definition

      t2_keys          in    Number of keys in second table

      t2_recs          in    Number of records in second table

      strict           in    Strict check switch

  DESCRIPTION

    This function compares two MyISAM definitions. By intention it was done

    to compare merge table definition against underlying table definition.

    It may also be used to compare dot-frm and MYI definitions of MyISAM

    table as well to compare different MyISAM table definitions.

    For merge table it is not required that number of keys in merge table

    must exactly match number of keys in underlying table. When calling this

    function for underlying table conformance check, 'strict' flag must be

    set to false, and converted merge definition must be passed as t1_*.

    Otherwise 'strict' flag must be set to 1 and it is not required to pass

    converted dot-frm definition as t1_*.

  RETURN VALUE

    0 - Equal definitions.

    1 - Different definitions.

  TODO

    - compare FULLTEXT keys;

    - compare SPATIAL keys;

    - compare FIELD_SKIP_ZERO which is converted to FIELD_NORMAL correctly

      (should be corretly detected in table2myisam).

*/

static int check_definition(MI_KEYDEF *t1_keyinfo, MI_COLUMNDEF *t1_recinfo,

                            uint32_t t1_keys, uint32_t t1_recs,

                            MI_KEYDEF *t2_keyinfo, MI_COLUMNDEF *t2_recinfo,

                            uint32_t t2_keys, uint32_t t2_recs, bool strict)

{

  uint32_t i, j;

  if ((strict ? t1_keys != t2_keys : t1_keys > t2_keys))

  {

    return(1);

  }

  if (t1_recs != t2_recs)

  {

    return(1);

  }

  for (i= 0; i < t1_keys; i++)

  {

    HA_KEYSEG *t1_keysegs= t1_keyinfo[i].seg;

    HA_KEYSEG *t2_keysegs= t2_keyinfo[i].seg;

    if (t1_keyinfo[i].keysegs != t2_keyinfo[i].keysegs ||

        t1_keyinfo[i].key_alg != t2_keyinfo[i].key_alg)

    {

      return(1);

    }

    for (j=  t1_keyinfo[i].keysegs; j--;)

    {

      uint8_t t1_keysegs_j__type= t1_keysegs[j].type;

      /*

        Table migration from 4.1 to 5.1. In 5.1 a *TEXT key part is

        always HA_KEYTYPE_VARTEXT2. In 4.1 we had only the equivalent of

        HA_KEYTYPE_VARTEXT1. Since we treat both the same on MyISAM

        level, we can ignore a mismatch between these types.

      */

      if ((t1_keysegs[j].flag & HA_BLOB_PART) &&

          (t2_keysegs[j].flag & HA_BLOB_PART))

      {

        if ((t1_keysegs_j__type == HA_KEYTYPE_VARTEXT2) &&

            (t2_keysegs[j].type == HA_KEYTYPE_VARTEXT1))

          t1_keysegs_j__type= HA_KEYTYPE_VARTEXT1; /* purecov: tested */

        else if ((t1_keysegs_j__type == HA_KEYTYPE_VARBINARY2) &&

                 (t2_keysegs[j].type == HA_KEYTYPE_VARBINARY1))

          t1_keysegs_j__type= HA_KEYTYPE_VARBINARY1; /* purecov: inspected */

      }

      if (t1_keysegs_j__type != t2_keysegs[j].type ||

          t1_keysegs[j].language != t2_keysegs[j].language ||

          t1_keysegs[j].null_bit != t2_keysegs[j].null_bit ||

          t1_keysegs[j].length != t2_keysegs[j].length)

      {

        return(1);

      }

    }

  }

  for (i= 0; i < t1_recs; i++)

  {

    MI_COLUMNDEF *t1_rec= &t1_recinfo[i];

    MI_COLUMNDEF *t2_rec= &t2_recinfo[i];

    /*

      FIELD_SKIP_ZERO can be changed to FIELD_NORMAL in mi_create,

      see NOTE1 in mi_create.c

    */

    if ((t1_rec->type != t2_rec->type &&

         !(t1_rec->type == (int) FIELD_SKIP_ZERO &&

           t1_rec->length == 1 &&

           t2_rec->type == (int) FIELD_NORMAL)) ||

        t1_rec->length != t2_rec->length ||

        t1_rec->null_bit != t2_rec->null_bit)

    {

      return(1);

    }

  }

  return(0);

}

extern "C" {

volatile int *killed_ptr(MI_CHECK *param)

{

  /* In theory Unsafe conversion, but should be ok for now */

  return (int*) &(((Session *)(param->session))->killed);

}

void mi_check_print_error(MI_CHECK *param, const char *fmt,...)

{

  param->error_printed|=1;

  param->out_flag|= O_DATA_LOST;

  va_list args;

  va_start(args, fmt);

  mi_check_print_msg(param, "error", fmt, args);

  va_end(args);

}

void mi_check_print_info(MI_CHECK *param, const char *fmt,...)

{

  va_list args;

  va_start(args, fmt);

  mi_check_print_msg(param, "info", fmt, args);

  va_end(args);

}

void mi_check_print_warning(MI_CHECK *param, const char *fmt,...)

{

  param->warning_printed=1;

  param->out_flag|= O_DATA_LOST;

  va_list args;

  va_start(args, fmt);

  mi_check_print_msg(param, "warning", fmt, args);

  va_end(args);

}

/**

  Report list of threads (and queries) accessing a table, thread_id of a

  thread that detected corruption, ource file name and line number where

  this corruption was detected, optional extra information (string).

  This function is intended to be used when table corruption is detected.

  @param[in] file      MI_INFO object.

  @param[in] message   Optional error message.

  @param[in] sfile     Name of source file.

  @param[in] sline     Line number in source file.

  @return void

*/

void _mi_report_crashed(MI_INFO *file, const char *message,

                        const char *sfile, uint32_t sline)

{

  Session *cur_session;

  pthread_mutex_lock(&file->s->intern_lock);

  if ((cur_session= file->in_use))

    errmsg_printf(ERRMSG_LVL_ERROR, _("Got an error from thread_id=%"PRIu64", %s:%d"),

                    cur_session->thread_id,

                    sfile, sline);

  else

    errmsg_printf(ERRMSG_LVL_ERROR, _("Got an error from unknown thread, %s:%d"), sfile, sline);

  if (message)

    errmsg_printf(ERRMSG_LVL_ERROR, "%s", message);

  list<Session *>::iterator it= file->s->in_use->begin();

  while (it != file->s->in_use->end())

  {

    errmsg_printf(ERRMSG_LVL_ERROR, "%s", _("Unknown thread accessing table"));

    ++it;

  }

  pthread_mutex_unlock(&file->s->intern_lock);

}

}

ha_myisam::ha_myisam(StorageEngine *engine_arg, TableShare *table_arg)

  :handler(engine_arg, table_arg), file(0),

  int_table_flags(HA_NULL_IN_KEY |

                  HA_DUPLICATE_POS |

                  HA_CAN_INDEX_BLOBS |

                  HA_AUTO_PART_KEY |

                  HA_NO_TRANSACTIONS |

                  HA_HAS_RECORDS |

                  HA_STATS_RECORDS_IS_EXACT |

                  HA_NEED_READ_RANGE_BUFFER |

                  HA_MRR_CANT_SORT),

   can_enable_indexes(1)

{}

handler *ha_myisam::clone(MEM_ROOT *mem_root)

{

  ha_myisam *new_handler= static_cast <ha_myisam *>(handler::clone(mem_root));

  if (new_handler)

    new_handler->file->state= file->state;

  return new_handler;

}

const char *ha_myisam::index_type(uint32_t )

{

  return "BTREE";

}

/* Name is here without an extension */

int ha_myisam::open(const char *name, int mode, uint32_t test_if_locked)

{

  MI_KEYDEF *keyinfo;

  MI_COLUMNDEF *recinfo= 0;

  uint32_t recs;

  uint32_t i;

  /*

    If the user wants to have memory mapped data files, add an

    open_flag. Do not memory map temporary tables because they are

    expected to be inserted and thus extended a lot. Memory mapping is

    efficient for files that keep their size, but very inefficient for

    growing files. Using an open_flag instead of calling mi_extra(...

    HA_EXTRA_MMAP ...) after mi_open() has the advantage that the

    mapping is not repeated for every open, but just done on the initial

    open, when the MyISAM share is created. Everytime the server

    requires to open a new instance of a table it calls this method. We

    will always supply HA_OPEN_MMAP for a permanent table. However, the

    MyISAM storage engine will ignore this flag if this is a secondary

    open of a table that is in use by other threads already (if the

    MyISAM share exists already).

  */

  if (!(file=mi_open(name, mode, test_if_locked)))

    return (my_errno ? my_errno : -1);

  if (!table->s->tmp_table) /* No need to perform a check for tmp table */

  {

    if ((my_errno= table2myisam(table, &keyinfo, &recinfo, &recs)))

    {

      /* purecov: begin inspected */

      goto err;

      /* purecov: end */

    }

    if (check_definition(keyinfo, recinfo, table->s->keys, recs,

                         file->s->keyinfo, file->s->rec,

                         file->s->base.keys, file->s->base.fields, true))

    {

      /* purecov: begin inspected */

      my_errno= HA_ERR_CRASHED;

      goto err;

      /* purecov: end */

    }

  }

  if (test_if_locked & (HA_OPEN_IGNORE_IF_LOCKED | HA_OPEN_TMP_TABLE))

    mi_extra(file, HA_EXTRA_NO_WAIT_LOCK, 0);

  info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);

  if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))

    mi_extra(file, HA_EXTRA_WAIT_LOCK, 0);

  if (!table->s->db_record_offset)

    int_table_flags|=HA_REC_NOT_IN_SEQ;

  if (file->s->options & (HA_OPTION_CHECKSUM | HA_OPTION_COMPRESS_RECORD))

    int_table_flags|=HA_HAS_CHECKSUM;

  keys_with_parts.reset();

  for (i= 0; i < table->s->keys; i++)

  {

    table->key_info[i].block_size= file->s->keyinfo[i].block_length;

    KEY_PART_INFO *kp= table->key_info[i].key_part;

    KEY_PART_INFO *kp_end= kp + table->key_info[i].key_parts;

    for (; kp != kp_end; kp++)

    {

      if (!kp->field->part_of_key.test(i))

      {

        keys_with_parts.set(i);

        break;

      }

    }

  }

  my_errno= 0;

  goto end;

 err:

  this->close();

 end:

  /*

    Both recinfo and keydef are allocated by my_multi_malloc(), thus only

    recinfo must be freed.

  */

  if (recinfo)

    free((unsigned char*) recinfo);

  return my_errno;

}

int ha_myisam::close(void)

{

  MI_INFO *tmp=file;

  file=0;

  return mi_close(tmp);

}

int ha_myisam::write_row(unsigned char *buf)

{

  ha_statistic_increment(&SSV::ha_write_count);

  /*

    If we have an auto_increment column and we are writing a changed row

    or a new row, then update the auto_increment value in the record.

  */

  if (table->next_number_field && buf == table->record[0])

  {

    int error;

    if ((error= update_auto_increment()))

      return error;

  }

  return mi_write(file,buf);

}

int ha_myisam::check(Session* session, HA_CHECK_OPT* check_opt)

{

  if (!file) return HA_ADMIN_INTERNAL_ERROR;

  int error;

  MI_CHECK param;

  MYISAM_SHARE* share = file->s;

  const char *old_proc_info= session->get_proc_info();

  session->set_proc_info("Checking table");

  myisamchk_init(&param);

  param.session = session;

  param.op_name =   "check";

  param.db_name=    table->s->db.str;

  param.table_name= table->alias;

  param.testflag = check_opt->flags | T_CHECK | T_SILENT;

  param.stats_method= MI_STATS_METHOD_NULLS_NOT_EQUAL;

  if (!(table->db_stat & HA_READ_ONLY))

    param.testflag|= T_STATISTICS;

  param.using_global_keycache = 1;

  if (!mi_is_crashed(file) &&

      (((param.testflag & T_CHECK_ONLY_CHANGED) &&

	!(share->state.changed & (STATE_CHANGED | STATE_CRASHED |

				  STATE_CRASHED_ON_REPAIR)) &&

	share->state.open_count == 0) ||

       ((param.testflag & T_FAST) && (share->state.open_count ==

				      (uint) (share->global_changed ? 1 : 0)))))

    return HA_ADMIN_ALREADY_DONE;

  error = chk_status(&param, file);		// Not fatal