sort.c 29.7 KB
Newer Older
bk@work.mysql.com's avatar
bk@work.mysql.com committed
1
/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
2

bk@work.mysql.com's avatar
bk@work.mysql.com committed
3 4 5 6
   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; either version 2 of the License, or
   (at your option) any later version.
7

bk@work.mysql.com's avatar
bk@work.mysql.com committed
8 9 10 11
   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.
12

bk@work.mysql.com's avatar
bk@work.mysql.com committed
13 14 15 16 17 18 19 20 21
   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.
*/

22
#include "fulltext.h"
bk@work.mysql.com's avatar
bk@work.mysql.com committed
23 24 25 26 27 28 29
#if defined(MSDOS) || defined(__WIN__)
#include <fcntl.h>
#else
#include <stddef.h>
#endif
#include <queues.h>

30 31
/* static variables */

monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
32 33 34
#undef MIN_SORT_MEMORY
#undef MYF_RW
#undef DISK_BUFFER_SIZE
bk@work.mysql.com's avatar
bk@work.mysql.com committed
35 36 37 38

#define MERGEBUFF 15
#define MERGEBUFF2 31
#define MIN_SORT_MEMORY (4096-MALLOC_OVERHEAD)
39
#define MYF_RW  MYF(MY_NABP | MY_WME | MY_WAIT_IF_FULL)
40
#define DISK_BUFFER_SIZE (IO_SIZE*16)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
41

42 43 44 45

/*
 Pointers of functions for store and read keys from temp file
*/
bk@work.mysql.com's avatar
bk@work.mysql.com committed
46 47 48

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

49
/* Functions defined in this file */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
50 51

static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info,uint keys,
52 53 54 55
                                    uchar **sort_keys,
                                    DYNAMIC_ARRAY *buffpek,int *maxbuffer,
                                    IO_CACHE *tempfile,
                                    IO_CACHE *tempfile_for_exceptions);
56
static int NEAR_F write_keys(MI_SORT_PARAM *info,uchar **sort_keys,
57
                             uint count, BUFFPEK *buffpek,IO_CACHE *tempfile);
58 59
static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key,
			    IO_CACHE *tempfile);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
60
static int NEAR_F write_index(MI_SORT_PARAM *info,uchar * *sort_keys,
61
                              uint count);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
62
static int NEAR_F merge_many_buff(MI_SORT_PARAM *info,uint keys,
63 64 65
                                  uchar * *sort_keys,
                                  BUFFPEK *buffpek,int *maxbuffer,
                                  IO_CACHE *t_file);
66
static uint NEAR_F read_to_buffer(IO_CACHE *fromfile,BUFFPEK *buffpek,
67
                                  uint sort_length);
68
static int NEAR_F merge_buffers(MI_SORT_PARAM *info,uint keys,
69 70 71
                                IO_CACHE *from_file, IO_CACHE *to_file,
                                uchar * *sort_keys, BUFFPEK *lastbuff,
                                BUFFPEK *Fb, BUFFPEK *Tb);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
72
static int NEAR_F merge_index(MI_SORT_PARAM *,uint,uchar **,BUFFPEK *, int,
73
                              IO_CACHE *);
74 75
static int flush_ft_buf(MI_SORT_PARAM *info);

76
static int NEAR_F write_keys_varlen(MI_SORT_PARAM *info,uchar **sort_keys,
77 78
                                    uint count, BUFFPEK *buffpek,
                                    IO_CACHE *tempfile);
79 80
static uint NEAR_F read_to_buffer_varlen(IO_CACHE *fromfile,BUFFPEK *buffpek,
                                         uint sort_length);
81
static int NEAR_F write_merge_key(MI_SORT_PARAM *info, IO_CACHE *to_file,
82 83 84 85 86
                                  char *key, uint sort_length, uint count);
static int NEAR_F write_merge_key_varlen(MI_SORT_PARAM *info,
					 IO_CACHE *to_file,
					 char* key, uint sort_length,
					 uint count);
87 88 89
static inline int
my_var_write(MI_SORT_PARAM *info, IO_CACHE *to_file, byte *bufs);

90 91 92 93 94 95 96 97 98 99 100 101 102
/*
  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
*/
bk@work.mysql.com's avatar
bk@work.mysql.com committed
103 104 105 106 107 108

