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

/*

  Creates a index for a database by reading keys, sorting them and outputing

  them in sorted order through SORT_INFO functions.

*/

#include "myisam_priv.h"

#include <stddef.h>

#include <queue>

#include <algorithm>

/* static variables */

#undef MIN_SORT_MEMORY

#undef MYF_RW

#undef DISK_BUFFER_SIZE

#define MERGEBUFF 15

#define MERGEBUFF2 31

#define MIN_SORT_MEMORY (4096-MALLOC_OVERHEAD)

#define MYF_RW  MYF(MY_NABP | MY_WME | MY_WAIT_IF_FULL)

#define DISK_BUFFER_SIZE (IO_SIZE*16)

using namespace std;

using namespace drizzled;

/*

 Pointers of functions for store and read keys from temp file

*/

extern void print_error(const char *fmt,...);

/* Functions defined in this file */

static ha_rows  find_all_keys(MI_SORT_PARAM *info,uint32_t keys,

			      unsigned char **sort_keys,

			      DYNAMIC_ARRAY *buffpek,

			      size_t *maxbuffer,

			      internal::IO_CACHE *tempfile,

			      internal::IO_CACHE *tempfile_for_exceptions);

static int  write_keys(MI_SORT_PARAM *info,unsigned char **sort_keys,

                             uint32_t count, BUFFPEK *buffpek,internal::IO_CACHE *tempfile);

static int  write_key(MI_SORT_PARAM *info, unsigned char *key,

			    internal::IO_CACHE *tempfile);

static int  write_index(MI_SORT_PARAM *info,unsigned char * *sort_keys,

                              uint32_t count);

static int  merge_many_buff(MI_SORT_PARAM *info,uint32_t keys,

			    unsigned char * *sort_keys,

			    BUFFPEK *buffpek,size_t *maxbuffer,

			    internal::IO_CACHE *t_file);

static uint32_t  read_to_buffer(internal::IO_CACHE *fromfile,BUFFPEK *buffpek,

                                  uint32_t sort_length);

static int  merge_buffers(MI_SORT_PARAM *info,uint32_t keys,

                                internal::IO_CACHE *from_file, internal::IO_CACHE *to_file,

                                unsigned char * *sort_keys, BUFFPEK *lastbuff,

                                BUFFPEK *Fb, BUFFPEK *Tb);

static int  merge_index(MI_SORT_PARAM *,uint,unsigned char **,BUFFPEK *, int,

                              internal::IO_CACHE *);

static int  write_keys_varlen(MI_SORT_PARAM *info,unsigned char **sort_keys,

                       uint32_t count, BUFFPEK *buffpek,

                       internal::IO_CACHE *tempfile);

static uint32_t  read_to_buffer_varlen(internal::IO_CACHE *fromfile,BUFFPEK *buffpek,

                                uint32_t sort_length);

static int  write_merge_key(MI_SORT_PARAM *info, internal::IO_CACHE *to_file,

                     unsigned char *key, uint32_t sort_length, uint32_t count);

static int  write_merge_key_varlen(MI_SORT_PARAM *info,

                            internal::IO_CACHE *to_file,

                            unsigned char* key, uint32_t sort_length,

                            uint32_t count);

inline int

my_var_write(MI_SORT_PARAM *info, internal::IO_CACHE *to_file, unsigned char *bufs);

/*

  Creates a index of sorted keys

  SYNOPSIS

    _create_index_by_sort()

    info		Sort parameters

    no_messages		Set to 1 if no output

    sortbuff_size	Size if sortbuffer to allocate

  RESULT

    0	ok

   <> 0 Error

*/

int _create_index_by_sort(MI_SORT_PARAM *info,bool no_messages,

			  size_t sortbuff_size)

