← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to storage/heap/hp_create.c

  • Committer: Mark Atwood
  • Date: 2008-10-03 01:39:40 UTC
  • mto: This revision was merged to the branch mainline in revision 437.
  • Revision ID: mark@fallenpegasus.com-20081003013940-mvefjo725dltz41h
rename logging_noop to logging_query

Show diffs side-by-side

added added

removed removed

Lines of Context:
storage/heap/hp_create.c
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 */
15
 
 
16
 
#include "heap_priv.h"
17
 
 
18
 
#include <drizzled/common.h>
19
 
#include <drizzled/error.h>
20
 
 
21
 
#include <string.h>
22
 
#include <algorithm>
23
 
 
24
 
using namespace std;
25
 
using namespace drizzled;
26
 
 
27
 
static void init_block(HP_BLOCK *block,uint32_t chunk_length, uint32_t min_records,
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
15
 
 
16
#include "heapdef.h"
 
17
#include "libdrizzle/drizzle_com.h"
 
18
#include "drizzled/error.h"
 
19
 
 
20
static int keys_compare(heap_rb_param *param, uchar *key1, uchar *key2);
 
21
static void init_block(HP_BLOCK *block,uint chunk_length, uint32_t min_records,
28
22
                        uint32_t max_records);
29
23
 
30
 
static const int FIXED_REC_OVERHEAD = (sizeof(unsigned char));
31
 
static const int VARIABLE_REC_OVERHEAD = (sizeof(unsigned char**) + ALIGN_SIZE(sizeof(unsigned char)));
32
 
 
 
24
#define FIXED_REC_OVERHEAD (sizeof(uchar))
 
25
#define VARIABLE_REC_OVERHEAD (sizeof(uchar**) + sizeof(uchar))
 
26
 
33
27
/* Minimum size that a chunk can take, 12 bytes on 32bit, 24 bytes on 64bit */
34
 
static const int VARIABLE_MIN_CHUNK_SIZE =
35
 
        ((sizeof(unsigned char**) + VARIABLE_REC_OVERHEAD + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1));
 
28
#define VARIABLE_MIN_CHUNK_SIZE \
 
29
        ((sizeof(uchar**) + VARIABLE_REC_OVERHEAD + sizeof(uchar**) - 1) & ~(sizeof(uchar**) - 1))
36
30
 
37
31
 
38
32
/* Create a heap table */
39
33
 
40
 
int heap_create(const char *name, uint32_t keys, HP_KEYDEF *keydef,
41
 
                uint32_t columns,
42
 
                uint32_t key_part_size,
43
 
                uint32_t reclength, uint32_t keys_memory_size,
44
 
                uint32_t max_records, uint32_t min_records,
45
 
                HP_CREATE_INFO *create_info, HP_SHARE **res)
 
34
int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
 
35
    uint columns, HP_COLUMNDEF *columndef,
 
36
    uint max_key_fieldnr, uint key_part_size,
 
37
    uint reclength, uint keys_memory_size,
 
38
    uint32_t max_records, uint32_t min_records,
 
39
    HP_CREATE_INFO *create_info, HP_SHARE **res)
46
40
{
47
 
  uint32_t i, key_segs, max_length, length;
 
41
  uint i, j, key_segs, max_length, length;
48
42
  uint32_t max_rows_for_stated_memory;
49
43
  HP_SHARE *share= 0;
50
44
  HA_KEYSEG *keyseg;
51
45
 
52
 
  if (not create_info->internal_table)
 
46
  if (!create_info->internal_table)
53
47
  {
54
 
    THR_LOCK_heap.lock();
 
48
    pthread_mutex_lock(&THR_LOCK_heap);
55
49
    if ((share= hp_find_named_heap(name)) && share->open_count == 0)
56
50
    {
57
51
      hp_free(share);
58
52
      share= 0;
59
53
    }
60
 
  }
 
54
  }  
61
55
 
62
56
  if (!share)