int _create_index_by_sort(MI_SORT_PARAM *info,my_bool no_messages,
			  ulong sortbuff_size)
{
  int error,maxbuffer,skr;
  uint memavl,old_memavl,keys,sort_length;
109
  DYNAMIC_ARRAY buffpek;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
110 111
  ha_rows records;
  uchar **sort_keys;
112
  IO_CACHE tempfile, tempfile_for_exceptions;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
113 114 115
  DBUG_ENTER("_create_index_by_sort");
  DBUG_PRINT("enter",("sort_length: %d", info->key_length));

116
  if (info->keyinfo->flag & HA_VAR_LENGTH_KEY)
117
  {
118 119 120
    info->write_keys=write_keys_varlen;
    info->read_to_buffer=read_to_buffer_varlen;
    info->write_key=write_merge_key_varlen;
121 122 123
  }
  else
  {
124 125 126
    info->write_keys=write_keys;
    info->read_to_buffer=read_to_buffer;
    info->write_key=write_merge_key;
127
  }
128

129
  my_b_clear(&tempfile);
130 131 132
  my_b_clear(&tempfile_for_exceptions);
  bzero((char*) &buffpek,sizeof(buffpek));
  sort_keys= (uchar **) NULL; error= 1;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
133 134 135
  maxbuffer=1;

  memavl=max(sortbuff_size,MIN_SORT_MEMORY);
136
  records=	info->sort_info->max_records;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159
  sort_length=	info->key_length;
  LINT_INIT(keys);

  while (memavl >= MIN_SORT_MEMORY)
  {
    if ((my_off_t) (records+1)*(sort_length+sizeof(char*)) <=
	(my_off_t) memavl)
      keys= records+1;
    else
      do
      {
	skr=maxbuffer;
	if (memavl < sizeof(BUFFPEK)*(uint) maxbuffer ||
	    (keys=(memavl-sizeof(BUFFPEK)*(uint) maxbuffer)/
	     (sort_length+sizeof(char*))) <= 1)
	{
	  mi_check_print_error(info->sort_info->param,
			       "sort_buffer_size is to small");
	  goto err;
	}
      }
      while ((maxbuffer= (int) (records/(keys-1)+1)) != skr);

160
    if ((sort_keys=(uchar **)my_malloc(keys*(sort_length+sizeof(char*))+
161
				       HA_FT_MAXBYTELEN, MYF(0))))
bk@work.mysql.com's avatar
bk@work.mysql.com committed
162
    {
163
      if (my_init_dynamic_array(&buffpek, sizeof(BUFFPEK), maxbuffer,
164
			     maxbuffer/2))
165
      {
bk@work.mysql.com's avatar
bk@work.mysql.com committed
166
	my_free((gptr) sort_keys,MYF(0));
167 168
        sort_keys= 0;
      }
169 170
      else
	break;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
171 172 173 174 175 176 177
    }
    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)
  {
178 179
    mi_check_print_error(info->sort_info->param,"Sort buffer to small"); /* purecov: tested */
    goto err; /* purecov: tested */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
180 181 182 183 184 185
  }
  (*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);

186 187
  if ((records=find_all_keys(info,keys,sort_keys,&buffpek,&maxbuffer,
                                  &tempfile,&tempfile_for_exceptions))
bk@work.mysql.com's avatar
bk@work.mysql.com committed
188
      == HA_POS_ERROR)
189
    goto err; /* purecov: tested */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
190 191 192
  if (maxbuffer == 0)
  {
    if (!no_messages)
193 194
      printf("  - Dumping %lu keys\n", (ulong) records);
    if (write_index(info,sort_keys, (uint) records))
195
      goto err; /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
196 197 198 199 200 201 202
  }
  else
  {
    keys=(keys*(sort_length+sizeof(char*)))/sort_length;
    if (maxbuffer >= MERGEBUFF2)
    {
      if (!no_messages)
203
	printf("  - Merging %lu keys\n", (ulong) records); /* purecov: tested */
204 205
      if (merge_many_buff(info,keys,sort_keys,
                  dynamic_element(&buffpek,0,BUFFPEK *),&maxbuffer,&tempfile))
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
206
	goto err;				/* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
207
    }
208 209
    if (flush_io_cache(&tempfile) ||
	reinit_io_cache(&tempfile,READ_CACHE,0L,0,0))
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
210
      goto err;					/* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
211
    if (!no_messages)
212
      printf("  - Last merge and dumping keys\n"); /* purecov: tested */
213 214
    if (merge_index(info,keys,sort_keys,dynamic_element(&buffpek,0,BUFFPEK *),
                    maxbuffer,&tempfile))
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
215
      goto err;					/* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
216
  }
217

218
  if (flush_ft_buf(info) || flush_pending_blocks(info))
219 220 221 222 223
    goto err;

  if (my_b_inited(&tempfile_for_exceptions))
  {
    MI_INFO *index=info->sort_info->info;
224
    uint     keyno=info->key;
225 226
    uint     key_length, ref_length=index->s->rec_reflength;

227 228
    if (!no_messages)
      printf("  - Adding exceptions\n"); /* purecov: tested */
229 230 231 232
    if (flush_io_cache(&tempfile_for_exceptions) ||
	reinit_io_cache(&tempfile_for_exceptions,READ_CACHE,0L,0,0))
      goto err;

monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
233 234 235 236
    while (!my_b_read(&tempfile_for_exceptions,(byte*)&key_length,
		      sizeof(key_length))
        && !my_b_read(&tempfile_for_exceptions,(byte*)sort_keys,
		      (uint) key_length))
237
    {
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
238 239
	if (_mi_ck_write(index,keyno,(uchar*) sort_keys,key_length-ref_length))
	  goto err;
240 241 242
    }
  }

bk@work.mysql.com's avatar
bk@work.mysql.com committed
243 244 245 246 247
  error =0;

err:
  if (sort_keys)
    my_free((gptr) sort_keys,MYF(0));
248
  delete_dynamic(&buffpek);
249
  close_cached_file(&tempfile);
250
  close_cached_file(&tempfile_for_exceptions);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
251 252 253 254 255

  DBUG_RETURN(error ? -1 : 0);
} /* _create_index_by_sort */


256
/* Search after all keys and place them in a temp. file */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
257 258

static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info, uint keys,
259 260 261
				    uchar **sort_keys, DYNAMIC_ARRAY *buffpek,
				    int *maxbuffer, IO_CACHE *tempfile,
				    IO_CACHE *tempfile_for_exceptions)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