{

  int error;

  size_t maxbuffer, skr;

  uint32_t memavl,old_memavl,keys,sort_length;

  DYNAMIC_ARRAY buffpek;

  ha_rows records;

  unsigned char **sort_keys;

  internal::IO_CACHE tempfile, tempfile_for_exceptions;

  if (info->keyinfo->flag & HA_VAR_LENGTH_KEY)

  {

    info->write_keys=write_keys_varlen;

    info->read_to_buffer=read_to_buffer_varlen;

    info->write_key= write_merge_key_varlen;

  }

  else

  {

    info->write_keys=write_keys;

    info->read_to_buffer=read_to_buffer;

    info->write_key=write_merge_key;

  }

  my_b_clear(&tempfile);

  my_b_clear(&tempfile_for_exceptions);

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

  sort_keys= (unsigned char **) NULL; error= 1;

  maxbuffer=1;

  memavl=max(sortbuff_size,(size_t)MIN_SORT_MEMORY);

  records=	info->sort_info->max_records;

  sort_length=	info->key_length;

  while (memavl >= MIN_SORT_MEMORY)

  {

    if ((records < UINT32_MAX) &&

       ((internal::my_off_t) (records + 1) *

        (sort_length + sizeof(char*)) <= (internal::my_off_t) memavl))

      keys= (uint)records+1;

    else

      do

      {

	skr=maxbuffer;

	if (memavl < sizeof(BUFFPEK)* maxbuffer ||

	    (keys=(memavl-sizeof(BUFFPEK)* maxbuffer)/

             (sort_length+sizeof(char*))) <= 1 ||

            keys < maxbuffer)

	{

	  mi_check_print_error(info->sort_info->param,

			       "myisam_sort_buffer_size is too small");

	  goto err;

	}

      }

      while ((maxbuffer= (size_t)(records/(keys-1)+1)) != skr);

    if ((sort_keys=(unsigned char **)malloc(keys*(sort_length+sizeof(char*)))))

    {

      if (my_init_dynamic_array(&buffpek, sizeof(BUFFPEK), maxbuffer,

			     maxbuffer/2))

      {

	free((unsigned char*) sort_keys);

        sort_keys= 0;

      }

      else

	break;

    }

    old_memavl=memavl;

    if ((memavl=memavl/4*3) < MIN_SORT_MEMORY && old_memavl > MIN_SORT_MEMORY)

      memavl=MIN_SORT_MEMORY;

  }

  if (memavl < MIN_SORT_MEMORY)

  {

    mi_check_print_error(info->sort_info->param,"MyISAM sort buffer too small");

    goto err;

  }

  (*info->lock_in_memory)(info->sort_info->param);/* Everything is allocated */

  if (!no_messages)

    printf("  - Searching for keys, allocating buffer for %d keys\n",keys);

  if ((records=find_all_keys(info,keys,sort_keys,&buffpek,&maxbuffer,

			     &tempfile,&tempfile_for_exceptions))

      == HA_POS_ERROR)

    goto err;

  if (maxbuffer == 0)

  {

    if (!no_messages)

      printf("  - Dumping %u keys\n", (uint32_t) records);

    if (write_index(info,sort_keys, (uint) records))

      goto err;

  }

  else

  {

    keys=(keys*(sort_length+sizeof(char*)))/sort_length;

    if (maxbuffer >= MERGEBUFF2)

    {

      if (!no_messages)

	printf("  - Merging %u keys\n", (uint32_t) records);

      if (merge_many_buff(info,keys,sort_keys,

                  dynamic_element(&buffpek,0,BUFFPEK *),&maxbuffer,&tempfile))

	goto err;

    }

    if (internal::flush_io_cache(&tempfile) ||

	internal::reinit_io_cache(&tempfile,internal::READ_CACHE,0L,0,0))

      goto err;

    if (!no_messages)

      printf("  - Last merge and dumping keys\n");

    if (merge_index(info,keys,sort_keys,dynamic_element(&buffpek,0,BUFFPEK *),

                    maxbuffer,&tempfile))

      goto err;

  }

  if (flush_pending_blocks(info))

    goto err;

  if (my_b_inited(&tempfile_for_exceptions))

  {

    MI_INFO *idx=info->sort_info->info;

    uint32_t     keyno=info->key;

    uint32_t     key_length, ref_length=idx->s->rec_reflength;

    if (!no_messages)

      printf("  - Adding exceptions\n");

    if (flush_io_cache(&tempfile_for_exceptions) ||

	reinit_io_cache(&tempfile_for_exceptions,internal::READ_CACHE,0L,0,0))

      goto err;

    while (!my_b_read(&tempfile_for_exceptions,(unsigned char*)&key_length,

		      sizeof(key_length))

        && !my_b_read(&tempfile_for_exceptions,(unsigned char*)sort_keys,

		      (uint) key_length))

    {

	if (_mi_ck_write(idx,keyno,(unsigned char*) sort_keys,key_length-ref_length))

	  goto err;

    }

  }

  error =0;

err:

  if (sort_keys)

    free((unsigned char*) sort_keys);

  delete_dynamic(&buffpek);

  close_cached_file(&tempfile);

  close_cached_file(&tempfile_for_exceptions);

  return(error ? -1 : 0);

} /* _create_index_by_sort */