63
57
  {
64
 
    size_t chunk_dataspace_length;
65
 
    uint32_t chunk_length;
66
 
    uint32_t fixed_data_length, fixed_column_count;
 
58
    uint chunk_dataspace_length, chunk_length, is_variable_size;
 
59
    uint fixed_data_length, fixed_column_count;
67
60
    HP_KEYDEF *keyinfo;
68
61
 
69
62
    if (create_info->max_chunk_size)
70
63
    {
71
 
      uint32_t configured_chunk_size= create_info->max_chunk_size;
 
64
      uint configured_chunk_size= create_info->max_chunk_size;
72
65
 
73
66
      /* User requested variable-size records, let's see if they're possible */
74
67
 
75
68
      if (configured_chunk_size < key_part_size)
76
69
      {
77
70
        /* Eventual chunk_size cannot be smaller than key data,
78
 
          which allows all keys to fit into the first chunk */
 
71
           which allows all keys to fit into the first chunk */
79
72
        my_error(ER_CANT_USE_OPTION_HERE, MYF(0), "block_size");
80
 
        THR_LOCK_heap.unlock();
 
73
        pthread_mutex_unlock(&THR_LOCK_heap);
81
74
        return(ER_CANT_USE_OPTION_HERE);
82
75
      }
83
76
 
84
 
      /* max_chunk_size is near the full reclength, let's use fixed size */
85
 
      chunk_dataspace_length= reclength;
 
77
      if ((reclength - configured_chunk_size) >= VARIABLE_MIN_CHUNK_SIZE<<1)
 
78
      {
 
79
        /* Allow variable size only if we're saving at least two smallest chunks */
 
80
        /* There has to be at least one field after indexed fields */
 
81
        /* Note that NULL bits are already included in key_part_size */
 
82
        is_variable_size= 1;
 
83
        chunk_dataspace_length= configured_chunk_size;
 
84
      }
 
85
      else
 
86
      {
 
87
        /* max_chunk_size is near the full reclength, let's use fixed size */
 
88
        is_variable_size= 0;
 
89
        chunk_dataspace_length= reclength;
 
90
      }
 
91
    }
 
92
    else if (create_info->is_dynamic)
 
93
    {
 
94
      /* User asked for dynamic records - use 256 as the chunk size */
 
95
      if ((key_part_size + VARIABLE_REC_OVERHEAD) > 256)
 
96
        chunk_dataspace_length= key_part_size;
 
97
      else
 
98
        chunk_dataspace_length= 256 - VARIABLE_REC_OVERHEAD;
 
99
 
 
100
      is_variable_size= 1;
86
101
    }
87
102
    else
88
103
    {
89
104
      /* if max_chunk_size is not specified, put the whole record in one chunk */
 
105
      is_variable_size= 0;
90
106
      chunk_dataspace_length= reclength;
91
107
    }
92
108
 
 
109
    if (is_variable_size)
 
110
    {
 
111
      /* Check whether we have any variable size records past key data */
 
112
      uint has_variable_fields= 0;
 
113
 
 
114
      fixed_data_length= key_part_size;
 
115
      fixed_column_count= max_key_fieldnr;
 
116
 
 
117
      for (i= max_key_fieldnr; i < columns; i++)
 
118
      {
 
119
        HP_COLUMNDEF* column= columndef + i;
 
120
        if (column->type == DRIZZLE_TYPE_VARCHAR && column->length >= 32)
 
121
        {
 
122
            /* The field has to be >= 5.0.3 true VARCHAR and have substantial length */
 
123
            /* TODO: do we want to calculate minimum length? */
 
124
            has_variable_fields= 1;
 
125
            break;
 
126
        }
 
127
 
 
128
        if (has_variable_fields)
 
129
        {
 
130
          break;
 
131
        }
 
132
 
 
133
        if ((column->offset + column->length) <= chunk_dataspace_length)
 
134
        {
 
135
          /* Still no variable-size columns, add one fixed-length */
 
136
          fixed_column_count= i + 1;
 
137
          fixed_data_length= column->offset + column->length;
 
138
        }
 
139
      }
 
140
 
 
141
      if (!has_variable_fields)
 
142
      {
 
143
        /* There is no need to use variable-size records without variable-size columns */
 
144
        /* Reset sizes if it's not variable size anymore */
 
145
        is_variable_size= 0;
 
146
        chunk_dataspace_length= reclength;
 
147
        fixed_data_length= reclength;
 
148
        fixed_column_count= columns;
 
149
      }
 
150
    }
 
151
    else
93
152
    {
94
153
      fixed_data_length= reclength;
95
154
      fixed_column_count= columns;
96
155
    }
97
156
 
98
157
    /*
99
 
      We store unsigned char* del_link inside the data area of deleted records,
100
 
      so the data length should be at least sizeof(unsigned char*)
 
158
      We store uchar* del_link inside the data area of deleted records,
 
159
      so the data length should be at least sizeof(uchar*)
101
160
    */
102
 
    set_if_bigger(chunk_dataspace_length, sizeof (unsigned char**));
 
161
    set_if_bigger(chunk_dataspace_length, sizeof (uchar**));
103
162
 
 
163
    if (is_variable_size)
 
164
    {
 
165
      chunk_length= chunk_dataspace_length + VARIABLE_REC_OVERHEAD;
 
166
    }
 
167
    else
104
168
    {
105
169
      chunk_length= chunk_dataspace_length + FIXED_REC_OVERHEAD;
106
170
    }
107
171
 
108
172
    /* Align chunk length to the next pointer */
109
 
    chunk_length= (uint) (chunk_length + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1);
110
 
 
111
 
 
112
 
 
 
173
    chunk_length= (uint) (chunk_length + sizeof(uchar**) - 1) & ~(sizeof(uchar**) - 1);
 
174
 
 
175
 
 
176
    
113
177
    for (i= key_segs= max_length= 0, keyinfo= keydef; i < keys; i++, keyinfo++)
114
178
    {
115
179
      memset(&keyinfo->block, 0, sizeof(keyinfo->block));
116
 
      for (uint32_t j= length= 0; j < keyinfo->keysegs; j++)
 
180
      memset(&keyinfo->rb_tree , 0, sizeof(keyinfo->rb_tree));
 
181
      for (j= length= 0; j < keyinfo->keysegs; j++)
117
182
      {
118
183
        length+= keyinfo->seg[j].length;
119
184
        if (keyinfo->seg[j].null_bit)
121
186
          length++;
122
187
          if (!(keyinfo->flag & HA_NULL_ARE_EQUAL))
123
188
            keyinfo->flag|= HA_NULL_PART_KEY;
 
189
          if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
 
190
            keyinfo->rb_tree.size_of_element++;
124
191
        }
125
192
        switch (keyinfo->seg[j].type) {
 
193
        case HA_KEYTYPE_SHORT_INT:
126
194
        case HA_KEYTYPE_LONG_INT:
 
195
        case HA_KEYTYPE_FLOAT:
127
196
        case HA_KEYTYPE_DOUBLE:
 
197
        case HA_KEYTYPE_USHORT_INT:
128
198
        case HA_KEYTYPE_ULONG_INT:
129
199
        case HA_KEYTYPE_LONGLONG:
130
200
        case HA_KEYTYPE_ULONGLONG:
 
201
        case HA_KEYTYPE_INT24:
 
202
        case HA_KEYTYPE_UINT24:
 
203
        case HA_KEYTYPE_INT8:
131
204
          keyinfo->seg[j].flag|= HA_SWAP_KEY;
132
205
          break;
133
206
        case HA_KEYTYPE_VARBINARY1:
159
232
        }
160
233
      }
161
234
      keyinfo->length= length;
 
235
      length+= keyinfo->rb_tree.size_of_element + 
 
236
               ((keyinfo->algorithm == HA_KEY_ALG_BTREE) ? sizeof(uchar*) : 0);
162
237
      if (length > max_length)
163
238
        max_length= length;
164
239
      key_segs+= keyinfo->keysegs;
 
240
      if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
 
241
      {
 
242
        key_segs++; /* additional HA_KEYTYPE_END segment */
 
243
        if (keyinfo->flag & HA_VAR_LENGTH_KEY)
 
244
          keyinfo->get_key_length= hp_rb_var_key_length;
 
245
        else if (keyinfo->flag & HA_NULL_PART_KEY)
 
246
          keyinfo->get_key_length= hp_rb_null_key_length;
 
247
        else
 
248
          keyinfo->get_key_length= hp_rb_key_length;
 
249
      }
165
250
    }