262 263
{
  int error;
264
  uint idx;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
265 266
  DBUG_ENTER("find_all_keys");

267
  idx=error=0;
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
268
  sort_keys[0]=(uchar*) (sort_keys+keys);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
269

270
  while (!(error=(*info->key_read)(info,sort_keys[idx])))
bk@work.mysql.com's avatar
bk@work.mysql.com committed
271
  {
272
    if (info->real_key_length > info->key_length)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
273
    {
274
      if (write_key(info,sort_keys[idx],tempfile_for_exceptions))
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
275
        DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
276 277 278 279 280
      continue;
    }

    if (++idx == keys)
    {
281
      if (info->write_keys(info,sort_keys,idx-1,(BUFFPEK *)alloc_dynamic(buffpek),
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
282 283
		     tempfile))
      DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
284

monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
285
      sort_keys[0]=(uchar*) (sort_keys+keys);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
286
      memcpy(sort_keys[0],sort_keys[idx-1],(size_t) info->key_length);
287
      idx=1;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
288
    }
289
    sort_keys[idx]=sort_keys[idx-1]+info->key_length;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
290 291
  }
  if (error > 0)
292
    DBUG_RETURN(HA_POS_ERROR);		/* Aborted by get_key */ /* purecov: inspected */
293 294
  if (buffpek->elements)
  {
295
    if (info->write_keys(info,sort_keys,idx,(BUFFPEK *)alloc_dynamic(buffpek),
296
		   tempfile))
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
297
      DBUG_RETURN(HA_POS_ERROR);		/* purecov: inspected */
298 299 300 301
    *maxbuffer=buffpek->elements-1;
  }
  else
    *maxbuffer=0;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
302

303 304
  DBUG_RETURN((*maxbuffer)*(keys-1)+idx);
} /* find_all_keys */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
305

306

307
#ifdef THREAD
308
/* Search after all keys and place them in a temp. file */
309

