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

/* open a isam-database */

#include "myisam_priv.h"

#include <string.h>

#include <algorithm>

#include <memory>

#include <boost/scoped_ptr.hpp>

#include <boost/scoped_array.hpp>

#include <drizzled/internal/m_string.h>

#include <drizzled/util/test.h>

#include <drizzled/charset.h>

#include <drizzled/memory/multi_malloc.h>

#include <drizzled/identifier.h>

using namespace std;

using namespace drizzled;

static void setup_key_functions(MI_KEYDEF *keyinfo);

static unsigned char *mi_keydef_read(unsigned char *ptr, MI_KEYDEF *keydef);

static unsigned char *mi_keyseg_read(unsigned char *ptr, HA_KEYSEG *keyseg);

static unsigned char *mi_recinfo_read(unsigned char *ptr, MI_COLUMNDEF *recinfo);

static uint64_t mi_safe_mul(uint64_t a, uint64_t b);

static unsigned char *mi_state_info_read(unsigned char *ptr, MI_STATE_INFO *state);

static unsigned char *mi_uniquedef_read(unsigned char *ptr, MI_UNIQUEDEF *def);

static unsigned char *my_n_base_info_read(unsigned char *ptr, MI_BASE_INFO *base);

#define disk_pos_assert(pos, end_pos) \

if (pos > end_pos)             \

{                              \

  errno=HA_ERR_CRASHED;     \

  goto err;                    \

}

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

** Return the shared struct if the table is already open.

** In MySQL the server will handle version issues.

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

MI_INFO *test_if_reopen(char *filename)

{

  list<MI_INFO *>::iterator it= myisam_open_list.begin();

  while (it != myisam_open_list.end())

  {

    MI_INFO *info= *it;

    MYISAM_SHARE *share=info->s;

    if (!strcmp(share->unique_file_name,filename) && share->last_version)

      return info;

    ++it;

  }

  return 0;

}

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

  open a MyISAM database.

  See my_base.h for the handle_locking argument

  if handle_locking and HA_OPEN_ABORT_IF_CRASHED then abort if the table

  is marked crashed or if we are not using locking and the table doesn't

  have an open count of 0.

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

MI_INFO *mi_open(const drizzled::identifier::Table &identifier, int mode, uint32_t open_flags)

