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

#include "config.h"

#include "drizzled/internal/my_bit.h"

#include "myisampack.h"

#include "ha_myisam.h"

#include "myisam_priv.h"

#include "drizzled/option.h"

#include "drizzled/internal/my_bit.h"

#include "drizzled/internal/m_string.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.h"

#include "drizzled/plugin/client.h"

#include "drizzled/table.h"

#include "drizzled/field/timestamp.h"

#include "drizzled/memory/multi_malloc.h"

#include "drizzled/plugin/daemon.h"

#include <boost/algorithm/string.hpp>

#include <string>

#include <sstream>

#include <map>

#include <algorithm>

#include <boost/program_options.hpp>

#include <drizzled/module/option_map.h>

namespace po= boost::program_options;

using namespace std;

using namespace drizzled;

static const string engine_name("MyISAM");

boost::mutex THR_LOCK_myisam;

static uint32_t myisam_key_cache_block_size= KEY_CACHE_BLOCK_SIZE;

static uint32_t myisam_key_cache_size;

static uint32_t myisam_key_cache_division_limit;

static uint32_t myisam_key_cache_age_threshold;

static uint64_t max_sort_file_size;

static uint64_t sort_buffer_size;

void st_mi_isam_share::setKeyCache()

{

  (void)init_key_cache(&key_cache,

                       myisam_key_cache_block_size,

                       myisam_key_cache_size,

                       myisam_key_cache_division_limit, 

                       myisam_key_cache_age_threshold);

}

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

** MyISAM tables

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

static const char *ha_myisam_exts[] = {

  ".MYI",

  ".MYD",

  NULL

};

class MyisamEngine : public plugin::StorageEngine

{

  MyisamEngine();

  MyisamEngine(const MyisamEngine&);

  MyisamEngine& operator=(const MyisamEngine&);

public:

  explicit MyisamEngine(string name_arg) :

    plugin::StorageEngine(name_arg,

                          HTON_CAN_INDEX_BLOBS |

                          HTON_STATS_RECORDS_IS_EXACT |

                          HTON_TEMPORARY_ONLY |

                          HTON_NULL_IN_KEY |

                          HTON_HAS_RECORDS |

                          HTON_DUPLICATE_POS |

                          HTON_AUTO_PART_KEY |

                          HTON_SKIP_STORE_LOCK)

  {

  }

  virtual ~MyisamEngine()

  { 

    mi_panic(HA_PANIC_CLOSE);

  }

  virtual Cursor *create(Table &table)

  {

    return new ha_myisam(*this, table);

  }

  const char **bas_ext() const {

    return ha_myisam_exts;

  }

  int doCreateTable(Session&,

                    Table& table_arg,

                    const TableIdentifier &identifier,

                    message::Table&);

  int doRenameTable(Session&, const TableIdentifier &from, const TableIdentifier &to);

  int doDropTable(Session&, const TableIdentifier &identifier);

  int doGetTableDefinition(Session& session,

                           const TableIdentifier &identifier,

                           message::Table &table_message);

  uint32_t max_supported_keys()          const { return MI_MAX_KEY; }

  uint32_t max_supported_key_length()    const { return MI_MAX_KEY_LENGTH; }

  uint32_t max_supported_key_part_length() const { return MI_MAX_KEY_LENGTH; }

  uint32_t index_flags(enum  ha_key_alg) const

  {

    return (HA_READ_NEXT |

            HA_READ_PREV |

            HA_READ_RANGE |

            HA_READ_ORDER |

            HA_KEYREAD_ONLY);

  }

  bool doDoesTableExist(Session& session, const TableIdentifier &identifier);

  void doGetTableIdentifiers(drizzled::CachedDirectory &directory,

                             const drizzled::SchemaIdentifier &schema_identifier,

                             drizzled::TableIdentifiers &set_of_identifiers);

  bool validateCreateTableOption(const std::string &key, const std::string &state)

  {

    (void)state;

    if (boost::iequals(key, "ROW_FORMAT"))

    {

      return true;

    }

    return false;

  }

};