310
pthread_handler_t thr_find_all_keys(void *arg)
311
{
312
  MI_SORT_PARAM *info= (MI_SORT_PARAM*) arg;
313
  int error;
314 315
  uint memavl,old_memavl,keys,sort_length;
  uint idx, maxbuffer;
316
  uchar **sort_keys=0;
serg@serg.mysql.com's avatar
serg@serg.mysql.com committed
317

318 319
  LINT_INIT(keys);

320 321 322 323 324 325
  error=1;

  if (my_thread_init())
    goto err;
  if (info->sort_info->got_error)
    goto err;
326

327
  if (info->keyinfo->flag & HA_VAR_LENGTH_KEY)
328
  {
329 330 331
    info->write_keys=write_keys_varlen;
    info->read_to_buffer=read_to_buffer_varlen;
    info->write_key=write_merge_key_varlen;
332 333 334
  }
  else
  {
335 336 337
    info->write_keys=write_keys;
    info->read_to_buffer=read_to_buffer;
    info->write_key=write_merge_key;
338 339
  }

340 341 342 343
  my_b_clear(&info->tempfile);
  my_b_clear(&info->tempfile_for_exceptions);
  bzero((char*) &info->buffpek,sizeof(info->buffpek));
  bzero((char*) &info->unique, sizeof(info->unique));
344
  sort_keys= (uchar **) NULL;
345 346 347 348

  memavl=max(info->sortbuff_size, MIN_SORT_MEMORY);
  idx=      info->sort_info->max_records;
  sort_length=  info->key_length;
349
  maxbuffer= 1;
350 351 352 353 354 355 356

  while (memavl >= MIN_SORT_MEMORY)
  {
    if ((my_off_t) (idx+1)*(sort_length+sizeof(char*)) <=
        (my_off_t) memavl)
      keys= idx+1;
    else
357 358
    {
      uint skr;
359 360 361 362 363 364 365 366 367 368 369 370 371
      do
      {
        skr=maxbuffer;
        if (memavl < sizeof(BUFFPEK)*maxbuffer ||
            (keys=(memavl-sizeof(BUFFPEK)*maxbuffer)/
             (sort_length+sizeof(char*))) <= 1)
        {
          mi_check_print_error(info->sort_info->param,
                               "sort_buffer_size is to small");
          goto err;
        }
      }
      while ((maxbuffer= (int) (idx/(keys-1)+1)) != skr);
372
    }
373
    if ((sort_keys=(uchar **)my_malloc(keys*(sort_length+sizeof(char*))+
374
				       ((info->keyinfo->flag & HA_FULLTEXT) ?
375
					HA_FT_MAXBYTELEN : 0), MYF(0))))
376 377
    {
      if (my_init_dynamic_array(&info->buffpek, sizeof(BUFFPEK),
378
				maxbuffer, maxbuffer/2))
379
      {
380
        my_free((gptr) sort_keys,MYF(0));
381 382
        sort_keys= (uchar **) NULL; /* for err: label */
      }
383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402
      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,"Sort buffer to small"); /* purecov: tested */
    goto err; /* purecov: tested */
  }

  if (info->sort_info->param->testflag & T_VERBOSE)
    printf("Key %d - Allocating buffer for %d keys\n",info->key+1,keys);
  info->sort_keys=sort_keys;

  idx=error=0;
  sort_keys[0]=(uchar*) (sort_keys+keys);

serg@serg.mysql.com's avatar
serg@serg.mysql.com committed
403 404
  while (!(error=info->sort_info->got_error) &&
         !(error=(*info->key_read)(info,sort_keys[idx])))
405 406 407
  {
    if (info->real_key_length > info->key_length)
    {
408
      if (write_key(info,sort_keys[idx], &info->tempfile_for_exceptions))
409 410 411 412 413 414
        goto err;
      continue;
    }

    if (++idx == keys)
    {
415
      if (info->write_keys(info,sort_keys,idx-1,
416 417
		     (BUFFPEK *)alloc_dynamic(&info->buffpek),
		     &info->tempfile))
418 419 420 421 422 423 424 425 426 427 428
        goto err;
      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;
  if (info->buffpek.elements)
  {
429
    if (info->write_keys(info,sort_keys, idx,
430
		   (BUFFPEK *) alloc_dynamic(&info->buffpek), &info->tempfile))
431 432 433 434 435 436 437 438 439 440 441 442 443 444 445
      goto err;
    info->keys=(info->buffpek.elements-1)*(keys-1)+idx;
  }
  else
    info->keys=idx;

  info->sort_keys_length=keys;
  goto ok;

err:
  info->sort_info->got_error=1; /* no need to protect this with a mutex */
  if (sort_keys)
    my_free((gptr) sort_keys,MYF(0));
  info->sort_keys=0;
  delete_dynamic(& info->buffpek);
446 447 448
  close_cached_file(&info->tempfile);
  close_cached_file(&info->tempfile_for_exceptions);

449
ok:
450 451
  remove_io_thread(&info->read_cache);
  pthread_mutex_lock(&info->sort_info->mutex);
452
  info->sort_info->threads_running--;
453 454
  pthread_cond_signal(&info->sort_info->cond);
  pthread_mutex_unlock(&info->sort_info->mutex);
455
  my_thread_end();
456
  return NULL;
457
}
458

459

460
int thr_write_keys(MI_SORT_PARAM *sort_param)
461 462 463 464 465
{
  SORT_INFO *sort_info=sort_param->sort_info;
  MI_CHECK *param=sort_info->param;
  ulong length, keys;
  ulong *rec_per_key_part=param->rec_per_key_part;
466 467
  int got_error=sort_info->got_error;
  uint i;
468
  MI_INFO *info=sort_info->info;
469
  MYISAM_SHARE *share=info->s;
470 471
  MI_SORT_PARAM *sinfo;
  byte *mergebuf=0;
472
  LINT_INIT(length);
473

474 475 476
  for (i= 0, sinfo= sort_param ;
       i < sort_info->total_keys ;
       i++, rec_per_key_part+=sinfo->keyinfo->keysegs, sinfo++)
477 478 479 480 481 482
  {
    if (!sinfo->sort_keys)
    {
      got_error=1;
      continue;
    }
483
    if (!got_error)
484
    {
485
      mi_set_key_active(share->state.key_map, sinfo->key);
486
      if (!sinfo->buffpek.elements)
487
      {
488 489 490 491 492 493
        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) ||
494
            flush_ft_buf(sinfo) || flush_pending_blocks(sinfo))
495
          got_error=1;
496
      }
497 498 499 500 501
      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,
                         (ulonglong) info->state->records);
502 503
    }
    my_free((gptr) sinfo->sort_keys,MYF(0));
504 505
    my_free(mi_get_rec_buff_ptr(info, sinfo->rec_buff),
	    MYF(MY_ALLOW_ZERO_PTR));
506 507 508
    sinfo->sort_keys=0;
  }