/* Search after all keys and place them in a temp. file */

static ha_rows  find_all_keys(MI_SORT_PARAM *info, uint32_t keys,

			      unsigned char **sort_keys,

			      DYNAMIC_ARRAY *buffpek,

			      size_t *maxbuffer, internal::IO_CACHE *tempfile,

			      internal::IO_CACHE *tempfile_for_exceptions)

{

  int error;

  uint32_t idx;

  idx=error=0;

  sort_keys[0]=(unsigned char*) (sort_keys+keys);

  while (!(error=(*info->key_read)(info,sort_keys[idx])))

  {

    if (info->real_key_length > info->key_length)

    {

      if (write_key(info,sort_keys[idx],tempfile_for_exceptions))

        return(HA_POS_ERROR);

      continue;

    }

    if (++idx == keys)

    {

      if (info->write_keys(info,sort_keys,idx-1,(BUFFPEK *)alloc_dynamic(buffpek),

		     tempfile))

      return(HA_POS_ERROR);

      sort_keys[0]=(unsigned char*) (sort_keys+keys);

      memcpy(sort_keys[0],sort_keys[idx-1],(size_t) info->key_length);

      idx=1;

    }

    sort_keys[idx]=sort_keys[idx-1]+info->key_length;

  }

  if (error > 0)

    return(HA_POS_ERROR);

  if (buffpek->elements)

  {

    if (info->write_keys(info,sort_keys,idx,(BUFFPEK *)alloc_dynamic(buffpek),

		   tempfile))

      return(HA_POS_ERROR);

    *maxbuffer=buffpek->elements-1;

  }

  else

    *maxbuffer=0;

  return((*maxbuffer)*(keys-1)+idx);

} /* find_all_keys */

int thr_write_keys(MI_SORT_PARAM *sort_param)

