← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/uniques.cc

  • Committer: Padraig O'Sullivan
  • Date: 2009-08-07 14:14:58 UTC
  • mfrom: (1112 staging)
  • mto: (1115.3.4 captain)
  • mto: This revision was merged to the branch mainline in revision 1117.
  • Revision ID: osullivan.padraig@gmail.com-20090807141458-qrc3don58s304ore
Merge from trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/uniques.cc
11
11
 
12
12
   You should have received a copy of the GNU General Public License
13
13
   along with this program; if not, write to the Free Software
14
 
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
15
15
 
16
16
/*
17
17
  Function to handle quick removal of duplicates
21
21
  The basic idea is as follows:
22
22
 
23
23
  Store first all strings in a binary tree, ignoring duplicates.
 
24
  When the tree uses more memory than 'max_heap_table_size',
 
25
  write the tree (in sorted order) out to disk and start with a new tree.
 
26
  When all data has been generated, merge the trees (removing any found
 
27
  duplicates).
24
28
 
25
29
  The unique entries will be returned in sort order, to ensure that we do the
26
30
  deletes in disk order.
27
31
*/
28
32
 
29
 
#include "config.h"
30
 
 
31
 
#include <math.h>
32
 
 
 
33
#include <drizzled/server_includes.h>
 
34
#include <drizzled/sql_sort.h>
 
35
#include <drizzled/session.h>
33
36
#include <queue>
34
37
 
35
 
#include "drizzled/sql_sort.h"
36
 
#include "drizzled/session.h"
37
 
#include "drizzled/sql_list.h"
38
 
#include "drizzled/internal/iocache.h"
39
 
 
40
38
#if defined(CMATH_NAMESPACE)
41
39
using namespace CMATH_NAMESPACE;
42
40
#endif
43
41
 
44
42
using namespace std;
45
43
 
46
 
namespace drizzled
 
44
 
 
45
int unique_write_to_file(unsigned char* key, element_count,
 
46
                         Unique *unique)
47
47
{
 
48
  /*
 
49
    Use unique->size (size of element stored in the tree) and not
 
50
    unique->tree.size_of_element. The latter is different from unique->size
 
51
    when tree implementation chooses to store pointer to key in TREE_ELEMENT
 
52
    (instead of storing the element itself there)
 
53
  */
 
54
  return my_b_write(&unique->file, key, unique->size) ? 1 : 0;
 
55
}
48
56
 
49
57
int unique_write_to_ptrs(unsigned char* key,
50
 
                         uint32_t, Unique *unique)
 
58
                         element_count, Unique *unique)