509 510 511 512 513 514 515
  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++)
516
  {
517 518
    if (got_error)
      continue;
519
    if (sinfo->keyinfo->flag & HA_VAR_LENGTH_KEY)
520
    {
521 522 523
      sinfo->write_keys=write_keys_varlen;
      sinfo->read_to_buffer=read_to_buffer_varlen;
      sinfo->write_key=write_merge_key_varlen;
524 525 526
    }
    else
    {
527 528 529
      sinfo->write_keys=write_keys;
      sinfo->read_to_buffer=read_to_buffer;
      sinfo->write_key=write_merge_key;
530
    }
531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551
    if (sinfo->buffpek.elements)
    {
      uint maxbuffer=sinfo->buffpek.elements-1;
      if (!mergebuf)
      {
        length=param->sort_buffer_length;
        while (length >= MIN_SORT_MEMORY && !mergebuf)
        {
          mergebuf=my_malloc(length, MYF(0));
          length=length*3/4;
        }
        if (!mergebuf)
        {
          got_error=1;
          continue;
        }
      }
      keys=length/sinfo->key_length;
      if (maxbuffer >= MERGEBUFF2)
      {
        if (param->testflag & T_VERBOSE)
552
          printf("Key %d  - Merging %u keys\n",sinfo->key+1, sinfo->keys);
553
        if (merge_many_buff(sinfo, keys, (uchar **)mergebuf,
554
			    dynamic_element(&sinfo->buffpek, 0, BUFFPEK *),
555
			    (int*) &maxbuffer, &sinfo->tempfile))
556 557 558 559 560 561 562 563 564 565 566 567
        {
          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)
568
        printf("Key %d  - Last merge and dumping keys\n", sinfo->key+1);
569 570
      if (merge_index(sinfo, keys, (uchar **)mergebuf,
                      dynamic_element(&sinfo->buffpek,0,BUFFPEK *),
571
                      maxbuffer,&sinfo->tempfile) ||
572
          flush_ft_buf(sinfo) ||
573
	  flush_pending_blocks(sinfo))
574 575 576 577 578 579 580 581 582 583
      {
        got_error=1;
        continue;
      }
    }
    if (my_b_inited(&sinfo->tempfile_for_exceptions))
    {
      uint key_length;

      if (param->testflag & T_VERBOSE)
584
        printf("Key %d  - Dumping 'long' keys\n", sinfo->key+1);
585 586 587 588 589 590 591 592

      if (flush_io_cache(&sinfo->tempfile_for_exceptions) ||
          reinit_io_cache(&sinfo->tempfile_for_exceptions,READ_CACHE,0L,0,0))
      {
        got_error=1;
        continue;
      }

593 594
      while (!got_error &&
	     !my_b_read(&sinfo->tempfile_for_exceptions,(byte*)&key_length,
svoj@mysql.com's avatar
svoj@mysql.com committed
595
			sizeof(key_length)))
596
      {
svoj@mysql.com's avatar
svoj@mysql.com committed
597
        byte ft_buf[HA_FT_MAXBYTELEN + HA_FT_WLEN + 10];
svoj@mysql.com's avatar
svoj@mysql.com committed
598 599 600 601 602 603
        if (key_length > sizeof(ft_buf) ||
            my_b_read(&sinfo->tempfile_for_exceptions, (byte*)ft_buf,
                      (uint)key_length) ||
            _mi_ck_write(info, sinfo->key, (uchar*)ft_buf,
                         key_length - info->s->rec_reflength))
          got_error=1;
604 605 606 607 608 609
      }
    }
  }
  my_free((gptr) mergebuf,MYF(MY_ALLOW_ZERO_PTR));
  return got_error;
}
610
#endif /* THREAD */
611 612

        /* Write all keys in memory to file for later merge */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
613 614

static int NEAR_F write_keys(MI_SORT_PARAM *info, register uchar **sort_keys,
615
                             uint count, BUFFPEK *buffpek, IO_CACHE *tempfile)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
616
{
617 618
  uchar **end;
  uint sort_length=info->key_length;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
619 620 621
  DBUG_ENTER("write_keys");

  qsort2((byte*) sort_keys,count,sizeof(byte*),(qsort2_cmp) info->key_cmp,
622
         info);
623
  if (!my_b_inited(tempfile) &&
624 625
      open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST",
                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))
626
    DBUG_RETURN(1); /* purecov: inspected */
627

628
  buffpek->file_pos=my_b_tell(tempfile);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
629
  buffpek->count=count;
630 631

  for (end=sort_keys+count ; sort_keys != end ; sort_keys++)
632
  {
633
    if (my_b_write(tempfile,(byte*) *sort_keys,(uint) sort_length))
634
      DBUG_RETURN(1); /* purecov: inspected */
635
  }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