{

  SORT_INFO *sort_info= sort_param->sort_info;

  MI_CHECK *param= sort_info->param;

  uint32_t length= 0, keys;

  ulong *rec_per_key_part= param->rec_per_key_part;

  int got_error=sort_info->got_error;

  uint32_t i;

  MI_INFO *info=sort_info->info;

  MYISAM_SHARE *share=info->s;

  MI_SORT_PARAM *sinfo;

  unsigned char *mergebuf=0;

  for (i= 0, sinfo= sort_param ;

       i < sort_info->total_keys ;

       i++, rec_per_key_part+=sinfo->keyinfo->keysegs, sinfo++)

  {

    if (!sinfo->sort_keys)

    {

      got_error=1;

      void * rec_buff_ptr= mi_get_rec_buff_ptr(info, sinfo->rec_buff);

      if (rec_buff_ptr != NULL)

        free(rec_buff_ptr);

      continue;

    }

    if (!got_error)

    {

      mi_set_key_active(share->state.key_map, sinfo->key);

      if (!sinfo->buffpek.elements)

      {

        if (param->testflag & T_VERBOSE)

        {

          printf("Key %d  - Dumping %u keys\n",sinfo->key+1, sinfo->keys);

          fflush(stdout);

        }

        if (write_index(sinfo, sinfo->sort_keys, sinfo->keys) || flush_pending_blocks(sinfo))

          got_error=1;

      }

      if (!got_error && param->testflag & T_STATISTICS)

        update_key_parts(sinfo->keyinfo, rec_per_key_part, sinfo->unique,

                         param->stats_method == MI_STATS_METHOD_IGNORE_NULLS?

                         sinfo->notnull: NULL,

                         (uint64_t) info->state->records);

    }

    free((unsigned char*) sinfo->sort_keys);

    void * rec_buff_ptr= mi_get_rec_buff_ptr(info, sinfo->rec_buff);

    if (rec_buff_ptr != NULL)

      free(rec_buff_ptr);

    sinfo->sort_keys=0;

  }

  for (i= 0, sinfo= sort_param ;

       i < sort_info->total_keys ;

       i++,

	 delete_dynamic(&sinfo->buffpek),

	 close_cached_file(&sinfo->tempfile),

	 close_cached_file(&sinfo->tempfile_for_exceptions),

	 sinfo++)

  {

    if (got_error)

      continue;

    if (sinfo->keyinfo->flag & HA_VAR_LENGTH_KEY)

    {

      sinfo->write_keys=write_keys_varlen;

      sinfo->read_to_buffer=read_to_buffer_varlen;

      sinfo->write_key=write_merge_key_varlen;

    }

    else

    {

      sinfo->write_keys=write_keys;

      sinfo->read_to_buffer=read_to_buffer;

      sinfo->write_key=write_merge_key;

    }

    if (sinfo->buffpek.elements)

    {

      size_t maxbuffer=sinfo->buffpek.elements-1;

      if (!mergebuf)

      {

        length=param->sort_buffer_length;

        while (length >= MIN_SORT_MEMORY)

        {

          if ((mergebuf= (unsigned char *)malloc(length)))

              break;

          length=length*3/4;

        }

        if (!mergebuf)

        {

          got_error=1;

          continue;

        }

      }

      keys=length/sinfo->key_length;

      if (maxbuffer >= MERGEBUFF2)

      {

        if (param->testflag & T_VERBOSE)

          printf("Key %d  - Merging %u keys\n",sinfo->key+1, sinfo->keys);

        if (merge_many_buff(sinfo, keys, (unsigned char **)mergebuf,

			    dynamic_element(&sinfo->buffpek, 0, BUFFPEK *),

			    &maxbuffer, &sinfo->tempfile))

        {

          got_error=1;

          continue;

        }

      }

      if (flush_io_cache(&sinfo->tempfile) ||

          reinit_io_cache(&sinfo->tempfile,internal::READ_CACHE,0L,0,0))

      {

        got_error=1;

        continue;

      }

      if (param->testflag & T_VERBOSE)

        printf("Key %d  - Last merge and dumping keys\n", sinfo->key+1);

      if (merge_index(sinfo, keys, (unsigned char **)mergebuf,

                      dynamic_element(&sinfo->buffpek,0,BUFFPEK *),

                      maxbuffer,&sinfo->tempfile) ||

	  flush_pending_blocks(sinfo))

      {

        got_error=1;

        continue;

      }

    }

    if (my_b_inited(&sinfo->tempfile_for_exceptions))

    {

      uint32_t key_length;

      if (param->testflag & T_VERBOSE)

        printf("Key %d  - Dumping 'long' keys\n", sinfo->key+1);

      if (flush_io_cache(&sinfo->tempfile_for_exceptions) ||

          reinit_io_cache(&sinfo->tempfile_for_exceptions,internal::READ_CACHE,0L,0,0))

      {

        got_error=1;

        continue;

      }

      while (!got_error &&

	     !my_b_read(&sinfo->tempfile_for_exceptions,(unsigned char*)&key_length,

			sizeof(key_length)))

      {

        unsigned char ft_buf[10];

        if (key_length > sizeof(ft_buf) ||

            my_b_read(&sinfo->tempfile_for_exceptions, (unsigned char*)ft_buf,

                      (uint)key_length) ||

            _mi_ck_write(info, sinfo->key, (unsigned char*)ft_buf,

                         key_length - info->s->rec_reflength))

          got_error=1;

      }

    }

  }

  free((unsigned char*) mergebuf);

  return(got_error);

}

        /* Write all keys in memory to file for later merge */

static int  write_keys(MI_SORT_PARAM *info, register unsigned char **sort_keys,

                             uint32_t count, BUFFPEK *buffpek, internal::IO_CACHE *tempfile)