{

  int lock_error,kfile,open_mode,save_errno,have_rtree=0;

  uint32_t i,j,len,errpos,head_length,base_pos,offset,info_length,keys,

    key_parts,unique_key_parts,uniques;

  char name_buff[FN_REFLEN], org_name[FN_REFLEN], index_name[FN_REFLEN],

       data_name[FN_REFLEN], rp_buff[PATH_MAX];

  unsigned char *disk_cache= NULL;

  unsigned char *disk_pos, *end_pos;

  MI_INFO info,*m_info,*old_info;

  boost::scoped_ptr<MYISAM_SHARE> share_buff_ap(new MYISAM_SHARE);

  MYISAM_SHARE &share_buff= *share_buff_ap.get();

  MYISAM_SHARE *share;

  boost::scoped_array<ulong> rec_per_key_part_ap(new ulong[HA_MAX_POSSIBLE_KEY*MI_MAX_KEY_SEG]);

  ulong *rec_per_key_part= rec_per_key_part_ap.get();

  internal::my_off_t key_root[HA_MAX_POSSIBLE_KEY],key_del[MI_MAX_KEY_BLOCK_SIZE];

  uint64_t max_key_file_length, max_data_file_length;

  kfile= -1;

  lock_error=1;

  errpos=0;

  head_length=sizeof(share_buff.state.header);

  memset(&info, 0, sizeof(info));

  (void)internal::fn_format(org_name,

                            identifier.getPath().c_str(), 

                            "",

                            MI_NAME_IEXT,

                            MY_UNPACK_FILENAME);

  if (!realpath(org_name,rp_buff))

    internal::my_load_path(rp_buff,org_name, NULL);

  rp_buff[FN_REFLEN-1]= '\0';

  strcpy(name_buff,rp_buff);

  THR_LOCK_myisam.lock();

  if (!(old_info=test_if_reopen(name_buff)))

  {

    share= &share_buff;

    memset(&share_buff, 0, sizeof(share_buff));

    share_buff.state.rec_per_key_part=rec_per_key_part;

    share_buff.state.key_root=key_root;

    share_buff.state.key_del=key_del;

    share_buff.setKeyCache();

    if ((kfile=internal::my_open(name_buff,(open_mode=O_RDWR),MYF(0))) < 0)

    {

      if ((errno != EROFS && errno != EACCES) ||

	  mode != O_RDONLY ||

	  (kfile=internal::my_open(name_buff,(open_mode=O_RDONLY),MYF(0))) < 0)

	goto err;

    }

    share->mode=open_mode;

    errpos=1;

    if (internal::my_read(kfile, share->state.header.file_version, head_length,

		MYF(MY_NABP)))

    {

      errno= HA_ERR_NOT_A_TABLE;

      goto err;

    }

    if (memcmp(share->state.header.file_version, myisam_file_magic, 4))

    {

      errno=HA_ERR_NOT_A_TABLE;

      goto err;

    }

    share->options= mi_uint2korr(share->state.header.options);

    static const uint64_t OLD_FILE_OPTIONS= HA_OPTION_PACK_RECORD |

	    HA_OPTION_PACK_KEYS |

	    HA_OPTION_COMPRESS_RECORD | HA_OPTION_READ_ONLY_DATA |

	    HA_OPTION_TEMP_COMPRESS_RECORD |

	    HA_OPTION_TMP_TABLE;

    if (share->options & ~OLD_FILE_OPTIONS)

    {

      errno=HA_ERR_OLD_FILE;

      goto err;

    }

    /* Don't call realpath() if the name can't be a link */

    ssize_t sym_link_size= readlink(org_name,index_name,FN_REFLEN-1);

    if (sym_link_size >= 0 )

      index_name[sym_link_size]= '\0';

    if (!strcmp(name_buff, org_name) || sym_link_size == -1)

      (void) strcpy(index_name, org_name);

    *strrchr(org_name, '.')= '\0';

    (void) internal::fn_format(data_name,org_name,"",MI_NAME_DEXT,

                     MY_APPEND_EXT|MY_UNPACK_FILENAME|MY_RESOLVE_SYMLINKS);

    info_length=mi_uint2korr(share->state.header.header_length);

    base_pos=mi_uint2korr(share->state.header.base_pos);

    if (!(disk_cache= (unsigned char*) malloc(info_length+128)))

    {

      errno=ENOMEM;

      goto err;

    }

    end_pos=disk_cache+info_length;

    errpos=2;

    lseek(kfile,0,SEEK_SET);

    errpos=3;

    if (internal::my_read(kfile,disk_cache,info_length,MYF(MY_NABP)))

    {

      errno=HA_ERR_CRASHED;

      goto err;

    }

    len=mi_uint2korr(share->state.header.state_info_length);

    keys=    (uint) share->state.header.keys;

    uniques= (uint) share->state.header.uniques;

    key_parts= mi_uint2korr(share->state.header.key_parts);

    unique_key_parts= mi_uint2korr(share->state.header.unique_key_parts);

    share->state_diff_length=len-MI_STATE_INFO_SIZE;

    mi_state_info_read(disk_cache, &share->state);

    len= mi_uint2korr(share->state.header.base_info_length);

    disk_pos= my_n_base_info_read(disk_cache + base_pos, &share->base);

    share->state.state_length=base_pos;

    if (share->state.changed & STATE_CRASHED)

    {

      errno=((share->state.changed & STATE_CRASHED_ON_REPAIR) ?

		HA_ERR_CRASHED_ON_REPAIR : HA_ERR_CRASHED_ON_USAGE);

      goto err;

    }

    /* sanity check */

    if (share->base.keystart > 65535 || share->base.rec_reflength > 8)

    {

      errno=HA_ERR_CRASHED;

      goto err;

    }

    if (share->base.max_key_length > MI_MAX_KEY_BUFF || keys > MI_MAX_KEY ||

	key_parts > MI_MAX_KEY * MI_MAX_KEY_SEG)

    {

      errno=HA_ERR_UNSUPPORTED;

      goto err;

    }

    /* Correct max_file_length based on length of sizeof(off_t) */

    max_data_file_length=

      (share->options & (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)) ?

      (((uint64_t) 1 << (share->base.rec_reflength*8))-1) :

      (mi_safe_mul(share->base.pack_reclength,

		   (uint64_t) 1 << (share->base.rec_reflength*8))-1);

    max_key_file_length=

      mi_safe_mul(MI_MIN_KEY_BLOCK_LENGTH,

		  ((uint64_t) 1 << (share->base.key_reflength*8))-1);

#if SIZEOF_OFF_T == 4

    set_if_smaller(max_data_file_length, INT32_MAX);

    set_if_smaller(max_key_file_length, INT32_MAX);

#endif

    if (share->base.raid_type)

    {

      errno=HA_ERR_UNSUPPORTED;

      goto err;

    }

    share->base.max_data_file_length=(internal::my_off_t) max_data_file_length;

    share->base.max_key_file_length=(internal::my_off_t) max_key_file_length;

    if (share->options & HA_OPTION_COMPRESS_RECORD)

      share->base.max_key_length+=2;	/* For safety */

    /* Add space for node pointer */

    share->base.max_key_length+= share->base.key_reflength;

    if (!drizzled::memory::multi_malloc(false,

           &share,sizeof(*share),

           &share->state.rec_per_key_part,sizeof(long)*key_parts,

           &share->keyinfo,keys*sizeof(MI_KEYDEF),

           &share->uniqueinfo,uniques*sizeof(MI_UNIQUEDEF),

           &share->keyparts,

           (key_parts+unique_key_parts+keys+uniques) * sizeof(HA_KEYSEG),

           &share->rec, (share->base.fields+1)*sizeof(MI_COLUMNDEF),

           &share->blobs,sizeof(MI_BLOB)*share->base.blobs,

           &share->unique_file_name,strlen(name_buff)+1,

           &share->index_file_name,strlen(index_name)+1,

           &share->data_file_name,strlen(data_name)+1,

           &share->state.key_root,keys*sizeof(uint64_t),

           &share->state.key_del,

           (share->state.header.max_block_size_index*sizeof(uint64_t)),

           NULL))

      goto err;

    errpos=4;

    *share=share_buff;

    memcpy(share->state.rec_per_key_part, rec_per_key_part,

           sizeof(long)*key_parts);

    memcpy(share->state.key_root, key_root,

           sizeof(internal::my_off_t)*keys);

    memcpy(share->state.key_del, key_del,

           sizeof(internal::my_off_t) * share->state.header.max_block_size_index);

    strcpy(share->unique_file_name, name_buff);

    share->unique_name_length= strlen(name_buff);

    strcpy(share->index_file_name,  index_name);

    strcpy(share->data_file_name,   data_name);

    share->blocksize=min((uint32_t)IO_SIZE,myisam_block_size);

    {

      HA_KEYSEG *pos=share->keyparts;

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

      {

        share->keyinfo[i].share= share;

	disk_pos=mi_keydef_read(disk_pos, &share->keyinfo[i]);

        disk_pos_assert(disk_pos + share->keyinfo[i].keysegs * HA_KEYSEG_SIZE,

 			end_pos);

	set_if_smaller(share->blocksize,(uint)share->keyinfo[i].block_length);

	share->keyinfo[i].seg=pos;

	for (j=0 ; j < share->keyinfo[i].keysegs; j++,pos++)

	{

	  disk_pos=mi_keyseg_read(disk_pos, pos);

          if (pos->flag & HA_BLOB_PART &&

              ! (share->options & (HA_OPTION_COMPRESS_RECORD |

                                   HA_OPTION_PACK_RECORD)))

          {

            errno= HA_ERR_CRASHED;

            goto err;

          }

	  if (pos->type == HA_KEYTYPE_TEXT ||

              pos->type == HA_KEYTYPE_VARTEXT1 ||

              pos->type == HA_KEYTYPE_VARTEXT2)

	  {

	    if (!pos->language)

	      pos->charset=default_charset_info;

	    else if (!(pos->charset= get_charset(pos->language)))

	    {

	      errno=HA_ERR_UNKNOWN_CHARSET;

	      goto err;

	    }

	  }

	  else if (pos->type == HA_KEYTYPE_BINARY)

	    pos->charset= &my_charset_bin;

	}

        setup_key_functions(share->keyinfo+i);

	share->keyinfo[i].end=pos;

	pos->type=HA_KEYTYPE_END;			/* End */

	pos->length=share->base.rec_reflength;

	pos->null_bit=0;

	pos->flag=0;					/* For purify */

	pos++;

      }

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

      {

	disk_pos=mi_uniquedef_read(disk_pos, &share->uniqueinfo[i]);

        disk_pos_assert(disk_pos + share->uniqueinfo[i].keysegs *

			HA_KEYSEG_SIZE, end_pos);

	share->uniqueinfo[i].seg=pos;

	for (j=0 ; j < share->uniqueinfo[i].keysegs; j++,pos++)

	{

	  disk_pos=mi_keyseg_read(disk_pos, pos);

	  if (pos->type == HA_KEYTYPE_TEXT ||

              pos->type == HA_KEYTYPE_VARTEXT1 ||

              pos->type == HA_KEYTYPE_VARTEXT2)

	  {

	    if (!pos->language)

	      pos->charset=default_charset_info;

	    else if (!(pos->charset= get_charset(pos->language)))

	    {

	      errno=HA_ERR_UNKNOWN_CHARSET;

	      goto err;

	    }

	  }

	}

	share->uniqueinfo[i].end=pos;

	pos->type=HA_KEYTYPE_END;			/* End */

	pos->null_bit=0;

	pos->flag=0;

	pos++;

      }

    }

    disk_pos_assert(disk_pos + share->base.fields *MI_COLUMNDEF_SIZE, end_pos);

    for (i=j=offset=0 ; i < share->base.fields ; i++)

    {

      disk_pos=mi_recinfo_read(disk_pos,&share->rec[i]);

      share->rec[i].pack_type=0;

      share->rec[i].huff_tree=0;

      share->rec[i].offset=offset;

      if (share->rec[i].type == (int) FIELD_BLOB)

      {

	share->blobs[j].pack_length=

	  share->rec[i].length-portable_sizeof_char_ptr;

	share->blobs[j].offset=offset;

	j++;

      }

      offset+=share->rec[i].length;

    }

    share->rec[i].type=(int) FIELD_LAST;	/* End marker */

    if (offset > share->base.reclength)

    {

      errno= HA_ERR_CRASHED;

      goto err;

    }

    if (! lock_error)

    {

      lock_error=1;			/* Database unlocked */

    }

    if (mi_open_datafile(&info, share, -1))

      goto err;

    errpos=5;

    share->kfile=kfile;

    share->this_process=(ulong) getpid();

    share->last_process= share->state.process;

    share->base.key_parts=key_parts;

    share->base.all_key_parts=key_parts+unique_key_parts;

    if (!(share->last_version=share->state.version))

      share->last_version=1;			/* Safety */

    share->rec_reflength=share->base.rec_reflength; /* May be changed */

    share->base.margin_key_file_length=(share->base.max_key_file_length -

					(keys ? MI_INDEX_BLOCK_MARGIN *

					 share->blocksize * keys : 0));

    share->blocksize=min((uint32_t)IO_SIZE,myisam_block_size);

    share->data_file_type=STATIC_RECORD;

    if (share->options & HA_OPTION_PACK_RECORD)

      share->data_file_type = DYNAMIC_RECORD;

    free(disk_cache);

    disk_cache= NULL;

    mi_setup_functions(share);

    share->is_log_table= false;

    if (myisam_concurrent_insert)

    {

      share->concurrent_insert=

	((share->options & (HA_OPTION_READ_ONLY_DATA | HA_OPTION_TMP_TABLE |

			   HA_OPTION_COMPRESS_RECORD |

			   HA_OPTION_TEMP_COMPRESS_RECORD)) ||

	 (open_flags & HA_OPEN_TMP_TABLE) || have_rtree) ? 0 : 1;

      if (share->concurrent_insert)

      {

        assert(0);

      }

    }

  }

  else

  {

    share= old_info->s;

    if (mode == O_RDWR && share->mode == O_RDONLY)

    {

      errno=EACCES;				/* Can't open in write mode */

      goto err;

    }

    if (mi_open_datafile(&info, share, old_info->dfile))

      goto err;

    errpos=5;

    have_rtree= old_info->rtree_recursion_state != NULL;

  }

  /* alloc and set up private structure parts */

  if (!drizzled::memory::multi_malloc(MY_WME,

         &m_info,sizeof(MI_INFO),

         &info.blobs,sizeof(MI_BLOB)*share->base.blobs,

         &info.buff,(share->base.max_key_block_length*2+

                     share->base.max_key_length),

         &info.lastkey,share->base.max_key_length*3+1,

         &info.first_mbr_key, share->base.max_key_length,

         &info.filename, identifier.getPath().length()+1,

         &info.rtree_recursion_state,have_rtree ? 1024 : 0,

         NULL))

    goto err;

  errpos=6;

  if (!have_rtree)

    info.rtree_recursion_state= NULL;

  strcpy(info.filename, identifier.getPath().c_str());

  memcpy(info.blobs,share->blobs,sizeof(MI_BLOB)*share->base.blobs);

  info.lastkey2=info.lastkey+share->base.max_key_length;

  info.s=share;

  info.lastpos= HA_OFFSET_ERROR;

  info.update= (short) (HA_STATE_NEXT_FOUND+HA_STATE_PREV_FOUND);

  info.opt_flag=READ_CHECK_USED;

  info.this_unique= (ulong) info.dfile; /* Uniq number in process */

  if (share->data_file_type == COMPRESSED_RECORD)

    info.this_unique= share->state.unique;

  info.this_loop=0;				/* Update counter */

  info.last_unique= share->state.unique;

  info.last_loop=   share->state.update_count;

  if (mode == O_RDONLY)

    share->options|=HA_OPTION_READ_ONLY_DATA;

  info.lock_type=F_UNLCK;

  info.quick_mode=0;

  info.bulk_insert=0;

  info.errkey= -1;

  info.page_changed=1;

  info.read_record=share->read_record;

  share->reopen++;

  share->write_flag=MYF(MY_NABP | MY_WAIT_IF_FULL);

  if (share->options & HA_OPTION_READ_ONLY_DATA)

  {

    info.lock_type=F_RDLCK;

    share->r_locks++;

    share->tot_locks++;

  }

  if ((open_flags & HA_OPEN_TMP_TABLE) ||

      (share->options & HA_OPTION_TMP_TABLE))

  {

    share->temporary=share->delay_key_write=1;

    share->write_flag=MYF(MY_NABP);

    /*

     * The following two statements are commented out as a fix of

     * bug https://bugs.launchpad.net/drizzle/+bug/387627

     *

     * UPDATE can be TRUNCATE on TEMPORARY TABLE (MyISAM).

     * The root cause of why this makes a difference hasn't

     * been found, but this fixes things for now.

     */

//    share->w_locks++;			// We don't have to update status

//    share->tot_locks++;

    info.lock_type=F_WRLCK;

  }

  share->delay_key_write= 1;

  info.state= &share->state.state;	/* Change global values by default */

  /* Allocate buffer for one record */

  /* prerequisites: memset(info, 0) && info->s=share; are met. */

  if (!mi_alloc_rec_buff(&info, SIZE_MAX, &info.rec_buff))

    goto err;

  memset(info.rec_buff, 0, mi_get_rec_buff_len(&info, info.rec_buff));

  *m_info=info;

  myisam_open_list.push_front(m_info);

  THR_LOCK_myisam.unlock();

  return(m_info);

err:

  free(disk_cache);

  save_errno=errno ? errno : HA_ERR_END_OF_FILE;

  if ((save_errno == HA_ERR_CRASHED) ||

      (save_errno == HA_ERR_CRASHED_ON_USAGE) ||

      (save_errno == HA_ERR_CRASHED_ON_REPAIR))

    mi_report_error(save_errno, identifier.getPath().c_str());

  switch (errpos) {

  case 6:

    free((unsigned char*) m_info);

    /* fall through */

  case 5:

    internal::my_close(info.dfile,MYF(0));

    if (old_info)

      break;					/* Don't remove open table */

    /* fall through */

  case 4:

    free((unsigned char*) share);

    /* fall through */

  case 3:

    /* fall through */

  case 1:

    internal::my_close(kfile,MYF(0));

    /* fall through */

  case 0:

  default:

    break;

  }

  THR_LOCK_myisam.unlock();

  errno=save_errno;

  return (NULL);

} /* mi_open */