636 637 638
  DBUG_RETURN(0);
} /* write_keys */

639

640 641
static inline int
my_var_write(MI_SORT_PARAM *info, IO_CACHE *to_file, byte *bufs)
642 643
{
  int err;
644
  uint16 len = _mi_keylength(info->keyinfo, (uchar*) bufs);
645

646 647
  /* The following is safe as this is a local file */
  if ((err= my_b_write(to_file, (byte*)&len, sizeof(len))))
648
    return (err);
649
  if ((err= my_b_write(to_file,bufs, (uint) len)))
650 651
    return (err);
  return (0);
652 653 654
}


655 656 657 658
static int NEAR_F write_keys_varlen(MI_SORT_PARAM *info,
				    register uchar **sort_keys,
                                    uint count, BUFFPEK *buffpek,
				    IO_CACHE *tempfile)
659 660
{
  uchar **end;
661
  int err;
662 663 664 665 666
  DBUG_ENTER("write_keys_varlen");

  qsort2((byte*) sort_keys,count,sizeof(byte*),(qsort2_cmp) info->key_cmp,
         info);
  if (!my_b_inited(tempfile) &&
667
      open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST",
668 669 670 671 672 673 674
                       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++)
  {
675
    if ((err= my_var_write(info,tempfile, (byte*) *sort_keys)))
676
      DBUG_RETURN(err);
677 678 679 680
  }
  DBUG_RETURN(0);
} /* write_keys_varlen */

bk@work.mysql.com's avatar
bk@work.mysql.com committed
681

682 683
static int NEAR_F write_key(MI_SORT_PARAM *info, uchar *key,
			    IO_CACHE *tempfile)
684
{
685
  uint key_length=info->real_key_length;
686 687 688
  DBUG_ENTER("write_key");

  if (!my_b_inited(tempfile) &&
689 690
      open_cached_file(tempfile, my_tmpdir(info->tmpdir), "ST",
                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))
691 692 693 694 695 696 697 698
    DBUG_RETURN(1);

  if (my_b_write(tempfile,(byte*)&key_length,sizeof(key_length)) ||
      my_b_write(tempfile,(byte*)key,(uint) key_length))
    DBUG_RETURN(1);
  DBUG_RETURN(0);
} /* write_key */

699

serg@serg.mysql.com's avatar
serg@serg.mysql.com committed
700
/* Write index */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
701

702
static int NEAR_F write_index(MI_SORT_PARAM *info, register uchar **sort_keys,
703
                              register uint count)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
704 705 706 707
{
  DBUG_ENTER("write_index");

  qsort2((gptr) sort_keys,(size_t) count,sizeof(byte*),
708
        (qsort2_cmp) info->key_cmp,info);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
709
  while (count--)
710
  {
711
    if ((*info->key_write)(info,*sort_keys++))
712
      DBUG_RETURN(-1); /* purecov: inspected */
713
  }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
714 715 716 717
  DBUG_RETURN(0);
} /* write_index */


718
        /* Merge buffers to make < MERGEBUFF2 buffers */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
719 720

static int NEAR_F merge_many_buff(MI_SORT_PARAM *info, uint keys,
721 722
                                  uchar **sort_keys, BUFFPEK *buffpek,
                                  int *maxbuffer, IO_CACHE *t_file)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
723 724
{
  register int i;
725
  IO_CACHE t_file2, *from_file, *to_file, *temp;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
726 727 728
  BUFFPEK *lastbuff;
  DBUG_ENTER("merge_many_buff");

729
  if (*maxbuffer < MERGEBUFF2)
730
    DBUG_RETURN(0);                             /* purecov: inspected */
731
  if (flush_io_cache(t_file) ||
732 733
      open_cached_file(&t_file2,my_tmpdir(info->tmpdir),"ST",
                       DISK_BUFFER_SIZE, info->sort_info->param->myf_rw))
734
    DBUG_RETURN(1);                             /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
735

736
  from_file= t_file ; to_file= &t_file2;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
737 738
  while (*maxbuffer >= MERGEBUFF2)
  {
739 740
    reinit_io_cache(from_file,READ_CACHE,0L,0,0);
    reinit_io_cache(to_file,WRITE_CACHE,0L,0,0);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
741 742 743 744
    lastbuff=buffpek;
    for (i=0 ; i <= *maxbuffer-MERGEBUFF*3/2 ; i+=MERGEBUFF)
    {
      if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,
745 746
                        buffpek+i,buffpek+i+MERGEBUFF-1))
        break; /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
747 748
    }
    if (merge_buffers(info,keys,from_file,to_file,sort_keys,lastbuff++,
749
                      buffpek+i,buffpek+ *maxbuffer))
750
      break; /* purecov: inspected */
751
    if (flush_io_cache(to_file))
752
      break;                                    /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
753
    temp=from_file; from_file=to_file; to_file=temp;