166
 
    share= new HP_SHARE;
167
 
 
168
 
    if (keys && !(share->keydef= new HP_KEYDEF[keys]))
169
 
      goto err;
170
 
    if (keys && !(share->keydef->seg= new HA_KEYSEG[key_segs]))
 
251
    if (!(share= (HP_SHARE*) my_malloc((uint) sizeof(HP_SHARE)+
 
252
                                       keys*sizeof(HP_KEYDEF)+
 
253
                                       columns*sizeof(HP_COLUMNDEF)+
 
254
                                       key_segs*sizeof(HA_KEYSEG),
 
255
                                       MYF(MY_ZEROFILL))))
171
256
      goto err;
172
257
 
173
258
    /*
175
260
       Calculate the very maximum number of rows (if everything was one chunk) and
176
261
       then take either that value or configured max_records (pick smallest one)
177
262
    */
178
 
    max_rows_for_stated_memory= (uint32_t)(create_info->max_table_size /
 
263
    max_rows_for_stated_memory= (ha_rows) (create_info->max_table_size /
179
264
      (keys_memory_size + chunk_length));
180
 
    max_records = ((max_records && max_records < max_rows_for_stated_memory) ?
 
265
    max_records = ((max_records && max_records < max_rows_for_stated_memory) ? 
181
266
                      max_records : max_rows_for_stated_memory);
 
267
 
 
268
    share->column_defs= (HP_COLUMNDEF*) (share + 1);
 
269
    memcpy(share->column_defs, columndef, (size_t) (sizeof(columndef[0]) * columns));
182
270
 
 
271
    share->keydef= (HP_KEYDEF*) (share->column_defs + columns);    
183
272
    share->key_stat_version= 1;
184
 
    keyseg= keys ? share->keydef->seg : NULL;
185
 
 
 
273
    keyseg= (HA_KEYSEG*) (share->keydef + keys);
186
274
    init_block(&share->recordspace.block, chunk_length, min_records, max_records);
187
275
    /* Fix keys */
188
276
    memcpy(share->keydef, keydef, (size_t) (sizeof(keydef[0]) * keys));
192
280
      memcpy(keyseg, keydef[i].seg,
193
281
             (size_t) (sizeof(keyseg[0]) * keydef[i].keysegs));
194
282
      keyseg+= keydef[i].keysegs;
 
283
 
 
284
      if (keydef[i].algorithm == HA_KEY_ALG_BTREE)
 
285
      {
 
286
        /* additional HA_KEYTYPE_END keyseg */
 
287
        keyseg->type=     HA_KEYTYPE_END;
 
288
        keyseg->length=   sizeof(uchar*);
 
289
        keyseg->flag=     0;
 
290
        keyseg->null_bit= 0;
 
291
        keyseg++;
 
292
 
 
293
        init_tree(&keyinfo->rb_tree, 0, 0, sizeof(uchar*),
 
294
                  (qsort_cmp2)keys_compare, 1, NULL, NULL);
 
295
        keyinfo->delete_key= hp_rb_delete_key;
 
296
        keyinfo->write_key= hp_rb_write_key;
 
297
      }
 
298
      else
195
299
      {
196
300
        init_block(&keyinfo->block, sizeof(HASH_INFO), min_records,
197
301
                   max_records);
 
302
        keyinfo->delete_key= hp_delete_key;
 
303
        keyinfo->write_key= hp_write_key;
198
304
        keyinfo->hash_buckets= 0;
199
305
      }
200
306
      if ((keyinfo->flag & HA_AUTO_KEY) && create_info->with_auto_increment)
218
324
 
219
325
    share->recordspace.chunk_length= chunk_length;
220
326
    share->recordspace.chunk_dataspace_length= chunk_dataspace_length;
 
327
    share->recordspace.is_variable_size= is_variable_size;
221
328
    share->recordspace.total_data_length= 0;
222
329
 
223
 
    {
 
330
    if (is_variable_size) {
 
331
      share->recordspace.offset_link= chunk_dataspace_length;
 
332
      share->recordspace.offset_status= share->recordspace.offset_link + sizeof(uchar**);
 
333
    } else {
224
334
      share->recordspace.offset_link= 1<<22; /* Make it likely to fail if anyone uses this offset */
225
335
      share->recordspace.offset_status= chunk_dataspace_length;
226
336
    }
227
337
 
228
338
    /* Must be allocated separately for rename to work */
229
 
    share->name.append(name);
 
339
    if (!(share->name= my_strdup(name,MYF(0))))
 
340
    {
 
341
      my_free((uchar*) share,MYF(0));
 
342
      goto err;
 
343
    }
 
344
    thr_lock_init(&share->lock);
 
345
    VOID(pthread_mutex_init(&share->intern_lock,MY_MUTEX_INIT_FAST));
230
346
    if (!create_info->internal_table)
231
347
    {
232
 
      heap_share_list.push_front(share);
 
348
      share->open_list.data= (void*) share;
 
349
      heap_share_list= list_add(heap_share_list,&share->open_list);
233
350
    }
234
351
    else
235
352
      share->delete_on_close= 1;
236
353
  }
237
354
  if (!create_info->internal_table)
238
 
    THR_LOCK_heap.unlock();
 
355
    pthread_mutex_unlock(&THR_LOCK_heap);
239
356
 
240
357
  *res= share;
241
358
  return(0);
242
359
 
243
360
err:
244
 
  if (share && share->keydef && share->keydef->seg)
245
 
    delete [] share->keydef->seg;
246
 
  if (share && share->keydef)
247
 
    delete [] share->keydef;
248
 
  if (share)
249
 
    delete share;
250
 
  if (not create_info->internal_table)
251
 
    THR_LOCK_heap.unlock();
 
361
  if (!create_info->internal_table)
 
362
    pthread_mutex_unlock(&THR_LOCK_heap);
252
363
  return(1);
253
364
} /* heap_create */
254
365
 
255
366
 
256
 
static void init_block(HP_BLOCK *block, uint32_t chunk_length, uint32_t min_records,
 
367
static int keys_compare(heap_rb_param *param, uchar *key1, uchar *key2)
 
368
{
 
369
  uint not_used[2];
 
370
  return ha_key_cmp(param->keyseg, key1, key2, param->key_length, 
 
371
                    param->search_flag, not_used);
 
372
}
 
373
 
 
374
static void init_block(HP_BLOCK *block, uint chunk_length, uint32_t min_records,
257
375
                       uint32_t max_records)
258
376
{
259
 
  uint32_t recbuffer,records_in_block;
 
377
  uint i,recbuffer,records_in_block;
260
378
 
261
379
  max_records= max(min_records,max_records);
262
380
  if (!max_records)
263
381
    max_records= 1000;                  /* As good as quess as anything */
264
 
 
 
382
  
265
383
  /* we want to start each chunk at 8 bytes boundary, round recbuffer to the next 8 */
266
 
  recbuffer= (uint) (chunk_length + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1);
 
384
  recbuffer= (uint) (chunk_length + sizeof(uchar**) - 1) & ~(sizeof(uchar**) - 1);  
267
385
  records_in_block= max_records / 10;
268
386
  if (records_in_block < 10 && max_records)
269
387
    records_in_block= 10;
270
388
  if (!records_in_block || records_in_block*recbuffer >
271
 
      (internal::my_default_record_cache_size-sizeof(HP_PTRS)*HP_MAX_LEVELS))
272
 
    records_in_block= (internal::my_default_record_cache_size - sizeof(HP_PTRS) *
 
389
      (my_default_record_cache_size-sizeof(HP_PTRS)*HP_MAX_LEVELS))
 
390
    records_in_block= (my_default_record_cache_size - sizeof(HP_PTRS) *
273
391
                      HP_MAX_LEVELS) / recbuffer + 1;
274
392
  block->records_in_block= records_in_block;
275
393
  block->recbuffer= recbuffer;
276
394
  block->last_allocated= 0L;
277
395
 
278
 
  for (uint32_t i= 0; i <= HP_MAX_LEVELS; i++)
279
 
  {
 
396
  for (i= 0; i <= HP_MAX_LEVELS; i++)
280
397
    block->level_info[i].records_under_level=
281
398
      (!i ? 1 : i == 1 ? records_in_block :
282
399
       HP_PTRS_IN_NOD * block->level_info[i - 1].records_under_level);
283
 
  }
284
400
}
285
401
 
286
402
 
298
414
  int result;
299
415
  register HP_SHARE *share;
300
416
 
301
 
  THR_LOCK_heap.lock();
 
417
  pthread_mutex_lock(&THR_LOCK_heap);
302
418
  if ((share= hp_find_named_heap(name)))
303
419
  {
304
420
    heap_try_free(share);
306
422
  }
307
423
  else
308
424
  {
309
 
    result= errno=ENOENT;
 
425
    result= my_errno=ENOENT;
310
426
  }
311
 
  THR_LOCK_heap.unlock();
 
427
  pthread_mutex_unlock(&THR_LOCK_heap);
312
428
  return(result);
313
429
}
314
430
 
315
431
 
 
432
void heap_drop_table(HP_INFO *info)
 
433
{
 
434
  pthread_mutex_lock(&THR_LOCK_heap);
 
435
  heap_try_free(info->s);
 
436
  pthread_mutex_unlock(&THR_LOCK_heap);
 
437
  return;
 
438
}
 
439
 
 
440
 
316
441
void hp_free(HP_SHARE *share)
317
442
{
318
 
  heap_share_list.remove(share);        /* If not internal table */
 
443
  if (share->open_list.data)                    /* If not internal table */
 
444
    heap_share_list= list_delete(heap_share_list, &share->open_list);
319
445
  hp_clear(share);                      /* Remove blocks from memory */
320
 
  if (share->keydef)
321
 
    delete [] share->keydef->seg;
322
 
  delete [] share->keydef;
323
 
  delete share;
 
446
  thr_lock_delete(&share->lock);
 
447
  VOID(pthread_mutex_destroy(&share->intern_lock));
 
448
  my_free((uchar*) share->name, MYF(0));
 
449
  my_free((uchar*) share, MYF(0));
 
450
  return;
324
451
}