void MyisamEngine::doGetTableIdentifiers(drizzled::CachedDirectory&,

                                         const drizzled::SchemaIdentifier&,

                                         drizzled::TableIdentifiers&)

{

}

bool MyisamEngine::doDoesTableExist(Session &session, const TableIdentifier &identifier)

{

  return session.doesTableMessageExist(identifier);

}

int MyisamEngine::doGetTableDefinition(Session &session,

                                       const TableIdentifier &identifier,

                                       message::Table &table_message)

{

  if (session.getTableMessage(identifier, table_message))

    return EEXIST;

  return ENOENT;

}

/* 

  Convert to push_Warnings if you ever care about this, otherwise, it is a no-op.

*/

static void mi_check_print_msg(MI_CHECK *,	const char* ,

                               const char *, va_list )

{

}

/*

  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 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;

  MI_KEYDEF *keydef;

  MI_COLUMNDEF *recinfo, *recinfo_pos;

  HA_KEYSEG *keyseg;

  TableShare *share= table_arg->getMutableShare();

  uint32_t options= share->db_options_in_use;

  if (!(memory::multi_malloc(false,

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

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

          &keyseg, (share->key_parts + share->sizeKeys()) * sizeof(HA_KEYSEG),

          NULL)))

    return(HA_ERR_OUT_OF_MEM);

  keydef= *keydef_out;

  recinfo= *recinfo_out;

  for (i= 0; i < share->sizeKeys(); i++)

  {

    KeyInfo *pos= &table_arg->key_info[i];

    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_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->getInsertRecord());

      }

      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->getInsertRecord();

  recpos= 0;

  recinfo_pos= recinfo;

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

  {

    Field **field, *found= 0;

    minpos= share->getRecordLength();

    length= 0;

    for (field= table_arg->getFields(); *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->getInsertRecord());

    }

    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);

}

int ha_myisam::reset_auto_increment(uint64_t value)

{

  file->s->state.auto_increment= value;

  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;

        else if ((t1_keysegs_j__type == HA_KEYTYPE_VARBINARY2) &&

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

          t1_keysegs_j__type= HA_KEYTYPE_VARBINARY1;

      }

      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);

}

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;

  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;

  }

}

ha_myisam::ha_myisam(plugin::StorageEngine &engine_arg,

                     Table &table_arg)

  : Cursor(engine_arg, table_arg),

  file(0),

  can_enable_indexes(true),

  is_ordered(true)

{ }

Cursor *ha_myisam::clone(memory::Root *mem_root)

{

  ha_myisam *new_handler= static_cast <ha_myisam *>(Cursor::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::doOpen(const drizzled::TableIdentifier &identifier, 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(identifier, mode, test_if_locked)))

    return (errno ? errno : -1);

  if (!getTable()->getShare()->getType()) /* No need to perform a check for tmp table */

  {

    if ((errno= table2myisam(getTable(), &keyinfo, &recinfo, &recs)))

    {

      goto err;

    }

    if (check_definition(keyinfo, recinfo, getTable()->getShare()->sizeKeys(), recs,

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

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

    {

      errno= HA_ERR_CRASHED;

      goto err;

    }

  }

  assert(test_if_locked);

  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 (!getTable()->getShare()->db_record_offset)

    is_ordered= false;

  keys_with_parts.reset();

  for (i= 0; i < getTable()->getShare()->sizeKeys(); i++)

  {

    getTable()->key_info[i].block_size= file->s->keyinfo[i].block_length;

    KeyPartInfo *kp= getTable()->key_info[i].key_part;

    KeyPartInfo *kp_end= kp + getTable()->key_info[i].key_parts;

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

    {

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

      {

        keys_with_parts.set(i);

        break;

      }

    }

  }

  errno= 0;

  goto end;

 err:

  this->close();

 end:

  /*

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

    recinfo must be freed.

  */

  if (recinfo)

    free((unsigned char*) recinfo);

  return errno;

}

int ha_myisam::close(void)

{

  MI_INFO *tmp=file;

  file=0;

  return mi_close(tmp);

}

int ha_myisam::doInsertRecord(unsigned char *buf)