754
    *maxbuffer= (int) (lastbuff-buffpek)-1;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
755
  }
756
  close_cached_file(to_file);                   /* This holds old result */
757
  if (to_file == t_file)
758
    *t_file=t_file2;                            /* Copy result file */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
759

760
  DBUG_RETURN(*maxbuffer >= MERGEBUFF2);        /* Return 1 if interrupted */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
761 762 763
} /* merge_many_buff */


764 765 766 767 768 769 770 771 772 773 774 775
/*
   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
*/
bk@work.mysql.com's avatar
bk@work.mysql.com committed
776

777
static uint NEAR_F read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek,
778
                                  uint sort_length)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
779 780 781 782
{
  register uint count;
  uint length;

783
  if ((count=(uint) min((ha_rows) buffpek->max_keys,buffpek->count)))
bk@work.mysql.com's avatar
bk@work.mysql.com committed
784
  {
785
    if (my_pread(fromfile->file,(byte*) buffpek->base,
786 787
                 (length= sort_length*count),buffpek->file_pos,MYF_RW))
      return((uint) -1);                        /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
788
    buffpek->key=buffpek->base;
789 790
    buffpek->file_pos+= length;                 /* New filepos */
    buffpek->count-=    count;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
791 792 793 794 795
    buffpek->mem_count= count;
  }
  return (count*sort_length);
} /* read_to_buffer */

796 797 798 799 800 801 802
static uint NEAR_F read_to_buffer_varlen(IO_CACHE *fromfile, BUFFPEK *buffpek,
                                         uint sort_length)
{
  register uint count;
  uint16 length_of_key = 0;
  uint idx;
  uchar *buffp;
803

804 805 806 807 808 809 810 811
  if ((count=(uint) min((ha_rows) buffpek->max_keys,buffpek->count)))
  {
    buffp = buffpek->base;

    for (idx=1;idx<=count;idx++)
    {
      if (my_pread(fromfile->file,(byte*)&length_of_key,sizeof(length_of_key),
                   buffpek->file_pos,MYF_RW))
812
        return((uint) -1);
813 814 815
      buffpek->file_pos+=sizeof(length_of_key);
      if (my_pread(fromfile->file,(byte*) buffp,length_of_key,
                   buffpek->file_pos,MYF_RW))
816
        return((uint) -1);
817 818
      buffpek->file_pos+=length_of_key;
      buffp = buffp + sort_length;
819
    }
820 821 822 823 824 825 826 827
    buffpek->key=buffpek->base;
    buffpek->count-=    count;
    buffpek->mem_count= count;
  }
  return (count*sort_length);
} /* read_to_buffer_varlen */


828 829
static int NEAR_F write_merge_key_varlen(MI_SORT_PARAM *info,
					 IO_CACHE *to_file,char* key,
830 831 832 833
                                         uint sort_length, uint count)
{
  uint idx;

834
  char *bufs = key;
835 836
  for (idx=1;idx<=count;idx++)
  {
837
    int err;
838
    if ((err= my_var_write(info,to_file, (byte*) bufs)))
839
      return (err);
840 841 842 843 844
    bufs=bufs+sort_length;
  }
  return(0);
}

845 846 847 848

static int NEAR_F write_merge_key(MI_SORT_PARAM *info __attribute__((unused)),
				  IO_CACHE *to_file, char* key,
				  uint sort_length, uint count)
849
{
850
  return my_b_write(to_file,(byte*) key,(uint) sort_length*count);
851
}
bk@work.mysql.com's avatar
bk@work.mysql.com committed
852

853 854 855 856
/*
  Merge buffers to one buffer
  If to_file == 0 then use info->key_write
*/
bk@work.mysql.com's avatar
bk@work.mysql.com committed
857 858

static int NEAR_F
859
merge_buffers(MI_SORT_PARAM *info, uint keys, IO_CACHE *from_file,
860 861
              IO_CACHE *to_file, uchar **sort_keys, BUFFPEK *lastbuff,
              BUFFPEK *Fb, BUFFPEK *Tb)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
862 863 864 865 866 867 868 869
{
  int error;
  uint sort_length,maxcount;
  ha_rows count;
  my_off_t to_start_filepos;
  uchar *strpos;
  BUFFPEK *buffpek,**refpek;
  QUEUE queue;
870
  volatile int *killed= killed_ptr(info->sort_info->param);
hf@genie.(none)'s avatar
SCRUM  
hf@genie.(none) committed
871

bk@work.mysql.com's avatar
bk@work.mysql.com committed
872 873 874 875 876 877
  DBUG_ENTER("merge_buffers");

  count=error=0;
  maxcount=keys/((uint) (Tb-Fb) +1);
  LINT_INIT(to_start_filepos);
  if (to_file)
878
    to_start_filepos=my_b_tell(to_file);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
879
  strpos=(uchar*) sort_keys;
880
  sort_length=info->key_length;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
881 882

  if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0,
883
                 (int (*)(void*, byte *,byte*)) info->key_cmp,
884
                 (void*) info))