51
59
{
52
60
  memcpy(unique->record_pointers, key, unique->size);
53
61
  unique->record_pointers+=unique->size;
56
64
 
57
65
Unique::Unique(qsort_cmp2 comp_func, void * comp_func_fixed_arg,
58
66
               uint32_t size_arg, size_t max_in_memory_size_arg)
59
 
  : max_in_memory_size(max_in_memory_size_arg),
60
 
    size(size_arg),
61
 
    elements(0)
 
67
  :max_in_memory_size(max_in_memory_size_arg), size(size_arg), elements(0)
62
68
{
63
 
  // Second element is max size for memory use in unique sort
64
 
  init_tree(&tree, 0, 0, size, comp_func, false,
 
69
  my_b_clear(&file);
 
70
  init_tree(&tree, (ulong) (max_in_memory_size / 16), 0, size, comp_func, 0,
65
71
            NULL, comp_func_fixed_arg);
 
72
  /* If the following fail's the next add will also fail */
 
73
  my_init_dynamic_array(&file_ptrs, sizeof(BUFFPEK), 16, 16);
 
74
  /*
 
75
    If you change the following, change it in get_max_elements function, too.
 
76
  */
 
77
  max_elements= (ulong) (max_in_memory_size /
 
78
                         ALIGN_SIZE(sizeof(TREE_ELEMENT)+size));
 
79
  open_cached_file(&file, drizzle_tmpdir,TEMP_PREFIX, DISK_BUFFER_SIZE,
 
80
                   MYF(MY_WME));
66
81
}
67
82
 
68
83
 
88
103
 
89
104
 
90
105
/*
 
106
  Calculate cost of merge_buffers function call for given sequence of
 
107
  input stream lengths and store the number of rows in result stream in *last.
 
108
 
 
109
  SYNOPSIS
 
110
    get_merge_buffers_cost()
 
111
      buff_elems  Array of #s of elements in buffers
 
112
      elem_size   Size of element stored in buffer
 
113
      first       Pointer to first merged element size
 
114
      last        Pointer to last merged element size
 
115
 
 
116
  RETURN
 
117
    Cost of merge_buffers operation in disk seeks.
 
118
 
 
119
  NOTES
 
120
    It is assumed that no rows are eliminated during merge.
 
121
    The cost is calculated as
 
122
 
 
123
      cost(read_and_write) + cost(merge_comparisons).
 
124
 
 
125
    All bytes in the sequences is read and written back during merge so cost
 
126
    of disk io is 2*elem_size*total_buf_elems/IO_SIZE (2 is for read + write)
 
127
 
 
128
    For comparisons cost calculations we assume that all merged sequences have
 
129
    the same length, so each of total_buf_size elements will be added to a sort
 
130
    heap with (n_buffers-1) elements. This gives the comparison cost:
 
131
 
 
132
      total_buf_elems* log2(n_buffers) / TIME_FOR_COMPARE_ROWID;
 
133
*/
 
134
 
 
135
static double get_merge_buffers_cost(uint32_t *, uint32_t elem_size,
 
136
                                     uint32_t *first, uint32_t *last)
 
137
{
 
138
  uint32_t total_buf_elems= 0;
 
139
  for (uint32_t *pbuf= first; pbuf <= last; pbuf++)
 
140
    total_buf_elems+= *pbuf;
 
141
  *last= total_buf_elems;
 
142
 
 
143
  int n_buffers= last - first + 1;
 
144
 
 
145
  /* Using log2(n)=log(n)/log(2) formula */
 
146
  return 2*((double)total_buf_elems*elem_size) / IO_SIZE +
 
147
     total_buf_elems*log((double) n_buffers) / (TIME_FOR_COMPARE_ROWID * M_LN2);
 
148
}
 
149
 
 
150
 
 
151
/*
 
152
  Calculate cost of merging buffers into one in Unique::get, i.e. calculate
 
153
  how long (in terms of disk seeks) the two calls
 
154
    merge_many_buffs(...);
 
155
    merge_buffers(...);
 
156
  will take.
 
157
 
 
158
  SYNOPSIS
 
159
    get_merge_many_buffs_cost()
 
160
      buffer        buffer space for temporary data, at least
 
161
                    Unique::get_cost_calc_buff_size bytes
 
162
      maxbuffer     # of full buffers
 
163
      max_n_elems   # of elements in first maxbuffer buffers
 
164
      last_n_elems  # of elements in last buffer
 
165
      elem_size     size of buffer element
 
166
 
 
167
  NOTES
 
168
    maxbuffer+1 buffers are merged, where first maxbuffer buffers contain
 
169
    max_n_elems elements each and last buffer contains last_n_elems elements.
 
170
 
 
171
    The current implementation does a dumb simulation of merge_many_buffs
 
172
    function actions.
 
173
 
 
174
  RETURN
 
175
    Cost of merge in disk seeks.
 
176
*/
 
177
 
 
178
static double get_merge_many_buffs_cost(uint32_t *buffer,
 
179
                                        uint32_t maxbuffer, uint32_t max_n_elems,
 
180
                                        uint32_t last_n_elems, int elem_size)
 
181
{
 
182
  register int i;
 
183
  double total_cost= 0.0;
 
184
  uint32_t *buff_elems= buffer; /* #s of elements in each of merged sequences */
 
185
 
 
186
  /*
 
187
    Set initial state: first maxbuffer sequences contain max_n_elems elements
 
188
    each, last sequence contains last_n_elems elements.
 
189
  */
 
190
  for (i = 0; i < (int)maxbuffer; i++)
 
191
    buff_elems[i]= max_n_elems;
 
192
  buff_elems[maxbuffer]= last_n_elems;
 
193
 
 
194
  /*
 
195
    Do it exactly as merge_many_buff function does, calling
 
196
    get_merge_buffers_cost to get cost of merge_buffers.
 
197
  */
 
198
  if (maxbuffer >= MERGEBUFF2)
 
199
  {
 
200
    while (maxbuffer >= MERGEBUFF2)
 
201
    {
 
202
      uint32_t lastbuff= 0;
 
203
      for (i = 0; i <= (int) maxbuffer - MERGEBUFF*3/2; i += MERGEBUFF)
 
204
      {
 
205
        total_cost+=get_merge_buffers_cost(buff_elems, elem_size,
 
206
                                           buff_elems + i,
 
207
                                           buff_elems + i + MERGEBUFF-1);
 
208
        lastbuff++;
 
209
      }
 
210
      total_cost+=get_merge_buffers_cost(buff_elems, elem_size,
 
211
                                         buff_elems + i,
 
212
                                         buff_elems + maxbuffer);
 
213
      maxbuffer= lastbuff;
 
214
    }
 
215
  }
 
216
 
 
217
  /* Simulate final merge_buff call. */
 
218
  total_cost += get_merge_buffers_cost(buff_elems, elem_size,
 
219
                                       buff_elems, buff_elems + maxbuffer);
 
220
  return total_cost;
 
221
}
 
222
 
 
223
 
 
224
/*
91
225
  Calculate cost of using Unique for processing nkeys elements of size
92
226
  key_size using max_in_memory_size memory.
93
227
 
122
256
 
123
257
      Approximate value of log2(N!) is calculated by log2_n_fact function.
124
258
 
125
 
       
126
 
      (The Next two are historical, we do all unique operations in memory or fail)
127
 
 
128
259
    2. Cost of merging.
129
260
      If only one tree is created by Unique no merging will be necessary.
130
261
      Otherwise, we model execution of merge_many_buff function and count
137
268
      these will be random seeks.
138
269
*/
139
270
 
140
 
double Unique::get_use_cost(uint32_t *, uint32_t nkeys, uint32_t key_size,
141
 
                            size_t max_in_memory_size_arg)
 
271
double Unique::get_use_cost(uint32_t *buffer, uint32_t nkeys, uint32_t key_size,
 
272
                            size_t max_in_memory_size)
142
273
{
143
274
  ulong max_elements_in_tree;
144
275
  ulong last_tree_elems;
 
276
  int   n_full_trees; /* number of trees in unique - 1 */
145
277
  double result;
146
278
 
147
 
  max_elements_in_tree= ((ulong) max_in_memory_size_arg /
 
279
  max_elements_in_tree= ((ulong) max_in_memory_size /
148
280
                         ALIGN_SIZE(sizeof(TREE_ELEMENT)+key_size));
149
281
 
 
282
  n_full_trees=    nkeys / max_elements_in_tree;
150
283
  last_tree_elems= nkeys % max_elements_in_tree;
151
284
 
152
285
  /* Calculate cost of creating trees */
153
286
  result= 2*log2_n_fact(last_tree_elems + 1.0);
 
287
  if (n_full_trees)
 
288
    result+= n_full_trees * log2_n_fact(max_elements_in_tree + 1.0);
154
289
  result /= TIME_FOR_COMPARE_ROWID;
155
290
 
 
291
 
 
292
  if (!n_full_trees)
 
293
    return result;
 
294
 
 
295
  /*
 
296
    There is more then one tree and merging is necessary.
 
297
    First, add cost of writing all trees to disk, assuming that all disk
 
298
    writes are sequential.
 
299
  */
 
300
  result += DISK_SEEK_BASE_COST * n_full_trees *
 
301
              ceil(((double) key_size)*max_elements_in_tree / IO_SIZE);
 
302
  result += DISK_SEEK_BASE_COST * ceil(((double) key_size)*last_tree_elems / IO_SIZE);
 
303
 
 
304
  /* Cost of merge */
 
305
  double merge_cost= get_merge_many_buffs_cost(buffer, n_full_trees,
 
306
                                               max_elements_in_tree,
 
307
                                               last_tree_elems, key_size);
 
308
  if (merge_cost < 0.0)
 
309
    return merge_cost;
 
310
 
 
311
  result += merge_cost;
 
312
  /*
 
313
    Add cost of reading the resulting sequence, assuming there were no
 
314
    duplicate elements.
 
315
  */
 
316
  result += ceil((double)key_size*nkeys/IO_SIZE);
 
317
 
156
318
  return result;
157
319
}
158
320
 
159
321
Unique::~Unique()
160
322
{
161
 
  delete_tree(&tree);
 
323
  close_cached_file(&file);
 
324
  delete_tree(&tree);
 
325
  delete_dynamic(&file_ptrs);
 
326
}
 
327
 
 
328
 
 
329
    /* Write tree to disk; clear tree */
 
330
bool Unique::flush()
 
331
{
 
332
  BUFFPEK file_ptr;
 
333
  elements+= tree.elements_in_tree;
 
334
  file_ptr.count=tree.elements_in_tree;
 
335
  file_ptr.file_pos=my_b_tell(&file);
 
336
 
 
337
  if (tree_walk(&tree, (tree_walk_action) unique_write_to_file,
 
338
                (void*) this, left_root_right) ||
 
339
      insert_dynamic(&file_ptrs, (unsigned char*) &file_ptr))
 
340
    return 1;
 
341
  delete_tree(&tree);
 
342
  return 0;
162
343
}
163
344
 
164
345
 
165
346
/*
166
 
  Clear the tree.
 
347
  Clear the tree and the file.
167
348
  You must call reset() if you want to reuse Unique after walk().
168
349
*/
169
350
 
171
352
Unique::reset()
172
353
{
173
354
  reset_tree(&tree);
174
 
  assert(elements == 0);
 
355
  /*
 
356
    If elements != 0, some trees were stored in the file (see how
 
357
    flush() works). Note, that we can not count on my_b_tell(&file) == 0
 
358
    here, because it can return 0 right after walk(), and walk() does not
 
359
    reset any Unique member.
 
360
  */
 
361
  if (elements)
 
362
  {
 
363
    reset_dynamic(&file_ptrs);
 
364
    reinit_io_cache(&file, WRITE_CACHE, 0L, 0, 1);
 
365
  }
 
366
  elements= 0;
 
367
}
 
368
 
 
369
/*
 
370
  The comparison function, passed to queue_init() in merge_walk() and in
 
371
  merge_buffers() when the latter is called from Uniques::get() must
 
372
  use comparison function of Uniques::tree, but compare members of struct
 
373
  BUFFPEK.
 
374
*/
 
375
 
 
376
#ifdef __cplusplus
 
377
extern "C" {
 
378
#endif
 
379
 
 
380
static int buffpek_compare(void *arg, unsigned char *key_ptr1, unsigned char *key_ptr2)
 
381
{
 
382
  BUFFPEK_COMPARE_CONTEXT *ctx= (BUFFPEK_COMPARE_CONTEXT *) arg;
 
383
  return ctx->key_compare(ctx->key_compare_arg,
 
384
                          *((unsigned char **) key_ptr1), *((unsigned char **)key_ptr2));
 
385
}
 
386
 
 
387
#ifdef __cplusplus
 
388
}
 
389
#endif
 
390
 
 
391
/*
 
392
 The comparison function object, passed to a priority_queue in merge_walk()
 
393
 as its sort function parameter.
 
394
*/
 
395
 
 
396
class buffpek_compare_functor
 
397
{
 
398
  qsort_cmp2 key_compare;
 
399
  void *key_compare_arg;
 
400
  public:
 
401
  buffpek_compare_functor(qsort_cmp2 in_key_compare, void *in_compare_arg)
 
402
    : key_compare(in_key_compare), key_compare_arg(in_compare_arg) { }
 
403
  inline bool operator()(const BUFFPEK *i, const BUFFPEK *j)
 
404
  {
 
405
    return key_compare(key_compare_arg,
 
406
                    i->key, j->key);
 
407
  }
 
408
};
 
409
 
 
410
/*
 
411
  DESCRIPTION
 
412
 
 
413
    Function is very similar to merge_buffers, but instead of writing sorted
 
414
    unique keys to the output file, it invokes walk_action for each key.
 
415
    This saves I/O if you need to pass through all unique keys only once.
 
416
 
 
417
  SYNOPSIS
 
418
    merge_walk()
 
419
  All params are 'IN' (but see comment for begin, end):
 
420
    merge_buffer       buffer to perform cached piece-by-piece loading
 
421
                       of trees; initially the buffer is empty
 
422
    merge_buffer_size  size of merge_buffer. Must be aligned with
 
423
                       key_length
 
424
    key_length         size of tree element; key_length * (end - begin)
 
425
                       must be less or equal than merge_buffer_size.
 
426
    begin              pointer to BUFFPEK struct for the first tree.
 
427
    end                pointer to BUFFPEK struct for the last tree;
 
428
                       end > begin and [begin, end) form a consecutive
 
429
                       range. BUFFPEKs structs in that range are used and
 
430
                       overwritten in merge_walk().
 
431
    walk_action        element visitor. Action is called for each unique
 
432
                       key.
 
433
    walk_action_arg    argument to walk action. Passed to it on each call.
 
434
    compare            elements comparison function
 
435
    compare_arg        comparison function argument
 
436
    file               file with all trees dumped. Trees in the file
 
437
                       must contain sorted unique values. Cache must be
 
438
                       initialized in read mode.
 
439
  RETURN VALUE
 
440
    0     ok
 
441
    <> 0  error
 
442
*/
 
443
 
 
444
static bool merge_walk(unsigned char *merge_buffer, ulong merge_buffer_size,
 
445
                       uint32_t key_length, BUFFPEK *begin, BUFFPEK *end,
 
446
                       tree_walk_action walk_action, void *walk_action_arg,
 
447
                       qsort_cmp2 compare, void *compare_arg,
 
448
                       IO_CACHE *file)
 
449
{
 
450
  if (end <= begin ||
 
451
      merge_buffer_size < (ulong) (key_length * (end - begin + 1))) 
 
452
    return 1;
 
453
  priority_queue<BUFFPEK *, vector<BUFFPEK *>, buffpek_compare_functor >
 
454
    queue(buffpek_compare_functor(compare, compare_arg));
 
455
  /* we need space for one key when a piece of merge buffer is re-read */
 
456
  merge_buffer_size-= key_length;
 
457
  unsigned char *save_key_buff= merge_buffer + merge_buffer_size;
 
458
  uint32_t max_key_count_per_piece= (uint32_t) (merge_buffer_size/(end-begin) /
 
459
                                        key_length);
 
460
  /* if piece_size is aligned reuse_freed_buffer will always hit */
 
461
  uint32_t piece_size= max_key_count_per_piece * key_length;
 
462
  uint32_t bytes_read;               /* to hold return value of read_to_buffer */
 
463
  BUFFPEK *top;
 
464
  int res= 1;
 
465
  /*
 
466
    Invariant: queue must contain top element from each tree, until a tree
 
467
    is not completely walked through.
 
468
    Here we're forcing the invariant, inserting one element from each tree
 
469
    to the queue.
 
470
  */
 
471
  for (top= begin; top != end; ++top)
 
472
  {
 
473
    top->base= merge_buffer + (top - begin) * piece_size;
 
474
    top->max_keys= max_key_count_per_piece;
 
475
    bytes_read= read_to_buffer(file, top, key_length);
 
476
    if (bytes_read == (uint32_t) (-1))
 
477
      goto end;
 
478
    assert(bytes_read);
 
479
    queue.push(top);
 
480
  }
 
481
  top= queue.top();
 
482
  while (queue.size() > 1)
 
483
  {
 
484
    /*
 
485
      Every iteration one element is removed from the queue, and one is
 
486
      inserted by the rules of the invariant. If two adjacent elements on
 
487
      the top of the queue are not equal, biggest one is unique, because all
 
488
      elements in each tree are unique. Action is applied only to unique
 
489
      elements.
 
490
    */
 
491
    void *old_key= top->key;
 
492
    /*
 
493
      read next key from the cache or from the file and push it to the
 
494
      queue; this gives new top.
 
495
    */
 
496
    top->key+= key_length;
 
497
    if (--top->mem_count)
 
498
    {
 
499
      queue.pop();
 
500
      queue.push(top);
 
501
    }
 
502
    else /* next piece should be read */
 
503
    {
 
504
      /* save old_key not to overwrite it in read_to_buffer */
 
505
      memcpy(save_key_buff, old_key, key_length);
 
506
      old_key= save_key_buff;
 
507
      bytes_read= read_to_buffer(file, top, key_length);
 
508
      if (bytes_read == (uint32_t) (-1))
 
509
        goto end;
 
510
      else if (bytes_read > 0) /* top->key, top->mem_count are reset */
 
511
      {                        /* in read_to_buffer */
 
512
        queue.pop();
 
513
        queue.push(top);
 
514
      }
 
515
      else
 
516
      {
 
517
        /*
 
518
          Tree for old 'top' element is empty: remove it from the queue. 
 
519
        */
 
520
        queue.pop();
 
521
      }
 
522
    }
 
523
    top= queue.top();
 
524
    /* new top has been obtained; if old top is unique, apply the action */
 
525
    if (compare(compare_arg, old_key, top->key))
 
526
    {
 
527
      if (walk_action(old_key, 1, walk_action_arg))
 
528
        goto end;
 
529
    }
 
530
  }
 
531
  /*
 
532
    Applying walk_action to the tail of the last tree: this is safe because
 
533
    either we had only one tree in the beginning, either we work with the
 
534
    last tree in the queue.
 
535
  */
 
536
  do
 
537
  {
 
538
    do
 
539
    {
 
540
      if (walk_action(top->key, 1, walk_action_arg))
 
541
        goto end;
 
542
      top->key+= key_length;
 
543
    }
 
544
    while (--top->mem_count);
 
545
    bytes_read= read_to_buffer(file, top, key_length);
 
546
    if (bytes_read == (uint32_t) (-1))
 
547
      goto end;
 
548
  }
 
549
  while (bytes_read);
 
550
  res= 0;
 
551
end:
 
552
  return res;
175
553
}
176
554
 
177
555
 
187
565
  SYNOPSIS
188
566
    Unique:walk()
189
567
  All params are 'IN':
190
 
    action  function-visitor, typed in include/tree.h
 
568
    action  function-visitor, typed in include/my_tree.h
191
569
            function is called for each unique element
192
570
    arg     argument for visitor, which is passed to it on each call
193
571
  RETURN VALUE
197
575
 
198
576
bool Unique::walk(tree_walk_action action, void *walk_action_arg)
199
577
{
200
 
  return tree_walk(&tree, action, walk_action_arg, left_root_right);
 
578
  int res;
 
579
  unsigned char *merge_buffer;
 
580
 
 
581
  if (elements == 0)                       /* the whole tree is in memory */
 
582
    return tree_walk(&tree, action, walk_action_arg, left_root_right);
 
583
 
 
584
  /* flush current tree to the file to have some memory for merge buffer */
 
585
  if (flush())
 
586
    return 1;
 
587
  if (flush_io_cache(&file) || reinit_io_cache(&file, READ_CACHE, 0L, 0, 0))
 
588
    return 1;
 
589
  if (!(merge_buffer= (unsigned char *) malloc(max_in_memory_size)))
 
590
    return 1;
 
591
  res= merge_walk(merge_buffer, (ulong) max_in_memory_size, size,
 
592
                  (BUFFPEK *) file_ptrs.buffer,
 
593
                  (BUFFPEK *) file_ptrs.buffer + file_ptrs.elements,
 
594
                  action, walk_action_arg,
 
595
                  tree.compare, tree.custom_arg, &file);
 
596
  free((char*) merge_buffer);
 
597
  return res;
201
598
}
202
599
 
203
600
/*
207
604
 
208
605
bool Unique::get(Table *table)
209
606
{
210
 
  table->sort.found_records= elements+tree.elements_in_tree;
 
607
  SORTPARAM sort_param;
 
608
  table->sort.found_records=elements+tree.elements_in_tree;
211
609
 
212
 
  if ((record_pointers=table->sort.record_pointers= (unsigned char*)
213
 
       malloc(size * tree.elements_in_tree)))
 
610
  if (my_b_tell(&file) == 0)
214
611
  {
215
 
    (void) tree_walk(&tree, (tree_walk_action) unique_write_to_ptrs,
216
 
                     this, left_root_right);
217
 
    return 0;
 
612
    /* Whole tree is in memory;  Don't use disk if you don't need to */
 
613
    if ((record_pointers=table->sort.record_pointers= (unsigned char*)
 
614
         malloc(size * tree.elements_in_tree)))
 
615
    {
 
616
      (void) tree_walk(&tree, (tree_walk_action) unique_write_to_ptrs,
 
617
                       this, left_root_right);
 
618
      return 0;
 
619
    }
218
620
  }
219
 
  /* Not enough memory */
220
 
  return 1;
 
621
  /* Not enough memory; Save the result to file && free memory used by tree */
 
622
  if (flush())
 
623
    return 1;
 
624
 
 
625
  IO_CACHE *outfile=table->sort.io_cache;
 
626
  BUFFPEK *file_ptr= (BUFFPEK*) file_ptrs.buffer;
 
627
  uint32_t maxbuffer= file_ptrs.elements - 1;
 
628
  unsigned char *sort_buffer;
 
629
  my_off_t save_pos;
 
630
  bool error=1;
 
631
 
 
632
      /* Open cached file if it isn't open */
 
633
  outfile=table->sort.io_cache= new IO_CACHE;
 
634
  memset(outfile, 0, sizeof(IO_CACHE));
 
635
 
 
636
  if (!outfile || (! my_b_inited(outfile) && open_cached_file(outfile,drizzle_tmpdir,TEMP_PREFIX,READ_RECORD_BUFFER, MYF(MY_WME))))
 
637
    return 1;
 
638
  reinit_io_cache(outfile, WRITE_CACHE, 0L, 0, 0);
 
639
 
 
640
  memset(&sort_param, 0, sizeof(sort_param));
 
641
  sort_param.max_rows= elements;
 
642
  sort_param.sort_form=table;
 
643
  sort_param.rec_length= sort_param.sort_length= sort_param.ref_length=
 
644
    size;
 
645
  sort_param.keys= (uint32_t) (max_in_memory_size / sort_param.sort_length);
 
646
  sort_param.not_killable=1;
 
647
 
 
648
  if (!(sort_buffer=(unsigned char*) malloc((sort_param.keys+1) *
 
649
                                            sort_param.sort_length)))
 
650
    return 1;
 
651
  sort_param.unique_buff= sort_buffer+(sort_param.keys*
 
652
                                       sort_param.sort_length);
 
653
 
 
654
  sort_param.compare= (qsort2_cmp) buffpek_compare;
 
655
  sort_param.cmp_context.key_compare= tree.compare;
 
656
  sort_param.cmp_context.key_compare_arg= tree.custom_arg;
 
657
 
 
658
  /* Merge the buffers to one file, removing duplicates */
 
659
  if (merge_many_buff(&sort_param,sort_buffer,file_ptr,&maxbuffer,&file))
 
660
    goto err;
 
661
  if (flush_io_cache(&file) ||
 
662
      reinit_io_cache(&file,READ_CACHE,0L,0,0))
 
663
    goto err;
 
664
  if (merge_buffers(&sort_param, &file, outfile, sort_buffer, file_ptr,
 
665
                    file_ptr, file_ptr+maxbuffer,0))
 
666
    goto err;
 
667
  error=0;
 
668
err:
 
669
  if (sort_buffer)
 
670
    free(sort_buffer);
 
671
  if (flush_io_cache(outfile))
 
672
    error=1;
 
673
 
 
674
  /* Setup io_cache for reading */
 
675
  save_pos=outfile->pos_in_file;
 
676
  if (reinit_io_cache(outfile,READ_CACHE,0L,0,0))
 
677
    error=1;
 
678
  outfile->end_of_file=save_pos;
 
679
  return error;
221
680
}
222
 
 
223
 
} /* namespace drizzled */