{

  unsigned char **end;

  uint32_t sort_length=info->key_length;

  internal::my_qsort2((unsigned char*) sort_keys,count,sizeof(unsigned char*),(qsort2_cmp) info->key_cmp,

            info);

  if (!my_b_inited(tempfile) &&

      open_cached_file(tempfile, P_tmpdir, "ST",

                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))

    return(1);

  buffpek->file_pos=my_b_tell(tempfile);

  buffpek->count=count;

  for (end=sort_keys+count ; sort_keys != end ; sort_keys++)

  {

    if (my_b_write(tempfile,(unsigned char*) *sort_keys,(uint) sort_length))

      return(1);

  }

  return(0);

} /* write_keys */

inline int

my_var_write(MI_SORT_PARAM *info, internal::IO_CACHE *to_file, unsigned char *bufs)

{

  int err;

  uint16_t len = _mi_keylength(info->keyinfo, (unsigned char*) bufs);

  /* The following is safe as this is a local file */

  if ((err= my_b_write(to_file, (unsigned char*)&len, sizeof(len))))

    return (err);

  if ((err= my_b_write(to_file,bufs, (uint) len)))

    return (err);

  return (0);

}

static int  write_keys_varlen(MI_SORT_PARAM *info,

				    register unsigned char **sort_keys,

                                    uint32_t count, BUFFPEK *buffpek,

				    internal::IO_CACHE *tempfile)

{

  unsigned char **end;

  int err;

  internal::my_qsort2((unsigned char*) sort_keys,count,sizeof(unsigned char*),(qsort2_cmp) info->key_cmp,

            info);

  if (!my_b_inited(tempfile) &&

      open_cached_file(tempfile, P_tmpdir, "ST",

                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))

    return(1);

  buffpek->file_pos=my_b_tell(tempfile);

  buffpek->count=count;

  for (end=sort_keys+count ; sort_keys != end ; sort_keys++)

  {

    if ((err= my_var_write(info,tempfile, (unsigned char*) *sort_keys)))

      return(err);

  }

  return(0);

} /* write_keys_varlen */

static int  write_key(MI_SORT_PARAM *info, unsigned char *key,

			    internal::IO_CACHE *tempfile)

{

  uint32_t key_length=info->real_key_length;

  if (!my_b_inited(tempfile) &&

      open_cached_file(tempfile, P_tmpdir, "ST",

                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))

    return(1);

  if (my_b_write(tempfile,(unsigned char*)&key_length,sizeof(key_length)) ||

      my_b_write(tempfile,(unsigned char*)key,(uint) key_length))

    return(1);

  return(0);

} /* write_key */

/* Write index */

static int  write_index(MI_SORT_PARAM *info, register unsigned char **sort_keys,

                              register uint32_t count)

{

  internal::my_qsort2((unsigned char*) sort_keys,(size_t) count,sizeof(unsigned char*),

           (qsort2_cmp) info->key_cmp,info);

  while (count--)

  {

    if ((*info->key_write)(info,*sort_keys++))

      return(-1);

  }

  return(0);

} /* write_index */

        /* Merge buffers to make < MERGEBUFF2 buffers */

static int  merge_many_buff(MI_SORT_PARAM *info, uint32_t keys,

			    unsigned char **sort_keys, BUFFPEK *buffpek,

			    size_t *maxbuffer, internal::IO_CACHE *t_file)

{

  uint32_t i;

  internal::IO_CACHE t_file2, *from_file, *to_file, *temp;

  BUFFPEK *lastbuff;

  if (*maxbuffer < MERGEBUFF2)

    return(0);

  if (flush_io_cache(t_file) ||

      open_cached_file(&t_file2, P_tmpdir, "ST",

                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))

    return(1);

  from_file= t_file ; to_file= &t_file2;

  while (*maxbuffer >= MERGEBUFF2)

  {

    reinit_io_cache(from_file,internal::READ_CACHE,0L,0,0);

    reinit_io_cache(to_file,internal::WRITE_CACHE,0L,0,0);

    lastbuff=buffpek;

    for (i=0 ; i <= *maxbuffer-MERGEBUFF*3/2 ; i+=MERGEBUFF)

    {

      if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,

                        buffpek+i,buffpek+i+MERGEBUFF-1))

        goto cleanup;

    }

    if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,

                      buffpek+i,buffpek+ *maxbuffer))

      break;

    if (flush_io_cache(to_file))

      break;

    temp=from_file; from_file=to_file; to_file=temp;