885
    DBUG_RETURN(1); /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
886

887
  for (buffpek= Fb ; buffpek <= Tb ; buffpek++)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
888 889 890 891
  {
    count+= buffpek->count;
    buffpek->base= strpos;
    buffpek->max_keys=maxcount;
monty@mysql.com's avatar
monty@mysql.com committed
892 893
    strpos+= (uint) (error=(int) info->read_to_buffer(from_file,buffpek,
                                                      sort_length));
894 895
    if (error == -1)
      goto err; /* purecov: inspected */
monty@hundin.mysql.fi's avatar
monty@hundin.mysql.fi committed
896
    queue_insert(&queue,(char*) buffpek);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
897 898 899 900 901 902
  }

  while (queue.elements > 1)
  {
    for (;;)
    {
serg@serg.mysql.com's avatar
serg@serg.mysql.com committed
903 904 905 906
      if (*killed)
      {
        error=1; goto err;
      }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
907 908 909
      buffpek=(BUFFPEK*) queue_top(&queue);
      if (to_file)
      {
monty@mysql.com's avatar
monty@mysql.com committed
910 911
        if (info->write_key(info,to_file,(byte*) buffpek->key,
                            (uint) sort_length,1))
912 913 914
        {
          error=1; goto err; /* purecov: inspected */
        }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
915 916 917
      }
      else
      {
918 919 920 921
        if ((*info->key_write)(info,(void*) buffpek->key))
        {
          error=1; goto err; /* purecov: inspected */
        }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
922 923 924 925
      }
      buffpek->key+=sort_length;
      if (! --buffpek->mem_count)
      {
926
        if (!(error=(int) info->read_to_buffer(from_file,buffpek,sort_length)))
927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952
        {
          uchar *base=buffpek->base;
          uint max_keys=buffpek->max_keys;

          VOID(queue_remove(&queue,0));

          /* 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 */
        }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
953
      }
954
      else if (error == -1)
955 956
        goto err;               /* purecov: inspected */
      queue_replaced(&queue);   /* Top element has been replaced */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
957 958 959 960 961 962 963 964 965
    }
  }
  buffpek=(BUFFPEK*) queue_top(&queue);
  buffpek->base=(uchar *) sort_keys;
  buffpek->max_keys=keys;
  do
  {
    if (to_file)
    {
966
      if (info->write_key(info,to_file,(byte*) buffpek->key,
967
                         sort_length,buffpek->mem_count))
bk@work.mysql.com's avatar
bk@work.mysql.com committed
968
      {
969
        error=1; goto err; /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
970 971 972 973 974 975 976
      }
    }
    else
    {
      register uchar *end;
      strpos= buffpek->key;
      for (end=strpos+buffpek->mem_count*sort_length;
977 978
           strpos != end ;
           strpos+=sort_length)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
979
      {
980 981 982 983
        if ((*info->key_write)(info,(void*) strpos))
        {
          error=1; goto err; /* purecov: inspected */
        }
bk@work.mysql.com's avatar
bk@work.mysql.com committed
984 985 986
      }
    }
  }
987
  while ((error=(int) info->read_to_buffer(from_file,buffpek,sort_length)) != -1 &&
988
         error != 0);
bk@work.mysql.com's avatar
bk@work.mysql.com committed
989 990 991

  lastbuff->count=count;
  if (to_file)
992
    lastbuff->file_pos=to_start_filepos;
bk@work.mysql.com's avatar
bk@work.mysql.com committed
993 994 995 996 997 998
err:
  delete_queue(&queue);
  DBUG_RETURN(error);
} /* merge_buffers */


999
        /* Do a merge to output-file (save only positions) */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
1000 1001 1002

static int NEAR_F
merge_index(MI_SORT_PARAM *info, uint keys, uchar **sort_keys,
1003
            BUFFPEK *buffpek, int maxbuffer, IO_CACHE *tempfile)
bk@work.mysql.com's avatar
bk@work.mysql.com committed
1004 1005
{
  DBUG_ENTER("merge_index");
1006
  if (merge_buffers(info,keys,tempfile,(IO_CACHE*) 0,sort_keys,buffpek,buffpek,
1007
                    buffpek+maxbuffer))
1008
    DBUG_RETURN(1); /* purecov: inspected */
bk@work.mysql.com's avatar
bk@work.mysql.com committed
1009 1010 1011
  DBUG_RETURN(0);
} /* merge_index */

1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024
static int
flush_ft_buf(MI_SORT_PARAM *info)
{
  int err=0;
  if (info->sort_info->ft_buf)
  {
    err=sort_ft_buf_flush(info);
    my_free((gptr)info->sort_info->ft_buf, MYF(0));
    info->sort_info->ft_buf=0;
  }
  return err;
}