unsigned char *mi_alloc_rec_buff(MI_INFO *info, size_t length, unsigned char **buf)

{

  uint32_t extra;

  uint32_t old_length= 0;

  if (! *buf || length > (old_length=mi_get_rec_buff_len(info, *buf)))

  {

    unsigned char *newptr = *buf;

    /* to simplify initial init of info->rec_buf in mi_open and mi_extra */

    if (length == SIZE_MAX)

    {

      if (info->s->options & HA_OPTION_COMPRESS_RECORD)

        length= max(info->s->base.pack_reclength, info->s->max_pack_length);

      else

        length= info->s->base.pack_reclength;

      length= max((uint32_t)length, info->s->base.max_key_length);

      /* Avoid unnecessary realloc */

      if (newptr && length == old_length)

	return newptr;

    }

    extra= ((info->s->options & HA_OPTION_PACK_RECORD) ?

	    ALIGN_SIZE(MI_MAX_DYN_BLOCK_HEADER)+MI_SPLIT_LENGTH+

	    MI_REC_BUFF_OFFSET : 0);

    if (extra && newptr)

      newptr-= MI_REC_BUFF_OFFSET;

    void *tmpnewptr= NULL;

    if (!(tmpnewptr= realloc(newptr, length+extra+8))) 

      return newptr;

    newptr= (unsigned char *)tmpnewptr;

    *((uint32_t *) newptr)= (uint32_t) length;

    *buf= newptr+(extra ?  MI_REC_BUFF_OFFSET : 0);

  }

  return *buf;

}

static uint64_t mi_safe_mul(uint64_t a, uint64_t b)

{

  uint64_t max_val= ~ (uint64_t) 0;		/* internal::my_off_t is unsigned */

  if (!a || max_val / a < b)

    return max_val;

  return a*b;

}

	/* Set up functions in structs */