diff options
Diffstat (limited to 'storage/maria/ma_sort.c')
| -rw-r--r-- | storage/maria/ma_sort.c | 1185 |
1 files changed, 1185 insertions, 0 deletions
diff --git a/storage/maria/ma_sort.c b/storage/maria/ma_sort.c new file mode 100644 index 00000000..de6f5b8b --- /dev/null +++ b/storage/maria/ma_sort.c @@ -0,0 +1,1185 @@ +/* Copyright (C) 2006 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; 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 Street, Fifth Floor, Boston, MA 02110-1335 USA */ + +/* + Creates a index for a database by reading keys, sorting them and outputing + them in sorted order through MARIA_SORT_INFO functions. +*/ + +#include "ma_fulltext.h" +#include <my_check_opt.h> +#if defined(_WIN32) +#include <fcntl.h> +#else +#include <stddef.h> +#endif +#include <queues.h> + +/* static variables */ + +#undef DISK_BUFFER_SIZE + +#define MERGEBUFF 15 +#define MERGEBUFF2 31 +#define DISK_BUFFER_SIZE (IO_SIZE*128) + +/* How many keys we can keep in memory */ +typedef ulonglong ha_keys; + +/* + 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(MARIA_SORT_PARAM *info, ha_keys keys, + uchar **sort_keys, + DYNAMIC_ARRAY *buffpek,size_t *maxbuffer, + IO_CACHE *tempfile, + IO_CACHE *tempfile_for_exceptions); +static int write_keys(MARIA_SORT_PARAM *info,uchar **sort_keys, + ha_keys count, BUFFPEK *buffpek,IO_CACHE *tempfile); +static int write_key(MARIA_SORT_PARAM *info, uchar *key, + IO_CACHE *tempfile); +static int write_index(MARIA_SORT_PARAM *info, uchar **sort_keys, + ha_keys count); +static int merge_many_buff(MARIA_SORT_PARAM *info, ha_keys keys, + uchar **sort_keys, + BUFFPEK *buffpek, size_t *maxbuffer, + IO_CACHE *t_file); +static my_off_t read_to_buffer(IO_CACHE *fromfile,BUFFPEK *buffpek, + uint sort_length); +static int merge_buffers(MARIA_SORT_PARAM *info, ha_keys keys, + IO_CACHE *from_file, IO_CACHE *to_file, + uchar **sort_keys, BUFFPEK *lastbuff, + BUFFPEK *Fb, BUFFPEK *Tb); +static int merge_index(MARIA_SORT_PARAM *,ha_keys,uchar **,BUFFPEK *, size_t, + IO_CACHE *); +static int flush_maria_ft_buf(MARIA_SORT_PARAM *info); + +static int write_keys_varlen(MARIA_SORT_PARAM *info,uchar **sort_keys, + ha_keys count, BUFFPEK *buffpek, + IO_CACHE *tempfile); +static my_off_t read_to_buffer_varlen(IO_CACHE *fromfile,BUFFPEK *buffpek, + uint sort_length); +static int write_merge_key(MARIA_SORT_PARAM *info, IO_CACHE *to_file, + uchar *key, uint sort_length, ha_keys count); +static int write_merge_key_varlen(MARIA_SORT_PARAM *info, + IO_CACHE *to_file, + uchar* key, uint sort_length, + ha_keys count); +static inline int +my_var_write(MARIA_SORT_PARAM *info, IO_CACHE *to_file, uchar *bufs); + +/* + Sets the appropriate read and write methods for the MARIA_SORT_PARAM + based on the variable length key flag. +*/ +static void set_sort_param_read_write(MARIA_SORT_PARAM *sort_param) +{ + if (sort_param->keyinfo->flag & HA_VAR_LENGTH_KEY) + { + sort_param->write_keys= write_keys_varlen; + sort_param->read_to_buffer= read_to_buffer_varlen; + sort_param->write_key= write_merge_key_varlen; + } + else + { + sort_param->write_keys= write_keys; + sort_param->read_to_buffer= read_to_buffer; + sort_param->write_key= write_merge_key; + } +} + + +/* + Creates a index of sorted keys + + SYNOPSIS + _ma_create_index_by_sort() + info Sort parameters + no_messages Set to 1 if no output + sortbuff_size Size of sortbuffer to allocate + + RESULT + 0 ok + <> 0 Error +*/ + +int _ma_create_index_by_sort(MARIA_SORT_PARAM *info, my_bool no_messages, + size_t sortbuff_size) +{ + int error; + uint sort_length; + size_t memavl, old_memavl, maxbuffer; + DYNAMIC_ARRAY buffpek; + ha_rows records, UNINIT_VAR(keys); + uchar **sort_keys; + IO_CACHE tempfile, tempfile_for_exceptions; + DBUG_ENTER("_ma_create_index_by_sort"); + DBUG_PRINT("enter",("sort_buff_size: %lu sort_length: %d max_records: %lu", + (ulong) sortbuff_size, info->key_length, + (ulong) info->sort_info->max_records)); + + set_sort_param_read_write(info); + + my_b_clear(&tempfile); + my_b_clear(&tempfile_for_exceptions); + bzero((char*) &buffpek,sizeof(buffpek)); + sort_keys= (uchar **) NULL; error= 1; + maxbuffer=1; + + memavl=MY_MAX(sortbuff_size,MARIA_MIN_SORT_MEMORY); + records= info->sort_info->max_records; + sort_length= info->key_length; + + while (memavl >= MARIA_MIN_SORT_MEMORY) + { + /* Check if we can fit all keys into memory */ + if (((ulonglong) (records + 1) * + (sort_length + sizeof(char*)) <= memavl)) + keys= records+1; + else if ((info->sort_info->param->testflag & + (T_FORCE_SORT_MEMORY | T_CREATE_MISSING_KEYS)) == + T_FORCE_SORT_MEMORY) + { + /* + Use all of the given sort buffer for key data. + Allocate 1000 buffers at a start for new data. More buffers + will be allocated when needed. + */ + keys= memavl / (sort_length+sizeof(char*)); + maxbuffer= (size_t) MY_MIN((ulonglong) 1000, (records / keys)+1); + } + else + { + /* + All keys can't fit in memory. + Calculate how many keys + buffers we can keep in memory + */ + size_t maxbuffer_org; + do + { + maxbuffer_org= maxbuffer; + if (memavl < sizeof(BUFFPEK) * maxbuffer || + (keys= (memavl-sizeof(BUFFPEK)*maxbuffer)/ + (sort_length+sizeof(char*))) <= 1) + { + _ma_check_print_error(info->sort_info->param, + "aria_sort_buffer_size is too small. Current aria_sort_buffer_size: %llu rows: %llu sort_length: %u", + (ulonglong) sortbuff_size, (ulonglong) records, + sort_length); + my_errno= ENOMEM; + goto err; + } + if (keys < maxbuffer) + { + /* + There must be sufficient memory for at least one key per BUFFPEK, + otherwise repair by sort/parallel repair cannot operate. + */ + maxbuffer= (uint) keys; + break; + } + } + while ((maxbuffer= (size_t) (records/(keys-1)+1)) != maxbuffer_org); + } + + if ((sort_keys= ((uchar**) + my_malloc(PSI_INSTRUMENT_ME, (size_t) (keys*(sort_length+sizeof(char*))+ + HA_FT_MAXBYTELEN), + MYF(0))))) + { + if (my_init_dynamic_array(PSI_INSTRUMENT_ME, &buffpek, sizeof(BUFFPEK), + maxbuffer, MY_MIN(maxbuffer/2, 1000), MYF(0))) + { + my_free(sort_keys); + sort_keys= 0; + } + else + break; + } + old_memavl=memavl; + if ((memavl=memavl/4*3) < MARIA_MIN_SORT_MEMORY && old_memavl > MARIA_MIN_SORT_MEMORY) + memavl=MARIA_MIN_SORT_MEMORY; + } + if (memavl < MARIA_MIN_SORT_MEMORY) + { + /* purecov: begin inspected */ + _ma_check_print_error(info->sort_info->param, + "aria_sort_buffer_size is too small. Current aria_sort_buffer_size: %llu rows: %llu sort_length: %u", + (ulonglong) sortbuff_size, (ulonglong) records, sort_length); + my_errno= ENOMEM; + goto err; + /* purecov: end inspected */ + } + (*info->lock_in_memory)(info->sort_info->param);/* Everything is allocated */ + + if (!no_messages) + my_fprintf(stdout, + " - Searching for keys, allocating buffer for %llu keys\n", + (ulonglong) keys); + + if ((records=find_all_keys(info,keys,sort_keys,&buffpek,&maxbuffer, + &tempfile,&tempfile_for_exceptions)) + == HA_POS_ERROR) + goto err; /* purecov: tested */ + + if (maxbuffer >= keys) + { + /* + merge_many_buff will crash if maxbuffer > keys as then we cannot store in memory + the keys for each buffer. + */ + keys= maxbuffer + 1; + if (!(sort_keys= ((uchar **) + my_realloc(PSI_INSTRUMENT_ME, sort_keys, + (size_t) (keys*(sort_length+sizeof(char*))+ + HA_FT_MAXBYTELEN), MYF(MY_FREE_ON_ERROR))))) + goto err; + } + + info->sort_info->param->stage++; /* Merge stage */ + + if (maxbuffer == 0) + { + if (!no_messages) + my_fprintf(stdout, " - Dumping %llu keys\n", (ulonglong) records); + if (write_index(info, sort_keys, (ha_keys) records)) + goto err; /* purecov: inspected */ + } + else + { + keys=(keys*(sort_length+sizeof(char*)))/sort_length; + if (maxbuffer >= MERGEBUFF2) + { + if (!no_messages) + my_fprintf(stdout, " - Merging %llu keys\n", + (ulonglong) records); /* purecov: tested */ + if (merge_many_buff(info,keys,sort_keys, + dynamic_element(&buffpek,0,BUFFPEK *),&maxbuffer,&tempfile)) + goto err; /* purecov: inspected */ + } + if (flush_io_cache(&tempfile) || + reinit_io_cache(&tempfile,READ_CACHE,0L,0,0)) + goto err; /* purecov: inspected */ + if (!no_messages) + printf(" - Last merge and dumping keys\n"); /* purecov: tested */ + if (merge_index(info,keys,sort_keys,dynamic_element(&buffpek,0,BUFFPEK *), + maxbuffer,&tempfile)) + goto err; /* purecov: inspected */ + } + + if (flush_maria_ft_buf(info) || _ma_flush_pending_blocks(info)) + goto err; + + if (my_b_inited(&tempfile_for_exceptions)) + { + MARIA_HA *idx=info->sort_info->info; + uint16 key_length; + MARIA_KEY key; + key.keyinfo= idx->s->keyinfo + info->key; + + if (!no_messages) + printf(" - Adding exceptions\n"); /* purecov: tested */ + if (flush_io_cache(&tempfile_for_exceptions) || + reinit_io_cache(&tempfile_for_exceptions,READ_CACHE,0L,0,0)) + goto err; + + while (!my_b_read(&tempfile_for_exceptions,(uchar*)&key_length, + sizeof(key_length)) + && !my_b_read(&tempfile_for_exceptions,(uchar*)sort_keys, + (uint) key_length)) + { + key.data= (uchar*) sort_keys; + key.ref_length= idx->s->rec_reflength; + key.data_length= key_length - key.ref_length; + key.flag= 0; + if (_ma_ck_write(idx, &key)) + goto err; + } + } + + error =0; + +err: + my_free(sort_keys); + delete_dynamic(&buffpek); + close_cached_file(&tempfile); + close_cached_file(&tempfile_for_exceptions); + + DBUG_RETURN(error ? -1 : 0); +} /* _ma_create_index_by_sort */ + + +/* Search after all keys and place them in a temp. file */ + +static ha_rows find_all_keys(MARIA_SORT_PARAM *info, ha_rows keys, + uchar **sort_keys, DYNAMIC_ARRAY *buffpek, + size_t *maxbuffer, IO_CACHE *tempfile, + IO_CACHE *tempfile_for_exceptions) +{ + int error; + ha_rows idx; + DBUG_ENTER("find_all_keys"); + + idx=error=0; + sort_keys[0]= (uchar*) (sort_keys+keys); + + info->sort_info->info->in_check_table= 1; + 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)) + goto err; /* purecov: inspected */ + continue; + } + + if (++idx == keys) + { + if (info->write_keys(info,sort_keys,idx-1, + (BUFFPEK *)alloc_dynamic(buffpek), + tempfile)) + goto err; /* purecov: inspected */ + + sort_keys[0]=(uchar*) (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) + goto err; /* purecov: inspected */ + if (buffpek->elements) + { + if (info->write_keys(info,sort_keys,idx,(BUFFPEK *)alloc_dynamic(buffpek), + tempfile)) + goto err; /* purecov: inspected */ + *maxbuffer=buffpek->elements-1; + } + else + *maxbuffer=0; + + info->sort_info->info->in_check_table= 0; + DBUG_RETURN((*maxbuffer)*(keys-1)+idx); + +err: + info->sort_info->info->in_check_table= 0; /* purecov: inspected */ + DBUG_RETURN(HA_POS_ERROR); /* purecov: inspected */ +} /* find_all_keys */ + + +static my_bool _ma_thr_find_all_keys_exec(MARIA_SORT_PARAM* sort_param) +{ + int error= 0; + ulonglong memavl, old_memavl; + longlong sortbuff_size; + ha_keys UNINIT_VAR(keys), idx; + uint sort_length; + size_t maxbuffer; + uchar **sort_keys= NULL; + DBUG_ENTER("_ma_thr_find_all_keys_exec"); + DBUG_PRINT("enter", ("master: %d", sort_param->master)); + + if (sort_param->sort_info->got_error) + DBUG_RETURN(TRUE); + + set_sort_param_read_write(sort_param); + + my_b_clear(&sort_param->tempfile); + my_b_clear(&sort_param->tempfile_for_exceptions); + bzero((char*) &sort_param->buffpek, sizeof(sort_param->buffpek)); + bzero((char*) &sort_param->unique, sizeof(sort_param->unique)); + + sortbuff_size= sort_param->sortbuff_size; + memavl= MY_MAX(sortbuff_size, MARIA_MIN_SORT_MEMORY); + idx= (ha_keys) sort_param->sort_info->max_records; + sort_length= sort_param->key_length; + maxbuffer= 1; + + while (memavl >= MARIA_MIN_SORT_MEMORY) + { + if ((my_off_t) (idx+1)*(sort_length+sizeof(char*)) <= (my_off_t) memavl) + keys= idx+1; + else if ((sort_param->sort_info->param->testflag & + (T_FORCE_SORT_MEMORY | T_CREATE_MISSING_KEYS)) == + T_FORCE_SORT_MEMORY) + { + /* + Use all of the given sort buffer for key data. + Allocate 1000 buffers at a start for new data. More buffers + will be allocated when needed. + */ + keys= memavl / (sort_length+sizeof(char*)); + maxbuffer= (size_t) MY_MIN((ulonglong) 1000, (idx / keys)+1); + } + else + { + size_t maxbuffer_org; + do + { + maxbuffer_org= maxbuffer; + if (memavl < sizeof(BUFFPEK)*maxbuffer || + (keys=(memavl-sizeof(BUFFPEK)*maxbuffer)/ + (sort_length+sizeof(char*))) <= 1 || + keys < maxbuffer) + { + _ma_check_print_error(sort_param->sort_info->param, + "aria_sort_buffer_size is too small. Current aria_sort_buffer_size: %llu rows: %llu sort_length: %u", + sortbuff_size, (ulonglong) idx, sort_length); + goto err; + } + } + while ((maxbuffer= (uint) (idx/(keys-1)+1)) != maxbuffer_org); + } + if ((sort_keys= (uchar **) + my_malloc(PSI_INSTRUMENT_ME, (size_t)(keys*(sort_length+sizeof(char*))+ + ((sort_param->keyinfo->flag & HA_FULLTEXT) ? + HA_FT_MAXBYTELEN : 0)), MYF(0)))) + { + if (my_init_dynamic_array(PSI_INSTRUMENT_ME, &sort_param->buffpek, sizeof(BUFFPEK), + maxbuffer, MY_MIN(maxbuffer / 2, 1000), MYF(0))) + { + my_free(sort_keys); + sort_keys= NULL; /* Safety against double free on error. */ + } + else + break; + } + old_memavl= memavl; + if ((memavl= memavl/4*3) < MARIA_MIN_SORT_MEMORY && + old_memavl > MARIA_MIN_SORT_MEMORY) + memavl= MARIA_MIN_SORT_MEMORY; + } + if (memavl < MARIA_MIN_SORT_MEMORY) + { + /* purecov: begin inspected */ + _ma_check_print_error(sort_param->sort_info->param, + "aria_sort_buffer_size is too small. Current aria_sort_buffer_size: %llu rows: %llu sort_length: %u", + sortbuff_size, (ulonglong) idx, sort_length); + my_errno= ENOMEM; + goto err; + /* purecov: end inspected */ + } + + if (sort_param->sort_info->param->testflag & T_VERBOSE) + my_fprintf(stdout, + "Key %d - Allocating buffer for %llu keys\n", + sort_param->key + 1, (ulonglong) keys); + sort_param->sort_keys= sort_keys; + + idx= error= 0; + sort_keys[0]= (uchar*) (sort_keys+keys); + + DBUG_PRINT("info", ("reading keys")); + while (!(error= sort_param->sort_info->got_error) && + !(error= (*sort_param->key_read)(sort_param, sort_keys[idx]))) + { + if (sort_param->real_key_length > sort_param->key_length) + { + if (write_key(sort_param, sort_keys[idx], + &sort_param->tempfile_for_exceptions)) + goto err; + continue; + } + + if (++idx == keys) + { + if (sort_param->write_keys(sort_param, sort_keys, idx - 1, + (BUFFPEK *)alloc_dynamic(&sort_param->buffpek), + &sort_param->tempfile)) + goto err; + sort_keys[0]= (uchar*) (sort_keys+keys); + memcpy(sort_keys[0], sort_keys[idx - 1], (size_t) sort_param->key_length); + idx= 1; + } + sort_keys[idx]= sort_keys[idx - 1] + sort_param->key_length; + } + if (error > 0) + goto err; + if (sort_param->buffpek.elements) + { + if (sort_param->write_keys(sort_param,sort_keys, idx, + (BUFFPEK *) alloc_dynamic(&sort_param->buffpek), + &sort_param->tempfile)) + goto err; + sort_param->keys= (uint)((sort_param->buffpek.elements - 1) * (keys - 1) + idx); + } + else + sort_param->keys= (uint)idx; + + DBUG_RETURN(FALSE); + +err: + DBUG_PRINT("error", ("got some error")); + my_free(sort_keys); + sort_param->sort_keys= 0; + delete_dynamic(& sort_param->buffpek); + close_cached_file(&sort_param->tempfile); + close_cached_file(&sort_param->tempfile_for_exceptions); + + DBUG_RETURN(TRUE); +} + +/* Search after all keys and place them in a temp. file */ + +pthread_handler_t _ma_thr_find_all_keys(void *arg) +{ + MARIA_SORT_PARAM *sort_param= (MARIA_SORT_PARAM*) arg; + my_bool error= FALSE; + /* If my_thread_init fails */ + if (my_thread_init() || _ma_thr_find_all_keys_exec(sort_param)) + error= TRUE; + + /* + Thread must clean up after itself. + */ + free_root(&sort_param->wordroot, MYF(0)); + /* + Detach from the share if the writer is involved. Avoid others to + be blocked. This includes a flush of the write buffer. This will + also indicate EOF to the readers. + That means that a writer always gets here first and readers - + only when they see EOF. But if a reader finishes prematurely + because of an error it may reach this earlier - don't allow it + to detach the writer thread. + */ + if (sort_param->master && sort_param->sort_info->info->rec_cache.share) + remove_io_thread(&sort_param->sort_info->info->rec_cache); + + /* Readers detach from the share if any. Avoid others to be blocked. */ + if (sort_param->read_cache.share) + remove_io_thread(&sort_param->read_cache); + + mysql_mutex_lock(&sort_param->sort_info->mutex); + if (error) + sort_param->sort_info->got_error= 1; + + if (!--sort_param->sort_info->threads_running) + mysql_cond_signal(&sort_param->sort_info->cond); + mysql_mutex_unlock(&sort_param->sort_info->mutex); + + my_thread_end(); + return NULL; +} + + +int _ma_thr_write_keys(MARIA_SORT_PARAM *sort_param) +{ + MARIA_SORT_INFO *sort_info=sort_param->sort_info; + HA_CHECK *param=sort_info->param; + size_t UNINIT_VAR(length), keys; + double *rec_per_key_part= param->new_rec_per_key_part; + int got_error=sort_info->got_error; + uint i; + MARIA_HA *info=sort_info->info; + MARIA_SHARE *share= info->s; + MARIA_SORT_PARAM *sinfo; + uchar *mergebuf=0; + DBUG_ENTER("_ma_thr_write_keys"); + + for (i= 0, sinfo= sort_param ; + i < sort_info->total_keys ; + i++, sinfo++) + { + if (!sinfo->sort_keys) + { + got_error=1; + my_free(sinfo->rec_buff); + continue; + } + if (!got_error) + { + maria_set_key_active(share->state.key_map, sinfo->key); + + if (!sinfo->buffpek.elements) + { + if (param->testflag & T_VERBOSE) + { + my_fprintf(stdout, + "Key %d - Dumping %llu keys\n", sinfo->key+1, + (ulonglong) sinfo->keys); + fflush(stdout); + } + if (write_index(sinfo, sinfo->sort_keys, sinfo->keys) || + flush_maria_ft_buf(sinfo) || _ma_flush_pending_blocks(sinfo)) + got_error=1; + } + } + my_free(sinfo->sort_keys); + my_free(sinfo->rec_buff); + 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), + rec_per_key_part+= sinfo->keyinfo->keysegs, + sinfo++) + { + if (got_error) + continue; + + set_sort_param_read_write(sinfo); + + if (sinfo->buffpek.elements) + { + size_t maxbuffer=sinfo->buffpek.elements-1; + if (!mergebuf) + { + length=(size_t)param->sort_buffer_length; + while (length >= MARIA_MIN_SORT_MEMORY) + { + if ((mergebuf= my_malloc(PSI_INSTRUMENT_ME, (size_t) length, MYF(0)))) + break; + length=length*3/4; + } + if (!mergebuf) + { + got_error=1; + continue; + } + } + keys=length/sinfo->key_length; + if (maxbuffer >= MERGEBUFF2) + { + if (param->testflag & T_VERBOSE) + my_fprintf(stdout, + "Key %d - Merging %llu keys\n", + sinfo->key+1, (ulonglong) sinfo->keys); + if (merge_many_buff(sinfo, keys, (uchar **)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,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, (uchar**) mergebuf, + dynamic_element(&sinfo->buffpek,0,BUFFPEK *), + maxbuffer,&sinfo->tempfile) || + flush_maria_ft_buf(sinfo) || + _ma_flush_pending_blocks(sinfo)) + { + got_error=1; + continue; + } + } + if (my_b_inited(&sinfo->tempfile_for_exceptions)) + { + uint16 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,READ_CACHE,0L,0,0)) + { + got_error=1; + continue; + } + + while (!got_error && + !my_b_read(&sinfo->tempfile_for_exceptions,(uchar*)&key_length, + sizeof(key_length))) + { + uchar maria_ft_buf[HA_FT_MAXBYTELEN + HA_FT_WLEN + 10]; + if (key_length > sizeof(maria_ft_buf) || + my_b_read(&sinfo->tempfile_for_exceptions, (uchar*)maria_ft_buf, + (uint) key_length)) + got_error= 1; + else + { + MARIA_KEY tmp_key; + tmp_key.keyinfo= info->s->keyinfo + sinfo->key; + tmp_key.data= maria_ft_buf; + tmp_key.ref_length= info->s->rec_reflength; + tmp_key.data_length= key_length - info->s->rec_reflength; + tmp_key.flag= 0; + if (_ma_ck_write(info, &tmp_key)) + got_error=1; + } + } + } + if (!got_error && (param->testflag & T_STATISTICS)) + maria_update_key_parts(sinfo->keyinfo, rec_per_key_part, sinfo->unique, + param->stats_method == + MI_STATS_METHOD_IGNORE_NULLS ? + sinfo->notnull : NULL, + (ulonglong) share->state.state.records); + + } + my_free(mergebuf); + DBUG_RETURN(got_error); +} + + +/* Write all keys in memory to file for later merge */ + +static int write_keys(MARIA_SORT_PARAM *info, register uchar **sort_keys, + ha_keys count, BUFFPEK *buffpek, IO_CACHE *tempfile) +{ + uchar **end; + uint sort_length=info->key_length; + DBUG_ENTER("write_keys"); + + if (!buffpek) + DBUG_RETURN(1); /* Out of memory */ + + my_qsort2((uchar*) sort_keys,(size_t) count, sizeof(uchar*), + (qsort2_cmp) info->key_cmp, info); + if (!my_b_inited(tempfile) && + open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST", + DISK_BUFFER_SIZE, info->sort_info->param->myf_rw)) + DBUG_RETURN(1); /* purecov: inspected */ + + 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, *sort_keys, sort_length)) + DBUG_RETURN(1); /* purecov: inspected */ + } + DBUG_RETURN(0); +} /* write_keys */ + + +static inline int +my_var_write(MARIA_SORT_PARAM *info, IO_CACHE *to_file, uchar *bufs) +{ + int err; + uint16 len= _ma_keylength(info->keyinfo, bufs); + + /* The following is safe as this is a local file */ + if ((err= my_b_write(to_file, (uchar*)&len, sizeof(len)))) + return (err); + if ((err= my_b_write(to_file,bufs, (uint) len))) + return (err); + return (0); +} + + +static int write_keys_varlen(MARIA_SORT_PARAM *info, + register uchar **sort_keys, + ha_keys count, BUFFPEK *buffpek, + IO_CACHE *tempfile) +{ + uchar **end; + int err; + DBUG_ENTER("write_keys_varlen"); + + if (!buffpek) + DBUG_RETURN(1); /* Out of memory */ + + my_qsort2((uchar*) sort_keys, (size_t) count, sizeof(uchar*), + (qsort2_cmp) info->key_cmp, info); + if (!my_b_inited(tempfile) && + open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST", + DISK_BUFFER_SIZE, info->sort_info->param->myf_rw)) + DBUG_RETURN(1); /* purecov: inspected */ + + 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, *sort_keys))) + DBUG_RETURN(err); + } + DBUG_RETURN(0); +} /* write_keys_varlen */ + + +static int write_key(MARIA_SORT_PARAM *info, uchar *key, + IO_CACHE *tempfile) +{ + uint16 key_length=info->real_key_length; + DBUG_ENTER("write_key"); + + if (!my_b_inited(tempfile) && + open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST", + DISK_BUFFER_SIZE, info->sort_info->param->myf_rw)) + DBUG_RETURN(1); + + if (my_b_write(tempfile, (uchar*)&key_length,sizeof(key_length)) || + my_b_write(tempfile, key, (uint) key_length)) + DBUG_RETURN(1); + DBUG_RETURN(0); +} /* write_key */ + + +/* Write index */ + +static int write_index(MARIA_SORT_PARAM *info, register uchar **sort_keys, + register ha_keys count) +{ + DBUG_ENTER("write_index"); + + my_qsort2((uchar*) sort_keys,(size_t) count,sizeof(uchar*), + (qsort2_cmp) info->key_cmp,info); + while (count--) + { + if ((*info->key_write)(info, *sort_keys++)) + DBUG_RETURN(-1); /* purecov: inspected */ + } + if (info->sort_info->param->max_stage != 1) /* If not parallel */ + _ma_report_progress(info->sort_info->param, 1, 1); + DBUG_RETURN(0); +} /* write_index */ + + + /* Merge buffers to make < MERGEBUFF2 buffers */ + +static int merge_many_buff(MARIA_SORT_PARAM *info, ha_keys keys, + uchar **sort_keys, BUFFPEK *buffpek, + size_t *maxbuffer, IO_CACHE *t_file) +{ + size_t tmp, merges, max_merges; + IO_CACHE t_file2, *from_file, *to_file, *temp; + BUFFPEK *lastbuff; + DBUG_ENTER("merge_many_buff"); + + if (*maxbuffer < MERGEBUFF2) + DBUG_RETURN(0); /* purecov: inspected */ + if (flush_io_cache(t_file) || + open_cached_file(&t_file2,my_tmpdir(info->tmpdir),"ST", + DISK_BUFFER_SIZE, info->sort_info->param->myf_rw)) + DBUG_RETURN(1); /* purecov: inspected */ + + /* Calculate how many merges are needed */ + max_merges= 1; /* Count merge_index */ + tmp= *maxbuffer; + while (tmp >= MERGEBUFF2) + { + merges= (tmp-MERGEBUFF*3/2 + 1) / MERGEBUFF + 1; + max_merges+= merges; + tmp= merges; + } + merges= 0; + + from_file= t_file ; to_file= &t_file2; + while (*maxbuffer >= MERGEBUFF2) + { + size_t i; + reinit_io_cache(from_file,READ_CACHE,0L,0,0); + reinit_io_cache(to_file,WRITE_CACHE,0L,0,0); + lastbuff=buffpek; + for (i=0 ; i + MERGEBUFF*3/2 <= *maxbuffer ; i+=MERGEBUFF) + { + if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++, + buffpek+i,buffpek+i+MERGEBUFF-1)) + goto cleanup; + if (info->sort_info->param->max_stage != 1) /* If not parallel */ + _ma_report_progress(info->sort_info->param, merges++, max_merges); + } + if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++, + buffpek+i,buffpek+ *maxbuffer)) + break; /* purecov: inspected */ + if (flush_io_cache(to_file)) + break; /* purecov: inspected */ + temp=from_file; from_file=to_file; to_file=temp; + *maxbuffer= (size_t) (lastbuff-buffpek)-1; + if (info->sort_info->param->max_stage != 1) /* If not parallel */ + _ma_report_progress(info->sort_info->param, merges++, max_merges); + } +cleanup: + close_cached_file(to_file); /* This holds old result */ + if (to_file == t_file) + { + DBUG_ASSERT(t_file2.type == WRITE_CACHE); + *t_file=t_file2; /* Copy result file */ + } + + DBUG_RETURN(*maxbuffer >= MERGEBUFF2); /* Return 1 if interrupted */ +} /* merge_many_buff */ + + +/* + Read data to buffer + + SYNOPSIS + read_to_buffer() + fromfile File to read from + buffpek Where to read from + sort_length max length to read + RESULT + > 0 Ammount of bytes read + -1 Error +*/ + +static my_off_t read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek, + uint sort_length) +{ + register ha_keys count; + size_t length; + + if ((count= (ha_keys) MY_MIN((ha_rows) buffpek->max_keys, + (ha_rows) buffpek->count))) + { + if (my_b_pread(fromfile, (uchar*) buffpek->base, + (length= sort_length * (size_t)count), buffpek->file_pos)) + return(HA_OFFSET_ERROR); /* purecov: inspected */ + buffpek->key=buffpek->base; + buffpek->file_pos+= length; /* New filepos */ + buffpek->count-= count; + buffpek->mem_count= count; + } + return (((my_off_t) count) * sort_length); +} /* read_to_buffer */ + + +static my_off_t read_to_buffer_varlen(IO_CACHE *fromfile, BUFFPEK *buffpek, + uint sort_length) +{ + register ha_keys count; + uint idx; + uchar *buffp; + + if ((count= (ha_keys) MY_MIN((ha_rows) buffpek->max_keys,buffpek->count))) + { + buffp= buffpek->base; + + for (idx=1;idx<=count;idx++) + { + uint16 length_of_key; + if (my_b_pread(fromfile, (uchar*)&length_of_key, + sizeof(length_of_key), buffpek->file_pos)) + return(HA_OFFSET_ERROR); + buffpek->file_pos+=sizeof(length_of_key); + if (my_b_pread(fromfile, (uchar*) buffp, + length_of_key, buffpek->file_pos)) + return((uint) -1); + buffpek->file_pos+=length_of_key; + buffp = buffp + sort_length; + } + buffpek->key=buffpek->base; + buffpek->count-= count; + buffpek->mem_count= count; + } + return (((my_off_t) count) * sort_length); +} /* read_to_buffer_varlen */ + + +static int write_merge_key_varlen(MARIA_SORT_PARAM *info, + IO_CACHE *to_file, uchar* key, + uint sort_length, ha_keys count) +{ + ha_keys idx; + uchar *bufs = key; + + for (idx=1;idx<=count;idx++) + { + int err; + if ((err= my_var_write(info, to_file, bufs))) + return (err); + bufs=bufs+sort_length; + } + return(0); +} + + +static int write_merge_key(MARIA_SORT_PARAM *info __attribute__((unused)), + IO_CACHE *to_file, uchar *key, + uint sort_length, ha_keys count) +{ + return my_b_write(to_file, key, (size_t) (sort_length * count)); +} + +/* + Merge buffers to one buffer + If to_file == 0 then use info->key_write + + Return: + 0 ok + 1 error +*/ + +static int +merge_buffers(MARIA_SORT_PARAM *info, ha_keys keys, IO_CACHE *from_file, + IO_CACHE *to_file, uchar **sort_keys, BUFFPEK *lastbuff, + BUFFPEK *Fb, BUFFPEK *Tb) +{ + int error= 1; + uint sort_length; + ha_keys maxcount; + ha_rows count; + my_off_t UNINIT_VAR(to_start_filepos), read_length; + uchar *strpos; + BUFFPEK *buffpek,**refpek; + QUEUE queue; + DBUG_ENTER("merge_buffers"); + + count= 0; + maxcount= keys/((uint) (Tb-Fb) +1); + DBUG_ASSERT(maxcount > 0); + if (to_file) + to_start_filepos=my_b_tell(to_file); + strpos= (uchar*) sort_keys; + sort_length=info->key_length; + + if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0, + (int (*)(void*, uchar *,uchar*)) info->key_cmp, + (void*) info, 0, 0)) + DBUG_RETURN(1); /* purecov: inspected */ + + for (buffpek= Fb ; buffpek <= Tb ; buffpek++) + { + count+= buffpek->count; + buffpek->base= strpos; + buffpek->max_keys= maxcount; + strpos+= (read_length= info->read_to_buffer(from_file,buffpek, + sort_length)); + if (read_length == HA_OFFSET_ERROR) + goto err; /* purecov: inspected */ + queue_insert(&queue,(uchar*) buffpek); + } + + while (queue.elements > 1) + { + for (;;) + { + buffpek=(BUFFPEK*) queue_top(&queue); + if (to_file) + { + if (info->write_key(info,to_file, buffpek->key, + sort_length, 1)) + goto err; /* purecov: inspected */ + } + else + { + if ((*info->key_write)(info,(void*) buffpek->key)) + goto err; /* purecov: inspected */ + } + buffpek->key+=sort_length; + if (! --buffpek->mem_count) + { + /* It's enough to check for killedptr before a slow operation */ + if (_ma_killed_ptr(info->sort_info->param)) + goto err; + if (!(read_length= info->read_to_buffer(from_file,buffpek,sort_length))) + { + uchar *base= buffpek->base; + ha_keys max_keys=buffpek->max_keys; + + queue_remove_top(&queue); + + /* Put room used by buffer to use in other buffer */ + for (refpek= (BUFFPEK**) &queue_top(&queue); + refpek <= (BUFFPEK**) &queue_end(&queue); + refpek++) + { + buffpek= *refpek; + if (buffpek->base+buffpek->max_keys*sort_length == base) + { + buffpek->max_keys+=max_keys; + break; + } + else if (base+max_keys*sort_length == buffpek->base) + { + buffpek->base=base; + buffpek->max_keys+=max_keys; + break; + } + } + break; /* One buffer have been removed */ + } + else if (read_length == HA_OFFSET_ERROR) + goto err; /* purecov: inspected */ + } + queue_replace_top(&queue); /* Top element has been replaced */ + } + } + buffpek=(BUFFPEK*) queue_top(&queue); + buffpek->base= (uchar*) sort_keys; + buffpek->max_keys=keys; + do + { + if (to_file) + { + if (info->write_key(info, to_file, buffpek->key, + sort_length,buffpek->mem_count)) + { + error=1; goto err; /* purecov: inspected */ + } + } + else + { + register uchar *end; + strpos= buffpek->key; + for (end= strpos+buffpek->mem_count*sort_length; + strpos != end ; + strpos+=sort_length) + { + if ((*info->key_write)(info, strpos)) + { + error=1; goto err; /* purecov: inspected */ + } + } + } + } + while ((read_length= info->read_to_buffer(from_file,buffpek,sort_length)) != HA_OFFSET_ERROR && read_length != 0); + if (read_length == 0) + error= 0; + + lastbuff->count=count; + if (to_file) + lastbuff->file_pos=to_start_filepos; +err: + delete_queue(&queue); + DBUG_RETURN(error); +} /* merge_buffers */ + + + /* Do a merge to output-file (save only positions) */ + +static int +merge_index(MARIA_SORT_PARAM *info, ha_keys keys, uchar **sort_keys, + BUFFPEK *buffpek, size_t maxbuffer, IO_CACHE *tempfile) +{ + DBUG_ENTER("merge_index"); + if (merge_buffers(info,keys,tempfile,(IO_CACHE*) 0,sort_keys,buffpek,buffpek, + buffpek+maxbuffer)) + DBUG_RETURN(1); /* purecov: inspected */ + if (info->sort_info->param->max_stage != 1) /* If not parallel */ + _ma_report_progress(info->sort_info->param, 1, 1); + DBUG_RETURN(0); +} /* merge_index */ + + +static int flush_maria_ft_buf(MARIA_SORT_PARAM *info) +{ + int err=0; + if (info->sort_info->ft_buf) + { + err=_ma_sort_ft_buf_flush(info); + my_free(info->sort_info->ft_buf); + info->sort_info->ft_buf=0; + } + return err; +} |